ScadaBridge/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/CachedCallTelemetryForwarder.cs

using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry;

/// <summary>
/// Site-side dual emitter for cached-call lifecycle telemetry. Sister to
/// <see cref="SiteAuditTelemetryActor"/>: where that actor
/// drains audit-only events, this forwarder takes a combined
/// <see cref="CachedCallTelemetry"/> packet and fans it out to the two
/// site-local stores in a single call:
/// <list type="bullet">
///   <item><description>The <see cref="AuditEvent"/> row is written via
///   <see cref="IAuditWriter"/> (the site <c>FallbackAuditWriter</c> +
///   <c>SqliteAuditWriter</c> chain).</description></item>
///   <item><description>The operational <see cref="SiteCallOperational"/> half
///   updates the site-local <c>OperationTracking</c> SQLite store via
///   <see cref="IOperationTrackingStore"/>, with the per-lifecycle method
///   (<c>Enqueue</c> / <c>Attempt</c> / <c>Terminal</c>) selected from the
///   audit row's <see cref="AuditKind"/>.</description></item>
/// </list>
/// </summary>
/// <remarks>
/// <para>
/// <b>Best-effort contract (alog.md §7):</b> a thrown writer OR a thrown
/// tracking store must never propagate to the calling script. Both emission
/// halves are wrapped in independent try/catch blocks so a SQLite outage on
/// one side cannot starve the other — the failure is logged and the call
/// returns normally.
/// </para>
/// <para>
/// <b>Local-write only — the wire push is the drain actor's job.</b> This
/// forwarder is deliberately synchronous against the two site-local SQLite
/// stores and never pushes to central itself. The site→central transport is
/// now live: <c>ClusterClientSiteAuditClient</c> is the production binding of
/// <see cref="ISiteStreamAuditClient"/> on site roles (with
/// <c>NoOpSiteStreamAuditClient</c> retained only for central/test composition
/// roots). The push happens out-of-band: <see cref="SiteAuditTelemetryActor"/>
/// sweeps the <c>AuditEvent</c> rows this forwarder wrote — they live in SQLite
/// tagged <see cref="AuditForwardState.Pending"/> — and drains them to central
/// via that client. A single drain loop therefore covers both the audit-only
/// emissions and the cached-call emissions this forwarder produces.
/// </para>
/// </remarks>
public sealed class CachedCallTelemetryForwarder : ICachedCallTelemetryForwarder
{
    private readonly IAuditWriter _auditWriter;
    private readonly IOperationTrackingStore? _trackingStore;
    private readonly ILogger<CachedCallTelemetryForwarder> _logger;

    /// <summary>
    /// SourceNode-stamping: local node identity provider used to
    /// stamp the tracking-store row's <c>SourceNode</c> column on
    /// <c>RecordEnqueueAsync</c>. Optional — when null (legacy / test hosts)
    /// the column stays NULL on the tracking row.
    /// </summary>
    private readonly INodeIdentityProvider? _nodeIdentity;

    /// <summary>
    /// Construct the forwarder. <paramref name="trackingStore"/> is optional —
    /// when null only the audit half of the packet is emitted, which matches
    /// the composition-root contract on Central nodes: the
    /// AuditLog DI surface registers the forwarder unconditionally (mirroring
    /// the IAuditWriter chain) but the site-only tracking store has no central
    /// registration. Production site nodes wire both — the central lazy
    /// resolution is a no-op path kept symmetric with the writer chain.
    /// </summary>
    /// <param name="auditWriter">Writer used to persist audit events from the telemetry packet.</param>
    /// <param name="trackingStore">Optional store for updating operation tracking state; null on central nodes.</param>
    /// <param name="logger">Logger for this forwarder.</param>
    /// <param name="nodeIdentity">Optional provider of the current node name stamped on emitted rows.</param>
    public CachedCallTelemetryForwarder(
        IAuditWriter auditWriter,
        IOperationTrackingStore? trackingStore,
        ILogger<CachedCallTelemetryForwarder> logger,
        INodeIdentityProvider? nodeIdentity = null)
    {
        _auditWriter = auditWriter ?? throw new ArgumentNullException(nameof(auditWriter));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _trackingStore = trackingStore;
        _nodeIdentity = nodeIdentity;
    }

    /// <inheritdoc />
    public async Task ForwardAsync(CachedCallTelemetry telemetry, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(telemetry);

        // Independent try/catch — a thrown audit writer must not prevent the
        // tracking-store update from running (and vice-versa). Both halves
        // are best-effort.
        await TryEmitAuditAsync(telemetry, ct).ConfigureAwait(false);
        await TryEmitTrackingAsync(telemetry, ct).ConfigureAwait(false);
    }

    private async Task TryEmitAuditAsync(CachedCallTelemetry telemetry, CancellationToken ct)
    {
        try
        {
            await _auditWriter.WriteAsync(telemetry.Audit, ct).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            // alog.md §7 best-effort contract — log and swallow. The audit
            // pipeline's own retry/recovery (RingBufferFallback in the
            // FallbackAuditWriter) handles transient writer failures upstream;
            // a throw bubbling up here means the writer's own swallow contract
            // failed, which is itself best-effort-handled.
            // Kind/Status are domain fields carried in DetailsJson — decompose to log them.
            var d = AuditRowProjection.Decompose(telemetry.Audit);
            _logger.LogWarning(ex,
                "CachedCallTelemetryForwarder: audit emission threw for EventId {EventId} (Kind {Kind}, Status {Status}, sqlite {SqliteError})",
                d.EventId, d.Kind, d.Status, SqliteErrorCodes.Describe(ex));
        }
    }

    private async Task TryEmitTrackingAsync(CachedCallTelemetry telemetry, CancellationToken ct)
    {
        if (_trackingStore is null)
        {
            // No site-local tracking store wired — Central composition root or
            // an integration-test host that skipped AddSiteRuntime. Emitting
            // through the audit half is still meaningful; the tracking half
            // is a no-op rather than an error.
            return;
        }

        // The audit half's domain fields (Kind/SourceInstanceId/SourceScript)
        // ride inside DetailsJson — decompose once for this packet.
        var audit = AuditRowProjection.Decompose(telemetry.Audit);
        try
        {
            switch (audit.Kind)
            {
                case AuditKind.CachedSubmit:
                    // Enqueue — insert-if-not-exists with the operational
                    // channel as the kind discriminator. RetryCount is fixed
                    // at 0 by the tracking store's INSERT contract.
                    // SourceNode-stamping: stamp the local node
                    // name (node-a/node-b) from the injected
                    // INodeIdentityProvider; null when no provider was wired
                    // so the tracking row's SourceNode column stays NULL.
                    await _trackingStore.RecordEnqueueAsync(
                        telemetry.Operational.TrackedOperationId,
                        telemetry.Operational.Channel,
                        telemetry.Operational.Target,
                        audit.SourceInstanceId,
                        audit.SourceScript,
                        sourceNode: _nodeIdentity?.NodeName,
                        ct).ConfigureAwait(false);
                    break;

                case AuditKind.ApiCallCached:
                case AuditKind.DbWriteCached:
                    // Attempt — advance retry counter + last-error/HTTP-status.
                    // Terminal rows are guarded by the store's WHERE clause.
                    await _trackingStore.RecordAttemptAsync(
                        telemetry.Operational.TrackedOperationId,
                        telemetry.Operational.Status,
                        telemetry.Operational.RetryCount,
                        telemetry.Operational.LastError,
                        telemetry.Operational.HttpStatus,
                        ct).ConfigureAwait(false);
                    break;

                case AuditKind.CachedResolve:
                    // Terminal — first-write-wins on the resolve flip.
                    await _trackingStore.RecordTerminalAsync(
                        telemetry.Operational.TrackedOperationId,
                        telemetry.Operational.Status,
                        telemetry.Operational.LastError,
                        telemetry.Operational.HttpStatus,
                        ct).ConfigureAwait(false);
                    break;

                default:
                    // Defensive — only the four cached-lifecycle kinds are
                    // expected on this path. Anything else is logged so a
                    // mis-routed packet is visible but never crashes the
                    // forwarder.
                    _logger.LogWarning(
                        "CachedCallTelemetryForwarder: unexpected audit kind {Kind} on tracking emission for EventId {EventId}",
                        audit.Kind, audit.EventId);
                    break;
            }
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "CachedCallTelemetryForwarder: tracking-store emission threw for TrackedOperationId {Id} (Status {Status}, sqlite {SqliteError})",
                telemetry.Operational.TrackedOperationId, telemetry.Operational.Status, SqliteErrorCodes.Describe(ex));
        }
    }
}