// Copyright 2019-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Adapted from server/stream.go and server/consumer.go in the NATS server Go source.

using System.Text.Json.Serialization;

namespace ZB.MOM.NatsNet.Server;

// ============================================================================
// Stream API types (from stream.go)
// ============================================================================

/// <summary>
/// A stream create request that extends <see cref="StreamConfig"/> with a pedantic flag.
/// Mirrors <c>streamConfigRequest</c> in server/stream.go.
/// </summary>
public sealed class StreamConfigRequest
{
    [JsonPropertyName("config")]
    public StreamConfig Config { get; set; } = new();

    /// <summary>If true, strict validation is applied during stream creation/update.</summary>
    [JsonPropertyName("pedantic")]
    public bool Pedantic { get; set; }
}

/// <summary>
/// Information about a stream, returned from info requests.
/// Mirrors <c>StreamInfo</c> in server/stream.go.
/// </summary>
public sealed class StreamInfo
{
    [JsonPropertyName("config")]
    public StreamConfig Config { get; set; } = new();

    [JsonPropertyName("created")]
    public DateTime Created { get; set; }

    [JsonPropertyName("state")]
    public StreamState State { get; set; } = new();

    [JsonPropertyName("mirror")]
    public StreamSourceInfo? Mirror { get; set; }

    [JsonPropertyName("sources")]
    public StreamSourceInfo[]? Sources { get; set; }

    [JsonPropertyName("cluster")]
    public ClusterInfo? Cluster { get; set; }

    [JsonPropertyName("mirror_direct")]
    public bool Mirror_Direct { get; set; }

    [JsonPropertyName("allow_direct")]
    public bool Allow_Direct { get; set; }

    /// <summary>Alternate cluster name.</summary>
    [JsonPropertyName("alternates")]
    public string? Alt { get; set; }
}

/// <summary>
/// Information about a stream mirror or source.
/// Mirrors <c>StreamSourceInfo</c> in server/stream.go.
/// </summary>
public sealed class StreamSourceInfo
{
    [JsonPropertyName("name")]
    public string Name { get; set; } = string.Empty;

    [JsonPropertyName("filter_subject")]
    public string? FilterSubject { get; set; }

    [JsonPropertyName("lag")]
    public ulong Lag { get; set; }

    [JsonPropertyName("active")]
    public DateTime? Active { get; set; }

    [JsonPropertyName("external")]
    public StreamSource? External { get; set; }

    [JsonPropertyName("error")]
    public string? Error { get; set; }

    internal bool IsCurrentSub(string reply)
    {
        if (string.IsNullOrWhiteSpace(reply) || string.IsNullOrWhiteSpace(Name))
            return false;

        return reply.Contains($".{Name}.", StringComparison.Ordinal) || reply.EndsWith($".{Name}", StringComparison.Ordinal);
    }

    internal byte[] GenSourceHeader(ulong sourceSequence)
    {
        var header = $"NATS/1.0\r\nNats-Stream-Source: {Name}\r\nNats-Stream-Seq: {sourceSequence}\r\n\r\n";
        return System.Text.Encoding.ASCII.GetBytes(header);
    }
}

/// <summary>
/// Request parameters for stream info, allowing filtering.
/// Mirrors <c>streamInfoRequest</c> in server/stream.go.
/// </summary>
public sealed class StreamInfoRequest
{
    [JsonPropertyName("subjects_filter")]
    public string? SubjectsFilter { get; set; }

    [JsonPropertyName("mirror_check_until")]
    public string? MirrorCheckUntil { get; set; }

    [JsonPropertyName("deleted_details")]
    public bool DeletedDetails { get; set; }

    [JsonPropertyName("subjects_detail")]
    public bool SubjectsDetail { get; set; }
}

/// <summary>
/// Request parameters for purging a stream.
/// Mirrors <c>StreamPurgeRequest</c> in server/stream.go.
/// </summary>
public sealed class StreamPurgeRequest
{
    [JsonPropertyName("filter")]
    public string? Filter { get; set; }

    [JsonPropertyName("seq")]
    public ulong Sequence { get; set; }

    [JsonPropertyName("keep")]
    public ulong Keep { get; set; }
}

/// <summary>
/// Request for deleting a specific stream message.
/// Mirrors <c>StreamMsgDeleteRequest</c> in server/stream.go.
/// </summary>
public sealed class StreamMsgDeleteRequest
{
    [JsonPropertyName("seq")]
    public ulong Seq { get; set; }

    [JsonPropertyName("no_erase")]
    public bool NoErase { get; set; }
}

/// <summary>
/// Request for retrieving a specific stream message.
/// Mirrors <c>StreamGetMsgRequest</c> in server/stream.go.
/// </summary>
public sealed class StreamGetMsgRequest
{
    [JsonPropertyName("seq")]
    public ulong Seq { get; set; }

    [JsonPropertyName("last_by_subj")]
    public string? LastBySubject { get; set; }

    [JsonPropertyName("next_by_subj")]
    public string? NextBySubject { get; set; }
}

/// <summary>
/// Publish acknowledgement response from JetStream.
/// Mirrors <c>JSPubAckResponse</c> in server/stream.go.
/// </summary>
public sealed class JSPubAckResponse
{
    [JsonPropertyName("stream")]
    public string Stream { get; set; } = string.Empty;

    [JsonPropertyName("seq")]
    public ulong Seq { get; set; }

    [JsonPropertyName("duplicate")]
    public bool Duplicate { get; set; }

    [JsonPropertyName("domain")]
    public string? Domain { get; set; }

    [JsonPropertyName("error")]
    public JsApiError? PubAckError { get; set; }

    /// <summary>
    /// Returns an exception if the response contains an error, otherwise null.
    /// Mirrors <c>ToError()</c> helper pattern in NATS Go server.
    /// </summary>
    public Exception? ToError()
    {
        if (PubAckError is { ErrCode: > 0 })
            return new InvalidOperationException($"{PubAckError.Description} (errCode={PubAckError.ErrCode})");

        Exception? error = null;
        return error;
    }
}

/// <summary>
/// A raw published message before JetStream processing.
/// Mirrors <c>pubMsg</c> (JetStream variant) in server/stream.go.
/// Note: renamed <c>JsStreamPubMsg</c> to avoid collision with the server-level
/// <c>PubMsg</c> (events.go) which lives in the same namespace.
/// </summary>
public sealed class JsStreamPubMsg
{
    public string Subject { get; set; } = string.Empty;
    public string? Reply { get; set; }
    public byte[]? Hdr { get; set; }
    public byte[]? Msg { get; set; }
    public Dictionary<string, string>? Meta { get; set; }
}

/// <summary>
/// A JetStream publish message with sync tracking.
/// Mirrors <c>jsPubMsg</c> in server/stream.go.
/// </summary>
public sealed partial class JsPubMsg
{
    public string Subject { get; set; } = string.Empty;
    public string? Reply { get; set; }
    public byte[]? Hdr { get; set; }
    public byte[]? Msg { get; set; }

    /// <summary>Publish argument (opaque, set at runtime).</summary>
    public object? Pa { get; set; }

    /// <summary>Sync/ack channel (opaque, set at runtime).</summary>
    public object? Sync { get; set; }
}

/// <summary>
/// An inbound message to be processed by the JetStream layer.
/// Mirrors <c>inMsg</c> in server/stream.go.
/// </summary>
public sealed partial class InMsg
{
    public string Subject { get; set; } = string.Empty;
    public string? Reply { get; set; }
    public byte[]? Hdr { get; set; }

    /// <summary>
    /// Optional sequence number, used for out-of-band sorting (e.g. scheduled messages).
    /// Mirrors Go <c>inMsg.seq uint64</c> in server/stream.go.
    /// </summary>
    public ulong Seq { get; set; }
    public byte[]? Msg { get; set; }

    /// <summary>The originating client (opaque, set at runtime).</summary>
    public object? Client { get; set; }

    internal bool IsControlMsg()
    {
        if (!string.IsNullOrEmpty(Subject))
        {
            if (Subject.StartsWith("$JS.FC.", StringComparison.Ordinal) ||
                Subject.StartsWith("$JS.SYNC.", StringComparison.Ordinal))
                return true;
        }

        if (Hdr == null || Hdr.Length == 0)
            return false;

        var headerText = System.Text.Encoding.ASCII.GetString(Hdr);
        return headerText.Contains("Status: 100", StringComparison.Ordinal) ||
               headerText.Contains("Nats-Consumer-Stalled", StringComparison.Ordinal);
    }
}

/// <summary>
/// A cached/clustered message for replication.
/// Mirrors <c>cMsg</c> in server/stream.go.
/// </summary>
public sealed partial class CMsg
{
    public string Subject { get; set; } = string.Empty;
    public byte[]? Msg { get; set; }
    public ulong Seq { get; set; }
}

// ============================================================================
// Consumer API types (from consumer.go)
// ============================================================================

/// <summary>
/// Information about a consumer, returned from info requests.
/// Mirrors <c>ConsumerInfo</c> in server/consumer.go.
/// </summary>
public sealed class ConsumerInfo
{
    [JsonPropertyName("stream_name")]
    public string Stream { get; set; } = string.Empty;

    [JsonPropertyName("name")]
    public string Name { get; set; } = string.Empty;

    [JsonPropertyName("created")]
    public DateTime Created { get; set; }

    [JsonPropertyName("config")]
    public ConsumerConfig? Config { get; set; }

    [JsonPropertyName("delivered")]
    public SequenceInfo Delivered { get; set; } = new();

    [JsonPropertyName("ack_floor")]
    public SequenceInfo AckFloor { get; set; } = new();

    [JsonPropertyName("num_ack_pending")]
    public int NumAckPending { get; set; }

    [JsonPropertyName("num_redelivered")]
    public int NumRedelivered { get; set; }

    [JsonPropertyName("num_waiting")]
    public int NumWaiting { get; set; }

    [JsonPropertyName("num_pending")]
    public ulong NumPending { get; set; }

    [JsonPropertyName("cluster")]
    public ClusterInfo? Cluster { get; set; }

    [JsonPropertyName("push_bound")]
    public bool PushBound { get; set; }

    [JsonPropertyName("paused")]
    public bool Paused { get; set; }

    [JsonPropertyName("pause_remaining")]
    public TimeSpan PauseRemaining { get; set; }

    [JsonPropertyName("ts")]
    public DateTime TimeStamp { get; set; }

    [JsonPropertyName("priority_groups")]
    public PriorityGroupState[]? PriorityGroups { get; set; }
}

/// <summary>
/// State information for a priority group on a pull consumer.
/// Mirrors <c>PriorityGroupState</c> in server/consumer.go.
/// </summary>
public sealed class PriorityGroupState
{
    [JsonPropertyName("group")]
    public string Group { get; set; } = string.Empty;

    [JsonPropertyName("pinned_client_id")]
    public string? PinnedClientId { get; set; }

    [JsonPropertyName("pinned_ts")]
    public DateTime PinnedTs { get; set; }
}

/// <summary>
/// Sequence information for consumer delivered/ack_floor positions.
/// Mirrors <c>SequenceInfo</c> in server/consumer.go.
/// </summary>
public sealed class SequenceInfo
{
    [JsonPropertyName("consumer_seq")]
    public ulong Consumer { get; set; }

    [JsonPropertyName("stream_seq")]
    public ulong Stream { get; set; }

    [JsonPropertyName("last_active")]
    public DateTime? Last { get; set; }
}

/// <summary>
/// Request to create or update a consumer.
/// Mirrors <c>CreateConsumerRequest</c> in server/consumer.go.
/// </summary>
public sealed class CreateConsumerRequest
{
    [JsonPropertyName("stream_name")]
    public string Stream { get; set; } = string.Empty;

    [JsonPropertyName("config")]
    public ConsumerConfig Config { get; set; } = new();

    [JsonPropertyName("action")]
    public ConsumerAction Action { get; set; }
}

/// <summary>
/// Specifies the intended action when creating a consumer.
/// Mirrors <c>ConsumerAction</c> in server/consumer.go.
/// </summary>
[JsonConverter(typeof(ConsumerActionJsonConverter))]
public enum ConsumerAction
{
    /// <summary>Create a new consumer or update if it already exists.</summary>
    CreateOrUpdate = 0,

    /// <summary>Create a new consumer; fail if it already exists.</summary>
    Create = 1,

    /// <summary>Update an existing consumer; fail if it does not exist.</summary>
    Update = 2,
}

/// <summary>
/// Response for a consumer deletion request.
/// Mirrors <c>ConsumerDeleteResponse</c> in server/consumer.go.
/// </summary>
public sealed class ConsumerDeleteResponse
{
    [JsonPropertyName("success")]
    public bool Success { get; set; }
}

/// <summary>
/// A pending pull request waiting in the wait queue.
/// Mirrors <c>waitingRequest</c> in server/consumer.go.
/// </summary>
public sealed class WaitingRequest
{
    public string Subject { get; set; } = string.Empty;
    public string? Reply { get; set; }

    /// <summary>Number of messages requested.</summary>
    public int N { get; set; }

    /// <summary>Number of messages delivered so far.</summary>
    public int D { get; set; }

    /// <summary>No-wait flag (1 = no wait).</summary>
    public int NoWait { get; set; }

    public DateTime? Expires { get; set; }

    /// <summary>Max byte limit for this batch.</summary>
    public int MaxBytes { get; set; }

    /// <summary>Bytes accumulated so far.</summary>
    public int B { get; set; }

    /// <summary>Optional pull request priority group metadata.</summary>
    public PriorityGroup? PriorityGroup { get; set; }

    public bool RecycleIfDone()
    {
        if (N > 0 || MaxBytes > 0 && B < MaxBytes)
            return false;

        Recycle();
        return true;
    }

    public void Recycle()
    {
        Subject = string.Empty;
        Reply = null;
        N = 0;
        D = 0;
        NoWait = 0;
        Expires = null;
        MaxBytes = 0;
        B = 0;
        PriorityGroup = null;
    }
}

public sealed class WaitingDelivery
{
    public string Reply { get; set; } = string.Empty;
    public ulong Sequence { get; set; }
    public DateTime Created { get; set; } = DateTime.UtcNow;

    public void Recycle()
    {
        Reply = string.Empty;
        Sequence = 0;
        Created = DateTime.UnixEpoch;
    }
}

/// <summary>
/// A circular wait queue for pending pull requests.
/// Mirrors <c>waitQueue</c> in server/consumer.go.
/// </summary>
public sealed class WaitQueue
{
    private readonly List<WaitingRequest> _reqs = new();
    private readonly int _max;
    private int _head;
    private int _tail;

    public WaitQueue(int max = 0)
    {
        _max = max;
    }

    /// <summary>Number of pending requests in the queue.</summary>
    public int Len => _tail - _head;

    /// <summary>Add a waiting request to the tail of the queue.</summary>
    public void Add(WaitingRequest req)
    {
        ArgumentNullException.ThrowIfNull(req);
        _reqs.Add(req);
        _tail++;
    }

    /// <summary>
    /// Add a waiting request ordered by priority while preserving FIFO order
    /// within each priority level.
    /// </summary>
    public bool AddPrioritized(WaitingRequest req)
    {
        ArgumentNullException.ThrowIfNull(req);
        if (IsFull(_max))
            return false;
        InsertSorted(req);
        return true;
    }

    /// <summary>Insert a request in priority order (lower number = higher priority).</summary>
    public void InsertSorted(WaitingRequest req)
    {
        ArgumentNullException.ThrowIfNull(req);

        if (Len == 0)
        {
            Add(req);
            return;
        }

        var priority = PriorityOf(req);
        var insertAt = _head;
        while (insertAt < _tail)
        {
            if (PriorityOf(_reqs[insertAt]) > priority)
                break;
            insertAt++;
        }

        _reqs.Insert(insertAt, req);
        _tail++;
    }

    /// <summary>Peek at the head request without removing it.</summary>
    public WaitingRequest? Peek()
    {
        if (Len == 0)
        {
            WaitingRequest? none = null;
            return none;
        }
        return _reqs[_head];
    }

    /// <summary>Remove and return the head request.</summary>
    public WaitingRequest? Pop()
    {
        var wr = Peek();
        if (wr is null)
        {
            WaitingRequest? none = null;
            return none;
        }

        wr.D++;
        wr.N--;
        if (wr.N > 0 && Len > 1)
        {
            RemoveCurrent();
            Add(wr);
        }
        else if (wr.N <= 0)
        {
            RemoveCurrent();
        }

        return wr;
    }

    /// <summary>Returns true if the queue contains no active requests.</summary>
    public bool IsEmpty() => Len == 0;

    /// <summary>Rotate the head request to the tail.</summary>
    public void Cycle()
    {
        var wr = Peek();
        if (wr is null)
            return;

        RemoveCurrent();
        Add(wr);
    }

    /// <summary>Pop strategy used by pull consumers based on priority policy.</summary>
    public WaitingRequest? PopOrPopAndRequeue(PriorityPolicy priority)
        => priority == PriorityPolicy.PriorityPrioritized ? PopAndRequeue() : Pop();

    /// <summary>
    /// Pop and requeue to the end of the same priority band while preserving
    /// stable order within that band.
    /// </summary>
    public WaitingRequest? PopAndRequeue()
    {
        var wr = Peek();
        if (wr is null)
        {
            WaitingRequest? none = null;
            return none;
        }

        wr.D++;
        wr.N--;

        if (wr.N > 0 && Len > 1)
        {
            // Remove the current head and insert it back in priority order.
            _reqs.RemoveAt(_head);
            _tail--;
            InsertSorted(wr);
        }
        else if (wr.N <= 0)
        {
            RemoveCurrent();
        }

        return wr;
    }

    /// <summary>Remove the current head request from the queue.</summary>
    public void RemoveCurrent() => Remove(null, Peek());

    /// <summary>Remove a specific request from the queue.</summary>
    public void Remove(WaitingRequest? pre, WaitingRequest? wr)
    {
        if (wr is null || Len == 0)
            return;

        var removeAt = -1;

        if (pre is not null)
        {
            for (var i = _head; i < _tail; i++)
            {
                if (!ReferenceEquals(_reqs[i], pre))
                    continue;

                var candidate = i + 1;
                if (candidate < _tail && ReferenceEquals(_reqs[candidate], wr))
                    removeAt = candidate;
                break;
            }
        }

        if (removeAt < 0)
        {
            for (var i = _head; i < _tail; i++)
            {
                if (ReferenceEquals(_reqs[i], wr))
                {
                    removeAt = i;
                    break;
                }
            }
        }

        if (removeAt < 0)
            return;

        if (removeAt == _head)
        {
            _head++;
        }
        else
        {
            _reqs.RemoveAt(removeAt);
            _tail--;
        }

        if (_head > 32 && _head * 2 >= _tail)
            Compress();
    }

    /// <summary>Compact the internal backing list to reclaim removed slots.</summary>
    public void Compress()
    {
        if (_head == 0)
            return;

        _reqs.RemoveRange(0, _head);
        _tail -= _head;
        _head = 0;
    }

    /// <summary>Returns true if the queue is at capacity (head == tail when full).</summary>
    public bool IsFull(int max)
    {
        if (max <= 0)
            return false;
        return Len >= max;
    }

    /// <summary>
    /// Returns an ordered snapshot of active queue entries from head to tail.
    /// </summary>
    public IReadOnlyList<WaitingRequest> Snapshot()
    {
        if (Len == 0)
            return [];

        var snapshot = new List<WaitingRequest>(Len);
        for (var i = _head; i < _tail; i++)
            snapshot.Add(_reqs[i]);
        return snapshot;
    }

    private static int PriorityOf(WaitingRequest req) => req.PriorityGroup?.Priority ?? int.MaxValue;

    public static WaitQueue NewWaitQueue(int max = 0) => new(max);
}

/// <summary>
/// Cluster membership and leadership information for a stream or consumer.
/// Mirrors <c>ClusterInfo</c> in server/consumer.go and server/stream.go.
/// </summary>
public sealed class ClusterInfo
{
    [JsonPropertyName("name")]
    public string? Name { get; set; }

    [JsonPropertyName("leader")]
    public string? Leader { get; set; }

    [JsonPropertyName("replicas")]
    public PeerInfo[]? Replicas { get; set; }
}

/// <summary>
/// Information about a peer in a JetStream Raft group.
/// Mirrors <c>PeerInfo</c> in server/consumer.go and server/stream.go.
/// </summary>
public sealed class PeerInfo
{
    [JsonPropertyName("name")]
    public string Name { get; set; } = string.Empty;

    [JsonPropertyName("current")]
    public bool Current { get; set; }

    [JsonPropertyName("offline")]
    public bool Offline { get; set; }

    [JsonPropertyName("active")]
    public TimeSpan Active { get; set; }

    [JsonPropertyName("lag")]
    public ulong Lag { get; set; }
}
