refactor(sf): extract StoreAndForwardSchema.Apply from the storage class
Task 3 of the Phase 2 plan. Mirrors OperationTrackingSchema: the sf_messages
DDL, the four additive ALTERs, the last_attempt_at_ms backfill and the due-index
move verbatim into a static sync Apply(SqliteConnection), depending only on
Microsoft.Data.Sqlite. The Host needs to apply this to a LocalDb-managed
connection before RegisterReplicated installs the capture triggers; the store
still calls it so a directly-constructed store stays self-sufficient.
Two deviations from the plan as written, both deliberate:
1. The PRAGMA journal_mode=WAL in InitializeAsync STAYS. The plan's Step 4
snippet drops it, but Task 4 explicitly says not to move the equivalent
pragma ("LocalDb owns the connection's pragmas") — the two tasks contradict
each other. Keeping it preserves behaviour for the intermediate commits and
for directly-constructed stores; it becomes moot in Task 5 when the store
stops opening its own connections. Dropping it now would quietly regress the
concurrent-writer support the pragma's own comment documents.
2. Added a second test beyond the plan's. The specified test asserts only
against a freshly-created table, where CREATE TABLE already lists all 16
columns — it would pass with every ALTER deleted. The added test starts from
the pre-upgrade 12-column shape with a row in it and proves the upgrade path
runs and preserves data.
That second test also surfaced that the backfill lands 1 ms low: julianday()'s
double day-fraction cannot represent every millisecond. Pre-existing behaviour,
carried over verbatim, asserted with a 1 ms tolerance rather than pinning a
precision the implementation never had.
Suite: 154 passed, 0 failed.
Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
Joseph Doherty
@@ -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");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adds a column to <c>sf_messages</c>
|
||||
/// only when it is not already present. SQLite lacks <c>ADD COLUMN IF NOT
|
||||
/// EXISTS</c>, so the schema is probed via <c>PRAGMA table_info</c> first.
|
||||
/// Idempotent — safe to run on every <see cref="InitializeAsync"/>.
|
||||
/// </summary>
|
||||
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();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
|
||||
Reference in New Issue
Block a user