ScadaBridge/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs

using System.Diagnostics;
using Akka.Actor;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.AuditLog.Configuration;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Types;

namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;

/// <summary>
/// Central singleton that drives the daily AuditLog partition
/// purge. On a configurable timer (default 24 hours) the actor:
/// <list type="number">
/// <item>Queries <see cref="IAuditLogRepository.GetPartitionBoundariesOlderThanAsync"/>
///       for monthly boundaries whose latest <c>OccurredAtUtc</c> is older
///       than <c>DateTime.UtcNow - RetentionDays</c>.</item>
/// <item>For each eligible boundary, calls
///       <see cref="IAuditLogRepository.SwitchOutPartitionAsync"/>, a
///       metadata-only staging-table switch (no index
///       drop/rebuild brackets it).</item>
/// <item>Publishes <see cref="AuditLogPurgedEvent"/> on the actor-system
///       EventStream so the central health collector + ops surfaces
///       can subscribe without coupling to this actor.</item>
/// </list>
/// </summary>
/// <remarks>
/// <para>
/// <b>Daily cadence.</b> The partition switch is metadata-only, but it still
/// takes schema-modification locks on a table the ingest path is writing to;
/// running more often than necessary trades contention for negligible freshness
/// wins. The default 24-hour interval matches alog.md §10's retention policy.
/// </para>
/// <para>
/// <b>Continue-on-error.</b> A single boundary that throws (transient SQL
/// failure, contention with backup, missing object) must NOT prevent the
/// other eligible boundaries from being purged on the same tick. Per-boundary
/// work runs inside its own try/catch — that per-boundary catch is what
/// keeps the singleton alive across handler throws. The
/// <see cref="SupervisorStrategy"/> override returns the Akka default
/// decider (Restart) and governs children only; this actor has no children
/// today, so the override is a forward-compat placeholder.
/// </para>
/// <para>
/// <b>DI scopes.</b> <see cref="IAuditLogRepository"/> is a scoped EF Core
/// service registered by <c>AddConfigurationDatabase</c>. The singleton
/// opens one DI scope per tick and reuses the same repository across every
/// boundary in that tick — mirrors the
/// <see cref="SiteAuditReconciliationActor"/> pattern.
/// </para>
/// <para>
/// <b>EventStream.</b> Publishing <see cref="AuditLogPurgedEvent"/> through
/// the EventStream rather than direct messaging avoids coupling this actor
/// to its consumers; a central health-counter bridge will subscribe to
/// surface purge progress on the central health report.
/// </para>
/// </remarks>
public class AuditLogPurgeActor : ReceiveActor
{
    private readonly IServiceProvider _services;
    private readonly AuditLogPurgeOptions _purgeOptions;
    private readonly AuditLogOptions _auditOptions;
    private readonly ILogger<AuditLogPurgeActor> _logger;
    private ICancelable? _timer;

    /// <summary>Initializes a new instance of <see cref="AuditLogPurgeActor"/> and registers the tick handler.</summary>
    /// <param name="services">DI service provider used to create scoped repository instances per tick.</param>
    /// <param name="purgeOptions">Options controlling the purge interval.</param>
    /// <param name="auditOptions">Options controlling retention policy (RetentionDays).</param>
    /// <param name="logger">Logger instance.</param>
    public AuditLogPurgeActor(
        IServiceProvider services,
        IOptions<AuditLogPurgeOptions> purgeOptions,
        IOptions<AuditLogOptions> auditOptions,
        ILogger<AuditLogPurgeActor> logger)
    {
        ArgumentNullException.ThrowIfNull(services);
        ArgumentNullException.ThrowIfNull(purgeOptions);
        ArgumentNullException.ThrowIfNull(auditOptions);
        ArgumentNullException.ThrowIfNull(logger);

        _services = services;
        _purgeOptions = purgeOptions.Value;
        _auditOptions = auditOptions.Value;
        _logger = logger;

        ReceiveAsync<PurgeTick>(_ => OnTickAsync());
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        base.PreStart();
        var interval = _purgeOptions.Interval;
        _timer = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
            // Short first tick (min(interval, 5min)): purges are idempotent, and using the
            // full interval as the initial delay meant a daily-recycled node never purged
            // (arch-review 04, S9). Fast test cadences (interval < 5min) are unchanged.
            initialDelay: PurgeTimerSchedule.InitialDelay(interval),
            interval: interval,
            receiver: Self,
            message: PurgeTick.Instance,
            sender: Self);
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        _timer?.Cancel();
        base.PostStop();
    }

    /// <inheritdoc />
    protected override SupervisorStrategy SupervisorStrategy()
    {
        return new OneForOneStrategy(
            maxNrOfRetries: 0,
            withinTimeRange: TimeSpan.Zero,
            decider: Akka.Actor.SupervisorStrategy.DefaultDecider);
    }

    private async Task OnTickAsync()
    {
        // Capture EventStream BEFORE the first await. Accessing Context (and
        // therefore Context.System) after an await is unsafe because Akka's
        // ActorBase.Context throws "no active ActorContext" once the
        // continuation runs on a thread that isn't currently dispatching this
        // actor — mirrors the same Sender-capture pattern in
        // AuditLogIngestActor.OnIngestAsync.
        var eventStream = Context.System.EventStream;

        // Compute the retention threshold from AuditLogOptions.RetentionDays
        // each tick — the options class supports hot reload via
        // IOptionsMonitor for the redaction policy and similar settings; we
        // read the snapshot per-tick so an operator who lowers RetentionDays
        // sees the change applied on the next purge without an actor
        // restart.
        var threshold = DateTime.UtcNow - TimeSpan.FromDays(_auditOptions.RetentionDays);

        // Use CreateAsyncScope + await using so scoped EF Core
        // services (IAsyncDisposable DbContexts) dispose asynchronously
        // without blocking on sync Dispose() of pending connection cleanup.
        await using var scope = _services.CreateAsyncScope();
        IAuditLogRepository repository;
        try
        {
            repository = scope.ServiceProvider.GetRequiredService<IAuditLogRepository>();
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to resolve IAuditLogRepository for AuditLog purge tick.");
            return;
        }

        // Optional purge-failure counter — resolved from the same per-tick scope the repository
        // comes from. NoOp default on site/test roots; the central composition root binds the
        // AuditCentralHealthSnapshot so failures surface on the central health dashboard. Mirrors
        // AuditLogIngestActor's ICentralAuditWriteFailureCounter resolution shape.
        var purgeFailureCounter = scope.ServiceProvider.GetService<IAuditPurgeFailureCounter>();

        IReadOnlyList<DateTime> boundaries;
        var enumSw = Stopwatch.StartNew();
        try
        {
            boundaries = await repository
                .GetPartitionBoundariesOlderThanAsync(threshold)
                .ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            enumSw.Stop();
            _logger.LogError(
                ex,
                "Failed to enumerate eligible AuditLog partition boundaries (threshold {ThresholdUtc:o}); skipping purge tick.",
                threshold);
            // Enumeration threw before any boundary was selected — publish with the
            // DateTime.MinValue sentinel so the failure is still observable on the health surface.
            PublishPurgeFailure(eventStream, purgeFailureCounter, DateTime.MinValue, ex, enumSw.ElapsedMilliseconds);
            return;
        }

        if (boundaries.Count == 0)
        {
            // No whole-month partitions are eligible, but per-channel overrides may
            // still expire rows earlier than the global window — run them below.
            await RunPerChannelOverridesAsync(repository, eventStream, purgeFailureCounter).ConfigureAwait(false);
            return;
        }

        foreach (var boundary in boundaries)
        {
            // Per-boundary try/catch: one bad partition (transient SQL
            // failure, missing object, contention with backup) does NOT
            // abandon the rest of the tick.
            var sw = Stopwatch.StartNew();
            try
            {
                var rowsDeleted = await repository
                    .SwitchOutPartitionAsync(boundary, _purgeOptions.ResolvedMaintenanceCommandTimeout)
                    .ConfigureAwait(false);
                sw.Stop();

                eventStream.Publish(
                    new AuditLogPurgedEvent(boundary, rowsDeleted, sw.ElapsedMilliseconds));

                _logger.LogInformation(
                    "Purged AuditLog partition {MonthBoundary:yyyy-MM-dd}; {RowsDeleted} rows in {DurationMs} ms.",
                    boundary,
                    rowsDeleted,
                    sw.ElapsedMilliseconds);
            }
            catch (Exception ex)
            {
                sw.Stop();
                _logger.LogError(
                    ex,
                    "Failed to purge AuditLog partition {MonthBoundary:yyyy-MM-dd}; other partitions continue. Elapsed {DurationMs} ms.",
                    boundary,
                    sw.ElapsedMilliseconds);
                PublishPurgeFailure(eventStream, purgeFailureCounter, boundary, ex, sw.ElapsedMilliseconds);
            }
        }

        // After the channel-blind global partition switch-out, apply any
        // per-channel retention overrides that are SHORTER than the global window via
        // a bounded, batched row DELETE on the same maintenance path. The global
        // switch-out has already dropped whole months older than RetentionDays; these
        // deletes only ever expire rows EARLIER than that, so they run last and are a
        // strict tightening.
        await RunPerChannelOverridesAsync(repository, eventStream, purgeFailureCounter).ConfigureAwait(false);
    }

    /// <summary>
    /// Publishes an <see cref="AuditLogPurgeFailedEvent"/> on the EventStream and increments the
    /// central purge-failure health counter (when wired). Best-effort — the symmetric failure twin
    /// of the success-path <see cref="AuditLogPurgedEvent"/> publish, so a silently failing
    /// retention job is a health-surface condition, not just a log line (arch-review 04, S2).
    /// </summary>
    private static void PublishPurgeFailure(
        Akka.Event.EventStream eventStream,
        IAuditPurgeFailureCounter? counter,
        DateTime monthBoundary,
        Exception ex,
        long elapsedMilliseconds)
    {
        eventStream.Publish(new AuditLogPurgeFailedEvent(monthBoundary, ex.Message, elapsedMilliseconds));
        counter?.Increment();
    }

    /// <summary>
    /// Runs each per-channel retention override whose window is strictly
    /// shorter than the global <see cref="AuditLogOptions.RetentionDays"/>, deleting
    /// rows of that channel older than the channel-specific threshold via a bounded,
    /// batched maintenance-path DELETE. Each channel runs inside its own try/catch so
    /// one bad channel does not abandon the others on the same tick, mirroring the
    /// per-boundary error isolation of the partition switch-out loop.
    /// </summary>
    /// <param name="repository">The repository resolved for this tick's DI scope.</param>
    /// <param name="eventStream">The actor-system EventStream captured before the first await.</param>
    /// <param name="purgeFailureCounter">Optional central purge-failure health counter (NoOp on site/test roots).</param>
    private async Task RunPerChannelOverridesAsync(
        IAuditLogRepository repository,
        Akka.Event.EventStream eventStream,
        IAuditPurgeFailureCounter? purgeFailureCounter)
    {
        var overrides = _auditOptions.PerChannelRetentionDays;
        if (overrides is null || overrides.Count == 0)
        {
            return;
        }

        var globalDays = _auditOptions.RetentionDays;

        foreach (var (channel, days) in overrides)
        {
            // Only act when the per-channel window is strictly shorter than the global
            // one. Equal/longer windows are already covered by the global partition
            // switch-out, so a row DELETE would be redundant work (and a longer window
            // is meaningless — the partition is dropped on the global schedule).
            if (days >= globalDays)
            {
                continue;
            }

            var channelThreshold = DateTime.UtcNow - TimeSpan.FromDays(days);
            var sw = Stopwatch.StartNew();
            try
            {
                var rowsDeleted = await repository
                    .PurgeChannelOlderThanAsync(
                        channel,
                        channelThreshold,
                        _purgeOptions.ChannelPurgeBatchSize,
                        _purgeOptions.ResolvedMaintenanceCommandTimeout,
                        maxRows: _purgeOptions.MaxPurgedRowsPerChannelPerTick)
                    .ConfigureAwait(false);
                sw.Stop();

                if (rowsDeleted > 0)
                {
                    _logger.LogInformation(
                        "Purged {RowsDeleted} AuditLog rows for channel {Channel} older than {Threshold:o} " +
                        "(per-channel override {Days}d < global {GlobalDays}d) in {DurationMs} ms.",
                        rowsDeleted,
                        channel,
                        channelThreshold,
                        days,
                        globalDays,
                        sw.ElapsedMilliseconds);
                }
            }
            catch (Exception ex)
            {
                sw.Stop();
                _logger.LogError(
                    ex,
                    "Failed to apply per-channel retention override for channel {Channel} " +
                    "({Days}d); other channels continue. Elapsed {DurationMs} ms.",
                    channel,
                    days,
                    sw.ElapsedMilliseconds);
                // A per-channel DELETE failure is a purge failure too — surface it alongside the
                // switch-out failures. MonthBoundary is DateTime.MinValue (this is not a
                // whole-month switch), matching the enumeration-phase sentinel.
                PublishPurgeFailure(eventStream, purgeFailureCounter, DateTime.MinValue, ex, sw.ElapsedMilliseconds);
            }
        }
    }

    /// <summary>Self-tick triggering a purge pass across all eligible partitions.</summary>
    internal sealed class PurgeTick
    {
        public static readonly PurgeTick Instance = new();
        private PurgeTick() { }
    }
}