feat(localdb): SqliteLocalDb host (WAL, UDF-registered connections, HLC recovery, consumer SQL API)

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs

@@ -0,0 +1,39 @@
+using Microsoft.Data.Sqlite;
+
+namespace ZB.MOM.WW.LocalDb;
+
+/// <summary>
+/// Consumer SQL API over the embedded SQLite local database. Every connection this
+/// hands out has the <c>zb_hlc_next()</c> UDF registered, so capture triggers on
+/// registered tables can stamp writes fail-closed.
+/// </summary>
+public interface ILocalDb
+{
+    /// <summary>Executes a non-query statement; returns the affected row count.</summary>
+    Task<int> ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default);
+
+    /// <summary>Executes a query and projects each row through <paramref name="map"/>.</summary>
+    Task<IReadOnlyList<T>> QueryAsync<T>(string sql, Func<SqliteDataReader, T> map, object? parameters = null, CancellationToken ct = default);
+
+    /// <summary>Begins a transaction pinned to a single UDF-registered connection.</summary>
+    Task<ILocalDbTransaction> BeginTransactionAsync(CancellationToken ct = default);
+
+    /// <summary>Escape hatch: an open, UDF-registered connection the caller owns and disposes.</summary>
+    SqliteConnection CreateConnection();
+}
+
+/// <summary>A transaction scope over the local database. Dispose without commit rolls back.</summary>
+public interface ILocalDbTransaction : IAsyncDisposable
+{
+    /// <summary>Executes a non-query statement inside the transaction.</summary>
+    Task<int> ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default);
+
+    /// <summary>Executes a query inside the transaction.</summary>
+    Task<IReadOnlyList<T>> QueryAsync<T>(string sql, Func<SqliteDataReader, T> map, object? parameters = null, CancellationToken ct = default);
+
+    /// <summary>Commits the transaction.</summary>
+    Task CommitAsync(CancellationToken ct = default);
+
+    /// <summary>Rolls the transaction back.</summary>
+    Task RollbackAsync(CancellationToken ct = default);
+}

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs

@@ -0,0 +1,168 @@
+using System.Collections.Concurrent;
+using System.Globalization;
+using System.Reflection;
+using Microsoft.Data.Sqlite;
+using ZB.MOM.WW.LocalDb.Hlc;
+
+namespace ZB.MOM.WW.LocalDb.Internal;
+
+/// <summary>
+/// SQLite-backed <see cref="ILocalDb"/> host: opens/creates the file in WAL mode, ensures the
+/// library schema, recovers the HLC from durable state, and hands out UDF-registered connections.
+/// </summary>
+internal sealed class SqliteLocalDb : ILocalDb, IDisposable
+{
+    private static readonly ConcurrentDictionary<Type, PropertyInfo[]> PropertyCache = new();
+
+    private readonly LocalDbOptions _options;
+    private readonly string _connectionString;
+    private readonly SqliteConnection _master;
+    private readonly HybridLogicalClock _clock;
+    private readonly string _nodeId;
+
+    internal HybridLogicalClock Clock => _clock;
+    internal string NodeId => _nodeId;
+
+    public SqliteLocalDb(LocalDbOptions options)
+    {
+        _options = options;
+        _connectionString = new SqliteConnectionStringBuilder { DataSource = options.Path }.ToString();
+
+        _master = new SqliteConnection(_connectionString);
+        _master.Open();
+        ExecuteRaw(_master,
+            $"PRAGMA journal_mode=WAL; PRAGMA synchronous={_options.Synchronous}; " +
+            $"PRAGMA busy_timeout={_options.BusyTimeoutMs}; PRAGMA foreign_keys=ON;");
+
+        LocalDbSchema.EnsureCreated(_master);
+
+        var initial = ScalarLong(_master,
+            "SELECT MAX(" +
+            "(SELECT hlc_high_water FROM __localdb_meta WHERE id=1)," +
+            "COALESCE((SELECT MAX(hlc) FROM __localdb_oplog),0)," +
+            "COALESCE((SELECT MAX(hlc) FROM __localdb_row_version),0))");
+        _clock = new HybridLogicalClock(initial);
+
+        (_nodeId, _) = LocalDbSchema.ReadMeta(_master);
+    }
+
+    public SqliteConnection CreateConnection()
+    {
+        var conn = new SqliteConnection(_connectionString);
+        conn.Open();
+        // journal_mode is a durable DB property already set on the file; synchronous is per-connection.
+        ExecuteRaw(conn,
+            $"PRAGMA synchronous={_options.Synchronous}; " +
+            $"PRAGMA busy_timeout={_options.BusyTimeoutMs}; PRAGMA foreign_keys=ON;");
+        conn.CreateFunction("zb_hlc_next", () => _clock.Next());
+        return conn;
+    }
+
+    public async Task<int> ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default)
+    {
+        await using var conn = CreateConnection();
+        await using var cmd = conn.CreateCommand();
+        cmd.CommandText = sql;
+        Bind(cmd, parameters);
+        return await cmd.ExecuteNonQueryAsync(ct);
+    }
+
+    public async Task<IReadOnlyList<T>> QueryAsync<T>(string sql, Func<SqliteDataReader, T> map, object? parameters = null, CancellationToken ct = default)
+    {
+        await using var conn = CreateConnection();
+        await using var cmd = conn.CreateCommand();
+        cmd.CommandText = sql;
+        Bind(cmd, parameters);
+        return await ReadAllAsync(cmd, map, ct);
+    }
+
+    public async Task<ILocalDbTransaction> BeginTransactionAsync(CancellationToken ct = default)
+    {
+        var conn = CreateConnection();
+        try
+        {
+            var tx = (SqliteTransaction)await conn.BeginTransactionAsync(ct);
+            return new Transaction(conn, tx);
+        }
+        catch
+        {
+            await conn.DisposeAsync();
+            throw;
+        }
+    }
+
+    public void Dispose()
+    {
+        LocalDbSchema.FlushHlcHighWater(_master, _clock.Current);
+        _master.Dispose();
+    }
+
+    private static async Task<IReadOnlyList<T>> ReadAllAsync<T>(SqliteCommand cmd, Func<SqliteDataReader, T> map, CancellationToken ct)
+    {
+        var results = new List<T>();
+        await using var reader = await cmd.ExecuteReaderAsync(ct);
+        while (await reader.ReadAsync(ct))
+            results.Add(map((SqliteDataReader)reader));
+        return results;
+    }
+
+    private static void Bind(SqliteCommand cmd, object? parameters)
+    {
+        if (parameters is null)
+            return;
+
+        foreach (var prop in PropertiesOf(parameters.GetType()))
+        {
+            var value = prop.GetValue(parameters);
+            cmd.Parameters.AddWithValue("@" + prop.Name, value ?? DBNull.Value);
+        }
+    }
+
+    private static PropertyInfo[] PropertiesOf(Type type) =>
+        PropertyCache.GetOrAdd(type, static t => t.GetProperties(BindingFlags.Public | BindingFlags.Instance));
+
+    private static void ExecuteRaw(SqliteConnection conn, string sql)
+    {
+        using var cmd = conn.CreateCommand();
+        cmd.CommandText = sql;
+        cmd.ExecuteNonQuery();
+    }
+
+    private static long ScalarLong(SqliteConnection conn, string sql)
+    {
+        using var cmd = conn.CreateCommand();
+        cmd.CommandText = sql;
+        return Convert.ToInt64(cmd.ExecuteScalar(), CultureInfo.InvariantCulture);
+    }
+
+    private sealed class Transaction(SqliteConnection connection, SqliteTransaction transaction) : ILocalDbTransaction
+    {
+        public async Task<int> ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default)
+        {
+            await using var cmd = connection.CreateCommand();
+            cmd.Transaction = transaction;
+            cmd.CommandText = sql;
+            Bind(cmd, parameters);
+            return await cmd.ExecuteNonQueryAsync(ct);
+        }
+
+        public async Task<IReadOnlyList<T>> QueryAsync<T>(string sql, Func<SqliteDataReader, T> map, object? parameters = null, CancellationToken ct = default)
+        {
+            await using var cmd = connection.CreateCommand();
+            cmd.Transaction = transaction;
+            cmd.CommandText = sql;
+            Bind(cmd, parameters);
+            return await ReadAllAsync(cmd, map, ct);
+        }
+
+        public Task CommitAsync(CancellationToken ct = default) => transaction.CommitAsync(ct);
+
+        public Task RollbackAsync(CancellationToken ct = default) => transaction.RollbackAsync(ct);
+
+        public async ValueTask DisposeAsync()
+        {
+            await transaction.DisposeAsync();
+            await connection.DisposeAsync();
+        }
+    }
+}

ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/LocalDbOptions.cs

@@ -0,0 +1,14 @@
+namespace ZB.MOM.WW.LocalDb;
+
+/// <summary>Host configuration for the embedded SQLite local database.</summary>
+public sealed class LocalDbOptions
+{
+    /// <summary>Filesystem path to the SQLite database file.</summary>
+    public string Path { get; set; } = "";
+
+    /// <summary>SQLite <c>busy_timeout</c> pragma, in milliseconds.</summary>
+    public int BusyTimeoutMs { get; set; } = 5000;
+
+    /// <summary>SQLite <c>synchronous</c> pragma level (e.g. OFF / NORMAL / FULL).</summary>
+    public string Synchronous { get; set; } = "NORMAL";
+}

ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs

@@ -0,0 +1,118 @@
+using Microsoft.Data.Sqlite;
+using ZB.MOM.WW.LocalDb.Internal;
+
+namespace ZB.MOM.WW.LocalDb.Tests;
+
+public sealed class SqliteLocalDbTests : 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 });
+
+    [Fact]
+    public void Open_CreatesDbFile_WalMode()
+    {
+        using var db = Create();
+
+        Assert.True(File.Exists(_path));
+
+        using var conn = db.CreateConnection();
+        using var cmd = conn.CreateCommand();
+        cmd.CommandText = "PRAGMA journal_mode";
+        var mode = (string)cmd.ExecuteScalar()!;
+        Assert.Equal("wal", mode, ignoreCase: true);
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_ParameterizedInsert_Works()
+    {
+        using var db = Create();
+
+        await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)");
+        await db.ExecuteAsync("INSERT INTO t (id, name) VALUES (@id, @name)", new { id = 1, name = "x" });
+
+        var rows = await db.QueryAsync("SELECT COUNT(*) FROM t", r => r.GetInt64(0));
+        Assert.Equal(1L, rows[0]);
+    }
+
+    [Fact]
+    public async Task QueryAsync_MapsRows()
+    {
+        using var db = Create();
+
+        await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)");
+        await db.ExecuteAsync("INSERT INTO t (id, name) VALUES (@id, @name)", new { id = 1, name = "a" });
+        await db.ExecuteAsync("INSERT INTO t (id, name) VALUES (@id, @name)", new { id = 2, name = "b" });
+
+        var rows = await db.QueryAsync(
+            "SELECT id, name FROM t ORDER BY id",
+            r => (Id: r.GetInt64(0), Name: r.GetString(1)));
+
+        Assert.Equal(2, rows.Count);
+        Assert.Equal((1L, "a"), rows[0]);
+        Assert.Equal((2L, "b"), rows[1]);
+    }
+
+    [Fact]
+    public async Task Transaction_CommitAndRollback()
+    {
+        using var db = Create();
+        await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY)");
+
+        await using (var tx = await db.BeginTransactionAsync())
+        {
+            await tx.ExecuteAsync("INSERT INTO t (id) VALUES (@id)", new { id = 1 });
+            await tx.CommitAsync(default);
+        }
+
+        await using (var tx = await db.BeginTransactionAsync())
+        {
+            await tx.ExecuteAsync("INSERT INTO t (id) VALUES (@id)", new { id = 2 });
+            await tx.RollbackAsync(default);
+        }
+
+        var ids = await db.QueryAsync("SELECT id FROM t ORDER BY id", r => r.GetInt64(0));
+        Assert.Equal(new[] { 1L }, ids);
+    }
+
+    [Fact]
+    public async Task HlcUdf_RegisteredOnConnection()
+    {
+        using var db = Create();
+
+        var values = await db.QueryAsync(
+            "SELECT zb_hlc_next() UNION ALL SELECT zb_hlc_next()",
+            r => r.GetInt64(0));
+
+        Assert.Equal(2, values.Count);
+        Assert.True(values[1] > values[0], $"{values[1]} should be greater than {values[0]}");
+    }
+
+    [Fact]
+    public async Task Hlc_ResumesFromDurableMax_OnReopen()
+    {
+        const long huge = 9_000_000_000_000_000L;
+
+        using (var db = Create())
+        {
+            await db.ExecuteAsync(
+                "INSERT INTO __localdb_oplog (table_name, pk_json, hlc, node_id) VALUES ('t', '{}', @hlc, 'n')",
+                new { hlc = huge });
+        }
+
+        SqliteConnection.ClearAllPools();
+
+        using (var db = Create())
+        {
+            var next = await db.QueryAsync("SELECT zb_hlc_next()", r => r.GetInt64(0));
+            Assert.True(next[0] > huge, $"{next[0]} should be greater than {huge}");
+        }
+    }
+}