using Parquet;
using Parquet.Schema;

namespace ZB.MOM.WW.ScadaBridge.ManagementService;

/// <summary>
/// <c>GET /api/audit/export?format=parquet</c> (ScadaBridge#42): the same 24 columns as the
/// csv / jsonl export (<see cref="AuditExportRow"/>, in <see cref="AuditEndpoints.CsvHeader"/>
/// order) as one Parquet file, one row group per <see cref="RowGroupRows"/> rows.
/// </summary>
/// <remarks>
/// <para>
/// Types: ids are <c>D</c>-format strings, timestamps are UTC Parquet timestamps
/// (microseconds), enums are their names, <c>HttpStatus</c>/<c>DurationMs</c> are int32,
/// <c>PayloadTruncated</c> is boolean, everything else is a UTF-8 string.
/// </para>
/// <para>
/// <b>Streaming.</b> Kestrel forbids synchronous writes to the response body and the Parquet
/// writer issues some, so each row group is encoded into a small in-memory buffer and then
/// copied to the response asynchronously. Memory is bounded by one row group, and the client
/// starts receiving bytes after the first one. A Parquet file's footer is written last, so a
/// response cut off mid-stream is not a readable file — the CLI treats it as a failed export.
/// </para>
/// <para>
/// This is the ad-hoc export. The retention archive (alog.md §15.2) is a different layout
/// that carries the hash-chain columns; see <c>AuditArchiveParquet</c>.
/// </para>
/// </remarks>
public static class AuditExportParquet
{
    /// <summary>Rows per Parquet row group.</summary>
    public const int RowGroupRows = 10_000;

    private static readonly DataField<string> EventId = new(nameof(AuditExportRow.EventId), false);
    private static readonly DateTimeDataField OccurredAtUtc = Timestamp(nameof(AuditExportRow.OccurredAtUtc), false);
    private static readonly DateTimeDataField IngestedAtUtc = Timestamp(nameof(AuditExportRow.IngestedAtUtc), true);
    private static readonly DataField<string> Channel = new(nameof(AuditExportRow.Channel), false);
    private static readonly DataField<string> Kind = new(nameof(AuditExportRow.Kind), false);
    private static readonly DataField<string> CorrelationId = Text(nameof(AuditExportRow.CorrelationId));
    private static readonly DataField<string> ExecutionId = Text(nameof(AuditExportRow.ExecutionId));
    private static readonly DataField<string> ParentExecutionId = Text(nameof(AuditExportRow.ParentExecutionId));
    private static readonly DataField<string> SourceSiteId = Text(nameof(AuditExportRow.SourceSiteId));
    private static readonly DataField<string> SourceNode = Text(nameof(AuditExportRow.SourceNode));
    private static readonly DataField<string> SourceInstanceId = Text(nameof(AuditExportRow.SourceInstanceId));
    private static readonly DataField<string> SourceScript = Text(nameof(AuditExportRow.SourceScript));
    private static readonly DataField<string> Actor = Text(nameof(AuditExportRow.Actor));
    private static readonly DataField<string> Target = Text(nameof(AuditExportRow.Target));
    private static readonly DataField<string> Status = new(nameof(AuditExportRow.Status), false);
    private static readonly DataField<int> HttpStatus = new(nameof(AuditExportRow.HttpStatus), true);
    private static readonly DataField<int> DurationMs = new(nameof(AuditExportRow.DurationMs), true);
    private static readonly DataField<string> ErrorMessage = Text(nameof(AuditExportRow.ErrorMessage));
    private static readonly DataField<string> ErrorDetail = Text(nameof(AuditExportRow.ErrorDetail));
    private static readonly DataField<string> RequestSummary = Text(nameof(AuditExportRow.RequestSummary));
    private static readonly DataField<string> ResponseSummary = Text(nameof(AuditExportRow.ResponseSummary));
    private static readonly DataField<bool> PayloadTruncated = new(nameof(AuditExportRow.PayloadTruncated), false);
    private static readonly DataField<string> Extra = Text(nameof(AuditExportRow.Extra));
    private static readonly DataField<string> ForwardState = Text(nameof(AuditExportRow.ForwardState));

    /// <summary>The export schema; column names and order equal <see cref="AuditEndpoints.CsvHeader"/>.</summary>
    public static ParquetSchema Schema { get; } = new(
        EventId, OccurredAtUtc, IngestedAtUtc, Channel, Kind, CorrelationId, ExecutionId, ParentExecutionId,
        SourceSiteId, SourceNode, SourceInstanceId, SourceScript, Actor, Target, Status, HttpStatus, DurationMs,
        ErrorMessage, ErrorDetail, RequestSummary, ResponseSummary, PayloadTruncated, Extra, ForwardState);

    /// <summary>Writes every row of <paramref name="pages"/> to <paramref name="output"/> as one Parquet file.</summary>
    /// <param name="pages">Export rows, page by page.</param>
    /// <param name="output">The destination (e.g. the HTTP response body); written asynchronously only.</param>
    /// <param name="ct">Cancellation token.</param>
    /// <returns>The number of rows written.</returns>
    public static async Task<long> WriteAsync(
        IAsyncEnumerable<IReadOnlyList<AuditExportRow>> pages, Stream output, CancellationToken ct = default)
    {
        ArgumentNullException.ThrowIfNull(pages);
        ArgumentNullException.ThrowIfNull(output);

        // Non-seekable on purpose: the writer then tracks file offsets itself, so emptying the
        // buffer after each drain cannot corrupt the column-chunk offsets in the footer.
        var staging = new StagingStream();
        long total = 0;
        var buffer = new List<AuditExportRow>(RowGroupRows);
        await using (var writer = await ParquetWriter.CreateAsync(Schema, staging, cancellationToken: ct).ConfigureAwait(false))
        {
            await foreach (var page in pages.WithCancellation(ct).ConfigureAwait(false))
            {
                foreach (var row in page)
                {
                    buffer.Add(row);
                    if (buffer.Count == RowGroupRows)
                    {
                        total += await WriteGroupAsync(writer, buffer, ct).ConfigureAwait(false);
                        await DrainAsync(staging, output, ct).ConfigureAwait(false);
                    }
                }
            }

            total += await WriteGroupAsync(writer, buffer, ct).ConfigureAwait(false);
        }

        // The footer is written when the writer is disposed.
        await DrainAsync(staging, output, ct).ConfigureAwait(false);
        return total;
    }

    private static async Task<int> WriteGroupAsync(ParquetWriter writer, List<AuditExportRow> rows, CancellationToken ct)
    {
        if (rows.Count == 0)
        {
            return 0;
        }

        using (var group = writer.CreateRowGroup())
        {
            await group.WriteAsync(EventId, Map(rows, r => r.EventId.ToString("D"))).ConfigureAwait(false);
            await group.WriteAsync<DateTime>(OccurredAtUtc, rows.Select(r => DateTime.SpecifyKind(r.OccurredAtUtc, DateTimeKind.Utc)).ToArray(), cancellationToken: ct).ConfigureAwait(false);
            await group.WriteAsync<DateTime>(IngestedAtUtc, rows.Select(r => r.IngestedAtUtc is { } i ? DateTime.SpecifyKind(i, DateTimeKind.Utc) : (DateTime?)null).ToArray(), cancellationToken: ct).ConfigureAwait(false);
            await group.WriteAsync(Channel, Map(rows, r => r.Channel.ToString())).ConfigureAwait(false);
            await group.WriteAsync(Kind, Map(rows, r => r.Kind.ToString())).ConfigureAwait(false);
            await group.WriteAsync(CorrelationId, Map(rows, r => r.CorrelationId?.ToString("D"))!).ConfigureAwait(false);
            await group.WriteAsync(ExecutionId, Map(rows, r => r.ExecutionId?.ToString("D"))!).ConfigureAwait(false);
            await group.WriteAsync(ParentExecutionId, Map(rows, r => r.ParentExecutionId?.ToString("D"))!).ConfigureAwait(false);
            await group.WriteAsync(SourceSiteId, Map(rows, r => r.SourceSiteId)!).ConfigureAwait(false);
            await group.WriteAsync(SourceNode, Map(rows, r => r.SourceNode)!).ConfigureAwait(false);
            await group.WriteAsync(SourceInstanceId, Map(rows, r => r.SourceInstanceId)!).ConfigureAwait(false);
            await group.WriteAsync(SourceScript, Map(rows, r => r.SourceScript)!).ConfigureAwait(false);
            await group.WriteAsync(Actor, Map(rows, r => r.Actor)!).ConfigureAwait(false);
            await group.WriteAsync(Target, Map(rows, r => r.Target)!).ConfigureAwait(false);
            await group.WriteAsync(Status, Map(rows, r => r.Status.ToString())).ConfigureAwait(false);
            await group.WriteAsync<int>(HttpStatus, rows.Select(r => r.HttpStatus).ToArray(), cancellationToken: ct).ConfigureAwait(false);
            await group.WriteAsync<int>(DurationMs, rows.Select(r => r.DurationMs).ToArray(), cancellationToken: ct).ConfigureAwait(false);
            await group.WriteAsync(ErrorMessage, Map(rows, r => r.ErrorMessage)!).ConfigureAwait(false);
            await group.WriteAsync(ErrorDetail, Map(rows, r => r.ErrorDetail)!).ConfigureAwait(false);
            await group.WriteAsync(RequestSummary, Map(rows, r => r.RequestSummary)!).ConfigureAwait(false);
            await group.WriteAsync(ResponseSummary, Map(rows, r => r.ResponseSummary)!).ConfigureAwait(false);
            await group.WriteAsync<bool>(PayloadTruncated, rows.Select(r => r.PayloadTruncated).ToArray(), cancellationToken: ct).ConfigureAwait(false);
            await group.WriteAsync(Extra, Map(rows, r => r.Extra)!).ConfigureAwait(false);
            await group.WriteAsync(ForwardState, Map(rows, r => r.ForwardState?.ToString())!).ConfigureAwait(false);
        }

        var count = rows.Count;
        rows.Clear();
        return count;
    }

    private static async Task DrainAsync(StagingStream staging, Stream output, CancellationToken ct)
    {
        var pending = staging.Pending;
        if (pending.Length == 0)
        {
            return;
        }

        await output.WriteAsync(pending, ct).ConfigureAwait(false);
        await output.FlushAsync(ct).ConfigureAwait(false);
        staging.Clear();
    }

    private static string?[] Map(List<AuditExportRow> rows, Func<AuditExportRow, string?> select)
        => rows.Select(select).ToArray();

    private static DataField<string> Text(string name) => new(name, true);

    private static DateTimeDataField Timestamp(string name, bool nullable)
        => new(name, DateTimeFormat.Timestamp, isAdjustedToUTC: true, unit: DateTimeTimeUnit.Micros, isNullable: nullable);

    /// <summary>
    /// Forward-only, non-seekable buffer the Parquet writer writes into (synchronously or
    /// not); <see cref="DrainAsync"/> moves its bytes to the real output asynchronously.
    /// </summary>
    private sealed class StagingStream : Stream
    {
        private readonly MemoryStream _buffer = new();

        public ReadOnlyMemory<byte> Pending => _buffer.GetBuffer().AsMemory(0, (int)_buffer.Length);

        public override bool CanRead => false;

        public override bool CanSeek => false;

        public override bool CanWrite => true;

        public override long Length => throw new NotSupportedException();

        public override long Position
        {
            get => throw new NotSupportedException();
            set => throw new NotSupportedException();
        }

        public void Clear() => _buffer.SetLength(0);

        public override void Write(byte[] buffer, int offset, int count) => _buffer.Write(buffer, offset, count);

        public override void Write(ReadOnlySpan<byte> buffer) => _buffer.Write(buffer);

        public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
        {
            _buffer.Write(buffer.Span);
            return ValueTask.CompletedTask;
        }

        public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        {
            _buffer.Write(buffer, offset, count);
            return Task.CompletedTask;
        }

        public override void Flush()
        {
        }

        public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;

        public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();

        public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();

        public override void SetLength(long value) => throw new NotSupportedException();
    }
}