Merge pull request 'Phase 7 follow-up #246 — Phase7Composer + Program.cs wire-in' (#193) from phase-7-fu-246-program-wireup into v2

src/ZB.MOM.WW.OtOpcUa.Server/OpcUa/OpcUaApplicationHost.cs

@@ -34,9 +34,11 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
     // Phase 7 Stream G follow-up (task #239). When composed with the VirtualTagEngine +
     // ScriptedAlarmEngine sources these route node reads to the engines instead of the
     // driver. Null = Phase 7 engines not enabled for this deployment (identical to pre-
-    // Phase-7 behaviour).
+    // Phase-7 behaviour). Late-bindable via SetPhase7Sources because the engines need
-    private readonly ZB.MOM.WW.OtOpcUa.Core.Abstractions.IReadable? _virtualReadable;
+    // the bootstrapped generation id before they can compose, which is only known after
-    private readonly ZB.MOM.WW.OtOpcUa.Core.Abstractions.IReadable? _scriptedAlarmReadable;
+    // the host has been DI-constructed (task #246).
+    private ZB.MOM.WW.OtOpcUa.Core.Abstractions.IReadable? _virtualReadable;
+    private ZB.MOM.WW.OtOpcUa.Core.Abstractions.IReadable? _scriptedAlarmReadable;
 
     private readonly ILoggerFactory _loggerFactory;
     private readonly ILogger<OpcUaApplicationHost> _logger;
@@ -75,6 +77,24 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
 
     public OtOpcUaServer? Server => _server;
 
+    /// <summary>
+    ///     Late-bind the Phase 7 engine-backed <c>IReadable</c> sources. Must be
+    ///     called BEFORE <see cref="StartAsync"/> — once the OPC UA server starts, the
+    ///     <see cref="OtOpcUaServer"/> ctor captures the field values + per-node
+    ///     <see cref="DriverNodeManager"/>s are constructed. Calling this after start has
+    ///     no effect on already-materialized node managers.
+    /// </summary>
+    public void SetPhase7Sources(
+        ZB.MOM.WW.OtOpcUa.Core.Abstractions.IReadable? virtualReadable,
+        ZB.MOM.WW.OtOpcUa.Core.Abstractions.IReadable? scriptedAlarmReadable)
+    {
+        if (_server is not null)
+            throw new InvalidOperationException(
+                "Phase 7 sources must be set before OpcUaApplicationHost.StartAsync; the OtOpcUaServer + DriverNodeManagers have already captured the previous values.");
+        _virtualReadable = virtualReadable;
+        _scriptedAlarmReadable = scriptedAlarmReadable;
+    }
+
     /// <summary>
     ///     Builds the <see cref="ApplicationConfiguration"/>, validates/creates the application
     ///     certificate, constructs + starts the <see cref="OtOpcUaServer"/>, then drives

src/ZB.MOM.WW.OtOpcUa.Server/OpcUaServerService.cs

@@ -3,6 +3,7 @@ using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using ZB.MOM.WW.OtOpcUa.Core.Hosting;
 using ZB.MOM.WW.OtOpcUa.Server.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Server.Phase7;
 
 namespace ZB.MOM.WW.OtOpcUa.Server;
 
@@ -17,6 +18,7 @@ public sealed class OpcUaServerService(
     DriverHost driverHost,
     OpcUaApplicationHost applicationHost,
     DriverEquipmentContentRegistry equipmentContentRegistry,
+    Phase7Composer phase7Composer,
     IServiceScopeFactory scopeFactory,
     ILogger<OpcUaServerService> logger) : BackgroundService
 {
@@ -34,12 +36,19 @@ public sealed class OpcUaServerService(
         // Skipped when no generation is Published yet — the fleet boots into a UNS-less
         // address space until the first publish, then the registry fills on next restart.
         if (result.GenerationId is { } gen)
+        {
             await PopulateEquipmentContentAsync(gen, stoppingToken);
 
-        // PR 17: stand up the OPC UA server + drive discovery per registered driver. Driver
+            // Phase 7 follow-up #246 — load Script + VirtualTag + ScriptedAlarm rows,
-        // registration itself (RegisterAsync on DriverHost) happens during an earlier DI
+            // compose VirtualTagEngine + ScriptedAlarmEngine, start the driver-bridge
-        // extension once the central config DB query + per-driver factory land; for now the
+            // feed. SetPhase7Sources MUST run before applicationHost.StartAsync because
-        // server comes up with whatever drivers are in DriverHost at start time.
+            // OtOpcUaServer + DriverNodeManager construction captures the field values
+            // — late binding after server start is rejected with InvalidOperationException.
+            // No-op when the generation has no virtual tags or scripted alarms.
+            var phase7 = await phase7Composer.PrepareAsync(gen, stoppingToken);
+            applicationHost.SetPhase7Sources(phase7.VirtualReadable, phase7.ScriptedAlarmReadable);
+        }
+
         await applicationHost.StartAsync(stoppingToken);
 
         logger.LogInformation("OtOpcUa.Server running. Hosted drivers: {Count}", driverHost.RegisteredDriverIds.Count);
@@ -57,6 +66,11 @@ public sealed class OpcUaServerService(
     public override async Task StopAsync(CancellationToken cancellationToken)
     {
         await base.StopAsync(cancellationToken);
+        // Dispose Phase 7 first so the bridge stops feeding the cache + the engines
+        // stop firing alarm/historian events before the OPC UA server tears down its
+        // node managers. Otherwise an in-flight cascade could try to push through a
+        // disposed source and surface as a noisy shutdown warning.
+        await phase7Composer.DisposeAsync();
         await applicationHost.DisposeAsync();
         await driverHost.DisposeAsync();
     }

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

@@ -0,0 +1,183 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+using ZB.MOM.WW.OtOpcUa.Core.Hosting;
+using ZB.MOM.WW.OtOpcUa.Core.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
+using ZB.MOM.WW.OtOpcUa.Server.OpcUa;
+
+namespace ZB.MOM.WW.OtOpcUa.Server.Phase7;
+
+/// <summary>
+///     Phase 7 follow-up (task #246) — orchestrates the runtime composition of virtual
+///     tags + scripted alarms + the historian sink + the driver-bridge that feeds the
+///     engines. Called by <see cref="OpcUaServerService"/> after the bootstrap generation
+///     loads + before <see cref="OpcUaApplicationHost.StartAsync"/>.
+/// </summary>
+/// <remarks>
+///     <para>
+///         <see cref="PrepareAsync"/> reads Script / VirtualTag / ScriptedAlarm rows from
+///         the central config DB at the bootstrapped generation, instantiates a
+///         <see cref="CachedTagUpstreamSource"/>, runs <see cref="Phase7EngineComposer.Compose"/>,
+///         starts a <see cref="DriverSubscriptionBridge"/> per registered driver feeding
+///         <see cref="EquipmentNamespaceContent"/>'s tag rows into the cache, and returns
+///         the engine-backed <see cref="Core.Abstractions.IReadable"/> sources for
+///         <see cref="OpcUaApplicationHost.SetPhase7Sources"/>.
+///     </para>
+///     <para>
+///         <see cref="DisposeAsync"/> tears down the bridge first (so no more events
+///         arrive at the cache), then the engines (so cascades + timer ticks stop), then
+///         the SQLite sink (which flushes any in-flight drain). Lifetime is owned by the
+///         host; <see cref="OpcUaServerService.StopAsync"/> calls dispose during graceful
+///         shutdown.
+///     </para>
+/// </remarks>
+public sealed class Phase7Composer : IAsyncDisposable
+{
+    private readonly IServiceScopeFactory _scopeFactory;
+    private readonly DriverHost _driverHost;
+    private readonly DriverEquipmentContentRegistry _equipmentRegistry;
+    private readonly IAlarmHistorianSink _historianSink;
+    private readonly ILoggerFactory _loggerFactory;
+    private readonly Serilog.ILogger _scriptLogger;
+    private readonly ILogger<Phase7Composer> _logger;
+
+    private DriverSubscriptionBridge? _bridge;
+    private Phase7ComposedSources _sources = Phase7ComposedSources.Empty;
+    private bool _disposed;
+
+    public Phase7Composer(
+        IServiceScopeFactory scopeFactory,
+        DriverHost driverHost,
+        DriverEquipmentContentRegistry equipmentRegistry,
+        IAlarmHistorianSink historianSink,
+        ILoggerFactory loggerFactory,
+        Serilog.ILogger scriptLogger,
+        ILogger<Phase7Composer> logger)
+    {
+        _scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
+        _driverHost = driverHost ?? throw new ArgumentNullException(nameof(driverHost));
+        _equipmentRegistry = equipmentRegistry ?? throw new ArgumentNullException(nameof(equipmentRegistry));
+        _historianSink = historianSink ?? throw new ArgumentNullException(nameof(historianSink));
+        _loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
+        _scriptLogger = scriptLogger ?? throw new ArgumentNullException(nameof(scriptLogger));
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+    }
+
+    public Phase7ComposedSources Sources => _sources;
+
+    public async Task<Phase7ComposedSources> PrepareAsync(long generationId, CancellationToken ct)
+    {
+        if (_disposed) throw new ObjectDisposedException(nameof(Phase7Composer));
+
+        // Load the three Phase 7 row sets in one DB scope.
+        List<Script> scripts;
+        List<VirtualTag> virtualTags;
+        List<ScriptedAlarm> scriptedAlarms;
+        using (var scope = _scopeFactory.CreateScope())
+        {
+            var db = scope.ServiceProvider.GetRequiredService<OtOpcUaConfigDbContext>();
+            scripts = await db.Scripts.AsNoTracking()
+                .Where(s => s.GenerationId == generationId).ToListAsync(ct).ConfigureAwait(false);
+            virtualTags = await db.VirtualTags.AsNoTracking()
+                .Where(v => v.GenerationId == generationId && v.Enabled).ToListAsync(ct).ConfigureAwait(false);
+            scriptedAlarms = await db.ScriptedAlarms.AsNoTracking()
+                .Where(a => a.GenerationId == generationId && a.Enabled).ToListAsync(ct).ConfigureAwait(false);
+        }
+
+        if (virtualTags.Count == 0 && scriptedAlarms.Count == 0)
+        {
+            _logger.LogInformation("Phase 7: no virtual tags or scripted alarms in generation {Gen}; engines dormant", generationId);
+            return Phase7ComposedSources.Empty;
+        }
+
+        var upstream = new CachedTagUpstreamSource();
+
+        _sources = Phase7EngineComposer.Compose(
+            scripts: scripts,
+            virtualTags: virtualTags,
+            scriptedAlarms: scriptedAlarms,
+            upstream: upstream,
+            alarmStateStore: new InMemoryAlarmStateStore(),
+            historianSink: _historianSink,
+            rootScriptLogger: _scriptLogger,
+            loggerFactory: _loggerFactory);
+
+        _logger.LogInformation(
+            "Phase 7: composed engines from generation {Gen} — {Vt} virtual tag(s), {Al} scripted alarm(s), {Sc} script(s)",
+            generationId, virtualTags.Count, scriptedAlarms.Count, scripts.Count);
+
+        // Build driver feeds from each registered driver's EquipmentNamespaceContent + start
+        // the bridge. Drivers without populated content (Galaxy SystemPlatform-kind, drivers
+        // whose Equipment rows haven't been published yet) contribute an empty feed which
+        // the bridge silently skips.
+        _bridge = new DriverSubscriptionBridge(upstream, _loggerFactory.CreateLogger<DriverSubscriptionBridge>());
+        var feeds = BuildDriverFeeds(_driverHost, _equipmentRegistry);
+        await _bridge.StartAsync(feeds, ct).ConfigureAwait(false);
+
+        return _sources;
+    }
+
+    /// <summary>
+    ///     For each registered driver that exposes <see cref="Core.Abstractions.ISubscribable"/>,
+    ///     build a UNS-path → driver-fullRef map from its EquipmentNamespaceContent.
+    ///     Path convention: <c>/{areaName}/{lineName}/{equipmentName}/{tagName}</c> matching
+    ///     what the EquipmentNodeWalker emits into the OPC UA browse tree, so script literals
+    ///     written against the operator-visible tree work without translation.
+    /// </summary>
+    internal static IReadOnlyList<DriverFeed> BuildDriverFeeds(
+        DriverHost driverHost, DriverEquipmentContentRegistry equipmentRegistry)
+    {
+        var feeds = new List<DriverFeed>();
+        foreach (var driverId in driverHost.RegisteredDriverIds)
+        {
+            var driver = driverHost.GetDriver(driverId);
+            if (driver is not Core.Abstractions.ISubscribable subscribable) continue;
+
+            var content = equipmentRegistry.Get(driverId);
+            if (content is null) continue;
+
+            var pathToFullRef = MapPathsToFullRefs(content);
+            if (pathToFullRef.Count == 0) continue;
+
+            feeds.Add(new DriverFeed(subscribable, pathToFullRef, TimeSpan.FromSeconds(1)));
+        }
+        return feeds;
+    }
+
+    internal static IReadOnlyDictionary<string, string> MapPathsToFullRefs(EquipmentNamespaceContent content)
+    {
+        var result = new Dictionary<string, string>(StringComparer.Ordinal);
+        var areaById = content.Areas.ToDictionary(a => a.UnsAreaId, StringComparer.OrdinalIgnoreCase);
+        var lineById = content.Lines.ToDictionary(l => l.UnsLineId, StringComparer.OrdinalIgnoreCase);
+        var equipmentById = content.Equipment.ToDictionary(e => e.EquipmentId, StringComparer.OrdinalIgnoreCase);
+
+        foreach (var tag in content.Tags)
+        {
+            if (string.IsNullOrEmpty(tag.EquipmentId)) continue;
+            if (!equipmentById.TryGetValue(tag.EquipmentId!, out var eq)) continue;
+            if (!lineById.TryGetValue(eq.UnsLineId, out var line)) continue;
+            if (!areaById.TryGetValue(line.UnsAreaId, out var area)) continue;
+
+            var path = $"/{area.Name}/{line.Name}/{eq.Name}/{tag.Name}";
+            result[path] = tag.TagConfig;  // duplicate-path collisions naturally win-last; UI publish-validation rules out duplicate names
+        }
+        return result;
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        if (_disposed) return;
+        _disposed = true;
+        if (_bridge is not null) await _bridge.DisposeAsync().ConfigureAwait(false);
+        foreach (var d in _sources.Disposables)
+        {
+            try { d.Dispose(); }
+            catch (Exception ex) { _logger.LogWarning(ex, "Phase 7 disposable threw during shutdown"); }
+        }
+        if (_historianSink is IDisposable disposableSink) disposableSink.Dispose();
+    }
+}

src/ZB.MOM.WW.OtOpcUa.Server/Program.cs

@@ -8,8 +8,10 @@ using Serilog.Formatting.Compact;
 using ZB.MOM.WW.OtOpcUa.Configuration;
 using ZB.MOM.WW.OtOpcUa.Configuration.LocalCache;
 using ZB.MOM.WW.OtOpcUa.Core.Hosting;
+using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
 using ZB.MOM.WW.OtOpcUa.Server;
 using ZB.MOM.WW.OtOpcUa.Server.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Server.Phase7;
 using ZB.MOM.WW.OtOpcUa.Server.Security;
 
 var builder = Host.CreateApplicationBuilder(args);
@@ -113,5 +115,13 @@ builder.Services.AddDbContext<OtOpcUaConfigDbContext>(opt =>
     opt.UseSqlServer(options.ConfigDbConnectionString));
 builder.Services.AddHostedService<HostStatusPublisher>();
 
+// Phase 7 follow-up #246 — historian sink + engine composer. NullAlarmHistorianSink
+// is the default until the Galaxy.Host SqliteStoreAndForwardSink writer adapter
+// lands (task #248). The composer reads Script/VirtualTag/ScriptedAlarm rows on
+// generation bootstrap, builds the engines, and starts the driver-bridge feed.
+builder.Services.AddSingleton<IAlarmHistorianSink>(NullAlarmHistorianSink.Instance);
+builder.Services.AddSingleton(Log.Logger);  // Serilog root for ScriptLoggerFactory
+builder.Services.AddSingleton<Phase7Composer>();
+
 var host = builder.Build();
 await host.RunAsync();

tests/ZB.MOM.WW.OtOpcUa.Server.Tests/Phase7/Phase7ComposerMappingTests.cs

@@ -0,0 +1,93 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
+using ZB.MOM.WW.OtOpcUa.Core.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Server.Phase7;
+
+namespace ZB.MOM.WW.OtOpcUa.Server.Tests.Phase7;
+
+/// <summary>
+///     Task #246 — covers the deterministic mapping inside <see cref="Phase7Composer"/>
+///     that turns <see cref="EquipmentNamespaceContent"/> into the path → fullRef map
+///     <see cref="DriverFeed.PathToFullRef"/> consumes. Pure function; no DI / DB needed.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class Phase7ComposerMappingTests
+{
+    private static UnsArea Area(string id, string name) =>
+        new() { UnsAreaId = id, ClusterId = "c", Name = name, GenerationId = 1 };
+
+    private static UnsLine Line(string id, string areaId, string name) =>
+        new() { UnsLineId = id, UnsAreaId = areaId, Name = name, GenerationId = 1 };
+
+    private static Equipment Eq(string id, string lineId, string name) => new()
+    {
+        EquipmentRowId = Guid.NewGuid(), GenerationId = 1, EquipmentId = id,
+        EquipmentUuid = Guid.NewGuid(), DriverInstanceId = "drv",
+        UnsLineId = lineId, Name = name, MachineCode = "m",
+    };
+
+    private static Tag T(string id, string name, string fullRef, string equipmentId) => new()
+    {
+        TagRowId = Guid.NewGuid(), GenerationId = 1, TagId = id,
+        DriverInstanceId = "drv", EquipmentId = equipmentId,
+        Name = name, DataType = "Float32",
+        AccessLevel = TagAccessLevel.Read, TagConfig = fullRef,
+    };
+
+    [Fact]
+    public void Maps_tag_to_UNS_path_walker_emits()
+    {
+        var content = new EquipmentNamespaceContent(
+            Areas: [Area("a1", "warsaw")],
+            Lines: [Line("l1", "a1", "oven-line")],
+            Equipment: [Eq("e1", "l1", "oven-3")],
+            Tags: [T("t1", "Temp", "DR.Temp", "e1")]);
+
+        var map = Phase7Composer.MapPathsToFullRefs(content);
+
+        map.ShouldContainKeyAndValue("/warsaw/oven-line/oven-3/Temp", "DR.Temp");
+    }
+
+    [Fact]
+    public void Skips_tag_with_null_EquipmentId()
+    {
+        var content = new EquipmentNamespaceContent(
+            [Area("a1", "warsaw")], [Line("l1", "a1", "ol")], [Eq("e1", "l1", "ov")],
+            [T("t1", "Bare", "DR.Bare", null!)]);  // SystemPlatform-style orphan
+
+        Phase7Composer.MapPathsToFullRefs(content).ShouldBeEmpty();
+    }
+
+    [Fact]
+    public void Skips_tag_pointing_at_unknown_Equipment()
+    {
+        var content = new EquipmentNamespaceContent(
+            [Area("a1", "warsaw")], [Line("l1", "a1", "ol")], [Eq("e1", "l1", "ov")],
+            [T("t1", "Lost", "DR.Lost", "e-missing")]);
+
+        Phase7Composer.MapPathsToFullRefs(content).ShouldBeEmpty();
+    }
+
+    [Fact]
+    public void Maps_multiple_tags_under_same_equipment_distinctly()
+    {
+        var content = new EquipmentNamespaceContent(
+            [Area("a1", "site")], [Line("l1", "a1", "line1")], [Eq("e1", "l1", "cell")],
+            [T("t1", "Temp", "DR.T", "e1"), T("t2", "Pressure", "DR.P", "e1")]);
+
+        var map = Phase7Composer.MapPathsToFullRefs(content);
+
+        map.Count.ShouldBe(2);
+        map["/site/line1/cell/Temp"].ShouldBe("DR.T");
+        map["/site/line1/cell/Pressure"].ShouldBe("DR.P");
+    }
+
+    [Fact]
+    public void Empty_content_yields_empty_map()
+    {
+        Phase7Composer.MapPathsToFullRefs(new EquipmentNamespaceContent([], [], [], []))
+            .ShouldBeEmpty();
+    }
+}