fix(localdb): Task 4 review fixes — synchronous allow-list, dictionary params, async conn setup, dispose guard

2026-07-17 21:11:07 -04:00
parent 827bf44684
commit 12b8839f33
3 changed files with 148 additions and 20 deletions
@@ -6,13 +6,24 @@ namespace ZB.MOM.WW.LocalDb;
 /// 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.
+/// SQL references parameters with the <c>@name</c> marker; supported parameter shapes are
+/// an anonymous object / POCO (public instance properties bind as <c>@PropertyName</c>) or an
+/// <c>IDictionary&lt;string, object?&gt;</c> (each entry binds as <c>@Key</c>). Null values bind as SQL NULL.
 /// </summary>
 public interface ILocalDb
 {
-    /// <summary>Executes a non-query statement; returns the affected row count.</summary>
+    /// <summary>
+    /// Executes a non-query statement; returns the affected row count.
+    /// <paramref name="parameters"/> may be an anonymous object / POCO (properties → <c>@name</c>)
+    /// or an <c>IDictionary&lt;string, object?&gt;</c> (entries → <c>@key</c>).
+    /// </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>
+    /// <summary>
+    /// Executes a query and projects each row through <paramref name="map"/>.
+    /// <paramref name="parameters"/> may be an anonymous object / POCO (properties → <c>@name</c>)
+    /// or an <c>IDictionary&lt;string, object?&gt;</c> (entries → <c>@key</c>).
+    /// </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>
@@ -12,27 +12,40 @@ namespace ZB.MOM.WW.LocalDb.Internal;
 /// </summary>
 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"];
+
    private static readonly ConcurrentDictionary<Type, PropertyInfo[]> PropertyCache = new();

-    private readonly LocalDbOptions _options;
+    private readonly string _synchronous;
+    private readonly int _busyTimeoutMs;
    private readonly string _connectionString;
+    // Held open for the object lifetime: it anchors the WAL journal and is the connection
+    // Dispose flushes the HLC high-water mark through, so the flush cannot fail on reopen.
    private readonly SqliteConnection _master;
    private readonly HybridLogicalClock _clock;
    private readonly string _nodeId;
+    private bool _disposed;

    internal HybridLogicalClock Clock => _clock;
    internal string NodeId => _nodeId;

    public SqliteLocalDb(LocalDbOptions options)
    {
-        _options = options;
+        if (string.IsNullOrWhiteSpace(options.Path))
+            throw new ArgumentException("LocalDbOptions.Path is required.", nameof(options));
+
+        _synchronous = Array.Find(AllowedSynchronous,
+                s => s.Equals(options.Synchronous, StringComparison.OrdinalIgnoreCase))
+            ?? throw new ArgumentException(
+                $"LocalDbOptions.Synchronous must be one of {string.Join("/", AllowedSynchronous)}; got '{options.Synchronous}'.",
+                nameof(options));
+        _busyTimeoutMs = options.BusyTimeoutMs;
        _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;");
+        ExecuteRaw(_master, "PRAGMA journal_mode=WAL; " + PerConnectionPragmas());

        LocalDbSchema.EnsureCreated(_master);

@@ -46,21 +59,43 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable
        (_nodeId, _) = LocalDbSchema.ReadMeta(_master);
    }

+    // journal_mode is a durable DB property already set on the file; synchronous is per-connection.
+    private string PerConnectionPragmas() =>
+        $"PRAGMA synchronous={_synchronous}; PRAGMA busy_timeout={_busyTimeoutMs}; PRAGMA foreign_keys=ON;";
+
    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;");
+        ExecuteRaw(conn, PerConnectionPragmas());
        conn.CreateFunction("zb_hlc_next", () => _clock.Next());
        return conn;
    }

+    internal async Task<SqliteConnection> CreateConnectionAsync(CancellationToken ct)
+    {
+        var conn = new SqliteConnection(_connectionString);
+        try
+        {
+            await conn.OpenAsync(ct);
+            await using (var cmd = conn.CreateCommand())
+            {
+                cmd.CommandText = PerConnectionPragmas();
+                await cmd.ExecuteNonQueryAsync(ct);
+            }
+            conn.CreateFunction("zb_hlc_next", () => _clock.Next());
+            return conn;
+        }
+        catch
+        {
+            await conn.DisposeAsync();
+            throw;
+        }
+    }
+
    public async Task<int> ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default)
    {
-        await using var conn = CreateConnection();
+        await using var conn = await CreateConnectionAsync(ct);
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = sql;
        Bind(cmd, parameters);
@@ -69,7 +104,7 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable

    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 conn = await CreateConnectionAsync(ct);
        await using var cmd = conn.CreateCommand();
        cmd.CommandText = sql;
        Bind(cmd, parameters);
@@ -78,7 +113,7 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable

    public async Task<ILocalDbTransaction> BeginTransactionAsync(CancellationToken ct = default)
    {
-        var conn = CreateConnection();
+        var conn = await CreateConnectionAsync(ct);
        try
        {
            var tx = (SqliteTransaction)await conn.BeginTransactionAsync(ct);
@@ -93,6 +128,9 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable

    public void Dispose()
    {
+        if (_disposed)
+            return;
+        _disposed = true;
        LocalDbSchema.FlushHlcHighWater(_master, _clock.Current);
        _master.Dispose();
    }
@@ -108,13 +146,18 @@ internal sealed class SqliteLocalDb : ILocalDb, IDisposable

    private static void Bind(SqliteCommand cmd, object? parameters)
    {
-        if (parameters is null)
-            return;
-
-        foreach (var prop in PropertiesOf(parameters.GetType()))
+        switch (parameters)
        {
-            var value = prop.GetValue(parameters);
-            cmd.Parameters.AddWithValue("@" + prop.Name, value ?? DBNull.Value);
+            case null:
+                return;
+            case IDictionary<string, object?> dict:
+                foreach (var (key, value) in dict)
+                    cmd.Parameters.AddWithValue("@" + key, value ?? DBNull.Value);
+                return;
+            default:
+                foreach (var prop in PropertiesOf(parameters.GetType()))
+                    cmd.Parameters.AddWithValue("@" + prop.Name, prop.GetValue(parameters) ?? DBNull.Value);
+                return;
        }
    }

@@ -82,6 +82,80 @@ public sealed class SqliteLocalDbTests : IDisposable
        Assert.Equal(new[] { 1L }, ids);
    }

+    [Fact]
+    public async Task Transaction_DisposeWithoutCommit_RollsBack()
+    {
+        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 });
+        }
+
+        var count = await db.QueryAsync("SELECT COUNT(*) FROM t", r => r.GetInt64(0));
+        Assert.Equal(0L, count[0]);
+    }
+
+    [Fact]
+    public async Task Bind_GuidByteArrayAndNull_RoundTrip()
+    {
+        using var db = Create();
+        await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY, g TEXT, blob BLOB, missing TEXT)");
+
+        var guid = Guid.NewGuid();
+        var bytes = new byte[] { 1, 2, 3 };
+        await db.ExecuteAsync(
+            "INSERT INTO t (id, g, blob, missing) VALUES (@id, @g, @blob, @missing)",
+            new { id = 1, g = guid, blob = bytes, missing = (string?)null });
+
+        var rows = await db.QueryAsync(
+            "SELECT g, blob, missing FROM t WHERE id = 1",
+            r => (Guid: r.GetGuid(0), Blob: (byte[])r.GetValue(1), MissingIsNull: r.IsDBNull(2)));
+
+        Assert.Equal(guid, rows[0].Guid);
+        Assert.Equal(bytes, rows[0].Blob);
+        Assert.True(rows[0].MissingIsNull);
+    }
+
+    [Fact]
+    public async Task Bind_DictionaryParameters_BindByKey()
+    {
+        using var db = Create();
+        await db.ExecuteAsync("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)");
+
+        var parameters = new Dictionary<string, object?> { ["id"] = 7, ["name"] = null };
+        await db.ExecuteAsync("INSERT INTO t (id, name) VALUES (@id, @name)", parameters);
+
+        var rows = await db.QueryAsync(
+            "SELECT id, name FROM t",
+            r => (Id: r.GetInt64(0), NameIsNull: r.IsDBNull(1)));
+
+        Assert.Equal(7L, rows[0].Id);
+        Assert.True(rows[0].NameIsNull);
+    }
+
+    [Fact]
+    public void Ctor_EmptyPath_Throws()
+    {
+        Assert.Throws<ArgumentException>(() => new SqliteLocalDb(new LocalDbOptions { Path = " " }));
+    }
+
+    [Fact]
+    public void Ctor_InvalidSynchronous_Throws()
+    {
+        Assert.Throws<ArgumentException>(() =>
+            new SqliteLocalDb(new LocalDbOptions { Path = _path, Synchronous = "NORMAL; PRAGMA foreign_keys=OFF" }));
+    }
+
+    [Fact]
+    public void Dispose_IsIdempotent()
+    {
+        var db = Create();
+        db.Dispose();
+        db.Dispose();
+    }
+
    [Fact]
    public async Task HlcUdf_RegisteredOnConnection()
    {