ScadaBridge/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs

using Akka.Actor;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Notifications;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Notifications;
using ZB.MOM.WW.ScadaBridge.NotificationOutbox.Delivery;
using ZB.MOM.WW.ScadaBridge.NotificationOutbox.Messages;

namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox;

/// <summary>
/// Central-side actor that owns the notification outbox. It accepts
/// <see cref="NotificationSubmit"/> messages forwarded from sites and persists each as a
/// <see cref="Notification"/> row (the ingest path), and runs a periodic dispatch loop
/// that claims due notifications, delivers them through the matching channel adapter, and
/// applies the resulting status transition. It also runs a periodic purge that bulk-deletes
/// terminal notification rows once they age past the configured retention window.
/// </summary>
public class NotificationOutboxActor : ReceiveActor, IWithTimers
{
    private const string DispatchTimerKey = "dispatch";
    private const string PurgeTimerKey = "purge";

    /// <summary>Retry policy fallback used when no SMTP configuration row is present.</summary>
    private const int FallbackMaxRetries = 10;
    private static readonly TimeSpan FallbackRetryDelay = TimeSpan.FromMinutes(1);

    /// <summary>
    /// Audit <c>Actor</c> stamped on central-dispatch (<c>NotifyDeliver</c>) rows.
    /// The Actor-column spec assigns central-originated audit rows a system
    /// identity — there is no per-call authenticated user at dispatch time.
    /// </summary>
    private const string SystemActor = "system";

    private readonly IServiceProvider _serviceProvider;
    private readonly NotificationOutboxOptions _options;
    private readonly ICentralAuditWriter _auditWriter;
    private readonly ILogger<NotificationOutboxActor> _logger;

    /// <summary>
    /// In-flight guard for the dispatch loop. Set true at the start of a sweep and cleared
    /// when the sweep's <see cref="InternalMessages.DispatchComplete"/> arrives. While true,
    /// further <see cref="InternalMessages.DispatchTick"/>s are dropped so sweeps never overlap.
    /// </summary>
    private bool _dispatching;

    /// <summary>
    /// Cached <see cref="NotificationType"/> → adapter lookup, built
    /// lazily on the first dispatch sweep and reused for the lifetime of the actor. The
    /// adapter registration is decided at startup by <c>AddNotificationOutbox</c> (the set is
    /// keyed by <see cref="NotificationType"/> and is static per process lifetime), so
    /// rebuilding this dictionary on every sweep was pure allocation waste.
    /// </summary>
    /// <remarks>
    /// The cache is paired with <see cref="_adaptersScope"/>, an actor-lifetime
    /// <see cref="IServiceScope"/> created on first use so the cached scoped adapter
    /// instances and their dependencies live as long as the cache itself. The scope is
    /// disposed in <see cref="PostStop"/>. The adapters are stateless wrappers that
    /// resolve their per-call collaborators (e.g. <see cref="INotificationRepository"/>'s
    /// underlying DbContext) through their own injected dependencies; holding them for
    /// the actor's lifetime is consistent with the actor's own singleton lifetime on the
    /// active central node.
    /// </remarks>
    private IReadOnlyDictionary<NotificationType, INotificationDeliveryAdapter>? _adaptersCache;

    /// <summary>
    /// Actor-lifetime DI scope that owns the cached
    /// <see cref="_adaptersCache"/> adapter instances. Created lazily on the first
    /// dispatch sweep that needs adapters; disposed in <see cref="PostStop"/> so the
    /// scoped adapter graph (and any disposable dependencies it transitively holds) is
    /// torn down with the actor.
    /// </summary>
    private IServiceScope? _adaptersScope;

    /// <summary>
    /// Lifecycle-scoped cancellation source, cancelled in <see cref="PostStop"/> so
    /// any in-flight dispatch sweep — including a long-running SMTP send via the channel
    /// adapter — observes shutdown promptly instead of blocking <c>CoordinatedShutdown</c>
    /// for the full SMTP connect/auth/send timeout per in-progress notification.
    /// </summary>
    private CancellationTokenSource? _shutdownCts;

    /// <summary>Akka timer scheduler, assigned by the actor system via <see cref="IWithTimers"/>.</summary>
    public ITimerScheduler Timers { get; set; } = null!;

    /// <summary>
    /// Initializes the actor with its dependencies and registers all message handlers.
    /// </summary>
    /// <param name="serviceProvider">DI service provider used to open scopes for database operations.</param>
    /// <param name="options">Notification outbox configuration options.</param>
    /// <param name="auditWriter">Central audit writer for recording dispatch events.</param>
    /// <param name="logger">Logger for this actor.</param>
    public NotificationOutboxActor(
        IServiceProvider serviceProvider,
        NotificationOutboxOptions options,
        ICentralAuditWriter auditWriter,
        ILogger<NotificationOutboxActor> logger)
    {
        _serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _auditWriter = auditWriter ?? throw new ArgumentNullException(nameof(auditWriter));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));

        Receive<NotificationSubmit>(HandleSubmit);
        Receive<InternalMessages.IngestPersisted>(HandleIngestPersisted);
        Receive<InternalMessages.DispatchTick>(_ => HandleDispatchTick());
        Receive<InternalMessages.DispatchComplete>(_ => _dispatching = false);
        Receive<InternalMessages.PurgeTick>(_ => HandlePurgeTick());
        // No-op: purge has no in-flight guard to lower, and the outcome is already logged
        // by the PipeTo projections, so PurgeComplete carries nothing to act on.
        Receive<InternalMessages.PurgeComplete>(_ => { });
        Receive<NotificationOutboxQueryRequest>(HandleQuery);
        Receive<NotificationStatusQuery>(HandleStatusQuery);
        Receive<NotificationDetailRequest>(HandleDetailRequest);
        Receive<RetryNotificationRequest>(HandleRetry);
        Receive<DiscardNotificationRequest>(HandleDiscard);
        Receive<NotificationKpiRequest>(HandleKpiRequest);
        Receive<PerSiteNotificationKpiRequest>(HandlePerSiteKpiRequest);
        Receive<PerNodeNotificationKpiRequest>(HandlePerNodeKpiRequest);
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        base.PreStart();
        // Shutdown token is alive for the lifetime of the actor; cancelled in PostStop
        // so dispatcher sweeps and the SMTP send beneath them observe coordinated shutdown.
        _shutdownCts = new CancellationTokenSource();
        Timers.StartPeriodicTimer(
            DispatchTimerKey, InternalMessages.DispatchTick.Instance, _options.DispatchInterval);
        Timers.StartPeriodicTimer(
            PurgeTimerKey, InternalMessages.PurgeTick.Instance, _options.PurgeInterval);
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        // Cancel the shutdown token first so the in-flight sweep's adapter call
        // observes cancellation, then dispose the source. Order matters — disposing first
        // would race with an in-flight sweep registering with the token.
        try
        {
            _shutdownCts?.Cancel();
        }
        catch (ObjectDisposedException)
        {
            // Already disposed under a restarted-actor race; nothing to do.
        }

        _shutdownCts?.Dispose();
        _shutdownCts = null;

        // Dispose the actor-lifetime adapter scope so the cached
        // scoped adapter instances and their disposable dependencies are torn down with
        // the actor (e.g. on a CoordinatedShutdown / failover that stops the singleton).
        _adaptersScope?.Dispose();
        _adaptersScope = null;
        _adaptersCache = null;

        base.PostStop();
    }

    /// <summary>
    /// Maps an inbound <see cref="NotificationSubmit"/> onto a <see cref="Notification"/>,
    /// persists it idempotently, and pipes the outcome back to <see cref="Self"/> so the
    /// ack is sent from the actor thread with the original sender preserved.
    /// </summary>
    private void HandleSubmit(NotificationSubmit msg)
    {
        var sender = Sender;

        // The success projection fires for both a fresh insert and an existing row;
        // only a thrown repository error reaches the failure projection. The list
        // lookup that stamps the channel Type happens inside PersistAsync so it can
        // share the same DI scope as the insert.
        PersistAsync(msg).PipeTo(
            Self,
            success: () => new InternalMessages.IngestPersisted(
                msg.NotificationId, sender, Succeeded: true, Error: null),
            failure: ex => new InternalMessages.IngestPersisted(
                msg.NotificationId, sender, Succeeded: false, Error: ex.GetBaseException().Message));
    }

    /// <summary>
    /// Resolves the target notification list to stamp the delivery channel
    /// <see cref="NotificationType"/> on a fresh row, then inserts the notification if a row
    /// with the same id does not already exist. The boolean result of
    /// <c>InsertIfNotExistsAsync</c> is intentionally ignored: an existing row is an
    /// idempotent re-submission and is acked just like a fresh insert (and its persisted
    /// Type is left untouched) so the site can clear its forward buffer. Only a thrown
    /// error must surface to the caller.
    /// </summary>
    private async Task PersistAsync(NotificationSubmit msg)
    {
        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();

        // The list's Type is the authoritative delivery channel. Resolve it here so a
        // fresh row is stamped with the right channel (Email/Sms/...). GetService (not
        // GetRequiredService) so the lookup is optional: a host without INotification
        // Repository, or a missing list, falls back to Email — the notification then
        // parks at delivery with "list not found", which is the unchanged behaviour.
        var listRepository = scope.ServiceProvider.GetService<INotificationRepository>();
        var list = listRepository is null
            ? null
            : await listRepository.GetListByNameAsync(msg.ListName);
        var type = list?.Type ?? NotificationType.Email;

        var notification = BuildNotification(msg, type);
        await repository.InsertIfNotExistsAsync(notification);
    }

    /// <summary>
    /// Acks the original submitter once persistence completes. <see cref="NotificationSubmitAck"/>
    /// is <c>Accepted</c> for both a fresh insert and an existing row; only a thrown
    /// repository error produces <c>Accepted: false</c> so the site retries the forward.
    /// </summary>
    private void HandleIngestPersisted(InternalMessages.IngestPersisted msg)
    {
        if (msg.Succeeded)
        {
            _logger.LogDebug("Notification {NotificationId} ingested into outbox.", msg.NotificationId);
            msg.Sender.Tell(new NotificationSubmitAck(msg.NotificationId, Accepted: true, Error: null));
        }
        else
        {
            _logger.LogWarning(
                "Failed to ingest notification {NotificationId}: {Error}",
                msg.NotificationId, msg.Error);
            msg.Sender.Tell(new NotificationSubmitAck(msg.NotificationId, Accepted: false, Error: msg.Error));
        }
    }

    /// <summary>
    /// Handles a dispatch tick. If a sweep is already in flight the tick is dropped so
    /// sweeps never overlap; otherwise the guard is raised and an asynchronous sweep is
    /// launched, with a <see cref="InternalMessages.DispatchComplete"/> piped back to
    /// <see cref="Self"/> to lower the guard on the actor thread.
    /// </summary>
    private void HandleDispatchTick()
    {
        if (_dispatching)
        {
            return;
        }

        _dispatching = true;
        var now = DateTimeOffset.UtcNow;
        // Hand the lifecycle token to the sweep so cancellation reaches the
        // adapter. A null token (very early start / post-stop race) is replaced with
        // None — no behaviour change vs. the previous dispatcher.
        var cancellationToken = _shutdownCts?.Token ?? CancellationToken.None;

        // RunDispatchPass swallows its own errors, but the failure projection is kept as a
        // belt-and-braces guard so even a faulted task still lowers the in-flight guard —
        // otherwise the dispatcher would wedge permanently.
        RunDispatchPass(now, cancellationToken).PipeTo(
            Self,
            success: () => InternalMessages.DispatchComplete.Instance,
            failure: ex =>
            {
                _logger.LogError(ex, "Dispatch sweep faulted unexpectedly.");
                return InternalMessages.DispatchComplete.Instance;
            });
    }

    /// <summary>
    /// Runs a single dispatch sweep: claims the due batch, resolves the retry policy, and
    /// delivers each notification sequentially. Per-notification failures are caught and
    /// logged so one bad row never aborts the rest of the batch. The whole body is wrapped
    /// in a try/catch so the returned task never faults — scope creation, service resolution,
    /// and retry-policy resolution can all throw, and a faulted task would otherwise leave
    /// the dispatcher's in-flight guard stuck and wedge the loop permanently.
    ///
    /// The per-sweep DI scope still owns the repository graph
    /// (<see cref="INotificationOutboxRepository"/> + <see cref="INotificationRepository"/>),
    /// which is correct because those services back a fresh DbContext per sweep. The
    /// channel delivery adapters, however, are cached for the actor's lifetime via
    /// <see cref="ResolveAdapters"/> — see <see cref="_adaptersCache"/> for the
    /// rationale.
    /// </summary>
    private async Task RunDispatchPass(DateTimeOffset now, CancellationToken cancellationToken)
    {
        try
        {
            using var scope = _serviceProvider.CreateScope();
            var outboxRepository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
            var notificationRepository = scope.ServiceProvider.GetRequiredService<INotificationRepository>();
            var adapters = ResolveAdapters();

            IReadOnlyList<Notification> due;
            try
            {
                due = await outboxRepository.GetDueAsync(now, _options.DispatchBatchSize, cancellationToken);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                // Shutdown cancelled the claim; row stays Pending and the next active
                // node picks it up. Not a failure.
                return;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Dispatch sweep failed to claim due notifications.");
                return;
            }

            if (due.Count == 0)
            {
                return;
            }

            var policies = await ResolveRetryPoliciesAsync(notificationRepository);

            // Deliver the claimed batch with bounded parallelism. Delivery is I/O-bound
            // (multi-second SMTP/Twilio round trips), so a strictly sequential sweep caps
            // throughput at ~1/attempt-latency — an alarm-storm backlog behind one slow
            // host would drain far too slowly (arch-review 04). A SemaphoreSlim caps the
            // number of in-flight deliveries; each delivery runs on its OWN DI scope/
            // repository (the EF DbContext is not thread-safe and rows are distinct per
            // notification, so concurrent deliveries never contend). The cached adapters
            // and the lifetime-scoped audit writer are already safe for concurrent use.
            // MaxParallelDeliveries = 1 preserves the strictly sequential behaviour.
            using var deliveryGate = new SemaphoreSlim(
                _options.ResolvedMaxParallelDeliveries, _options.ResolvedMaxParallelDeliveries);
            var deliveries = new List<Task>(due.Count);
            foreach (var notification in due)
            {
                deliveries.Add(DeliverGatedAsync(
                    notification, now, policies, adapters, deliveryGate, cancellationToken));
            }

            // Each DeliverGatedAsync isolates its own faults, so WhenAll never throws.
            await Task.WhenAll(deliveries);
        }
        catch (Exception ex)
        {
            // Scope/service resolution or retry-policy resolution faulted; swallow and log so
            // the returned task completes normally and the in-flight guard is always cleared.
            _logger.LogError(ex, "Dispatch sweep failed unexpectedly.");
        }
    }

    /// <summary>
    /// Delivers one notification under the sweep's concurrency gate. Acquires a permit,
    /// runs <see cref="DeliverOneAsync"/> on a fresh per-notification DI scope (so parallel
    /// deliveries never share an EF <c>DbContext</c>), and always releases the permit.
    /// Faults are isolated here — shutdown cancellation is swallowed (the row stays in its
    /// pre-attempt state for the next active sweep) and any other exception is logged — so
    /// one bad delivery never aborts the rest of the batch and <c>Task.WhenAll</c> never faults.
    /// </summary>
    private async Task DeliverGatedAsync(
        Notification notification,
        DateTimeOffset now,
        RetryPolicies policies,
        IReadOnlyDictionary<NotificationType, INotificationDeliveryAdapter> adapters,
        SemaphoreSlim gate,
        CancellationToken cancellationToken)
    {
        try
        {
            await gate.WaitAsync(cancellationToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            // Shutdown cancelled before this delivery acquired a permit; the row stays
            // Pending and the next active node's sweep re-claims it. No permit to release.
            return;
        }

        try
        {
            if (cancellationToken.IsCancellationRequested)
            {
                return;
            }

            // Each delivery owns its DI scope + repository so parallel UpdateAsync calls
            // never share a DbContext. The adapters map + audit writer are already safe
            // for concurrent use.
            using var scope = _serviceProvider.CreateScope();
            var outboxRepository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
            await DeliverOneAsync(
                    notification, now, policies, outboxRepository, adapters, cancellationToken)
                .ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // In-flight delivery interrupted by shutdown. Row remains in its pre-attempt
            // state; next active sweep retries.
        }
        catch (Exception ex)
        {
            // Isolate per-notification failures so the remainder of the batch still runs.
            _logger.LogError(
                ex, "Dispatch failed for notification {NotificationId}.", notification.NotificationId);
        }
        finally
        {
            gate.Release();
        }
    }

    /// <summary>
    /// A single channel's transient-failure retry policy: the max attempt count before a
    /// row parks and the fixed delay between attempts. Resolved once per sweep and selected
    /// per notification by <see cref="NotificationType"/> in <see cref="DeliverOneAsync"/>.
    /// </summary>
    private readonly record struct RetryPolicy(int MaxRetries, TimeSpan RetryDelay);

    /// <summary>
    /// The per-type retry policies resolved once per dispatch sweep: <see cref="SmtpPolicy"/>
    /// drives Email (and every non-SMS type), <see cref="SmsPolicy"/> drives SMS.
    /// </summary>
    private readonly record struct RetryPolicies(RetryPolicy SmtpPolicy, RetryPolicy SmsPolicy);

    /// <summary>
    /// Resolves the per-type retry policies for the sweep: the SMTP policy from the first
    /// SMTP configuration row (drives Email and every non-SMS type), and the SMS policy from
    /// the <see cref="SmsConfiguration"/> row — falling back to the SMTP policy when no SMS
    /// configuration exists. Resolved once and reused for the whole batch;
    /// <see cref="DeliverOneAsync"/> selects the right policy by
    /// <see cref="Notification.Type"/>.
    /// </summary>
    private async Task<RetryPolicies> ResolveRetryPoliciesAsync(
        INotificationRepository notificationRepository)
    {
        var smtpPolicy = await ResolveSmtpPolicyAsync(notificationRepository);
        var smsPolicy = await ResolveSmsPolicyAsync(notificationRepository, smtpPolicy);
        return new RetryPolicies(smtpPolicy, smsPolicy);
    }

    /// <summary>
    /// Resolves the SMTP-derived retry policy from the lowest-Id SMTP configuration row. When no
    /// SMTP configuration exists, falls back to the conservative default constants — delivery
    /// itself will permanently fail in that case, so the policy only acts as a guard.
    /// </summary>
    private async Task<RetryPolicy> ResolveSmtpPolicyAsync(
        INotificationRepository notificationRepository)
    {
        var configurations = await notificationRepository.GetAllSmtpConfigurationsAsync();
        // Deterministic pick: the lowest-Id row wins regardless of repository order, so the
        // retry policy resolves the same way the delivery adapter selects its config (S7).
        var configuration = configurations.OrderBy(c => c.Id).FirstOrDefault();
        if (configuration is null)
        {
            return new RetryPolicy(FallbackMaxRetries, FallbackRetryDelay);
        }

        if (configurations.Count > 1)
        {
            _logger.LogWarning(
                "Multiple SMTP configurations exist; using the lowest-Id row ({Id}). "
                + "Enforce a single row or delete extras.",
                configuration.Id);
        }

        return ClampRetryPolicy(
            nameof(SmtpConfiguration), configuration.MaxRetries, configuration.RetryDelay);
    }

    /// <summary>
    /// Resolves the SMS-specific retry policy from the <see cref="SmsConfiguration"/> row,
    /// honoring its own <see cref="SmsConfiguration.MaxRetries"/>/<see cref="SmsConfiguration.RetryDelay"/>.
    /// When no SMS configuration row exists the SMTP policy (<paramref name="smtpFallback"/>)
    /// is reused so SMS delivery keeps the historical shared-SMTP behaviour until an operator
    /// configures SMS.
    /// </summary>
    private async Task<RetryPolicy> ResolveSmsPolicyAsync(
        INotificationRepository notificationRepository, RetryPolicy smtpFallback)
    {
        var configuration = await notificationRepository.GetSmsConfigurationAsync();
        if (configuration is null)
        {
            return smtpFallback;
        }

        return ClampRetryPolicy(
            nameof(SmsConfiguration), configuration.MaxRetries, configuration.RetryDelay);
    }

    /// <summary>
    /// Clamps a configured retry policy's non-positive values to the fallback constants,
    /// logging a Warning (parameterized by <paramref name="configName"/> so an operator can
    /// tell which configuration is misconfigured).
    /// </summary>
    /// <remarks>
    /// A non-positive <c>MaxRetries</c> (zero or negative) would otherwise satisfy
    /// <c>RetryCount >= maxRetries</c> on the very first transient failure and park the row
    /// without a single retry — silently halving the outbox's delivery guarantees. The same
    /// applies to a non-positive <c>RetryDelay</c>, which would burn-loop the dispatcher.
    /// </remarks>
    private RetryPolicy ClampRetryPolicy(string configName, int maxRetries, TimeSpan retryDelay)
    {
        if (maxRetries <= 0)
        {
            _logger.LogWarning(
                "{ConfigName}.MaxRetries={ConfiguredMaxRetries} is non-positive; " +
                "clamping to FallbackMaxRetries={FallbackMaxRetries} so transient failures " +
                "actually retry before parking.",
                configName, maxRetries, FallbackMaxRetries);
            maxRetries = FallbackMaxRetries;
        }

        if (retryDelay <= TimeSpan.Zero)
        {
            _logger.LogWarning(
                "{ConfigName}.RetryDelay={ConfiguredRetryDelay} is non-positive; " +
                "clamping to FallbackRetryDelay={FallbackRetryDelay} so the dispatcher does " +
                "not burn-loop on transient failures.",
                configName, retryDelay, FallbackRetryDelay);
            retryDelay = FallbackRetryDelay;
        }

        return new RetryPolicy(maxRetries, retryDelay);
    }

    /// <summary>
    /// Returns the <see cref="NotificationType"/> → adapter lookup, building it lazily on
    /// the first call and caching it on <see cref="_adaptersCache"/> for the actor's
    /// lifetime. The last adapter registered for a given type wins, mirroring DI's
    /// last-wins resolution semantics.
    /// </summary>
    /// <remarks>
    /// The lookup used to be rebuilt on every dispatch sweep
    /// from the per-sweep DI scope. Adapter registration is static per process
    /// lifetime, so the dict is now built ONCE — on the first sweep that needs it —
    /// and reused. To respect each adapter's scoped lifetime
    /// (<see cref="EmailNotificationDeliveryAdapter"/> takes a scoped
    /// <see cref="INotificationRepository"/>), the cache is paired with
    /// <see cref="_adaptersScope"/>, an actor-lifetime <see cref="IServiceScope"/> that
    /// owns the cached adapter instances and is disposed in <see cref="PostStop"/>.
    /// </remarks>
    private IReadOnlyDictionary<NotificationType, INotificationDeliveryAdapter> ResolveAdapters()
    {
        if (_adaptersCache is not null)
        {
            return _adaptersCache;
        }

        _adaptersScope = _serviceProvider.CreateScope();
        var adapters = new Dictionary<NotificationType, INotificationDeliveryAdapter>();
        foreach (var adapter in _adaptersScope.ServiceProvider.GetServices<INotificationDeliveryAdapter>())
        {
            adapters[adapter.Type] = adapter;
        }

        _adaptersCache = adapters;
        return _adaptersCache;
    }

    /// <summary>
    /// Delivers a single notification through its channel adapter and applies the resulting
    /// status transition. A missing adapter parks the notification; otherwise the
    /// <see cref="DeliveryOutcome"/> drives the transition. The updated row is always persisted.
    /// </summary>
    /// <remarks>
    /// <para>
    /// A single
    /// <see cref="AuditChannel.Notification"/>/<see cref="AuditKind.NotifyDeliver"/>
    /// row is emitted with <see cref="AuditStatus.Attempted"/> per attempt
    /// (success, transient, permanent); when the post-outcome status is a
    /// terminal one (Delivered, Parked) a SECOND row is emitted carrying
    /// that terminal status. Both emissions are wrapped in a try/catch so a
    /// thrown audit writer NEVER aborts the user-facing dispatch — the
    /// <see cref="CentralAuditWriter"/> itself swallows internal failures,
    /// but the dispatcher wraps defensively per alog.md §13. The
    /// missing-adapter park path also emits both rows because it IS an
    /// attempt that resolved to a park from the dispatcher's point of view.
    /// </para>
    /// <para>
    /// Attempt duration is measured around the adapter call and recorded on
    /// the Attempted row so downstream KPIs can compute per-attempt latency
    /// without joining to the row update timestamps.
    /// </para>
    /// </remarks>
    private async Task DeliverOneAsync(
        Notification notification,
        DateTimeOffset now,
        RetryPolicies policies,
        INotificationOutboxRepository outboxRepository,
        IReadOnlyDictionary<NotificationType, INotificationDeliveryAdapter> adapters,
        CancellationToken cancellationToken)
    {
        // Select the retry policy by delivery channel: SMS honors SmsConfiguration's own
        // MaxRetries/RetryDelay; every other type (Email today) uses the SMTP policy.
        var (maxRetries, retryDelay) = notification.Type == NotificationType.Sms
            ? policies.SmsPolicy
            : policies.SmtpPolicy;

        if (!adapters.TryGetValue(notification.Type, out var adapter))
        {
            // Missing-adapter park: from the dispatcher's perspective this is an
            // attempt that resolved to a terminal park. Emit Attempted then the
            // terminal Parked row, both carrying the same explanatory error.
            var missingAdapterError = $"no delivery adapter for type {notification.Type}";
            notification.Status = NotificationStatus.Parked;
            notification.LastError = missingAdapterError;
            notification.LastAttemptAt = now;
            await WarnIfVanishedAsync(outboxRepository, notification, cancellationToken);
            await EmitAttemptAuditAsync(
                notification,
                now,
                durationMs: 0,
                errorMessage: missingAdapterError);
            await EmitTerminalAuditAsync(notification, now, errorMessage: missingAdapterError);
            return;
        }

        // Measure the attempt duration around the adapter call so the
        // Attempted row carries it for KPI use.
        // Pass the lifecycle token so a coordinated shutdown promptly cancels the
        // in-flight SMTP send instead of waiting for the SMTP connect/auth/send timeout.
        var attemptStart = DateTimeOffset.UtcNow;
        var outcome = await adapter.DeliverAsync(notification, cancellationToken);
        var durationMs = (int)Math.Min(
            int.MaxValue, Math.Max(0, (DateTimeOffset.UtcNow - attemptStart).TotalMilliseconds));

        switch (outcome.Result)
        {
            case DeliveryResult.Success:
                notification.Status = NotificationStatus.Delivered;
                notification.DeliveredAt = now;
                notification.LastAttemptAt = now;
                notification.ResolvedTargets = outcome.ResolvedTargets;
                notification.LastError = null;
                break;

            case DeliveryResult.TransientFailure:
                notification.LastAttemptAt = now;
                notification.RetryCount++;
                notification.LastError = outcome.Error;
                if (notification.RetryCount >= maxRetries)
                {
                    notification.Status = NotificationStatus.Parked;
                }
                else
                {
                    notification.Status = NotificationStatus.Retrying;
                    notification.NextAttemptAt = now + retryDelay;
                }
                break;

            case DeliveryResult.PermanentFailure:
                notification.Status = NotificationStatus.Parked;
                notification.LastAttemptAt = now;
                notification.LastError = outcome.Error;
                break;
        }

        await WarnIfVanishedAsync(outboxRepository, notification, cancellationToken);

        // Emit the per-attempt Attempted row exactly once regardless of the
        // outcome (B2). The error message comes from the outcome, not from
        // notification.LastError, so a success row is null and a transient
        // row carries the SMTP failure reason verbatim.
        await EmitAttemptAuditAsync(
            notification,
            now,
            durationMs: durationMs,
            errorMessage: outcome.Result == DeliveryResult.Success ? null : outcome.Error);

        // If the post-outcome status is terminal (Delivered or Parked — the
        // dispatcher never sets Discarded; that lives on the manual discard
        // path), emit the terminal NotifyDeliver row (B3). The error message
        // on a Delivered terminal is null; on Parked it carries the outcome's
        // reason so downstream consumers can link Attempted+Parked rows.
        if (IsTerminal(notification.Status))
        {
            await EmitTerminalAuditAsync(
                notification,
                now,
                errorMessage: outcome.Result == DeliveryResult.Success ? null : outcome.Error);
        }
    }

    /// <summary>
    /// Persists the dispatcher's delivery-state write and logs when the row has
    /// VANISHED underneath it — <see cref="INotificationOutboxRepository.UpdateAsync"/>
    /// returning false means the retention purge deleted the notification between
    /// the claim and the write.
    /// </summary>
    /// <remarks>
    /// The dispatcher deliberately does not treat this as an error: the delivery
    /// itself already happened (or failed) and there is no row left to record the
    /// outcome on, so there is nothing to retry or roll back. The audit rows are
    /// still emitted — they are the durable record. The operator one-shots
    /// (retry/discard) take the opposite stance and answer "not found", because a
    /// human is waiting on that answer.
    /// </remarks>
    private async Task WarnIfVanishedAsync(
        INotificationOutboxRepository repository,
        Notification notification,
        CancellationToken cancellationToken)
    {
        var persisted = await repository.UpdateAsync(notification, cancellationToken);
        if (!persisted)
        {
            _logger.LogWarning(
                "Notification {NotificationId} disappeared before its delivery state could be written (status {Status}); the row was most likely purged mid-flight.",
                notification.NotificationId,
                notification.Status);
        }
    }

    /// <summary>
    /// True for <see cref="NotificationStatus.Delivered"/>,
    /// <see cref="NotificationStatus.Parked"/>, or
    /// <see cref="NotificationStatus.Discarded"/> — the three terminal states
    /// on the central outbox lifecycle. Used by the dispatcher and the manual
    /// discard handler to decide when to emit the terminal NotifyDeliver row.
    /// </summary>
    private static bool IsTerminal(NotificationStatus status)
    {
        return status is NotificationStatus.Delivered
            or NotificationStatus.Parked
            or NotificationStatus.Discarded;
    }

    /// <summary>
    /// Emits a single
    /// <see cref="AuditChannel.Notification"/>/<see cref="AuditKind.NotifyDeliver"/>
    /// audit row carrying the terminal status (Delivered, Parked, or
    /// Discarded) of <paramref name="notification"/>. Wrapped in try/catch
    /// for the same defensive reason as <see cref="EmitAttemptAuditAsync"/>.
    /// </summary>
    /// <remarks>
    /// <see cref="ICentralAuditWriter.WriteAsync"/> is awaited inside the
    /// try/catch so the catch is actually reachable for writer faults and so the
    /// audit task does not outlive the per-sweep DI scope. The audit-write-never-
    /// affects-delivery invariant is preserved by the surrounding catch.
    /// </remarks>
    private async Task EmitTerminalAuditAsync(
        Notification notification,
        DateTimeOffset now,
        string? errorMessage,
        string? actorOverride = null)
    {
        try
        {
            var terminalStatus = MapNotificationStatusToAuditStatus(notification.Status);
            var evt = BuildNotifyDeliverEvent(
                notification, now, terminalStatus, errorMessage, actorOverride: actorOverride);
            await _auditWriter.WriteAsync(evt);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(
                ex,
                "Failed to emit terminal {Status} audit row for notification {NotificationId}.",
                notification.Status, notification.NotificationId);
        }
    }

    /// <summary>
    /// Maps the central-outbox <see cref="NotificationStatus"/> terminal
    /// values onto the corresponding <see cref="AuditStatus"/> values used by
    /// AuditLog. Non-terminal statuses throw — the caller must gate on
    /// <see cref="IsTerminal"/>.
    /// </summary>
    private static AuditStatus MapNotificationStatusToAuditStatus(NotificationStatus status)
    {
        return status switch
        {
            NotificationStatus.Delivered => AuditStatus.Delivered,
            NotificationStatus.Parked => AuditStatus.Parked,
            NotificationStatus.Discarded => AuditStatus.Discarded,
            _ => throw new ArgumentOutOfRangeException(
                nameof(status), status, "non-terminal status has no audit terminal mapping"),
        };
    }

    /// <summary>
    /// Emits a single
    /// <see cref="AuditChannel.Notification"/>/<see cref="AuditKind.NotifyDeliver"/>
    /// audit row with <see cref="AuditStatus.Attempted"/>. Wrapped in
    /// try/catch so an audit-write failure never propagates back into the
    /// dispatcher loop (alog.md §13).
    /// </summary>
    /// <remarks>
    /// Previously the writer task was discarded (<c>_ = WriteAsync(...)</c>),
    /// which made the surrounding catch unreachable for any fault originating in the
    /// awaited body of <c>WriteAsync</c> and let the audit task outlive the dispatcher's
    /// per-sweep DI scope. The task is now awaited inside the try/catch: the
    /// audit-write-never-affects-delivery invariant is preserved by the catch, and
    /// writer faults reach the operator log instead of being silently lost.
    /// </remarks>
    private async Task EmitAttemptAuditAsync(
        Notification notification,
        DateTimeOffset now,
        int durationMs,
        string? errorMessage)
    {
        try
        {
            var evt = BuildNotifyDeliverEvent(
                notification, now, AuditStatus.Attempted, errorMessage, durationMs);
            await _auditWriter.WriteAsync(evt);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(
                ex,
                "Failed to emit Attempted audit row for notification {NotificationId}.",
                notification.NotificationId);
        }
    }

    /// <summary>
    /// Builds a <see cref="AuditChannel.Notification"/>/<see cref="AuditKind.NotifyDeliver"/>
    /// row with the per-notification provenance fields (correlation id, list
    /// name, source site/instance/script) populated from
    /// <paramref name="notification"/>. <see cref="AuditEvent.CorrelationId"/>
    /// parses the notification's id as a Guid; sites generate the id with
    /// <c>Guid.NewGuid().ToString("N")</c> so the parse always succeeds, but
    /// a non-Guid id is recorded as null rather than crashing the dispatcher.
    /// <see cref="AuditEvent.ExecutionId"/> is copied straight from
    /// <see cref="Notification.OriginExecutionId"/> so the dispatcher's
    /// <c>NotifyDeliver</c> rows carry the same per-run id as the site's
    /// <c>NotifySend</c> row; <see cref="AuditEvent.ParentExecutionId"/>
    /// is likewise copied from <see cref="Notification.OriginParentExecutionId"/>.
    /// </summary>
    private static AuditEvent BuildNotifyDeliverEvent(
        Notification notification,
        DateTimeOffset now,
        AuditStatus status,
        string? errorMessage,
        int? durationMs = null,
        string? actorOverride = null)
    {
        Guid? correlationId = Guid.TryParse(notification.NotificationId, out var parsed)
            ? parsed
            : null;

        return ScadaBridgeAuditEventFactory.Create(
            channel: AuditChannel.Notification,
            kind: AuditKind.NotifyDeliver,
            status: status,
            occurredAtUtc: now.UtcDateTime,
            // Central dispatch is a system identity per the Actor-column spec —
            // there is no per-call authenticated user for automatic delivery. An
            // operator-initiated transition (Retry/Discard) passes actorOverride
            // so the row is attributable to the person who un-parked/cancelled it.
            // The originating script is still captured on SourceScript.
            actor: actorOverride ?? SystemActor,
            target: notification.ListName,
            correlationId: correlationId,
            // ExecutionId: the originating script execution's id,
            // carried from the site on NotificationSubmit and persisted on the
            // Notification row. Echoing it here links the central NotifyDeliver
            // rows to the site-emitted NotifySend row for the same run. Null when
            // the notification was raised outside a script execution.
            executionId: notification.OriginExecutionId,
            // ParentExecutionId: the originating routed run's
            // parent ExecutionId, carried from the site on NotificationSubmit and
            // persisted on the Notification row. Echoing it here links the central
            // NotifyDeliver rows to the routed run's parent. Null for non-routed runs.
            parentExecutionId: notification.OriginParentExecutionId,
            sourceSiteId: notification.SourceSiteId,
            sourceInstanceId: notification.SourceInstanceId,
            sourceScript: notification.SourceScript,
            durationMs: durationMs,
            errorMessage: errorMessage);
    }

    /// <summary>
    /// Handles a purge tick by launching an asynchronous sweep that bulk-deletes terminal
    /// notification rows older than <see cref="NotificationOutboxOptions.TerminalRetention"/>.
    /// Purges are daily and idempotent, so no in-flight guard is needed. <see cref="RunPurgePass"/>
    /// self-isolates its faults — it logs internally and never faults its task — so the
    /// success projection is the normal completion path that logs the deleted count. The
    /// failure projection is kept as a belt-and-braces backup, consistent with
    /// <see cref="HandleDispatchTick"/>/<see cref="RunDispatchPass"/>.
    /// </summary>
    private void HandlePurgeTick()
    {
        var cutoff = DateTimeOffset.UtcNow - _options.TerminalRetention;

        RunPurgePass(cutoff).PipeTo(
            Self,
            success: deleted =>
            {
                _logger.LogInformation(
                    "Purge removed {DeletedCount} terminal notification(s) older than {Cutoff:o}.",
                    deleted, cutoff);
                return InternalMessages.PurgeComplete.Instance;
            },
            failure: ex =>
            {
                _logger.LogError(ex, "Purge sweep faulted unexpectedly.");
                return InternalMessages.PurgeComplete.Instance;
            });
    }

    /// <summary>
    /// Runs a single purge sweep: resolves a scoped <see cref="INotificationOutboxRepository"/>
    /// and bulk-deletes terminal rows created before <paramref name="cutoff"/>, returning the
    /// deleted count. The whole body is wrapped in a try/catch so the returned task never
    /// faults — scope creation, service resolution, and the bulk delete can all throw, and
    /// self-isolating the fault here keeps the fault-handling strategy symmetric with
    /// <see cref="RunDispatchPass"/>. On failure the exception is logged and 0 is returned.
    /// </summary>
    private async Task<int> RunPurgePass(DateTimeOffset cutoff)
    {
        try
        {
            using var scope = _serviceProvider.CreateScope();
            var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
            return await repository.DeleteTerminalOlderThanAsync(cutoff);
        }
        catch (Exception ex)
        {
            // Scope/service resolution or the bulk delete faulted; swallow and log so the
            // returned task completes normally, mirroring RunDispatchPass.
            _logger.LogError(ex, "Purge sweep failed unexpectedly.");
            return 0;
        }
    }

    /// <summary>
    /// Handles a paginated, filtered query over the outbox. Builds a
    /// <see cref="NotificationOutboxFilter"/> from the request (parsing the string status/type
    /// filters to their enums and deriving the stuck cutoff when <c>StuckOnly</c> is set),
    /// runs the query on a scoped repository, and pipes the mapped response back to the
    /// captured sender. A repository fault yields a failure response with an empty list.
    /// </summary>
    private void HandleQuery(NotificationOutboxQueryRequest request)
    {
        var sender = Sender;
        var now = DateTimeOffset.UtcNow;

        QueryOutboxAsync(request, now).PipeTo(
            sender,
            success: response => response,
            failure: ex => new NotificationOutboxQueryResponse(
                request.CorrelationId,
                Success: false,
                ErrorMessage: ex.GetBaseException().Message,
                Notifications: Array.Empty<NotificationSummary>(),
                TotalCount: 0));
    }

    private async Task<NotificationOutboxQueryResponse> QueryOutboxAsync(
        NotificationOutboxQueryRequest request, DateTimeOffset now)
    {
        var filter = new NotificationOutboxFilter(
            Status: ParseEnum<NotificationStatus>(request.StatusFilter),
            Type: ParseEnum<NotificationType>(request.TypeFilter),
            SourceSiteId: request.SourceSiteFilter,
            ListName: request.ListNameFilter,
            SubjectKeyword: request.SubjectKeyword,
            StuckOnly: request.StuckOnly,
            StuckCutoff: request.StuckOnly ? StuckCutoff(now) : null,
            From: request.From,
            To: request.To,
            SourceNode: request.SourceNodeFilter);

        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
        var (rows, totalCount) = await repository.QueryAsync(filter, request.PageNumber, request.PageSize);

        var stuckCutoff = StuckCutoff(now);
        var summaries = rows
            .Select(row => new NotificationSummary(
                row.NotificationId,
                row.Type.ToString(),
                row.ListName,
                row.Subject,
                row.Status.ToString(),
                row.RetryCount,
                row.LastError,
                row.SourceSiteId,
                row.SourceInstanceId,
                row.CreatedAt,
                row.DeliveredAt,
                IsStuck: IsStuck(row, stuckCutoff),
                SourceNode: row.SourceNode))
            .ToList();

        return new NotificationOutboxQueryResponse(
            request.CorrelationId, Success: true, ErrorMessage: null, summaries, totalCount);
    }

    /// <summary>
    /// Handles a single-notification status query. Replies <c>Found: false</c> with empty
    /// detail when no row matches, otherwise the row's current status, retry count, last
    /// error, and delivery time.
    /// </summary>
    private void HandleStatusQuery(NotificationStatusQuery query)
    {
        var sender = Sender;

        StatusQueryAsync(query).PipeTo(
            sender,
            success: response => response,
            failure: ex =>
            {
                // NotificationStatusResponse has no error field, so a repository fault is
                // reported as Found: false — log the fault so a transient DB error is not
                // silently indistinguishable from a genuinely-missing notification.
                _logger.LogWarning(
                    ex, "Status query for notification {NotificationId} failed.", query.NotificationId);
                return new NotificationStatusResponse(
                    query.CorrelationId, Found: false, Status: string.Empty,
                    RetryCount: 0, LastError: null, DeliveredAt: null);
            });
    }

    private async Task<NotificationStatusResponse> StatusQueryAsync(NotificationStatusQuery query)
    {
        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
        var notification = await repository.GetByIdAsync(query.NotificationId);

        if (notification is null)
        {
            return new NotificationStatusResponse(
                query.CorrelationId, Found: false, Status: string.Empty,
                RetryCount: 0, LastError: null, DeliveredAt: null);
        }

        return new NotificationStatusResponse(
            query.CorrelationId,
            Found: true,
            Status: notification.Status.ToString(),
            RetryCount: notification.RetryCount,
            LastError: notification.LastError,
            DeliveredAt: notification.DeliveredAt);
    }

    /// <summary>
    /// Handles a full-detail query for a single notification — backs the report detail
    /// modal, which needs the Body and resolved recipients that the grid summary omits.
    /// </summary>
    private void HandleDetailRequest(NotificationDetailRequest request)
    {
        var sender = Sender;

        DetailAsync(request).PipeTo(
            sender,
            success: response => response,
            failure: ex => new NotificationDetailResponse(
                request.CorrelationId, Success: false,
                ErrorMessage: ex.GetBaseException().Message, Detail: null));
    }

    private async Task<NotificationDetailResponse> DetailAsync(NotificationDetailRequest request)
    {
        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
        var notification = await repository.GetByIdAsync(request.NotificationId);

        if (notification is null)
        {
            return new NotificationDetailResponse(
                request.CorrelationId, Success: false,
                ErrorMessage: "notification not found", Detail: null);
        }

        var detail = new NotificationDetail(
            notification.NotificationId,
            notification.Type.ToString(),
            notification.ListName,
            notification.Subject,
            notification.Body,
            notification.Status.ToString(),
            notification.RetryCount,
            notification.LastError,
            notification.ResolvedTargets,
            notification.TypeData,
            notification.SourceSiteId,
            notification.SourceInstanceId,
            notification.SourceScript,
            notification.SiteEnqueuedAt,
            notification.CreatedAt,
            notification.LastAttemptAt,
            notification.NextAttemptAt,
            notification.DeliveredAt,
            notification.SourceNode);

        return new NotificationDetailResponse(
            request.CorrelationId, Success: true, ErrorMessage: null, detail);
    }

    /// <summary>
    /// Handles a manual retry request. Only a <c>Parked</c> notification can be retried;
    /// it is reset to <c>Pending</c> with a cleared retry count, next-attempt time, and
    /// last error so the dispatch loop re-claims it on the next sweep.
    /// </summary>
    private void HandleRetry(RetryNotificationRequest request)
    {
        var sender = Sender;

        RetryAsync(request).PipeTo(
            sender,
            success: response => response,
            failure: ex => new RetryNotificationResponse(
                request.CorrelationId, Success: false, ErrorMessage: ex.GetBaseException().Message));
    }

    private async Task<RetryNotificationResponse> RetryAsync(RetryNotificationRequest request)
    {
        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
        var notification = await repository.GetByIdAsync(request.NotificationId);

        if (notification is null)
        {
            return new RetryNotificationResponse(
                request.CorrelationId, Success: false, ErrorMessage: "notification not found");
        }

        if (notification.Status != NotificationStatus.Parked)
        {
            return new RetryNotificationResponse(
                request.CorrelationId, Success: false,
                ErrorMessage: "only parked notifications can be retried");
        }

        notification.Status = NotificationStatus.Pending;
        notification.RetryCount = 0;
        notification.NextAttemptAt = null;
        notification.LastError = null;

        // Zero rows updated means the row was purged between the read above and
        // this write. Answer the operator honestly instead of reporting a
        // re-queue that never happened — and emit no un-park audit row, because
        // there is nothing to attribute it to.
        if (!await repository.UpdateAsync(notification))
        {
            return new RetryNotificationResponse(
                request.CorrelationId, Success: false, ErrorMessage: "notification not found");
        }

        // Operator re-queued a parked notification. Emit a Submitted NotifyDeliver
        // row attributing the un-park to the operator — otherwise the lifecycle
        // reads Parked → Attempted → Delivered with no record of who un-parked it.
        // Best-effort: audit failure never aborts the retry.
        await EmitRetrySubmittedAuditAsync(notification, DateTimeOffset.UtcNow, request.RequestedBy);

        return new RetryNotificationResponse(request.CorrelationId, Success: true, ErrorMessage: null);
    }

    /// <summary>
    /// Emits a single <see cref="AuditKind.NotifyDeliver"/> row with
    /// <see cref="AuditStatus.Submitted"/> recording that <paramref name="actorOverride"/>
    /// re-queued a parked notification. Best-effort with the same
    /// audit-never-affects-the-action shape as <see cref="EmitTerminalAuditAsync"/>.
    /// </summary>
    private async Task EmitRetrySubmittedAuditAsync(
        Notification notification,
        DateTimeOffset now,
        string? actorOverride)
    {
        try
        {
            var evt = BuildNotifyDeliverEvent(
                notification, now, AuditStatus.Submitted, errorMessage: null, actorOverride: actorOverride);
            await _auditWriter.WriteAsync(evt);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(
                ex,
                "Failed to emit Submitted (operator retry) audit row for notification {NotificationId}.",
                notification.NotificationId);
        }
    }

    /// <summary>
    /// Handles a manual discard request. Only a <c>Parked</c> notification can be discarded;
    /// it is moved to the terminal <c>Discarded</c> status.
    /// </summary>
    private void HandleDiscard(DiscardNotificationRequest request)
    {
        var sender = Sender;

        DiscardAsync(request).PipeTo(
            sender,
            success: response => response,
            failure: ex => new DiscardNotificationResponse(
                request.CorrelationId, Success: false, ErrorMessage: ex.GetBaseException().Message));
    }

    private async Task<DiscardNotificationResponse> DiscardAsync(DiscardNotificationRequest request)
    {
        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
        var notification = await repository.GetByIdAsync(request.NotificationId);

        if (notification is null)
        {
            return new DiscardNotificationResponse(
                request.CorrelationId, Success: false, ErrorMessage: "notification not found");
        }

        if (notification.Status != NotificationStatus.Parked)
        {
            return new DiscardNotificationResponse(
                request.CorrelationId, Success: false,
                ErrorMessage: "only parked notifications can be discarded");
        }

        notification.Status = NotificationStatus.Discarded;

        // Same vanished-row honesty as the retry path: a purge between the read
        // and the write leaves nothing to discard, so report not-found rather
        // than success (and skip the terminal audit row below).
        if (!await repository.UpdateAsync(notification))
        {
            return new DiscardNotificationResponse(
                request.CorrelationId, Success: false, ErrorMessage: "notification not found");
        }

        // A manual discard is the OTHER code path that produces
        // a terminal NotificationStatus transition (alongside the dispatcher).
        // Emit a Discarded NotifyDeliver row to match the dispatcher's
        // Delivered/Parked emissions; the row carries no error message because
        // the discard is an operator-driven cancellation, not a delivery error.
        // The operator identity is stamped as Actor so the cancellation is
        // attributable.
        await EmitTerminalAuditAsync(
            notification, DateTimeOffset.UtcNow, errorMessage: null, actorOverride: request.RequestedBy);

        return new DiscardNotificationResponse(request.CorrelationId, Success: true, ErrorMessage: null);
    }

    /// <summary>
    /// Handles a KPI snapshot request, computing the outbox metrics with the stuck cutoff
    /// derived from <see cref="NotificationOutboxOptions.StuckAgeThreshold"/> and the
    /// delivered window from <see cref="NotificationOutboxOptions.DeliveredKpiWindow"/>.
    /// </summary>
    private void HandleKpiRequest(NotificationKpiRequest request)
    {
        var sender = Sender;
        var now = DateTimeOffset.UtcNow;
        var stuckCutoff = StuckCutoff(now);
        var deliveredSince = now - _options.DeliveredKpiWindow;

        ComputeKpisAsync(request.CorrelationId, stuckCutoff, deliveredSince).PipeTo(
            sender,
            success: response => response,
            failure: ex => new NotificationKpiResponse(
                request.CorrelationId,
                Success: false,
                ErrorMessage: ex.GetBaseException().Message,
                QueueDepth: 0,
                StuckCount: 0,
                ParkedCount: 0,
                DeliveredLastInterval: 0,
                OldestPendingAge: null));
    }

    private async Task<NotificationKpiResponse> ComputeKpisAsync(
        string correlationId, DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince)
    {
        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
        var snapshot = await repository.ComputeKpisAsync(stuckCutoff, deliveredSince);

        return new NotificationKpiResponse(
            correlationId,
            Success: true,
            ErrorMessage: null,
            snapshot.QueueDepth,
            snapshot.StuckCount,
            snapshot.ParkedCount,
            snapshot.DeliveredLastInterval,
            snapshot.OldestPendingAge);
    }

    /// <summary>
    /// Handles a per-site KPI request, computing the per-source-site outbox metrics with the
    /// same stuck cutoff and delivered window as <see cref="HandleKpiRequest"/>.
    /// </summary>
    private void HandlePerSiteKpiRequest(PerSiteNotificationKpiRequest request)
    {
        var sender = Sender;
        var now = DateTimeOffset.UtcNow;
        var stuckCutoff = StuckCutoff(now);
        var deliveredSince = now - _options.DeliveredKpiWindow;

        ComputePerSiteKpisAsync(request.CorrelationId, stuckCutoff, deliveredSince).PipeTo(
            sender,
            success: response => response,
            failure: ex => new PerSiteNotificationKpiResponse(
                request.CorrelationId,
                Success: false,
                ErrorMessage: ex.GetBaseException().Message,
                Sites: Array.Empty<SiteNotificationKpiSnapshot>()));
    }

    private async Task<PerSiteNotificationKpiResponse> ComputePerSiteKpisAsync(
        string correlationId, DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince)
    {
        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
        var sites = await repository.ComputePerSiteKpisAsync(stuckCutoff, deliveredSince);

        return new PerSiteNotificationKpiResponse(correlationId, Success: true, ErrorMessage: null, sites);
    }

    /// <summary>
    /// Handles a per-node KPI request, computing the per-source-node outbox metrics with the
    /// same stuck cutoff and delivered window as <see cref="HandleKpiRequest"/>. Additive
    /// alongside <see cref="HandlePerSiteKpiRequest"/> — does not change per-site behaviour.
    /// </summary>
    private void HandlePerNodeKpiRequest(PerNodeNotificationKpiRequest request)
    {
        var sender = Sender;
        var now = DateTimeOffset.UtcNow;
        var stuckCutoff = StuckCutoff(now);
        var deliveredSince = now - _options.DeliveredKpiWindow;

        ComputePerNodeKpisAsync(request.CorrelationId, stuckCutoff, deliveredSince).PipeTo(
            sender,
            success: response => response,
            failure: ex => new PerNodeNotificationKpiResponse(
                request.CorrelationId,
                Success: false,
                ErrorMessage: ex.GetBaseException().Message,
                Nodes: Array.Empty<NodeNotificationKpiSnapshot>()));
    }

    private async Task<PerNodeNotificationKpiResponse> ComputePerNodeKpisAsync(
        string correlationId, DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince)
    {
        using var scope = _serviceProvider.CreateScope();
        var repository = scope.ServiceProvider.GetRequiredService<INotificationOutboxRepository>();
        var nodes = await repository.ComputePerNodeKpisAsync(stuckCutoff, deliveredSince);

        return new PerNodeNotificationKpiResponse(correlationId, Success: true, ErrorMessage: null, nodes);
    }

    /// <summary>
    /// The instant before which a still-pending notification counts as stuck — <paramref name="now"/>
    /// offset back by <see cref="NotificationOutboxOptions.StuckAgeThreshold"/>.
    /// </summary>
    private DateTimeOffset StuckCutoff(DateTimeOffset now) => now - _options.StuckAgeThreshold;

    /// <summary>
    /// A notification counts as stuck when it is still in a non-terminal status
    /// (<c>Pending</c> or <c>Retrying</c>) and was created before the supplied cutoff.
    /// </summary>
    private static bool IsStuck(Notification notification, DateTimeOffset stuckCutoff)
    {
        return notification.Status is NotificationStatus.Pending or NotificationStatus.Retrying
            && notification.CreatedAt < stuckCutoff;
    }

    /// <summary>
    /// Parses a string filter value to a nullable enum, ignoring case. An empty, whitespace,
    /// or unrecognised value yields <c>null</c> — meaning "no constraint on that dimension".
    /// </summary>
    private static TEnum? ParseEnum<TEnum>(string? value) where TEnum : struct, Enum
    {
        return Enum.TryParse<TEnum>(value, ignoreCase: true, out var parsed) ? parsed : null;
    }

    private static Notification BuildNotification(NotificationSubmit msg, NotificationType type)
    {
        // The delivery channel is taken from the target list's Type (resolved by the
        // caller); it defaults to Email when the list cannot be resolved.
        return new Notification(
            msg.NotificationId,
            type,
            msg.ListName,
            msg.Subject,
            msg.Body,
            msg.SourceSiteId)
        {
            SourceInstanceId = msg.SourceInstanceId,
            SourceScript = msg.SourceScript,
            // SourceNode: the cluster node on which the
            // notification was emitted (node-a/node-b for site rows). Stamped by the
            // emitting site from INodeIdentityProvider and carried, inside the
            // serialized payload, through the S&F buffer to central. EF tracked-entity
            // insert flows it through to the Notifications.SourceNode column. Null on
            // submissions buffered before the field existed.
            SourceNode = msg.SourceNode,
            // OriginExecutionId: the originating script execution's id,
            // carried from the site so the dispatcher can echo it onto NotifyDeliver rows.
            OriginExecutionId = msg.OriginExecutionId,
            // OriginParentExecutionId: the originating routed run's parent
            // ExecutionId, carried from the site so the dispatcher can echo it onto
            // NotifyDeliver rows.
            OriginParentExecutionId = msg.OriginParentExecutionId,
            SiteEnqueuedAt = msg.SiteEnqueuedAt,
            CreatedAt = DateTimeOffset.UtcNow,
            // Status stays at its Pending default for the dispatch sweep to claim.
        };
    }
}