using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using ZB.MOM.WW.Audit; using ZB.MOM.WW.Configuration; using ZB.MOM.WW.ScadaBridge.AuditLog.Central; using ZB.MOM.WW.ScadaBridge.AuditLog.Configuration; using ZB.MOM.WW.ScadaBridge.AuditLog.Kpi; using ZB.MOM.WW.ScadaBridge.AuditLog.Payload; using ZB.MOM.WW.ScadaBridge.AuditLog.Redaction; using ZB.MOM.WW.ScadaBridge.AuditLog.Site; using ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; using IAuditWriter = ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.IAuditWriter; namespace ZB.MOM.WW.ScadaBridge.AuditLog; /// /// Composition root for the Audit Log component. /// /// /// /// Registered + the validator. Extends the /// surface with the site-side writer chain /// ( + + /// ) and the telemetry collaborators /// (, , /// , , /// ). /// /// /// Audit Log sits alongside Notification Outbox and Site Call /// Audit. IAuditLogRepository is registered by /// ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.ServiceCollectionExtensions.AddConfigurationDatabase, /// so the caller (the Host on the central node) must also call that. /// /// public static class ServiceCollectionExtensions { /// Configuration section bound to . public const string ConfigSectionName = "AuditLog"; /// Configuration section bound to . public const string SiteWriterSectionName = "AuditLog:SiteWriter"; /// Configuration section bound to . public const string SiteTelemetrySectionName = "AuditLog:SiteTelemetry"; /// Configuration section bound to . public const string PartitionMaintenanceSectionName = "AuditLog:PartitionMaintenance"; /// Configuration section bound to . public const string PurgeSectionName = "AuditLog:Purge"; /// Configuration section bound to . public const string ReconciliationSectionName = "AuditLog:Reconciliation"; /// /// Registers the Audit Log component services: options, the site /// SQLite writer chain (primary + ring fallback + failure-counter sink), /// and the site-→central telemetry collaborators. Idempotent re-registration /// is not supported; call this exactly once per . /// /// The service collection to register into. /// Application configuration used to bind and related options sections. /// The same for chaining. public static IServiceCollection AddAuditLog(this IServiceCollection services, IConfiguration config) { ArgumentNullException.ThrowIfNull(services); ArgumentNullException.ThrowIfNull(config); // Top-level AuditLogOptions + validator (redaction policy, payload caps, etc.). // Collapsed onto the shared ZB.MOM.WW.Configuration helper: it binds the // "AuditLog" section, registers the validator, and enables ValidateOnStart in // one call. Same section path as before; AddAuditLog is call-once per // collection, and the helper's TryAddEnumerable is idempotent for the // validator (a strict improvement over the previous AddSingleton). services.AddValidatedOptions(config, ConfigSectionName); // The canonical IAuditRedactor is wired as // ScadaBridgeAuditRedactor — same truncation + header / body / // SQL-parameter redaction as the original pipeline, applied between // event construction and persistence. Singleton — stateless; the // IOptionsMonitor dependency picks up hot reloads on its own. services.AddSingleton(); // Per-stage redactor-failure counter. NoOp default; this binding is // later replaced with the Site Health Monitoring bridge that surfaces // failures as AuditRedactionFailure on the site health report. services.TryAddSingleton(); // Site writer + telemetry options bindings. // BindConfiguration is not used because the configuration root supplied // by the caller may not be the application root — we go through the // section explicitly so a partial IConfiguration (e.g. a test stub // anchored on the AuditLog section's parent) still works. services.AddOptions() .Bind(config.GetSection(SiteWriterSectionName)); services.AddOptions() .Bind(config.GetSection(SiteTelemetrySectionName)); // Retention purge options (the SiteAuditRetentionService hosted job is // registered site-only in AddAuditLogHealthMetricsBridge). Binding here is // inert on central — no hosted service consumes it there. services.AddOptions() .Bind(config.GetSection(SiteAuditRetentionOptions.SectionName)); // SqliteAuditWriter is a singleton with a single owned SqliteConnection // and a background writer Task; multiple instances would race on the // same file. Registered concretely so the ISiteAuditQueue + IAuditWriter // forwards below resolve to the same instance — the actor must observe // the writes made via the hot-path interface. services.AddSingleton(); services.AddSingleton(sp => sp.GetRequiredService()); // RingBufferFallback: drop-oldest in-memory ring used by // FallbackAuditWriter when the primary SQLite writer throws. Default // capacity is fine (1024). services.AddSingleton(); // IAuditWriteFailureCounter: NoOp default. This binding is later // overridden with the real Site Health Monitoring counter. Registered // before FallbackAuditWriter so the factory can resolve it. services.AddSingleton(); // The script-thread surface is FallbackAuditWriter (primary + ring + // counter), not the raw SqliteAuditWriter — primary failures must NEVER // abort the user-facing action. // The canonical IAuditRedactor singleton above is wired // through the factory so every event written through this surface is // truncated + redacted before it hits SQLite (and the ring on // failure). services.AddSingleton(sp => new FallbackAuditWriter( primary: sp.GetRequiredService(), ring: sp.GetRequiredService(), failureCounter: sp.GetRequiredService(), logger: sp.GetRequiredService>(), redactor: sp.GetRequiredService())); // ISiteStreamAuditClient: NoOp default. This binding remains correct for // central/test composition roots that have no SiteCommunicationActor. // The real implementation is ClusterClientSiteAuditClient, which pushes // audit telemetry to central over Akka ClusterClient via the site's // SiteCommunicationActor — the Host wires it directly into the // SiteAuditTelemetryActor's Props.Create call for site roles (it cannot // be a DI singleton because it needs the SiteCommunicationActor IActorRef, // created during Akka bootstrap, not at DI-composition time). services.AddSingleton(); // Site-side dual emitter for cached-call lifecycle // telemetry. ScriptRuntimeContext.ExternalSystem.CachedCall / // Database.CachedWrite resolves this through DI and pushes one combined // packet per lifecycle event; the forwarder writes the audit half // through IAuditWriter and the operational half through the // IOperationTrackingStore. The audit writer is always wired (the // chain above); the operational tracking store is SITE-ONLY (registered // by ZB.MOM.WW.ScadaBridge.SiteRuntime). On a Central composition root the tracking // store has no registration, so the factory resolves it with GetService // (returning null) — the forwarder degrades to "audit-only" emission, // mirroring the lazy IAuditWriter chain established above. services.AddSingleton(sp => new CachedCallTelemetryForwarder( sp.GetRequiredService(), sp.GetService(), sp.GetRequiredService>(), // INodeIdentityProvider is now required across // every consumer in AddAuditLog. The Host's // SiteServiceRegistration registers it as a singleton on both // site and central paths, and the AddAuditLogTests fixture provides a // FakeNodeIdentityProvider; a silent GetService() that // returned null would mask a future composition root that // forgot to register the provider. sp.GetRequiredService())); // Bridge the store-and-forward retry-loop observer hook // to the cached-call forwarder so per-attempt + terminal telemetry // emitted from the S&F retry sweep lands on the same SQLite hot-path // as the script-thread CachedSubmit row. Registered as a singleton // and also bound to ICachedCallLifecycleObserver so AddStoreAndForward // can resolve it through DI. // SourceNode-stamping: factory-resolved so the // INodeIdentityProvider singleton can be threaded through — the // bridge stamps SiteCallOperational.SourceNode from // INodeIdentityProvider.NodeName on every cached-call lifecycle row. // The provider is resolved with GetRequiredService — // SiteServiceRegistration.BindSharedOptions registers it on both // site and central paths, so a missing registration is a // composition-root bug, not a silent null-SourceNode degradation. services.AddSingleton(sp => new CachedCallLifecycleBridge( sp.GetRequiredService(), sp.GetRequiredService>(), // Required, matches the other consumers in this // composition root — the provider is always registered by // SiteServiceRegistration. sp.GetRequiredService())); services.AddSingleton( sp => sp.GetRequiredService()); // Central audit-write failure counter — NoOp default // for site/test composition roots that don't wire the central health // snapshot. AddAuditLogCentralMaintenance below replaces this binding // with the AuditCentralHealthSnapshot implementation so increments // surface on the central dashboard. services.TryAddSingleton(); // Inbound body-ceiling hit counter — NoOp default for // site/test roots. AddAuditLogCentralMaintenance replaces this binding // with the AuditCentralHealthSnapshot implementation so ceiling-hit // counts surface on the central dashboard alongside write-failure and // redaction-failure counters. services.TryAddSingleton(); // AuditLog purge-failure counter — NoOp default for site/test // composition roots. AddAuditLogCentralMaintenance below replaces this binding // with the AuditCentralHealthSnapshot implementation so partition/channel purge // failures surface on the central dashboard alongside the write-failure counter. services.TryAddSingleton(); // Central direct-write audit writer used by // NotificationOutboxActor and Inbound API to // emit AuditLog rows that originate ON central, not via site telemetry. // Singleton — the writer is stateless; its per-call scope opens a fresh // IAuditLogRepository (a SCOPED EF Core service registered by // ZB.MOM.WW.ScadaBridge.ConfigurationDatabase). The interface (ICentralAuditWriter) // is intentionally distinct from IAuditWriter so site composition roots // do not accidentally bind it; central composition roots that include // AddConfigurationDatabase get a working implementation transparently. // Wire the canonical IAuditRedactor into the factory so // NotificationOutboxActor + Inbound API rows are truncated + redacted // before they hit MS SQL. // Also wire the ICentralAuditWriteFailureCounter // so swallowed repo throws bump the central health counter. services.AddSingleton(sp => new CentralAuditWriter( sp, sp.GetRequiredService>(), sp.GetRequiredService(), sp.GetRequiredService(), // SourceNode-stamping: wire the local node identity so // central-origin rows (Notification Outbox dispatch, Inbound API) // carry the writing node's identifier when the caller hasn't // already supplied one. GetRequiredService — the production // composition root in SiteServiceRegistration registers the // provider as a singleton on both site and central paths. sp.GetRequiredService())); // KPI History & Trends: the AuditLog source the central KPI // recorder enumerates each sampling pass to snapshot Global-scope Audit // Log KPIs (volume / error rate / backlog) into the KpiSample history // table. Scoped to match its IAuditLogRepository dependency — a SCOPED // EF Core service; the recorder opens a fresh scope per sampling pass. // TryAdd-Enumerable so multiple sources can register the same interface // (NotificationOutbox + SiteCallAudit + SiteHealth add their own) without // any one composition root clobbering another, and re-registration is a // no-op if AddAuditLog were ever called twice. services.TryAddEnumerable( ServiceDescriptor.Scoped()); return services; } /// /// Swap the default /// and /// registrations for the /// real / /// bridges so the /// FallbackAuditWriter primary-failure counter AND the /// redactor-failure counter both surface in the /// site health report payload as /// SiteHealthReport.SiteAuditWriteFailures + /// SiteHealthReport.AuditRedactionFailure. /// /// /// /// Must be called AFTER both (registers the /// NoOp defaults this method replaces) and /// ZB.MOM.WW.ScadaBridge.HealthMonitoring.ServiceCollectionExtensions.AddHealthMonitoring /// or AddSiteHealthMonitoring (registers the /// the bridges depend on). Resolving /// or /// without the latter throws /// at GetRequiredService /// time — by design, since a silent NoOp would mask a misconfiguration. /// /// /// Idempotent — a sentinel check on the /// hosted-service descriptor /// short-circuits subsequent calls so the hosted service is not /// double-registered (AddHostedService has no TryAdd variant). /// /// /// Site-side only: the central composition root keeps the NoOp /// defaults; the central health-metric surface that would expose /// AuditRedactionFailure next to the existing central counters /// ships later. /// /// /// The service collection to register into. /// The same for chaining. public static IServiceCollection AddAuditLogHealthMetricsBridge(this IServiceCollection services) { ArgumentNullException.ThrowIfNull(services); // Guard against double-registration. AddHostedService is // additive (no TryAdd variant) so a second call without this sentinel // would spin up a second SiteAuditBacklogReporter, doubling the 30 s // SQL probe rate and racing on the ISiteHealthCollector snapshot. The // SiteAuditBacklogReporter descriptor is the discriminator: it's only // registered by this helper, so its presence proves the bridge has // already been wired. if (services.Any(d => d.ImplementationType == typeof(SiteAuditBacklogReporter))) { return services; } services.Replace( ServiceDescriptor.Singleton()); services.Replace( ServiceDescriptor.Singleton()); // The site-side backlog reporter polls the // SqliteAuditWriter every 30 s and pushes the snapshot into the // collector so the next SiteHealthReport carries a fresh // SiteAuditBacklog field. Registered alongside the other site-only // metric bridges so AddAuditLog (which runs on central too) stays // free of hosted-service registrations that would resolve a missing // ISiteHealthCollector on central. services.AddHostedService(); // Site-only retention purge: runs ISiteAuditQueue.PurgeExpiredAsync on a // daily timer so the ephemeral site SQLite store does not grow unbounded. // Registered here (not in AddAuditLog, which also runs on central) so the // hosted service only starts on site composition roots. Guarded by the same // SiteAuditBacklogReporter sentinel above against double-registration. services.AddHostedService(); return services; } /// /// Central-only registration for the /// hosted service plus /// its binding. Must be /// called from the Central role's composition root (not from a site /// composition root); the underlying IPartitionMaintenance /// implementation is registered by AddConfigurationDatabase and /// only exists on the central node. /// /// /// /// Separated from because AddAuditLog is /// also invoked from site composition roots — silently starting a /// hosted service that resolves an unregistered dependency on a site /// would fail every tick. Keeping the central-only registration in its /// own helper preserves the "every Add* call is safe to issue /// from any composition root" invariant. /// /// /// The service collection to register into. /// Application configuration used to bind partition maintenance options. /// The same for chaining. public static IServiceCollection AddAuditLogCentralMaintenance( this IServiceCollection services, IConfiguration config) { ArgumentNullException.ThrowIfNull(services); ArgumentNullException.ThrowIfNull(config); services.AddOptions() .Bind(config.GetSection(PartitionMaintenanceSectionName)); services.AddHostedService(); // I1 (review): bind the two central-singleton options HERE rather than in // AddAuditLogCentralReconciliationClient. AkkaHostedService.RegisterCentralActors // resolves IOptions / // via GetRequiredService when it wires the AuditLogPurgeActor + // SiteAuditReconciliationActor singletons; AddAuditLogCentralMaintenance is // ALWAYS called on the central path (the reconciliation-client helper is the // one that could in principle be dropped), so binding the options here means // the singletons get a valid IOptions even if the gRPC-client helper is not // wired — instead of a cryptic InvalidOperationException at GetRequiredService. // Defaults are fine when the section is absent (24 h purge cadence / // 5 min reconciliation tick); production exposes IntervalHours / // ReconciliationIntervalSeconds only — the test-only *Override knobs are // not intended to be set from config (see the options classes' remarks). services.AddOptions() .Bind(config.GetSection(PurgeSectionName)); services.AddOptions() .Bind(config.GetSection(ReconciliationSectionName)); // Central health snapshot — a single object // that owns the CentralAuditWriteFailures + AuditRedactionFailure // Interlocked counters AND surfaces them on // IAuditCentralHealthSnapshot. The same instance is bound to BOTH // writer-side interfaces (ICentralAuditWriteFailureCounter + // IAuditRedactionFailureCounter) so every central-side increment // routes into the shared counters; site nodes keep their existing // Site bridges (registered by AddAuditLogHealthMetricsBridge) so // the same counter type does not shadow the site-side metric. // The snapshot itself has no actor-system dependency — the // per-site stalled latch is fed by SiteAuditTelemetryStalledTracker // which the Akka bootstrap wires up after ActorSystem.Create returns // (the tracker is NOT registered here because its construction // requires ActorSystem, which is not a DI-resolvable singleton). services.AddSingleton(); services.AddSingleton( sp => sp.GetRequiredService()); services.Replace(ServiceDescriptor.Singleton( sp => sp.GetRequiredService())); // Override the NoOp IAuditRedactionFailureCounter // (registered by AddAuditLog) with the CentralAuditRedactionFailureCounter // bridge so payload-filter throws on CentralAuditWriter / // AuditLogIngestActor paths surface on the central dashboard. The // bridge is a thin wrapper around the AuditCentralHealthSnapshot // singleton so all central redactor failures route into the same // counter as CentralAuditWriteFailures. The site composition root // overrides this binding AGAIN via AddAuditLogHealthMetricsBridge — // central nodes do not call that bridge, so this is the final // binding on a central host. Mirrors the // HealthMetricsAuditRedactionFailureCounter shape one-for-one. services.Replace(ServiceDescriptor.Singleton()); // Replace the NoOp IAuditInboundCeilingHitsCounter with the // AuditCentralHealthSnapshot so ceiling-hit counts surface on the // central dashboard. Same singleton-forward pattern as // ICentralAuditWriteFailureCounter above. services.Replace(ServiceDescriptor.Singleton( sp => sp.GetRequiredService())); // Replace the NoOp IAuditPurgeFailureCounter with the AuditCentralHealthSnapshot so // partition/channel purge failures surface on the central dashboard. Same // singleton-forward pattern as ICentralAuditWriteFailureCounter above (arch-review 04, S2). services.Replace(ServiceDescriptor.Singleton( sp => sp.GetRequiredService())); return services; } /// /// Audit Log — central-only registration of the production /// () /// and its unary-call invoker () used /// by to pull reconciliation /// batches from each site over the PullAuditEvents gRPC RPC. /// /// /// /// Kept out of — which also runs on site /// composition roots — because the client dials sites and resolves /// (a central-only collaborator wired /// alongside the reconciliation singleton). Folding it into /// would register a site-dialing client on every /// site host, violating the "every Add* call is safe from any /// composition root" invariant. This helper is the central analogue of /// . /// /// /// The binds with default /// /// keepalive unless an IOptions<CommunicationOptions> is /// already registered, in which case the configured timings flow through — /// matching how SiteStreamGrpcClientFactory takes its keepalive from /// the same options. /// /// /// The production (, /// wrapping the scoped ISiteRepository) IS registered here — so the /// singleton wired in the Host can /// resolve its enumerator + gRPC client from this central-only helper. Keeping /// the enumerator on this central path preserves the "every Add* call is /// safe from any composition root" invariant: a site host never calls this /// helper, so it never registers a site-dialing enumerator. The /// + /// bindings live in instead (I1 /// review fix) — that helper is unconditionally called on the central path, so /// the two maintenance singletons get a valid IOptions even if this /// gRPC-client helper is ever dropped. /// /// /// The service collection to register into. /// Application configuration used to bind the gRPC client's communication options (purge + reconciliation options are bound by ). /// The same for chaining. public static IServiceCollection AddAuditLogCentralReconciliationClient( this IServiceCollection services, IConfiguration config) { ArgumentNullException.ThrowIfNull(services); ArgumentNullException.ThrowIfNull(config); // Production ISiteEnumerator: projects the config-DB Site rows into the // reconciliation targets the SiteAuditReconciliationActor polls. Scoped // ISiteRepository is resolved per call inside the enumerator, so the // singleton takes the ROOT provider (mirrors the per-tick scope pattern // in SiteAuditReconciliationActor / AuditLogPurgeActor). services.TryAddSingleton(sp => new SiteEnumerator(sp)); // I1 (review): the AuditLogPurgeOptions / SiteAuditReconciliationOptions // bindings moved to AddAuditLogCentralMaintenance — that helper is always // called on the central path, so the two maintenance singletons resolve a // valid IOptions even if this gRPC-client helper is ever dropped. Keep the // ISiteEnumerator + gRPC client registrations here (they dial sites and are // central-only by design). // The invoker owns the per-endpoint GrpcChannel cache, so it must be a // singleton — a fresh invoker per resolution would leak channels. // Resolve CommunicationOptions if present (the central Host binds it), // otherwise fall back to defaults so this helper stays standalone. services.TryAddSingleton(sp => { var options = sp .GetService>(); return options is null ? new GrpcPullAuditEventsInvoker() : new GrpcPullAuditEventsInvoker(options.Value); }); services.TryAddSingleton( sp => sp.GetRequiredService()); services.TryAddSingleton(sp => new GrpcPullAuditEventsClient( sp.GetRequiredService(), sp.GetRequiredService(), sp.GetRequiredService>())); // Site Call Audit reconciliation pull client — central-only, the // sibling of the audit pull client above. Lives here (not in the // SiteCallAudit project) so it can reuse the central-only // ISiteEnumerator registered just above; SiteCallAudit does not // reference AuditLog. The invoker owns the per-endpoint GrpcChannel // cache, so it must be a singleton (a fresh invoker per resolution // would leak channels). CommunicationOptions flow through when bound by // the central Host, else defaults — mirrors the audit invoker. services.TryAddSingleton(sp => { var options = sp .GetService>(); return options is null ? new GrpcPullSiteCallsInvoker() : new GrpcPullSiteCallsInvoker(options.Value); }); services.TryAddSingleton( sp => sp.GetRequiredService()); services.TryAddSingleton(sp => new GrpcPullSiteCallsClient( sp.GetRequiredService(), sp.GetRequiredService(), sp.GetRequiredService>())); return services; } }