lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/LocalDbDeploymentArtifactCache.cs

using System.Globalization;
using System.Security.Cryptography;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.LocalDb;

namespace ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;

/// <summary>
///     <see cref="IDeploymentArtifactCache"/> backed by the replicated LocalDb deployment-cache
///     tables.
/// </summary>
/// <remarks>
///     <para>
///         <b>Why the artifact is stored as base64 chunks.</b> See
///         <see cref="DeploymentCacheSchema"/> — replication rejects BLOB columns and batches by row
///         count against gRPC's message cap, so a whole-artifact row could never be delivered.
///         Everything in this class that looks like ceremony (chunk indices, a chunk count, a
///         separate SHA over the raw bytes) exists to make that split safely reversible.
///     </para>
/// </remarks>
public sealed class LocalDbDeploymentArtifactCache : IDeploymentArtifactCache
{
    /// <summary>
    ///     Raw bytes per chunk, before base64 expansion.
    /// </summary>
    /// <remarks>
    ///     128 KiB raw is ≈ 171 KB encoded, which keeps a worst-case replication batch inside gRPC's
    ///     4 MB default cap with room to spare.
    /// </remarks>
    private const int ChunkSize = 128 * 1024;

    /// <summary>
    ///     How many deployments per cluster survive a store.
    /// </summary>
    /// <remarks>
    ///     Two, not one: the previous artifact is what a rollback would need, and keeping it costs
    ///     one artifact's worth of disk. Three would only add a generation nobody rolls back to.
    /// </remarks>
    private const int RetainedDeployments = 2;

    private readonly ILocalDb _db;
    private readonly ILogger<LocalDbDeploymentArtifactCache> _logger;

    public LocalDbDeploymentArtifactCache(ILocalDb db, ILogger<LocalDbDeploymentArtifactCache> logger)
    {
        ArgumentNullException.ThrowIfNull(db);
        ArgumentNullException.ThrowIfNull(logger);

        _db = db;
        _logger = logger;
    }

    /// <inheritdoc/>
    public async Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
                                 byte[] artifact, CancellationToken ct = default)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(clusterId);
        ArgumentException.ThrowIfNullOrWhiteSpace(deploymentId);
        ArgumentException.ThrowIfNullOrWhiteSpace(revisionHash);
        ArgumentNullException.ThrowIfNull(artifact);

        var sha = Convert.ToHexString(SHA256.HashData(artifact));
        var cachedAtUtc = FormatTimestamp(DateTimeOffset.UtcNow);
        var chunkCount = (artifact.Length + ChunkSize - 1) / ChunkSize;

        if (await IsAlreadyCachedAsync(clusterId, deploymentId, revisionHash, sha, chunkCount, ct))
        {
            _logger.LogDebug(
                "Deployment {DeploymentId} (revision {RevisionHash}) is already cached intact for cluster {ClusterId}; skipping the re-store.",
                deploymentId, revisionHash, clusterId);
            return;
        }

        // One transaction for the whole store. A pointer that commits without its chunks — or
        // chunks that commit without the pointer — is a cache that reads back as a corrupt hit
        // rather than a clean miss, which is the one outcome this type exists to prevent.
        await using var tx = await _db.BeginTransactionAsync(ct);

        // Delete-then-insert rather than upsert-per-chunk: a re-store with FEWER chunks than last
        // time would otherwise leave the old tail behind, and those orphans read back as a
        // chunk-count mismatch forever.
        await tx.ExecuteAsync(
            "DELETE FROM deployment_artifacts WHERE deployment_id = @DeploymentId",
            new { DeploymentId = deploymentId },
            ct);

        for (var index = 0; index < chunkCount; index++)
        {
            var offset = index * ChunkSize;
            var length = Math.Min(ChunkSize, artifact.Length - offset);

            await tx.ExecuteAsync(
                """
                INSERT INTO deployment_artifacts
                    (deployment_id, chunk_index, cluster_id, revision_hash, chunk_count,
                     chunk_base64, cached_at_utc)
                VALUES
                    (@DeploymentId, @ChunkIndex, @ClusterId, @RevisionHash, @ChunkCount,
                     @ChunkBase64, @CachedAtUtc)
                """,
                new
                {
                    DeploymentId = deploymentId,
                    ChunkIndex = index,
                    ClusterId = clusterId,
                    RevisionHash = revisionHash,
                    ChunkCount = chunkCount,
                    ChunkBase64 = Convert.ToBase64String(artifact, offset, length),
                    CachedAtUtc = cachedAtUtc,
                },
                ct);
        }

        await tx.ExecuteAsync(
            """
            INSERT INTO deployment_pointer
                (cluster_id, deployment_id, revision_hash, artifact_sha256, applied_at_utc)
            VALUES
                (@ClusterId, @DeploymentId, @RevisionHash, @Sha, @AppliedAtUtc)
            ON CONFLICT(cluster_id) DO UPDATE SET
                deployment_id   = excluded.deployment_id,
                revision_hash   = excluded.revision_hash,
                artifact_sha256 = excluded.artifact_sha256,
                applied_at_utc  = excluded.applied_at_utc
            """,
            new
            {
                ClusterId = clusterId,
                DeploymentId = deploymentId,
                RevisionHash = revisionHash,
                Sha = sha,
                AppliedAtUtc = cachedAtUtc,
            },
            ct);

        // Prune inside the same transaction so the cache is never briefly unbounded.
        //
        // The `deployment_id <> @DeploymentId` clause is load-bearing, not belt-and-braces: it makes
        // "the deployment the pointer names is always present" a structural invariant instead of a
        // consequence of clock resolution. Without it, three stores landing inside one timestamp
        // tick fall through to the `deployment_id DESC` tiebreak — and since real deployment ids are
        // GUIDs, that ordering is effectively random, so the row just written can lose. The pointer
        // would then name chunks that no longer exist, which reads back as a cache miss on every
        // subsequent boot until the next deploy overwrites it. Silent and permanent: exactly the
        // failure this cache exists to prevent.
        await tx.ExecuteAsync(
            $"""
             DELETE FROM deployment_artifacts
             WHERE cluster_id = @ClusterId
               AND deployment_id <> @DeploymentId
               AND deployment_id NOT IN (
                   SELECT deployment_id
                   FROM deployment_artifacts
                   WHERE cluster_id = @ClusterId
                   GROUP BY deployment_id
                   ORDER BY MAX(cached_at_utc) DESC, deployment_id DESC
                   LIMIT {RetainedDeployments}
               )
             """,
            new { ClusterId = clusterId, DeploymentId = deploymentId },
            ct);

        await tx.CommitAsync(ct);
    }

    /// <summary>
    ///     Whether this exact artifact is already cached <b>and readable</b> for the cluster, so
    ///     storing it again would rewrite every row to its current value.
    /// </summary>
    /// <remarks>
    ///     <para>
    ///         <b>Why this exists.</b> Every write to a replicated table fires a capture trigger and
    ///         mints an oplog row. Re-caching an artifact the node already holds — which is what a
    ///         restart's boot-from-cache and every <c>RestoreApplied</c> recovery does — would
    ///         therefore cost a delete plus an insert per chunk plus a pointer update, all carrying
    ///         values identical to what is already there. On a replicating pair that is pure churn;
    ///         on an unpaired (default-OFF) node there is no peer to ack those rows, so they
    ///         accumulate in the oplog until the library's age cap prunes them. The Phase 1 live gate
    ///         saw exactly that (check 8).
    ///     </para>
    ///     <para>
    ///         <b>Identity is over the bytes, not the ids.</b> The pointer must match on
    ///         <c>artifact_sha256</c> as well as the deployment id and revision hash: a re-composed
    ///         artifact can legitimately carry the same ids with different bytes, and keeping the
    ///         stale copy would serve a boot-from-cache node the wrong address space while every
    ///         id-shaped signal claimed it was right.
    ///     </para>
    ///     <para>
    ///         <b>The chunk count is load-bearing.</b> A pointer can name a deployment whose chunks
    ///         are missing or partial — a half-replicated artifact, or a prune that raced the
    ///         pointer. Skipping on the pointer alone would make that state permanent, because the
    ///         re-store that repairs it is the call being skipped.
    ///     </para>
    ///     <para>
    ///         <b>Accepted consequence:</b> a skipped store leaves <c>applied_at_utc</c> at the
    ///         moment the artifact was <i>first</i> cached rather than last confirmed. Refreshing it
    ///         would mint the very oplog row this check exists to avoid. Nothing reads it as a
    ///         liveness signal; retention ranks distinct deployments, which are unaffected.
    ///     </para>
    /// </remarks>
    private async Task<bool> IsAlreadyCachedAsync(
        string clusterId, string deploymentId, string revisionHash, string sha, int chunkCount,
        CancellationToken ct)
    {
        var matches = await _db.QueryAsync(
            """
            SELECT (
                SELECT COUNT(*)
                FROM deployment_artifacts
                WHERE cluster_id = @ClusterId
                  AND deployment_id = @DeploymentId
                  AND chunk_count = @ChunkCount
            )
            FROM deployment_pointer
            WHERE cluster_id = @ClusterId
              AND deployment_id = @DeploymentId
              AND revision_hash = @RevisionHash
              AND artifact_sha256 = @Sha
            """,
            r => r.GetInt32(0),
            new
            {
                ClusterId = clusterId,
                DeploymentId = deploymentId,
                RevisionHash = revisionHash,
                Sha = sha,
                ChunkCount = chunkCount,
            },
            ct);

        return matches.Count == 1 && matches[0] == chunkCount;
    }

    /// <inheritdoc/>
    public async Task<CachedDeploymentArtifact?> GetCurrentAsync(
        string clusterId, CancellationToken ct = default)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(clusterId);

        var pointers = await _db.QueryAsync(
            """
            SELECT deployment_id, revision_hash, artifact_sha256, applied_at_utc
            FROM deployment_pointer
            WHERE cluster_id = @ClusterId
            """,
            ReadPointer,
            new { ClusterId = clusterId },
            ct);

        if (pointers.Count == 0)
        {
            _logger.LogDebug("No cached deployment pointer for cluster {ClusterId}.", clusterId);
            return null;
        }

        return await ReassembleAsync(clusterId, pointers[0], ct);
    }

    /// <inheritdoc/>
    public async Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
    {
        // applied_at_utc is round-trip ISO-8601 UTC, so ordering it as TEXT is chronological —
        // that is the reason the format is pinned rather than left to the current culture.
        var pointers = await _db.QueryAsync(
            """
            SELECT cluster_id, deployment_id, revision_hash, artifact_sha256, applied_at_utc
            FROM deployment_pointer
            ORDER BY applied_at_utc DESC
            """,
            r => (ClusterId: r.GetString(0), Pointer: new PointerRow(
                DeploymentId: r.GetString(1),
                RevisionHash: r.GetString(2),
                ArtifactSha256: r.GetString(3),
                AppliedAtUtc: r.GetString(4))),
            parameters: null,
            ct);

        if (pointers.Count == 0)
        {
            _logger.LogDebug("No cached deployment pointer of any cluster.");
            return null;
        }

        if (pointers.Count > 1)
        {
            // The node was re-homed between clusters. Booting the newest is the only defensible
            // choice, but it must never be a silent one — a wrong-cluster address space presents as
            // a plausible configuration, so the operator needs the cluster names to spot it.
            _logger.LogWarning(
                "Local deployment cache holds pointers for {PointerCount} clusters ({ClusterIds}); " +
                "booting the newest ({ChosenClusterId}). This node appears to have been re-homed — " +
                "clear the stale cache if that is unexpected.",
                pointers.Count,
                string.Join(", ", pointers.Select(p => p.ClusterId)),
                pointers[0].ClusterId);
        }

        return await ReassembleAsync(pointers[0].ClusterId, pointers[0].Pointer, ct);
    }

    /// <summary>
    ///     Rebuilds the artifact the pointer names, returning <see langword="null"/> unless every
    ///     integrity check passes.
    /// </summary>
    private async Task<CachedDeploymentArtifact?> ReassembleAsync(
        string clusterId, PointerRow pointer, CancellationToken ct)
    {
        var chunks = await _db.QueryAsync(
            """
            SELECT chunk_count, chunk_base64
            FROM deployment_artifacts
            WHERE deployment_id = @DeploymentId
            ORDER BY chunk_index
            """,
            r => (ChunkCount: r.GetInt32(0), Base64: r.GetString(1)),
            new { pointer.DeploymentId },
            ct);

        // The chunk_count carried on every row is what makes a partial replica detectable. Without
        // it a missing tail chunk is indistinguishable from a shorter artifact.
        var expectedChunkCount = chunks.Count > 0 ? chunks[0].ChunkCount : 0;

        if (chunks.Count != expectedChunkCount)
        {
            _logger.LogWarning(
                "Cached deployment {DeploymentId} for cluster {ClusterId} is incomplete: found " +
                "{FoundChunks} of {ExpectedChunks} chunks. Treating as a cache miss.",
                pointer.DeploymentId, clusterId, chunks.Count, expectedChunkCount);
            return null;
        }

        byte[] artifact;
        try
        {
            artifact = Decode(chunks.Select(c => c.Base64));
        }
        catch (FormatException ex)
        {
            _logger.LogWarning(
                ex,
                "Cached deployment {DeploymentId} for cluster {ClusterId} has a chunk that is not " +
                "valid base64. Treating as a cache miss.",
                pointer.DeploymentId, clusterId);
            return null;
        }

        var actualSha = Convert.ToHexString(SHA256.HashData(artifact));
        if (!string.Equals(actualSha, pointer.ArtifactSha256, StringComparison.OrdinalIgnoreCase))
        {
            _logger.LogWarning(
                "Cached deployment {DeploymentId} for cluster {ClusterId} failed its SHA-256 check " +
                "(expected {ExpectedSha}, computed {ActualSha}). Treating as a cache miss.",
                pointer.DeploymentId, clusterId, pointer.ArtifactSha256, actualSha);
            return null;
        }

        if (!DateTimeOffset.TryParse(pointer.AppliedAtUtc, CultureInfo.InvariantCulture,
                DateTimeStyles.RoundtripKind, out var appliedAtUtc))
        {
            _logger.LogWarning(
                "Cached deployment {DeploymentId} for cluster {ClusterId} has an unparseable " +
                "applied_at_utc ({AppliedAtUtc}). Treating as a cache miss.",
                pointer.DeploymentId, clusterId, pointer.AppliedAtUtc);
            return null;
        }

        return new CachedDeploymentArtifact(
            pointer.DeploymentId, pointer.RevisionHash, artifact, appliedAtUtc);
    }

    private static byte[] Decode(IEnumerable<string> chunksInOrder)
    {
        using var buffer = new MemoryStream();

        foreach (var chunk in chunksInOrder)
        {
            var bytes = Convert.FromBase64String(chunk);
            buffer.Write(bytes, 0, bytes.Length);
        }

        return buffer.ToArray();
    }

    private static PointerRow ReadPointer(Microsoft.Data.Sqlite.SqliteDataReader reader)
        => new(
            DeploymentId: reader.GetString(0),
            RevisionHash: reader.GetString(1),
            ArtifactSha256: reader.GetString(2),
            AppliedAtUtc: reader.GetString(3));

    /// <summary>
    ///     Round-trip ("O") UTC, so lexicographic ordering of the stored TEXT is chronological
    ///     ordering — the retention query sorts on it directly.
    /// </summary>
    private static string FormatTimestamp(DateTimeOffset value)
        => value.ToUniversalTime().ToString("O", CultureInfo.InvariantCulture);

    private sealed record PointerRow(
        string DeploymentId, string RevisionHash, string ArtifactSha256, string AppliedAtUtc);
}