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));
// 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();
// 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();
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()));
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;
}
}