From f4063f811f557a04fb0bb8c760cf2f26f73df7a0 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 17 Jul 2026 21:18:08 -0400 Subject: [PATCH] feat(localdb): opt-in table registration + capture triggers (oplog + row_version, applying guard) --- .../src/ZB.MOM.WW.LocalDb/ILocalDb.cs | 13 ++ .../Internal/SqliteLocalDb.cs | 53 ++++++ .../LocalDbRegistrationException.cs | 13 ++ .../Registration/ReplicatedTable.cs | 29 ++++ .../Registration/TriggerSqlGenerator.cs | 74 ++++++++ .../CaptureTriggerTests.cs | 163 ++++++++++++++++++ .../RegistrationTests.cs | 119 +++++++++++++ 7 files changed, 464 insertions(+) create mode 100644 ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/LocalDbRegistrationException.cs create mode 100644 ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/ReplicatedTable.cs create mode 100644 ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/TriggerSqlGenerator.cs create mode 100644 ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/CaptureTriggerTests.cs create mode 100644 ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs index d864328..480f68e 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs @@ -1,4 +1,5 @@ using Microsoft.Data.Sqlite; +using ZB.MOM.WW.LocalDb.Registration; namespace ZB.MOM.WW.LocalDb; @@ -31,6 +32,18 @@ public interface ILocalDb /// Escape hatch: an open, UDF-registered connection the caller owns and disposes. SqliteConnection CreateConnection(); + + /// + /// Opts a table into replication: validates it has an explicit primary key, computes its + /// schema digest, and installs the three AFTER triggers that capture every change into the + /// oplog + row-version tables within the same transaction as the consumer's write. Idempotent + /// (re-registering regenerates the triggers). Throws + /// if the name is unsafe, the table is missing, or it has no primary key. + /// + ReplicatedTable RegisterReplicated(string tableName); + + /// The tables currently opted into replication, keyed by table name. + IReadOnlyDictionary ReplicatedTables { get; } } /// A transaction scope over the local database. Dispose without commit rolls back. diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs index 6efdfd7..4b3c3d5 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs @@ -1,8 +1,10 @@ using System.Collections.Concurrent; using System.Globalization; using System.Reflection; +using System.Text.RegularExpressions; using Microsoft.Data.Sqlite; using ZB.MOM.WW.LocalDb.Hlc; +using ZB.MOM.WW.LocalDb.Registration; namespace ZB.MOM.WW.LocalDb.Internal; @@ -15,8 +17,13 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable // Allow-list because the value is interpolated into a PRAGMA statement (pragmas cannot be parameterized). private static readonly string[] AllowedSynchronous = ["OFF", "NORMAL", "FULL", "EXTRA"]; + // Gates every registered table name: it is interpolated into PRAGMA table_info and the trigger DDL. + private static readonly Regex TableNameRegex = new("^[A-Za-z_][A-Za-z0-9_]*$", RegexOptions.Compiled); + private static readonly ConcurrentDictionary PropertyCache = new(); + private readonly ConcurrentDictionary _replicatedTables = new(StringComparer.Ordinal); + private readonly string _synchronous; private readonly int _busyTimeoutMs; private readonly string _connectionString; @@ -126,6 +133,52 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable } } + public IReadOnlyDictionary ReplicatedTables => _replicatedTables; + + public ReplicatedTable RegisterReplicated(string tableName) + { + if (string.IsNullOrWhiteSpace(tableName) || !TableNameRegex.IsMatch(tableName)) + throw new LocalDbRegistrationException( + $"Cannot register table '{tableName}': name must be a plain identifier matching {TableNameRegex}."); + + var columns = new List<(string Name, string Type, int Pk)>(); + using (var cmd = _master.CreateCommand()) + { + cmd.CommandText = $"PRAGMA table_info('{tableName}')"; + using var reader = cmd.ExecuteReader(); + while (reader.Read()) + // table_info cols: 0=cid, 1=name, 2=type, 3=notnull, 4=dflt_value, 5=pk (1-based PK ordinal, 0 if none). + columns.Add((reader.GetString(1), reader.IsDBNull(2) ? "" : reader.GetString(2), reader.GetInt32(5))); + } + + if (columns.Count == 0) + throw new LocalDbRegistrationException($"Cannot register table '{tableName}': it does not exist."); + + var pkColumns = columns.Where(c => c.Pk > 0).OrderBy(c => c.Pk).Select(c => c.Name).ToList(); + if (pkColumns.Count == 0) + throw new LocalDbRegistrationException( + $"Cannot register table '{tableName}': replication requires an explicit primary key."); + + var allColumns = columns.Select(c => (c.Name, c.Type)).ToList(); + var digest = ReplicatedTable.ComputeDigest(tableName, pkColumns, allColumns); + var table = new ReplicatedTable(tableName, pkColumns, allColumns, digest); + + var script = TriggerSqlGenerator.BuildInstallScript(table); + using (var tx = _master.BeginTransaction()) + { + using (var cmd = _master.CreateCommand()) + { + cmd.Transaction = tx; + cmd.CommandText = script; + cmd.ExecuteNonQuery(); + } + tx.Commit(); + } + + _replicatedTables[tableName] = table; + return table; + } + public void Dispose() { if (_disposed) diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/LocalDbRegistrationException.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/LocalDbRegistrationException.cs new file mode 100644 index 0000000..fd7d512 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/LocalDbRegistrationException.cs @@ -0,0 +1,13 @@ +namespace ZB.MOM.WW.LocalDb.Registration; + +/// +/// Thrown when a table cannot be opted into replication: the name is unsafe, the table does +/// not exist, or it has no explicit primary key (required for stable per-row capture keys). +/// +public sealed class LocalDbRegistrationException : Exception +{ + public LocalDbRegistrationException(string message) : base(message) { } + + public LocalDbRegistrationException(string message, Exception innerException) + : base(message, innerException) { } +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/ReplicatedTable.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/ReplicatedTable.cs new file mode 100644 index 0000000..6154ee3 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/ReplicatedTable.cs @@ -0,0 +1,29 @@ +using System.Security.Cryptography; +using System.Text; + +namespace ZB.MOM.WW.LocalDb.Registration; + +/// +/// A table opted into replication: its primary-key columns (in PK ordinal order), all its +/// columns (in PRAGMA cid order), and a schema digest used to detect drift. +/// +public sealed record ReplicatedTable( + string Name, + IReadOnlyList PkColumns, + IReadOnlyList<(string Name, string Type)> Columns, + string Digest) +{ + /// + /// Lowercase-hex SHA-256 of "{name}|pk:{p1,p2}|cols:{c1:TYPE,c2:TYPE,...}" — PK columns + /// in ordinal order, all columns in cid order with uppercased declared types. + /// + internal static string ComputeDigest( + string name, IReadOnlyList pkColumns, IReadOnlyList<(string Name, string Type)> columns) + { + var pk = string.Join(",", pkColumns); + var cols = string.Join(",", columns.Select(c => $"{c.Name}:{c.Type.ToUpperInvariant()}")); + var canonical = $"{name}|pk:{pk}|cols:{cols}"; + var hash = SHA256.HashData(Encoding.UTF8.GetBytes(canonical)); + return Convert.ToHexStringLower(hash); + } +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/TriggerSqlGenerator.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/TriggerSqlGenerator.cs new file mode 100644 index 0000000..39d78ee --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Registration/TriggerSqlGenerator.cs @@ -0,0 +1,74 @@ +using System.Text; + +namespace ZB.MOM.WW.LocalDb.Registration; + +/// +/// Emits the three AFTER triggers (insert/update/delete) that capture every row change on a +/// registered table into __localdb_oplog + __localdb_row_version within the +/// consumer's own write transaction. The WHEN applying = 0 guard skips replicated applies; +/// the last_insert_rowid() back-reference binds each row-version row to the SAME +/// zb_hlc_next() stamp as its oplog entry (one stamp per row change). +/// +internal static class TriggerSqlGenerator +{ + public static string TriggerName(string table, string suffix) => $"__localdb_{table}_{suffix}"; + + /// DROP IF EXISTS + CREATE for all three triggers, ready to run in one transaction. + public static string BuildInstallScript(ReplicatedTable table) + { + var sb = new StringBuilder(); + foreach (var suffix in new[] { "ai", "au", "ad" }) + sb.Append("DROP TRIGGER IF EXISTS \"").Append(TriggerName(table.Name, suffix)).Append("\";\n"); + + sb.Append(BuildInsert(table)).Append('\n'); + sb.Append(BuildUpdate(table)).Append('\n'); + sb.Append(BuildDelete(table)).Append('\n'); + return sb.ToString(); + } + + private static string BuildInsert(ReplicatedTable t) => Capture(t, "ai", "INSERT", "NEW", + rowJson: JsonObject("NEW", t.Columns.Select(c => c.Name)), isTombstone: "0", + rvTail: "0, NULL"); + + private static string BuildUpdate(ReplicatedTable t) => Capture(t, "au", "UPDATE", "NEW", + rowJson: JsonObject("NEW", t.Columns.Select(c => c.Name)), isTombstone: "0", + rvTail: "0, NULL"); + + private static string BuildDelete(ReplicatedTable t) => Capture(t, "ad", "DELETE", "OLD", + rowJson: "NULL", isTombstone: "1", + rvTail: "1, strftime('%Y-%m-%dT%H:%M:%fZ','now')"); + + private static string Capture( + ReplicatedTable t, string suffix, string verb, string alias, + string rowJson, string isTombstone, string rvTail) + { + var name = TriggerName(t.Name, suffix); + var pkJson = JsonObject(alias, t.PkColumns); + var tableLit = Literal(t.Name); + return + $""" + CREATE TRIGGER "{name}" AFTER {verb} ON "{t.Name}" + WHEN (SELECT applying FROM __localdb_applying WHERE id = 1) = 0 + BEGIN + INSERT INTO __localdb_oplog (table_name, pk_json, row_json, hlc, node_id, is_tombstone) + VALUES ({tableLit}, + {pkJson}, + {rowJson}, + zb_hlc_next(), + (SELECT node_id FROM __localdb_meta WHERE id = 1), + {isTombstone}); + INSERT OR REPLACE INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) + SELECT table_name, pk_json, hlc, node_id, {rvTail} FROM __localdb_oplog WHERE seq = last_insert_rowid(); + END; + """; + } + + // json_object('col', ALIAS."col", ...) — natively handles NULLs and SQLite types (BLOB columns error, a v1 limitation). + private static string JsonObject(string alias, IEnumerable columns) + { + var args = columns.Select(c => $"{Literal(c)}, {alias}.\"{c}\""); + return $"json_object({string.Join(", ", args)})"; + } + + private static string Literal(string value) => "'" + value.Replace("'", "''") + "'"; +} diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/CaptureTriggerTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/CaptureTriggerTests.cs new file mode 100644 index 0000000..98fb94f --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/CaptureTriggerTests.cs @@ -0,0 +1,163 @@ +using Microsoft.Data.Sqlite; +using ZB.MOM.WW.LocalDb.Internal; + +namespace ZB.MOM.WW.LocalDb.Tests; + +public sealed class CaptureTriggerTests : IDisposable +{ + private readonly string _path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); + + public void Dispose() + { + SqliteConnection.ClearAllPools(); + if (File.Exists(_path)) + File.Delete(_path); + } + + private SqliteLocalDb Create() => new(new LocalDbOptions { Path = _path }); + + private async Task NewOrdersDb() + { + var db = Create(); + await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)"); + db.RegisterReplicated("orders"); + return db; + } + + private sealed record OplogRow(long Seq, string PkJson, string? RowJson, long Hlc, string NodeId, long IsTombstone); + + private Task> Oplog(SqliteLocalDb db) => + db.QueryAsync( + "SELECT seq, pk_json, row_json, hlc, node_id, is_tombstone FROM __localdb_oplog WHERE table_name='orders' ORDER BY seq", + r => new OplogRow( + r.GetInt64(0), r.GetString(1), r.IsDBNull(2) ? null : r.GetString(2), + r.GetInt64(3), r.GetString(4), r.GetInt64(5))); + + [Fact] + public async Task Insert_WritesOplogAndRowVersion_SameHlc() + { + using var db = await NewOrdersDb(); + + await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'ABC', 5)"); + + var oplog = await Oplog(db); + var row = Assert.Single(oplog); + Assert.Equal("{\"id\":1}", row.PkJson); + Assert.Equal("{\"id\":1,\"sku\":\"ABC\",\"qty\":5}", row.RowJson); + Assert.Equal(0, row.IsTombstone); + Assert.Equal(db.NodeId, row.NodeId); + + var rvHlc = await db.QueryAsync( + "SELECT hlc FROM __localdb_row_version WHERE table_name='orders' AND pk_json='{\"id\":1}'", + r => r.GetInt64(0)); + Assert.Equal(row.Hlc, Assert.Single(rvHlc)); + } + + [Fact] + public async Task Update_WritesNewFullRow() + { + using var db = await NewOrdersDb(); + await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'ABC', 5)"); + var insertHlc = (await Oplog(db)).Single().Hlc; + + await db.ExecuteAsync("UPDATE orders SET qty = 9 WHERE id = 1"); + + var oplog = await Oplog(db); + Assert.Equal(2, oplog.Count); + var newest = oplog[^1]; + Assert.Equal("{\"id\":1,\"sku\":\"ABC\",\"qty\":9}", newest.RowJson); + + var rvHlc = await db.QueryAsync( + "SELECT hlc FROM __localdb_row_version WHERE table_name='orders' AND pk_json='{\"id\":1}'", + r => r.GetInt64(0)); + Assert.Equal(newest.Hlc, Assert.Single(rvHlc)); + Assert.True(newest.Hlc > insertHlc); + } + + [Fact] + public async Task Delete_WritesTombstone_NullRowJson_RowVersionTombstoned() + { + using var db = await NewOrdersDb(); + await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'ABC', 5)"); + + await db.ExecuteAsync("DELETE FROM orders WHERE id = 1"); + + var oplog = await Oplog(db); + var tomb = oplog[^1]; + Assert.Null(tomb.RowJson); + Assert.Equal(1, tomb.IsTombstone); + + var rv = await db.QueryAsync( + "SELECT is_tombstone, tombstone_utc FROM __localdb_row_version WHERE table_name='orders' AND pk_json='{\"id\":1}'", + r => (Tomb: r.GetInt64(0), Utc: r.IsDBNull(1) ? null : r.GetString(1))); + var v = Assert.Single(rv); + Assert.Equal(1, v.Tomb); + Assert.NotNull(v.Utc); + Assert.EndsWith("Z", v.Utc); + Assert.True(DateTimeOffset.TryParse(v.Utc, out _), $"tombstone_utc '{v.Utc}' should be ISO-8601"); + } + + [Fact] + public async Task MultiRowStatement_OneOplogEntryPerRow_DistinctHlcs() + { + using var db = await NewOrdersDb(); + + await db.ExecuteAsync( + "INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1), (2, 'B', 2), (3, 'C', 3)"); + + var oplog = await Oplog(db); + Assert.Equal(3, oplog.Count); + Assert.Equal(3, oplog.Select(o => o.Hlc).Distinct().Count()); + } + + [Fact] + public async Task ApplyingFlag_SuppressesCapture() + { + using var db = await NewOrdersDb(); + + await using (var tx = await db.BeginTransactionAsync()) + { + await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 1 WHERE id = 1"); + await tx.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1)"); + await tx.ExecuteAsync("UPDATE __localdb_applying SET applying = 0 WHERE id = 1"); + await tx.CommitAsync(default); + } + + var oplog = await Oplog(db); + Assert.Empty(oplog); + + var rv = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_row_version", r => r.GetInt64(0)); + Assert.Equal(0L, rv[0]); + } + + [Fact] + public async Task UnregisteredTable_NoCapture() + { + using var db = await NewOrdersDb(); + await db.ExecuteAsync("CREATE TABLE other (id INTEGER PRIMARY KEY, v TEXT)"); + + await db.ExecuteAsync("INSERT INTO other (id, v) VALUES (1, 'x')"); + + var oplog = await db.QueryAsync( + "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name='other'", r => r.GetInt64(0)); + Assert.Equal(0L, oplog[0]); + } + + [Fact] + public async Task Capture_InsideConsumerTransaction_AtomicWithRollback() + { + using var db = await NewOrdersDb(); + + await using (var tx = await db.BeginTransactionAsync()) + { + await tx.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1)"); + await tx.RollbackAsync(default); + } + + var oplog = await Oplog(db); + Assert.Empty(oplog); + + var orders = await db.QueryAsync("SELECT COUNT(*) FROM orders", r => r.GetInt64(0)); + Assert.Equal(0L, orders[0]); + } +} diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs new file mode 100644 index 0000000..0c68608 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/RegistrationTests.cs @@ -0,0 +1,119 @@ +using Microsoft.Data.Sqlite; +using ZB.MOM.WW.LocalDb.Internal; +using ZB.MOM.WW.LocalDb.Registration; + +namespace ZB.MOM.WW.LocalDb.Tests; + +public sealed class RegistrationTests : IDisposable +{ + private readonly string _path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db"); + + public void Dispose() + { + SqliteConnection.ClearAllPools(); + if (File.Exists(_path)) + File.Delete(_path); + } + + private SqliteLocalDb Create() => new(new LocalDbOptions { Path = _path }); + + private static async Task CreateOrders(SqliteLocalDb db) => + await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)"); + + private Task> Triggers(SqliteLocalDb db) => + db.QueryAsync( + "SELECT name FROM sqlite_master WHERE type='trigger' AND tbl_name='orders' ORDER BY name", + r => r.GetString(0)); + + [Fact] + public async Task Register_TableWithPk_InstallsThreeTriggers() + { + using var db = Create(); + await CreateOrders(db); + + db.RegisterReplicated("orders"); + + var triggers = await Triggers(db); + Assert.Equal( + new[] { "__localdb_orders_ad", "__localdb_orders_ai", "__localdb_orders_au" }, + triggers); + } + + [Fact] + public async Task Register_NoPk_Throws() + { + using var db = Create(); + await db.ExecuteAsync("CREATE TABLE nopk (a TEXT, b TEXT)"); + + var ex = Assert.Throws(() => db.RegisterReplicated("nopk")); + Assert.Contains("nopk", ex.Message); + } + + [Fact] + public void Register_MissingTable_Throws() + { + using var db = Create(); + + var ex = Assert.Throws(() => db.RegisterReplicated("ghost")); + Assert.Contains("ghost", ex.Message); + } + + [Fact] + public async Task Register_IsIdempotent_RegeneratesTriggers() + { + using var db = Create(); + await CreateOrders(db); + + db.RegisterReplicated("orders"); + db.RegisterReplicated("orders"); + + var triggers = await Triggers(db); + Assert.Equal(3, triggers.Count); + } + + [Fact] + public async Task Register_CompositePk_Works() + { + using var db = Create(); + await db.ExecuteAsync("CREATE TABLE kv (a INTEGER, b INTEGER, v TEXT, PRIMARY KEY (a, b))"); + + db.RegisterReplicated("kv"); + await db.ExecuteAsync("INSERT INTO kv (a, b, v) VALUES (1, 2, 'x')"); + + var pk = await db.QueryAsync( + "SELECT pk_json FROM __localdb_oplog WHERE table_name='kv'", + r => r.GetString(0)); + Assert.Single(pk); + Assert.Equal("{\"a\":1,\"b\":2}", pk[0]); + } + + [Fact] + public async Task Digest_ChangesWhenSchemaChanges() + { + using var db = Create(); + await CreateOrders(db); + + db.RegisterReplicated("orders"); + var before = db.ReplicatedTables["orders"].Digest; + + await db.ExecuteAsync("ALTER TABLE orders ADD COLUMN note TEXT"); + db.RegisterReplicated("orders"); + var after = db.ReplicatedTables["orders"].Digest; + + Assert.NotEqual(before, after); + } + + [Fact] + public async Task ReplicatedTables_ExposesRegistry() + { + using var db = Create(); + await CreateOrders(db); + + db.RegisterReplicated("orders"); + + var table = db.ReplicatedTables["orders"]; + Assert.Equal("orders", table.Name); + Assert.Equal(new[] { "id" }, table.PkColumns); + Assert.Equal(new[] { "id", "sku", "qty" }, table.Columns.Select(c => c.Name).ToArray()); + } +}