ScadaBridge/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/HeadExport/AuditChainHeadExportService.cs

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;

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

/// <summary>
/// Central hosted service that exports the AuditLog hash-chain heads off-box
/// (<c>AuditLog:HashChain:HeadExport</c>, ScadaBridge#107 item 1). Default off.
/// </summary>
/// <remarks>
/// <para>
/// Runs on both central nodes (each record names its node). A tick reads every month's head in
/// one consistent snapshot (<see cref="IAuditChainHeadStore.ReadHeadsAsync"/>, chain lock
/// SHARED) and writes the heads that changed since this node's last successful export — all of
/// them on the first tick after start — to the configured <see cref="IAuditChainHeadSink"/>.
/// A failed write is logged and retried on the next tick with the same heads (the in-memory
/// "last exported" state only advances on success). No signing: the protection is that the
/// sink sits outside the database login's reach.
/// </para>
/// </remarks>
public sealed class AuditChainHeadExportService : IHostedService, IDisposable
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly AuditChainHeadExportOptions _options;
    private readonly IAuditChainHeadSink? _sink;
    private readonly string? _node;
    private readonly TimeProvider _time;
    private readonly ILogger<AuditChainHeadExportService> _logger;
    private readonly AuditChainHeadExportStatus _status;
    private readonly Dictionary<DateTime, (long Seq, long? Purged, long Links)> _lastExported = new();
    private CancellationTokenSource? _cts;
    private Task? _loop;

    /// <summary>Initializes the service.</summary>
    /// <param name="scopeFactory">Scope factory for the scoped head store.</param>
    /// <param name="options">Hash-chain options (the <c>HeadExport</c> sub-section is used).</param>
    /// <param name="sink">The sink (null when the export is disabled).</param>
    /// <param name="nodeName">This node's name for the records.</param>
    /// <param name="logger">Logger.</param>
    /// <param name="time">Clock.</param>
    /// <param name="status">Health status the tick outcome is recorded on (ScadaBridge#115).</param>
    public AuditChainHeadExportService(
        IServiceScopeFactory scopeFactory,
        IOptions<AuditLogHashChainOptions> options,
        IAuditChainHeadSink? sink,
        string? nodeName,
        ILogger<AuditChainHeadExportService> logger,
        TimeProvider? time = null,
        AuditChainHeadExportStatus? status = null)
    {
        _status = status ?? new AuditChainHeadExportStatus();
        _scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
        _options = (options ?? throw new ArgumentNullException(nameof(options))).Value.HeadExport;
        _sink = sink;
        _node = nodeName;
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _time = time ?? TimeProvider.System;
    }

    /// <inheritdoc />
    public Task StartAsync(CancellationToken cancellationToken)
    {
        if (!_options.Enabled || _sink is null)
        {
            _logger.LogInformation("AuditLog chain-head export disabled (AuditLog:HashChain:HeadExport:Enabled=false).");
            return Task.CompletedTask;
        }

        _status.MarkStarted(_options.IntervalMinutes, _time.GetUtcNow().UtcDateTime);
        _logger.LogInformation("AuditLog chain-head export every {Minutes} min to {Sink}.", _options.IntervalMinutes, _sink.Describe);
        var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        _cts = cts;
        var token = cts.Token;
        _loop = Task.Run(() => RunLoopAsync(token), CancellationToken.None);
        return Task.CompletedTask;
    }

    private async Task RunLoopAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            await TickAsync(ct).ConfigureAwait(false);
            try
            {
                await Task.Delay(TimeSpan.FromMinutes(_options.IntervalMinutes), ct).ConfigureAwait(false);
            }
            catch (OperationCanceledException)
            {
                break;
            }
        }
    }

    /// <summary>Runs one export tick.</summary>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>The number of records written (0 on failure or nothing changed).</returns>
    internal async Task<int> TickAsync(CancellationToken ct)
    {
        if (_sink is null)
        {
            return 0;
        }

        try
        {
            IReadOnlyList<AuditChainHeadSnapshot> heads;
            await using (var scope = _scopeFactory.CreateAsyncScope())
            {
                heads = await scope.ServiceProvider.GetRequiredService<IAuditChainHeadStore>()
                    .ReadHeadsAsync(ct).ConfigureAwait(false);
            }

            var now = _time.GetUtcNow().UtcDateTime;
            var changed = heads
                .Where(h => _options.ExportUnchanged
                    || !_lastExported.TryGetValue(h.Month, out var last)
                    || last != (h.LastSeq, h.PurgedThroughSeq, h.ChainedRows + h.Tombstones))
                .ToList();
            if (changed.Count == 0)
            {
                _status.RecordSuccess(now);
                return 0;
            }

            var records = changed.Select(h => ToRecord(h, now, _node)).ToList();
            await _sink.WriteAsync(records, ct).ConfigureAwait(false);
            foreach (var h in changed)
            {
                _lastExported[h.Month] = (h.LastSeq, h.PurgedThroughSeq, h.ChainedRows + h.Tombstones);
            }

            _status.RecordSuccess(now);
            _logger.LogDebug("Exported {Count} AuditLog chain head(s) to {Sink}.", records.Count, _sink.Describe);
            return records.Count;
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            return 0;
        }
        catch (Exception ex)
        {
            // Contained: the next tick retries the same heads (nothing was marked exported).
            _logger.LogError(ex, "AuditLog chain-head export to {Sink} failed; retrying next tick.", _sink.Describe);
            _status.RecordFailure(_time.GetUtcNow().UtcDateTime, ex.Message);
            return 0;
        }
    }

    /// <summary>Maps a database head to its exported record.</summary>
    /// <param name="head">The head.</param>
    /// <param name="exportedAtUtc">Export time.</param>
    /// <param name="node">Exporting node.</param>
    /// <returns>The record.</returns>
    public static AuditChainHeadRecord ToRecord(AuditChainHeadSnapshot head, DateTime exportedAtUtc, string? node)
    {
        ArgumentNullException.ThrowIfNull(head);
        return new AuditChainHeadRecord(
            AuditChainHeadRecord.CurrentVersion,
            AuditChainHasher.PartitionLabel(head.Month),
            head.LastSeq,
            AuditChainHeadComparer.ToHex(head.LastHash),
            head.PurgedThroughSeq,
            head.PurgedThroughHash is null ? null : AuditChainHeadComparer.ToHex(head.PurgedThroughHash),
            head.ChainedRows,
            head.Tombstones,
            head.UpdatedAtUtc,
            exportedAtUtc,
            node);
    }

    /// <inheritdoc />
    public Task StopAsync(CancellationToken cancellationToken)
    {
        try
        {
            _cts?.Cancel();
        }
        catch (ObjectDisposedException)
        {
            // Dispose already cancelled the loop.
        }

        return _loop ?? Task.CompletedTask;
    }

    /// <inheritdoc />
    public void Dispose()
    {
        try
        {
            _cts?.Cancel();
        }
        catch (ObjectDisposedException)
        {
            // Idempotent.
        }

        _cts?.Dispose();
    }
}