lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Deployments/DeploymentDriftProbe.cs

using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;

namespace ZB.MOM.WW.OtOpcUa.AdminUI.Deployments;

/// <summary>
///     Answers the /deployments page's "Configuration drift" chip without recomposing the whole
///     configuration on every page load.
/// </summary>
/// <remarks>
///     <para>
///         <b>Why.</b> The chip's question — "does the live-edit state still hash to the latest sealed
///         revision?" — was answered by calling <c>ConfigComposer.SnapshotAndFlattenAsync</c> on every
///         <c>ReloadAsync</c>: eighteen unfiltered table reads, a deterministic JSON serialisation of the
///         entire fleet configuration, a gzip pass and a SHA-256, all to render one badge. Opening
///         /deployments, deploying, or simply re-rendering the page paid for it again.
///     </para>
///     <para>
///         <b>The watermark.</b> <c>ConfigEdit</c> is the append-only audit row that every mutating
///         live-edit operation writes (see the entity's remarks — <c>AdminOperationsActor</c> appends one
///         per mutation), so <c>(MAX(EditedAtUtc), COUNT(*))</c> moves whenever the configuration could
///         have changed. That pair is one indexed aggregate (<c>IX_ConfigEdit_EditedAt</c>) against a
///         full recomposition. When neither the watermark nor the sealed revision hash has moved since
///         the last verdict, the cached verdict is returned unchanged.
///     </para>
///     <para>
///         <b>Accepted staleness.</b> An edit path that mutated a configuration table WITHOUT writing a
///         <c>ConfigEdit</c> row would not invalidate this cache, and the chip could then read "In sync"
///         over a genuinely drifted configuration until the next audited edit. That is acceptable
///         precisely because the chip is advisory: the authoritative comparison is
///         <c>AdminOperationsActor</c>'s own recomposition at deploy time, which always reads live and
///         which reports <c>NoChanges</c> for a deploy that would be a no-op. Today every
///         <c>AdminOperationsActor</c> mutation writes the audit row, so the gap is theoretical — but a
///         future write path that skips the audit must not be relied on to move this chip. The
///         <c>COUNT(*)</c> half is what keeps a same-timestamp edit (two mutations inside one tick, or
///         an edit deleted and replaced) from hiding behind an unchanged maximum.
///     </para>
///     <para>
///         <b>Concurrency.</b> The cheap watermark read runs outside the gate, so concurrent circuits do
///         not serialise on it. Only the recompose path takes the <see cref="SemaphoreSlim"/>, and it
///         double-checks the cache after acquiring it — so N circuits that miss at the same instant
///         share ONE recomposition rather than launching N. A throw leaves the previous verdict in
///         place and propagates to its caller alone.
///     </para>
/// </remarks>
/// <param name="dbFactory">Factory for the configuration database context.</param>
/// <param name="composeRevisionHash">
///     The compose seam — returns the revision hash of a fresh snapshot of the supplied context.
///     Defaults to the real <c>ConfigComposer.SnapshotAndFlattenAsync(...).RevisionHash</c>; tests
///     inject a counting fake to prove the cache actually elides recompositions.
/// </param>
public sealed class DeploymentDriftProbe(
    IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
    Func<OtOpcUaConfigDbContext, Task<string>>? composeRevisionHash = null)
{
    private readonly Func<OtOpcUaConfigDbContext, Task<string>> _composeRevisionHash =
        composeRevisionHash ?? ComposeRevisionHashAsync;

    private readonly SemaphoreSlim _recomposeGate = new(1, 1);

    /// <summary>The last computed verdict, or <see langword="null"/> before the first one. Volatile: the
    /// fast path reads it off the gate, so the publishing write must be visible to every circuit.</summary>
    private volatile Verdict? _verdict;

    /// <summary>
    ///     Reports whether the live-edit configuration has drifted from the supplied sealed revision.
    /// </summary>
    /// <param name="latestSealedRevisionHash">
    ///     The revision hash of the most recent sealed deployment. Empty means there is nothing sealed to
    ///     compare against.
    /// </param>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>
    ///     <see langword="true"/> when the current configuration hashes to something other than
    ///     <paramref name="latestSealedRevisionHash"/>, <see langword="false"/> when it matches, and
    ///     <see langword="null"/> when no verdict applies (no sealed revision supplied).
    /// </returns>
    public async Task<bool?> GetDriftAsync(string latestSealedRevisionHash, CancellationToken ct = default)
    {
        if (string.IsNullOrEmpty(latestSealedRevisionHash)) return null;

        await using var db = await dbFactory.CreateDbContextAsync(ct);
        var watermark = await ReadWatermarkAsync(db, ct);

        // Fast path: no gate, because a hit reads one immutable record reference.
        if (IsCurrent(_verdict, watermark, latestSealedRevisionHash, out var cached)) return cached;

        await _recomposeGate.WaitAsync(ct);
        try
        {
            // Double-check: a caller ahead of us in the queue may have just recomposed the same state.
            if (IsCurrent(_verdict, watermark, latestSealedRevisionHash, out cached)) return cached;

            var current = await _composeRevisionHash(db);
            var drift = !string.Equals(current, latestSealedRevisionHash, StringComparison.Ordinal);
            _verdict = new Verdict(watermark, latestSealedRevisionHash, drift);
            return drift;
        }
        finally
        {
            _recomposeGate.Release();
        }
    }

    private static bool IsCurrent(
        Verdict? verdict, Watermark watermark, string sealedHash, out bool drift)
    {
        if (verdict is not null
            && verdict.Watermark == watermark
            && string.Equals(verdict.SealedRevisionHash, sealedHash, StringComparison.Ordinal))
        {
            drift = verdict.Drift;
            return true;
        }

        drift = false;
        return false;
    }

    /// <summary>Reads the audit watermark as a single aggregate row (no rows at all ⇒ the empty watermark).</summary>
    private static async Task<Watermark> ReadWatermarkAsync(OtOpcUaConfigDbContext db, CancellationToken ct)
    {
        // GroupBy(_ => 1) collapses the whole table to one aggregate row, so this is ONE round-trip for
        // both scalars rather than two. An empty table produces no group at all, so FirstOrDefaultAsync
        // lands on null and we return the default (never-edited) watermark; the Max is cast to nullable
        // so the projection stays valid for the no-rows shape either way.
        var row = await db.ConfigEdits
            .AsNoTracking()
            .GroupBy(_ => 1)
            .Select(g => new { LastEditUtc = g.Max(e => (DateTime?)e.EditedAtUtc), EditCount = g.Count() })
            .FirstOrDefaultAsync(ct);

        return row is null ? default : new Watermark(row.LastEditUtc, row.EditCount);
    }

    private static async Task<string> ComposeRevisionHashAsync(OtOpcUaConfigDbContext db) =>
        (await ConfigComposer.SnapshotAndFlattenAsync(db)).RevisionHash;

    /// <summary>The ConfigEdit audit watermark: newest edit timestamp plus the total edit count.</summary>
    private readonly record struct Watermark(DateTime? LastEditUtc, int EditCount);

    /// <summary>A drift answer, valid only for the watermark and sealed revision it was computed against.</summary>
    private sealed record Verdict(Watermark Watermark, string SealedRevisionHash, bool Drift);
}