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 f8a9466..d864328 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
@@ -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 zb_hlc_next() UDF registered, so capture triggers on
/// registered tables can stamp writes fail-closed.
+/// SQL references parameters with the @name marker; supported parameter shapes are
+/// an anonymous object / POCO (public instance properties bind as @PropertyName) or an
+/// IDictionary<string, object?> (each entry binds as @Key). Null values bind as SQL NULL.
///
public interface ILocalDb
{
- /// Executes a non-query statement; returns the affected row count.
+ ///
+ /// Executes a non-query statement; returns the affected row count.
+ /// may be an anonymous object / POCO (properties → @name)
+ /// or an IDictionary<string, object?> (entries → @key).
+ ///
Task ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default);
- /// Executes a query and projects each row through .
+ ///
+ /// Executes a query and projects each row through .
+ /// may be an anonymous object / POCO (properties → @name)
+ /// or an IDictionary<string, object?> (entries → @key).
+ ///
Task> QueryAsync(string sql, Func map, object? parameters = null, CancellationToken ct = default);
/// Begins a transaction pinned to a single UDF-registered connection.
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 fc5084b..6efdfd7 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
@@ -12,27 +12,40 @@ namespace ZB.MOM.WW.LocalDb.Internal;
///
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 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 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 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> QueryAsync(string sql, Func 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 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 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;
}
}
diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs
index fbec825..572ac1d 100644
--- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs
+++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.cs
@@ -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 { ["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(() => new SqliteLocalDb(new LocalDbOptions { Path = " " }));
+ }
+
+ [Fact]
+ public void Ctor_InvalidSynchronous_Throws()
+ {
+ Assert.Throws(() =>
+ 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()
{