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

namespace ZB.MOM.NatsNet.Server.Mqtt;

// ============================================================================
// Enumerations
// ============================================================================

/// <summary>
/// State machine states for parsing incoming MQTT byte streams.
/// Mirrors the <c>mqttParseState*</c> iota in server/mqtt.go (implicit from
/// the read-loop logic).
/// </summary>
internal enum MqttParseState : byte
{
    /// <summary>Waiting for the first fixed-header byte.</summary>
    MqttStateHeader = 0,

    /// <summary>Reading the remaining-length variable-integer bytes.</summary>
    MqttStateFixedHeader,

    /// <summary>Reading the variable-header + payload bytes of the current packet.</summary>
    MqttStateControlPacket,
}

// ============================================================================
// Will
// ============================================================================

/// <summary>
/// MQTT Will message parameters extracted from a CONNECT packet.
/// Mirrors Go <c>mqttWill</c> struct in server/mqtt.go.
/// </summary>
internal sealed class MqttWill
{
    /// <summary>NATS subject derived from the MQTT will topic.</summary>
    public string Subject { get; set; } = string.Empty;

    /// <summary>Original MQTT will topic string.</summary>
    public string Topic { get; set; } = string.Empty;

    /// <summary>Will message payload bytes, or <c>null</c> if empty.</summary>
    public byte[]? Msg { get; set; }

    /// <summary>QoS level for the will message (0, 1, or 2).</summary>
    public byte Qos { get; set; }

    /// <summary>Whether the will message should be retained.</summary>
    public bool Retain { get; set; }
}

// ============================================================================
// Connect protocol
// ============================================================================

/// <summary>
/// MQTT CONNECT packet parsed payload.
/// Mirrors Go <c>mqttConnectProto</c> struct in server/mqtt.go (extended with
/// the fields surfaced by the parse helpers).
/// </summary>
internal sealed class MqttConnectProto
{
    /// <summary>MQTT client identifier.</summary>
    public string ClientId { get; set; } = string.Empty;

    /// <summary>Raw CONNECT packet bytes (for forwarding / replay).</summary>
    public byte[] Connect { get; set; } = [];

    /// <summary>Parsed Will parameters, or <c>null</c> if the Will flag is not set.</summary>
    public MqttWill? Will { get; set; }

    /// <summary>Username presented in the CONNECT packet.</summary>
    public string Username { get; set; } = string.Empty;

    /// <summary>Password bytes presented in the CONNECT packet, or <c>null</c> if absent.</summary>
    public byte[]? Password { get; set; }

    /// <summary>Whether the Clean Session flag was set.</summary>
    public bool CleanSession { get; set; }

    /// <summary>Keep-alive interval in seconds (0 = disabled).</summary>
    public ushort KeepAlive { get; set; }
}

// ============================================================================
// Subscription
// ============================================================================

/// <summary>
/// A single MQTT topic filter subscription entry stored in a session.
/// Mirrors the per-entry semantics of <c>mqttSession.subs</c> map in server/mqtt.go.
/// </summary>
internal sealed class MqttSubscription
{
    /// <summary>NATS subject derived from the MQTT topic filter.</summary>
    public string Subject { get; set; } = string.Empty;

    /// <summary>Maximum QoS level granted for this subscription.</summary>
    public byte Qos { get; set; }
}

// ============================================================================
// Publish info
// ============================================================================

/// <summary>
/// Parsed metadata for an inbound MQTT PUBLISH packet.
/// Mirrors Go <c>mqttPublish</c> struct in server/mqtt.go.
/// </summary>
internal sealed class MqttPublishInfo
{
    /// <summary>NATS subject derived from the MQTT topic.</summary>
    public string Subject { get; set; } = string.Empty;

    /// <summary>Original MQTT topic string.</summary>
    public string Topic { get; set; } = string.Empty;

    /// <summary>Message payload bytes, or <c>null</c> if empty.</summary>
    public byte[]? Msg { get; set; }

    /// <summary>QoS level of the PUBLISH packet.</summary>
    public byte Qos { get; set; }

    /// <summary>Whether the Retain flag is set.</summary>
    public bool Retain { get; set; }

    /// <summary>Whether the DUP flag is set (re-delivery of a QoS &gt;0 packet).</summary>
    public bool Dup { get; set; }

    /// <summary>Packet identifier (only meaningful for QoS 1 and 2).</summary>
    public ushort Pi { get; set; }
}

// ============================================================================
// Pending ack
// ============================================================================

/// <summary>
/// Tracks a single in-flight QoS 1 or QoS 2 message pending acknowledgement.
/// Mirrors Go <c>mqttPending</c> struct in server/mqtt.go.
/// </summary>
internal sealed class MqttPending
{
    /// <summary>JetStream stream sequence number for this message.</summary>
    public ulong SSeq { get; set; }

    /// <summary>JetStream ACK subject to send the acknowledgement to.</summary>
    public string JsAckSubject { get; set; } = string.Empty;

    /// <summary>JetStream durable consumer name.</summary>
    public string JsDur { get; set; } = string.Empty;
}

// ============================================================================
// Retained message
// ============================================================================

/// <summary>
/// A retained MQTT message stored in JetStream.
/// Mirrors Go <c>mqttRetainedMsg</c> struct in server/mqtt.go.
/// </summary>
internal sealed class MqttRetainedMsg
{
    /// <summary>Origin server name.</summary>
    public string Origin { get; set; } = string.Empty;

    /// <summary>NATS subject for this retained message.</summary>
    public string Subject { get; set; } = string.Empty;

    /// <summary>Original MQTT topic.</summary>
    public string Topic { get; set; } = string.Empty;

    /// <summary>Message payload bytes.</summary>
    public byte[]? Msg { get; set; }

    /// <summary>Message flags byte.</summary>
    public byte Flags { get; set; }

    /// <summary>Source identifier.</summary>
    public string Source { get; set; } = string.Empty;

    /// <summary>
    /// Cache expiry time for this retained message in the in-memory RmsCache.
    /// Not persisted to JetStream.
    /// </summary>
    public DateTime ExpiresFromCache { get; set; }
}

// ============================================================================
// Persisted session
// ============================================================================

/// <summary>
/// The JSON-serialisable representation of an MQTT session stored in JetStream.
/// Mirrors Go <c>mqttPersistedSession</c> struct in server/mqtt.go.
/// </summary>
internal sealed class MqttPersistedSession
{
    /// <summary>Server that originally created this session.</summary>
    public string Origin { get; set; } = string.Empty;

    /// <summary>MQTT client identifier.</summary>
    public string Id { get; set; } = string.Empty;

    /// <summary>Whether this was a clean session.</summary>
    public bool Clean { get; set; }

    /// <summary>Map of MQTT topic filters to granted QoS levels.</summary>
    public Dictionary<string, byte> Subs { get; set; } = new();
}

// ============================================================================
// Session
// ============================================================================

/// <summary>
/// In-memory MQTT session state.
/// Mirrors Go <c>mqttSession</c> struct in server/mqtt.go.
/// </summary>
internal sealed class MqttSession
{
    private readonly Lock _mu = new();

    /// <summary>Lock for this session (matches Go <c>sess.mu</c>).</summary>
    public Lock Mu => _mu;

    // ------------------------------------------------------------------
    // Identity
    // ------------------------------------------------------------------

    /// <summary>MQTT client identifier.</summary>
    public string Id { get; set; } = string.Empty;

    /// <summary>Hash of the client identifier (used as JetStream key).</summary>
    public string IdHash { get; set; } = string.Empty;

    /// <summary>Whether this is a clean session.</summary>
    public bool Clean { get; set; }

    /// <summary>Domain token (domain with trailing '.', or empty).</summary>
    public string DomainTk { get; set; } = string.Empty;

    // ------------------------------------------------------------------
    // Client link
    // ------------------------------------------------------------------

    /// <summary>
    /// The <see cref="ClientConnection"/> that currently owns this session.
    /// Protected by <see cref="Mu"/>. Set to <c>null</c> when the client disconnects.
    /// Mirrors Go <c>mqttSession.c *client</c>.
    /// </summary>
    public ClientConnection? Client { get; set; }

    // ------------------------------------------------------------------
    // JetStream sequence
    // ------------------------------------------------------------------

    /// <summary>
    /// JetStream stream sequence number of the last persisted session record.
    /// Protected by <see cref="Mu"/>.
    /// Mirrors Go <c>mqttSession.seq uint64</c>.
    /// </summary>
    public ulong Seq { get; set; }

    // ------------------------------------------------------------------
    // Subscriptions
    // ------------------------------------------------------------------

    /// <summary>
    /// Map from MQTT SUBSCRIBE filter to granted QoS level.
    /// Mirrors Go <c>mqttSession.subs map[string]byte</c>.
    /// </summary>
    public Dictionary<string, byte> Subs { get; } = new();

    // ------------------------------------------------------------------
    // Pending acks
    // ------------------------------------------------------------------

    /// <summary>Maximum number of in-flight QoS-1/2 PUBLISH acks.</summary>
    public ushort MaxPending { get; set; }

    /// <summary>
    /// In-flight QoS-1 PUBLISH packets pending PUBACK from the client.
    /// Key is the packet identifier.
    /// </summary>
    public Dictionary<ushort, MqttPending> PendingPublish { get; } = new();

    /// <summary>
    /// In-flight QoS-2 PUBREL packets pending PUBCOMP from the client.
    /// Key is the packet identifier.
    /// </summary>
    public Dictionary<ushort, MqttPending> PendingPubRel { get; } = new();

    /// <summary>"Last used" packet identifier; used as the starting point when allocating the next one.</summary>
    public ushort LastPi { get; set; }

    // ------------------------------------------------------------------
    // Constructor
    // ------------------------------------------------------------------

    /// <summary>Initialises a new session with the given identity.</summary>
    public MqttSession(string id, string idHash, bool clean)
    {
        Id     = id;
        IdHash = idHash;
        Clean  = clean;
    }
}

// ============================================================================
// Global session manager
// ============================================================================

/// <summary>
/// Server-wide MQTT session manager.
/// Mirrors Go <c>mqttSessionManager</c> struct in server/mqtt.go.
/// </summary>
internal sealed class MqttSessionManager
{
    private readonly ReaderWriterLockSlim _mu = new(LockRecursionPolicy.NoRecursion);

    /// <summary>Per-account session managers keyed by account name.</summary>
    public Dictionary<string, MqttAccountSessionManager> Sessions { get; } = new();

    /// <summary>Read lock for this manager.</summary>
    public ReaderWriterLockSlim Mu => _mu;
}
