ScadaBridge/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Secrets/DeployedSecretsMirror.cs

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.Secrets.Abstractions;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Secrets;

/// <summary>
/// The outcome of one <see cref="DeployedSecretsMirror.ReconcileAsync"/> pass: the secret
/// names applied (inserted or updated in this node's own secret store), the names removed
/// (deleted because central tombstoned them and this node had previously mirrored them), and
/// a human-readable failure message per live secret that could not be applied and/or verified
/// on this node.
/// </summary>
/// <param name="Applied">Names applied (inserted or updated) this pass.</param>
/// <param name="Removed">Names deleted this pass because they were previously mirrored and are now tombstoned upstream.</param>
/// <param name="Failed">One message per live secret that failed to apply and/or decrypt on this node — never includes secret values.</param>
public sealed record MirrorResult(
    IReadOnlyList<string> Applied, IReadOnlyList<string> Removed, IReadOnlyList<string> Failed)
{
    /// <summary>Whether this pass applied or removed anything (used to decide the summary log level).</summary>
    public bool Changed => Applied.Count > 0 || Removed.Count > 0;
}

/// <summary>
/// Per-node reconcile of the site-wide, LocalDb-replicated <c>deployed_secrets</c> table into
/// THIS node's own <c>ZB.MOM.WW.Secrets</c> store. Design §3.5 step 2.
/// </summary>
/// <remarks>
/// <para>
/// Runs on EVERY site node — not a singleton — because <c>deployed_secrets</c> row content
/// (ciphertext) is site-wide (replicated by LocalDb), but which names THIS node has actually
/// applied into its own local secret store is node-local fact: <c>mirrored_secrets</c> is
/// deliberately NOT one of LocalDb's replicated tables (see
/// <see cref="SiteStorageService.GetMirroredSecretsAsync"/>). Each node reconciles for
/// itself, and — because the two nodes' secret stores are otherwise independent — each node
/// also independently verifies that the deployed ciphertext actually decrypts under ITS OWN
/// master key.
/// </para>
/// <para>
/// <b>Apply semantics and the confirm-then-ledger fix (fix round 1, review Important #1).</b>
/// <see cref="ISecretStore.ApplyReplicatedAsync"/> applies last-writer-wins strictly on
/// (<c>UpdatedUtc</c>, <c>Revision</c>) — see
/// <see cref="ZB.MOM.WW.Secrets.Abstractions.SecretLastWriterWins"/> — and a row that is not
/// strictly newer than the local row is silently ignored, so this mirror never trusts the call
/// blindly: every apply is followed by an <see cref="ISecretStore.GetAsync"/> read-back, and
/// <c>mirrored_secrets</c> is ledgered only when the local row is confirmed LIVE and its
/// ciphertext (<c>Ciphertext</c> + <c>WrappedDek</c>) matches what was just shipped. Two
/// non-ledgered outcomes are possible, both reported as a <see cref="MirrorResult.Failed"/>
/// entry (never as the decrypt-failure message, which would misdiagnose either case) and never
/// overwritten a second way:
/// <list type="bullet">
/// <item>
/// <b>The local row is a tombstone</b> (this node previously deleted it — e.g. the site
/// stopped referencing it, then referenced it again unchanged at central, so the shipped row's
/// <c>UpdatedUtc</c> is older than the tombstone's). Re-stamping is safe here: neither
/// <c>UpdatedUtc</c> nor <c>Revision</c> is bound into the AEAD associated data — confirmed
/// against <c>ZB.MOM.WW.Secrets.Crypto.AesGcmEnvelopeCipher</c>, whose body layer binds only
/// <c>SecretName.Value</c> and whose wrap layer binds only <c>KekId</c> — so the mirror
/// re-applies the exact shipped ciphertext with <c>UpdatedUtc = max(now, tombstone.UpdatedUtc +
/// 1 tick)</c> and <c>Revision = tombstone.Revision + 1</c>, which beats the tombstone under LWW
/// without touching anything the cipher verifies. If that confirmed re-apply still does not
/// read back live and matching (a concurrent conflict), the row is reported failed instead of
/// silently left broken.
/// </item>
/// <item>
/// <b>The local row is LIVE with different ciphertext</b> — an out-of-band secret created
/// directly on this node under the same name, stamped strictly newer than what central shipped.
/// It wins LWW legitimately. This mirror does NOT overwrite it, does NOT ledger it (so a later
/// central tombstone of the name can never delete someone's out-of-band secret via this
/// mirror), and does NOT run the decrypt-verify against it (verifying would trivially "succeed"
/// against the operator's own value and mask that the deployed revision never took effect).
/// </item>
/// </list>
/// Ledgering always uses the SHIPPED artifact's own revision (not any bumped local revision)
/// so a later pass's "is this ledgered at the shipped revision" check stays correct.
/// </para>
/// <para>
/// <b>Per-row isolation (review Important #2).</b> Each row's apply/verify body runs in its own
/// try/catch; an exception — an unrecognised <c>ContentType</c> from a newer central, a
/// per-row store fault, a malformed name — is caught, reported as a <see cref="MirrorResult.Failed"/>
/// entry naming the secret and the exception type/message (never a value), logged with the same
/// de-spam rule, and does not stop the rest of the pass. Only an exception attributable to the
/// caller's own cancellation (<c>ct.IsCancellationRequested</c>) propagates.
/// </para>
/// <para>
/// <b>Cache invalidation (review Important #3).</b> The concrete resolver
/// (<c>DefaultSecretResolver</c>) caches decrypted plaintext for a TTL and also implements
/// <see cref="ISecretCacheInvalidator"/>; every ZB.MOM.WW.Secrets write path invalidates after a
/// write for exactly this reason. This mirror does the same: after every successful apply
/// (including a tombstone re-materialization) or delete, and before the verify step, it calls
/// <see cref="ISecretCacheInvalidator.Invalidate"/> (injected optionally — <c>null</c> if the
/// host never registered one) so neither the verify nor a script running immediately afterward
/// can observe a stale pre-rotation plaintext for up to the cache TTL.
/// </para>
/// <para>
/// <b>Node-local store assumption (review Minor #5).</b> This mirror assumes the site's
/// <see cref="ISecretStore"/> is node-local (plain local SQLite, or the gRPC-hub follower half,
/// both of which keep writes local). It is NOT safe to run against a site node whose store is
/// decorated by SQL-Server secrets REPLICATION (<c>Secrets:Replication:Mode=SqlServer</c>) — its
/// <c>DeleteAsync</c> publishes the tombstone to the shared hub, so a site merely un-referencing
/// a secret would delete it fleet-wide. Task 11 rejects that configuration on the Site role;
/// no code change is made here.
/// </para>
/// <para>
/// <b>Resolve semantics and the resolver cache.</b> The trial-resolve verify step goes through
/// <see cref="ISecretResolver.GetAsync"/>, which the concrete resolver caches for a TTL
/// (<c>ResolveCacheTtl</c>, 30 s in the shipped implementation). Because every successful apply
/// or delete now invalidates that cache first (see above), the remaining exposure is narrow: a
/// KEK rotation on this node's OWN master key (not a secret rotation at central) landing inside
/// the cache window of an unrelated resolve. That residual case is still an acceptable false
/// negative for the "did the key change" question — the cache hit still proves this node once
/// decrypted a value with a key that produced that cache entry — and is not a bug in this
/// mirror.
/// </para>
/// <para>
/// <b>Error-log de-spam.</b> A secret that keeps failing (to apply OR to decrypt) would
/// otherwise log an Error line every reconcile pass forever. This mirror logs Error only the
/// FIRST time a given failure fingerprint (kind + revision + relevant detail, e.g. KekId for a
/// decrypt failure) occurs for a name on this node; identical repeats log at Debug instead. The
/// moment that same name applies/resolves successfully, the remembered fingerprint is cleared,
/// so a LATER failure — even one with an identical fingerprint — logs at Error again. This state
/// is in-memory only (per node process), which matches the ledger's own node-local scope.
/// </para>
/// </remarks>
public sealed class DeployedSecretsMirror : BackgroundService
{
    private const string DeleteActor = "deployed-secrets-mirror";

    private readonly SiteStorageService _storage;
    private readonly ISecretStore _store;
    private readonly ISecretResolver _resolver;
    private readonly ISecretCacheInvalidator? _invalidator;
    private readonly IOptions<SiteRuntimeOptions> _options;
    private readonly ILogger<DeployedSecretsMirror> _logger;
    private readonly TimeProvider _time;

    private readonly SemaphoreSlim _gate = new(1, 1);
    private bool _disposed;

    /// <summary>
    /// Per-secret-name record of the failure fingerprint this node has already logged at Error
    /// (an apply-conflict or a decrypt failure — see the class remarks). Cleared for a name the
    /// moment it applies/resolves successfully, so a later failure — even an identical one —
    /// logs at Error again. In-memory only; not a durable ledger, and deliberately so (it exists
    /// purely to de-spam this process's logs).
    /// </summary>
    private readonly Dictionary<string, string> _lastLoggedFailure = new(StringComparer.Ordinal);

    /// <summary>Initializes a new instance of <see cref="DeployedSecretsMirror"/>.</summary>
    /// <param name="storage">Site-local storage exposing <c>deployed_secrets</c> and the <c>mirrored_secrets</c> ledger.</param>
    /// <param name="store">This node's own secret store.</param>
    /// <param name="resolver">This node's own secret resolver, used to trial-decrypt each deployed secret.</param>
    /// <param name="invalidator">
    /// Optional cache invalidator for <paramref name="resolver"/> (the concrete
    /// <c>DefaultSecretResolver</c> implements this itself; <c>null</c> if the host never
    /// registered one, in which case no invalidation is attempted).
    /// </param>
    /// <param name="options">Supplies <see cref="SiteRuntimeOptions.DeployedSecretsMirrorInterval"/>.</param>
    /// <param name="logger">Logger instance. Never logs secret values — names, revisions, and KEK ids only.</param>
    /// <param name="time">Clock driving the reconcile loop's <see cref="PeriodicTimer"/> and the tombstone-recovery stamp.</param>
    public DeployedSecretsMirror(
        SiteStorageService storage,
        ISecretStore store,
        ISecretResolver resolver,
        ISecretCacheInvalidator? invalidator,
        IOptions<SiteRuntimeOptions> options,
        ILogger<DeployedSecretsMirror> logger,
        TimeProvider time)
    {
        _storage = storage ?? throw new ArgumentNullException(nameof(storage));
        _store = store ?? throw new ArgumentNullException(nameof(store));
        _resolver = resolver ?? throw new ArgumentNullException(nameof(resolver));
        _invalidator = invalidator;
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _time = time ?? throw new ArgumentNullException(nameof(time));
    }

    /// <summary>
    /// Runs one reconcile pass: applies every live <c>deployed_secrets</c> row into this
    /// node's own secret store (confirmed by read-back before ledgering), deletes rows this
    /// node previously mirrored that are now tombstoned upstream, and trial-decrypts every
    /// live, confirmed-applied row to verify it resolves under this node's own master key.
    /// Safe to call concurrently with itself (and with the hosted loop) — a single in-flight
    /// pass is enforced internally.
    /// </summary>
    /// <param name="ct">A token to cancel the operation.</param>
    /// <returns>The pass's <see cref="MirrorResult"/>.</returns>
    public async Task<MirrorResult> ReconcileAsync(CancellationToken ct)
    {
        await _gate.WaitAsync(ct).ConfigureAwait(false);
        try
        {
            var rows = await _storage.GetDeployedSecretsAsync().ConfigureAwait(false);
            var mirrored = (await _storage.GetMirroredSecretsAsync().ConfigureAwait(false))
                .ToDictionary(m => m.Name, m => m.Revision, StringComparer.Ordinal);

            List<string> applied = [];
            List<string> removed = [];
            List<string> failed = [];

            foreach (var row in rows)
            {
                try
                {
                    if (row.IsRemoved)
                    {
                        await ProcessRemovedRowAsync(row, mirrored, removed, ct).ConfigureAwait(false);
                    }
                    else
                    {
                        await ProcessLiveRowAsync(row, mirrored, applied, failed, ct).ConfigureAwait(false);
                    }
                }
                catch (Exception ex) when (!ct.IsCancellationRequested)
                {
                    ReportFailure(
                        row.Name,
                        $"error:{row.Secret.Revision}:{ex.GetType().Name}:{ex.Message}",
                        $"Secret '{row.Name}' failed to process on this node ({ex.GetType().Name}: {ex.Message})",
                        failed);
                }
            }

            var result = new MirrorResult(applied, removed, failed);

            if (result.Changed)
            {
                _logger.LogInformation(
                    "Deployed secrets mirror: {Applied} applied, {Removed} removed, {Failed} failed",
                    applied.Count, removed.Count, failed.Count);
            }
            else
            {
                _logger.LogDebug(
                    "Deployed secrets mirror in sync ({Count} secret(s), {FailedCount} failing)",
                    rows.Count(r => !r.IsRemoved), failed.Count);
            }

            return result;
        }
        finally
        {
            _gate.Release();
        }
    }

    /// <summary>
    /// Handles a tombstoned <c>deployed_secrets</c> row: deletes it from this node's store, but
    /// only if this node had previously ledgered it as mirrored — an out-of-band, same-named
    /// local secret this mirror never applied is left alone.
    /// </summary>
    private async Task ProcessRemovedRowAsync(
        DeployedSecretRow row, Dictionary<string, long> mirrored, List<string> removed, CancellationToken ct)
    {
        if (!mirrored.ContainsKey(row.Name))
        {
            return;
        }

        await _store.DeleteAsync(new SecretName(row.Name), DeleteActor, ct).ConfigureAwait(false);
        _invalidator?.Invalidate(new SecretName(row.Name));
        await _storage.RemoveMirroredSecretAsync(row.Name).ConfigureAwait(false);
        removed.Add(row.Name);
        _lastLoggedFailure.Remove(row.Name);
        _logger.LogInformation(
            "Removed deployed secret {SecretName} (no longer referenced by this site)",
            row.Name);
    }

    /// <summary>
    /// Handles a live <c>deployed_secrets</c> row: applies it (confirmed by read-back) when not
    /// already ledgered at the shipped revision, then verifies it decrypts — unless the apply
    /// outcome makes a verify meaningless (an out-of-band local secret won LWW), in which case a
    /// single, distinct failure is reported instead.
    /// </summary>
    private async Task ProcessLiveRowAsync(
        DeployedSecretRow row,
        Dictionary<string, long> mirrored,
        List<string> applied,
        List<string> failed,
        CancellationToken ct)
    {
        var name = new SecretName(row.Name);
        var alreadyLedgered =
            mirrored.TryGetValue(row.Name, out var mirroredRevision) && mirroredRevision == row.Secret.Revision;

        if (!alreadyLedgered)
        {
            var outcome = await ApplyAndConfirmAsync(row, name, ct).ConfigureAwait(false);

            switch (outcome)
            {
                case ApplyOutcome.Applied:
                    await _storage.SetMirroredSecretAsync(row.Name, row.Secret.Revision).ConfigureAwait(false);
                    _invalidator?.Invalidate(name);
                    applied.Add(row.Name);
                    _lastLoggedFailure.Remove(row.Name);
                    _logger.LogInformation(
                        "Applied deployed secret {SecretName} r{Revision} (kek {KekId})",
                        row.Name, row.Secret.Revision, row.Secret.KekId);
                    break;

                case ApplyOutcome.LocalNewer:
                    ReportFailure(
                        row.Name,
                        $"outofband:{row.Secret.Revision}",
                        $"Secret '{row.Name}' exists on this node with a newer local value; " +
                            "the deployed revision was not applied",
                        failed);
                    return; // Do not verify — that would decrypt the operator's own value.

                case ApplyOutcome.Unconfirmed:
                    ReportFailure(
                        row.Name,
                        $"unconfirmed:{row.Secret.Revision}",
                        $"Secret '{row.Name}' could not be confirmed applied on this node " +
                            "after a tombstone re-apply attempt",
                        failed);
                    return;
            }
        }

        await VerifyAsync(row, failed, ct).ConfigureAwait(false);
    }

    /// <summary>The confirmed outcome of applying one shipped row into this node's store. See the class remarks.</summary>
    private enum ApplyOutcome
    {
        /// <summary>The local row is now live and its ciphertext matches what was shipped.</summary>
        Applied,

        /// <summary>The local row is live with DIFFERENT ciphertext — an out-of-band secret that legitimately won LWW.</summary>
        LocalNewer,

        /// <summary>Neither of the above — a tombstone re-apply attempt did not read back as applied.</summary>
        Unconfirmed,
    }

    /// <summary>
    /// Calls <see cref="ISecretStore.ApplyReplicatedAsync"/> and confirms the outcome by reading
    /// the row back — see the class remarks for why this mirror never trusts the call blindly.
    /// </summary>
    private async Task<ApplyOutcome> ApplyAndConfirmAsync(DeployedSecretRow row, SecretName name, CancellationToken ct)
    {
        var shipped = ToStored(row.Secret);
        await _store.ApplyReplicatedAsync(shipped, ct).ConfigureAwait(false);

        var local = await _store.GetAsync(name, ct).ConfigureAwait(false);

        if (local is not null && !local.IsDeleted && MatchesShippedCiphertext(local, shipped))
        {
            return ApplyOutcome.Applied;
        }

        if (local is not null && local.IsDeleted)
        {
            // Neither UpdatedUtc nor Revision is bound into the AEAD associated data (confirmed
            // against AesGcmEnvelopeCipher — only SecretName and KekId are AAD), so re-stamping
            // strictly past the tombstone is safe: it changes nothing the cipher verifies.
            var nowUtc = _time.GetUtcNow();
            var tombstoneUtcPlusOneTick = local.UpdatedUtc.AddTicks(1);
            var bumpedUpdatedUtc = nowUtc > tombstoneUtcPlusOneTick ? nowUtc : tombstoneUtcPlusOneTick;
            var bumped = shipped with { UpdatedUtc = bumpedUpdatedUtc, Revision = local.Revision + 1 };

            await _store.ApplyReplicatedAsync(bumped, ct).ConfigureAwait(false);
            var reconfirmed = await _store.GetAsync(name, ct).ConfigureAwait(false);

            return reconfirmed is not null && !reconfirmed.IsDeleted && MatchesShippedCiphertext(reconfirmed, shipped)
                ? ApplyOutcome.Applied
                : ApplyOutcome.Unconfirmed;
        }

        // local is live but its ciphertext differs from what was shipped: an out-of-band value
        // with a strictly newer (UpdatedUtc, Revision) won LWW. (local is null only in a
        // pathological concurrent-delete race; Unconfirmed covers that too.)
        return local is null ? ApplyOutcome.Unconfirmed : ApplyOutcome.LocalNewer;
    }

    private static bool MatchesShippedCiphertext(StoredSecret local, StoredSecret shipped) =>
        local.Ciphertext.AsSpan().SequenceEqual(shipped.Ciphertext)
        && local.WrappedDek.AsSpan().SequenceEqual(shipped.WrappedDek);

    /// <summary>
    /// Trial-resolves <paramref name="row"/>'s secret through <see cref="ISecretResolver"/> to
    /// verify it decrypts on this node. On failure, appends a message to
    /// <paramref name="failed"/> and logs with the shared de-spam rule (see the class remarks).
    /// </summary>
    private async Task VerifyAsync(DeployedSecretRow row, List<string> failed, CancellationToken ct)
    {
        bool resolved;
        try
        {
            resolved = await _resolver.GetAsync(new SecretName(row.Name), ct).ConfigureAwait(false) is not null;
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            throw;
        }
        catch (Exception)
        {
            resolved = false;
        }

        if (resolved)
        {
            _lastLoggedFailure.Remove(row.Name);
            return;
        }

        var message =
            $"Secret '{row.Name}' (kek {row.Secret.KekId}) did not decrypt on this node — " +
            "the node's master key differs from central's";
        ReportFailure(row.Name, $"decrypt:{row.Secret.Revision}:{row.Secret.KekId}", message, failed);
    }

    /// <summary>
    /// Appends <paramref name="message"/> to <paramref name="failed"/> and logs it — Error the
    /// first time <paramref name="fingerprint"/> is seen for <paramref name="name"/> since it
    /// last succeeded, Debug on every identical repeat.
    /// </summary>
    private void ReportFailure(string name, string fingerprint, string message, List<string> failed)
    {
        failed.Add(message);

        if (_lastLoggedFailure.TryGetValue(name, out var previous) && previous == fingerprint)
        {
            _logger.LogDebug("{Failure}", message);
        }
        else
        {
            _lastLoggedFailure[name] = fingerprint;
            _logger.LogError("{Failure}", message);
        }
    }

    /// <inheritdoc />
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using var timer = new PeriodicTimer(_options.Value.DeployedSecretsMirrorInterval, _time);
        do
        {
            try
            {
                await ReconcileAsync(stoppingToken).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                // Covers every other failure, including an OperationCanceledException NOT
                // attributable to this service's own shutdown token (e.g. a store/command
                // timeout) — that must not silently end the loop (review Minor #4).
                _logger.LogWarning(ex, "Deployed secrets mirror pass failed; retrying next interval.");
            }
        } while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false));
    }

    /// <summary>Maps a wire <see cref="SecretArtifact"/> row back to the store's <see cref="StoredSecret"/> shape.</summary>
    private static StoredSecret ToStored(SecretArtifact secret) => new()
    {
        Name = new SecretName(secret.Name),
        Description = secret.Description,
        ContentType = Enum.Parse<SecretContentType>(secret.ContentType, ignoreCase: false),
        Ciphertext = secret.Ciphertext,
        Nonce = secret.Nonce,
        Tag = secret.Tag,
        WrappedDek = secret.WrappedDek,
        WrapNonce = secret.WrapNonce,
        WrapTag = secret.WrapTag,
        KekId = secret.KekId,
        Revision = secret.Revision,
        CreatedUtc = secret.CreatedUtc,
        UpdatedUtc = secret.UpdatedUtc,
    };

    /// <inheritdoc />
    public override void Dispose()
    {
        if (!_disposed)
        {
            _disposed = true;
            _gate.Dispose();
        }

        base.Dispose();
    }
}