// 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;

// ============================================================================
// Per-client MQTT state
// ============================================================================

/// <summary>
/// Per-client MQTT state attached to every connection established via the MQTT
/// listener or WebSocket upgrade.
/// Mirrors Go <c>mqtt</c> struct in server/mqtt.go.
/// </summary>
internal sealed class MqttHandler
{
    private readonly Lock _mu = new();

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

    /// <summary>MQTT client identifier presented in the CONNECT packet.</summary>
    public string ClientId { get; set; } = string.Empty;

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

    // ------------------------------------------------------------------
    // Session / Will
    // ------------------------------------------------------------------

    /// <summary>Session associated with this connection after a successful CONNECT.</summary>
    public MqttSession? Session { get; set; }

    /// <summary>
    /// Quick reference to the account session manager.
    /// Immutable after <c>processConnect()</c> completes.
    /// </summary>
    public MqttAccountSessionManager? AccountSessionManager { get; set; }

    /// <summary>Will message to publish when this connection closes unexpectedly.</summary>
    public MqttWill? Will { get; set; }

    // ------------------------------------------------------------------
    // Keep-alive
    // ------------------------------------------------------------------

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

    // ------------------------------------------------------------------
    // QoS pending / packet identifiers
    // ------------------------------------------------------------------

    /// <summary>Next packet identifier to use for QoS &gt;0 outbound messages.</summary>
    public ushort NextPi { get; set; }

    /// <summary>
    /// Pending ack map: packet identifier → pending state.
    /// Used for tracking in-flight QoS 1/2 PUBLISH packets.
    /// </summary>
    public Dictionary<ushort, MqttPending?> Pending { get; } = new();

    // ------------------------------------------------------------------
    // Protocol flags
    // ------------------------------------------------------------------

    /// <summary>
    /// When <c>true</c>, the server rejects QoS-2 PUBLISH from this client
    /// and terminates the connection on receipt of such a packet.
    /// Mirrors Go <c>mqtt.rejectQoS2Pub</c>.
    /// </summary>
    public bool RejectQoS2Pub { get; set; }

    /// <summary>
    /// When <c>true</c>, QoS-2 SUBSCRIBE requests are silently downgraded to QoS-1.
    /// Mirrors Go <c>mqtt.downgradeQoS2Sub</c>.
    /// </summary>
    public bool DowngradeQoS2Sub { get; set; }

    // ------------------------------------------------------------------
    // QoS 2 in-memory pending store (full JetStream persistence in Task 6)
    // ------------------------------------------------------------------

    /// <summary>
    /// In-memory store for QoS 2 inbound messages awaiting PUBREL.
    /// Keyed by packet identifier. Replaces JetStream $MQTT_qos2in stream.
    /// </summary>
    public Dictionary<ushort, MqttPublishInfo> QoS2Pending { get; } = new();

    // ------------------------------------------------------------------
    // Parse state (used by the read-loop MQTT byte-stream parser)
    // ------------------------------------------------------------------

    /// <summary>Current state of the fixed-header / remaining-length state machine.</summary>
    public byte ParseState { get; set; }

    /// <summary>Control packet type byte extracted from the current fixed header.</summary>
    public byte PktType { get; set; }

    /// <summary>Remaining length of the current control packet (bytes still to read).</summary>
    public int RemLen { get; set; }

    /// <summary>Buffer accumulating the current packet's variable-header and payload.</summary>
    public byte[]? Buf { get; set; }

    /// <summary>Multiplier accumulator used during multi-byte remaining-length decoding.</summary>
    public int RemLenMult { get; set; }

    // ------------------------------------------------------------------
    // Reader
    // ------------------------------------------------------------------

    /// <summary>
    /// Per-connection MQTT reader for binary packet parsing.
    /// Created lazily on first use.
    /// </summary>
    public MqttReader Reader { get; } = new();

    // ------------------------------------------------------------------
    // Thread safety
    // ------------------------------------------------------------------

    /// <summary>Lock protecting mutable fields on this instance.</summary>
    public Lock Mu => _mu;
}

// ============================================================================
// Server-side MQTT extension methods
// ============================================================================

/// <summary>
/// Extension methods on <see cref="NatsServer"/> for MQTT server operations.
/// Mirrors the server-receiver MQTT functions in server/mqtt.go.
/// <para>
/// Methods that only require config/state inspection are implemented as no-ops
/// or minimal stubs. Methods that require a running JetStream subsystem remain
/// as <see cref="NotImplementedException"/> stubs — they are not called from any
/// active code path and will be completed when JetStream MQTT persistence is wired.
/// </para>
/// </summary>
internal static class MqttServerExtensions
{
    /// <summary>
    /// Start listening for MQTT client connections.
    /// Mirrors Go <c>(*Server).startMQTT()</c>.
    /// </summary>
    public static void StartMqtt(this NatsServer server)
    {
        server.StartMqttListener();
    }

    // ------------------------------------------------------------------
    // Config / state — implemented as minimal stubs
    // ------------------------------------------------------------------

    /// <summary>
    /// Configure MQTT authentication overrides from the MQTT options block.
    /// No-op: auth overrides are applied through the standard auth pipeline.
    /// Mirrors Go <c>(*Server).mqttConfigAuth()</c>.
    /// </summary>
    public static void MqttConfigAuth(this NatsServer _server, object _mqttOpts)
    {
        // No-op: MQTT auth merges into the server's standard auth configuration.
    }

    /// <summary>
    /// Handle cleanup when an MQTT client connection closes.
    /// Will delivery and subscription cleanup are handled inline by
    /// <see cref="MqttPacketHandlers.HandleDisconnect"/> and
    /// <see cref="MqttPacketHandlers.DeliverWill"/>.
    /// Mirrors Go <c>(*Server).mqttHandleClosedClient()</c>.
    /// </summary>
    public static void MqttHandleClosedClient(this NatsServer _server, ClientConnection _client)
    {
        // No-op: cleanup is handled by HandleDisconnect/DeliverWill in MqttPacketHandlers.
    }

    /// <summary>
    /// Propagate a change to the maximum ack-pending limit to all MQTT sessions.
    /// No-op: ack-pending limits are enforced at the JetStream consumer level.
    /// Mirrors Go <c>(*Server).mqttUpdateMaxAckPending()</c>.
    /// </summary>
    public static void MqttUpdateMaxAckPending(this NatsServer _server, ushort _maxp)
    {
        // No-op: will be wired when JetStream consumer ack-pending is implemented.
    }

    /// <summary>
    /// Determine how many JetStream replicas to use for MQTT streams.
    /// Returns 1 (standalone mode). Cluster-aware replica selection requires
    /// the clustering subsystem.
    /// Mirrors Go <c>(*Server).mqttDetermineReplicas()</c>.
    /// </summary>
    public static int MqttDetermineReplicas(this NatsServer _server) => 1;

    // ------------------------------------------------------------------
    // JetStream-dependent — stubs (not called from any active code path)
    // ------------------------------------------------------------------

    /// <summary>
    /// Retrieve or lazily-create the JSA for the named account.
    /// Requires JetStream subsystem.
    /// Mirrors Go <c>(*Server).mqttGetJSAForAccount()</c>.
    /// </summary>
    public static MqttJsa MqttGetJsaForAccount(this NatsServer server, string account) =>
        throw new NotImplementedException("requires JetStream subsystem");

    /// <summary>
    /// Store a QoS message for an account on a (possibly new) NATS subject.
    /// Requires JetStream subsystem.
    /// Mirrors Go <c>(*Server).mqttStoreQoSMsgForAccountOnNewSubject()</c>.
    /// </summary>
    public static void MqttStoreQosMsgForAccountOnNewSubject(
        this NatsServer server,
        int hdr, byte[] msg, string account, string subject) =>
        throw new NotImplementedException("requires JetStream subsystem");

    /// <summary>
    /// Get or create the <see cref="MqttAccountSessionManager"/> for the client's account.
    /// Requires JetStream subsystem.
    /// Mirrors Go <c>(*Server).getOrCreateMQTTAccountSessionManager()</c>.
    /// </summary>
    public static MqttAccountSessionManager GetOrCreateMqttAccountSessionManager(
        this NatsServer server, object client) =>
        throw new NotImplementedException("requires JetStream subsystem");

    /// <summary>
    /// Create a new <see cref="MqttAccountSessionManager"/> for the given account.
    /// Requires JetStream subsystem.
    /// Mirrors Go <c>(*Server).mqttCreateAccountSessionManager()</c>.
    /// </summary>
    public static MqttAccountSessionManager MqttCreateAccountSessionManager(
        this NatsServer server, object account, System.Threading.CancellationToken cancel) =>
        throw new NotImplementedException("requires JetStream subsystem");

    /// <summary>
    /// Initiate delivery of a PUBLISH message via JetStream.
    /// Requires JetStream subsystem.
    /// Mirrors Go <c>(*Server).mqttInitiateMsgDelivery()</c>.
    /// </summary>
    public static void MqttInitiateMsgDelivery(
        this NatsServer server, object client, MqttPublishInfo pp) =>
        throw new NotImplementedException("requires JetStream subsystem");

    /// <summary>
    /// Store a QoS-2 PUBLISH exactly once (idempotent).
    /// Requires JetStream subsystem.
    /// Mirrors Go <c>(*Server).mqttStoreQoS2MsgOnce()</c>.
    /// </summary>
    public static void MqttStoreQoS2MsgOnce(
        this NatsServer server, object client, MqttPublishInfo pp) =>
        throw new NotImplementedException("requires JetStream subsystem");

    /// <summary>
    /// Audit retained-message permissions after a configuration reload.
    /// Requires JetStream subsystem.
    /// Mirrors Go <c>(*Server).mqttCheckPubRetainedPerms()</c>.
    /// </summary>
    public static void MqttCheckPubRetainedPerms(this NatsServer server) =>
        throw new NotImplementedException("requires JetStream subsystem");
}
