{$mode objfpc}{$H+}

/// <summary>
/// uCICMQTTClient — MQTT 5.0 client (with 3.1.1 fallback) for Free Pascal / Lazarus.
/// Uses FPC built-in ssockets/sslsockets for TCP/TLS transport (TLS 1.3 capable).
/// No LCL dependency. Cross-platform (Windows + Linux).
/// Supports QoS 0 and QoS 1 publish, subscribe, unsubscribe, keepalive, and
/// automatic reconnection with exponential back-off.
/// </summary>
unit uCICMQTTClient;

interface

uses
  SysUtils, Classes, SyncObjs,
  ssockets, sslsockets, opensslsockets, sslbase,
  uCICMQTTProtocol;

// ---------------------------------------------------------------------------
// Public types
// ---------------------------------------------------------------------------

type
  TCICMQTTQoS = (mqQoS0, mqQoS1, mqQoS2);

  TCICMQTTConnectionStatus = (
    mqDisconnected,
    mqConnecting,
    mqConnected,
    mqDisconnecting
  );

  TCICMQTTReasonCode = (
    mqrcSuccess                   = $00,
    mqrcNoMatchingSubscribers     = $10,
    mqrcUnspecifiedError          = $80,
    mqrcMalformedPacket           = $81,
    mqrcNotAuthorized             = $87,
    mqrcServerBusy                = $89,
    mqrcTopicNameInvalid          = $90,
    mqrcPacketIdInUse             = $91,
    mqrcQuotaExceeded             = $97,
    mqrcUnsupportedProtocolVersion = $84
  );

  TCICMQTTMessageEvent    = procedure(const ATopic, APayload: String;
                              AQoS: TCICMQTTQoS) of object;
  TCICMQTTDisconnectEvent = procedure(AReasonCode: TCICMQTTReasonCode;
                              const AReason: String) of object;
  TCICMQTTLogEvent        = procedure(const AMessage: String) of object;
  /// <summary>Fires after every successful connect (initial AND each
  /// reconnect), from the connecting thread — use it to (re)subscribe, since
  /// CleanStart drops server-side subscriptions on reconnect.</summary>
  TCICMQTTConnectEvent    = procedure of object;

// ---------------------------------------------------------------------------
// Forward declarations
// ---------------------------------------------------------------------------

  TCICMQTTClient = class;

// ---------------------------------------------------------------------------
// Internal thread classes
// ---------------------------------------------------------------------------

  /// <summary>
  /// Background thread that reads incoming MQTT packets from the socket.
  /// </summary>
  TCICMQTTReaderThread = class(TThread)
  private
    FClient: TCICMQTTClient;
  protected
    procedure Execute; override;
  public
    constructor Create(AClient: TCICMQTTClient);
  end;

  /// <summary>
  /// Background thread that sends PINGREQ packets to maintain the keepalive,
  /// and triggers reconnection when the broker stops responding.
  /// </summary>
  TCICMQTTKeepaliveThread = class(TThread)
  private
    FClient: TCICMQTTClient;
  protected
    procedure Execute; override;
  public
    constructor Create(AClient: TCICMQTTClient);
  end;

// ---------------------------------------------------------------------------
// Main client class
// ---------------------------------------------------------------------------

  TCICMQTTClient = class
  private
    // TCP + SSL
    FSocket:      TInetSocket;
    FSSLHandler:  TSSLSocketHandler;
    FConnected:   Boolean;
    FStatus:      TCICMQTTConnectionStatus;
    FNextPacketId: Word;
    FProtocolLevel: Byte;          // 5 = MQTT 5.0, 4 = MQTT 3.1.1 fallback
    FLastConnectReasonCode: Byte;
    FKeepaliveThread: TThread;
    FReaderThread:    TThread;
    FLock:            TCriticalSection;
    FConnectEvent:    TEvent;      // signalled when CONNACK arrives
    FLastSendTime:    TDateTime;
    FLastRecvTime:    TDateTime;
    FUserDisconnect:  Boolean;     // True when Disconnect() was called by user
    FReconnecting:    Boolean;
    FForceReconnect:  Boolean;     // keepalive timeout signals the I/O thread here

    // Outbound byte-buffers awaiting write. The I/O (reader) thread is the SOLE
    // toucher of the socket: it drains this queue (all writes) then reads, so
    // reads and writes never overlap on the OpenSSL SSL*. FLock guards ONLY this
    // queue and shared flags — never a socket read/write.
    FOutQueue:    TList;           // stores PBytes (heap-copied packet buffers)

    // Pending QoS-1 packet IDs (simple list)
    FPendingIds:  TList;           // stores Word-sized IDs as Pointers

    // Internal helpers
    procedure DoLog(const AMsg: String);
    procedure DoDisconnectEvent(ARC: TCICMQTTReasonCode; const AReason: String);
    procedure DoMessageEvent(const ATopic, APayload: String; AQoS: TCICMQTTQoS);

    function  NextPacketId: Word;
    procedure AddPendingId(AId: Word);
    procedure RemovePendingId(AId: Word);

    // Packet builders
    function  EncodeVarInt(AValue: Cardinal): TBytes;
    function  EncodeUTF8(const AStr: String): TBytes;
    function  BuildConnect: TBytes;
    function  BuildPublish(const ATopic, APayload: String;
                AQoS: TCICMQTTQoS; APacketId: Word; ARetain: Boolean = False): TBytes;
    function  BuildPubAck(APacketId: Word): TBytes;
    function  BuildSubscribe(const ATopic: String;
                AQoS: TCICMQTTQoS; APacketId: Word): TBytes;
    function  BuildUnsubscribe(const ATopic: String; APacketId: Word): TBytes;
    function  BuildPingReq: TBytes;
    function  BuildDisconnect(AReasonCode: Byte): TBytes;

    // Outbound queue: SendBytes enqueues (any thread); FlushOut drains + writes
    // (I/O thread ONLY); ClearOutQueue discards pending (on disconnect).
    procedure SendBytes(const AData: TBytes);
    procedure FlushOut;
    procedure ClearOutQueue;

    // Packet reading helpers (called from reader thread, no lock needed)
    function  ReadByte: Byte;
    function  ReadWord: Word;
    function  ReadVarInt(out AValue: Cardinal): Boolean;
    function  ReadUTF8String: String;
    procedure SkipBytes(ACount: Cardinal);
    procedure SkipPacketBody;

    // Packet dispatchers
    procedure HandleConnAck;
    procedure HandlePublish(AFlags: Byte);
    procedure HandlePubAck;
    procedure HandleSubAck;
    procedure HandleUnsubAck;
    procedure HandlePingResp;
    procedure HandleDisconnect;

    // Connection management
    procedure DoConnect;
    procedure DoDisconnect(AReasonCode: TCICMQTTReasonCode; AUserInitiated: Boolean);
    procedure StartThreads;
    procedure StopThreads;
    procedure ReconnectLoop;

  public
    BrokerHostName:       String;
    BrokerPort:           Integer;
    UseSSL:               Boolean;
    ClientID:             String;
    Username:             String;
    Password:             String;
    KeepAliveSecs:        Integer;
    CleanStart:           Boolean;
    SessionExpiryInterval: UInt32;
    ConnectTimeoutSecs:   Integer;

    OnMessage:    TCICMQTTMessageEvent;
    OnDisconnect: TCICMQTTDisconnectEvent;
    OnLog:        TCICMQTTLogEvent;
    OnConnect:    TCICMQTTConnectEvent;   // fires after every successful connect

    /// <summary>Returns True when the client is in the Connected state.</summary>
    function  IsConnected: Boolean;

    /// <summary>Returns the current connection status.</summary>
    function  ConnectionStatus: TCICMQTTConnectionStatus;

    /// <summary>Returns the reason code received in the last CONNACK packet.</summary>
    function  LastConnectReasonCode: TCICMQTTReasonCode;

    /// <summary>
    /// Initiates a TCP (and optionally TLS) connection to the broker and
    /// sends an MQTT CONNECT packet. Blocks until CONNACK is received or
    /// ConnectTimeoutSecs elapses. Raises an exception on failure.
    /// </summary>
    procedure Connect;

    /// <summary>
    /// Sends an MQTT DISCONNECT packet and closes the TCP connection.
    /// Setting AReasonCode to mqrcSuccess is normal closure.
    /// </summary>
    procedure Disconnect(AReasonCode: TCICMQTTReasonCode = mqrcSuccess);

    /// <summary>
    /// Publishes a message. Returns the Packet Identifier used (0 for QoS 0).
    /// For QoS 1 the ID is tracked until PUBACK is received.
    /// </summary>
    function  Publish(const ATopic, APayload: String;
                AQoS: TCICMQTTQoS = mqQoS0; ARetain: Boolean = False): Word;

    /// <summary>
    /// Sends a SUBSCRIBE request for the given topic filter at the given QoS.
    /// Returns the Packet Identifier.
    /// </summary>
    function  Subscribe(const ATopic: String;
                AQoS: TCICMQTTQoS = mqQoS0): Word;

    /// <summary>
    /// Sends an UNSUBSCRIBE request for the given topic filter.
    /// Returns the Packet Identifier.
    /// </summary>
    function  Unsubscribe(const ATopic: String): Word;

    constructor Create;
    destructor  Destroy; override;
  end;

// ---------------------------------------------------------------------------
// All MQTT packet type constants and protocol level constants are defined in
// uCICMQTTProtocol.  This unit uses them from there directly.
// ---------------------------------------------------------------------------

implementation

type
  PBytes = ^TBytes;

const
  // Idle-poll timeout (ms) for the I/O thread's fixed-header read. It bounds two
  // things: idle CPU (one wakeup per interval) and the worst-case latency before
  // a just-enqueued outbound packet is flushed (the I/O thread drains the whole
  // queue at the top of each loop). Throughput is NOT bounded by this — a drain
  // writes every queued packet in one pass. Once a packet starts arriving the
  // read switches to the full operation timeout so a fragmented body is not
  // misread as a disconnect.
  READ_POLL_TIMEOUT_MS = 25;

// ===========================================================================
// TCICMQTTReaderThread
// ===========================================================================

constructor TCICMQTTReaderThread.Create(AClient: TCICMQTTClient);
begin
  FClient    := AClient;
  FreeOnTerminate := False;
  inherited Create(False);
end;

procedure TCICMQTTReaderThread.Execute;
var
  FixedHeader: Byte;
  PacketType:  Byte;
  Flags:       Byte;
  N:           LongInt;
  B:           Byte;
begin
  while not Terminated do
  begin
    try
      if not FClient.FConnected or (FClient.FSocket = nil) then
      begin
        Sleep(100);
        Continue;
      end;

      // The keepalive thread only SIGNALS a timeout (sets FForceReconnect); the
      // teardown + reconnect happens here so the socket is torn down and rebuilt
      // exclusively on this thread — no other thread ever frees it mid-read.
      if FClient.FForceReconnect then
      begin
        FClient.FForceReconnect := False;
        FClient.DoLog('Keepalive timeout — reconnecting');
        FClient.DoDisconnect(mqrcUnspecifiedError, False);
        if Terminated or FClient.FUserDisconnect or not FClient.FConnected then Break;
        Continue;
      end;

      // This thread is the SOLE toucher of the socket: drain every queued
      // outbound packet (all writes), then read. Reads and writes are on one
      // thread, so they never overlap on the OpenSSL SSL* — no lock spans I/O.
      FClient.FlushOut;

      FClient.FSocket.IOTimeout := READ_POLL_TIMEOUT_MS;
      N := FClient.FSocket.Read(B, 1); // 0 on idle timeout
      if N > 0 then
      begin
        FClient.FSocket.IOTimeout := FClient.KeepAliveSecs * 1000;
        FixedHeader := B;
        PacketType  := (FixedHeader shr 4) and $0F;
        Flags       := FixedHeader and $0F;

        case PacketType of
          MQTT_CONNACK:     FClient.HandleConnAck;
          MQTT_PUBLISH:     FClient.HandlePublish(Flags);
          MQTT_PUBACK:      FClient.HandlePubAck;
          MQTT_SUBACK:      FClient.HandleSubAck;
          MQTT_UNSUBACK:    FClient.HandleUnsubAck;
          MQTT_PINGRESP:    FClient.HandlePingResp;
          MQTT_DISCONNECT:  FClient.HandleDisconnect;
        else
          // Unknown packet type — skip remaining length + payload
          FClient.SkipPacketBody;
          FClient.DoLog(Format('Unknown packet type $%2.2X - discarded', [PacketType]));
        end;
      end
      else if Terminated or FClient.FUserDisconnect then
        Break; // idle-timeout poll returned nothing and we are shutting down

    except
      on E: Exception do
      begin
        if Terminated or FClient.FUserDisconnect then
          Break; // Clean shutdown — do not reconnect
        FClient.DoLog('I/O thread error: ' + E.ClassName + ': ' + E.Message);
        // We own the socket — tear down and reconnect on THIS thread, then
        // resume the loop against the new socket (StartThreads is a no-op for
        // the reader while this thread is still running).
        FClient.DoDisconnect(mqrcUnspecifiedError, False);
        if Terminated or FClient.FUserDisconnect or not FClient.FConnected then Break;
      end;
    end;
  end;
end;

// ===========================================================================
// TCICMQTTKeepaliveThread
// ===========================================================================

constructor TCICMQTTKeepaliveThread.Create(AClient: TCICMQTTClient);
begin
  FClient := AClient;
  FreeOnTerminate := False;
  inherited Create(False);
end;

procedure TCICMQTTKeepaliveThread.Execute;
var
  HalfInterval:  Integer;
  FullInterval:  Integer;
  ElapsedSend:   Double;
  ElapsedRecv:   Double;
begin
  HalfInterval := FClient.KeepAliveSecs div 2;
  if HalfInterval < 1 then HalfInterval := 1;
  FullInterval := FClient.KeepAliveSecs;

  while not Terminated do
  begin
    Sleep(1000); // check every second

    if Terminated then Break;
    if FClient.FStatus <> mqConnected then Continue;

    ElapsedSend := (Now - FClient.FLastSendTime) * SecsPerDay;
    ElapsedRecv := (Now - FClient.FLastRecvTime) * SecsPerDay;

    // Send PINGREQ if we have not sent anything recently
    if ElapsedSend >= HalfInterval then
    begin
      try
        FClient.SendBytes(FClient.BuildPingReq);
        FClient.DoLog('PINGREQ sent');
      except
        on E: Exception do
          FClient.DoLog('Keepalive PINGREQ failed: ' + E.Message);
      end;
    end;

    // If no data received within the full keepalive window, SIGNAL the I/O
    // thread to reconnect. We must not tear the socket down from here — only the
    // I/O thread (the sole socket owner) may free/rebuild it, or it could vanish
    // out from under an in-progress read.
    if ElapsedRecv >= FullInterval then
    begin
      FClient.DoLog(Format(
        'No data received for %ds — keepalive timeout, signalling reconnect', [FullInterval]));
      FClient.FForceReconnect := True;
    end;
  end;
end;

// ===========================================================================
// TCICMQTTClient — constructor / destructor
// ===========================================================================

constructor TCICMQTTClient.Create;
begin
  inherited Create;
  FLock         := TCriticalSection.Create;
  FConnectEvent := TEvent.Create(nil, True, False, '');
  FPendingIds   := TList.Create;
  FOutQueue     := TList.Create;
  FForceReconnect := False;

  // Defaults
  BrokerPort           := 8883;
  UseSSL               := True;
  KeepAliveSecs        := 60;
  CleanStart           := True;
  SessionExpiryInterval := 0;
  ConnectTimeoutSecs   := 10;
  FProtocolLevel       := MQTT_PROTOCOL_LEVEL_50;
  FNextPacketId        := 1;
  FStatus              := mqDisconnected;
  FUserDisconnect      := False;
  FReconnecting        := False;

  FLastSendTime := Now;
  FLastRecvTime := Now;

  // Socket created on connect
  FSocket     := nil;
  FSSLHandler := nil;
  FConnected  := False;
end;

destructor TCICMQTTClient.Destroy;
begin
  if FStatus <> mqDisconnected then
    Disconnect(mqrcSuccess);

  StopThreads;

  FreeAndNil(FSocket);
  // FSSLHandler is owned by TInetSocket, freed automatically
  ClearOutQueue;
  FreeAndNil(FOutQueue);
  FreeAndNil(FPendingIds);
  FreeAndNil(FConnectEvent);
  FreeAndNil(FLock);
  inherited Destroy;
end;

// ===========================================================================
// Logging + event dispatchers
// ===========================================================================

procedure TCICMQTTClient.DoLog(const AMsg: String);
begin
  if Assigned(OnLog) then
    OnLog(AMsg);
end;

procedure TCICMQTTClient.DoDisconnectEvent(ARC: TCICMQTTReasonCode;
  const AReason: String);
begin
  if Assigned(OnDisconnect) then
    OnDisconnect(ARC, AReason);
end;

procedure TCICMQTTClient.DoMessageEvent(const ATopic, APayload: String;
  AQoS: TCICMQTTQoS);
begin
  if Assigned(OnMessage) then
    OnMessage(ATopic, APayload, AQoS);
end;

// ===========================================================================
// Packet ID management
// ===========================================================================

function TCICMQTTClient.NextPacketId: Word;
begin
  FLock.Acquire;
  try
    Result := FNextPacketId;
    Inc(FNextPacketId);
    if FNextPacketId = 0 then FNextPacketId := 1; // wrap, avoid 0
  finally
    FLock.Release;
  end;
end;

procedure TCICMQTTClient.AddPendingId(AId: Word);
begin
  FLock.Acquire;
  try
    FPendingIds.Add(Pointer(PtrUInt(AId)));
  finally
    FLock.Release;
  end;
end;

procedure TCICMQTTClient.RemovePendingId(AId: Word);
var
  I: Integer;
begin
  FLock.Acquire;
  try
    for I := 0 to FPendingIds.Count - 1 do
      if Word(PtrUInt(FPendingIds[I])) = AId then
      begin
        FPendingIds.Delete(I);
        Break;
      end;
  finally
    FLock.Release;
  end;
end;

// ===========================================================================
// Variable-length integer encoding / decoding
// ===========================================================================

function TCICMQTTClient.EncodeVarInt(AValue: Cardinal): TBytes;
begin
  Result := uCICMQTTProtocol.EncodeVarInt(AValue);
end;

function TCICMQTTClient.ReadVarInt(out AValue: Cardinal): Boolean;
var
  Multiplier: Cardinal;
  B:          Byte;
  Iterations: Integer;
begin
  Result     := False;
  AValue     := 0;
  Multiplier := 1;
  Iterations := 0;
  repeat
    B := ReadByte;
    Inc(Iterations);
    AValue := AValue + (B and $7F) * Multiplier;
    Multiplier := Multiplier * 128;
    if Iterations > 4 then Exit; // malformed
  until (B and $80) = 0;
  Result := True;
end;

// ===========================================================================
// UTF-8 string encoding (length-prefixed, 2-byte big-endian length)
// ===========================================================================

function TCICMQTTClient.EncodeUTF8(const AStr: String): TBytes;
begin
  Result := uCICMQTTProtocol.EncodeMQTTString(AStr);
end;

// ===========================================================================
// Low-level read helpers (called from reader thread — no lock)
// ===========================================================================

function TCICMQTTClient.ReadByte: Byte;
var
  B: Byte;
  N: LongInt;
begin
  N := FSocket.Read(B, 1);
  if N <= 0 then
    raise ESocketError.Create('Connection closed or read timeout');
  Result := B;
end;

function TCICMQTTClient.ReadWord: Word;
var
  Hi, Lo: Byte;
begin
  Hi := ReadByte;
  Lo := ReadByte;
  Result := (Word(Hi) shl 8) or Lo;
end;

function TCICMQTTClient.ReadUTF8String: String;
var
  Len: Word;
  Buf: TBytes;
  N:   LongInt;
begin
  Len := ReadWord;
  if Len = 0 then
  begin
    Result := '';
    Exit;
  end;
  SetLength(Buf, Len);
  N := FSocket.Read(Buf[0], Len);
  if N < Len then
    raise ESocketError.Create('Short read on UTF8 string');
  Result := TEncoding.UTF8.GetString(Buf);
end;

procedure TCICMQTTClient.SkipBytes(ACount: Cardinal);
var
  Buf:       array[0..1023] of Byte;
  Remaining: Cardinal;
  ToRead:    Integer;
begin
  Remaining := ACount;
  while Remaining > 0 do
  begin
    ToRead := Remaining;
    if ToRead > SizeOf(Buf) then ToRead := SizeOf(Buf);
    FSocket.Read(Buf, ToRead);
    Dec(Remaining, ToRead);
  end;
end;

procedure TCICMQTTClient.SkipPacketBody;
var
  RemLen: Cardinal;
begin
  if ReadVarInt(RemLen) then
    SkipBytes(RemLen);
end;

// ===========================================================================
// Packet builders
// ===========================================================================

function TCICMQTTClient.BuildConnect: TBytes;
begin
  Result := uCICMQTTProtocol.BuildConnectPacket(
    FProtocolLevel, CleanStart, KeepAliveSecs, ClientID,
    Username, Password, SessionExpiryInterval);
end;

function TCICMQTTClient.BuildPublish(const ATopic, APayload: String;
  AQoS: TCICMQTTQoS; APacketId: Word; ARetain: Boolean = False): TBytes;
begin
  Result := uCICMQTTProtocol.BuildPublishPacket(
    ATopic, TEncoding.UTF8.GetBytes(APayload),
    Byte(Ord(AQoS)), APacketId, FProtocolLevel, ARetain);
end;

function TCICMQTTClient.BuildPubAck(APacketId: Word): TBytes;
begin
  Result := uCICMQTTProtocol.BuildPubAckPacket(APacketId, 0, FProtocolLevel);
end;

function TCICMQTTClient.BuildSubscribe(const ATopic: String;
  AQoS: TCICMQTTQoS; APacketId: Word): TBytes;
begin
  Result := uCICMQTTProtocol.BuildSubscribePacket(
    ATopic, Byte(Ord(AQoS)), APacketId, FProtocolLevel);
end;

function TCICMQTTClient.BuildUnsubscribe(const ATopic: String;
  APacketId: Word): TBytes;
begin
  Result := uCICMQTTProtocol.BuildUnsubscribePacket(ATopic, APacketId, FProtocolLevel);
end;

function TCICMQTTClient.BuildPingReq: TBytes;
begin
  Result := uCICMQTTProtocol.BuildPingReqPacket;
end;

function TCICMQTTClient.BuildDisconnect(AReasonCode: Byte): TBytes;
begin
  Result := uCICMQTTProtocol.BuildDisconnectPacket(AReasonCode, FProtocolLevel);
end;

// ===========================================================================
// Thread-safe send
// ===========================================================================

// Enqueue bytes for the I/O thread to write. Safe to call from ANY thread — it
// never touches the socket, only the lock-guarded queue. (During the connect
// handshake, DoConnect writes directly since no I/O thread is running yet.)
procedure TCICMQTTClient.SendBytes(const AData: TBytes);
var
  p: PBytes;
begin
  if Length(AData) = 0 then Exit;
  New(p);
  p^ := Copy(AData, 0, Length(AData));
  FLock.Acquire;
  try
    FOutQueue.Add(p);
  finally
    FLock.Release;
  end;
end;

// Drain the outbound queue, writing each buffer to the socket. MUST be called
// only where nothing else is reading/writing the socket — i.e. the I/O thread's
// loop, or StopThreads after the I/O thread has exited. Only the brief dequeue
// is locked; the writes happen unlocked (this is the sole writer at that point).
procedure TCICMQTTClient.FlushOut;
var
  Pending: array of PBytes;
  i, n: Integer;
begin
  FLock.Acquire;
  try
    n := FOutQueue.Count;
    SetLength(Pending, n);
    for i := 0 to n - 1 do
      Pending[i] := PBytes(FOutQueue[i]);
    FOutQueue.Clear;
  finally
    FLock.Release;
  end;
  if n = 0 then Exit;

  try
    for i := 0 to n - 1 do
      if (FSocket <> nil) and FConnected and (Length(Pending[i]^) > 0) then
      begin
        FSocket.IOTimeout := KeepAliveSecs * 1000;
        FSocket.Write(Pending[i]^[0], Length(Pending[i]^));
        FLastSendTime := Now;
      end;
  finally
    for i := 0 to n - 1 do
      Dispose(Pending[i]);
  end;
end;

// Discard any queued outbound packets (e.g. stale ones on disconnect).
procedure TCICMQTTClient.ClearOutQueue;
var
  i: Integer;
  p: PBytes;
begin
  FLock.Acquire;
  try
    for i := 0 to FOutQueue.Count - 1 do
    begin
      p := PBytes(FOutQueue[i]);
      Dispose(p);
    end;
    FOutQueue.Clear;
  finally
    FLock.Release;
  end;
end;

// ===========================================================================
// Packet handlers (called from reader thread)
// ===========================================================================

procedure TCICMQTTClient.HandleConnAck;
var
  RemLen:         Cardinal;
  SessionPresent: Byte;
  ReasonCode:     Byte;
  PropLen:        Cardinal;
begin
  if not ReadVarInt(RemLen) then Exit;

  SessionPresent := ReadByte and $01;
  ReasonCode     := ReadByte;

  FLastConnectReasonCode := ReasonCode;
  FLastRecvTime := Now;

  // Parse (and discard) properties for MQTT 5.0
  if (FProtocolLevel = MQTT_PROTOCOL_LEVEL_50) and (RemLen > 2) then
  begin
    if ReadVarInt(PropLen) then
      SkipBytes(PropLen);
  end;

  DoLog(Format('CONNACK received — session_present=%d reason=$%2.2X',
    [SessionPresent, ReasonCode]));

  if ReasonCode = Byte(mqrcUnsupportedProtocolVersion) then
  begin
    // Trigger 3.1.1 fallback — signal ConnectEvent so Connect() can retry
    DoLog('Broker rejected MQTT 5.0 — will retry with MQTT 3.1.1');
    FProtocolLevel := MQTT_PROTOCOL_LEVEL_311;
  end;

  if ReasonCode = 0 then
  begin
    FLock.Acquire;
    try
      FStatus := mqConnected;
    finally
      FLock.Release;
    end;
  end;

  FConnectEvent.SetEvent;
end;

procedure TCICMQTTClient.HandlePublish(AFlags: Byte);
var
  RemLen:         Cardinal;
  QoS:            TCICMQTTQoS;
  Retain:         Boolean;
  Topic:          String;
  PacketId:       Word;
  PropLen:        Cardinal;
  PayloadLen:     Integer;
  PayloadBuf:     TBytes;
  Payload:        String;
  TopicLen:       Word;
  HeaderConsumed: Cardinal;
  TopicBuf:       TBytes;
  N:              LongInt;
begin
  QoS    := TCICMQTTQoS((AFlags shr 1) and $03);
  Retain := (AFlags and $01) <> 0;

  if not ReadVarInt(RemLen) then Exit;
  FLastRecvTime := Now;

  // Track how many bytes we read from the variable header
  // so we know how many bytes remain for the payload.
  TopicLen := ReadWord;
  HeaderConsumed := 2 + TopicLen;

  SetLength(TopicBuf, TopicLen);
  N := FSocket.Read(TopicBuf[0], TopicLen);
  if N < TopicLen then
    raise ESocketError.Create('Short read on topic');
  Topic := TEncoding.UTF8.GetString(TopicBuf);

  PacketId := 0;
  if QoS in [mqQoS1, mqQoS2] then
  begin
    PacketId := ReadWord;
    Inc(HeaderConsumed, 2);
  end;

  // Properties (MQTT 5.0)
  if FProtocolLevel = MQTT_PROTOCOL_LEVEL_50 then
  begin
    if ReadVarInt(PropLen) then
    begin
      // PropLen bytes for the varint itself varies; approximate 1 byte for small values
      // We track consumed bytes by using RemLen - HeaderConsumed - PropLen - PropLenFieldSize
      // Simpler: just skip the props and compute payload from what's left.
      SkipBytes(PropLen);
      // varint for 0..127 is 1 byte
      if PropLen <= $7F then
        Inc(HeaderConsumed, 1 + PropLen)
      else if PropLen <= $3FFF then
        Inc(HeaderConsumed, 2 + PropLen)
      else if PropLen <= $1FFFFF then
        Inc(HeaderConsumed, 3 + PropLen)
      else
        Inc(HeaderConsumed, 4 + PropLen);
    end;
  end;

  PayloadLen := Integer(RemLen) - Integer(HeaderConsumed);
  if PayloadLen < 0 then PayloadLen := 0;

  if PayloadLen > 0 then
  begin
    SetLength(PayloadBuf, PayloadLen);
    N := FSocket.Read(PayloadBuf[0], PayloadLen);
    if N < PayloadLen then
      raise ESocketError.Create('Short read on payload');
    Payload := TEncoding.UTF8.GetString(PayloadBuf);
  end
  else
    Payload := '';

  DoLog(Format('PUBLISH received topic="%s" QoS=%d retain=%s payload_len=%d',
    [Topic, Ord(QoS), BoolToStr(Retain, True), PayloadLen]));

  // Send PUBACK for QoS 1
  if QoS = mqQoS1 then
  begin
    try
      SendBytes(BuildPubAck(PacketId));
      DoLog(Format('PUBACK sent for packet id %d', [PacketId]));
    except
      on E: Exception do
        DoLog('Failed to send PUBACK: ' + E.Message);
    end;
  end;

  DoMessageEvent(Topic, Payload, QoS);
end;

procedure TCICMQTTClient.HandlePubAck;
var
  RemLen:     Cardinal;
  PacketId:   Word;
  ReasonCode: Byte;
  PropLen:    Cardinal;
begin
  if not ReadVarInt(RemLen) then Exit;
  FLastRecvTime := Now;

  PacketId   := ReadWord;
  ReasonCode := 0;

  if (FProtocolLevel = MQTT_PROTOCOL_LEVEL_50) and (RemLen >= 3) then
  begin
    ReasonCode := ReadByte;
    if RemLen > 3 then
    begin
      if ReadVarInt(PropLen) then
        SkipBytes(PropLen);
    end;
  end
  else if RemLen > 2 then
    SkipBytes(RemLen - 2);

  DoLog(Format('PUBACK received packet_id=%d reason=$%2.2X', [PacketId, ReasonCode]));
  RemovePendingId(PacketId);
end;

procedure TCICMQTTClient.HandleSubAck;
var
  RemLen:    Cardinal;
  PacketId:  Word;
  PropLen:   Cardinal;
  RC:        Byte;
  BytesLeft: Integer;
begin
  if not ReadVarInt(RemLen) then Exit;
  FLastRecvTime := Now;

  PacketId  := ReadWord;
  BytesLeft := Integer(RemLen) - 2;

  if (FProtocolLevel = MQTT_PROTOCOL_LEVEL_50) and (BytesLeft > 0) then
  begin
    if ReadVarInt(PropLen) then
    begin
      SkipBytes(PropLen);
      // Approximate varint field size
      if PropLen <= $7F then Dec(BytesLeft, 1 + Integer(PropLen))
      else if PropLen <= $3FFF then Dec(BytesLeft, 2 + Integer(PropLen))
      else Dec(BytesLeft, 3 + Integer(PropLen));
    end;
  end;

  // Read per-topic reason codes
  while BytesLeft > 0 do
  begin
    RC := ReadByte;
    Dec(BytesLeft);
    DoLog(Format('SUBACK packet_id=%d reason=$%2.2X', [PacketId, RC]));
  end;
end;

procedure TCICMQTTClient.HandleUnsubAck;
var
  RemLen:    Cardinal;
  PacketId:  Word;
  PropLen:   Cardinal;
  RC:        Byte;
  BytesLeft: Integer;
begin
  if not ReadVarInt(RemLen) then Exit;
  FLastRecvTime := Now;

  PacketId  := ReadWord;
  BytesLeft := Integer(RemLen) - 2;

  if (FProtocolLevel = MQTT_PROTOCOL_LEVEL_50) and (BytesLeft > 0) then
  begin
    if ReadVarInt(PropLen) then
    begin
      SkipBytes(PropLen);
      if PropLen <= $7F then Dec(BytesLeft, 1 + Integer(PropLen))
      else if PropLen <= $3FFF then Dec(BytesLeft, 2 + Integer(PropLen))
      else Dec(BytesLeft, 3 + Integer(PropLen));
    end;
  end;

  while BytesLeft > 0 do
  begin
    RC := ReadByte;
    Dec(BytesLeft);
    DoLog(Format('UNSUBACK packet_id=%d reason=$%2.2X', [PacketId, RC]));
  end;
end;

procedure TCICMQTTClient.HandlePingResp;
var
  RemLen: Cardinal;
begin
  ReadVarInt(RemLen); // should be 0
  FLastRecvTime := Now;
  DoLog('PINGRESP received');
end;

procedure TCICMQTTClient.HandleDisconnect;
var
  RemLen:     Cardinal;
  ReasonCode: Byte;
  PropLen:    Cardinal;
begin
  ReasonCode := 0;
  if ReadVarInt(RemLen) then
  begin
    if RemLen >= 1 then
    begin
      ReasonCode := ReadByte;
      if (FProtocolLevel = MQTT_PROTOCOL_LEVEL_50) and (RemLen > 1) then
      begin
        if ReadVarInt(PropLen) then
          SkipBytes(PropLen);
      end;
    end;
  end;

  DoLog(Format('DISCONNECT received from broker reason=$%2.2X', [ReasonCode]));

  FLock.Acquire;
  try
    FStatus := mqDisconnected;
  finally
    FLock.Release;
  end;

  DoDisconnectEvent(TCICMQTTReasonCode(ReasonCode),
    Format('Broker disconnected (reason=$%2.2X)', [ReasonCode]));

  if not FUserDisconnect then
    ReconnectLoop;
end;

// ===========================================================================
// Connection lifecycle
// ===========================================================================

procedure TCICMQTTClient.DoConnect;
var
  ConnPkt:    TBytes;
  B:          Byte;
  N:          LongInt;
  PacketType: Byte;
begin
  DoLog(Format('Connecting to %s:%d (SSL=%s, protocol=%d)',
    [BrokerHostName, BrokerPort, BoolToStr(UseSSL, True), FProtocolLevel]));

  FConnectEvent.ResetEvent;
  FForceReconnect := False;

  // Free any previous socket
  FreeAndNil(FSocket);

  if UseSSL then
  begin
    FSSLHandler := TSSLSocketHandler.GetDefaultHandler;
    FSSLHandler.SSLType := stAny; // negotiates highest available (TLS 1.3 if supported)
    FSSLHandler.VerifyPeerCert := False; // accept self-signed certs
    FSSLHandler.SendHostAsSNI := True;
    FSocket := TInetSocket.Create(BrokerHostName, BrokerPort, ConnectTimeoutSecs * 1000, FSSLHandler);
    FSocket.Connect;
  end
  else
  begin
    FSSLHandler := nil;
    FSocket := TInetSocket.Create(BrokerHostName, BrokerPort, ConnectTimeoutSecs * 1000);
  end;

  FConnected := True;
  DoLog('TCP connected');

  // --- CONNECT/CONNACK handshake, done SYNCHRONOUSLY on this thread. No I/O
  // thread is running yet (initial connect) or it IS this thread (reconnect),
  // so the CONNECT write and the CONNACK read own the socket exclusively —
  // no concurrent access, and the 3.1.1 fallback can free/rebuild the socket
  // safely. The I/O + keepalive threads start only AFTER the session is up. ---
  FSocket.IOTimeout := ConnectTimeoutSecs * 1000;
  ConnPkt := BuildConnect;
  FSocket.Write(ConnPkt[0], Length(ConnPkt));
  FLastSendTime := Now;
  DoLog('CONNECT packet sent');

  N := FSocket.Read(B, 1);
  if N <= 0 then
    raise Exception.Create('Timeout waiting for CONNACK');
  PacketType := (B shr 4) and $0F;
  if PacketType <> MQTT_CONNACK then
    raise Exception.CreateFmt('Expected CONNACK, got packet type $%2.2X', [PacketType]);
  HandleConnAck; // reads the remainder; sets reason code, protocol level, status

  if FLastConnectReasonCode = Byte(mqrcUnsupportedProtocolVersion) then
  begin
    // 3.1.1 fallback: FProtocolLevel already set by HandleConnAck; reconnect
    DoLog('Retrying with MQTT 3.1.1');
    FreeAndNil(FSocket);
    FConnected := False;
    DoConnect; // recursive call — will use FProtocolLevel = 4 this time
    Exit;
  end;

  if FLastConnectReasonCode <> 0 then
  begin
    FConnected := False;
    FreeAndNil(FSocket);
    FStatus := mqDisconnected;
    raise Exception.CreateFmt('Broker refused connection (reason=$%2.2X)',
      [FLastConnectReasonCode]);
  end;

  FSocket.IOTimeout := KeepAliveSecs * 1000;
  FLastSendTime := Now;
  FLastRecvTime := Now;

  // Session established — now start the I/O + keepalive threads. From here the
  // I/O thread is the sole toucher of the socket.
  StartThreads;
  DoLog('MQTT session established');

  // Notify the app so it can (re)subscribe. Runs on the connecting thread; the
  // handler must only enqueue work (e.g. Subscribe, which is thread-safe).
  if Assigned(OnConnect) then
    OnConnect;
end;

procedure TCICMQTTClient.StartThreads;
begin
  if (FReaderThread = nil) or FReaderThread.Finished then
  begin
    FReaderThread := TCICMQTTReaderThread.Create(Self);
    DoLog('Reader thread started');
  end;
  if (FKeepaliveThread = nil) or FKeepaliveThread.Finished then
  begin
    FKeepaliveThread := TCICMQTTKeepaliveThread.Create(Self);
    DoLog('Keepalive thread started');
  end;
end;

procedure TCICMQTTClient.StopThreads;
begin
  if FReaderThread <> nil then
    FReaderThread.Terminate;
  if FKeepaliveThread <> nil then
    FKeepaliveThread.Terminate;

  FConnectEvent.SetEvent; // unblock anything waiting on it

  // Nudge a blocked read to return promptly so WaitFor doesn't stall on the full
  // operation timeout. A plain integer property set — harmless even if the I/O
  // thread reads it concurrently.
  if FSocket <> nil then
    try FSocket.IOTimeout := READ_POLL_TIMEOUT_MS; except end;

  // Wait for the I/O thread to actually EXIT before freeing the socket. It is
  // the sole reader/writer, so once it has gone nothing can touch the socket and
  // the free below cannot race a read. (A user Disconnect sets FUserDisconnect
  // first, so the I/O thread breaks out rather than reconnecting.)
  if FReaderThread <> nil then
  begin
    FReaderThread.WaitFor;
    FreeAndNil(FReaderThread);
  end;
  if FKeepaliveThread <> nil then
  begin
    FKeepaliveThread.WaitFor;
    FreeAndNil(FKeepaliveThread);
  end;

  // I/O thread is gone — flush any last queued packet (e.g. the DISCONNECT that
  // Disconnect enqueued) while the socket is still open, then close it.
  try FlushOut; except end;

  FLock.Acquire;
  try
    FConnected := False;
    FreeAndNil(FSocket);
  finally
    FLock.Release;
  end;

  ClearOutQueue;
end;

procedure TCICMQTTClient.DoDisconnect(AReasonCode: TCICMQTTReasonCode;
  AUserInitiated: Boolean);
begin
  FLock.Acquire;
  try
    if not FConnected then Exit;
  finally
    FLock.Release;
  end;

  try
    if AUserInitiated then
      SendBytes(BuildDisconnect(Byte(AReasonCode)));
  except
  end;

  FLock.Acquire;
  try
    FConnected := False;
    FreeAndNil(FSocket);
  finally
    FLock.Release;
  end;

  FLock.Acquire;
  try
    FStatus := mqDisconnected;
  finally
    FLock.Release;
  end;

  if not AUserInitiated then
  begin
    DoDisconnectEvent(AReasonCode, Format('Disconnected (reason=$%2.2X)', [Byte(AReasonCode)]));
    ReconnectLoop;
  end;
end;

procedure TCICMQTTClient.ReconnectLoop;
var
  DelaySecs: Integer;
  Attempt:   Integer;
begin
  FLock.Acquire;
  try
    if FReconnecting or FUserDisconnect then Exit;
    FReconnecting := True;
  finally
    FLock.Release;
  end;

  try
    DelaySecs := 1;
    Attempt   := 0;
    while not FUserDisconnect do
    begin
      Inc(Attempt);
      DoLog(Format('Reconnect attempt %d in %ds...', [Attempt, DelaySecs]));
      Sleep(DelaySecs * 1000);
      if FUserDisconnect then Break;

      try
        DoConnect;
        DoLog('Reconnected successfully');
        Break;
      except
        on E: Exception do
        begin
          DoLog(Format('Reconnect attempt %d failed: %s', [Attempt, E.Message]));
          DelaySecs := DelaySecs * 2;
          if DelaySecs > 60 then DelaySecs := 60;
        end;
      end;
    end;
  finally
    FLock.Acquire;
    try
      FReconnecting := False;
    finally
      FLock.Release;
    end;
  end;
end;

// ===========================================================================
// Public API
// ===========================================================================

function TCICMQTTClient.IsConnected: Boolean;
begin
  FLock.Acquire;
  try
    Result := FConnected and (FSocket <> nil);
  finally
    FLock.Release;
  end;
end;

function TCICMQTTClient.ConnectionStatus: TCICMQTTConnectionStatus;
begin
  Result := FStatus;
end;

function TCICMQTTClient.LastConnectReasonCode: TCICMQTTReasonCode;
begin
  Result := TCICMQTTReasonCode(FLastConnectReasonCode);
end;

/// <summary>
/// Initiates connection to the MQTT broker. Sets status to mqConnecting,
/// establishes TCP (and TLS if UseSSL is True), sends CONNECT, and waits
/// for CONNACK. Raises an exception if the connection fails or times out.
/// Automatically falls back to MQTT 3.1.1 if the broker rejects MQTT 5.0.
/// </summary>
procedure TCICMQTTClient.Connect;
begin
  FLock.Acquire;
  try
    if FStatus in [mqConnecting, mqConnected] then Exit;
    FStatus         := mqConnecting;
    FUserDisconnect := False;
  finally
    FLock.Release;
  end;

  try
    DoConnect;
  except
    FLock.Acquire;
    try
      FStatus := mqDisconnected;
    finally
      FLock.Release;
    end;
    raise;
  end;
end;

/// <summary>
/// Sends an MQTT DISCONNECT packet and closes the TCP connection.
/// Stops background threads. Does not trigger automatic reconnection.
/// </summary>
procedure TCICMQTTClient.Disconnect(AReasonCode: TCICMQTTReasonCode = mqrcSuccess);
begin
  FLock.Acquire;
  try
    FUserDisconnect := True;
  finally
    FLock.Release;
  end;

  // Send DISCONNECT packet while socket is still open
  try
    if FConnected and (FSocket <> nil) then
      SendBytes(BuildDisconnect(Byte(AReasonCode)));
  except
  end;

  StopThreads;

  FLock.Acquire;
  try
    FStatus := mqDisconnected;
  finally
    FLock.Release;
  end;

  DoLog('Disconnected by user');
end;

/// <summary>
/// Publishes an MQTT message to the given topic with the specified QoS level.
/// For QoS 0 the returned Packet Identifier is always 0.
/// For QoS 1 the Packet Identifier is tracked until PUBACK is received.
/// Raises an exception if not connected.
/// </summary>
function TCICMQTTClient.Publish(const ATopic, APayload: String;
  AQoS: TCICMQTTQoS = mqQoS0; ARetain: Boolean = False): Word;
var
  PktId: Word;
begin
  if FStatus <> mqConnected then
    raise Exception.Create('MQTT client is not connected');

  if AQoS = mqQoS0 then
    PktId := 0
  else
    PktId := NextPacketId;

  SendBytes(BuildPublish(ATopic, APayload, AQoS, PktId, ARetain));

  if AQoS = mqQoS1 then
    AddPendingId(PktId);

  DoLog(Format('PUBLISH sent topic="%s" QoS=%d packet_id=%d payload_len=%d',
    [ATopic, Ord(AQoS), PktId, Length(APayload)]));

  Result := PktId;
end;

/// <summary>
/// Subscribes to the given topic filter at the given QoS level.
/// Returns the Packet Identifier used in the SUBSCRIBE packet.
/// Raises an exception if not connected.
/// </summary>
function TCICMQTTClient.Subscribe(const ATopic: String;
  AQoS: TCICMQTTQoS = mqQoS0): Word;
var
  PktId: Word;
begin
  if FStatus <> mqConnected then
    raise Exception.Create('MQTT client is not connected');

  PktId := NextPacketId;
  SendBytes(BuildSubscribe(ATopic, AQoS, PktId));
  DoLog(Format('SUBSCRIBE sent topic="%s" QoS=%d packet_id=%d',
    [ATopic, Ord(AQoS), PktId]));
  Result := PktId;
end;

/// <summary>
/// Unsubscribes from the given topic filter.
/// Returns the Packet Identifier used in the UNSUBSCRIBE packet.
/// Raises an exception if not connected.
/// </summary>
function TCICMQTTClient.Unsubscribe(const ATopic: String): Word;
var
  PktId: Word;
begin
  if FStatus <> mqConnected then
    raise Exception.Create('MQTT client is not connected');

  PktId := NextPacketId;
  SendBytes(BuildUnsubscribe(ATopic, PktId));
  DoLog(Format('UNSUBSCRIBE sent topic="%s" packet_id=%d', [ATopic, PktId]));
  Result := PktId;
end;

end.