#include "ayla/session.hpp"

#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <mutex>
#include <new>

#include "ayla/envelope.hpp"
#include "ayla/httpc.hpp"
#include "ayla/json.hpp"
#include "ayla/log.hpp"
#include "ayla/platform/platform.hpp"

namespace fgl::ayla {

namespace {

constexpr uint32_t kLoopTickMs = 20;
constexpr size_t kMaxName = 40;
constexpr size_t kMaxQueue = 32;

struct Command {
  uint8_t type;  // 1=GET, 2=SET, 3=DELETE
  char name[kMaxName];
  int64_t value;
  char base_type[10];
  int cmd_id;
};

bool gen_random_token(char* out, size_t len) {
  static const char kAlpha[] =
      "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
  uint8_t rnd[32];
  size_t need = len <= sizeof(rnd) ? len : sizeof(rnd);
  if (!plat::random(rnd, need)) return false;
  for (size_t i = 0; i < len; i++) {
    out[i] = kAlpha[rnd[i % need] % 62];
  }
  out[len] = '\0';
  return true;
}

// "cmd_id=5&status=200" -> значения; отсутствующие остаются прежними.
void parse_query(const char* query, int* cmd_id, int* status) {
  const char* p = query;
  while (*p != '\0') {
    const char* eq = strchr(p, '=');
    const char* amp = strchr(p, '&');
    const char* end = (amp != nullptr) ? amp : p + strlen(p);
    if (eq != nullptr && eq < end) {
      size_t klen = static_cast<size_t>(eq - p);
      int v = atoi(eq + 1);
      if (klen == 6 && strncmp(p, "cmd_id", 6) == 0) *cmd_id = v;
      if (klen == 6 && strncmp(p, "status", 6) == 0) *status = v;
    }
    if (amp == nullptr) break;
    p = amp + 1;
  }
}

}  // namespace

struct Session::Impl {
  SessionConfig cfg{};
  SessionTimings timings{};
  SessionCallbacks cbs{};

  char host[64] = {};
  char dsn[40] = {};
  char lanip_key[64] = {};

  // ---- httpd-поток (используется только из httpd-потока) ----
  HttpServer httpd;
  SessionCrypto crypto;
  bool crypto_ready = false;
  int64_t out_seq = 0;
  char ke_random2[17] = {};
  // Рабочие буферы datapoint-пути (httpd однопоточен; вынесены из стека —
  // экономия ~7КБ стека потока httpd).
  char dp_body[kHttpdMaxBody + 1];
  char dp_enc_b64[kEnvelopeMaxB64];
  char dp_sign_b64[96];
  char dp_plain[kEnvelopeMaxPlain];
  char dp_data[kEnvelopeMaxPlain];

  // ---- разделяемое ----
  std::mutex queue_mu;
  Command queue[kMaxQueue] = {};
  uint8_t queue_len = 0;
  int next_cmd_id = 1;
  bool batch_open = false;
  Command batch[kMaxQueue] = {};
  uint8_t batch_len = 0;

  std::atomic<uint8_t> state{static_cast<uint8_t>(SessionState::kIdle)};
  std::atomic<int> last_error{static_cast<int>(SessionError::kNone)};
  std::atomic<uint64_t> ke_time_ms{0};       // время ответа на KE (0 — не было)
  std::atomic<uint64_t> last_local_reg_ms{0};
  std::atomic<uint64_t> quiet_until_ms{0};   // пауза local_reg (восстановление)
  std::atomic<uint64_t> retry_at_ms{0};
  std::atomic<bool> had_poll_since_ke{false};
  std::atomic<bool> ever_active{false};      // был online хотя бы раз
  std::atomic<bool> want_notify{false};      // после batch commit / перехода online
  std::atomic<bool> delete_pending{false};
  std::atomic<bool> delete_served{false};
  std::atomic<bool> decrypt_failed{false};
  std::atomic<bool> running{false};

  std::atomic<uint32_t> rekeys{0};
  std::atomic<uint32_t> pushes_ok{0};
  std::atomic<uint32_t> pushes_bad{0};
  std::atomic<uint32_t> cmds_served{0};

  plat::ThreadId thread = nullptr;
  uint8_t backoff_attempts = 0;
  uint32_t backoff_ms = 0;
  bool reg_ok = false;  // последний local_reg принят (202/200)

  uint16_t listen_port_actual = 0;

  // ---------- helpers (вызывается из обоих потоков) ----------
  void set_state(SessionState st, SessionError err) {
    uint8_t prev = state.exchange(static_cast<uint8_t>(st),
                                  std::memory_order_acq_rel);
    last_error.store(static_cast<int>(err), std::memory_order_release);
    if (prev != static_cast<uint8_t>(st) && cbs.on_state != nullptr) {
      cbs.on_state(cbs.ctx, st, err);  // без дублирования одинаковых состояний
    }
  }

  // ---------- очередь (mutex) ----------
  bool enqueue_locked(const Command& cmd) {
    // coalescing: SET замещает незабранный SET того же свойства;
    // GET-дубликат отбрасывается; DELETE — единственный.
    if (cmd.type == 3) {
      for (uint8_t i = 0; i < queue_len; i++) {
        if (queue[i].type == 3) return true;
      }
    } else {
      for (uint8_t i = 0; i < queue_len; i++) {
        if (queue[i].type == cmd.type &&
            strncmp(queue[i].name, cmd.name, kMaxName) == 0) {
          if (cmd.type == 2) {
            queue[i].value = cmd.value;  // замещаем
            return true;
          }
          if (cmd.type == 1) return true;  // дубликат GET
        }
      }
    }
    if (queue_len >= cfg.max_queue || queue_len >= kMaxQueue) return false;
    queue[queue_len++] = cmd;
    return true;
  }

  bool submit(Command cmd) {
    std::lock_guard<std::mutex> lk(queue_mu);
    bool ok;
    if (batch_open && cmd.type != 3) {
      if (batch_len >= kMaxQueue) return false;
      // coalescing внутри batch
      for (uint8_t i = 0; i < batch_len; i++) {
        if (batch[i].type == cmd.type &&
            strncmp(batch[i].name, cmd.name, kMaxName) == 0) {
          if (cmd.type == 2) {
            batch[i].value = cmd.value;
            return true;
          }
          if (cmd.type == 1) return true;
        }
      }
      batch[batch_len++] = cmd;
      ok = true;
    } else {
      ok = enqueue_locked(cmd);
    }
    return ok;
  }

  // ---------- httpd-обработчики (httpd-поток) ----------
  static bool http_handler(const HttpRequest& req, HttpResponse& resp,
                           void* ctx);

  void handle_key_exchange(const HttpRequest& req, HttpResponse& resp);
  void handle_commands(HttpResponse& resp);
  void handle_datapoint(const HttpRequest& req, HttpResponse& resp);

  void build_get_payload(char* out, size_t out_cap, const Command& c,
                         int* seq_out);
  void build_set_payload(char* out, size_t out_cap, const Command& c);

  // ---------- session-поток ----------
  static void session_thread_trampoline(void* ctx) {
    static_cast<Impl*>(ctx)->session_loop();
  }
  void session_loop();
  bool send_local_reg(bool notify, bool first);
};

// ---------------------------------------------------------------------------
// httpd
// ---------------------------------------------------------------------------

bool Session::Impl::http_handler(const HttpRequest& req, HttpResponse& resp,
                                 void* ctx) {
  auto* impl = static_cast<Impl*>(ctx);
  if (strcmp(req.method, "POST") == 0) {
    if (strcmp(req.target, "/local_lan/key_exchange.json") == 0) {
      impl->handle_key_exchange(req, resp);
      return true;
    }
    if (strcmp(req.target, "/local_lan/property/datapoint.json") == 0) {
      impl->handle_datapoint(req, resp);
      return true;
    }
    if (strcmp(req.target, "/local_lan/property/datapoint/ack.json") == 0 ||
        strcmp(req.target, "/local_lan/node/property/datapoint.json") == 0 ||
        strcmp(req.target, "/local_lan/node/property/datapoint/ack.json") == 0) {
      FGL_LOGD("session: ack/node datapoint (пусто ok)");
      resp.status = 200;
      return true;
    }
  } else if (strcmp(req.method, "GET") == 0) {
    if (strcmp(req.target, "/local_lan/commands.json") == 0) {
      if (!impl->crypto_ready) {
        resp.status = 401;
        return true;
      }
      impl->handle_commands(resp);
      return true;
    }
  }
  resp.status = 404;
  return true;
}

void Session::Impl::handle_key_exchange(const HttpRequest& req,
                                        HttpResponse& resp) {
  // Тело: {"key_exchange":{"ver":1,"proto":1,"key_id":N,"random_1":..,"time_1":N,"sec":""}}
  char body[kHttpdMaxBody + 1];
  size_t n = req.body_len < kHttpdMaxBody ? req.body_len : kHttpdMaxBody;
  memcpy(body, req.body, n);
  body[n] = '\0';

  json::Doc doc;
  if (!doc.parse(body)) {
    resp.status = 400;
    return;
  }
  int64_t ver = 0, proto = 0, key_id = 0, time_1 = 0;
  char random_1[32] = {}, sec[8] = {};
  bool have_r1 = doc.get_string("random_1", random_1, sizeof(random_1));
  bool have_t1 = doc.get_int("time_1", &time_1);
  bool have_sec = doc.get_string("sec", sec, sizeof(sec));
  bool have_ver = doc.get_int("ver", &ver);
  bool have_proto = doc.get_int("proto", &proto);
  bool have_kid = doc.get_int("key_id", &key_id);
  if (!have_r1 || !have_t1 || !have_ver || !have_proto || !have_kid) {
    FGL_LOGW("session: key_exchange неполный");
    resp.status = 400;
    return;
  }
  if (ver != 1 || proto != 1 || (have_sec && sec[0] != '\0')) {
    FGL_LOGW("session: key_exchange ver/proto/sec не поддержаны");
    set_state(SessionState::kKeyError, SessionError::kBadKeyExchange);
    resp.status = 426;
    return;
  }
  {
    // sec длиннее буфера -> get_string=false, но поле есть: setup-режим
    // не поддерживаем — тоже 426 (PROTOCOL §3.1).
    const char* sec_start = nullptr;
    size_t sec_len = 0;
    jsmntype_t sec_type;
    if (doc.find("sec", &sec_start, &sec_len, &sec_type) && sec_type == JSMN_STRING &&
        sec_len >= sizeof(sec)) {
      set_state(SessionState::kKeyError, SessionError::kBadKeyExchange);
      resp.status = 426;
      return;
    }
  }
  if (static_cast<uint32_t>(key_id) != cfg.lanip_key_id) {
    FGL_LOGE("session: key_id %lld != %u — ротация ключа?",
             static_cast<long long>(key_id),
             static_cast<unsigned>(cfg.lanip_key_id));
    set_state(SessionState::kKeyError, SessionError::kKeyMismatch);
    resp.status = 412;
    return;
  }

  if (!gen_random_token(ke_random2, 16)) {
    resp.status = 500;
    return;
  }
  int64_t time_2 = static_cast<int64_t>(plat::now_ms()) * 1000000ll;
  crypto_ready = crypto.init(lanip_key, random_1, ke_random2, time_1,
                             time_2);  // цепочки сбрасываются тут же
  if (!crypto_ready) {
    resp.status = 500;
    return;
  }
  had_poll_since_ke.store(false, std::memory_order_release);
  ke_time_ms.store(plat::now_ms(), std::memory_order_release);
  bool rekey = ever_active.load(std::memory_order_acquire);
  rekeys.fetch_add(1, std::memory_order_relaxed);
  if (!rekey && state.load(std::memory_order_acquire) !=
                    static_cast<uint8_t>(SessionState::kRegistering)) {
    set_state(SessionState::kRegistering, SessionError::kNone);
  }
  FGL_LOGI("session: key exchange #%u (rekey=%d)",
           static_cast<unsigned>(rekeys.load(std::memory_order_relaxed)), rekey);

  static thread_local char out[128];
  json::Writer w(out, sizeof(out));
  w.begin_object();
  w.key("random_2");
  w.string(ke_random2);
  w.key("time_2");
  w.integer(time_2);
  w.end_object();
  resp.status = 200;
  resp.body = reinterpret_cast<const uint8_t*>(out);
  resp.body_len = strlen(out);
  // буфер out живёт до конца ответа (send_response в handle_connection
  // выполняется синхронно в том же кадре стека httpd-потока).
}

void Session::Impl::build_get_payload(char* out, size_t out_cap,
                                      const Command& c, int* seq_out) {
  (void)seq_out;
  json::Writer w(out, out_cap);
  w.begin_object();
  w.key("cmds");
  w.begin_array();
  w.begin_object();
  w.key("cmd");
  w.begin_object();
  w.key("cmd_id");
  w.integer(c.cmd_id);
  w.key("method");
  w.string("GET");
  w.key("resource");
  char res[80];
  snprintf(res, sizeof(res), "property.json?name=%s", c.name);
  w.string(res);
  w.key("data");
  w.string("");
  w.key("uri");
  w.string("/local_lan/property/datapoint.json");
  w.end_object();
  w.end_object();
  w.end_array();
  w.end_object();
}

void Session::Impl::build_set_payload(char* out, size_t out_cap,
                                      const Command& c) {
  json::Writer w(out, out_cap);
  w.begin_object();
  w.key("properties");
  w.begin_array();
  w.begin_object();
  w.key("property");
  w.begin_object();
  w.key("base_type");
  w.string(c.base_type);
  w.key("name");
  w.string(c.name);
  w.key("value");
  if (strcmp(c.base_type, "boolean") == 0) {
    w.boolean(c.value != 0);
  } else {
    w.integer(c.value);
  }
  w.key("id");
  char id[9];
  gen_random_token(id, 8);
  w.string(id);
  w.end_object();
  w.end_object();
  w.end_array();
  w.end_object();
}

void Session::Impl::handle_commands(HttpResponse& resp) {
  Command head{};
  bool have = false;
  uint8_t remaining = 0;
  {
    std::lock_guard<std::mutex> lk(queue_mu);
    if (queue_len > 0) {
      head = queue[0];
      have = true;
      queue_len--;
      memmove(queue, queue + 1, queue_len * sizeof(Command));
      remaining = queue_len;
    }
  }
  char payload[512];
  if (!have) {
    payload[0] = '{';
    payload[1] = '}';
    payload[2] = '\0';
  } else if (head.type == 1) {
    build_get_payload(payload, sizeof(payload), head, nullptr);
  } else if (head.type == 2) {
    build_set_payload(payload, sizeof(payload), head);
  } else {  // DELETE session
    json::Writer w(payload, sizeof(payload));
    w.begin_object();
    w.key("cmds");
    w.begin_array();
    w.begin_object();
    w.key("cmd");
    w.begin_object();
    w.key("cmd_id");
    w.integer(0);
    w.key("method");
    w.string("DELETE");
    w.key("resource");
    w.string("local_reg.json");
    w.key("data");
    w.string("delete_session");
    w.key("uri");
    w.string("/local_lan");
    w.end_object();
    w.end_object();
    w.end_array();
    w.end_object();
    delete_served.store(true, std::memory_order_release);
    delete_pending.store(false, std::memory_order_release);
    FGL_LOGI("session: delete_session выдан модулю");
  }

  static thread_local char envelope[kEnvelopeMaxB64];
  int64_t seq = out_seq++;
  if (!envelope_pack(crypto.app, seq, payload, envelope, sizeof(envelope))) {
    resp.status = 500;
    return;
  }
  cmds_served.fetch_add(1, std::memory_order_relaxed);
  had_poll_since_ke.store(true, std::memory_order_release);
  uint8_t st8 = state.load(std::memory_order_acquire);
  if (st8 == static_cast<uint8_t>(SessionState::kRegistering) ||
      st8 == static_cast<uint8_t>(SessionState::kRecovering) ||
      st8 == static_cast<uint8_t>(SessionState::kOffline)) {
    // Опрос команд = сессия жива (в т.ч. после re-key при recovering/offline).
    ever_active.store(true, std::memory_order_release);
    decrypt_failed.store(false, std::memory_order_release);
    set_state(SessionState::kOnline, SessionError::kNone);
    want_notify.store(true, std::memory_order_release);
  }
  resp.status = remaining > 0 ? 206 : 200;
  resp.body = reinterpret_cast<const uint8_t*>(envelope);
  resp.body_len = strlen(envelope);
}

void Session::Impl::handle_datapoint(const HttpRequest& req,
                                     HttpResponse& resp) {
  size_t n = req.body_len < kHttpdMaxBody ? req.body_len : kHttpdMaxBody;
  memcpy(dp_body, req.body, n);
  dp_body[n] = '\0';
  json::Doc wrap;
  if (!wrap.parse(dp_body)) {
    resp.status = 400;
    return;
  }
  if (!wrap.get_string("enc", dp_enc_b64, sizeof(dp_enc_b64)) ||
      !wrap.get_string("sign", dp_sign_b64, sizeof(dp_sign_b64))) {
    resp.status = 400;
    return;
  }

  int64_t seq_no = -1;
  if (!envelope_unpack(crypto.dev, dp_enc_b64, dp_sign_b64, dp_plain,
                       sizeof(dp_plain), &seq_no)) {
    pushes_bad.fetch_add(1, std::memory_order_relaxed);
    FGL_LOGW("session: push не расшифрован/подпись (401); пауза и re-key");
    if (state.load(std::memory_order_acquire) ==
        static_cast<uint8_t>(SessionState::kOnline)) {
      decrypt_failed.store(true, std::memory_order_release);
      set_state(SessionState::kRecovering, SessionError::kDecryptFailed);
      // Стратегия восстановления (PROTOCOL §4.4): замолчать на > порога
      // «возврата» (~44-50с) — следующий local_reg заставит модуль re-key.
      quiet_until_ms.store(plat::now_ms() + timings.recovering_quiet_ms,
                           std::memory_order_release);
    }
    resp.status = 401;
    return;
  }

  // Восстановление после 401 с живой цепочкой (бракованная подпись):
  // сообщение расшифровано — отменяем тишину и возвращаем online.
  if (decrypt_failed.exchange(false, std::memory_order_acq_rel)) {
    quiet_until_ms.store(0, std::memory_order_release);
    set_state(SessionState::kOnline, SessionError::kNone);
    FGL_LOGI("session: цепочка восстановлена (успешный push после 401)");
  }

  // {"seq_no":N,"data":{"name":..,"value":..}}
  json::Doc top;
  if (!top.parse(dp_plain)) {
    pushes_ok.fetch_add(1, std::memory_order_relaxed);
    resp.status = 200;
    return;
  }
  const char* data_start = nullptr;
  size_t data_len = 0;
  jsmntype_t data_type;
  PropertyEvent ev{};
  ev.seq_no = seq_no;
  if (top.find("data", &data_start, &data_len, &data_type) &&
      data_type == JSMN_OBJECT && data_len < sizeof(dp_data)) {
    memcpy(dp_data, data_start, data_len);
    dp_data[data_len] = '\0';
    {
      json::Doc data_doc;
      if (data_doc.parse(dp_data)) {
        char name[kMaxName];
        if (data_doc.get_string("name", name, sizeof(name))) {
          snprintf(ev.name, sizeof(ev.name), "%s", name);
          int64_t iv = 0;
          bool bv = false;
          const char* sv = nullptr;
          size_t sv_len = 0;
          jsmntype_t vt;
          if (data_doc.get_int("value", &iv)) {
            ev.is_int = true;
            ev.int_value = iv;
          } else if (data_doc.get_bool("value", &bv)) {
            ev.is_bool = true;
            ev.bool_value = bv;
          } else if (data_doc.find("value", &sv, &sv_len, &vt) &&
                     vt == JSMN_STRING && sv_len + 1 <= sizeof(ev.str_value)) {
            memcpy(ev.str_value, sv, sv_len);
            ev.str_value[sv_len] = '\0';
          }
          int cmd_id = -1, status = 0;
          parse_query(req.query, &cmd_id, &status);
          ev.cmd_id = cmd_id;
          ev.status = status;
          if (cbs.on_property != nullptr) {
            cbs.on_property(cbs.ctx, ev);
          }
        }
      }
    }
  }
  pushes_ok.fetch_add(1, std::memory_order_relaxed);
  resp.status = 200;
}

// ---------------------------------------------------------------------------
// session-поток: local_reg / keep-alive / backoff / таймауты
// ---------------------------------------------------------------------------

bool Session::Impl::send_local_reg(bool notify, bool first) {
  char path[96];
  if (first) {
    snprintf(path, sizeof(path), "/local_reg.json?dsn=%s", dsn);
  } else {
    snprintf(path, sizeof(path), "/local_reg.json");
  }
  char local_ip[24];
  if (!plat::local_ip_for(host, local_ip, sizeof(local_ip))) {
    FGL_LOGW("session: local_ip_for(%s) failed", host);
    return false;
  }
  char body[160];
  json::Writer w(body, sizeof(body));
  w.begin_object();
  w.key("local_reg");
  w.begin_object();
  w.key("ip");
  w.string(local_ip);
  w.key("notify");
  w.boolean(notify);
  w.key("port");
  w.integer(listen_port_actual);
  w.key("uri");
  w.string("/local_lan");
  w.end_object();
  w.end_object();
  if (!w.ok()) return false;

  HttpcRequest req;
  req.method = first ? "POST" : "PUT";
  req.host = host;
  req.port = cfg.device_port;
  req.path = path;
  req.body = reinterpret_cast<const uint8_t*>(body);
  req.body_len = strlen(body);
  req.timeout_ms = 5000;
  HttpcResponse resp;
  if (!httpc_perform(req, &resp)) {
    return false;
  }
  last_local_reg_ms.store(plat::now_ms(), std::memory_order_release);
  if (resp.status == 503) {
    set_state(SessionState::kOffline, SessionError::kNoSlot);
    retry_at_ms.store(plat::now_ms() + timings.no_slot_retry_ms,
                      std::memory_order_release);
    FGL_LOGW("session: local_reg -> 503 (нет слотов)");
    backoff_attempts = 0;
    backoff_ms = 0;
    reg_ok = false;  // сессия не активировалась — следующий local_reg POST?dsn
    return true;  // транспорт ок — это протокольный ответ
  }
  if (resp.status != 202 && resp.status != 200) {
    FGL_LOGW("session: local_reg -> %d", resp.status);
    return false;
  }
  if (state.load(std::memory_order_acquire) ==
      static_cast<uint8_t>(SessionState::kIdle)) {
    set_state(SessionState::kRegistering, SessionError::kNone);
  }
  backoff_attempts = 0;
  backoff_ms = 0;
  reg_ok = true;
  return true;
}

void Session::Impl::session_loop() {
  bool first_reg = true;
  while (running.load(std::memory_order_acquire)) {
    uint64_t now = plat::now_ms();
    SessionState st =
        static_cast<SessionState>(state.load(std::memory_order_acquire));

    // kKeyError — устойчивая ошибка: конфиг менять вручную, не дёргаем модуль.
    if (st == SessionState::kKeyError) {
      plat::sleep_ms(kLoopTickMs);
      continue;
    }

    // Тишина (восстановление после «KE без poll»).
    if (now < quiet_until_ms.load(std::memory_order_acquire)) {
      plat::sleep_ms(kLoopTickMs);
      continue;
    }

    // Активация: KE отвечен, но опроса нет.
    uint64_t ke_time = ke_time_ms.load(std::memory_order_acquire);
    if (st == SessionState::kRegistering && ke_time != 0 &&
        !had_poll_since_ke.load(std::memory_order_acquire) &&
        now - ke_time > timings.activation_timeout_ms) {
      FGL_LOGW("session: активация не наступила (KE без poll) — пауза %ums",
               static_cast<unsigned>(timings.recovering_quiet_ms));
      ke_time_ms.store(0, std::memory_order_release);
      quiet_until_ms.store(now + timings.recovering_quiet_ms,
                           std::memory_order_release);
      set_state(SessionState::kRecovering, SessionError::kActivationTimeout);
      continue;
    }

    // Backoff / отложенный повтор.
    if (now < retry_at_ms.load(std::memory_order_acquire)) {
      plat::sleep_ms(kLoopTickMs);
      continue;
    }

    // local_reg: по keep-alive, по notify (batch/online) или первичный.
    bool queue_nonempty;
    {
      std::lock_guard<std::mutex> lk(queue_mu);
      queue_nonempty = queue_len > 0;
    }
    bool notify = queue_nonempty || want_notify.exchange(false,
                                                         std::memory_order_acq_rel);
    uint64_t last_reg = last_local_reg_ms.load(std::memory_order_acquire);
    bool due = notify || last_reg == 0 ||
               now - last_reg >= timings.keepalive_ms;
    if (!due) {
      plat::sleep_ms(kLoopTickMs);
      continue;
    }

    bool first = first_reg || !reg_ok;
    if (!send_local_reg(queue_nonempty, first)) {
      // Транспортная ошибка: backoff.
      backoff_ms = backoff_ms == 0 ? timings.backoff_base_ms
                                   : (backoff_ms * 8) / 5;  // x1.6
      if (backoff_ms > timings.backoff_max_ms) {
        backoff_ms = timings.backoff_max_ms;
      }
      backoff_attempts++;
      if (backoff_attempts >= timings.backoff_attempts) {
        set_state(SessionState::kOffline, SessionError::kUnreachable);
        retry_at_ms.store(now + timings.backoff_max_ms,
                          std::memory_order_release);
        backoff_attempts = 0;
        backoff_ms = 0;
      } else {
        retry_at_ms.store(now + backoff_ms, std::memory_order_release);
      }
      continue;
    }
    if (reg_ok) first_reg = false;
  }
}

// ---------------------------------------------------------------------------
// Session (публичный класс)
// ---------------------------------------------------------------------------

Session* Session::create(const SessionConfig& cfg, const SessionCallbacks& cbs) {
  if (cfg.host == nullptr || cfg.dsn == nullptr || cfg.lanip_key == nullptr) {
    return nullptr;
  }
  if (strlen(cfg.lanip_key) >= sizeof(Impl::lanip_key) ||
      strlen(cfg.dsn) >= sizeof(Impl::dsn) ||
      strlen(cfg.host) >= sizeof(Impl::host)) {
    return nullptr;  // не помещается во внутренние копии
  }
  auto* impl = new (std::nothrow) Impl();
  if (impl == nullptr) return nullptr;
  impl->cfg = cfg;
  impl->cbs = cbs;
  if (impl->cfg.max_queue == 0) impl->cfg.max_queue = 16;
  if (impl->cfg.keepalive_ms == 0) impl->cfg.keepalive_ms = 15000;
  impl->timings.keepalive_ms = impl->cfg.keepalive_ms;
  snprintf(impl->host, sizeof(impl->host), "%s", cfg.host);
  snprintf(impl->dsn, sizeof(impl->dsn), "%s", cfg.dsn);
  snprintf(impl->lanip_key, sizeof(impl->lanip_key), "%s", cfg.lanip_key);
  auto* s = new (std::nothrow) Session(cfg, cbs);
  if (s == nullptr) {
    delete impl;
    return nullptr;
  }
  s->impl_ = impl;
  return s;
}

Session::Session(const SessionConfig&, const SessionCallbacks&) : impl_(nullptr) {}

SessionError Session::last_error() const {
  if (impl_ == nullptr) return SessionError::kNone;
  return static_cast<SessionError>(impl_->last_error.load(std::memory_order_acquire));
}

SessionState Session::state() const {
  if (impl_ == nullptr) return SessionState::kIdle;
  return static_cast<SessionState>(impl_->state.load(std::memory_order_acquire));
}

uint16_t Session::listen_port() const {
  return impl_ != nullptr ? impl_->listen_port_actual : 0;
}

uint32_t Session::rekey_count() const {
  return impl_ != nullptr ? impl_->rekeys.load(std::memory_order_relaxed) : 0;
}
uint32_t Session::pushes_ok() const {
  return impl_ != nullptr ? impl_->pushes_ok.load(std::memory_order_relaxed) : 0;
}
uint32_t Session::pushes_bad() const {
  return impl_ != nullptr ? impl_->pushes_bad.load(std::memory_order_relaxed) : 0;
}
uint32_t Session::commands_served() const {
  return impl_ != nullptr ? impl_->cmds_served.load(std::memory_order_relaxed) : 0;
}
bool Session::had_activity() const {
  return impl_ != nullptr &&
         impl_->had_poll_since_ke.load(std::memory_order_acquire);
}
Session::~Session() {
  stop();
  delete impl_;
}

bool Session::start() {
  if (impl_ == nullptr || impl_->running.load(std::memory_order_acquire)) {
    return false;
  }
  if (!impl_->httpd.start(impl_->cfg.listen_port, Impl::http_handler, impl_,
                          "fgl_session")) {
    return false;
  }
  impl_->listen_port_actual = impl_->httpd.port();
  impl_->running.store(true, std::memory_order_release);
  if (!plat::thread_create(Impl::session_thread_trampoline, impl_,
                           "fgl_sess", 8192, &impl_->thread)) {
    impl_->running.store(false, std::memory_order_release);
    impl_->httpd.stop();
    return false;
  }
  return true;
}

void Session::stop() {
  if (impl_ == nullptr || !impl_->running.load(std::memory_order_acquire)) {
    return;
  }
  // Штатное завершение: DELETE-команда + notify local_reg. Сессионный поток
  // ещё работает и доставит notify; ждём выдачи команды модулю.
  if (impl_->state.load(std::memory_order_acquire) !=
      static_cast<uint8_t>(SessionState::kKeyError)) {
    delete_session();
    uint64_t deadline = plat::now_ms() + impl_->timings.delete_wait_ms;
    while (plat::now_ms() < deadline &&
           !impl_->delete_served.load(std::memory_order_acquire)) {
      plat::sleep_ms(10);
    }
  }
  impl_->running.store(false, std::memory_order_release);
  if (impl_->thread != nullptr) {
    plat::thread_join(impl_->thread);
    impl_->thread = nullptr;
  }
  impl_->httpd.stop();
  // Сброс для возможного рестарта.
  impl_->delete_served.store(false, std::memory_order_release);
  impl_->delete_pending.store(false, std::memory_order_release);
  impl_->quiet_until_ms.store(0, std::memory_order_release);
  impl_->retry_at_ms.store(0, std::memory_order_release);
  impl_->ke_time_ms.store(0, std::memory_order_release);
  impl_->set_state(SessionState::kIdle, SessionError::kNone);
}

bool Session::get_property(const char* name) {
  if (impl_ == nullptr || name == nullptr || strlen(name) >= kMaxName) {
    return false;
  }
  Command cmd{};
  cmd.type = 1;
  snprintf(cmd.name, sizeof(cmd.name), "%s", name);
  {
    std::lock_guard<std::mutex> lk(impl_->queue_mu);
    cmd.cmd_id = impl_->next_cmd_id++;
  }
  if (!impl_->submit(cmd)) return false;
  impl_->want_notify.store(true, std::memory_order_release);
  return true;
}

bool Session::set_property(const char* name, int64_t value,
                           const char* base_type) {
  if (impl_ == nullptr || name == nullptr || strlen(name) >= kMaxName) {
    return false;
  }
  Command cmd{};
  cmd.type = 2;
  snprintf(cmd.name, sizeof(cmd.name), "%s", name);
  cmd.value = value;
  snprintf(cmd.base_type, sizeof(cmd.base_type), "%s", base_type);
  {
    std::lock_guard<std::mutex> lk(impl_->queue_mu);
    cmd.cmd_id = impl_->next_cmd_id++;
  }
  if (!impl_->submit(cmd)) return false;
  impl_->want_notify.store(true, std::memory_order_release);
  return true;
}

void Session::set_timings_for_test(const SessionTimings& t) {
  if (impl_ == nullptr) return;
  SessionTimings tmp = t;
  if (tmp.keepalive_ms < 100) tmp.keepalive_ms = 100;  // анти-спам
  impl_->timings = tmp;
}

bool Session::begin_batch() {
  std::lock_guard<std::mutex> lk(impl_->queue_mu);
  if (impl_->batch_open) return false;
  impl_->batch_open = true;
  impl_->batch_len = 0;
  return true;
}

bool Session::commit_batch() {
  {
    std::lock_guard<std::mutex> lk(impl_->queue_mu);
    if (!impl_->batch_open) return false;
    bool ok = true;
    for (uint8_t i = 0; i < impl_->batch_len; i++) {
      if (!impl_->enqueue_locked(impl_->batch[i])) ok = false;
    }
    impl_->batch_open = false;
    impl_->batch_len = 0;
    if (!ok) return false;
  }
  impl_->want_notify.store(true, std::memory_order_release);
  return true;
}

bool Session::abort_batch() {
  std::lock_guard<std::mutex> lk(impl_->queue_mu);
  if (!impl_->batch_open) return false;
  impl_->batch_open = false;
  impl_->batch_len = 0;
  return true;
}

bool Session::delete_session() {
  if (impl_ == nullptr) return false;
  Command cmd{};
  cmd.type = 3;
  snprintf(cmd.name, sizeof(cmd.name), "local_reg.json");
  {
    std::lock_guard<std::mutex> lk(impl_->queue_mu);
    if (!impl_->enqueue_locked(cmd)) return false;
  }
  impl_->delete_pending.store(true, std::memory_order_release);
  impl_->want_notify.store(true, std::memory_order_release);
  return true;
}

}  // namespace fgl::ayla