feat(historian): SQL Server backend skeleton - schema DDL, locked bootstrapper, options, connection/mapper/tag-catalog helpers, status consts, env-gated fixture

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-09 12:52:15 -04:00
parent ff7c6c531a
commit 17962d4233
19 changed files with 1187 additions and 0 deletions
@@ -0,0 +1,155 @@
+using System.Data;
+using System.Reflection;
+using System.Text.RegularExpressions;
+using Microsoft.Data.SqlClient;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
+
+/// <summary>
+///     Lazily applies the embedded, idempotent schema scripts once per instance, under a SQL application lock
+///     so concurrent nodes/instances serialise. A failure is NOT latched: the next <see cref="EnsureAsync"/> retries.
+///     With <c>AutoCreateSchema=false</c> it only verifies <c>MAX(Version) &gt;= 1</c>.
+/// </summary>
+internal sealed partial class SqlServerSchemaBootstrapper
+{
+    /// <summary>The schema version this build requires.</summary>
+    public const int RequiredVersion = 1;
+
+    private const string LockResource = "OtOpcUaHistorianSchema";
+    private const int LockTimeoutMs = 30_000;
+
+    private readonly SqlHistorianConnection _conn;
+    private readonly bool _autoCreate;
+    private readonly ILogger _log;
+    private readonly SemaphoreSlim _gate = new(1, 1);
+    private volatile bool _ready;
+
+    /// <summary>Creates the bootstrapper.</summary>
+    /// <param name="conn">Connection helper.</param>
+    /// <param name="options">Validated options.</param>
+    /// <param name="log">Optional logger.</param>
+    public SqlServerSchemaBootstrapper(SqlHistorianConnection conn, SqlServerHistorianOptions options, ILogger? log = null)
+    {
+        _conn = conn;
+        _autoCreate = options.AutoCreateSchema;
+        _log = log ?? NullLogger.Instance;
+    }
+
+    /// <summary>Ensures the schema is ready; no-op after the first success.</summary>
+    /// <param name="ct">Cancellation.</param>
+    /// <returns>A task completing when the schema is usable.</returns>
+    /// <exception cref="HistorianSchemaMissingException">AutoCreateSchema is false and version 1 is absent.</exception>
+    public async Task EnsureAsync(CancellationToken ct)
+    {
+        if (_ready) return;
+        await _gate.WaitAsync(ct).ConfigureAwait(false);
+        try
+        {
+            if (_ready) return;
+            await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+            if (_autoCreate) await ApplyAsync(conn, ct).ConfigureAwait(false);
+            else await VerifyAsync(conn, ct).ConfigureAwait(false);
+            _ready = true;
+        }
+        finally
+        {
+            _gate.Release();
+        }
+    }
+
+    private async Task<int> ReadVersionAsync(SqlConnection conn, CancellationToken ct)
+    {
+        await using var cmd = _conn.CreateCommand(conn,
+            $"IF OBJECT_ID(N'{_conn.Q("SchemaVersion")}', N'U') IS NULL SELECT 0 ELSE SELECT ISNULL(MAX(Version), 0) FROM {_conn.Q("SchemaVersion")}");
+        return Convert.ToInt32(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false));
+    }
+
+    private async Task VerifyAsync(SqlConnection conn, CancellationToken ct)
+    {
+        var v = await ReadVersionAsync(conn, ct).ConfigureAwait(false);
+        if (v < RequiredVersion)
+            throw new HistorianSchemaMissingException(
+                $"Historian schema [{_conn.Schema}] is missing version {RequiredVersion} (found {v}) and AutoCreateSchema is false. " +
+                "Apply docs/historian/sqlserver/001_initial.sql or enable AutoCreateSchema.");
+    }
+
+    private async Task ApplyAsync(SqlConnection conn, CancellationToken ct)
+    {
+        await AcquireLockAsync(conn, ct).ConfigureAwait(false);
+        try
+        {
+            var current = await ReadVersionAsync(conn, ct).ConfigureAwait(false);
+            foreach (var (version, name, text) in LoadScripts())
+            {
+                if (version <= current) continue;
+                foreach (var batch in SplitBatches(text.Replace("$(Schema)", _conn.Schema, StringComparison.Ordinal)))
+                {
+                    await using var cmd = _conn.CreateCommand(conn, batch);
+                    await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+                }
+                _log.LogInformation("Applied historian schema script {Script} to [{Schema}]", name, _conn.Schema);
+            }
+            await VerifyAsync(conn, ct).ConfigureAwait(false);
+        }
+        finally
+        {
+            await ReleaseLockAsync(conn).ConfigureAwait(false);
+        }
+    }
+
+    private async Task AcquireLockAsync(SqlConnection conn, CancellationToken ct)
+    {
+        await using var cmd = _conn.CreateCommand(conn, "sp_getapplock");
+        cmd.CommandType = CommandType.StoredProcedure;
+        cmd.CommandTimeout = Math.Max(cmd.CommandTimeout, LockTimeoutMs / 1000 + 15);
+        cmd.Parameters.AddWithValue("@Resource", LockResource);
+        cmd.Parameters.AddWithValue("@LockMode", "Exclusive");
+        cmd.Parameters.AddWithValue("@LockOwner", "Session");
+        cmd.Parameters.AddWithValue("@LockTimeout", LockTimeoutMs);
+        var rc = new SqlParameter("@rc", SqlDbType.Int) { Direction = ParameterDirection.ReturnValue };
+        cmd.Parameters.Add(rc);
+        await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+        if ((int)rc.Value < 0)
+            throw new TimeoutException($"Could not acquire historian schema applock within {LockTimeoutMs} ms (sp_getapplock returned {rc.Value}).");
+    }
+
+    private async Task ReleaseLockAsync(SqlConnection conn)
+    {
+        try
+        {
+            await using var cmd = _conn.CreateCommand(conn, "sp_releaseapplock");
+            cmd.CommandType = CommandType.StoredProcedure;
+            cmd.Parameters.AddWithValue("@Resource", LockResource);
+            cmd.Parameters.AddWithValue("@LockOwner", "Session");
+            await cmd.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            // Closing the session releases a Session-owned lock anyway.
+            _log.LogDebug(ex, "sp_releaseapplock failed; the lock dies with the connection");
+        }
+    }
+
+    [GeneratedRegex(@"^\s*GO\s*$", RegexOptions.Multiline | RegexOptions.IgnoreCase)]
+    private static partial Regex GoLine();
+
+    internal static IEnumerable<string> SplitBatches(string script) =>
+        GoLine().Split(script).Select(b => b.Trim()).Where(b => b.Length > 0);
+
+    private static List<(int Version, string Name, string Text)> LoadScripts()
+    {
+        var asm = typeof(SqlServerSchemaBootstrapper).Assembly;
+        var list = new List<(int, string, string)>();
+        foreach (var res in asm.GetManifestResourceNames())
+        {
+            var m = Regex.Match(res, @"^Schema\.(\d+)_.*\.sql$");
+            if (!m.Success) continue;
+            using var s = asm.GetManifestResourceStream(res)!;
+            using var r = new StreamReader(s);
+            list.Add((int.Parse(m.Groups[1].Value), res, r.ReadToEnd()));
+        }
+        return list.OrderBy(x => x.Item1).ToList();
+    }
+}