dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(sf,site): both stores take ILocalDb instead of a connection string

Browse Source 
Tasks 5 and 6 of the Phase 2 plan, committed together because their test
fallout is entangled — several fixtures construct both stores.

StoreAndForwardStorage and SiteStorageService now take ILocalDb. Connections
come from ILocalDb.CreateConnection(), which hands out an already-open,
pragma-configured connection carrying the zb_hlc_next() UDF the capture triggers
call; a raw connection would lack the UDF and every write to a replicated table
would fail closed. Deleted with the connection strings: S&F's
EnsureDatabaseDirectoryExists and its per-open busy_timeout pragma, and the site
service's BusyTimeoutFloorSeconds normalization — LocalDb owns all of it now.

DI: AddSiteRuntime's string overload is gone (nothing left to supply), so the
Host calls the no-arg form. ScadaBridge:Database:SiteDbPath and
StoreAndForwardOptions.SqliteDbPath survive only as the migrator's source
locations in Tasks 8/9.

Two things the plan did not anticipate, both worth reading:

1. FOUND A REAL LATENT DEFECT, from Phase 1, now fixed. The plan assumed
   directory creation simply moved to LocalDb along with file ownership. It did
   not: the LocalDb library never creates the parent directory, and
   SqliteLocalDb opens the file eagerly in its constructor — so a missing
   directory is a hard boot failure ("SQLite Error 14: unable to open database
   file"), not a degraded start. The default site config points at the RELATIVE
   path ./data/site-localdb.db, so any site node without a pre-existing data/
   directory fails to boot. The docker rig escapes only because its volume mount
   happens to create /app/data — a coincidence that would have hidden this until
   a bare-metal or fresh deployment. This has been latent since Phase 1 made
   LocalDb:Path required; deleting S&F's EnsureDatabaseDirectoryExists here
   would have widened it. Re-established the guarantee at the layer that now
   owns the path (SiteLocalDbDirectory.Ensure, called before AddZbLocalDb) and
   pinned it with SiteLocalDbDirectoryTests. Non-vacuity is not assumed: two
   tests written against the wrong assumption failed with exactly this
   SQLite Error 14 before the fix existed.

2. Test fallout was ~7x the plan's estimate. The plan named "fixtures" in one
   project; the constructor change actually reaches 40 files across 7 test
   projects, and most used Mode=Memory;Cache=Shared — which LocalDb has no
   equivalent for, so every one had to move to a real temp file. Rather than
   copy the Phase 1 TestLocalDb fixture into 7 projects, added a shared
   tests/ZB.MOM.WW.ScadaBridge.TestSupport library (not a test project) so the
   WAL-sidecar cleanup and the "real, not stubbed" rationale live in one place.

Retargeted rather than deleted, in both directions: the S&F WAL test now asserts
against the LocalDb-backed store (WAL genuinely is LocalDb's job), while the
directory-creation test moved to Host.Tests (that guarantee is NOT LocalDb's).
SiteStorageServiceTests.Initialize_EnablesWalJournalMode got the same treatment.
DeploymentManagerMediumFindingsTests induced a persistence failure via an
unopenable path, which no longer reaches the assertion since the fixture now
throws first; it induces the same failure shape via an uninitialized store.

Verified: full solution build 0 warnings; SiteRuntime 532, Host 318,
AuditLog 355, ExternalSystemGateway 142, HealthMonitoring 97,
StoreAndForward 153 — 1597 passed, 0 failed.

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
Joseph Doherty
2026-07-20 03:05:45 -04:00
parent 3dfb288b74
commit f2efeb37b7
57 changed files with 1368 additions and 848 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Host/SiteLocalDbDirectory.cs
+59
View File
@@ -0,0 +1,59 @@
using Microsoft.Extensions.Configuration;
namespace ZB.MOM.WW.ScadaBridge.Host;
/// <summary>
/// Ensures the directory holding the consolidated site database (<c>LocalDb:Path</c>)
/// exists before LocalDb opens the file.
/// </summary>
/// <remarks>
/// <para>
/// SQLite creates the database file on demand but never its parent directory, and
/// <c>SqliteLocalDb</c> opens the file eagerly in its constructor — so a missing directory
/// is a hard boot failure ("SQLite Error 14: unable to open database file"), not a degraded
/// start. Neither the LocalDb library nor its options validation does this.
/// </para>
/// <para>
/// The guarantee previously lived in <c>StoreAndForwardStorage.EnsureDatabaseDirectoryExists</c>,
/// which created the directory for the store's own SQLite file. Once that file was folded
/// into the consolidated database the store stopped owning a path, so the guarantee had to
/// be re-established at the layer that now configures the path: the site host.
/// </para>
/// <para>
/// It matters because the default site configuration uses the <i>relative</i> path
/// <c>./data/site-localdb.db</c>. The docker rig never noticed the gap only because its
/// volume mount happens to create <c>/app/data</c>.
/// </para>
/// </remarks>
public static class SiteLocalDbDirectory
{
/// <summary>
/// Creates the directory containing <c>LocalDb:Path</c> when it does not already exist.
/// No-op when the key is unset.
/// </summary>
/// <remarks>
/// Best-effort by design: if the directory cannot be created (permissions, read-only
/// mount) this stays silent and lets LocalDb raise the real error, which names the
/// actual path and is a better diagnostic than anything thrown from here.
/// </remarks>
/// <param name="config">Configuration carrying <c>LocalDb:Path</c>.</param>
public static void Ensure(IConfiguration config)
{
ArgumentNullException.ThrowIfNull(config);
var path = config["LocalDb:Path"];
if (string.IsNullOrWhiteSpace(path))
return;
try
{
var directory = Path.GetDirectoryName(Path.GetFullPath(path));
if (!string.IsNullOrEmpty(directory) && !Directory.Exists(directory))
Directory.CreateDirectory(directory);
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or ArgumentException)
{
// Deliberately swallowed — see the remarks above.
}
}
}
src/ZB.MOM.WW.ScadaBridge.Host/SiteServiceRegistration.cs
+19 -4
View File
@@ -65,10 +65,13 @@ public static class SiteServiceRegistration
services.AddSingleton<ISiteIdentityProvider, SiteIdentityProvider>();
services.AddSingleton<IHealthReportTransport, AkkaHealthReportTransport>();
// Site-only components — AddSiteRuntime registers SiteStorageService with SQLite path
// and site-local repository implementations (IExternalSystemRepository, INotificationRepository)
var siteDbPath = config["ScadaBridge:Database:SiteDbPath"] ?? "site.db";
services.AddSiteRuntime($"Data Source={siteDbPath}");
// Site-only components — AddSiteRuntime registers SiteStorageService and the
// site-local repository implementations (IExternalSystemRepository,
// INotificationRepository). It takes no connection string any more:
// SiteStorageService persists to the consolidated LocalDb database registered
// just below (LocalDb:Path). ScadaBridge:Database:SiteDbPath survives only as
// the legacy migrator's source location.
services.AddSiteRuntime();
// Consolidated site database (LocalDb Phase 1). Holds OperationTracking and
// site_events as replicated tables so the pair stops losing them on failover.
@@ -78,6 +81,18 @@ public static class SiteServiceRegistration
// initiator idles.
//
// Design: scadaproj docs/plans/2026-07-19-scadabridge-localdb-design.md
//
// Create the parent directory FIRST. SQLite creates the database file on demand
// but not its directory, and neither the LocalDb library nor its options
// validation does this — SqliteLocalDb's constructor opens the file eagerly, so a
// missing directory is a hard boot failure ("SQLite Error 14: unable to open
// database file"), not a degraded start. The default site path is the relative
// "./data/site-localdb.db", so this bites any site node whose data directory has
// not been pre-created; the docker rig only escapes it because the volume mount
// creates /app/data. StoreAndForwardStorage used to do this for its own file
// (EnsureDatabaseDirectoryExists) and that guarantee has to keep existing
// somewhere now that LocalDb owns the file.
SiteLocalDbDirectory.Ensure(config);
services.AddZbLocalDb(config, db => SiteLocalDbSetup.OnReady(db, config));
// The replication engine, likewise unconditional but INERT by default: with no
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Persistence/SiteStorageService.cs
+57 -86
View File
@@ -1,5 +1,6 @@
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.LocalDb;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
@@ -10,70 +11,59 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
/// </summary>
public class SiteStorageService
{
private readonly string _connectionString;
private readonly ILocalDb _localDb;
private readonly ILogger<SiteStorageService> _logger;
/// <summary>
/// Initializes a new instance of the SiteStorageService with the specified SQLite connection string and logger.
/// Initializes the service over the consolidated site database.
/// </summary>
/// <param name="connectionString">SQLite connection string for the site database.</param>
/// <param name="localDb">
/// The consolidated site database. Every connection it hands out is already open and carries
/// the per-connection pragmas (including the busy timeout that this class used to pin itself)
/// plus the <c>zb_hlc_next()</c> UDF that the config tables' capture triggers call — which is
/// exactly why the service no longer builds its own connection string. A raw connection would
/// lack the UDF and every write to a replicated table would fail closed.
/// </param>
/// <param name="logger">Logger instance for diagnostic messages.</param>
public SiteStorageService(string connectionString, ILogger<SiteStorageService> logger)
public SiteStorageService(ILocalDb localDb, ILogger<SiteStorageService> logger)
{
// Normalize the connection string and pin a busy-timeout floor (S8). WAL lets a reader
// and a writer proceed concurrently, but two writers still contend; Microsoft.Data.Sqlite
// drives its busy handler off the command timeout, so a floor of 5 s means a briefly
// busy database is waited-on rather than failing fast with SQLITE_BUSY. Only raise a
// caller-supplied value that is lower than the floor (the library default of 30 s stays).
var builder = new SqliteConnectionStringBuilder(connectionString);
if (builder.DefaultTimeout < BusyTimeoutFloorSeconds)
builder.DefaultTimeout = BusyTimeoutFloorSeconds;
_connectionString = builder.ToString();
ArgumentNullException.ThrowIfNull(localDb);
ArgumentNullException.ThrowIfNull(logger);
_localDb = localDb;
_logger = logger;
}
/// <summary>Busy-timeout floor in seconds applied to the site connection string (S8).</summary>
private const int BusyTimeoutFloorSeconds = 5;
/// <summary>
/// Returns an <b>already-open</b> connection against the site database.
/// Exposed so site-local repositories can get their own connections without reaching
/// into private state via reflection. The caller owns the connection and is
/// responsible for disposing it.
/// </summary>
/// <remarks>
/// <b>Contract change:</b> this used to return an <i>unopened</i> connection that the
/// caller opened. LocalDb hands out connections already open, pragma-configured, and
/// carrying the <c>zb_hlc_next()</c> UDF — calling <c>Open</c>/<c>OpenAsync</c> on one
/// throws. Callers must not open it.
/// </remarks>
/// <returns>An open <see cref="SqliteConnection"/> against the site database.</returns>
public SqliteConnection CreateConnection() => _localDb.CreateConnection();
/// <summary>
/// Creates a new (unopened) SQLite connection against the site database.
/// Exposed so site-local repositories can open their own connections without
/// reaching into private state via reflection. The caller owns
/// the connection and is responsible for opening and disposing it.
/// <para>
/// The database runs in WAL journal mode (set once in <see cref="InitializeAsync"/>) with a
/// busy-timeout floor (see the constructor), so connections handed out here inherit
/// concurrent-reader/writer behavior and wait out a briefly-busy database instead of
/// failing with SQLITE_BUSY.
/// </para>
/// Convenience alias used internally, mirroring the other LocalDb-backed stores.
/// </summary>
/// <returns>A new, unopened <see cref="SqliteConnection"/> against the site database.</returns>
public SqliteConnection CreateConnection() => new(_connectionString);
private SqliteConnection OpenConnection() => _localDb.CreateConnection();
/// <summary>
/// Creates the SQLite tables if they do not exist.
/// Called once on site startup.
/// </summary>
/// <returns>A task that completes when all tables have been created or verified.</returns>
public async Task InitializeAsync()
public Task InitializeAsync()
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
// Switch to WAL journal mode (S8). WAL is persistent (database-level, survives across
// connections) so this one-time set is enough. It lets readers and a writer proceed
// concurrently instead of every access serializing on the rollback journal. A
// :memory: database or a filesystem that cannot support WAL falls back to its prior
// mode — we log the result rather than throwing.
await using (var walCommand = connection.CreateCommand())
{
walCommand.CommandText = "PRAGMA journal_mode=WAL;";
var resultingMode = (await walCommand.ExecuteScalarAsync()) as string ?? "unknown";
if (!string.Equals(resultingMode, "wal", StringComparison.OrdinalIgnoreCase))
_logger.LogWarning(
"Site SQLite could not enable WAL journal mode (got '{Mode}') — concurrent access may serialize",
resultingMode);
}
// No journal-mode pragma and no busy-timeout normalization here any more:
// LocalDb owns the file and sets WAL plus the per-connection pragmas on every
// connection it hands out.
using var connection = OpenConnection();
// The DDL itself lives in SiteStorageSchema so the Host can apply it to a
// LocalDb-managed connection before RegisterReplicated installs the capture
@@ -81,7 +71,8 @@ public class SiteStorageService
// (tests, tooling) remains self-sufficient.
SiteStorageSchema.Apply(connection);
_logger.LogInformation("Site SQLite storage initialized at {ConnectionString}", _connectionString);
_logger.LogInformation("Site SQLite storage initialized");
return Task.CompletedTask;
}
// ── Deployed Configuration CRUD ──
@@ -92,8 +83,7 @@ public class SiteStorageService
/// <returns>A task that resolves to the list of all deployed instance configurations.</returns>
public async Task<List<DeployedInstance>> GetAllDeployedConfigsAsync()
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -134,8 +124,7 @@ public class SiteStorageService
string revisionHash,
bool isEnabled)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -197,8 +186,7 @@ public class SiteStorageService
{
var deployedAt = (deployedAtOverride ?? DateTimeOffset.UtcNow).ToString("O");
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -231,8 +219,7 @@ public class SiteStorageService
/// <returns>A task that completes when the configuration and its overrides have been removed.</returns>
public async Task RemoveDeployedConfigAsync(string instanceName)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var transaction = await connection.BeginTransactionAsync();
@@ -272,8 +259,7 @@ public class SiteStorageService
/// <returns>A task that completes when the enabled flag has been updated.</returns>
public async Task SetInstanceEnabledAsync(string instanceName, bool isEnabled)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -300,8 +286,7 @@ public class SiteStorageService
/// <returns>A task that resolves to a dictionary mapping attribute names to their override values.</returns>
public async Task<Dictionary<string, string>> GetStaticOverridesAsync(string instanceName)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -329,8 +314,7 @@ public class SiteStorageService
/// <returns>A task that completes when the override has been saved.</returns>
public async Task SetStaticOverrideAsync(string instanceName, string attributeName, string value)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -356,8 +340,7 @@ public class SiteStorageService
/// <returns>A task that completes when all overrides for the instance have been deleted.</returns>
public async Task ClearStaticOverridesAsync(string instanceName)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = "DELETE FROM static_attribute_overrides WHERE instance_unique_name = @name";
@@ -384,8 +367,7 @@ public class SiteStorageService
string instanceName, string sourceCanonicalName, string sourceReference,
string conditionJson, DateTimeOffset lastTransitionAt, string? metadataJson = null)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -424,8 +406,7 @@ public class SiteStorageService
if (rows.Count == 0)
return;
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
await using var command = connection.CreateCommand();
@@ -469,8 +450,7 @@ public class SiteStorageService
/// <returns>A task that completes when the alarm condition row has been deleted.</returns>
public async Task DeleteNativeAlarmAsync(string instanceName, string sourceCanonicalName, string sourceReference)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -494,8 +474,7 @@ public class SiteStorageService
/// <returns>A task that resolves to the list of stored native alarm condition rows for the binding.</returns>
public async Task<IReadOnlyList<NativeAlarmRow>> GetNativeAlarmsAsync(string instanceName, string sourceCanonicalName)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -526,8 +505,7 @@ public class SiteStorageService
/// <returns>A task that completes when all native alarm rows for the instance have been deleted.</returns>
public async Task ClearNativeAlarmsForInstanceAsync(string instanceName)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = "DELETE FROM native_alarm_state WHERE instance_unique_name = @name";
@@ -549,8 +527,7 @@ public class SiteStorageService
/// <returns>A task that completes when the shared script has been stored or updated.</returns>
public async Task StoreSharedScriptAsync(string name, string code, string? parameterDefs, string? returnDef)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -578,8 +555,7 @@ public class SiteStorageService
/// <returns>A task that resolves to the list of all stored shared scripts.</returns>
public async Task<List<StoredSharedScript>> GetAllSharedScriptsAsync()
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = "SELECT name, code, parameter_definitions, return_definition FROM shared_scripts";
@@ -616,8 +592,7 @@ public class SiteStorageService
string name, string endpointUrl, string authType, string? authConfig, string? methodDefs,
int timeoutSeconds = 0)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -655,8 +630,7 @@ public class SiteStorageService
public async Task StoreDatabaseConnectionAsync(
string name, string connectionString, int maxRetries, TimeSpan retryDelay)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -699,8 +673,7 @@ public class SiteStorageService
/// <returns>A task that completes when both tables have been emptied.</returns>
public async Task PurgeCentralOnlyNotificationConfigAsync()
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -723,8 +696,7 @@ public class SiteStorageService
public async Task StoreDataConnectionDefinitionAsync(
string name, string protocol, string? configJson, string? backupConfigJson = null, int failoverRetryCount = 3)
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -754,8 +726,7 @@ public class SiteStorageService
/// <returns>A task that resolves to the list of all stored data connection definitions.</returns>
public async Task<List<StoredDataConnectionDefinition>> GetAllDataConnectionDefinitionsAsync()
{
await using var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var connection = OpenConnection();
await using var command = connection.CreateCommand();
command.CommandText = "SELECT name, protocol, configuration, backup_configuration, failover_retry_count FROM data_connection_definitions";
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Repositories/SiteExternalSystemRepository.cs
+10 -5
View File
@@ -31,8 +31,9 @@ public class SiteExternalSystemRepository : IExternalSystemRepository
public async Task<IReadOnlyList<ExternalSystemDefinition>> GetAllExternalSystemsAsync(
CancellationToken cancellationToken = default)
{
// Already open — SiteStorageService.CreateConnection now hands out a
// LocalDb-managed connection. Opening it again would throw.
await using var connection = CreateConnection();
await connection.OpenAsync(cancellationToken);
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -63,8 +64,9 @@ public class SiteExternalSystemRepository : IExternalSystemRepository
public async Task<ExternalSystemDefinition?> GetExternalSystemByNameAsync(
string name, CancellationToken cancellationToken = default)
{
// Already open — SiteStorageService.CreateConnection now hands out a
// LocalDb-managed connection. Opening it again would throw.
await using var connection = CreateConnection();
await connection.OpenAsync(cancellationToken);
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -91,8 +93,9 @@ public class SiteExternalSystemRepository : IExternalSystemRepository
if (system is null)
return Array.Empty<ExternalSystemMethod>();
// Already open — SiteStorageService.CreateConnection now hands out a
// LocalDb-managed connection. Opening it again would throw.
await using var connection = CreateConnection();
await connection.OpenAsync(cancellationToken);
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -140,8 +143,9 @@ public class SiteExternalSystemRepository : IExternalSystemRepository
public async Task<IReadOnlyList<DatabaseConnectionDefinition>> GetAllDatabaseConnectionsAsync(
CancellationToken cancellationToken = default)
{
// Already open — SiteStorageService.CreateConnection now hands out a
// LocalDb-managed connection. Opening it again would throw.
await using var connection = CreateConnection();
await connection.OpenAsync(cancellationToken);
await using var command = connection.CreateCommand();
command.CommandText = @"
@@ -178,8 +182,9 @@ public class SiteExternalSystemRepository : IExternalSystemRepository
public async Task<DatabaseConnectionDefinition?> GetDatabaseConnectionByNameAsync(
string name, CancellationToken cancellationToken = default)
{
// Already open — SiteStorageService.CreateConnection now hands out a
// LocalDb-managed connection. Opening it again would throw.
await using var connection = CreateConnection();
await connection.OpenAsync(cancellationToken);
await using var command = connection.CreateCommand();
command.CommandText = @"
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/ServiceCollectionExtensions.cs
+9 -21
View File
@@ -1,6 +1,7 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment;
@@ -14,32 +15,19 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime;
public static class ServiceCollectionExtensions
{
/// <summary>
/// Registers Site Runtime services including SiteStorageService for SQLite persistence.
/// The caller must register an <see cref="ISiteStorageConnectionProvider"/> or call the
/// overload with an explicit connection string.
/// Registers Site Runtime services including SiteStorageService, which persists to the
/// consolidated site database resolved from <c>ILocalDb</c> (<c>LocalDb:Path</c>).
/// </summary>
/// <param name="services">The DI service collection to register services into.</param>
/// <returns>The same <see cref="IServiceCollection"/> to allow chaining.</returns>
public static IServiceCollection AddSiteRuntime(this IServiceCollection services)
{
// SiteStorageService is registered by the Host using AddSiteRuntime(connectionString)
// This overload is for backward compatibility / skeleton placeholder
return services;
}
/// <summary>
/// Registers Site Runtime services with an explicit SQLite connection string.
/// </summary>
/// <param name="services">The DI service collection to register services into.</param>
/// <param name="siteDbConnectionString">The SQLite connection string for the site local storage database.</param>
/// <returns>The same <see cref="IServiceCollection"/> to allow chaining.</returns>
public static IServiceCollection AddSiteRuntime(this IServiceCollection services, string siteDbConnectionString)
{
services.AddSingleton(sp =>
{
var logger = sp.GetRequiredService<ILogger<SiteStorageService>>();
return new SiteStorageService(siteDbConnectionString, logger);
});
// SiteStorageService takes ILocalDb (the consolidated site database at
// LocalDb:Path) rather than a connection string, so there is nothing left for a
// caller to supply — the string overload is gone and this is the only entry point.
services.AddSingleton(sp => new SiteStorageService(
sp.GetRequiredService<ILocalDb>(),
sp.GetRequiredService<ILogger<SiteStorageService>>()));
services.AddHostedService<SiteStorageInitializer>();
src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ServiceCollectionExtensions.cs
+7 -8
View File
@@ -1,6 +1,7 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.SiteEventLogging;
@@ -15,14 +16,12 @@ public static class ServiceCollectionExtensions
/// <returns>The same <paramref name="services"/> collection, for chaining.</returns>
public static IServiceCollection AddStoreAndForward(this IServiceCollection services)
{
services.AddSingleton<StoreAndForwardStorage>(sp =>
{
var options = sp.GetRequiredService<IOptions<StoreAndForwardOptions>>().Value;
var logger = sp.GetRequiredService<ILogger<StoreAndForwardStorage>>();
return new StoreAndForwardStorage(
$"Data Source={options.SqliteDbPath}",
logger);
});
// The buffer now lives in the consolidated LocalDb database at LocalDb:Path, not
// at StoreAndForwardOptions.SqliteDbPath — that option survives only as the
// migrator's source location (Task 8).
services.AddSingleton<StoreAndForwardStorage>(sp => new StoreAndForwardStorage(
sp.GetRequiredService<ILocalDb>(),
sp.GetRequiredService<ILogger<StoreAndForwardStorage>>()));
services.AddSingleton<StoreAndForwardService>(sp =>
{
src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardStorage.cs
+42 -77
View File
@@ -1,5 +1,6 @@
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
namespace ZB.MOM.WW.ScadaBridge.StoreAndForward;
@@ -11,17 +12,25 @@ namespace ZB.MOM.WW.ScadaBridge.StoreAndForward;
/// </summary>
public class StoreAndForwardStorage
{
private readonly string _connectionString;
private readonly ILocalDb _localDb;
private readonly ILogger<StoreAndForwardStorage> _logger;
/// <summary>
/// Initializes a new instance of <see cref="StoreAndForwardStorage"/> with the given SQLite connection string.
/// Initializes the store over the consolidated site database and applies the schema.
/// </summary>
/// <param name="connectionString">SQLite connection string for the store-and-forward database.</param>
/// <param name="localDb">
/// The consolidated site database. Every connection it hands out is already open and carries
/// the per-connection pragmas (including <c>busy_timeout</c>) plus the <c>zb_hlc_next()</c>
/// UDF that <c>sf_messages</c>' capture triggers call — which is exactly why the store no
/// longer opens its own <see cref="SqliteConnection"/> from a connection string. A raw
/// connection would lack the UDF and every write to the replicated table would fail closed.
/// </param>
/// <param name="logger">Logger for diagnostics.</param>
public StoreAndForwardStorage(string connectionString, ILogger<StoreAndForwardStorage> logger)
public StoreAndForwardStorage(ILocalDb localDb, ILogger<StoreAndForwardStorage> logger)
{
_connectionString = connectionString;
ArgumentNullException.ThrowIfNull(localDb);
ArgumentNullException.ThrowIfNull(logger);
_localDb = localDb;
_logger = logger;
}
@@ -29,22 +38,12 @@ public class StoreAndForwardStorage
/// Creates the sf_messages table if it does not exist.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task InitializeAsync()
public Task InitializeAsync()
{
EnsureDatabaseDirectoryExists();
await using var connection = await OpenConnectionAsync();
// Enable WAL so the concurrent writers this store has by design (script
// enqueues, the retry sweep's per-target lanes, standby replication applies,
// central pull queries) read/write without "database is locked". WAL is
// persistent + file-scoped; in-memory DBs report "memory" instead — harmless,
// so it is not asserted here.
await using (var walCmd = connection.CreateCommand())
{
walCmd.CommandText = "PRAGMA journal_mode=WAL";
await walCmd.ExecuteNonQueryAsync();
}
// No directory creation and no journal-mode pragma here any more: LocalDb owns
// the file (it creates the directory) and sets WAL plus the per-connection
// pragmas on every connection it hands out.
using var connection = OpenConnection();
// The DDL itself lives in StoreAndForwardSchema so the Host can apply it to a
// LocalDb-managed connection before RegisterReplicated installs the capture
@@ -53,50 +52,16 @@ public class StoreAndForwardStorage
StoreAndForwardSchema.Apply(connection);
_logger.LogInformation("Store-and-forward SQLite storage initialized");
return Task.CompletedTask;
}
/// <summary>
/// Ensures the directory for a file-backed SQLite database exists. SQLite creates
/// the database file on demand but not its parent directory, so a configured path
/// such as "./data/store-and-forward.db" fails to open ("unable to open database
/// file") when the "data" directory does not yet exist. In-memory databases and
/// bare filenames in the working directory have no directory to create and are
/// skipped.
/// CreateConnection returns an ALREADY-OPEN, pragma-configured connection with the
/// <c>zb_hlc_next()</c> UDF registered. Calling <c>OpenAsync</c> on it throws, and a raw
/// <see cref="SqliteConnection"/> would lack the UDF, making every capture trigger fail
/// closed.
/// </summary>
private void EnsureDatabaseDirectoryExists()
{
var builder = new SqliteConnectionStringBuilder(_connectionString);
if (builder.Mode == SqliteOpenMode.Memory)
return;
var dataSource = builder.DataSource;
if (string.IsNullOrEmpty(dataSource) || dataSource == ":memory:")
return;
var directory = System.IO.Path.GetDirectoryName(System.IO.Path.GetFullPath(dataSource));
if (!string.IsNullOrEmpty(directory) && !System.IO.Directory.Exists(directory))
{
System.IO.Directory.CreateDirectory(directory);
_logger.LogInformation("Created store-and-forward database directory: {Directory}", directory);
}
}
/// <summary>
/// Opens a connection with a 5s busy_timeout. Concurrent writers exist by
/// design (script enqueues, the sweep's lanes, standby replication applies,
/// central pull queries); with connection-per-operation the pragma must be
/// set per open (it is per-connection, and Microsoft.Data.Sqlite pooling
/// makes the extra statement cheap on a pooled physical connection).
/// </summary>
private async Task<SqliteConnection> OpenConnectionAsync()
{
var connection = new SqliteConnection(_connectionString);
await connection.OpenAsync();
await using var pragma = connection.CreateCommand();
pragma.CommandText = "PRAGMA busy_timeout = 5000";
await pragma.ExecuteNonQueryAsync();
return connection;
}
private SqliteConnection OpenConnection() => _localDb.CreateConnection();
/// <summary>
/// INSERT statement for a full message row. Shared by <see cref="EnqueueAsync"/>
@@ -149,7 +114,7 @@ public class StoreAndForwardStorage
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task EnqueueAsync(StoreAndForwardMessage message)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = InsertMessageSql;
@@ -169,7 +134,7 @@ public class StoreAndForwardStorage
/// <returns>The oldest-first page (at most <paramref name="limit"/> rows) and whether more rows exist beyond it.</returns>
public async Task<(List<StoreAndForwardMessage> Messages, bool Truncated)> GetAllMessagesAsync(int limit)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -197,7 +162,7 @@ public class StoreAndForwardStorage
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task ReplaceAllAsync(IReadOnlyList<StoreAndForwardMessage> messages)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
await using (var deleteCmd = connection.CreateCommand())
@@ -232,7 +197,7 @@ public class StoreAndForwardStorage
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task UpsertMessageAsync(StoreAndForwardMessage message)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -289,7 +254,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to the list of messages due for retry, ordered by creation time ascending.</returns>
public async Task<List<StoreAndForwardMessage>> GetMessagesForRetryAsync(int limit = 0)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -323,7 +288,7 @@ public class StoreAndForwardStorage
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task UpdateMessageAsync(StoreAndForwardMessage message)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -360,7 +325,7 @@ public class StoreAndForwardStorage
StoreAndForwardMessage message,
StoreAndForwardMessageStatus expectedStatus)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -393,7 +358,7 @@ public class StoreAndForwardStorage
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task RemoveMessageAsync(string messageId)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = "DELETE FROM sf_messages WHERE id = @id";
@@ -414,7 +379,7 @@ public class StoreAndForwardStorage
int pageNumber = 1,
int pageSize = 50)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
@@ -459,7 +424,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to <c>true</c> if the message was found and reset to Pending; <c>false</c> if not found or not in Parked status.</returns>
public async Task<bool> RetryParkedMessageAsync(string messageId)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -483,7 +448,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to <c>true</c> if the message was found and deleted; <c>false</c> if not found or not in Parked status.</returns>
public async Task<bool> DiscardParkedMessageAsync(string messageId)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = "DELETE FROM sf_messages WHERE id = @id AND status = @parked";
@@ -500,7 +465,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to a dictionary mapping each category to its pending message count.</returns>
public async Task<Dictionary<StoreAndForwardCategory, int>> GetBufferDepthByCategoryAsync()
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -530,7 +495,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to the number of messages whose origin instance matches <paramref name="instanceName"/>.</returns>
public async Task<int> GetMessageCountByOriginInstanceAsync(string instanceName)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -549,7 +514,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to the matching message, or <c>null</c> if not found.</returns>
public async Task<StoreAndForwardMessage?> GetMessageByIdAsync(string messageId)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = @"
@@ -570,7 +535,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to the number of messages currently in Parked status.</returns>
public async Task<int> GetParkedMessageCountAsync()
{
await using var conn = await OpenConnectionAsync();
await using var conn = OpenConnection();
await using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @parked";
cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
@@ -588,7 +553,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to the oldest parked row's creation time, or <c>null</c> if none are parked.</returns>
public async Task<DateTimeOffset?> GetOldestParkedCreatedAtAsync()
{
await using var conn = await OpenConnectionAsync();
await using var conn = OpenConnection();
await using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT MIN(created_at) FROM sf_messages WHERE status = @parked";
cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
@@ -605,7 +570,7 @@ public class StoreAndForwardStorage
/// <returns>A task that resolves to the count of messages with the specified status.</returns>
public async Task<int> GetMessageCountByStatusAsync(StoreAndForwardMessageStatus status)
{
await using var connection = await OpenConnectionAsync();
await using var connection = OpenConnection();
await using var cmd = connection.CreateCommand();
cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @status";