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

using System.Net;
using System.Net.Sockets;
using ZB.MOM.NatsNet.Server.Internal;
using ZB.MOM.NatsNet.Server.Mqtt;

namespace ZB.MOM.NatsNet.Server;

public sealed partial class NatsServer
{
    // =========================================================================
    // MQTT Listener and Client Creation
    // Mirrors Go startMQTT() and createMQTTClient() in server/mqtt.go.
    // =========================================================================

    /// <summary>
    /// Starts the MQTT TCP listener and accept loop.
    /// Called from <see cref="MqttServerExtensions.StartMqtt"/> extension method.
    /// Mirrors Go <c>(*Server).startMQTT()</c>.
    /// </summary>
    internal void StartMqttListener()
    {
        if (IsShuttingDown()) return;

        var opts = GetOpts();
        var port = opts.Mqtt.Port;
        var host = opts.Mqtt.Host;

        if (string.IsNullOrEmpty(host))
            host = "0.0.0.0";

        // RandomPort (-1) means ephemeral — pass 0 to TcpListener.
        var listenPort = port < 0 ? 0 : port;

        TcpListener listener;
        try
        {
            var addr = host == "0.0.0.0" || host == "::"
                ? (host == "::" ? IPAddress.IPv6Any : IPAddress.Any)
                : IPAddress.Parse(host);

            listener = new TcpListener(addr, listenPort);
            listener.Start();
        }
        catch (Exception ex)
        {
            Fatalf("Can't listen for MQTT client connections: {0}", ex.Message);
            return;
        }

        var ep = (IPEndPoint)listener.LocalEndpoint!;
        var scheme = opts.Mqtt.TlsConfig != null ? "tls" : "mqtt";
        Noticef("Listening for MQTT client connections on {0}://{1}:{2}", scheme, host, ep.Port);

        // Write back resolved port if ephemeral (0 or -1).
        if (port <= 0)
            opts.Mqtt.Port = ep.Port;

        _mu.EnterWriteLock();
        _mqttListener = listener;
        _mu.ExitWriteLock();

        // Start accept loop in a goroutine.
        _ = Task.Run(() =>
        {
            AcceptConnections(listener, "Mqtt", tc => CreateMqttClient(tc));
        });
    }

    /// <summary>
    /// Creates and registers a new MQTT client connection from the accepted TCP client.
    /// Unlike NATS clients, MQTT clients do not receive an INFO line — the MQTT CONNECT
    /// handshake is initiated by the client.
    /// Mirrors Go <c>createMQTTClient()</c> in server/mqtt.go.
    /// </summary>
    private ClientConnection? CreateMqttClient(TcpClient tc)
    {
        var opts = GetOpts();
        var now = DateTime.UtcNow;
        var nc = tc.GetStream();

        var c = new ClientConnection(ClientKind.Client, this, nc)
        {
            Start = now,
            Last = now,
            Opts = ClientOptions.Default,
            Headers = true, // MQTT always uses NATS headers for QoS metadata
        };

        c.InitMqtt(new MqttHandler
        {
            RejectQoS2Pub = opts.Mqtt.RejectQoS2Pub,
            DowngradeQoS2Sub = opts.Mqtt.DowngradeQoS2Sub,
        });

        // Register with the global account.
        var globalAcc = GlobalAccount();
        if (globalAcc != null)
        {
            try { c.RegisterWithAccount(globalAcc); }
            catch (Exception ex)
            {
                c.ReportErrRegisterAccount(globalAcc, ex);
                return null;
            }
        }

        // Register with the server — no INFO sent for MQTT clients.
        _mu.EnterWriteLock();
        if (!IsRunning() || _ldm)
        {
            if (IsShuttingDown())
                nc.Close();
            _mu.ExitWriteLock();
            return c;
        }

        _totalClients++;

        if (opts.MaxConn > 0 && _clients.Count >= opts.MaxConn)
        {
            _mu.ExitWriteLock();
            c.MaxConnExceeded();
            return null;
        }

        _clients[c.Cid] = c;
        _mu.ExitWriteLock();

        lock (c)
        {
            if (c.IsClosed())
            {
                c.CloseConnection(ClosedState.WriteError);
                return null;
            }

            c.InitClient();
            c.Debugf("MQTT client connection created");

            // Set auth timer if authentication is required.
            if (_info.AuthRequired)
            {
                c.Flags |= ClientFlags.ExpectConnect;
                c.SetAuthTimer(TimeSpan.FromSeconds(
                    opts.Mqtt.AuthTimeout > 0 ? opts.Mqtt.AuthTimeout : opts.AuthTimeout));
            }

            c.SetPingTimer();

            // Start read/write loops.
            StartGoRoutine(() => c.ReadLoop(null));
            StartGoRoutine(() => c.WriteLoop());
        }

        return c;
    }

    /// <summary>
    /// Returns the MQTT listener address, or null if not listening.
    /// Mirrors Go <c>(*Server).MQTTAddr()</c>.
    /// </summary>
    public IPEndPoint? MqttAddr()
    {
        _mu.EnterReadLock();
        try { return _mqttListener?.LocalEndpoint as IPEndPoint; }
        finally { _mu.ExitReadLock(); }
    }
}
