ScadaBridge/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/DeploymentManagerRepository.cs

using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.ExternalSystems;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Deployment;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;

/// <summary>
/// EF Core implementation of <see cref="IDeploymentManagerRepository"/> covering
/// the deployment pipeline's persistence surface: <c>DeploymentRecord</c> CRUD
/// (with optimistic concurrency via <c>DeploymentRecord.RowVersion</c>),
/// <c>SystemArtifactDeploymentRecord</c> CRUD, <c>DeployedConfigSnapshot</c> CRUD,
/// and a Restrict-FK-aware <see cref="DeleteInstanceAsync"/> that explicitly
/// clears dependent deployment-record rows before removing an instance.
/// </summary>
public class DeploymentManagerRepository : IDeploymentManagerRepository
{
    private readonly ScadaBridgeDbContext _dbContext;

    /// <summary>
    /// Initializes a new instance of the DeploymentManagerRepository class.
    /// </summary>
    /// <param name="dbContext">The database context for accessing deployment data.</param>
    public DeploymentManagerRepository(ScadaBridgeDbContext dbContext)
    {
        _dbContext = dbContext ?? throw new ArgumentNullException(nameof(dbContext));
    }

    // --- DeploymentRecord ---

    /// <inheritdoc />
    public async Task<DeploymentRecord?> GetDeploymentRecordByIdAsync(int id, CancellationToken cancellationToken = default)
    {
        return await _dbContext.DeploymentRecords.FindAsync([id], cancellationToken);
    }

    /// <summary>
    /// Shared filtered join behind <see cref="QueryDeploymentListPageAsync"/> and
    /// <see cref="GetDeploymentStatusCountsAsync"/>, so the two reads that back one
    /// screen can never disagree about which rows they describe.
    ///
    /// <para>
    /// The join to <c>Instance</c> is an INNER join, which is exact rather than
    /// merely convenient: <c>DeploymentRecord</c> has a <c>Restrict</c> FK to
    /// <c>Instance</c> and <see cref="DeleteInstanceAsync"/> removes the records
    /// before the instance, so an orphaned deployment record cannot exist. The join
    /// is what lets site scoping run in SQL — <c>DeploymentRecord</c> itself has no
    /// <c>SiteId</c>.
    /// </para>
    /// </summary>
    /// <param name="filter">Filter to apply.</param>
    /// <param name="applyStatus">
    /// When <see langword="false"/>, the status constraint is skipped — the
    /// status-count aggregation needs every status of the otherwise-identical set.
    /// </param>
    /// <returns>The composed, unordered, untracked query.</returns>
    internal IQueryable<DeploymentInstanceJoin> BuildListQuery(DeploymentListFilter filter, bool applyStatus)
    {
        ArgumentNullException.ThrowIfNull(filter);

        var query =
            from d in _dbContext.DeploymentRecords.AsNoTracking()
            join i in _dbContext.Set<Instance>().AsNoTracking() on d.InstanceId equals i.Id
            select new DeploymentInstanceJoin
            {
                Id = d.Id,
                DeploymentId = d.DeploymentId,
                InstanceId = d.InstanceId,
                InstanceUniqueName = i.UniqueName,
                SiteId = i.SiteId,
                Status = d.Status,
                RevisionHash = d.RevisionHash,
                DeployedBy = d.DeployedBy,
                DeployedAt = d.DeployedAt,
                CompletedAt = d.CompletedAt,
                ErrorMessage = d.ErrorMessage
            };

        if (filter.InstanceId.HasValue)
            query = query.Where(r => r.InstanceId == filter.InstanceId.Value);

        if (applyStatus && filter.Status.HasValue)
            query = query.Where(r => r.Status == filter.Status.Value);

        if (filter.SiteIdScope != null)
        {
            // An EMPTY scope is a real filter — a user permitted no sites sees
            // nothing — so it must not be short-circuited into "no filter".
            var scope = filter.SiteIdScope as int[] ?? filter.SiteIdScope.ToArray();
            query = query.Where(r => scope.Contains(r.SiteId));
        }

        if (filter.EffectiveDeployedFrom(DateTimeOffset.UtcNow) is { } from)
            query = query.Where(r => r.DeployedAt >= from);

        if (filter.NormalizedSearch is { } search)
        {
            // Default: PREFIX match across the four identifiers a deployment is findable
            // by (LIKE 'term%', index-seekable; case-insensitivity is the database
            // collation's). Contains is the explicit opt-in scan (LIKE '%term%') and is
            // always bounded by the DeployedAt window applied above (ScadaBridge#50).
            query = filter.SearchMode == DeploymentSearchMode.Contains
                ? query.Where(r =>
                    r.InstanceUniqueName.Contains(search)
                    || r.DeploymentId.Contains(search)
                    || r.DeployedBy.Contains(search)
                    || (r.RevisionHash != null && r.RevisionHash.Contains(search)))
                : query.Where(r =>
                    r.InstanceUniqueName.StartsWith(search)
                    || r.DeploymentId.StartsWith(search)
                    || r.DeployedBy.StartsWith(search)
                    || (r.RevisionHash != null && r.RevisionHash.StartsWith(search)));
        }

        return query;
    }

    /// <inheritdoc />
    public async Task<DeploymentListPage> QueryDeploymentListPageAsync(
        DeploymentListFilter filter,
        int pageNumber,
        int pageSize,
        CancellationToken cancellationToken = default)
    {
        var page = Math.Max(1, pageNumber);
        var size = Math.Clamp(pageSize, 1, DeploymentListPage.MaxPageSize);

        var query = BuildListQuery(filter, applyStatus: true);

        // Count first: the pager needs a page count, and the caller must be able to
        // tell "page 7 is now past the end" (a concurrent purge or a narrowed filter)
        // from "page 7 is empty".
        var totalCount = await query.CountAsync(cancellationToken);

        // DeployedAt ties on rapid redeploys; without the Id tie-break the sort key
        // is unstable and offset paging repeats or drops rows between pages.
        var rows = await query
            .OrderByDescending(r => r.DeployedAt)
            .ThenByDescending(r => r.Id)
            .Skip((page - 1) * size)
            .Take(size)
            .Select(r => new DeploymentListRow(
                r.Id,
                r.DeploymentId,
                r.InstanceId,
                r.InstanceUniqueName,
                r.SiteId,
                r.Status,
                r.RevisionHash,
                r.DeployedBy,
                r.DeployedAt,
                r.CompletedAt,
                r.ErrorMessage))
            .ToListAsync(cancellationToken);

        return new DeploymentListPage(rows, totalCount);
    }

    /// <inheritdoc />
    public async Task<DeploymentStatusCounts> GetDeploymentStatusCountsAsync(
        DeploymentListFilter filter,
        CancellationToken cancellationToken = default)
    {
        // applyStatus: false — the tiles are the status breakdown of the filtered
        // set, so the active status filter is the one dimension they must NOT honour.
        var grouped = await BuildListQuery(filter, applyStatus: false)
            .GroupBy(r => r.Status)
            .Select(g => new { Status = g.Key, Count = g.Count() })
            .ToListAsync(cancellationToken);

        return DeploymentStatusCounts.FromGrouped(grouped.Select(g => (g.Status, g.Count)));
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<DeploymentRecord>> GetDeploymentsByInstanceIdAsync(int instanceId, CancellationToken cancellationToken = default)
    {
        return await _dbContext.DeploymentRecords
            .Where(d => d.InstanceId == instanceId)
            .OrderByDescending(d => d.DeployedAt)
            .ToListAsync(cancellationToken);
    }

    /// <inheritdoc />
    public async Task<DeploymentRecord?> GetCurrentDeploymentStatusAsync(int instanceId, CancellationToken cancellationToken = default)
    {
        // Deployments are insert-only (one row per deploy
        // attempt), so two records for the same instance can tie on DeployedAt when
        // they are created within the same clock tick (a rapid redeploy, or a
        // redeploy immediately after a timed-out attempt). SQL Server's choice
        // between equal sort keys is undefined, so reconciliation could read the
        // wrong "current" record. ThenByDescending(d => d.Id) makes the read
        // deterministic — the highest Id (the most recently inserted row) wins.
        return await _dbContext.DeploymentRecords
            .Where(d => d.InstanceId == instanceId)
            .OrderByDescending(d => d.DeployedAt)
            .ThenByDescending(d => d.Id)
            .FirstOrDefaultAsync(cancellationToken);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyDictionary<int, DeploymentRecord>> GetCurrentDeploymentStatusesAsync(
        IReadOnlyCollection<int> instanceIds, CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(instanceIds);
        if (instanceIds.Count == 0)
            return new Dictionary<int, DeploymentRecord>();

        var ids = instanceIds.Distinct().ToArray();

        // "Current" = no newer record for the same instance, with the same (DeployedAt, Id)
        // tie-break GetCurrentDeploymentStatusAsync uses. Resolved in SQL so only one row per
        // instance comes back however long each instance's deploy history is.
        var records = await _dbContext.DeploymentRecords
            .AsNoTracking()
            .Where(d => ids.Contains(d.InstanceId))
            .Where(d => !_dbContext.DeploymentRecords.Any(o =>
                o.InstanceId == d.InstanceId
                && (o.DeployedAt > d.DeployedAt || (o.DeployedAt == d.DeployedAt && o.Id > d.Id))))
            .ToListAsync(cancellationToken);

        return records.ToDictionary(d => d.InstanceId);
    }

    /// <inheritdoc />
    public async Task<DeploymentRecord?> GetDeploymentByDeploymentIdAsync(string deploymentId, CancellationToken cancellationToken = default)
    {
        return await _dbContext.DeploymentRecords
            .FirstOrDefaultAsync(d => d.DeploymentId == deploymentId, cancellationToken);
    }

    /// <inheritdoc />
    public async Task AddDeploymentRecordAsync(DeploymentRecord record, CancellationToken cancellationToken = default)
    {
        await _dbContext.DeploymentRecords.AddAsync(record, cancellationToken);
    }

    /// <inheritdoc />
    public Task UpdateDeploymentRecordAsync(DeploymentRecord record, CancellationToken cancellationToken = default)
    {
        _dbContext.DeploymentRecords.Update(record);
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public Task DeleteDeploymentRecordAsync(int id, byte[] expectedRowVersion, CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(expectedRowVersion);

        // DeploymentRecord carries a SQL Server rowversion concurrency token.
        // The stub-attach delete path must seed EF's OriginalValues["RowVersion"] with
        // the caller's last-observed value so the generated SQL becomes
        // `DELETE ... WHERE Id = @id AND RowVersion = @prior`. Without this seeding a
        // concurrent edit is silently overwritten; with it, EF raises
        // DbUpdateConcurrencyException on SaveChangesAsync — the documented
        // optimistic-concurrency contract on deployment status records.
        var record = _dbContext.DeploymentRecords.Local.FirstOrDefault(d => d.Id == id);
        if (record != null)
        {
            var entry = _dbContext.Entry(record);
            entry.OriginalValues["RowVersion"] = expectedRowVersion;
            _dbContext.DeploymentRecords.Remove(record);
        }
        else
        {
            var stub = new DeploymentRecord("stub", "stub") { Id = id };
            _dbContext.DeploymentRecords.Attach(stub);
            var entry = _dbContext.Entry(stub);
            entry.OriginalValues["RowVersion"] = expectedRowVersion;
            _dbContext.DeploymentRecords.Remove(stub);
        }
        return Task.CompletedTask;
    }

    // --- SystemArtifactDeploymentRecord ---

    /// <inheritdoc />
    public async Task<SystemArtifactDeploymentRecord?> GetSystemArtifactDeploymentByIdAsync(int id, CancellationToken cancellationToken = default)
    {
        return await _dbContext.SystemArtifactDeploymentRecords.FindAsync([id], cancellationToken);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<SystemArtifactDeploymentRecord>> GetAllSystemArtifactDeploymentsAsync(CancellationToken cancellationToken = default)
    {
        return await _dbContext.SystemArtifactDeploymentRecords
            .OrderByDescending(d => d.DeployedAt)
            .ToListAsync(cancellationToken);
    }

    /// <inheritdoc />
    public async Task AddSystemArtifactDeploymentAsync(SystemArtifactDeploymentRecord record, CancellationToken cancellationToken = default)
    {
        await _dbContext.SystemArtifactDeploymentRecords.AddAsync(record, cancellationToken);
    }

    /// <inheritdoc />
    public Task UpdateSystemArtifactDeploymentAsync(SystemArtifactDeploymentRecord record, CancellationToken cancellationToken = default)
    {
        _dbContext.SystemArtifactDeploymentRecords.Update(record);
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public Task DeleteSystemArtifactDeploymentAsync(int id, CancellationToken cancellationToken = default)
    {
        var record = _dbContext.SystemArtifactDeploymentRecords.Local.FirstOrDefault(d => d.Id == id);
        if (record != null)
        {
            _dbContext.SystemArtifactDeploymentRecords.Remove(record);
        }
        else
        {
            var stub = new SystemArtifactDeploymentRecord("stub", "stub") { Id = id };
            _dbContext.SystemArtifactDeploymentRecords.Attach(stub);
            _dbContext.SystemArtifactDeploymentRecords.Remove(stub);
        }
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public async Task<int> PurgeSystemArtifactDeploymentRecordsAsync(
        DateTimeOffset cutoffUtc,
        int batchSize,
        CancellationToken cancellationToken = default)
    {
        // Same bounded-batch shape as PurgeTerminalDeploymentRecordsAsync: the caller
        // re-invokes until fewer than batchSize rows come back. Every row is terminal by
        // construction (it is written once, after the artifact sweep finished).
        var stale = await _dbContext.SystemArtifactDeploymentRecords
            .Where(d => d.DeployedAt < cutoffUtc)
            .OrderBy(d => d.Id)
            .Take(batchSize)
            .ToListAsync(cancellationToken);

        if (stale.Count == 0)
            return 0;

        _dbContext.SystemArtifactDeploymentRecords.RemoveRange(stale);
        await _dbContext.SaveChangesAsync(cancellationToken);
        return stale.Count;
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<ExternalSystemMethod>> GetAllExternalSystemMethodsAsync(
        CancellationToken cancellationToken = default)
    {
        return await _dbContext.Set<ExternalSystemMethod>()
            .AsNoTracking()
            .OrderBy(m => m.ExternalSystemDefinitionId)
            .ThenBy(m => m.Id)
            .ToListAsync(cancellationToken);
    }

    // --- DeployedConfigSnapshot ---

    /// <inheritdoc />
    public async Task<DeployedConfigSnapshot?> GetDeployedSnapshotByInstanceIdAsync(int instanceId, CancellationToken cancellationToken = default)
    {
        return await _dbContext.Set<DeployedConfigSnapshot>()
            .FirstOrDefaultAsync(s => s.InstanceId == instanceId, cancellationToken);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyDictionary<int, string>> GetDeployedRevisionHashesAsync(CancellationToken cancellationToken = default)
    {
        // Projection only: ConfigurationJson is never materialised, so this stays cheap enough to
        // read for the whole fleet in one round trip.
        return await _dbContext.Set<DeployedConfigSnapshot>()
            .AsNoTracking()
            .Select(s => new { s.InstanceId, s.RevisionHash })
            .ToDictionaryAsync(s => s.InstanceId, s => s.RevisionHash, cancellationToken);
    }

    /// <inheritdoc />
    public async Task AddDeployedSnapshotAsync(DeployedConfigSnapshot snapshot, CancellationToken cancellationToken = default)
    {
        await _dbContext.Set<DeployedConfigSnapshot>().AddAsync(snapshot, cancellationToken);
    }

    /// <inheritdoc />
    public Task UpdateDeployedSnapshotAsync(DeployedConfigSnapshot snapshot, CancellationToken cancellationToken = default)
    {
        _dbContext.Set<DeployedConfigSnapshot>().Update(snapshot);
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public async Task DeleteDeployedSnapshotAsync(int instanceId, CancellationToken cancellationToken = default)
    {
        var snapshot = await _dbContext.Set<DeployedConfigSnapshot>()
            .FirstOrDefaultAsync(s => s.InstanceId == instanceId, cancellationToken);
        if (snapshot != null)
        {
            _dbContext.Set<DeployedConfigSnapshot>().Remove(snapshot);
        }
    }

    // --- Notify-and-fetch: PendingDeployment staging store ---

    /// <inheritdoc />
    public async Task AddPendingDeploymentAsync(PendingDeployment pending, CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(pending);

        // Supersession: a newer deploy for the same instance replaces any prior pending
        // row. Safe because the per-instance operation lock serializes same-instance
        // deploys. SaveChanges is deferred to the caller, matching the other Add methods.
        //
        // Only the prior rows' KEYS are read: each row carries the whole flattened
        // ConfigurationJson (tens of KB), and materialising it just to delete it was the
        // dominant cost of the supersede. The delete itself stays a tracked Remove (not
        // ExecuteDelete) so it still commits atomically with the new row in the caller's
        // SaveChanges.
        var priorIds = await _dbContext.Set<PendingDeployment>()
            .Where(p => p.InstanceId == pending.InstanceId)
            .Select(p => p.Id)
            .ToListAsync(cancellationToken);
        foreach (var id in priorIds)
        {
            RemovePendingByKey(id);
        }
        await _dbContext.Set<PendingDeployment>().AddAsync(pending, cancellationToken);
    }

    /// <inheritdoc />
    public Task<PendingDeployment?> GetPendingDeploymentByIdAsync(string deploymentId, CancellationToken cancellationToken = default)
    {
        return _dbContext.Set<PendingDeployment>()
            .FirstOrDefaultAsync(p => p.DeploymentId == deploymentId, cancellationToken);
    }

    /// <inheritdoc />
    public Task<PendingDeployment?> GetPendingDeploymentByInstanceIdAsync(int instanceId, DateTimeOffset? nowUtc = null, CancellationToken cancellationToken = default)
    {
        // At most one pending row per instance by design (supersession + stage-if-absent),
        // but order deterministically and take the most recent so a hypothetical duplicate
        // never makes the read non-deterministic — mirrors GetCurrentDeploymentStatusAsync.
        var query = _dbContext.Set<PendingDeployment>()
            .Where(p => p.InstanceId == instanceId);

        // Expiry-aware: never hand back an EXPIRED row. Pending rows are only TTL-purged, so an
        // expired-but-unpurged row would otherwise return a token the config-fetch endpoint 404s
        // (it correctly rejects expired rows), leaving the node unhealed.
        if (nowUtc.HasValue)
        {
            var now = nowUtc.Value;
            query = query.Where(p => p.ExpiresAtUtc > now);
        }

        return query
            .OrderByDescending(p => p.CreatedAtUtc)
            .ThenByDescending(p => p.Id)
            .FirstOrDefaultAsync(cancellationToken);
    }

    /// <inheritdoc />
    public async Task DeletePendingDeploymentByIdAsync(string deploymentId, CancellationToken cancellationToken = default)
    {
        var row = await _dbContext.Set<PendingDeployment>()
            .FirstOrDefaultAsync(p => p.DeploymentId == deploymentId, cancellationToken);
        if (row != null)
        {
            _dbContext.Set<PendingDeployment>().Remove(row);
        }
    }

    /// <inheritdoc />
    public async Task<int> PurgeExpiredPendingDeploymentsAsync(DateTimeOffset nowUtc, CancellationToken cancellationToken = default)
    {
        // Self-contained TTL maintenance op: commits its own delete and returns the count.
        // A set-based DELETE — the expired rows' ConfigurationJson is never read.
        return await DeletePendingWhereAsync(p => p.ExpiresAtUtc <= nowUtc, cancellationToken);
    }

    /// <summary>
    /// Deletes every pending row matching <paramref name="predicate"/> immediately, without
    /// materialising the rows (each carries its full flattened <c>ConfigurationJson</c>).
    /// A relational provider issues one set-based <c>DELETE</c>; a non-relational provider
    /// (the EF in-memory provider some host-level tests run on) cannot translate
    /// <c>ExecuteDelete</c>, so it falls back to load-and-remove with its own save.
    /// </summary>
    /// <param name="predicate">Rows to delete.</param>
    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>The number of rows deleted.</returns>
    private async Task<int> DeletePendingWhereAsync(
        System.Linq.Expressions.Expression<Func<PendingDeployment, bool>> predicate,
        CancellationToken cancellationToken)
    {
        var query = _dbContext.Set<PendingDeployment>().Where(predicate);
        if (_dbContext.Database.IsRelational())
        {
            return await query.ExecuteDeleteAsync(cancellationToken);
        }

        var rows = await query.ToListAsync(cancellationToken);
        if (rows.Count == 0)
        {
            return 0;
        }
        _dbContext.Set<PendingDeployment>().RemoveRange(rows);
        await _dbContext.SaveChangesAsync(cancellationToken);
        return rows.Count;
    }

    /// <summary>
    /// Marks the pending row with primary key <paramref name="id"/> for deletion on the next
    /// <c>SaveChanges</c> without loading it: an already-tracked instance is removed in place,
    /// otherwise a key-only stub is attached and removed (EF deletes by key).
    /// </summary>
    /// <param name="id">Primary key of the pending row.</param>
    private void RemovePendingByKey(int id)
    {
        var set = _dbContext.Set<PendingDeployment>();
        var tracked = set.Local.FirstOrDefault(p => p.Id == id);
        if (tracked != null)
        {
            set.Remove(tracked);
            return;
        }

        var stub = new PendingDeployment(
            string.Empty, 0, string.Empty, string.Empty, string.Empty,
            default, default) { Id = id };
        set.Attach(stub);
        set.Remove(stub);
    }

    /// <inheritdoc />
    public async Task<int> PurgeTerminalDeploymentRecordsAsync(
        DateTimeOffset cutoffUtc,
        int batchSize,
        CancellationToken cancellationToken = default)
    {
        // Terminal statuses ONLY. An InProgress (or Pending) row is never purged
        // regardless of age: it is precisely the row TryReconcileWithSiteAsync
        // looks for when deciding whether a prior deploy actually landed, and
        // deleting it would silently disable the query-before-redeploy idempotency
        // guard for that instance.
        //
        // Bounded batch rather than one unbounded DELETE so the first purge on a
        // long-lived system cannot take a table-scale lock on DeploymentRecords.
        // The caller re-invokes until it returns fewer rows than the batch size.
        var stale = await _dbContext.DeploymentRecords
            .Where(d => d.CompletedAt != null
                        && d.CompletedAt < cutoffUtc
                        && (d.Status == DeploymentStatus.Success || d.Status == DeploymentStatus.Failed))
            .OrderBy(d => d.Id)
            .Take(batchSize)
            .ToListAsync(cancellationToken);

        if (stale.Count == 0)
            return 0;

        _dbContext.DeploymentRecords.RemoveRange(stale);
        await _dbContext.SaveChangesAsync(cancellationToken);
        return stale.Count;
    }

    // --- Startup reconciliation ---

    /// <inheritdoc />
    public async Task<IReadOnlyList<ExpectedDeployment>> GetExpectedDeploymentsForSiteAsync(int siteId, CancellationToken cancellationToken = default)
    {
        // Inner join: only instances that have a DeployedConfigSnapshot are returned.
        // ConfigurationJson is intentionally excluded — the caller fetches the full config
        // on demand via the HTTP endpoint for gap instances only.
        return await (
            from snap in _dbContext.Set<DeployedConfigSnapshot>()
            join inst in _dbContext.Set<Instance>() on snap.InstanceId equals inst.Id
            where inst.SiteId == siteId
            select new ExpectedDeployment(
                inst.Id,
                inst.UniqueName,
                snap.RevisionHash,
                snap.DeploymentId,
                inst.State == InstanceState.Enabled))
            .ToListAsync(cancellationToken);
    }

    /// <inheritdoc />
    public async Task<bool> StagePendingIfAbsentAsync(
        int instanceId, string deploymentId, string revisionHash,
        string configurationJson, string token,
        DateTimeOffset createdAtUtc, DateTimeOffset expiresAtUtc,
        CancellationToken cancellationToken = default)
    {
        // Treat createdAtUtc as "now". FIRST remove any EXPIRED pending rows for this instance
        // (ExpiresAtUtc <= now). Pending rows are only TTL-purged — PendingDeploymentPurgeActor
        // sweeps them on its PendingDeploymentPurgeInterval cadence — so between sweeps an
        // EXPIRED-but-unpurged row would otherwise (a) read as "a deploy is in flight" and block
        // staging — handing the node an expired token (HTTP 404) and leaving it unhealed — and
        // (b) collide on the DeploymentId UNIQUE index when a reconcile re-stages the snapshot's
        // own DeploymentId. Dropping expired rows first fixes both. The delete is set-based and
        // immediate (an expired row is dead either way), so its JSON is never read.
        await DeletePendingWhereAsync(
            p => p.InstanceId == instanceId && p.ExpiresAtUtc <= createdAtUtc, cancellationToken);

        // THEN insert-if-absent against still-LIVE rows only. A live pending row means a genuine
        // in-flight deploy (or a concurrent reconcile) already owns the slot — do NOT supersede it;
        // clobbering it could make the node fetch the reconcile token while the original deliver is
        // mid-flight. (Expired rows just removed are disjoint from this future-expiry predicate.)
        var liveExists = await _dbContext.Set<PendingDeployment>()
            .AnyAsync(p => p.InstanceId == instanceId && p.ExpiresAtUtc > createdAtUtc, cancellationToken);
        if (!liveExists)
        {
            var pending = new PendingDeployment(
                deploymentId, instanceId, revisionHash,
                configurationJson, token, createdAtUtc, expiresAtUtc);
            await _dbContext.Set<PendingDeployment>().AddAsync(pending, cancellationToken);
        }

        // Self-contained: the expired-row cleanup has already run, so the same-DeploymentId insert
        // below cannot collide with it on the unique index. Returns true only when a fresh row was
        // staged.
        await _dbContext.SaveChangesAsync(cancellationToken);
        return !liveExists;
    }

    /// <inheritdoc />
    public async Task<IReadOnlyDictionary<int, DeployedConfigSnapshot>> GetDeployedSnapshotsByInstanceIdsAsync(
        IReadOnlyCollection<int> instanceIds, CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(instanceIds);
        if (instanceIds.Count == 0)
            return new Dictionary<int, DeployedConfigSnapshot>();

        var ids = instanceIds.Distinct().ToArray();
        return await _dbContext.Set<DeployedConfigSnapshot>()
            .AsNoTracking()
            .Where(s => ids.Contains(s.InstanceId))
            .ToDictionaryAsync(s => s.InstanceId, cancellationToken);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyDictionary<int, PendingDeployment>> StagePendingBatchIfAbsentAsync(
        IReadOnlyList<PendingDeployment> candidates, DateTimeOffset nowUtc,
        CancellationToken cancellationToken = default)
    {
        ArgumentNullException.ThrowIfNull(candidates);
        var byInstance = new Dictionary<int, PendingDeployment>();
        foreach (var candidate in candidates)
            byInstance.TryAdd(candidate.InstanceId, candidate);
        if (byInstance.Count == 0)
            return byInstance;

        var ids = byInstance.Keys.ToArray();

        // Same two-step contract as StagePendingIfAbsentAsync, set-based: drop EXPIRED rows
        // first (they would otherwise read as in-flight and collide on the DeploymentId unique
        // index when a snapshot's own DeploymentId is re-staged), then insert only where no
        // LIVE row exists. A live row is never superseded.
        await DeletePendingWhereAsync(
            p => ids.Contains(p.InstanceId) && p.ExpiresAtUtc <= nowUtc, cancellationToken);

        var live = await ReadLivePendingAsync(ids, nowUtc, cancellationToken);
        var added = new List<PendingDeployment>();
        foreach (var (instanceId, candidate) in byInstance)
        {
            if (live.ContainsKey(instanceId))
                continue;
            added.Add(candidate);
        }

        if (added.Count == 0)
            return live;

        await _dbContext.Set<PendingDeployment>().AddRangeAsync(added, cancellationToken);
        try
        {
            await _dbContext.SaveChangesAsync(cancellationToken);
        }
        catch (DbUpdateException)
        {
            // The batch lost a race — most commonly the peer site node's concurrent reconcile
            // inserted the same snapshot DeploymentId first (unique index), or an instance was
            // deleted mid-reconcile (FK). Un-track the failed inserts and fall back to staging
            // one row at a time, so one bad row costs only itself and a row the peer staged is
            // picked up as the live row it now is.
            foreach (var row in added)
                _dbContext.Entry(row).State = EntityState.Detached;

            return await StagePendingOneByOneAsync(byInstance, nowUtc, cancellationToken);
        }

        foreach (var row in added)
            live[row.InstanceId] = row;
        return live;
    }

    /// <summary>
    /// Fallback for <see cref="StagePendingBatchIfAbsentAsync"/> after its single save failed:
    /// stages each candidate through <see cref="StagePendingIfAbsentAsync"/>, skipping (and
    /// omitting from the result) any single row that still cannot be written.
    /// </summary>
    private async Task<Dictionary<int, PendingDeployment>> StagePendingOneByOneAsync(
        IReadOnlyDictionary<int, PendingDeployment> byInstance, DateTimeOffset nowUtc,
        CancellationToken cancellationToken)
    {
        var result = new Dictionary<int, PendingDeployment>();
        foreach (var (instanceId, candidate) in byInstance)
        {
            bool staged;
            try
            {
                staged = await StagePendingIfAbsentAsync(
                    instanceId, candidate.DeploymentId, candidate.RevisionHash,
                    candidate.ConfigurationJson, candidate.Token,
                    nowUtc, candidate.ExpiresAtUtc, cancellationToken);
            }
            catch (DbUpdateException)
            {
                DetachPendingChanges();
                continue;
            }

            if (staged)
            {
                result[instanceId] = candidate;
                continue;
            }

            var existing = await ReadLivePendingAsync([instanceId], nowUtc, cancellationToken);
            if (existing.TryGetValue(instanceId, out var row))
                result[instanceId] = row;
        }

        return result;
    }

    /// <summary>
    /// Reads the most recent LIVE pending row per instance, projected WITHOUT
    /// <c>ConfigurationJson</c> (the caller only needs the fetch coordinates), untracked.
    /// </summary>
    private async Task<Dictionary<int, PendingDeployment>> ReadLivePendingAsync(
        int[] instanceIds, DateTimeOffset nowUtc, CancellationToken cancellationToken)
    {
        var rows = await _dbContext.Set<PendingDeployment>()
            .AsNoTracking()
            .Where(p => instanceIds.Contains(p.InstanceId) && p.ExpiresAtUtc > nowUtc)
            .Select(p => new
            {
                p.Id, p.DeploymentId, p.InstanceId, p.RevisionHash, p.Token, p.CreatedAtUtc, p.ExpiresAtUtc
            })
            .ToListAsync(cancellationToken);

        // At most one live row per instance by design; on a hypothetical duplicate take the
        // most recent, mirroring GetPendingDeploymentByInstanceIdAsync's ordering.
        return rows
            .GroupBy(r => r.InstanceId)
            .Select(g => g.OrderByDescending(r => r.CreatedAtUtc).ThenByDescending(r => r.Id).First())
            .ToDictionary(
                r => r.InstanceId,
                r => new PendingDeployment(
                    r.DeploymentId, r.InstanceId, r.RevisionHash, string.Empty, r.Token,
                    r.CreatedAtUtc, r.ExpiresAtUtc) { Id = r.Id });
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<Instance>> GetDeployedInstanceSummariesAsync(
        CancellationToken cancellationToken = default)
    {
        return await _dbContext.Set<Instance>()
            .AsNoTracking()
            .Where(i => i.State != InstanceState.NotDeployed)
            .Select(i => new Instance(i.UniqueName)
            {
                Id = i.Id,
                SiteId = i.SiteId,
                TemplateId = i.TemplateId,
                State = i.State
            })
            .ToListAsync(cancellationToken);
    }

    // --- Instance lookups for deployment pipeline ---

    /// <inheritdoc />
    public async Task<Instance?> GetInstanceByIdAsync(int instanceId, CancellationToken cancellationToken = default)
    {
        return await _dbContext.Set<Instance>()
            .Include(i => i.AttributeOverrides)
            .Include(i => i.AlarmOverrides)
            .Include(i => i.ConnectionBindings)
            .Include(i => i.NativeAlarmSourceOverrides)
            .AsSplitQuery()
            .FirstOrDefaultAsync(i => i.Id == instanceId, cancellationToken);
    }

    /// <inheritdoc />
    public async Task<Instance?> GetInstanceByUniqueNameAsync(string uniqueName, CancellationToken cancellationToken = default)
    {
        return await _dbContext.Set<Instance>()
            .Include(i => i.AttributeOverrides)
            .Include(i => i.AlarmOverrides)
            .Include(i => i.ConnectionBindings)
            .Include(i => i.NativeAlarmSourceOverrides)
            .AsSplitQuery()
            .FirstOrDefaultAsync(i => i.UniqueName == uniqueName, cancellationToken);
    }

    /// <inheritdoc />
    public Task UpdateInstanceAsync(Instance instance, CancellationToken cancellationToken = default)
    {
        _dbContext.Set<Instance>().Update(instance);
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public async Task DeleteInstanceAsync(int instanceId, CancellationToken cancellationToken = default)
    {
        // DeploymentRecords have a Restrict FK to Instance — remove them
        // explicitly first. The snapshot, overrides, and connection bindings
        // are configured with cascade delete and go with the instance.
        var records = await _dbContext.DeploymentRecords
            .Where(d => d.InstanceId == instanceId)
            .ToListAsync(cancellationToken);
        if (records.Count > 0)
        {
            _dbContext.DeploymentRecords.RemoveRange(records);
        }

        var instance = await _dbContext.Set<Instance>()
            .FirstOrDefaultAsync(i => i.Id == instanceId, cancellationToken);
        if (instance != null)
        {
            _dbContext.Set<Instance>().Remove(instance);
        }
    }

    /// <inheritdoc />
    public async Task<int> SaveChangesAsync(CancellationToken cancellationToken = default)
    {
        return await _dbContext.SaveChangesAsync(cancellationToken);
    }

    /// <inheritdoc />
    public int DetachPendingChanges()
    {
        // ChangeTracker.Entries() is enumerated into a list FIRST: setting State
        // mutates the tracker, and mutating it while enumerating it throws.
        // Unchanged/Detached entries are left alone — they carry no pending write,
        // so keeping them tracked preserves identity resolution for anything the
        // caller still holds a reference to.
        var pending = _dbContext.ChangeTracker.Entries()
            .Where(e => e.State is EntityState.Added or EntityState.Modified or EntityState.Deleted)
            .ToList();

        foreach (var entry in pending)
            entry.State = EntityState.Detached;

        return pending.Count;
    }

    /// <inheritdoc />
    public int DetachSettledDeploymentRows(int instanceId)
    {
        // Enumerated into a list first: changing State mutates the tracker.
        var settled = _dbContext.ChangeTracker.Entries()
            .Where(e => e.State == EntityState.Unchanged
                        && e.Entity switch
                        {
                            PendingDeployment p => p.InstanceId == instanceId,
                            DeployedConfigSnapshot s => s.InstanceId == instanceId,
                            DeploymentRecord r => r.InstanceId == instanceId,
                            _ => false
                        })
            .ToList();

        foreach (var entry in settled)
            entry.State = EntityState.Detached;

        return settled.Count;
    }

    /// <summary>
    /// Flat shape of the deployment-record ⋈ instance join used to compose the list
    /// query. A named type rather than an anonymous one so
    /// <see cref="BuildListQuery"/> can hand the composed-but-unexecuted query back
    /// to both the paging read and the count aggregation, keeping the two in
    /// lockstep. Not a public contract — <see cref="DeploymentListRow"/> is what
    /// leaves this class.
    /// </summary>
    internal sealed class DeploymentInstanceJoin
    {
        /// <summary>Deployment record primary key.</summary>
        public int Id { get; init; }
        /// <summary>Unique deployment identifier.</summary>
        public string DeploymentId { get; init; } = string.Empty;
        /// <summary>Primary key of the target instance.</summary>
        public int InstanceId { get; init; }
        /// <summary>Unique name of the target instance.</summary>
        public string InstanceUniqueName { get; init; } = string.Empty;
        /// <summary>Primary key of the instance's site.</summary>
        public int SiteId { get; init; }
        /// <summary>Current deployment status.</summary>
        public DeploymentStatus Status { get; init; }
        /// <summary>Revision hash of the deployed configuration, if any.</summary>
        public string? RevisionHash { get; init; }
        /// <summary>User who initiated the deployment.</summary>
        public string DeployedBy { get; init; } = string.Empty;
        /// <summary>When the deployment was initiated.</summary>
        public DateTimeOffset DeployedAt { get; init; }
        /// <summary>When the deployment completed, if it has.</summary>
        public DateTimeOffset? CompletedAt { get; init; }
        /// <summary>Failure message, if the deployment failed.</summary>
        public string? ErrorMessage { get; init; }
    }
}