Merge pull request 'Phase 7 follow-up #247 — Galaxy.Host historian writer + SQLite sink activation' (#194) from phase-7-fu-247-galaxy-historian-writer into v2

src/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy/GalaxyProxyDriver.cs

@@ -1,4 +1,5 @@
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
 using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy.Ipc;
 using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;
 using IpcHostConnectivityStatus = ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts.HostConnectivityStatus;
@@ -22,6 +23,7 @@ public sealed class GalaxyProxyDriver(GalaxyProxyOptions options)
       IHistoryProvider,
       IRediscoverable,
       IHostConnectivityProbe,
+      IAlarmHistorianWriter,
       IDisposable
 {
     private GalaxyIpcClient? _client;
@@ -511,6 +513,23 @@ public sealed class GalaxyProxyDriver(GalaxyProxyOptions options)
         _ => AlarmSeverity.Critical,
     };
 
+    /// <summary>
+    ///     Phase 7 follow-up #247 — IAlarmHistorianWriter implementation. Forwards alarm
+    ///     batches to Galaxy.Host over the existing IPC channel, reusing the connection
+    ///     the driver already established for data-plane traffic. Throws
+    ///     <see cref="InvalidOperationException"/> when called before
+    ///     <see cref="InitializeAsync"/> has connected the client; the SQLite drain worker
+    ///     translates that to whole-batch RetryPlease per its catch contract.
+    /// </summary>
+    public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
+        IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
+    {
+        if (_client is null)
+            throw new InvalidOperationException(
+                "GalaxyProxyDriver IPC client not connected — historian writes rejected until InitializeAsync completes");
+        return new GalaxyHistorianWriter(_client).WriteBatchAsync(batch, cancellationToken);
+    }
+
     public void Dispose() => _client?.DisposeAsync().AsTask().GetAwaiter().GetResult();
 }
 

src/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy/Ipc/GalaxyHistorianWriter.cs

@@ -0,0 +1,90 @@
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy.Ipc;
+
+/// <summary>
+///     Phase 7 follow-up (task #247) — bridges <see cref="SqliteStoreAndForwardSink"/>'s
+///     drain worker to <c>Driver.Galaxy.Host</c> over the existing <see cref="GalaxyIpcClient"/>
+///     pipe. Translates <see cref="AlarmHistorianEvent"/> batches into the
+///     <see cref="HistorianAlarmEventDto"/> wire format the Host expects + maps per-event
+///     <see cref="HistorianAlarmEventOutcomeDto"/> responses back to
+///     <see cref="HistorianWriteOutcome"/> so the SQLite queue knows what to ack /
+///     dead-letter / retry.
+/// </summary>
+/// <remarks>
+///     <para>
+///         Reuses the IPC channel <see cref="GalaxyProxyDriver"/> already opens for the
+///         Galaxy data plane — no second pipe to <c>Driver.Galaxy.Host</c>, no separate
+///         auth handshake. The IPC client's call gate serializes historian batches with
+///         driver Reads/Writes/Subscribes; historian batches are infrequent (every few
+///         seconds at most under the SQLite sink's drain cadence) so the contention is
+///         negligible compared to per-tag-read pressure.
+///     </para>
+///     <para>
+///         Pipe-level transport faults (broken pipe, host crash) bubble up as
+///         <see cref="GalaxyIpcException"/> which the SQLite sink's drain worker catches +
+///         translates to a whole-batch RetryPlease per the
+///         <see cref="SqliteStoreAndForwardSink"/> docstring — failed events stay queued
+///         for the next drain tick after backoff.
+///     </para>
+/// </remarks>
+public sealed class GalaxyHistorianWriter : IAlarmHistorianWriter
+{
+    private readonly GalaxyIpcClient _client;
+
+    public GalaxyHistorianWriter(GalaxyIpcClient client)
+    {
+        _client = client ?? throw new ArgumentNullException(nameof(client));
+    }
+
+    public async Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
+        IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(batch);
+        if (batch.Count == 0) return [];
+
+        var request = new HistorianAlarmEventRequest
+        {
+            Events = batch.Select(ToDto).ToArray(),
+        };
+
+        var response = await _client.CallAsync<HistorianAlarmEventRequest, HistorianAlarmEventResponse>(
+            requestKind: MessageKind.HistorianAlarmEventRequest,
+            request: request,
+            expectedResponseKind: MessageKind.HistorianAlarmEventResponse,
+            ct: cancellationToken).ConfigureAwait(false);
+
+        if (response.Outcomes.Length != batch.Count)
+            throw new InvalidOperationException(
+                $"Galaxy.Host returned {response.Outcomes.Length} outcomes for a batch of {batch.Count} — protocol mismatch");
+
+        var outcomes = new HistorianWriteOutcome[response.Outcomes.Length];
+        for (var i = 0; i < response.Outcomes.Length; i++)
+            outcomes[i] = MapOutcome(response.Outcomes[i]);
+        return outcomes;
+    }
+
+    internal static HistorianAlarmEventDto ToDto(AlarmHistorianEvent e) => new()
+    {
+        AlarmId = e.AlarmId,
+        EquipmentPath = e.EquipmentPath,
+        AlarmName = e.AlarmName,
+        AlarmTypeName = e.AlarmTypeName,
+        Severity = (int)e.Severity,
+        EventKind = e.EventKind,
+        Message = e.Message,
+        User = e.User,
+        Comment = e.Comment,
+        TimestampUtcUnixMs = new DateTimeOffset(e.TimestampUtc, TimeSpan.Zero).ToUnixTimeMilliseconds(),
+    };
+
+    internal static HistorianWriteOutcome MapOutcome(HistorianAlarmEventOutcomeDto wire) => wire switch
+    {
+        HistorianAlarmEventOutcomeDto.Ack => HistorianWriteOutcome.Ack,
+        HistorianAlarmEventOutcomeDto.RetryPlease => HistorianWriteOutcome.RetryPlease,
+        HistorianAlarmEventOutcomeDto.PermanentFail => HistorianWriteOutcome.PermanentFail,
+        _ => throw new InvalidOperationException($"Unknown HistorianAlarmEventOutcomeDto byte {(byte)wire}"),
+    };
+}

src/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy.csproj

@@ -14,6 +14,7 @@
     <ItemGroup>
         <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.Abstractions\ZB.MOM.WW.OtOpcUa.Core.Abstractions.csproj"/>
         <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared\ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.csproj"/>
+        <ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian\ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.csproj"/>
     </ItemGroup>
 
     <ItemGroup>

src/ZB.MOM.WW.OtOpcUa.Server/Phase7/Phase7Composer.cs

@@ -47,6 +47,9 @@ public sealed class Phase7Composer : IAsyncDisposable
 
     private DriverSubscriptionBridge? _bridge;
     private Phase7ComposedSources _sources = Phase7ComposedSources.Empty;
+    // Sink we constructed in PrepareAsync (vs. the injected fallback). Held so
+    // DisposeAsync can flush + tear down the SQLite drain timer.
+    private SqliteStoreAndForwardSink? _ownedSink;
     private bool _disposed;
 
     public Phase7Composer(
@@ -96,13 +99,20 @@ public sealed class Phase7Composer : IAsyncDisposable
 
         var upstream = new CachedTagUpstreamSource();
 
+        // Phase 7 follow-up #247 — if any registered driver implements IAlarmHistorianWriter
+        // (today: GalaxyProxyDriver), wrap it in a SqliteStoreAndForwardSink at
+        // %ProgramData%/OtOpcUa/alarm-historian-queue.db with the 2s drain cadence the
+        // sink's docstring recommends. Otherwise fall back to the injected sink (Null in
+        // the default registration).
+        var historianSink = ResolveHistorianSink();
+
         _sources = Phase7EngineComposer.Compose(
             scripts: scripts,
             virtualTags: virtualTags,
             scriptedAlarms: scriptedAlarms,
             upstream: upstream,
             alarmStateStore: new InMemoryAlarmStateStore(),
-            historianSink: _historianSink,
+            historianSink: historianSink,
             rootScriptLogger: _scriptLogger,
             loggerFactory: _loggerFactory);
 
@@ -121,6 +131,47 @@ public sealed class Phase7Composer : IAsyncDisposable
         return _sources;
     }
 
+    private IAlarmHistorianSink ResolveHistorianSink()
+    {
+        IAlarmHistorianWriter? writer = null;
+        foreach (var driverId in _driverHost.RegisteredDriverIds)
+        {
+            if (_driverHost.GetDriver(driverId) is IAlarmHistorianWriter w)
+            {
+                writer = w;
+                _logger.LogInformation(
+                    "Phase 7 historian sink: driver {Driver} provides IAlarmHistorianWriter — wiring SqliteStoreAndForwardSink",
+                    driverId);
+                break;
+            }
+        }
+        if (writer is null)
+        {
+            _logger.LogInformation(
+                "Phase 7 historian sink: no driver provides IAlarmHistorianWriter — using {Sink}",
+                _historianSink.GetType().Name);
+            return _historianSink;
+        }
+
+        var queueRoot = Environment.GetFolderPath(Environment.SpecialFolder.CommonApplicationData);
+        if (string.IsNullOrEmpty(queueRoot)) queueRoot = Path.GetTempPath();
+        var queueDir = Path.Combine(queueRoot, "OtOpcUa");
+        Directory.CreateDirectory(queueDir);
+        var queuePath = Path.Combine(queueDir, "alarm-historian-queue.db");
+
+        var sinkLogger = _loggerFactory.CreateLogger<SqliteStoreAndForwardSink>();
+        // SqliteStoreAndForwardSink wants a Serilog logger for warn-on-eviction emissions;
+        // bridge the Microsoft logger via Serilog's null-safe path until the sink's
+        // dependency surface is reshaped (covered as part of release-readiness).
+        var serilogShim = _scriptLogger.ForContext("HistorianQueuePath", queuePath);
+        _ownedSink = new SqliteStoreAndForwardSink(
+            databasePath: queuePath,
+            writer: writer,
+            logger: serilogShim);
+        _ownedSink.StartDrainLoop(TimeSpan.FromSeconds(2));
+        return _ownedSink;
+    }
+
     /// <summary>
     ///     For each registered driver that exposes <see cref="Core.Abstractions.ISubscribable"/>,
     ///     build a UNS-path → driver-fullRef map from its EquipmentNamespaceContent.
@@ -178,6 +229,9 @@ public sealed class Phase7Composer : IAsyncDisposable
             try { d.Dispose(); }
             catch (Exception ex) { _logger.LogWarning(ex, "Phase 7 disposable threw during shutdown"); }
         }
+        // Owned SQLite sink: dispose first so the drain timer stops + final batch flushes
+        // before we release the writer-bearing driver via DriverHost.DisposeAsync upstream.
+        _ownedSink?.Dispose();
         if (_historianSink is IDisposable disposableSink) disposableSink.Dispose();
     }
 }

tests/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy.Tests/GalaxyHistorianWriterMappingTests.cs

@@ -0,0 +1,83 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy.Ipc;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Proxy.Tests;
+
+/// <summary>
+///     Phase 7 follow-up #247 — covers the wire-format translation between the
+///     <see cref="AlarmHistorianEvent"/> the SQLite sink hands to the writer + the
+///     <see cref="HistorianAlarmEventDto"/> the Galaxy.Host IPC contract expects, plus
+///     the per-event outcome enum mapping. Pure functions; the round-trip over a real
+///     pipe is exercised by the live Host suite (task #240).
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class GalaxyHistorianWriterMappingTests
+{
+    [Fact]
+    public void ToDto_round_trips_every_field()
+    {
+        var ts = new DateTime(2026, 4, 20, 14, 30, 0, DateTimeKind.Utc);
+        var e = new AlarmHistorianEvent(
+            AlarmId: "al-7",
+            EquipmentPath: "/Site/Line/Cell",
+            AlarmName: "HighTemp",
+            AlarmTypeName: "LimitAlarm",
+            Severity: AlarmSeverity.High,
+            EventKind: "RaiseEvent",
+            Message: "Temp 92°C exceeded 90°C",
+            User: "operator-7",
+            Comment: "ack with reason",
+            TimestampUtc: ts);
+
+        var dto = GalaxyHistorianWriter.ToDto(e);
+
+        dto.AlarmId.ShouldBe("al-7");
+        dto.EquipmentPath.ShouldBe("/Site/Line/Cell");
+        dto.AlarmName.ShouldBe("HighTemp");
+        dto.AlarmTypeName.ShouldBe("LimitAlarm");
+        dto.Severity.ShouldBe((int)AlarmSeverity.High);
+        dto.EventKind.ShouldBe("RaiseEvent");
+        dto.Message.ShouldBe("Temp 92°C exceeded 90°C");
+        dto.User.ShouldBe("operator-7");
+        dto.Comment.ShouldBe("ack with reason");
+        dto.TimestampUtcUnixMs.ShouldBe(new DateTimeOffset(ts, TimeSpan.Zero).ToUnixTimeMilliseconds());
+    }
+
+    [Fact]
+    public void ToDto_preserves_null_Comment()
+    {
+        var e = new AlarmHistorianEvent(
+            "a", "/p", "n", "AlarmCondition", AlarmSeverity.Low, "RaiseEvent", "m",
+            User: "system", Comment: null, TimestampUtc: DateTime.UtcNow);
+
+        GalaxyHistorianWriter.ToDto(e).Comment.ShouldBeNull();
+    }
+
+    [Theory]
+    [InlineData(HistorianAlarmEventOutcomeDto.Ack, HistorianWriteOutcome.Ack)]
+    [InlineData(HistorianAlarmEventOutcomeDto.RetryPlease, HistorianWriteOutcome.RetryPlease)]
+    [InlineData(HistorianAlarmEventOutcomeDto.PermanentFail, HistorianWriteOutcome.PermanentFail)]
+    public void MapOutcome_round_trips_every_byte(
+        HistorianAlarmEventOutcomeDto wire, HistorianWriteOutcome expected)
+    {
+        GalaxyHistorianWriter.MapOutcome(wire).ShouldBe(expected);
+    }
+
+    [Fact]
+    public void MapOutcome_unknown_byte_throws()
+    {
+        Should.Throw<InvalidOperationException>(
+            () => GalaxyHistorianWriter.MapOutcome((HistorianAlarmEventOutcomeDto)0xFF));
+    }
+
+    [Fact]
+    public void Null_client_rejected()
+    {
+        Should.Throw<ArgumentNullException>(() => new GalaxyHistorianWriter(null!));
+    }
+
+}