// 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/store.go, server/stream.go, server/consumer.go, server/filestore.go, server/disk_avail.go

using System.Text.Json.Serialization;
using ZB.MOM.NatsNet.Server.Internal.DataStructures;

namespace ZB.MOM.NatsNet.Server;

// ---------------------------------------------------------------------------
// StorageType
// ---------------------------------------------------------------------------

/// <summary>Determines how messages are stored for retention.</summary>
[JsonConverter(typeof(StorageTypeJsonConverter))]
public enum StorageType
{
    /// <summary>On disk, designated by the JetStream config StoreDir.</summary>
    FileStorage = 22,

    /// <summary>In memory only.</summary>
    MemoryStorage = 33,
}

// ---------------------------------------------------------------------------
// Well-known store errors
// ---------------------------------------------------------------------------

public static class StoreErrors
{
    /// <summary>Returned when the store has been closed.</summary>
    public static readonly Exception ErrStoreClosed = new InvalidOperationException("store is closed");

    /// <summary>Returned when a message was not found but was expected.</summary>
    public static readonly Exception ErrStoreMsgNotFound = new KeyNotFoundException("no message found");

    /// <summary>Returned when message seq is greater than the last sequence.</summary>
    public static readonly Exception ErrStoreEOF = new EndOfStreamException("stream store EOF");

    /// <summary>Returned when DiscardNew is set and the message limit is reached.</summary>
    public static readonly Exception ErrMaxMsgs = new InvalidOperationException("maximum messages exceeded");

    /// <summary>Returned when DiscardNew is set and the bytes limit is reached.</summary>
    public static readonly Exception ErrMaxBytes = new InvalidOperationException("maximum bytes exceeded");

    /// <summary>Returned when DiscardNew is set and the per-subject message limit is reached.</summary>
    public static readonly Exception ErrMaxMsgsPerSubject = new InvalidOperationException("maximum messages per subject exceeded");

    /// <summary>Returned when RemoveMsg or EraseMsg is called while a snapshot is in progress.</summary>
    public static readonly Exception ErrStoreSnapshotInProgress = new InvalidOperationException("snapshot in progress");

    /// <summary>Returned when a message is considered too large.</summary>
    public static readonly Exception ErrMsgTooLarge = new InvalidOperationException("message too large");

    /// <summary>Returned when accessing the wrong storage type.</summary>
    public static readonly Exception ErrStoreWrongType = new InvalidOperationException("wrong storage type");

    /// <summary>Returned when trying to update a consumer's acks with no ack policy.</summary>
    public static readonly Exception ErrNoAckPolicy = new InvalidOperationException("ack policy is none");

    /// <summary>Returned when storing a raw message with a mismatched expected sequence.</summary>
    public static readonly Exception ErrSequenceMismatch = new InvalidOperationException("expected sequence does not match store");

    /// <summary>Returned when a stream state snapshot is corrupt.</summary>
    public static readonly Exception ErrCorruptStreamState = new InvalidDataException("stream state snapshot is corrupt");

    /// <summary>Returned when a request yields too many matching results.</summary>
    public static readonly Exception ErrTooManyResults = new InvalidOperationException("too many matching results for request");

    /// <summary>Returned when a stream state encoding is malformed.</summary>
    public static readonly Exception ErrBadStreamStateEncoding = new InvalidDataException("bad stream state encoding");
}

// ---------------------------------------------------------------------------
// StoreMsg — stored message format
// ---------------------------------------------------------------------------

/// <summary>
/// The stored message format for messages retained by the Store layer.
/// </summary>
public sealed class StoreMsg
{
    public string Subject { get; set; } = string.Empty;
    public byte[] Hdr { get; set; } = Array.Empty<byte>();
    public byte[] Msg { get; set; } = Array.Empty<byte>();
    public byte[] Buf { get; set; } = Array.Empty<byte>();
    public ulong Seq { get; set; }
    public long Ts { get; set; }

    /// <summary>Copy this message's fields into <paramref name="dst"/>.</summary>
    public void Copy(StoreMsg dst)
    {
        ArgumentNullException.ThrowIfNull(dst);

        if (Buf.Length > 0)
        {
            var copiedBuffer = new byte[Buf.Length];
            Buffer.BlockCopy(Buf, 0, copiedBuffer, 0, Buf.Length);
            dst.Buf = copiedBuffer;
            var headerLength = Math.Min(Hdr.Length, copiedBuffer.Length);
            dst.Hdr = copiedBuffer[..headerLength];
            dst.Msg = copiedBuffer[headerLength..];
        }
        else
        {
            var headerCopy = new byte[Hdr.Length];
            Buffer.BlockCopy(Hdr, 0, headerCopy, 0, Hdr.Length);

            var messageCopy = new byte[Msg.Length];
            Buffer.BlockCopy(Msg, 0, messageCopy, 0, Msg.Length);

            dst.Hdr = headerCopy;
            dst.Msg = messageCopy;

            var combined = new byte[headerCopy.Length + messageCopy.Length];
            Buffer.BlockCopy(headerCopy, 0, combined, 0, headerCopy.Length);
            Buffer.BlockCopy(messageCopy, 0, combined, headerCopy.Length, messageCopy.Length);
            dst.Buf = combined;
        }

        dst.Subject = Subject;
        dst.Seq = Seq;
        dst.Ts = Ts;
    }

    /// <summary>Copy fields from another StoreMsg into this one.</summary>
    public void CopyFrom(StoreMsg src)
    {
        ArgumentNullException.ThrowIfNull(src);
        src.Copy(this);
    }

    /// <summary>Clear all fields, resetting to an empty message.</summary>
    public void Clear()
    {
        Subject = string.Empty;
        Hdr = Array.Empty<byte>();
        Msg = Array.Empty<byte>();
        Buf = Array.Empty<byte>();
        Seq = 0;
        Ts = 0;
    }
}

// ---------------------------------------------------------------------------
// Delegate types
// ---------------------------------------------------------------------------

/// <summary>
/// Callback into the upper layers to report on changes in storage resources.
/// For single-message cases, also supplies sequence number and subject.
/// </summary>
public delegate void StorageUpdateHandler(long msgs, long bytes, ulong seq, string subj);

/// <summary>Callback into the upper layers to remove a message.</summary>
public delegate void StorageRemoveMsgHandler(ulong seq);

/// <summary>
/// Callback into the upper layers to process a JetStream message.
/// Will propose the message if the stream is replicated.
/// </summary>
public delegate void ProcessJetStreamMsgHandler(object? msg);

// ---------------------------------------------------------------------------
// IStreamStore interface
// ---------------------------------------------------------------------------

/// <summary>Abstraction over JetStream stream storage backends.</summary>
public interface IStreamStore
{
    (ulong Seq, long Ts) StoreMsg(string subject, byte[]? hdr, byte[]? msg, long ttl);
    void StoreRawMsg(string subject, byte[]? hdr, byte[]? msg, ulong seq, long ts, long ttl, bool discardNewCheck);
    (ulong Seq, Exception? Error) SkipMsg(ulong seq);
    void SkipMsgs(ulong seq, ulong num);
    void FlushAllPending();
    StoreMsg? LoadMsg(ulong seq, StoreMsg? sm);
    (StoreMsg? Sm, ulong Skip) LoadNextMsg(string filter, bool wc, ulong start, StoreMsg? smp);
    (StoreMsg? Sm, ulong Skip) LoadNextMsgMulti(object? sl, ulong start, StoreMsg? smp);
    StoreMsg? LoadLastMsg(string subject, StoreMsg? sm);
    (StoreMsg? Sm, Exception? Error) LoadPrevMsg(ulong start, StoreMsg? smp);
    (StoreMsg? Sm, ulong Skip, Exception? Error) LoadPrevMsgMulti(object? sl, ulong start, StoreMsg? smp);
    (bool Removed, Exception? Error) RemoveMsg(ulong seq);
    (bool Removed, Exception? Error) EraseMsg(ulong seq);
    (ulong Purged, Exception? Error) Purge();
    (ulong Purged, Exception? Error) PurgeEx(string subject, ulong seq, ulong keep);
    (ulong Purged, Exception? Error) Compact(ulong seq);
    void Truncate(ulong seq);
    ulong GetSeqFromTime(DateTime t);
    SimpleState FilteredState(ulong seq, string subject);
    Dictionary<string, SimpleState> SubjectsState(string filterSubject);
    Dictionary<string, ulong> SubjectsTotals(string filterSubject);
    (ulong[] Seqs, Exception? Error) AllLastSeqs();
    (ulong[] Seqs, Exception? Error) MultiLastSeqs(string[] filters, ulong maxSeq, int maxAllowed);
    (string Subject, Exception? Error) SubjectForSeq(ulong seq);
    (ulong Total, ulong ValidThrough, Exception? Error) NumPending(ulong sseq, string filter, bool lastPerSubject);
    (ulong Total, ulong ValidThrough, Exception? Error) NumPendingMulti(ulong sseq, object? sl, bool lastPerSubject);
    StreamState State();
    void FastState(StreamState state);
    (byte[] Enc, Exception? Error) EncodedStreamState(ulong failed);
    void SyncDeleted(DeleteBlocks dbs);
    StorageType Type();
    void RegisterStorageUpdates(StorageUpdateHandler cb);
    void RegisterStorageRemoveMsg(StorageRemoveMsgHandler cb);
    void RegisterProcessJetStreamMsg(ProcessJetStreamMsgHandler cb);
    void UpdateConfig(StreamConfig cfg);
    void Delete(bool inline);
    void Stop();
    IConsumerStore ConsumerStore(string name, DateTime created, ConsumerConfig cfg);
    void AddConsumer(IConsumerStore o);
    void RemoveConsumer(IConsumerStore o);
    (SnapshotResult? Result, Exception? Error) Snapshot(TimeSpan deadline, bool includeConsumers, bool checkMsgs);
    (ulong Total, ulong Reported, Exception? Error) Utilization();
    void ResetState();
}

// ---------------------------------------------------------------------------
// RetentionPolicy
// ---------------------------------------------------------------------------

/// <summary>Determines how messages in a stream are retained.</summary>
[JsonConverter(typeof(RetentionPolicyJsonConverter))]
public enum RetentionPolicy
{
    /// <summary>Messages are retained until any given limit is reached.</summary>
    LimitsPolicy = 0,

    /// <summary>Messages are removed when all known consumers have acknowledged them.</summary>
    InterestPolicy,

    /// <summary>Messages are removed when the first worker acknowledges them.</summary>
    WorkQueuePolicy,
}

// ---------------------------------------------------------------------------
// DiscardPolicy
// ---------------------------------------------------------------------------

/// <summary>Determines how the store proceeds when message or byte limits are hit.</summary>
[JsonConverter(typeof(DiscardPolicyJsonConverter))]
public enum DiscardPolicy
{
    /// <summary>Remove older messages to return to the limits.</summary>
    DiscardOld = 0,

    /// <summary>Fail the StoreMsg call when limits are exceeded.</summary>
    DiscardNew,
}

// ---------------------------------------------------------------------------
// StreamState
// ---------------------------------------------------------------------------

/// <summary>Information about the given stream.</summary>
public sealed class StreamState
{
    [JsonPropertyName("messages")]
    public ulong Msgs { get; set; }

    [JsonPropertyName("bytes")]
    public ulong Bytes { get; set; }

    [JsonPropertyName("first_seq")]
    public ulong FirstSeq { get; set; }

    [JsonPropertyName("first_ts")]
    public DateTime FirstTime { get; set; }

    [JsonPropertyName("last_seq")]
    public ulong LastSeq { get; set; }

    [JsonPropertyName("last_ts")]
    public DateTime LastTime { get; set; }

    [JsonPropertyName("num_subjects")]
    public int NumSubjects { get; set; }

    [JsonPropertyName("subjects")]
    public Dictionary<string, ulong>? Subjects { get; set; }

    [JsonPropertyName("num_deleted")]
    public int NumDeleted { get; set; }

    [JsonPropertyName("deleted")]
    public ulong[]? Deleted { get; set; }

    [JsonPropertyName("lost")]
    public LostStreamData? Lost { get; set; }

    [JsonPropertyName("consumer_count")]
    public int Consumers { get; set; }
}

// ---------------------------------------------------------------------------
// SimpleState
// ---------------------------------------------------------------------------

/// <summary>Filtered subject-specific state.</summary>
public sealed class SimpleState
{
    [JsonPropertyName("messages")]
    public ulong Msgs { get; set; }

    [JsonPropertyName("first_seq")]
    public ulong First { get; set; }

    [JsonPropertyName("last_seq")]
    public ulong Last { get; set; }

    // Internal: the first/last need to be recalculated before use.
    internal bool FirstNeedsUpdate { get; set; }
    internal bool LastNeedsUpdate { get; set; }
}

// ---------------------------------------------------------------------------
// LostStreamData
// ---------------------------------------------------------------------------

/// <summary>Indicates messages that have been lost.</summary>
public sealed class LostStreamData
{
    [JsonPropertyName("msgs")]
    public ulong[] Msgs { get; set; } = Array.Empty<ulong>();

    [JsonPropertyName("bytes")]
    public ulong Bytes { get; set; }

    /// <summary>
    /// Returns the index of <paramref name="seq"/> in <see cref="Msgs"/> and whether it was found.
    /// Mirrors Go's <c>LostStreamData.exists</c>.
    /// </summary>
    public (int Index, bool Found) Exists(ulong seq)
    {
        var index = Array.IndexOf(Msgs, seq);
        return (index, index >= 0);
    }
}

// ---------------------------------------------------------------------------
// SnapshotResult
// ---------------------------------------------------------------------------

/// <summary>Contains information about a snapshot operation.</summary>
public sealed class SnapshotResult
{
    public Stream? Reader { get; set; }
    public StreamState State { get; set; } = new();
}

// ---------------------------------------------------------------------------
// DeleteBlock / DeleteBlocks
// ---------------------------------------------------------------------------

/// <summary>
/// Interface for a block of deleted sequences.
/// Implementations include AVL seqsets, run-length encoded ranges, and legacy slices.
/// </summary>
public interface IDeleteBlock
{
    (ulong First, ulong Last, ulong Num) GetState();
    void Range(Func<ulong, bool> f);
}

/// <summary>A list of delete blocks.</summary>
public sealed class DeleteBlocks : List<IDeleteBlock>
{
    public ulong NumDeleted()
    {
        ulong total = 0;
        foreach (var db in this)
        {
            var (_, _, num) = db.GetState();
            total += num;
        }
        return total;
    }
}

// ---------------------------------------------------------------------------
// DeleteRange — run-length encoded delete range
// ---------------------------------------------------------------------------

/// <summary>Run-length encoded delete range.</summary>
public sealed class DeleteRange : IDeleteBlock
{
    public ulong First { get; set; }
    public ulong Num { get; set; }

    public (ulong First, ulong Last, ulong Num) State()
        => GetState();

    public (ulong First, ulong Last, ulong Num) GetState()
    {
        var deletesAfterFirst = Num > 0 ? Num - 1 : 0;
        return (First, First + deletesAfterFirst, Num);
    }

    public void Range(Func<ulong, bool> f)
    {
        for (var seq = First; seq < First + Num; seq++)
        {
            if (!f(seq))
                return;
        }
    }
}

// ---------------------------------------------------------------------------
// DeleteSlice — legacy []uint64 equivalent
// ---------------------------------------------------------------------------

/// <summary>Legacy slice-based delete block.</summary>
public sealed class DeleteSlice : IDeleteBlock
{
    private readonly ulong[] _seqs;

    public DeleteSlice(ulong[] seqs)
    {
        _seqs = seqs;
    }

    public (ulong First, ulong Last, ulong Num) State()
        => GetState();

    public (ulong First, ulong Last, ulong Num) GetState()
    {
        if (_seqs.Length == 0)
            return (0, 0, 0);
        return (_seqs[0], _seqs[^1], (ulong)_seqs.Length);
    }

    public void Range(Func<ulong, bool> f)
    {
        foreach (var seq in _seqs)
        {
            if (!f(seq))
                return;
        }
    }
}

// ---------------------------------------------------------------------------
// StreamReplicatedState
// ---------------------------------------------------------------------------

/// <summary>
/// Represents what is encoded in a binary stream snapshot used for stream
/// replication in an NRG (NATS Raft Group).
/// </summary>
public sealed class StreamReplicatedState
{
    public ulong Msgs { get; set; }
    public ulong Bytes { get; set; }
    public ulong FirstSeq { get; set; }
    public ulong LastSeq { get; set; }
    public ulong Failed { get; set; }
    public DeleteBlocks Deleted { get; set; } = new();
}

// ---------------------------------------------------------------------------
// IConsumerStore interface
// ---------------------------------------------------------------------------

/// <summary>Stores state on consumers for streams.</summary>
public interface IConsumerStore
{
    void SetStarting(ulong sseq);
    void UpdateStarting(ulong sseq);
    void Reset(ulong sseq);
    bool HasState();
    void UpdateDelivered(ulong dseq, ulong sseq, ulong dc, long ts);
    void UpdateAcks(ulong dseq, ulong sseq);
    void UpdateConfig(ConsumerConfig cfg);
    void Update(ConsumerState state);
    (ConsumerState? State, Exception? Error) State();
    (ConsumerState? State, Exception? Error) BorrowState();
    byte[] EncodedState();
    StorageType Type();
    void Stop();
    void Delete();
    void StreamDelete();
}

// ---------------------------------------------------------------------------
// SequencePair
// ---------------------------------------------------------------------------

/// <summary>
/// Has both the consumer and the stream sequence. They point to the same message.
/// </summary>
public sealed class SequencePair
{
    [JsonPropertyName("consumer_seq")]
    public ulong Consumer { get; set; }

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

// ---------------------------------------------------------------------------
// Pending
// ---------------------------------------------------------------------------

/// <summary>
/// Represents a pending message for explicit ack or ack-all.
/// Sequence is the original consumer sequence.
/// </summary>
public sealed class Pending
{
    public ulong Sequence { get; set; }
    public long Timestamp { get; set; }
}

// ---------------------------------------------------------------------------
// ConsumerState
// ---------------------------------------------------------------------------

/// <summary>Represents a stored state for a consumer.</summary>
public sealed class ConsumerState
{
    [JsonPropertyName("delivered")]
    public SequencePair Delivered { get; set; } = new();

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

    /// <summary>
    /// Pending messages and the timestamp for the delivered time.
    /// Only present when AckPolicy is ExplicitAck.
    /// Keys are stream sequence numbers.
    /// </summary>
    [JsonPropertyName("pending")]
    public Dictionary<ulong, Pending>? Pending { get; set; }

    /// <summary>
    /// Messages that have been redelivered (count > 1).
    /// Keys are stream sequence numbers; values are delivery counts minus one.
    /// </summary>
    [JsonPropertyName("redelivered")]
    public Dictionary<ulong, ulong>? Redelivered { get; set; }
}

// ---------------------------------------------------------------------------
// AckPolicy
// ---------------------------------------------------------------------------

/// <summary>Determines how the consumer should acknowledge delivered messages.</summary>
[JsonConverter(typeof(AckPolicyJsonConverter))]
public enum AckPolicy
{
    /// <summary>No acks required for delivered messages.</summary>
    AckNone = 0,

    /// <summary>Acking a sequence implicitly acks all sequences below it.</summary>
    AckAll,

    /// <summary>Requires explicit ack or nack for all messages.</summary>
    AckExplicit,
}

// ---------------------------------------------------------------------------
// ReplayPolicy
// ---------------------------------------------------------------------------

/// <summary>Determines how the consumer replays messages already queued in the stream.</summary>
[JsonConverter(typeof(ReplayPolicyJsonConverter))]
public enum ReplayPolicy
{
    /// <summary>Replay messages as fast as possible.</summary>
    ReplayInstant = 0,

    /// <summary>Maintain the same timing as when the messages were received.</summary>
    ReplayOriginal,
}

// ---------------------------------------------------------------------------
// DeliverPolicy
// ---------------------------------------------------------------------------

/// <summary>Determines how the consumer selects the first message to deliver.</summary>
[JsonConverter(typeof(DeliverPolicyJsonConverter))]
public enum DeliverPolicy
{
    /// <summary>Deliver all messages (default).</summary>
    DeliverAll = 0,

    /// <summary>Start with the last sequence received.</summary>
    DeliverLast,

    /// <summary>Only deliver new messages sent after the consumer is created.</summary>
    DeliverNew,

    /// <summary>Look for a defined starting sequence.</summary>
    DeliverByStartSequence,

    /// <summary>Select the first message with a timestamp >= StartTime.</summary>
    DeliverByStartTime,

    /// <summary>Start with the last message for all subjects received.</summary>
    DeliverLastPerSubject,
}

// ---------------------------------------------------------------------------
// PriorityPolicy
// ---------------------------------------------------------------------------

/// <summary>Policy for selecting messages based on priority.</summary>
[JsonConverter(typeof(PriorityPolicyJsonConverter))]
public enum PriorityPolicy
{
    PriorityNone = 0,
    PriorityOverflow,
    PriorityPinnedClient,
    PriorityPrioritized,
}

// StoreCipher is already defined in ServerOptionTypes.cs — no duplicate needed here.

// ---------------------------------------------------------------------------
// StoreCompression
// ---------------------------------------------------------------------------

/// <summary>Compression algorithm used for stream data.</summary>
public enum StoreCompression : byte
{
    NoCompression = 0,
    S2Compression,
}

// ---------------------------------------------------------------------------
// PersistModeType
// ---------------------------------------------------------------------------

/// <summary>Determines what persistence mode the stream uses.</summary>
[JsonConverter(typeof(PersistModeTypeJsonConverter))]
public enum PersistModeType
{
    DefaultPersistMode = 0,
    AsyncPersistMode,
}

// ---------------------------------------------------------------------------
// Placement
// ---------------------------------------------------------------------------

/// <summary>
/// Guides placement of streams and meta controllers in clustered JetStream.
/// </summary>
public sealed class Placement
{
    [JsonPropertyName("cluster")]
    public string? Cluster { get; set; }

    [JsonPropertyName("tags")]
    public string[]? Tags { get; set; }

    [JsonPropertyName("preferred")]
    public string? Preferred { get; set; }
}

// ---------------------------------------------------------------------------
// SubjectTransformConfig
// ---------------------------------------------------------------------------

/// <summary>
/// Applies a subject transform to incoming messages before storing.
/// </summary>
public sealed class SubjectTransformConfig
{
    [JsonPropertyName("src")]
    public string Source { get; set; } = string.Empty;

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

// ---------------------------------------------------------------------------
// RePublish
// ---------------------------------------------------------------------------

/// <summary>Configuration for republishing messages once committed to a stream.</summary>
public sealed class RePublish
{
    [JsonPropertyName("src")]
    public string? Source { get; set; }

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

    [JsonPropertyName("headers_only")]
    public bool HeadersOnly { get; set; }
}

// ---------------------------------------------------------------------------
// ExternalStream
// ---------------------------------------------------------------------------

/// <summary>Qualifies access to a stream source in another account or domain.</summary>
public sealed class ExternalStream
{
    [JsonPropertyName("api")]
    public string ApiPrefix { get; set; } = string.Empty;

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

    public string Domain()
    {
        if (string.IsNullOrEmpty(ApiPrefix))
            return string.Empty;

        return SubscriptionIndex.TokenAt(ApiPrefix, 2);
    }
}

// ---------------------------------------------------------------------------
// StreamSource
// ---------------------------------------------------------------------------

/// <summary>Dictates how streams can source from other streams.</summary>
public sealed class StreamSource
{
    [JsonPropertyName("name")]
    public string Name { get; set; } = string.Empty;

    [JsonPropertyName("opt_start_seq")]
    public ulong OptStartSeq { get; set; }

    [JsonPropertyName("opt_start_time")]
    public DateTime? OptStartTime { get; set; }

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

    [JsonPropertyName("subject_transforms")]
    public SubjectTransformConfig[]? SubjectTransforms { get; set; }

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

    [JsonIgnore]
    public string IndexName { get; private set; } = string.Empty;

    public string ComposeIName()
    {
        var name = Name;
        if (External != null)
            name = $"{name}:{NatsServer.GetHash(External.ApiPrefix)}";

        var source = FilterSubject ?? string.Empty;
        var destination = ">";

        if (SubjectTransforms is null || SubjectTransforms.Length == 0)
        {
            if (string.IsNullOrEmpty(source))
                source = ">";
            if (string.IsNullOrEmpty(destination))
                destination = ">";
        }
        else
        {
            var sources = new List<string>(SubjectTransforms.Length);
            var destinations = new List<string>(SubjectTransforms.Length);
            foreach (var transform in SubjectTransforms)
            {
                sources.Add(string.IsNullOrEmpty(transform.Source) ? ">" : transform.Source);
                destinations.Add(string.IsNullOrEmpty(transform.Destination) ? ">" : transform.Destination);
            }

            source = string.Join('\f', sources);
            destination = string.Join('\f', destinations);
        }

        return string.Join(' ', [name, source, destination]);
    }

    public void SetIndexName()
    {
        IndexName = ComposeIName();
    }
}

// ---------------------------------------------------------------------------
// StreamConsumerLimits
// ---------------------------------------------------------------------------

/// <summary>Limits applied to consumers created against a stream.</summary>
public sealed class StreamConsumerLimits
{
    [JsonPropertyName("inactive_threshold")]
    public TimeSpan InactiveThreshold { get; set; }

    [JsonPropertyName("max_ack_pending")]
    public int MaxAckPending { get; set; }
}

// ---------------------------------------------------------------------------
// StreamConfig
// ---------------------------------------------------------------------------

/// <summary>
/// Determines the name, subjects, and retention policy for a given stream.
/// If Subjects is empty the Name will be used.
/// </summary>
public sealed class StreamConfig
{
    [JsonPropertyName("name")]
    public string Name { get; set; } = string.Empty;

    [JsonPropertyName("description")]
    public string? Description { get; set; }

    [JsonPropertyName("subjects")]
    public string[]? Subjects { get; set; }

    [JsonPropertyName("retention")]
    public RetentionPolicy Retention { get; set; }

    [JsonPropertyName("max_consumers")]
    public int MaxConsumers { get; set; }

    [JsonPropertyName("max_msgs")]
    public long MaxMsgs { get; set; }

    [JsonPropertyName("max_bytes")]
    public long MaxBytes { get; set; }

    [JsonPropertyName("max_age")]
    public TimeSpan MaxAge { get; set; }

    [JsonPropertyName("max_msgs_per_subject")]
    public long MaxMsgsPer { get; set; }

    [JsonPropertyName("max_msg_size")]
    public int MaxMsgSize { get; set; }

    [JsonPropertyName("discard")]
    public DiscardPolicy Discard { get; set; }

    [JsonPropertyName("storage")]
    public StorageType Storage { get; set; }

    [JsonPropertyName("num_replicas")]
    public int Replicas { get; set; }

    [JsonPropertyName("no_ack")]
    public bool NoAck { get; set; }

    [JsonPropertyName("duplicate_window")]
    public TimeSpan Duplicates { get; set; }

    [JsonPropertyName("placement")]
    public Placement? Placement { get; set; }

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

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

    [JsonPropertyName("compression")]
    public StoreCompression Compression { get; set; }

    [JsonPropertyName("first_seq")]
    public ulong FirstSeq { get; set; }

    [JsonPropertyName("subject_transform")]
    public SubjectTransformConfig? SubjectTransform { get; set; }

    [JsonPropertyName("republish")]
    public RePublish? RePublish { get; set; }

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

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

    [JsonPropertyName("discard_new_per_subject")]
    public bool DiscardNewPer { get; set; }

    [JsonPropertyName("sealed")]
    public bool Sealed { get; set; }

    [JsonPropertyName("deny_delete")]
    public bool DenyDelete { get; set; }

    [JsonPropertyName("deny_purge")]
    public bool DenyPurge { get; set; }

    [JsonPropertyName("allow_rollup_hdrs")]
    public bool AllowRollup { get; set; }

    [JsonPropertyName("consumer_limits")]
    public StreamConsumerLimits ConsumerLimits { get; set; } = new();

    [JsonPropertyName("allow_msg_ttl")]
    public bool AllowMsgTTL { get; set; }

    [JsonPropertyName("subject_delete_marker_ttl")]
    public TimeSpan SubjectDeleteMarkerTTL { get; set; }

    [JsonPropertyName("allow_msg_counter")]
    public bool AllowMsgCounter { get; set; }

    [JsonPropertyName("allow_atomic")]
    public bool AllowAtomicPublish { get; set; }

    [JsonPropertyName("allow_msg_schedules")]
    public bool AllowMsgSchedules { get; set; }

    [JsonPropertyName("persist_mode")]
    public PersistModeType PersistMode { get; set; }

    [JsonPropertyName("metadata")]
    public Dictionary<string, string>? Metadata { get; set; }

    /// <summary>Performs a deep copy of this StreamConfig.</summary>
    public StreamConfig Clone()
    {
        var clone = (StreamConfig)MemberwiseClone();
        if (Placement != null)
            clone.Placement = new Placement { Cluster = Placement.Cluster, Tags = Placement.Tags?.ToArray(), Preferred = Placement.Preferred };
        if (Mirror != null)
            clone.Mirror = new StreamSource { Name = Mirror.Name, OptStartSeq = Mirror.OptStartSeq, OptStartTime = Mirror.OptStartTime, FilterSubject = Mirror.FilterSubject, SubjectTransforms = Mirror.SubjectTransforms?.ToArray(), External = Mirror.External };
        if (Sources != null)
            clone.Sources = Sources.ToArray();
        if (SubjectTransform != null)
            clone.SubjectTransform = new SubjectTransformConfig { Source = SubjectTransform.Source, Destination = SubjectTransform.Destination };
        if (RePublish != null)
            clone.RePublish = new RePublish { Source = RePublish.Source, Destination = RePublish.Destination, HeadersOnly = RePublish.HeadersOnly };
        if (Metadata != null)
            clone.Metadata = new Dictionary<string, string>(Metadata);
        if (Subjects != null)
            clone.Subjects = Subjects.ToArray();
        return clone;
    }
}

// ---------------------------------------------------------------------------
// ConsumerConfig
// ---------------------------------------------------------------------------

/// <summary>Configuration for a JetStream consumer.</summary>
public sealed class ConsumerConfig
{
    [JsonPropertyName("durable_name")]
    public string? Durable { get; set; }

    [JsonPropertyName("name")]
    public string? Name { get; set; }

    [JsonPropertyName("description")]
    public string? Description { get; set; }

    [JsonPropertyName("deliver_policy")]
    public DeliverPolicy DeliverPolicy { get; set; }

    [JsonPropertyName("opt_start_seq")]
    public ulong OptStartSeq { get; set; }

    [JsonPropertyName("opt_start_time")]
    public DateTime? OptStartTime { get; set; }

    [JsonPropertyName("ack_policy")]
    public AckPolicy AckPolicy { get; set; }

    [JsonPropertyName("ack_wait")]
    public TimeSpan AckWait { get; set; }

    [JsonPropertyName("max_deliver")]
    public int MaxDeliver { get; set; }

    [JsonPropertyName("backoff")]
    public TimeSpan[]? BackOff { get; set; }

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

    [JsonPropertyName("filter_subjects")]
    public string[]? FilterSubjects { get; set; }

    [JsonPropertyName("replay_policy")]
    public ReplayPolicy ReplayPolicy { get; set; }

    [JsonPropertyName("rate_limit_bps")]
    public ulong RateLimit { get; set; }

    [JsonPropertyName("sample_freq")]
    public string? SampleFrequency { get; set; }

    [JsonPropertyName("max_waiting")]
    public int MaxWaiting { get; set; }

    [JsonPropertyName("max_ack_pending")]
    public int MaxAckPending { get; set; }

    [JsonPropertyName("flow_control")]
    public bool FlowControl { get; set; }

    [JsonPropertyName("headers_only")]
    public bool HeadersOnly { get; set; }

    // Pull-based options
    [JsonPropertyName("max_batch")]
    public int MaxRequestBatch { get; set; }

    [JsonPropertyName("max_expires")]
    public TimeSpan MaxRequestExpires { get; set; }

    [JsonPropertyName("max_bytes")]
    public int MaxRequestMaxBytes { get; set; }

    // Push-based consumers
    [JsonPropertyName("deliver_subject")]
    public string? DeliverSubject { get; set; }

    [JsonPropertyName("deliver_group")]
    public string? DeliverGroup { get; set; }

    [JsonPropertyName("idle_heartbeat")]
    public TimeSpan Heartbeat { get; set; }

    [JsonPropertyName("inactive_threshold")]
    public TimeSpan InactiveThreshold { get; set; }

    [JsonPropertyName("num_replicas")]
    public int Replicas { get; set; }

    [JsonPropertyName("mem_storage")]
    public bool MemoryStorage { get; set; }

    [JsonPropertyName("direct")]
    public bool Direct { get; set; }

    [JsonPropertyName("metadata")]
    public Dictionary<string, string>? Metadata { get; set; }

    [JsonPropertyName("pause_until")]
    public DateTime? PauseUntil { get; set; }

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

    [JsonPropertyName("priority_policy")]
    public PriorityPolicy PriorityPolicy { get; set; }

    [JsonPropertyName("priority_timeout")]
    public TimeSpan PinnedTTL { get; set; }
}

// ---------------------------------------------------------------------------
// DiskAvailability — stub for disk_avail.go
// ---------------------------------------------------------------------------

/// <summary>
/// Utility for checking disk space availability.
/// Adapted from server/disk_avail.go (uses statfs on Unix).
/// </summary>
public static class DiskAvailability
{
    // Default large store limit: 1 TB (used as fallback when statfs fails).
    private const long JetStreamMaxStoreDefault = 1L * 1024 * 1024 * 1024 * 1024;

    /// <summary>
    /// Returns approximately 75% of available disk space at <paramref name="storeDir"/>.
    /// Ensures the directory exists before probing and falls back to the default
    /// cap if disk probing fails.
    /// </summary>
    public static long DiskAvailable(string storeDir)
    {
        try
        {
            if (!string.IsNullOrWhiteSpace(storeDir))
                Directory.CreateDirectory(storeDir);

            var root = Path.GetPathRoot(Path.GetFullPath(storeDir));
            var drive = new DriveInfo(root ?? storeDir);
            if (drive.IsReady)
                return drive.AvailableFreeSpace / 4 * 3;
        }
        catch
        {
            // Fall through to default.
        }

        return JetStreamMaxStoreDefault;
    }

    /// <summary>
    /// Returns approximately 75% of available disk space at <paramref name="path"/>.
    /// Returns <see cref="JetStreamMaxStoreDefault"/> (1 TB) if the check fails.
    /// </summary>
    public static long Available(string path) => DiskAvailable(path);

    /// <summary>
    /// Returns true if at least <paramref name="needed"/> bytes are available at <paramref name="path"/>.
    /// </summary>
    public static bool Check(string path, long needed) => DiskAvailable(path) >= needed;
}
