feat(site): deployed_secrets (LocalDb-replicated ciphertext) + node-local mirrored_secrets ledger
This commit is contained in:
Joseph Doherty
@@ -1,6 +1,8 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Artifacts;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
|
||||
|
||||
@@ -658,6 +660,230 @@ public class SiteStorageService
|
||||
_logger.LogDebug("Cleared native alarm state for {Instance}", instanceName);
|
||||
}
|
||||
|
||||
// ── Deployed Secrets store (LocalDb-replicated ciphertext) ──
|
||||
|
||||
/// <summary>
|
||||
/// Full-set upsert of the site's deployed secrets: every <see cref="SecretArtifact"/> in
|
||||
/// <paramref name="secrets"/> is written (<c>is_removed = 0</c>), and every row not named
|
||||
/// in the set is soft-deleted (<c>is_removed = 1</c>) rather than actually removed.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Central ships the complete set of secrets referenced by a site's artifacts on every
|
||||
/// artifact deploy, mirroring the reconcile contract of
|
||||
/// <see cref="DeleteRowsExceptAsync"/> for the other artifact tables. Unlike those tables,
|
||||
/// <c>deployed_secrets</c> is never hard-deleted: it is one of
|
||||
/// <c>SiteLocalDbSetup.ReplicatedTables</c>, and a mass <c>DELETE</c> on a replicated table
|
||||
/// would be captured and shipped to the peer as a destructive resync — the same hazard
|
||||
/// <c>StoreAndForwardStorage.ReplaceAllAsync</c> was deleted for in LocalDb Phase 2. A row
|
||||
/// that a later deploy re-adds simply flips back to <c>is_removed = 0</c>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <c>row_json</c> is <c>JsonSerializer.Serialize(secret)</c> — the <see cref="SecretArtifact"/>
|
||||
/// record verbatim, so its <c>byte[]</c> ciphertext fields serialize as base64 and nothing
|
||||
/// but ciphertext ever lands in this column.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="secrets">The complete set of secrets that should exist on this site.</param>
|
||||
/// <param name="deploymentId">The deployment identifier that applied this set.</param>
|
||||
/// <returns>
|
||||
/// A task that resolves to the names that were applied (inserted or updated) and the
|
||||
/// names that were newly marked removed by this call (rows that were previously
|
||||
/// <c>is_removed = 0</c>; a name already removed by an earlier call is not repeated here).
|
||||
/// </returns>
|
||||
public async Task<(IReadOnlyList<string> Applied, IReadOnlyList<string> Removed)> UpsertDeployedSecretsAsync(
|
||||
IReadOnlyList<SecretArtifact> secrets, string deploymentId)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(secrets);
|
||||
ArgumentException.ThrowIfNullOrWhiteSpace(deploymentId);
|
||||
|
||||
await using var connection = OpenConnection();
|
||||
await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
|
||||
|
||||
var appliedAt = DateTimeOffset.UtcNow.ToString("O");
|
||||
var applied = new List<string>(secrets.Count);
|
||||
|
||||
await using (var upsert = connection.CreateCommand())
|
||||
{
|
||||
upsert.Transaction = transaction;
|
||||
upsert.CommandText = @"
|
||||
INSERT INTO deployed_secrets (name, row_json, revision, updated_utc, is_removed, deployment_id, applied_at)
|
||||
VALUES (@name, @json, @revision, @updatedUtc, 0, @deploymentId, @appliedAt)
|
||||
ON CONFLICT(name) DO UPDATE SET
|
||||
row_json = excluded.row_json,
|
||||
revision = excluded.revision,
|
||||
updated_utc = excluded.updated_utc,
|
||||
is_removed = 0,
|
||||
deployment_id = excluded.deployment_id,
|
||||
applied_at = excluded.applied_at";
|
||||
|
||||
var pName = upsert.Parameters.Add("@name", SqliteType.Text);
|
||||
var pJson = upsert.Parameters.Add("@json", SqliteType.Text);
|
||||
var pRevision = upsert.Parameters.Add("@revision", SqliteType.Integer);
|
||||
var pUpdatedUtc = upsert.Parameters.Add("@updatedUtc", SqliteType.Text);
|
||||
var pDeploymentId = upsert.Parameters.Add("@deploymentId", SqliteType.Text);
|
||||
var pAppliedAt = upsert.Parameters.Add("@appliedAt", SqliteType.Text);
|
||||
pDeploymentId.Value = deploymentId;
|
||||
pAppliedAt.Value = appliedAt;
|
||||
|
||||
foreach (var secret in secrets)
|
||||
{
|
||||
pName.Value = secret.Name;
|
||||
pJson.Value = JsonSerializer.Serialize(secret);
|
||||
pRevision.Value = secret.Revision;
|
||||
pUpdatedUtc.Value = secret.UpdatedUtc.ToString("O");
|
||||
await upsert.ExecuteNonQueryAsync();
|
||||
applied.Add(secret.Name);
|
||||
}
|
||||
}
|
||||
|
||||
// Names present in the incoming set are never candidates for removal, including
|
||||
// one that was previously soft-deleted and has just been re-applied above.
|
||||
var keepParams = secrets.Select((s, i) => (Key: $"@k{i}", Value: s.Name)).ToList();
|
||||
var predicate = keepParams.Count == 0
|
||||
? " WHERE is_removed = 0"
|
||||
: $" WHERE is_removed = 0 AND name NOT IN ({string.Join(", ", keepParams.Select(p => p.Key))})";
|
||||
|
||||
var removed = new List<string>();
|
||||
await using (var select = connection.CreateCommand())
|
||||
{
|
||||
select.Transaction = transaction;
|
||||
select.CommandText = $"SELECT name FROM deployed_secrets{predicate}";
|
||||
foreach (var (key, value) in keepParams)
|
||||
select.Parameters.AddWithValue(key, value);
|
||||
|
||||
await using var reader = await select.ExecuteReaderAsync();
|
||||
while (await reader.ReadAsync())
|
||||
removed.Add(reader.GetString(0));
|
||||
}
|
||||
|
||||
if (removed.Count > 0)
|
||||
{
|
||||
await using var markRemoved = connection.CreateCommand();
|
||||
markRemoved.Transaction = transaction;
|
||||
markRemoved.CommandText = $"UPDATE deployed_secrets SET is_removed = 1{predicate}";
|
||||
foreach (var (key, value) in keepParams)
|
||||
markRemoved.Parameters.AddWithValue(key, value);
|
||||
|
||||
await markRemoved.ExecuteNonQueryAsync();
|
||||
}
|
||||
|
||||
await transaction.CommitAsync();
|
||||
|
||||
_logger.LogInformation(
|
||||
"Deployed secrets reconcile for deployment {DeploymentId}: applied {AppliedCount}, newly removed {RemovedCount}",
|
||||
deploymentId, applied.Count, removed.Count);
|
||||
|
||||
return (applied, removed);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns every row in <c>deployed_secrets</c>, including soft-deleted (<c>is_removed</c>)
|
||||
/// ones, with each row's <see cref="SecretArtifact"/> round-tripped from <c>row_json</c>.
|
||||
/// </summary>
|
||||
/// <returns>A task that resolves to every stored deployed-secret row.</returns>
|
||||
public async Task<IReadOnlyList<DeployedSecretRow>> GetDeployedSecretsAsync()
|
||||
{
|
||||
await using var connection = OpenConnection();
|
||||
|
||||
await using var command = connection.CreateCommand();
|
||||
command.CommandText = "SELECT name, row_json, is_removed, deployment_id FROM deployed_secrets";
|
||||
|
||||
var results = new List<DeployedSecretRow>();
|
||||
await using var reader = await command.ExecuteReaderAsync();
|
||||
while (await reader.ReadAsync())
|
||||
{
|
||||
var name = reader.GetString(0);
|
||||
var rowJson = reader.GetString(1);
|
||||
var secret = JsonSerializer.Deserialize<SecretArtifact>(rowJson)
|
||||
?? throw new InvalidOperationException($"deployed_secrets row '{name}' has invalid row_json");
|
||||
|
||||
results.Add(new DeployedSecretRow(
|
||||
name,
|
||||
secret,
|
||||
reader.GetInt64(2) != 0,
|
||||
reader.GetString(3)));
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
// ── Mirrored Secrets ledger (node-local only — deliberately NOT LocalDb-replicated) ──
|
||||
|
||||
/// <summary>
|
||||
/// Returns every name this node has recorded as mirrored into its own secret store.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <c>mirrored_secrets</c> is not in <c>SiteLocalDbSetup.ReplicatedTables</c> on purpose:
|
||||
/// mirroring writes a name into THIS node's own local secret store, so which names are
|
||||
/// mirrored — and at what revision — is per-node fact, not site-wide state. Replicating it
|
||||
/// would tell this node's ledger about the OTHER node's mirroring progress, which is not
|
||||
/// what it records.
|
||||
/// </remarks>
|
||||
/// <returns>A task that resolves to every mirrored-secret ledger row.</returns>
|
||||
public async Task<IReadOnlyList<MirroredSecretRow>> GetMirroredSecretsAsync()
|
||||
{
|
||||
await using var connection = OpenConnection();
|
||||
|
||||
await using var command = connection.CreateCommand();
|
||||
command.CommandText = "SELECT name, revision, mirrored_at FROM mirrored_secrets";
|
||||
|
||||
var results = new List<MirroredSecretRow>();
|
||||
await using var reader = await command.ExecuteReaderAsync();
|
||||
while (await reader.ReadAsync())
|
||||
{
|
||||
results.Add(new MirroredSecretRow(
|
||||
reader.GetString(0),
|
||||
reader.GetInt64(1),
|
||||
DateTimeOffset.Parse(reader.GetString(2), null, System.Globalization.DateTimeStyles.RoundtripKind)));
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Records that this node has mirrored <paramref name="name"/> at <paramref name="revision"/>
|
||||
/// into its own secret store. Uses UPSERT semantics.
|
||||
/// </summary>
|
||||
/// <param name="name">The mirrored secret's name.</param>
|
||||
/// <param name="revision">The revision that was mirrored.</param>
|
||||
/// <returns>A task that completes when the ledger row has been stored or updated.</returns>
|
||||
public async Task SetMirroredSecretAsync(string name, long revision)
|
||||
{
|
||||
await using var connection = OpenConnection();
|
||||
|
||||
await using var command = connection.CreateCommand();
|
||||
command.CommandText = @"
|
||||
INSERT INTO mirrored_secrets (name, revision, mirrored_at)
|
||||
VALUES (@name, @revision, @mirroredAt)
|
||||
ON CONFLICT(name) DO UPDATE SET
|
||||
revision = excluded.revision,
|
||||
mirrored_at = excluded.mirrored_at";
|
||||
|
||||
command.Parameters.AddWithValue("@name", name);
|
||||
command.Parameters.AddWithValue("@revision", revision);
|
||||
command.Parameters.AddWithValue("@mirroredAt", DateTimeOffset.UtcNow.ToString("O"));
|
||||
|
||||
await command.ExecuteNonQueryAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Removes the mirrored-secret ledger row for <paramref name="name"/> (e.g. once the
|
||||
/// secret is no longer deployed to this site and this node has un-mirrored it).
|
||||
/// </summary>
|
||||
/// <param name="name">The mirrored secret's name.</param>
|
||||
/// <returns>A task that completes when the ledger row has been deleted.</returns>
|
||||
public async Task RemoveMirroredSecretAsync(string name)
|
||||
{
|
||||
await using var connection = OpenConnection();
|
||||
|
||||
await using var command = connection.CreateCommand();
|
||||
command.CommandText = "DELETE FROM mirrored_secrets WHERE name = @name";
|
||||
command.Parameters.AddWithValue("@name", name);
|
||||
|
||||
await command.ExecuteNonQueryAsync();
|
||||
}
|
||||
|
||||
// ── Shared Script CRUD ──
|
||||
|
||||
/// <summary>
|
||||
@@ -1095,3 +1321,18 @@ public class StoredDataConnectionDefinition
|
||||
/// </summary>
|
||||
public record NativeAlarmRow(
|
||||
string SourceReference, string ConditionJson, DateTimeOffset LastTransitionAt, string? MetadataJson = null);
|
||||
|
||||
/// <summary>
|
||||
/// A single row from the site-local, LocalDb-replicated <c>deployed_secrets</c> table.
|
||||
/// <paramref name="Secret"/> is the <see cref="SecretArtifact"/> round-tripped from
|
||||
/// <c>row_json</c> — ciphertext only, never plaintext. <paramref name="IsRemoved"/> is true
|
||||
/// for a soft-deleted row (see <see cref="SiteStorageService.UpsertDeployedSecretsAsync"/>);
|
||||
/// the row is retained, not hard-deleted, because this table replicates.
|
||||
/// </summary>
|
||||
public record DeployedSecretRow(string Name, SecretArtifact Secret, bool IsRemoved, string DeploymentId);
|
||||
|
||||
/// <summary>
|
||||
/// A single row from the site-local, node-local (NOT LocalDb-replicated) <c>mirrored_secrets</c>
|
||||
/// ledger — the names this node has mirrored into its own secret store, and at what revision.
|
||||
/// </summary>
|
||||
public record MirroredSecretRow(string Name, long Revision, DateTimeOffset MirroredAt);
|
||||
|
||||
Reference in New Issue
Block a user