diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbLegacyMigrator.cs b/src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbLegacyMigrator.cs
index 3e52ab8e..1abb67fc 100644
--- a/src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbLegacyMigrator.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbLegacyMigrator.cs
@@ -5,8 +5,9 @@ using ZB.MOM.WW.LocalDb;
namespace ZB.MOM.WW.ScadaBridge.Host;
///
-/// One-time copy of the pre-Phase-1 site databases (site-tracking.db and
-/// site_events.db) into the consolidated ZB.MOM.WW.LocalDb database.
+/// One-time copy of the pre-consolidation site databases — Phase 1's
+/// site-tracking.db and site_events.db, and Phase 2's
+/// store-and-forward.db — into the consolidated ZB.MOM.WW.LocalDb database.
///
///
///
@@ -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.
///
///
-/// Finding nothing is the expected case on the docker rig. Neither legacy config key
+/// Finding nothing is the expected case on the docker rig for the two Phase 1 files.
+/// Neither legacy config key
/// is set in any rig appsettings, so both fall back to CWD-relative code defaults
/// (/app/site-tracking.db, /app/site_events.db) 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
/// /app/data/site-localdb.db. A no-op here is a legitimate result, not a failure.
///
+///
+/// Store-and-forward is the exception. Its default path is inside the data
+/// volume (./data/store-and-forward.db), 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.
+///
///
public static class SiteLocalDbLegacyMigrator
{
@@ -49,6 +57,22 @@ public static class SiteLocalDbLegacyMigrator
/// Default legacy event-log path (SiteEventLogOptions.DatabasePath).
private const string DefaultEventLogPath = "site_events.db";
+ /// Default legacy store-and-forward path (StoreAndForwardOptions.SqliteDbPath).
+ private const string DefaultStoreAndForwardPath = "./data/store-and-forward.db";
+
+ ///
+ /// Every column of the current sf_messages schema, in a fixed order. Columns
+ /// absent from an older legacy file are dropped from the copy rather than failing it —
+ /// see .
+ ///
+ 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",
+ ];
+
///
/// Copies any legacy site databases into , then renames them.
///
@@ -63,6 +87,7 @@ public static class SiteLocalDbLegacyMigrator
MigrateTracking(db, ResolveTrackingPath(config));
MigrateEvents(db, ResolveEventLogPath(config), nodeName);
+ MigrateStoreAndForward(db, ResolveStoreAndForwardPath(config));
}
///
@@ -113,6 +138,86 @@ public static class SiteLocalDbLegacyMigrator
return Path.GetFullPath(path);
}
+ ///
+ /// 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 (./data/), so on the docker rig there is a real file here
+ /// with real buffered messages — this migration is not the usual no-op.
+ ///
+ 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);
+ }
+
+ ///
+ /// Copies buffered store-and-forward messages out of the legacy file.
+ ///
+ ///
+ /// No id synthesis, unlike : sf_messages.id is already
+ /// a caller-assigned TEXT primary key, so INSERT OR IGNORE is naturally idempotent
+ /// across a crash-then-rerun.
+ ///
+ /// 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.
+ ///
+ ///
+ private static void MigrateStoreAndForward(ILocalDb db, string legacyPath)
+ => MigrateTable(db, legacyPath, "sf_messages", "id", StoreAndForwardColumns);
+
+ ///
+ /// Copies one table from a legacy file into the consolidated database, then renames the
+ /// legacy file.
+ ///
+ ///
+ /// The copy is restricted to the columns the legacy table actually has. 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.
+ ///
+ /// The consolidated database.
+ /// The legacy file, which may not exist.
+ /// Table name, identical on both sides.
+ ///
+ /// 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.
+ ///
+ /// The current schema's full column list, in a fixed order.
+ private static void MigrateTable(
+ ILocalDb db, string legacyPath, string table, string requiredColumn, IReadOnlyList 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;
+ }
+
+ ///
+ /// Returns the subset of that the legacy table actually has,
+ /// in the caller's order. An absent table yields an empty list rather than throwing.
+ ///
+ private static List PresentColumns(
+ SqliteConnection legacy, string table, IReadOnlyList wanted)
+ {
+ var present = new HashSet(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)];
+ }
+
+ /// Streams every row of from the legacy table into the target.
+ private static void CopyRows(
+ SqliteConnection legacy,
+ SqliteConnection target,
+ SqliteTransaction transaction,
+ string table,
+ IReadOnlyList 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++)
diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbLegacyMigratorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbLegacyMigratorTests.cs
index d3065f3c..e12e6be8 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbLegacyMigratorTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbLegacyMigratorTests.cs
@@ -40,7 +40,14 @@ public class SiteLocalDbLegacyMigratorTests : IDisposable
private string Path_(string name) => System.IO.Path.Combine(_root, name);
- /// A consolidated database with both tables created and registered, as the host has it.
+ /// A consolidated database with the tables created and registered, as the host has it.
+ ///
+ /// sf_messages is registered here even though production OnReady does not
+ /// register it until the Task 14 cutover. The oplog assertion below needs live capture
+ /// triggers to mean anything, and the property under test — that the migrator runs after
+ /// registration — is the same either way. The corollary is a real constraint on Task 14:
+ /// Migrate must stay the LAST call in OnReady, after every registration.
+ ///
private ILocalDb CreateConsolidated(IConfiguration config)
{
var provider = new ServiceCollection()
@@ -49,8 +56,10 @@ public class SiteLocalDbLegacyMigratorTests : IDisposable
using var connection = db.CreateConnection();
ZB.MOM.WW.ScadaBridge.SiteRuntime.Tracking.OperationTrackingSchema.Apply(connection);
ZB.MOM.WW.ScadaBridge.SiteEventLogging.SiteEventLogSchema.Apply(connection);
+ ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardSchema.Apply(connection);
db.RegisterReplicated("OperationTracking");
db.RegisterReplicated("site_events");
+ db.RegisterReplicated("sf_messages");
})
.BuildServiceProvider();
_providers.Add(provider);
@@ -59,12 +68,22 @@ public class SiteLocalDbLegacyMigratorTests : IDisposable
}
private IConfiguration Config(
- string? trackingPath = null, string? eventsPath = null, string nodeName = "node-a")
+ string? trackingPath = null,
+ string? eventsPath = null,
+ string nodeName = "node-a",
+ string? storeAndForwardPath = null)
{
var values = new Dictionary
{
["LocalDb:Path"] = Path_("consolidated.db"),
["ScadaBridge:Node:NodeName"] = nodeName,
+
+ // Always pinned inside the test's own directory, even when a test does not care
+ // about it. Left unset, the resolver falls back to the CWD-relative code default
+ // "./data/store-and-forward.db" — and the test run's CWD is the test binary's
+ // output directory, so an unlucky run could migrate (and RENAME) a real file.
+ ["ScadaBridge:StoreAndForward:SqliteDbPath"] =
+ storeAndForwardPath ?? Path_("absent-store-and-forward.db"),
};
if (trackingPath is not null)
@@ -75,6 +94,28 @@ public class SiteLocalDbLegacyMigratorTests : IDisposable
return new ConfigurationBuilder().AddInMemoryCollection(values).Build();
}
+ /// Seeds a legacy store-and-forward file with the CURRENT column set.
+ private static void SeedLegacyStoreAndForward(string path, params string[] ids)
+ {
+ using var connection = new SqliteConnection($"Data Source={path}");
+ connection.Open();
+ ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardSchema.Apply(connection);
+
+ foreach (var id in ids)
+ {
+ using var cmd = connection.CreateCommand();
+ cmd.CommandText = """
+ INSERT INTO sf_messages (
+ id, category, target, payload_json, retry_count, max_retries,
+ retry_interval_ms, created_at, status, execution_id)
+ VALUES ($id, 0, 'ERP', '{"order":1}', 0, 50, 30000, $now, 0, 'exec-1');
+ """;
+ cmd.Parameters.AddWithValue("$id", id);
+ cmd.Parameters.AddWithValue("$now", DateTime.UtcNow.ToString("o"));
+ cmd.ExecuteNonQuery();
+ }
+ }
+
private static void SeedLegacyTracking(string path, params string[] ids)
{
using var connection = new SqliteConnection($"Data Source={path}");
@@ -261,6 +302,99 @@ public class SiteLocalDbLegacyMigratorTests : IDisposable
Assert.Equal(2L, (long)cmd.ExecuteScalar()!);
}
+ [Fact]
+ public void SfMessages_AreCopiedFromTheLegacyFile()
+ {
+ // Phase 2 (Task 8). Unlike the two Phase 1 files, the store-and-forward default path
+ // is inside the data volume, so on a real deployment this actually moves data:
+ // undelivered messages that a lost migration would silently discard.
+ var sfPath = Path_("store-and-forward.db");
+ SeedLegacyStoreAndForward(sfPath, "msg-1", "msg-2", "msg-3");
+ var config = Config(storeAndForwardPath: sfPath);
+ var db = CreateConsolidated(config);
+
+ SiteLocalDbLegacyMigrator.Migrate(db, config);
+
+ Assert.Equal(3, CountRows(db, "sf_messages"));
+ Assert.Equal(["msg-1", "msg-2", "msg-3"], SelectIds(db, "sf_messages", "id"));
+
+ // Renamed, not deleted — same contract as the Phase 1 files.
+ Assert.False(File.Exists(sfPath));
+ Assert.True(File.Exists(sfPath + ".migrated"));
+ }
+
+ [Fact]
+ public void SfMessages_Migration_IsIdempotent_WhenRerunAfterACrashBeforeRename()
+ {
+ // The crash window: copy committed, rename did not. sf_messages.id is a natural
+ // TEXT key, so INSERT OR IGNORE absorbs the re-copy with no id synthesis needed.
+ var sfPath = Path_("store-and-forward.db");
+ SeedLegacyStoreAndForward(sfPath, "msg-1", "msg-2");
+ var config = Config(storeAndForwardPath: sfPath);
+ var db = CreateConsolidated(config);
+
+ SiteLocalDbLegacyMigrator.Migrate(db, config);
+ File.Move(sfPath + ".migrated", sfPath);
+ SiteLocalDbLegacyMigrator.Migrate(db, config);
+
+ Assert.Equal(2, CountRows(db, "sf_messages"));
+ }
+
+ [Fact]
+ public void MigratedSfMessages_EnterTheOplog_SoTheyActuallyReplicate()
+ {
+ // Same ordering trap as the Phase 1 tables: migrate before RegisterReplicated and
+ // the rows never enter the oplog, never reach the peer, and nothing errors. Only an
+ // assertion on __localdb_oplog catches the ordering being reversed.
+ var sfPath = Path_("store-and-forward.db");
+ SeedLegacyStoreAndForward(sfPath, "msg-1", "msg-2");
+ var config = Config(storeAndForwardPath: sfPath);
+ var db = CreateConsolidated(config);
+
+ SiteLocalDbLegacyMigrator.Migrate(db, config);
+
+ using var connection = db.CreateConnection();
+ using var cmd = connection.CreateCommand();
+ cmd.CommandText = "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'sf_messages'";
+ Assert.Equal(2L, (long)cmd.ExecuteScalar()!);
+ }
+
+ [Fact]
+ public void SfMessages_FromAnOlderBuild_MigrateDespiteMissingColumns()
+ {
+ // A legacy file written before execution_id / parent_execution_id / last_attempt_at_ms
+ // existed. Naming a missing column in the SELECT throws "no such column", and the
+ // reader treats that as an unrecognised shape — which would silently discard every
+ // buffered message. The copy intersects the column sets instead.
+ var sfPath = Path_("store-and-forward.db");
+ using (var legacy = new SqliteConnection($"Data Source={sfPath}"))
+ {
+ legacy.Open();
+ using var ddl = legacy.CreateCommand();
+ ddl.CommandText = """
+ CREATE TABLE sf_messages (
+ id TEXT PRIMARY KEY, category INTEGER NOT NULL, target TEXT NOT NULL,
+ payload_json TEXT NOT NULL, retry_count INTEGER NOT NULL DEFAULT 0,
+ max_retries INTEGER NOT NULL DEFAULT 50,
+ retry_interval_ms INTEGER NOT NULL DEFAULT 30000,
+ created_at TEXT NOT NULL, last_attempt_at TEXT,
+ status INTEGER NOT NULL DEFAULT 0, last_error TEXT, origin_instance TEXT
+ );
+ INSERT INTO sf_messages (id, category, target, payload_json, created_at)
+ VALUES ('old-1', 0, 'ERP', '{}', '2026-01-01T00:00:00Z');
+ """;
+ ddl.ExecuteNonQuery();
+ }
+
+ var config = Config(storeAndForwardPath: sfPath);
+ var db = CreateConsolidated(config);
+
+ SiteLocalDbLegacyMigrator.Migrate(db, config);
+
+ Assert.Equal(1, CountRows(db, "sf_messages"));
+ Assert.Equal(["old-1"], SelectIds(db, "sf_messages", "id"));
+ }
+
[Fact]
public void LegacyFileWithoutTheExpectedTable_IsTreatedAsEmpty_NotAFailure()
{