feat(localdb): migrate legacy store-and-forward.db into the consolidated DB
Task 8. Adds ResolveStoreAndForwardPath + MigrateStoreAndForward, wired as a third call in Migrate alongside the two Phase 1 files. Unlike the Phase 1 paths, the store-and-forward default sits INSIDE the data volume, so a real deployment has a real file here holding undelivered messages. This migration genuinely moves data; losing it would discard exactly the buffered calls store-and-forward exists to protect. No id synthesis: sf_messages.id is already a caller-assigned TEXT primary key, so INSERT OR IGNORE is idempotent across a crash-then-rerun. The copy intersects the legacy column set with the current one rather than naming all 16 columns outright. A file from an older build predates execution_id / parent_execution_id / last_attempt_at_ms, and naming a missing column throws 'no such column' — which the existing reader treats as an unrecognised shape and silently discards every row. A required-column (PK) guard keeps that tolerance from degrading into copying NULL-keyed rows. Four tests: copy+rename, crash-before-rename idempotence, the __localdb_oplog assertion that catches migrate-before-register, and the older-column-set case. All four verified to fail without the Migrate call. Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
@@ -5,8 +5,9 @@ using ZB.MOM.WW.LocalDb;
|
||||
namespace ZB.MOM.WW.ScadaBridge.Host;
|
||||
|
||||
/// <summary>
|
||||
/// One-time copy of the pre-Phase-1 site databases (<c>site-tracking.db</c> and
|
||||
/// <c>site_events.db</c>) into the consolidated <c>ZB.MOM.WW.LocalDb</c> database.
|
||||
/// One-time copy of the pre-consolidation site databases — Phase 1's
|
||||
/// <c>site-tracking.db</c> and <c>site_events.db</c>, and Phase 2's
|
||||
/// <c>store-and-forward.db</c> — into the consolidated <c>ZB.MOM.WW.LocalDb</c> database.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
@@ -31,13 +32,20 @@ namespace ZB.MOM.WW.ScadaBridge.Host;
|
||||
/// half-migrated state to reason about. A second boot sees the renamed file and no-ops.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Finding nothing is the expected case on the docker rig.</b> Neither legacy config key
|
||||
/// <b>Finding nothing is the expected case on the docker rig for the two Phase 1 files.</b>
|
||||
/// Neither legacy config key
|
||||
/// is set in any rig appsettings, so both fall back to CWD-relative code defaults
|
||||
/// (<c>/app/site-tracking.db</c>, <c>/app/site_events.db</c>) that sit OUTSIDE the mounted
|
||||
/// data volume — meaning they were already being discarded on every container recreate.
|
||||
/// Phase 1 incidentally fixes that data-loss bug by consolidating into
|
||||
/// <c>/app/data/site-localdb.db</c>. A no-op here is a legitimate result, not a failure.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Store-and-forward is the exception.</b> Its default path <i>is</i> inside the data
|
||||
/// volume (<c>./data/store-and-forward.db</c>), so a real deployment has a real file there
|
||||
/// holding undelivered messages. That migration genuinely moves data, and dropping it would
|
||||
/// silently discard exactly the buffered calls store-and-forward exists to protect.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static class SiteLocalDbLegacyMigrator
|
||||
{
|
||||
@@ -49,6 +57,22 @@ public static class SiteLocalDbLegacyMigrator
|
||||
/// <summary>Default legacy event-log path (<c>SiteEventLogOptions.DatabasePath</c>).</summary>
|
||||
private const string DefaultEventLogPath = "site_events.db";
|
||||
|
||||
/// <summary>Default legacy store-and-forward path (<c>StoreAndForwardOptions.SqliteDbPath</c>).</summary>
|
||||
private const string DefaultStoreAndForwardPath = "./data/store-and-forward.db";
|
||||
|
||||
/// <summary>
|
||||
/// Every column of the current <c>sf_messages</c> schema, in a fixed order. Columns
|
||||
/// absent from an older legacy file are dropped from the copy rather than failing it —
|
||||
/// see <see cref="PresentColumns"/>.
|
||||
/// </summary>
|
||||
private static readonly string[] StoreAndForwardColumns =
|
||||
[
|
||||
"id", "category", "target", "payload_json",
|
||||
"retry_count", "max_retries", "retry_interval_ms",
|
||||
"created_at", "last_attempt_at", "status", "last_error", "origin_instance",
|
||||
"execution_id", "source_script", "parent_execution_id", "last_attempt_at_ms",
|
||||
];
|
||||
|
||||
/// <summary>
|
||||
/// Copies any legacy site databases into <paramref name="db"/>, then renames them.
|
||||
/// </summary>
|
||||
@@ -63,6 +87,7 @@ public static class SiteLocalDbLegacyMigrator
|
||||
|
||||
MigrateTracking(db, ResolveTrackingPath(config));
|
||||
MigrateEvents(db, ResolveEventLogPath(config), nodeName);
|
||||
MigrateStoreAndForward(db, ResolveStoreAndForwardPath(config));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -113,6 +138,86 @@ public static class SiteLocalDbLegacyMigrator
|
||||
return Path.GetFullPath(path);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves the legacy store-and-forward database path from the OLD key, falling back
|
||||
/// to the code default. Unlike the two Phase 1 paths, this default sits INSIDE the
|
||||
/// mounted data volume (<c>./data/</c>), so on the docker rig there is a real file here
|
||||
/// with real buffered messages — this migration is not the usual no-op.
|
||||
/// </summary>
|
||||
internal static string ResolveStoreAndForwardPath(IConfiguration config)
|
||||
{
|
||||
var path = config["ScadaBridge:StoreAndForward:SqliteDbPath"] ?? DefaultStoreAndForwardPath;
|
||||
|
||||
if (string.IsNullOrWhiteSpace(path) ||
|
||||
path.Equals(":memory:", StringComparison.OrdinalIgnoreCase) ||
|
||||
path.StartsWith("file:", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
return string.Empty;
|
||||
}
|
||||
|
||||
return Path.GetFullPath(path);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Copies buffered store-and-forward messages out of the legacy file.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// No id synthesis, unlike <see cref="MigrateEvents"/>: <c>sf_messages.id</c> is already
|
||||
/// a caller-assigned TEXT primary key, so <c>INSERT OR IGNORE</c> is naturally idempotent
|
||||
/// across a crash-then-rerun.
|
||||
/// <para>
|
||||
/// These are undelivered messages, so dropping them is real data loss — a buffered call
|
||||
/// that never reaches its external system is exactly what store-and-forward exists to
|
||||
/// prevent. That is why the copy tolerates an older column set rather than bailing.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private static void MigrateStoreAndForward(ILocalDb db, string legacyPath)
|
||||
=> MigrateTable(db, legacyPath, "sf_messages", "id", StoreAndForwardColumns);
|
||||
|
||||
/// <summary>
|
||||
/// Copies one table from a legacy file into the consolidated database, then renames the
|
||||
/// legacy file.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The copy is restricted to the columns the legacy table <i>actually has</i>. A file
|
||||
/// written by an older build predates some columns, and naming a missing column in the
|
||||
/// SELECT would throw "no such column" — which the reader treats as an unrecognised
|
||||
/// shape, silently discarding every row in the table. Intersecting first means an old
|
||||
/// file migrates its data and simply leaves the newer columns at their schema defaults.
|
||||
/// </remarks>
|
||||
/// <param name="db">The consolidated database.</param>
|
||||
/// <param name="legacyPath">The legacy file, which may not exist.</param>
|
||||
/// <param name="table">Table name, identical on both sides.</param>
|
||||
/// <param name="requiredColumn">
|
||||
/// A column without which the table is not the one we mean — normally the primary key.
|
||||
/// Copying rows with a NULL PK would be worse than copying nothing.
|
||||
/// </param>
|
||||
/// <param name="columns">The current schema's full column list, in a fixed order.</param>
|
||||
private static void MigrateTable(
|
||||
ILocalDb db, string legacyPath, string table, string requiredColumn, IReadOnlyList<string> columns)
|
||||
{
|
||||
if (!ShouldMigrate(legacyPath)) return;
|
||||
|
||||
using (var legacy = OpenLegacyReadOnly(legacyPath))
|
||||
{
|
||||
var present = PresentColumns(legacy, table, columns);
|
||||
|
||||
// An absent table probes as zero columns, so this one guard covers both "old
|
||||
// file predating the table" and "file we do not recognise".
|
||||
if (present.Contains(requiredColumn))
|
||||
{
|
||||
using var connection = db.CreateConnection();
|
||||
using var transaction = connection.BeginTransaction();
|
||||
|
||||
CopyRows(legacy, connection, transaction, table, present);
|
||||
|
||||
transaction.Commit();
|
||||
}
|
||||
}
|
||||
|
||||
MarkMigrated(legacyPath);
|
||||
}
|
||||
|
||||
private static void MigrateTracking(ILocalDb db, string legacyPath)
|
||||
{
|
||||
if (!ShouldMigrate(legacyPath)) return;
|
||||
@@ -249,6 +354,63 @@ public static class SiteLocalDbLegacyMigrator
|
||||
return rows;
|
||||
}
|
||||
|
||||
private static SqliteConnection OpenLegacyReadOnly(string legacyPath)
|
||||
{
|
||||
var connection = new SqliteConnection($"Data Source={legacyPath};Mode=ReadOnly");
|
||||
connection.Open();
|
||||
return connection;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns the subset of <paramref name="wanted"/> that the legacy table actually has,
|
||||
/// in the caller's order. An absent table yields an empty list rather than throwing.
|
||||
/// </summary>
|
||||
private static List<string> PresentColumns(
|
||||
SqliteConnection legacy, string table, IReadOnlyList<string> wanted)
|
||||
{
|
||||
var present = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
using (var probe = legacy.CreateCommand())
|
||||
{
|
||||
// Table name is a caller-controlled constant, never user input — safe to
|
||||
// interpolate (parameters are not permitted as a pragma-function argument).
|
||||
probe.CommandText = $"SELECT name FROM pragma_table_info('{table}')";
|
||||
using var reader = probe.ExecuteReader();
|
||||
while (reader.Read()) present.Add(reader.GetString(0));
|
||||
}
|
||||
|
||||
return [.. wanted.Where(present.Contains)];
|
||||
}
|
||||
|
||||
/// <summary>Streams every row of <paramref name="columns"/> from the legacy table into the target.</summary>
|
||||
private static void CopyRows(
|
||||
SqliteConnection legacy,
|
||||
SqliteConnection target,
|
||||
SqliteTransaction transaction,
|
||||
string table,
|
||||
IReadOnlyList<string> columns)
|
||||
{
|
||||
var columnList = string.Join(", ", columns);
|
||||
var parameterList = string.Join(", ", columns.Select((_, i) => $"$p{i}"));
|
||||
|
||||
using var read = legacy.CreateCommand();
|
||||
read.CommandText = $"SELECT {columnList} FROM {table};";
|
||||
using var reader = read.ExecuteReader();
|
||||
|
||||
while (reader.Read())
|
||||
{
|
||||
using var write = target.CreateCommand();
|
||||
write.Transaction = transaction;
|
||||
write.CommandText =
|
||||
$"INSERT OR IGNORE INTO {table} ({columnList}) VALUES ({parameterList});";
|
||||
|
||||
for (var i = 0; i < columns.Count; i++)
|
||||
write.Parameters.AddWithValue($"$p{i}", reader.IsDBNull(i) ? DBNull.Value : reader.GetValue(i));
|
||||
|
||||
write.ExecuteNonQuery();
|
||||
}
|
||||
}
|
||||
|
||||
private static void Bind(SqliteCommand cmd, object?[] row)
|
||||
{
|
||||
for (var i = 0; i < row.Length; i++)
|
||||
|
||||
Reference in New Issue
Block a user