From 83f384247cb6dd090fa4dc23a04540cc678082f9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 17 Jul 2026 21:03:39 -0400 Subject: [PATCH] feat(localdb): SqliteLocalDb host (WAL, UDF-registered connections, HLC recovery, consumer SQL API) --- .../src/ZB.MOM.WW.LocalDb/ILocalDb.cs | 39 ++++ .../Internal/SqliteLocalDb.cs | 168 ++++++++++++++++++ .../src/ZB.MOM.WW.LocalDb/LocalDbOptions.cs | 14 ++ .../SqliteLocalDbTests.cs | 118 ++++++++++++ 4 files changed, 339 insertions(+) create mode 100644 ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs create mode 100644 ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/Internal/SqliteLocalDb.cs create mode 100644 ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/LocalDbOptions.cs create mode 100644 ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SqliteLocalDbTests.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 new file mode 100644 index 0000000..f8a9466 --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/ILocalDb.cs @@ -0,0 +1,39 @@ +using Microsoft.Data.Sqlite; + +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. +/// +public interface ILocalDb +{ + /// Executes a non-query statement; returns the affected row count. + Task ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default); + + /// Executes a query and projects each row through . + Task> QueryAsync(string sql, Func map, object? parameters = null, CancellationToken ct = default); + + /// Begins a transaction pinned to a single UDF-registered connection. + Task BeginTransactionAsync(CancellationToken ct = default); + + /// Escape hatch: an open, UDF-registered connection the caller owns and disposes. + SqliteConnection CreateConnection(); +} + +/// A transaction scope over the local database. Dispose without commit rolls back. +public interface ILocalDbTransaction : IAsyncDisposable +{ + /// Executes a non-query statement inside the transaction. + Task ExecuteAsync(string sql, object? parameters = null, CancellationToken ct = default); + + /// Executes a query inside the transaction. + Task> QueryAsync(string sql, Func map, object? parameters = null, CancellationToken ct = default); + + /// Commits the transaction. + Task CommitAsync(CancellationToken ct = default); + + /// Rolls the transaction back. + Task RollbackAsync(CancellationToken ct = default); +} 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 new file mode 100644 index 0000000..fc5084b --- /dev/null +++ b/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; + +/// +/// SQLite-backed host: opens/creates the file in WAL mode, ensures the +/// library schema, recovers the HLC from durable state, and hands out UDF-registered connections. +/// +internal sealed class SqliteLocalDb : ILocalDb, IDisposable +{ + private static readonly ConcurrentDictionary 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 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> QueryAsync(string sql, Func 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 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> ReadAllAsync(SqliteCommand cmd, Func map, CancellationToken ct) + { + var results = new List(); + 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 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> QueryAsync(string sql, Func 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(); + } + } +} diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/LocalDbOptions.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/LocalDbOptions.cs new file mode 100644 index 0000000..daf270c --- /dev/null +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb/LocalDbOptions.cs @@ -0,0 +1,14 @@ +namespace ZB.MOM.WW.LocalDb; + +/// Host configuration for the embedded SQLite local database. +public sealed class LocalDbOptions +{ + /// Filesystem path to the SQLite database file. + public string Path { get; set; } = ""; + + /// SQLite busy_timeout pragma, in milliseconds. + public int BusyTimeoutMs { get; set; } = 5000; + + /// SQLite synchronous pragma level (e.g. OFF / NORMAL / FULL). + public string Synchronous { get; set; } = "NORMAL"; +} 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 new file mode 100644 index 0000000..fbec825 --- /dev/null +++ b/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}"); + } + } +}