using Akka.Actor; using Akka.Hosting; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using ZB.MOM.WW.OtOpcUa.Commons.Engines; using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Configuration; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; using ZB.MOM.WW.OtOpcUa.Core.Scripting; using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; using ZB.MOM.WW.OtOpcUa.OpcUaServer; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.Health; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; namespace ZB.MOM.WW.OtOpcUa.Runtime; public static class ServiceCollectionExtensions { public const string DriverRole = "driver"; public const string DriverHostActorName = "driver-host"; public const string DbHealthProbeActorName = "db-health"; public const string HistorianAdapterActorName = "historian-adapter"; public const string DependencyMuxActorName = "dependency-mux"; public const string OpcUaPublishActorName = "opcua-publish"; public const string PeerProbeSupervisorName = "peer-probe-supervisor"; public const string ContinuousHistorizationRecorderActorName = "continuous-historization-recorder"; /// /// Registers shared runtime services. Currently binds /// to as the default; production deployments /// override this with SqliteStoreAndForwardSink wrapping the HistorianGateway alarm writer. /// Call this BEFORE AddAkka. /// /// The service collection to register with. /// The same instance for chaining. public static IServiceCollection AddOtOpcUaRuntime(this IServiceCollection services) { services.TryAddSingleton(NullAlarmHistorianSink.Instance); services.TryAddSingleton(NullHistorianDataSource.Instance); // Historian tag provisioning. Null default (no-op) so the AddressSpaceApplier resolves a real // IHistorianProvisioning only when AddHistorianProvisioning registers the gateway-backed one // (ServerHistorian:Enabled). TryAddSingleton so the gateway registration wins last. services.TryAddSingleton(NullHistorianProvisioning.Instance); // VirtualTag historization sink. Null default — the durable AVEVA sink is infra-gated (there is // no live-data historian write RPC). TryAddSingleton so a deployment that bound a real // IHistoryWriter earlier wins. services.TryAddSingleton(NullHistoryWriter.Instance); services.TryAddSingleton(NullDriverFactory.Instance); services.TryAddSingleton(NullOpcUaAddressSpaceSink.Instance); services.TryAddSingleton(NullServiceLevelPublisher.Instance); services.TryAddSingleton(); return services; } /// /// Config-gated durable alarm-historian sink. When the AlarmHistorian section has /// Enabled=true, registers a (draining via the /// -supplied writer) as the , /// overriding the default. Otherwise a no-op (Null stays). /// The writer is injected so the durable downstream (the HistorianGateway alarm writer) can be /// supplied by the Host, which is the only project that references it. /// /// The service collection to register with. /// The configuration carrying the AlarmHistorian section. /// /// Factory the Host supplies to build the concrete /// (the HistorianGateway alarm writer) from the bound options + the resolving provider. /// /// The same instance for chaining. public static IServiceCollection AddAlarmHistorian( this IServiceCollection services, IConfiguration configuration, Func writerFactory) { var opts = configuration.GetSection(AlarmHistorianOptions.SectionName).Get(); if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime foreach (var warning in opts.Validate()) Serilog.Log.Logger.ForContext().Warning("Historian config: {HistorianConfigWarning}", warning); services.AddSingleton(sp => { // SqliteStoreAndForwardSink takes a Serilog ILogger (not Microsoft.Extensions.Logging). // Resolve it off the host's configured static logger so the drain worker's WARN/INFO // lines land in the same sinks as the rest of the process. var sink = new SqliteStoreAndForwardSink( opts.DatabasePath, writerFactory(opts, sp), Serilog.Log.Logger.ForContext(), batchSize: opts.BatchSize, capacity: opts.Capacity, deadLetterRetention: TimeSpan.FromDays(opts.DeadLetterRetentionDays), maxAttempts: opts.MaxAttempts); sink.StartDrainLoop(TimeSpan.FromSeconds(opts.DrainIntervalSeconds)); return sink; }); return services; } /// /// Config-gated server-side HistoryRead backend. When the ServerHistorian section has /// Enabled=true, registers the -supplied /// (the read-only HistorianGateway-backed data source) overriding /// the default from . Otherwise /// a no-op (the Null default stays and the node manager's HistoryRead returns /// GoodNoData-empty). The data source is injected so the gateway-backed client can be supplied /// by the Host, which is the only project that references the driver. /// /// The service collection to register with. /// The configuration carrying the ServerHistorian section. /// /// Factory the Host supplies to build the concrete read /// (the gateway-backed data source) from the bound options + the resolving provider. /// /// The same instance for chaining. public static IServiceCollection AddServerHistorian( this IServiceCollection services, IConfiguration configuration, Func dataSourceFactory) { var opts = configuration.GetSection(ServerHistorianOptions.SectionName).Get(); if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime foreach (var warning in opts.Validate()) Serilog.Log.Logger.ForContext().Warning("ServerHistorian config: {ServerHistorianConfigWarning}", warning); // Last-registration-wins over the TryAddSingleton Null default seeded by AddOtOpcUaRuntime. services.AddSingleton(sp => dataSourceFactory(opts, sp)); return services; } /// /// Config-gated historian tag provisioning. When the ServerHistorian section has /// Enabled=true, registers the -supplied /// (the gateway-backed GatewayTagProvisioner that calls /// the gateway's EnsureTags) overriding the default from /// . Otherwise a no-op (the Null default stays and deploying historized /// tags provisions nothing). The provisioner is consumed by the AddressSpaceApplier, which fires /// a non-blocking for added historized tags on /// every deploy. Gated on the same ServerHistorian:Enabled flag as the read path /// () — provisioning targets the same single gateway. The provisioner /// is injected so the gateway-backed client can be supplied by the Host, which is the only project that /// references the driver. /// /// The service collection to register with. /// The configuration carrying the ServerHistorian section. /// /// Factory the Host supplies to build the concrete /// (the gateway-backed provisioner) from the bound options + the resolving provider. /// /// The same instance for chaining. public static IServiceCollection AddHistorianProvisioning( this IServiceCollection services, IConfiguration configuration, Func provisioningFactory) { var opts = configuration.GetSection(ServerHistorianOptions.SectionName).Get(); if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime // Last-registration-wins over the TryAddSingleton Null default seeded by AddOtOpcUaRuntime. services.AddSingleton(sp => provisioningFactory(opts, sp)); return services; } /// /// Spawns the per-node driver-role actors on the host's : /// (one per node), /// (consumed by the health endpoint + redundancy calc), and /// wrapping the registered . /// /// Mirror of WithOtOpcUaControlPlaneSingletons for the driver role. Both must /// be registered on the same as the cluster /// bootstrap so the actors share the host's ActorSystem. /// /// Wire from the fused Host's Program.cs when the node carries the driver role: /// /// services.AddOtOpcUaRuntime(); /// services.AddAkka("otopcua", (ab, sp) => { ab.WithOtOpcUaClusterBootstrap(sp); if (hasDriver) ab.WithOtOpcUaRuntimeActors(); }); /// /// /// The Akka configuration builder. /// The same instance for chaining. public static AkkaConfigurationBuilder WithOtOpcUaRuntimeActors(this AkkaConfigurationBuilder builder) { // Production cluster HOCON (akka.conf) carries this dispatcher block, but consumers that // bootstrap their own HOCON (e.g. ServiceCollectionExtensionsTests) wouldn't pick it up // — OpcUaPublishActor.Props pins itself to opcua-synchronized-dispatcher and Akka throws // ConfigurationException if it doesn't exist. Prepend a fallback so the runtime extension // is self-contained. builder.AddHocon(@" opcua-synchronized-dispatcher { type = ""PinnedDispatcher"" executor = ""thread-pool-executor"" throughput = 1 } ", HoconAddMode.Prepend); builder.WithActors((system, registry, resolver) => { var dbFactory = resolver.GetService>(); var roleInfo = resolver.GetService(); // Fallback to Null* if AddOtOpcUaRuntime wasn't called (e.g., test harnesses). var historianSink = resolver.GetService() ?? NullAlarmHistorianSink.Instance; var driverFactory = resolver.GetService() ?? NullDriverFactory.Instance; // Phase 6.1 resilience-invoker factory — bound by the fused Host (which references Core + // Polly); pass-through on nodes/harnesses that didn't register it. Resolved through the // Core.Abstractions seam so this Polly-free assembly stays that way. var invokerFactory = resolver.GetService() ?? NullDriverCapabilityInvokerFactory.Instance; var addressSpaceSink = resolver.GetService() ?? NullOpcUaAddressSpaceSink.Instance; var serviceLevel = resolver.GetService() ?? NullServiceLevelPublisher.Instance; var loggerFactory = resolver.GetService() ?? NullLoggerFactory.Instance; var healthPublisher = resolver.GetService() ?? NullDriverHealthPublisher.Instance; // Root script logger backs the ScriptedAlarm host's engine + script logging. Registered in // Host DI inside the hasDriver block; may be absent in some role configs / test harnesses, // in which case the DriverHostActor gracefully skips spawning the ScriptedAlarm host. var scriptRootLogger = resolver.GetService(); // Production evaluator is the Host's RoslynVirtualTagEvaluator (registered as // IVirtualTagEvaluator); fall back to the null evaluator for test harnesses that don't // register one (VirtualTagActor children then evaluate to nothing). var virtualTagEvaluator = resolver.GetService(); if (virtualTagEvaluator is null) { loggerFactory.CreateLogger("ZB.MOM.WW.OtOpcUa.Runtime.ServiceCollectionExtensions") .LogWarning("IVirtualTagEvaluator not registered; Equipment VirtualTags will evaluate to NoChange (no live values). Expected only in test harnesses — driver-role nodes should register RoslynVirtualTagEvaluator."); virtualTagEvaluator = NullVirtualTagEvaluator.Instance; } // VirtualTag historization sink threaded to the spawned VirtualTagHostActor. Null default // (durable AVEVA sink is infra-gated); a deployment binding a real IHistoryWriter overrides. var historyWriter = resolver.GetService() ?? NullHistoryWriter.Instance; var dbHealth = system.ActorOf( DbHealthProbeActor.Props(dbFactory), DbHealthProbeActorName); registry.Register(dbHealth); // Dependency mux must be spawned before DriverHostActor so the host can forward // AttributeValuePublished into it from the very first driver spawn. var mux = system.ActorOf(DependencyMuxActor.Props(), DependencyMuxActorName); registry.Register(mux); // Continuous-historization recorder — gated on ContinuousHistorization:Enabled AND the // gateway-backed IHistorianValueWriter + the durable IHistorizationOutbox being registered // (the Host registers both ONLY when historization is enabled and the ServerHistorian gateway // is configured). The recorder taps the dependency mux's value fan-out, so it is spawned after // (and fed) the same `mux` ref the DriverHostActor uses. It is spawned BEFORE the applier so // the applier's historized-ref subscription sink can wrap this recorder's IActorRef and feed it // the add/remove delta of historized refs on every deploy (closing the T18 ref-feed gap). IActorRef? continuousRecorder = null; var continuousOptions = resolver.GetService(); if (continuousOptions is { Enabled: true }) { var valueWriter = resolver.GetService(); var outbox = resolver.GetService(); if (valueWriter is not null && outbox is not null) { // Initial ref set is EMPTY: the deployed address space (and thus the historized-ref // set) is built later at deploy time, not here. The applier's per-deploy add/remove // feed populates the recorder's interest from that point on. continuousRecorder = system.ActorOf( ContinuousHistorizationRecorder.Props( dependencyMux: mux, writer: valueWriter, outbox: outbox, historizedRefs: Array.Empty(), drainBatchSize: continuousOptions.DrainBatchSize, drainInterval: TimeSpan.FromSeconds(continuousOptions.DrainIntervalSeconds), minBackoff: TimeSpan.FromSeconds(continuousOptions.MinBackoffSeconds), maxBackoff: TimeSpan.FromSeconds(continuousOptions.MaxBackoffSeconds)), ContinuousHistorizationRecorderActorName); registry.Register(continuousRecorder); } else { loggerFactory.CreateLogger("ZB.MOM.WW.OtOpcUa.Runtime.ServiceCollectionExtensions") .LogWarning("ContinuousHistorization is enabled but IHistorianValueWriter and/or IHistorizationOutbox are not registered; the recorder will not be spawned. Expected only in misconfigured deployments or test harnesses."); } } // Historized-ref subscription sink fed by the applier on every deploy. When the recorder was // spawned, an adapter wraps its IActorRef (a non-blocking Tell of the add/remove delta); // otherwise the Null no-op sink, so the applier behaves identically when historization is off. IHistorizedTagSubscriptionSink historizedSubscriptions = continuousRecorder is not null ? new ActorHistorizedTagSubscriptionSink(continuousRecorder) : NullHistorizedTagSubscriptionSink.Instance; // Historian tag provisioner fed to the applier so deploying historized tags auto-ensures // them in the historian (EnsureTags). The Host registers the gateway-backed provisioner via // AddHistorianProvisioning when ServerHistorian:Enabled; otherwise this resolves the no-op // NullHistorianProvisioning default seeded by AddOtOpcUaRuntime (so the applier's hook is inert). var provisioning = resolver.GetService(); // OPC UA publish actor — pinned dispatcher, owns the address-space side of the // pipeline. AddressSpaceApplier is constructed here so the actor + applier share the // same sink reference (when DeferredAddressSpaceSink swaps later, both see it). var applier = new AddressSpaceApplier( addressSpaceSink, loggerFactory.CreateLogger(), provisioning: provisioning, historizedSubscriptions: historizedSubscriptions); var publishActor = system.ActorOf( OpcUaPublishActor.Props( sink: addressSpaceSink, serviceLevel: serviceLevel, localNode: roleInfo.LocalNode, dbFactory: dbFactory, applier: applier, dbHealthProbe: dbHealth), OpcUaPublishActorName); registry.Register(publishActor); // Per-node peer-probe supervisor — keeps one OPC UA TCP probe per OTHER non-Detached // driver node, so every node is continuously probed by all its peers. OpcUaPublishActor // consumes the resulting probe verdicts to learn this node's own reachability. var peerProbes = system.ActorOf( PeerProbeSupervisor.Props(roleInfo.LocalNode), PeerProbeSupervisorName); registry.Register(peerProbes); var driverHost = system.ActorOf( DriverHostActor.Props(dbFactory, roleInfo.LocalNode, coordinator: null, driverFactory: driverFactory, localRoles: roleInfo.LocalRoles, dependencyMux: mux, opcUaPublishActor: publishActor, healthPublisher: healthPublisher, virtualTagEvaluator: virtualTagEvaluator, historyWriter: historyWriter, loggerFactory: loggerFactory, scriptRootLogger: scriptRootLogger, invokerFactory: invokerFactory), DriverHostActorName); registry.Register(driverHost); var historian = system.ActorOf( HistorianAdapterActor.Props(historianSink, roleInfo.LocalNode), HistorianAdapterActorName); registry.Register(historian); }); return builder; } } /// Marker key types used by Akka.Hosting to resolve runtime actors from the registry. public sealed class DriverHostActorKey { } public sealed class DbHealthProbeActorKey { } public sealed class HistorianAdapterActorKey { } public sealed class DependencyMuxActorKey { } public sealed class OpcUaPublishActorKey { } /// Marker key for the per-node ContinuousHistorizationRecorder (spawned only when /// ContinuousHistorization:Enabled=true and the gateway value-writer + outbox are registered). public sealed class ContinuousHistorizationRecorderKey { } /// Marker key for the per-node PeerProbeSupervisor. public sealed class PeerProbeSupervisorKey { }