using System.Globalization;
using System.Security.Cryptography;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.LocalDb;
namespace ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
///
/// backed by the replicated LocalDb deployment-cache
/// tables.
///
///
///
/// Why the artifact is stored as base64 chunks. See
/// — 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.
///
///
public sealed class LocalDbDeploymentArtifactCache : IDeploymentArtifactCache
{
///
/// Raw bytes per chunk, before base64 expansion.
///
///
/// 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.
///
private const int ChunkSize = 128 * 1024;
///
/// How many deployments per cluster survive a store.
///
///
/// 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.
///
private const int RetainedDeployments = 2;
private readonly ILocalDb _db;
private readonly ILogger _logger;
public LocalDbDeploymentArtifactCache(ILocalDb db, ILogger logger)
{
ArgumentNullException.ThrowIfNull(db);
ArgumentNullException.ThrowIfNull(logger);
_db = db;
_logger = logger;
}
///
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;
// 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);
}
///
public async Task 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);
}
///
public async Task 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);
}
///
/// Rebuilds the artifact the pointer names, returning unless every
/// integrity check passes.
///
private async Task 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 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));
///
/// Round-trip ("O") UTC, so lexicographic ordering of the stored TEXT is chronological
/// ordering — the retention query sorts on it directly.
///
private static string FormatTimestamp(DateTimeOffset value)
=> value.ToUniversalTime().ToString("O", CultureInfo.InvariantCulture);
private sealed record PointerRow(
string DeploymentId, string RevisionHash, string ArtifactSha256, string AppliedAtUtc);
}