diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardSchema.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardSchema.cs
new file mode 100644
index 00000000..70a8c270
--- /dev/null
+++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardSchema.cs
@@ -0,0 +1,124 @@
+using Microsoft.Data.Sqlite;
+
+namespace ZB.MOM.WW.ScadaBridge.StoreAndForward;
+
+///
+/// DDL for the sf_messages table, extracted from
+/// so it can be applied by whoever owns the
+/// database file.
+///
+/// Deliberately depends only on Microsoft.Data.Sqlite, not on the LocalDb library.
+/// The Host applies this DDL to a LocalDb-managed connection before
+/// RegisterReplicated installs the capture triggers; nothing about the schema
+/// itself is LocalDb-specific, and the store still calls it so a directly-constructed
+/// store (tests, tooling) remains self-sufficient.
+///
+///
+public static class StoreAndForwardSchema
+{
+ ///
+ /// Creates the sf_messages table and its indexes when absent, and additively
+ /// upgrades a table created by an older build. Idempotent — safe to run on every
+ /// startup.
+ ///
+ /// An open connection to the database that should hold the table.
+ public static void Apply(SqliteConnection connection)
+ {
+ ArgumentNullException.ThrowIfNull(connection);
+
+ using (var command = connection.CreateCommand())
+ {
+ command.CommandText = @"
+ CREATE TABLE IF NOT EXISTS 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
+ );
+
+ CREATE INDEX IF NOT EXISTS idx_sf_messages_status ON sf_messages(status);
+ CREATE INDEX IF NOT EXISTS idx_sf_messages_category ON sf_messages(category);
+ ";
+ command.ExecuteNonQuery();
+ }
+
+ // Additively add the execution_id /
+ // source_script columns. CREATE TABLE IF NOT EXISTS above does NOT add
+ // columns to a table that already exists from before these fields, so a
+ // databases created by an older build needs the columns ALTER-ed in.
+ // SQLite has no "ADD COLUMN IF NOT EXISTS"; the column presence is
+ // probed first and the ALTER skipped when already there. Both columns
+ // are nullable with no default, so any row buffered before this
+ // migration reads back ExecutionId/SourceScript = null (back-compat).
+ AddColumnIfMissing(connection, "execution_id", "TEXT");
+ AddColumnIfMissing(connection, "source_script", "TEXT");
+
+ // Additively add the
+ // parent_execution_id column the same way — a sibling to execution_id.
+ // Nullable with no default, so any row buffered before this migration
+ // reads back ParentExecutionId = null (back-compat).
+ AddColumnIfMissing(connection, "parent_execution_id", "TEXT");
+
+ // Additively add the epoch-ms sibling of last_attempt_at. The
+ // ISO-8601 text column stays authoritative for reads / back-compat; this
+ // INTEGER column drives the due predicate so the sweep no longer parses
+ // julianday() per row.
+ AddColumnIfMissing(connection, "last_attempt_at_ms", "INTEGER");
+
+ // One-time backfill for rows persisted before the ms column existed: derive
+ // epoch-ms from the text timestamp. The "... IS NULL" guard makes this run
+ // once per legacy DB and never match again — this is the only remaining
+ // julianday() use in the store.
+ using (var backfill = connection.CreateCommand())
+ {
+ backfill.CommandText = @"
+ UPDATE sf_messages
+ SET last_attempt_at_ms = CAST((julianday(last_attempt_at) - 2440587.5) * 86400000 AS INTEGER)
+ WHERE last_attempt_at IS NOT NULL AND last_attempt_at_ms IS NULL";
+ backfill.ExecuteNonQuery();
+ }
+
+ // Covering index for the due query (status filter + ms ordering column).
+ // Created after the ALTER above so the column exists.
+ using (var dueIndex = connection.CreateCommand())
+ {
+ dueIndex.CommandText =
+ "CREATE INDEX IF NOT EXISTS idx_sf_messages_status_due ON sf_messages(status, last_attempt_at_ms)";
+ dueIndex.ExecuteNonQuery();
+ }
+ }
+
+ ///
+ /// Adds a column to sf_messages
+ /// only when it is not already present. SQLite lacks ADD COLUMN IF NOT
+ /// EXISTS, so the schema is probed via PRAGMA table_info first.
+ /// Idempotent — safe to run on every .
+ ///
+ private static void AddColumnIfMissing(
+ SqliteConnection connection, string columnName, string columnType)
+ {
+ using (var probe = connection.CreateCommand())
+ {
+ probe.CommandText = "SELECT COUNT(*) FROM pragma_table_info('sf_messages') WHERE name = @name";
+ probe.Parameters.AddWithValue("@name", columnName);
+ if (Convert.ToInt32(probe.ExecuteScalar()) > 0)
+ {
+ return;
+ }
+ }
+
+ using var alter = connection.CreateCommand();
+ // Column name + type are caller-controlled constants, never user input —
+ // safe to interpolate (parameters are not permitted in DDL).
+ alter.CommandText = $"ALTER TABLE sf_messages ADD COLUMN {columnName} {columnType}";
+ alter.ExecuteNonQuery();
+ }
+}
diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
index 65f1162d..187d4436 100644
--- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
@@ -46,101 +46,15 @@ public class StoreAndForwardStorage
await walCmd.ExecuteNonQueryAsync();
}
- await using var command = connection.CreateCommand();
- command.CommandText = @"
- CREATE TABLE IF NOT EXISTS 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
- );
-
- CREATE INDEX IF NOT EXISTS idx_sf_messages_status ON sf_messages(status);
- CREATE INDEX IF NOT EXISTS idx_sf_messages_category ON sf_messages(category);
- ";
- await command.ExecuteNonQueryAsync();
-
- // Additively add the execution_id /
- // source_script columns. CREATE TABLE IF NOT EXISTS above does NOT add
- // columns to a table that already exists from before these fields, so a
- // databases created by an older build needs the columns ALTER-ed in.
- // SQLite has no "ADD COLUMN IF NOT EXISTS"; the column presence is
- // probed first and the ALTER skipped when already there. Both columns
- // are nullable with no default, so any row buffered before this
- // migration reads back ExecutionId/SourceScript = null (back-compat).
- await AddColumnIfMissingAsync(connection, "execution_id", "TEXT");
- await AddColumnIfMissingAsync(connection, "source_script", "TEXT");
-
- // Additively add the
- // parent_execution_id column the same way — a sibling to execution_id.
- // Nullable with no default, so any row buffered before this migration
- // reads back ParentExecutionId = null (back-compat).
- await AddColumnIfMissingAsync(connection, "parent_execution_id", "TEXT");
-
- // Additively add the epoch-ms sibling of last_attempt_at. The
- // ISO-8601 text column stays authoritative for reads / back-compat; this
- // INTEGER column drives the due predicate so the sweep no longer parses
- // julianday() per row.
- await AddColumnIfMissingAsync(connection, "last_attempt_at_ms", "INTEGER");
-
- // One-time backfill for rows persisted before the ms column existed: derive
- // epoch-ms from the text timestamp. The "... IS NULL" guard makes this run
- // once per legacy DB and never match again — this is the only remaining
- // julianday() use in the store.
- await using (var backfill = connection.CreateCommand())
- {
- backfill.CommandText = @"
- UPDATE sf_messages
- SET last_attempt_at_ms = CAST((julianday(last_attempt_at) - 2440587.5) * 86400000 AS INTEGER)
- WHERE last_attempt_at IS NOT NULL AND last_attempt_at_ms IS NULL";
- await backfill.ExecuteNonQueryAsync();
- }
-
- // Covering index for the due query (status filter + ms ordering column).
- // Created after the ALTER above so the column exists.
- await using (var dueIndex = connection.CreateCommand())
- {
- dueIndex.CommandText =
- "CREATE INDEX IF NOT EXISTS idx_sf_messages_status_due ON sf_messages(status, last_attempt_at_ms)";
- await dueIndex.ExecuteNonQueryAsync();
- }
+ // The DDL itself lives in StoreAndForwardSchema so the Host can apply it to a
+ // LocalDb-managed connection before RegisterReplicated installs the capture
+ // triggers. The store still calls it, so a directly-constructed store (tests,
+ // tooling) remains self-sufficient.
+ StoreAndForwardSchema.Apply(connection);
_logger.LogInformation("Store-and-forward SQLite storage initialized");
}
- ///
- /// Adds a column to sf_messages
- /// only when it is not already present. SQLite lacks ADD COLUMN IF NOT
- /// EXISTS, so the schema is probed via PRAGMA table_info first.
- /// Idempotent — safe to run on every .
- ///
- private static async Task AddColumnIfMissingAsync(
- SqliteConnection connection, string columnName, string columnType)
- {
- await using var probe = connection.CreateCommand();
- probe.CommandText = "SELECT COUNT(*) FROM pragma_table_info('sf_messages') WHERE name = @name";
- probe.Parameters.AddWithValue("@name", columnName);
- var exists = Convert.ToInt32(await probe.ExecuteScalarAsync()) > 0;
- if (exists)
- {
- return;
- }
-
- await using var alter = connection.CreateCommand();
- // Column name + type are caller-controlled constants, never user input —
- // safe to interpolate (parameters are not permitted in DDL).
- alter.CommandText = $"ALTER TABLE sf_messages ADD COLUMN {columnName} {columnType}";
- await alter.ExecuteNonQueryAsync();
- }
-
///
/// Ensures the directory for a file-backed SQLite database exists. SQLite creates
/// the database file on demand but not its parent directory, so a configured path
diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardSchemaTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardSchemaTests.cs
new file mode 100644
index 00000000..b4279e5a
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardSchemaTests.cs
@@ -0,0 +1,113 @@
+using Microsoft.Data.Sqlite;
+
+namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;
+
+///
+/// Pins as the single owner of the
+/// sf_messages DDL. The Host applies this to a LocalDb-managed connection
+/// before RegisterReplicated installs the capture triggers, so the shape it
+/// produces is the shape that replicates.
+///
+public class StoreAndForwardSchemaTests
+{
+ [Fact]
+ public void Apply_IsIdempotent_AndCreatesEveryColumn()
+ {
+ var path = Path.Combine(Path.GetTempPath(), $"sf-schema-{Guid.NewGuid():N}.db");
+ try
+ {
+ using var connection = new SqliteConnection($"Data Source={path}");
+ connection.Open();
+
+ StoreAndForwardSchema.Apply(connection);
+ StoreAndForwardSchema.Apply(connection); // second run must not throw
+
+ using var cmd = connection.CreateCommand();
+ cmd.CommandText = "SELECT name FROM pragma_table_info('sf_messages')";
+ var columns = new List();
+ using var reader = cmd.ExecuteReader();
+ while (reader.Read()) columns.Add(reader.GetString(0));
+
+ // 12 original + 4 additive
+ Assert.Equal(16, columns.Count);
+ Assert.Contains("last_attempt_at_ms", columns);
+ Assert.Contains("parent_execution_id", columns);
+ }
+ finally
+ {
+ SqliteConnection.ClearAllPools();
+ File.Delete(path);
+ }
+ }
+
+ ///
+ /// The additive columns exist to upgrade a database created by an older build in
+ /// place. Asserting only against a freshly-created table would pass even if every
+ /// ALTER were deleted, because CREATE TABLE already lists them — so
+ /// this starts from the pre-upgrade shape and proves the ALTER path runs.
+ ///
+ [Fact]
+ public void Apply_UpgradesALegacyTable_WithoutLosingRows()
+ {
+ var path = Path.Combine(Path.GetTempPath(), $"sf-schema-legacy-{Guid.NewGuid():N}.db");
+ try
+ {
+ using var connection = new SqliteConnection($"Data Source={path}");
+ connection.Open();
+
+ // The 12-column shape as it existed before execution_id / source_script /
+ // parent_execution_id / last_attempt_at_ms were added.
+ using (var legacy = connection.CreateCommand())
+ {
+ legacy.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, last_attempt_at)
+ VALUES ('legacy-1', 0, 'target-a', '{}', '2026-01-01T00:00:00Z', '2026-01-02T03:04:05Z');
+ """;
+ legacy.ExecuteNonQuery();
+ }
+
+ StoreAndForwardSchema.Apply(connection);
+
+ using var check = connection.CreateCommand();
+ check.CommandText =
+ "SELECT COUNT(*) FROM pragma_table_info('sf_messages') WHERE name = 'last_attempt_at_ms'";
+ Assert.Equal(1, Convert.ToInt32(check.ExecuteScalar()));
+
+ // The pre-existing row survives and its epoch-ms sibling is backfilled from
+ // the authoritative text timestamp.
+ using var backfilled = connection.CreateCommand();
+ backfilled.CommandText =
+ "SELECT last_attempt_at_ms FROM sf_messages WHERE id = 'legacy-1'";
+ var ms = Convert.ToInt64(backfilled.ExecuteScalar());
+ var expected = DateTimeOffset.Parse("2026-01-02T03:04:05Z").ToUnixTimeMilliseconds();
+
+ // Tolerance of 1 ms, not sloppiness: the backfill derives epoch-ms from
+ // julianday(), whose double-precision day-fraction cannot represent every
+ // millisecond exactly, so it lands 1 ms low here. That rounding is
+ // pre-existing production behaviour carried over verbatim from
+ // StoreAndForwardStorage — the column is a sweep-ordering hint, with the
+ // ISO-8601 text column remaining authoritative for reads. Asserting exact
+ // equality would pin a precision the implementation never had.
+ Assert.InRange(ms, expected - 1, expected + 1);
+ }
+ finally
+ {
+ SqliteConnection.ClearAllPools();
+ File.Delete(path);
+ }
+ }
+}