// Copyright 2020-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/jetstream.go in the NATS server Go source.

using System.Text.Json.Serialization;

namespace ZB.MOM.NatsNet.Server;

// ---------------------------------------------------------------------------
// JetStreamConfig
// ---------------------------------------------------------------------------

/// <summary>
/// Configuration for the JetStream subsystem.
/// Mirrors <c>JetStreamConfig</c> in server/jetstream.go.
/// </summary>
public sealed class JetStreamConfig
{
    /// <summary>Maximum size of memory-backed streams.</summary>
    [JsonPropertyName("max_memory")]
    public long MaxMemory { get; set; }

    /// <summary>Maximum size of file-backed streams.</summary>
    [JsonPropertyName("max_storage")]
    public long MaxStore { get; set; }

    /// <summary>Directory where storage files are stored.</summary>
    [JsonPropertyName("store_dir")]
    public string StoreDir { get; set; } = string.Empty;

    /// <summary>How frequently we sync to disk in the background via fsync.</summary>
    [JsonPropertyName("sync_interval")]
    public TimeSpan SyncInterval { get; set; }

    /// <summary>If true, flushes are done after every write.</summary>
    [JsonPropertyName("sync_always")]
    public bool SyncAlways { get; set; }

    /// <summary>The JetStream domain for this server.</summary>
    [JsonPropertyName("domain")]
    public string Domain { get; set; } = string.Empty;

    /// <summary>Whether compression is supported.</summary>
    [JsonPropertyName("compress_ok")]
    public bool CompressOK { get; set; }

    /// <summary>Unique tag assigned to this server instance.</summary>
    [JsonPropertyName("unique_tag")]
    public string UniqueTag { get; set; } = string.Empty;

    /// <summary>Whether strict JSON parsing is performed.</summary>
    [JsonPropertyName("strict")]
    public bool Strict { get; set; }
}

// ---------------------------------------------------------------------------
// JetStreamApiStats
// ---------------------------------------------------------------------------

/// <summary>
/// Statistics about the JetStream API usage for this server.
/// Mirrors <c>JetStreamAPIStats</c> in server/jetstream.go.
/// </summary>
public sealed class JetStreamApiStats
{
    /// <summary>Active API level implemented by this server.</summary>
    [JsonPropertyName("level")]
    public int Level { get; set; }

    /// <summary>Total API requests received since server start.</summary>
    [JsonPropertyName("total")]
    public ulong Total { get; set; }

    /// <summary>Total API requests that resulted in error responses.</summary>
    [JsonPropertyName("errors")]
    public ulong Errors { get; set; }

    /// <summary>Number of API requests currently being served.</summary>
    [JsonPropertyName("inflight")]
    public ulong Inflight { get; set; }
}

// ---------------------------------------------------------------------------
// JetStreamStats
// ---------------------------------------------------------------------------

/// <summary>
/// Statistics about JetStream for this server.
/// Mirrors <c>JetStreamStats</c> in server/jetstream.go.
/// </summary>
public sealed class JetStreamStats
{
    [JsonPropertyName("memory")]           public ulong Memory         { get; set; }
    [JsonPropertyName("storage")]          public ulong Store          { get; set; }
    [JsonPropertyName("reserved_memory")]  public ulong ReservedMemory { get; set; }
    [JsonPropertyName("reserved_storage")] public ulong ReservedStore  { get; set; }
    [JsonPropertyName("accounts")]         public int   Accounts       { get; set; }
    [JsonPropertyName("ha_assets")]        public int   HAAssets       { get; set; }
    [JsonPropertyName("api")]             public JetStreamApiStats Api { get; set; } = new();
}

// ---------------------------------------------------------------------------
// JetStreamAccountLimits
// ---------------------------------------------------------------------------

/// <summary>
/// Per-account JetStream limits.
/// Mirrors <c>JetStreamAccountLimits</c> in server/jetstream.go.
/// </summary>
public sealed class JetStreamAccountLimits
{
    [JsonPropertyName("max_memory")]               public long MaxMemory            { get; set; }
    [JsonPropertyName("max_storage")]              public long MaxStore             { get; set; }
    [JsonPropertyName("max_streams")]              public int  MaxStreams           { get; set; }
    [JsonPropertyName("max_consumers")]            public int  MaxConsumers         { get; set; }
    [JsonPropertyName("max_ack_pending")]          public int  MaxAckPending        { get; set; }
    [JsonPropertyName("memory_max_stream_bytes")]  public long MemoryMaxStreamBytes { get; set; }
    [JsonPropertyName("storage_max_stream_bytes")] public long StoreMaxStreamBytes  { get; set; }
    [JsonPropertyName("max_bytes_required")]       public bool MaxBytesRequired     { get; set; }
}

// ---------------------------------------------------------------------------
// JetStreamTier
// ---------------------------------------------------------------------------

/// <summary>
/// Per-tier JetStream usage and limits.
/// Mirrors <c>JetStreamTier</c> in server/jetstream.go.
/// </summary>
public sealed class JetStreamTier
{
    [JsonPropertyName("memory")]          public ulong                  Memory         { get; set; }
    [JsonPropertyName("storage")]         public ulong                  Store          { get; set; }
    [JsonPropertyName("reserved_memory")] public ulong                  ReservedMemory { get; set; }
    [JsonPropertyName("reserved_storage")]public ulong                  ReservedStore  { get; set; }
    [JsonPropertyName("streams")]         public int                    Streams        { get; set; }
    [JsonPropertyName("consumers")]       public int                    Consumers      { get; set; }
    [JsonPropertyName("limits")]          public JetStreamAccountLimits Limits         { get; set; } = new();
}

// ---------------------------------------------------------------------------
// JetStreamAccountStats
// ---------------------------------------------------------------------------

/// <summary>
/// Current statistics about an account's JetStream usage.
/// Mirrors <c>JetStreamAccountStats</c> in server/jetstream.go.
/// Embeds <see cref="JetStreamTier"/> for totals.
/// </summary>
public sealed class JetStreamAccountStats
{
    // Embedded JetStreamTier fields (Go struct embedding)
    [JsonPropertyName("memory")]          public ulong                  Memory         { get; set; }
    [JsonPropertyName("storage")]         public ulong                  Store          { get; set; }
    [JsonPropertyName("reserved_memory")] public ulong                  ReservedMemory { get; set; }
    [JsonPropertyName("reserved_storage")]public ulong                  ReservedStore  { get; set; }
    [JsonPropertyName("streams")]         public int                    Streams        { get; set; }
    [JsonPropertyName("consumers")]       public int                    Consumers      { get; set; }
    [JsonPropertyName("limits")]          public JetStreamAccountLimits Limits         { get; set; } = new();

    [JsonPropertyName("domain")] public string?                           Domain { get; set; }
    [JsonPropertyName("api")]    public JetStreamApiStats                  Api    { get; set; } = new();
    [JsonPropertyName("tiers")]  public Dictionary<string, JetStreamTier>? Tiers  { get; set; }
}

// ---------------------------------------------------------------------------
// Internal JetStream engine types
// ---------------------------------------------------------------------------

/// <summary>
/// The main JetStream engine, one per server.
/// Mirrors <c>jetStream</c> struct in server/jetstream.go.
/// </summary>
internal sealed partial class JetStream
{
    // Atomic counters (use Interlocked for thread-safety)
    public long ApiInflight;
    public long ApiTotal;
    public long ApiErrors;
    public long MemReserved;
    public long StoreReserved;
    public long MemUsed;
    public long StoreUsed;
    public long QueueLimit;
    public int  Clustered;   // atomic int32

    private readonly ReaderWriterLockSlim _mu = new();

    public object?           Server   { get; set; }  // *Server — set at runtime
    public JetStreamConfig   Config   { get; set; } = new();
    public object?           Cluster  { get; set; }  // *jetStreamCluster — session 20+
    public Dictionary<string, JsAccount> Accounts { get; } = new(StringComparer.Ordinal);
    public DateTime          Started  { get; set; }

    // State booleans
    public bool MetaRecovering { get; set; }
    public bool StandAlone     { get; set; }
    public bool Oos            { get; set; }
    public bool ShuttingDown   { get; set; }
    public int  Disabled;      // atomic bool (0=false, 1=true)

    public ReaderWriterLockSlim Lock => _mu;
}

/// <summary>
/// Tracks remote per-tier usage for a JetStream account.
/// Mirrors <c>remoteUsage</c> in server/jetstream.go.
/// </summary>
internal sealed class RemoteUsage
{
    public Dictionary<string, JsaUsage> Tiers { get; } = new(StringComparer.Ordinal);
    public ulong Api;
    public ulong Err;
}

/// <summary>
/// Per-tier storage accounting (total + local split).
/// Mirrors <c>jsaStorage</c> in server/jetstream.go.
/// </summary>
internal sealed class JsaStorage
{
    public JsaUsage Total { get; set; } = new();
    public JsaUsage Local { get; set; } = new();
}

/// <summary>
/// A JetStream-enabled account, holding streams, limits and usage tracking.
/// Mirrors <c>jsAccount</c> in server/jetstream.go.
/// </summary>
internal sealed partial class JsAccount
{
    private readonly ReaderWriterLockSlim _mu = new();

    public object? Js      { get; set; }  // *jetStream
    public object? Account { get; set; }  // *Account
    public string  StoreDir { get; set; } = string.Empty;

    // Concurrent inflight map (mirrors sync.Map)
    public readonly System.Collections.Concurrent.ConcurrentDictionary<string, object?> Inflight = new();

    // Streams keyed by stream name
    public Dictionary<string, object?> Streams { get; } = new(StringComparer.Ordinal);  // *stream

    // Send queue (mirrors *ipQueue[*pubMsg])
    public object? SendQ { get; set; }

    // Atomic sync flag (0=false, 1=true)
    public int Sync;

    // Usage/limits (protected by UsageMu)
    private readonly ReaderWriterLockSlim _usageMu = new();
    public Dictionary<string, JetStreamAccountLimits> Limits { get; } = new(StringComparer.Ordinal);
    public Dictionary<string, JsaStorage>             Usage  { get; } = new(StringComparer.Ordinal);
    public Dictionary<string, RemoteUsage>            RUsage { get; } = new(StringComparer.Ordinal);

    public ulong   ApiTotal   { get; set; }
    public ulong   ApiErrors  { get; set; }
    public ulong   UsageApi   { get; set; }
    public ulong   UsageErr   { get; set; }
    public string  UpdatesPub { get; set; } = string.Empty;
    public object? UpdatesSub { get; set; }  // *subscription
    public DateTime LUpdate   { get; set; }

    public ReaderWriterLockSlim Lock      => _mu;
    public ReaderWriterLockSlim UsageLock => _usageMu;
}

/// <summary>
/// Memory and store byte usage.
/// Mirrors <c>jsaUsage</c> in server/jetstream.go.
/// </summary>
internal sealed class JsaUsage
{
    public long Mem   { get; set; }
    public long Store { get; set; }
}

/// <summary>
/// Delegate for a function that generates a key-encryption key from a context byte array.
/// Mirrors <c>keyGen</c> in server/jetstream.go.
/// </summary>
public delegate byte[] KeyGen(byte[] context);

// ---------------------------------------------------------------------------
// JetStream message header helpers
// ---------------------------------------------------------------------------

/// <summary>
/// Static helpers for extracting TTL, scheduling, and scheduler information
/// from JetStream message headers.
/// Mirrors <c>getMessageTTL</c>, <c>nextMessageSchedule</c>, <c>getMessageScheduler</c>
/// in server/stream.go.
/// </summary>
public static class JetStreamHeaderHelpers
{
    /// <summary>
    /// Extracts the TTL value (in seconds) from the message header.
    /// Returns 0 if no TTL header is present. Returns -1 for "never".
    /// Mirrors Go <c>getMessageTTL</c>.
    /// </summary>
    public static (long Ttl, Exception? Error) GetMessageTtl(byte[] hdr)
    {
        var raw = NatsMessageHeaders.GetHeader(NatsHeaderConstants.JsMessageTtl, hdr);
        if (raw == null || raw.Length == 0)
            return (0, null);

        return ParseMessageTtl(System.Text.Encoding.ASCII.GetString(raw));
    }

    /// <summary>
    /// Parses a TTL string value into seconds.
    /// Supports "never" (-1), Go-style duration strings ("30s", "5m"), or plain integer seconds.
    /// Mirrors Go <c>parseMessageTTL</c>.
    /// </summary>
    public static (long Ttl, Exception? Error) ParseMessageTtl(string ttl)
    {
        if (string.Equals(ttl, "never", StringComparison.OrdinalIgnoreCase))
            return (-1, null);

        // Try parsing as a Go-style duration.
        if (TryParseDuration(ttl, out var dur))
        {
            if (dur.TotalSeconds < 1)
                return (0, new InvalidOperationException("message TTL invalid"));
            return ((long)dur.TotalSeconds, null);
        }

        // Try as plain integer (seconds).
        if (long.TryParse(ttl, out var t))
        {
            if (t < 0)
                return (0, new InvalidOperationException("message TTL invalid"));
            return (t, null);
        }

        return (0, new InvalidOperationException("message TTL invalid"));
    }

    /// <summary>
    /// Extracts the next scheduled fire time from the message header.
    /// Returns (DateTime, true) if valid, (default, true) if no header, (default, false) on parse error.
    /// Mirrors Go <c>nextMessageSchedule</c>.
    /// </summary>
    public static (DateTime Schedule, bool Ok) NextMessageSchedule(byte[] hdr, long ts)
    {
        if (hdr.Length == 0)
            return (default, true);

        var slice = NatsMessageHeaders.SliceHeader(NatsHeaderConstants.JsSchedulePattern, hdr);
        if (slice == null || slice.Value.Length == 0)
            return (default, true);

        var val = System.Text.Encoding.ASCII.GetString(slice.Value.Span);
        var (schedule, _, ok) = Internal.MsgScheduling.ParseMsgSchedule(val, ts);
        return (schedule, ok);
    }

    /// <summary>
    /// Extracts the scheduler identifier from the message header.
    /// Returns empty string if not present.
    /// Mirrors Go <c>getMessageScheduler</c>.
    /// </summary>
    public static string GetMessageScheduler(byte[] hdr)
    {
        if (hdr.Length == 0)
            return string.Empty;

        var raw = NatsMessageHeaders.GetHeader(NatsHeaderConstants.JsScheduler, hdr);
        if (raw == null || raw.Length == 0)
            return string.Empty;

        return System.Text.Encoding.ASCII.GetString(raw);
    }

    private static bool TryParseDuration(string s, out TimeSpan result)
    {
        result = default;
        if (s.EndsWith("ms", StringComparison.Ordinal) && double.TryParse(s[..^2], out var ms))
        { result = TimeSpan.FromMilliseconds(ms); return true; }
        if (s.EndsWith('s') && double.TryParse(s[..^1], out var sec))
        { result = TimeSpan.FromSeconds(sec); return true; }
        if (s.EndsWith('m') && double.TryParse(s[..^1], out var min))
        { result = TimeSpan.FromMinutes(min); return true; }
        if (s.EndsWith('h') && double.TryParse(s[..^1], out var hr))
        { result = TimeSpan.FromHours(hr); return true; }
        return TimeSpan.TryParse(s, out result);
    }
}
