feat(localdb): opt-in table registration + capture triggers (oplog + row_version, applying guard)

2026-07-17 21:18:08 -04:00
parent 8b921e3106
commit f4063f811f
7 changed files with 464 additions and 0 deletions
@@ -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

    /// <summary>Escape hatch: an open, UDF-registered connection the caller owns and disposes.</summary>
    SqliteConnection CreateConnection();
+
+    /// <summary>
+    /// 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 <see cref="LocalDbRegistrationException"/>
+    /// if the name is unsafe, the table is missing, or it has no primary key.
+    /// </summary>
+    ReplicatedTable RegisterReplicated(string tableName);
+
+    /// <summary>The tables currently opted into replication, keyed by table name.</summary>
+    IReadOnlyDictionary<string, ReplicatedTable> ReplicatedTables { get; }
 }

 /// <summary>A transaction scope over the local database. Dispose without commit rolls back.</summary>
@@ -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<Type, PropertyInfo[]> PropertyCache = new();

+    private readonly ConcurrentDictionary<string, ReplicatedTable> _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<string, ReplicatedTable> 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)
@@ -0,0 +1,13 @@
+namespace ZB.MOM.WW.LocalDb.Registration;
+
+/// <summary>
+/// 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).
+/// </summary>
+public sealed class LocalDbRegistrationException : Exception
+{
+    public LocalDbRegistrationException(string message) : base(message) { }
+
+    public LocalDbRegistrationException(string message, Exception innerException)
+        : base(message, innerException) { }
+}
@@ -0,0 +1,29 @@
+using System.Security.Cryptography;
+using System.Text;
+
+namespace ZB.MOM.WW.LocalDb.Registration;
+
+/// <summary>
+/// A table opted into replication: its primary-key columns (in PK ordinal order), all its
+/// columns (in PRAGMA <c>cid</c> order), and a schema digest used to detect drift.
+/// </summary>
+public sealed record ReplicatedTable(
+    string Name,
+    IReadOnlyList<string> PkColumns,
+    IReadOnlyList<(string Name, string Type)> Columns,
+    string Digest)
+{
+    /// <summary>
+    /// Lowercase-hex SHA-256 of <c>"{name}|pk:{p1,p2}|cols:{c1:TYPE,c2:TYPE,...}"</c> — PK columns
+    /// in ordinal order, all columns in cid order with uppercased declared types.
+    /// </summary>
+    internal static string ComputeDigest(
+        string name, IReadOnlyList<string> 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);
+    }
+}
@@ -0,0 +1,74 @@
+using System.Text;
+
+namespace ZB.MOM.WW.LocalDb.Registration;
+
+/// <summary>
+/// Emits the three AFTER triggers (insert/update/delete) that capture every row change on a
+/// registered table into <c>__localdb_oplog</c> + <c>__localdb_row_version</c> within the
+/// consumer's own write transaction. The <c>WHEN applying = 0</c> guard skips replicated applies;
+/// the <c>last_insert_rowid()</c> back-reference binds each row-version row to the SAME
+/// <c>zb_hlc_next()</c> stamp as its oplog entry (one stamp per row change).
+/// </summary>
+internal static class TriggerSqlGenerator
+{
+    public static string TriggerName(string table, string suffix) => $"__localdb_{table}_{suffix}";
+
+    /// <summary>DROP IF EXISTS + CREATE for all three triggers, ready to run in one transaction.</summary>
+    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<string> columns)
+    {
+        var args = columns.Select(c => $"{Literal(c)}, {alias}.\"{c}\"");
+        return $"json_object({string.Join(", ", args)})";
+    }
+
+    private static string Literal(string value) => "'" + value.Replace("'", "''") + "'";
+}
@@ -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<SqliteLocalDb> 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<IReadOnlyList<OplogRow>> 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]);
+    }
+}
@@ -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<IReadOnlyList<string>> 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<LocalDbRegistrationException>(() => db.RegisterReplicated("nopk"));
+        Assert.Contains("nopk", ex.Message);
+    }
+
+    [Fact]
+    public void Register_MissingTable_Throws()
+    {
+        using var db = Create();
+
+        var ex = Assert.Throws<LocalDbRegistrationException>(() => 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());
+    }
+}