using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Deployment; using ZB.MOM.WW.ScadaBridge.Commons.Types.Deployment; namespace ZB.MOM.WW.ScadaBridge.Communication; /// /// Central-side startup-reconciliation handler. A site node, on startup, reports its /// local deployed inventory via (delivered over /// ClusterClient to ); this service diffs /// it against central's expected deployed set and replies with fresh fetch tokens for the /// gap — instances the node is missing or has at a stale revision — plus the orphan names /// (present locally but no longer deployed centrally, which the node only logs). /// /// This self-heals a node that was DOWN during a deploy and therefore missed the /// replicate. The node fetches each gap item's config over the existing token-gated HTTP /// endpoint; this service stages those tokens as PendingDeployment rows. /// /// Scoped service (holds scoped repositories); resolved per-request by the actor inside a /// DI scope. The actor stays thin: it captures Sender, resolves this service in a /// scope, awaits , and pipes the response back. /// public class ReconcileService { private readonly IDeploymentManagerRepository _deploymentRepository; private readonly ISiteRepository _siteRepository; private readonly CommunicationOptions _options; private readonly ILogger _logger; /// Initializes the reconciliation service. /// Repository for the expected-set query and pending-row staging. /// Repository used to resolve a site's numeric id from its identifier. /// Communication options carrying and . /// Logger. public ReconcileService( IDeploymentManagerRepository deploymentRepository, ISiteRepository siteRepository, IOptions options, ILogger logger) { _deploymentRepository = deploymentRepository; _siteRepository = siteRepository; _options = options.Value; _logger = logger; } /// /// Diffs the node's reported inventory against central's expected deployed set, /// stages fresh fetch tokens for the gap, and returns the reconcile response. /// /// The site node's reconcile request (its local name→revision-hash map). /// A cancellation token. /// The gap (with fresh tokens), orphan names, and the central fetch base URL. public async Task ReconcileAsync( ReconcileSiteRequest request, CancellationToken cancellationToken = default) { var baseUrl = _options.CentralFetchBaseUrl; // 1. Resolve the numeric site id. An unknown site is non-fatal: reply empty // (the node simply finds no gap to fetch) and log a warning so a // misconfigured SiteIdentifier is visible to operators. var site = await _siteRepository .GetSiteByIdentifierAsync(request.SiteIdentifier, cancellationToken) .ConfigureAwait(false); if (site == null) { _logger.LogWarning( "Reconcile request from unknown site '{SiteIdentifier}' (node {NodeId}); replying with empty gap", request.SiteIdentifier, request.NodeId); return new ReconcileSiteResponse( Array.Empty(), Array.Empty(), baseUrl); } // 2. Central's expected deployed set for this site (instances with a snapshot). var expected = await _deploymentRepository .GetExpectedDeploymentsForSiteAsync(site.Id, cancellationToken) .ConfigureAwait(false); var localMap = request.LocalNameToRevisionHash; var gap = new List(); var now = DateTimeOffset.UtcNow; var expiresAt = now + _options.PendingDeploymentTtl; // 3. GAP = expected items the node is MISSING (name absent locally) or STALE // (local revision != expected revision). Current items are omitted. foreach (var exp in expected) { var present = localMap.TryGetValue(exp.InstanceUniqueName, out var localHash); var stale = present && !string.Equals(localHash, exp.RevisionHash, StringComparison.Ordinal); if (present && !stale) continue; // node already has the current revision // 4. Read the frozen snapshot config to stage. Null = the snapshot was deleted // between the expected-set query and now (instance removed mid-reconcile); // skip it — best-effort reconcile re-runs on the next node startup. var snapshot = await _deploymentRepository .GetDeployedSnapshotByInstanceIdAsync(exp.InstanceId, cancellationToken) .ConfigureAwait(false); if (snapshot == null) { _logger.LogDebug( "Reconcile: snapshot for instance {Instance} disappeared (deleted race); skipping", exp.InstanceUniqueName); continue; } var token = DeploymentFetchToken.Generate(); // Stage with the snapshot's DeploymentId as the deploymentId so the gap item's // DeploymentId + token point the node at the right pending row to fetch. // // StagePendingIfAbsent is insert-if-absent: if BOTH site nodes are concurrently // missing the same instance (e.g. fresh container start / cleared SQLite after a // successful deploy), both attempt to stage here. The first succeeds (true); the // second gets false and is handled in the !staged branch below — it returns the // existing pending row's token so it heals in the same round, rather than being // omitted. Deploy-time supersession serializes via the per-instance operation lock. var staged = await _deploymentRepository.StagePendingIfAbsentAsync( exp.InstanceId, snapshot.DeploymentId, exp.RevisionHash, snapshot.ConfigurationJson, token, now, expiresAt, cancellationToken) .ConfigureAwait(false); if (!staged) { // A pending row already exists for this instance — either a CONCURRENT reconcile // from the other site node (both nodes' SQLite empty after a fresh/cleared deploy, // both reconciling at startup) or an in-flight deploy. Do NOT omit the item: if we // did, the second concurrently-missing node would get 0 fetched and stay unhealed // until a later restart. Instead, read the EXISTING pending row and emit a gap item // carrying ITS DeploymentId/RevisionHash/Token. The fetch token is multi-use within // its TTL, so both nodes fetch the same pending config and heal in the same round. // (If the existing row is from an in-flight deploy its config is newer than the // snapshot — fetching it is still correct; the site's guarded write handles ordering.) var existing = await _deploymentRepository .GetPendingDeploymentByInstanceIdAsync(exp.InstanceId, now, cancellationToken) .ConfigureAwait(false); if (existing != null) { _logger.LogDebug( "Reconcile: pending row already exists for instance {Instance} (concurrent reconcile or in-flight deploy); returning existing token so this node heals too", exp.InstanceUniqueName); gap.Add(new ReconcileGapItem( exp.InstanceUniqueName, existing.DeploymentId, existing.RevisionHash, exp.IsEnabled, existing.Token)); } else { // Raced away: the pending row was purged between the stage attempt and this // read. Omit it — reconcile is best-effort and the node retries next round. _logger.LogDebug( "Reconcile: pending row for instance {Instance} disappeared between stage and read (purged race); omitting from gap", exp.InstanceUniqueName); } continue; } gap.Add(new ReconcileGapItem( exp.InstanceUniqueName, snapshot.DeploymentId, exp.RevisionHash, exp.IsEnabled, token)); } // 5. ORPHANS = names the node has locally that central no longer considers deployed. // The node only LOGS these (never deletes). var expectedNames = new HashSet( expected.Select(e => e.InstanceUniqueName), StringComparer.Ordinal); var orphans = localMap.Keys .Where(name => !expectedNames.Contains(name)) .ToList(); _logger.LogDebug( "Reconcile for site {SiteIdentifier} (node {NodeId}): {GapCount} gap, {OrphanCount} orphan(s)", request.SiteIdentifier, request.NodeId, gap.Count, orphans.Count); // 6. Reply. return new ReconcileSiteResponse(gap, orphans, baseUrl); } }