ScadaBridge/src/ZB.MOM.WW.ScadaBridge.CLI/Commands/AuditExportHelpers.cs

using System.Globalization;
using System.Net;
using System.Text.Json;

namespace ZB.MOM.WW.ScadaBridge.CLI.Commands;

/// <summary>
/// Filter + destination arguments for an <c>audit export</c> invocation. Mirrors the
/// <c>GET /api/audit/export</c> parameters.
/// <see cref="Channel"/>/<see cref="Kind"/>/<see cref="Status"/>/<see cref="Site"/>
/// are multi-valued — each supplied value becomes a repeated query-string param so
/// the server's multi-value <c>IN (…)</c> filter sees the full set, exactly like
/// the <c>audit query</c> subcommand.
/// </summary>
public sealed class AuditExportArgs
{
    /// <summary>
    /// Start timestamp for the export time window.
    /// </summary>
    public string Since { get; set; } = string.Empty;
    /// <summary>
    /// End timestamp for the export time window.
    /// </summary>
    public string Until { get; set; } = string.Empty;
    /// <summary>
    /// Export format (e.g., 'json', 'csv', 'parquet').
    /// </summary>
    public string Format { get; set; } = string.Empty;
    /// <summary>
    /// Output file path for the exported audit log.
    /// </summary>
    public string Output { get; set; } = string.Empty;
    /// <summary>
    /// Channel filter values (repeated query parameter).
    /// </summary>
    public string[] Channel { get; set; } = Array.Empty<string>();
    /// <summary>
    /// Kind filter values (repeated query parameter).
    /// </summary>
    public string[] Kind { get; set; } = Array.Empty<string>();
    /// <summary>
    /// Status filter values (repeated query parameter).
    /// </summary>
    public string[] Status { get; set; } = Array.Empty<string>();
    /// <summary>
    /// Site identifier filter values (repeated query parameter).
    /// </summary>
    public string[] Site { get; set; } = Array.Empty<string>();
    /// <summary>
    /// Optional target system filter.
    /// </summary>
    public string? Target { get; set; }
    /// <summary>
    /// Optional actor/user filter.
    /// </summary>
    public string? Actor { get; set; }
}

/// <summary>
/// Helpers for the <c>audit export</c> subcommand: builds the export query string and
/// streams the HTTP response body straight to the destination file without buffering
/// the (potentially multi-megabyte) export in memory.
/// </summary>
public static class AuditExportHelpers
{
    /// <summary>
    /// Builds the <c>?...</c> query string for <c>GET /api/audit/export</c>: the required
    /// time window + format, plus optional filters. Time-specs are resolved via
    /// <see cref="AuditQueryHelpers.ResolveTimeSpec"/>. The multi-valued
    /// <c>--channel</c>/<c>--kind</c>/<c>--status</c>/<c>--site</c> filters each emit ONE
    /// repeated query-string key per value (e.g. <c>channel=A&amp;channel=B</c>) so the
    /// server's multi-value <c>IN (…)</c> filter receives the full set — mirroring
    /// <see cref="AuditQueryHelpers.BuildQueryString"/>.
    /// </summary>
    /// <param name="args">The export arguments containing filters and format.</param>
    /// <param name="now">The current time for resolving relative time specifications.</param>
    /// <returns>The full query string (including the leading <c>?</c>) for the export endpoint.</returns>
    public static string BuildQueryString(AuditExportArgs args, DateTimeOffset now)
    {
        var parts = new List<string>();

        void Add(string key, string? value)
        {
            if (!string.IsNullOrWhiteSpace(value))
                parts.Add($"{key}={Uri.EscapeDataString(value)}");
        }

        void AddEach(string key, IReadOnlyList<string> values)
        {
            foreach (var value in values)
            {
                Add(key, value);
            }
        }

        Add("fromUtc", AuditQueryHelpers.ResolveTimeSpec(args.Since, now).ToString("o", CultureInfo.InvariantCulture));
        Add("toUtc", AuditQueryHelpers.ResolveTimeSpec(args.Until, now).ToString("o", CultureInfo.InvariantCulture));
        Add("format", args.Format);
        AddEach("channel", args.Channel);
        AddEach("kind", args.Kind);
        AddEach("status", args.Status);
        AddEach("sourceSiteId", args.Site);
        Add("target", args.Target);
        Add("actor", args.Actor);

        return "?" + string.Join("&", parts);
    }

    /// <summary>
    /// Deadline for receiving the export response's headers (i.e. the server has
    /// started responding), measured from the start of the request. This is the
    /// one bound <see cref="ManagementHttpClient.SendGetStreamAsync"/> does NOT
    /// carry on its own — <see cref="ManagementHttpClient"/>'s <c>HttpClient.Timeout</c>
    /// is deliberately <see cref="Timeout.InfiniteTimeSpan"/> (see its doc comment)
    /// and <see cref="SocketsHttpHandler.ConnectTimeout"/> bounds only the TCP/TLS
    /// connect phase, not a connected-but-silent server. Reuses
    /// <see cref="ManagementHttpClient.DefaultConnectTimeout"/> as the standard CLI
    /// per-call default.
    /// </summary>
    public static readonly TimeSpan DefaultHeadersTimeout = ManagementHttpClient.DefaultConnectTimeout;

    /// <summary>
    /// Inactivity deadline for the streaming body copy: reset on every chunk
    /// successfully read, so a large export keeps streaming indefinitely as long
    /// as bytes keep arriving, but a stream that goes silent mid-transfer (a
    /// hung/crashed server holding the connection open) is bounded rather than
    /// hanging the CLI forever — the residual gap the timeout-honesty fix
    /// (<c>ManagementHttpClient.DefaultConnectTimeout</c> doc comment) left in this
    /// one path.
    /// </summary>
    public static readonly TimeSpan DefaultInactivityTimeout = ManagementHttpClient.DefaultConnectTimeout;

    /// <summary>
    /// Executes the export: GETs <c>/api/audit/export</c> and copies the response body
    /// stream directly to <see cref="AuditExportArgs.Output"/>. The body is never fully
    /// buffered — the copy streams in fixed-size chunks via <see cref="Stream.ReadAsync(Memory{byte}, CancellationToken)"/>.
    /// A <c>501 Not Implemented</c> (parquet not yet supported server-side) prints the
    /// server message and returns a non-zero exit code.
    /// </summary>
    /// <param name="client">The management HTTP client for API communication.</param>
    /// <param name="args">The export arguments containing filters and output file path.</param>
    /// <param name="output">Text writer for command output messages.</param>
    /// <param name="now">The current time for resolving relative time specifications.</param>
    /// <param name="cancellationToken">
    /// External cancellation (e.g. the CLI's Ctrl-C handler) — cancelling this
    /// stops the export promptly at either the headers wait or the next stream
    /// read, with a "cancelled" (not timed-out) message.
    /// </param>
    /// <param name="headersTimeout">
    /// Override for <see cref="DefaultHeadersTimeout"/> — test seam.
    /// </param>
    /// <param name="inactivityTimeout">
    /// Override for <see cref="DefaultInactivityTimeout"/> — test seam.
    /// </param>
    /// <returns>0 on success, 1 on general error, or 2 on authorization failure.</returns>
    public static async Task<int> RunExportAsync(
        ManagementHttpClient client, AuditExportArgs args, TextWriter output, DateTimeOffset now,
        CancellationToken cancellationToken = default,
        TimeSpan? headersTimeout = null,
        TimeSpan? inactivityTimeout = null)
    {
        var qs = BuildQueryString(args, now);

        using var headersCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        headersCts.CancelAfter(headersTimeout ?? DefaultHeadersTimeout);

        HttpResponseMessage response;
        try
        {
            response = await client.SendGetStreamAsync("api/audit/export" + qs, headersCts.Token);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            OutputFormatter.WriteError("Export cancelled.", "CANCELLED");
            return 1;
        }
        catch (OperationCanceledException)
        {
            OutputFormatter.WriteError("Request timed out waiting for the server to respond.", "TIMEOUT");
            return 1;
        }
        catch (HttpRequestException ex)
        {
            OutputFormatter.WriteError($"Connection failed: {ex.Message}", "CONNECTION_FAILED");
            return 1;
        }

        using (response)
        {
            if (response.StatusCode == HttpStatusCode.NotImplemented)
            {
                var message = await response.Content.ReadAsStringAsync(cancellationToken);
                OutputFormatter.WriteError(
                    string.IsNullOrWhiteSpace(message)
                        ? "Export format not implemented by the server."
                        : message,
                    "NOT_IMPLEMENTED");
                return 1;
            }

            if (!response.IsSuccessStatusCode)
            {
                var message = await response.Content.ReadAsStringAsync(cancellationToken);
                // Honour the documented "authorization failure → exit 2"
                // contract on the REST audit surface as well. HTTP 403 is the
                // primary signal; the server may also surface UNAUTHORIZED /
                // FORBIDDEN via the JSON error envelope on a status other than 403.
                var errorCode = TryExtractErrorCode(message);
                var isAuthFailure = (int)response.StatusCode == 403
                    || string.Equals(errorCode, "FORBIDDEN", StringComparison.OrdinalIgnoreCase)
                    || string.Equals(errorCode, "UNAUTHORIZED", StringComparison.OrdinalIgnoreCase);
                OutputFormatter.WriteError(
                    string.IsNullOrWhiteSpace(message) ? $"Export failed (HTTP {(int)response.StatusCode})." : message,
                    errorCode ?? "ERROR");
                return isAuthFailure ? 2 : 1;
            }

            await using var source = await response.Content.ReadAsStreamAsync(cancellationToken);
            await using var destination = new FileStream(
                args.Output, FileMode.Create, FileAccess.Write, FileShare.None,
                bufferSize: 81920, useAsync: true);

            try
            {
                await CopyWithInactivityTimeoutAsync(
                    source, destination, inactivityTimeout ?? DefaultInactivityTimeout, cancellationToken);
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                OutputFormatter.WriteError($"Export cancelled; partial file written to {args.Output}.", "CANCELLED");
                return 1;
            }
            catch (OperationCanceledException)
            {
                OutputFormatter.WriteError(
                    "Export stalled — no data received from the server for "
                        + $"{(inactivityTimeout ?? DefaultInactivityTimeout).TotalSeconds:0}s; "
                        + $"partial file written to {args.Output}.",
                    "TIMEOUT");
                return 1;
            }
            catch (Exception ex) when (ex is IOException or HttpRequestException)
            {
                OutputFormatter.WriteError(
                    $"Export failed while streaming the response: {ex.Message}; partial file written to {args.Output}.",
                    "STREAM_FAILED");
                return 1;
            }
        }

        output.WriteLine($"Exported audit log to {args.Output}");
        return 0;
    }

    /// <summary>
    /// Copies <paramref name="source"/> to <paramref name="destination"/> in fixed-size
    /// chunks, rearming an inactivity deadline before every read so the copy is bounded
    /// only by silence, not by total size — a multi-hundred-MB export keeps streaming as
    /// long as bytes keep arriving. <paramref name="cancellationToken"/> (external, e.g.
    /// Ctrl-C) and the inactivity deadline share one linked token; callers distinguish the
    /// two causes via <c>cancellationToken.IsCancellationRequested</c> after catching
    /// <see cref="OperationCanceledException"/>.
    /// </summary>
    private static async Task CopyWithInactivityTimeoutAsync(
        Stream source, Stream destination, TimeSpan inactivityTimeout, CancellationToken cancellationToken)
    {
        using var readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        var buffer = new byte[81920];
        while (true)
        {
            readCts.CancelAfter(inactivityTimeout);
            var bytesRead = await source.ReadAsync(buffer.AsMemory(), readCts.Token);
            if (bytesRead == 0)
            {
                break;
            }
            await destination.WriteAsync(buffer.AsMemory(0, bytesRead), cancellationToken);
        }
    }

    /// <summary>
    /// Best-effort parse of the server's JSON error envelope (<c>{ "error": ..., "code": ... }</c>)
    /// to extract the <c>code</c> field. Returns null if the body is empty, not valid JSON, or
    /// has no <c>code</c> property — callers fall back to "ERROR" in that case.
    /// </summary>
    /// <param name="body">The HTTP response body string to parse for an error code.</param>
    /// <returns>The <c>code</c> string from the JSON error envelope, or null if absent or unparseable.</returns>
    internal static string? TryExtractErrorCode(string body)
    {
        if (string.IsNullOrWhiteSpace(body))
            return null;

        try
        {
            using var doc = JsonDocument.Parse(body);
            if (doc.RootElement.ValueKind == JsonValueKind.Object
                && doc.RootElement.TryGetProperty("code", out var codeProp)
                && codeProp.ValueKind == JsonValueKind.String)
            {
                return codeProp.GetString();
            }
        }
        catch (JsonException)
        {
            // Body is not a JSON envelope (e.g. an HTML proxy error page); no code to extract.
        }

        return null;
    }
}