signalk_ws_client.cpp

Go to the documentation of this file.

#include "sensesp.h"
 
#include "signalk_ws_client.h"
 
#include <ArduinoJson.h>
#include <ESPmDNS.h>
#include <esp_http_client.h>
 
#include <map>
 
#include "sensesp/signalk/signalk_ws_delta_size.h"
 
#ifdef SENSESP_SSL_SUPPORT
#include <mbedtls/pem.h>
#include <mbedtls/sha256.h>
#include <mbedtls/ssl.h>
#include <mbedtls/x509_crt.h>
 
#include "sensesp/signalk/signalk_tofu.h"
#endif
 
#include <memory>
#include <new>
 
#include "Arduino.h"
#include "elapsedMillis.h"
#include "esp_arduino_version.h"
#include "sensesp/signalk/signalk_listener.h"
#include "sensesp/signalk/signalk_put_request.h"
#include "sensesp/signalk/signalk_put_request_listener.h"
#include "sensesp/system/uuid.h"
#include "sensesp_app.h"
 
namespace sensesp {
 
constexpr int kWsClientTaskStackSize = 8192;  // Stack for the connect worker
constexpr int kWsTransportTaskStackSize = 6144;  // Stack for esp_websocket_client internal task
constexpr TickType_t kWsSendTimeoutTicks = pdMS_TO_TICKS(5000);
// Periodic delta telemetry must never block the caller (it will be driven from
// the event loop). 0 = enqueue if the ws-client lock and transport are free
// right now, otherwise fail fast. Deltas are supersedable. See SignalK/SensESP#1033.
constexpr TickType_t kWsDeltaSendTimeoutTicks = 0;
// A device that keeps producing a delta larger than the buffer would drop one
// every send cycle; rate-limit the warning so it does not flood the log.
constexpr uint32_t kOversizeDropLogIntervalMs = 10000;
 
SKWSClient* ws_client;
 
static const char* kRequestPermission = "readwrite";
 
#ifdef SENSESP_SSL_SUPPORT
// Convert a SHA256 hash to hex string
static void sha256_to_hex(const uint8_t* sha256, char* hex) {
  for (int i = 0; i < 32; i++) {
    sprintf(hex + (i * 2), "%02x", sha256[i]);
  }
  hex[64] = '\0';
}
 
// SHA256 of a certificate's raw DER, as a 64-char hex String.
static String cert_fingerprint(const mbedtls_x509_crt* crt) {
  uint8_t sha256[32];
  mbedtls_sha256_context ctx;
  mbedtls_sha256_init(&ctx);
  mbedtls_sha256_starts(&ctx, 0);  // 0 = SHA256 (not SHA224)
  mbedtls_sha256_update(&ctx, crt->raw.p, crt->raw.len);
  mbedtls_sha256_finish(&ctx, sha256);
  mbedtls_sha256_free(&ctx);
  char hex[65];
  sha256_to_hex(sha256, hex);
  return String(hex);
}
 
// Maximum stored/displayed length of a certificate CN.
static constexpr size_t kMaxPinCnLen = 64;
 
// Extract the CN from a certificate subject, for display. The CN is
// attacker-controlled, so the result is length-bounded and stripped of
// non-printable and quote/backslash characters before it is stored or shown.
static String cert_common_name(const mbedtls_x509_crt* crt) {
  char dn[256];
  int len = mbedtls_x509_dn_gets(dn, sizeof(dn), &crt->subject);
  if (len <= 0) {
    return String("");
  }
  const char* cn = strstr(dn, "CN=");
  if (cn == nullptr) {
    return String("");
  }
  cn += 3;  // skip "CN="
  String out;
  for (size_t i = 0; i < kMaxPinCnLen && cn[i] != '\0' && cn[i] != ','; i++) {
    char c = cn[i];
    if (c >= 0x20 && c < 0x7f && c != '"' && c != '\\') {
      out += c;
    }
  }
  return out;
}
 
// PEM-encode a certificate's DER for storage. The encode buffer is allocated on
// demand and freed on return rather than held for the device's lifetime: TOFU CA
// capture happens only during the TLS handshake, so a permanent .bss buffer
// would waste ~4 KB for the whole uptime. It is heap- rather than stack-
// allocated because the verify callback runs on a small TLS task stack.
static String cert_to_pem(const mbedtls_x509_crt* crt) {
  // Sized for a large CA cert (RSA-4096 + SANs/extensions). The size is fixed,
  // not derived from crt->raw.len: PEM is base64 (~4/3 of the DER) plus the
  // header/footer and line breaks, so the encoded form is always larger than the
  // DER. On overflow mbedtls_pem_write_buffer returns an error and this returns
  // "" -- callers must treat an empty PEM as "no usable CA" and fail safe, never
  // store it.
  constexpr size_t kPemBufSize = 4096;
  std::unique_ptr<unsigned char[]> pem_buf(
      new (std::nothrow) unsigned char[kPemBufSize]);
  if (!pem_buf) {
    ESP_LOGE("SKWSClient", "TOFU: PEM buffer allocation failed");
    return String("");
  }
  size_t olen = 0;
  int r = mbedtls_pem_write_buffer(
      "-----BEGIN CERTIFICATE-----\n", "-----END CERTIFICATE-----\n",
      crt->raw.p, crt->raw.len, pem_buf.get(), kPemBufSize, &olen);
  if (r != 0) {
    ESP_LOGE("SKWSClient", "TOFU: PEM encode failed (-0x%x)", -r);
    return String("");
  }
  return String(reinterpret_cast<const char*>(pem_buf.get()));
}
 
// Normalized (lowercase, sorted, deduplicated, comma-joined) set of the
// certificate's dNSName SANs, for TOFU identity binding. Empty if the cert has
// no DNS SAN. Two certs with the same set of names produce the same string
// regardless of order, so an exact-match comparison is stable across leaf
// rotation but changes when the identity itself changes.
static String cert_dns_sans(const mbedtls_x509_crt* crt) {
  std::set<String> names;
  for (const mbedtls_x509_sequence* cur = &crt->subject_alt_names;
       cur != nullptr && cur->buf.p != nullptr; cur = cur->next) {
    mbedtls_x509_subject_alternative_name san;
    memset(&san, 0, sizeof(san));
    if (mbedtls_x509_parse_subject_alt_name(&cur->buf, &san) != 0) {
      continue;
    }
    if (san.type == MBEDTLS_X509_SAN_DNS_NAME &&
        san.san.unstructured_name.p != nullptr &&
        san.san.unstructured_name.len > 0) {
      size_t n = san.san.unstructured_name.len;
      if (n > 255) {
        n = 255;
      }
      String name;
      name.reserve(n);
      for (size_t i = 0; i < n; i++) {
        char c = static_cast<char>(san.san.unstructured_name.p[i]);
        if (c >= 'A' && c <= 'Z') {
          c = static_cast<char>(c + ('a' - 'A'));
        }
        name += c;
      }
      names.insert(name);
    }
    mbedtls_x509_free_subject_alt_name(&san);
  }
  String out;
  for (const String& s : names) {
    if (!out.isEmpty()) {
      out += ",";
    }
    out += s;
  }
  return out;
}
 
// TOFU verification callback - called during the TLS handshake, once per
// presented certificate, highest depth (CA) first down to depth 0 (leaf).
// Returns 0 to allow, non-zero to reject.
static int tofu_verify_callback(void* ctx, mbedtls_x509_crt* crt, int depth,
                                uint32_t* flags) {
  SKWSClient* client = static_cast<SKWSClient*>(ctx);
  if (client == nullptr) {
    ESP_LOGW("SKWSClient", "TOFU: no client context, allowing connection");
    *flags = 0;
    return 0;
  }
 
  if (!client->is_tofu_enabled()) {
    // Verification disabled: accept any certificate (insecure opt-out).
    *flags = 0;
    return 0;
  }
 
  // CA-anchor mode: the stored CA was installed as the trust anchor and esp-tls
  // runs VERIFY_REQUIRED, so mbedTLS has already validated this certificate and
  // set *flags. Honor that result rather than clearing it.
  if (client->has_tofu_ca()) {
    if (*flags != 0) {
      ESP_LOGE("SKWSClient", "TOFU: certificate failed CA validation (0x%lx)",
               (unsigned long)*flags);
      client->flag_cert_error();
      return MBEDTLS_ERR_X509_CERT_VERIFY_FAILED;
    }
    // Identity binding: at the leaf, require the same SAN identity captured when
    // the CA was pinned. This is what keeps a public CA safe — a valid leaf for
    // a different name signed by the same CA (e.g. any Let's Encrypt cert) is
    // rejected here even though it chains to the pinned CA.
    if (depth == 0 && !client->get_tofu_san().isEmpty()) {
      if (cert_dns_sans(crt) != client->get_tofu_san()) {
        ESP_LOGE("SKWSClient", "TOFU: leaf identity (SAN) mismatch, rejecting");
        client->flag_cert_error();
        return MBEDTLS_ERR_X509_CERT_VERIFY_FAILED;
      }
    }
    return 0;
  }
 
  // Capture / leaf-fingerprint mode (VERIFY_OPTIONAL). Collect a CA candidate
  // from the higher-depth certificates (which arrive first), then decide at the
  // leaf. The first CA:TRUE certificate seen is the highest in the presented
  // chain (closest to the root), so it is the preferred anchor.
  // Certificates above the leaf (depth > 0): collect a CA candidate -- the
  // highest CA:TRUE cert, which arrives first -- and never fail on chain-trust
  // flags during capture. A depth-0 certificate is ALWAYS treated as the leaf
  // by the fingerprint/role decision below, even if it is self-signed with
  // CA:TRUE: a single presented certificate is pinned as a leaf, never adopted
  // as a CA trust anchor. (Adopting a CA:TRUE leaf here would skip the
  // fingerprint check and let a mismatched self-signed cert be accepted.)
  if (depth > 0) {
    // basicConstraints CA:TRUE has no public getter in mbedTLS 3.x, so the
    // flag is read through the MBEDTLS_PRIVATE accessor macro.
    bool is_ca = crt->MBEDTLS_PRIVATE(ca_istrue) != 0;
    if (is_ca && !client->has_pending_ca()) {
      String ca_pem = cert_to_pem(crt);
      // Skip on encode failure: leaving no pending CA fails safe to leaf-
      // fingerprint mode rather than committing an empty (pin-disabling) anchor.
      if (!ca_pem.isEmpty()) {
        client->stash_pending_ca(ca_pem, cert_common_name(crt));
      }
    }
    *flags = 0;
    return 0;
  }
 
  // depth == 0: the leaf. Always run the fingerprint/role decision.
  String leaf_fp = cert_fingerprint(crt);
  String leaf_san = cert_dns_sans(crt);
  bool leaf_matches =
      client->has_tofu_fingerprint() && client->get_tofu_fingerprint() == leaf_fp;
  TofuCaptureDecision decision = tofu_decide_capture(
      client->has_tofu_fingerprint(), leaf_matches, client->has_pending_ca(),
      !leaf_san.isEmpty());
 
  switch (decision) {
    case TofuCaptureDecision::kReject:
      ESP_LOGE("SKWSClient", "TOFU: leaf fingerprint mismatch, rejecting");
      client->flag_cert_error();
      return MBEDTLS_ERR_X509_CERT_VERIFY_FAILED;
    case TofuCaptureDecision::kCaptureLeaf:
      ESP_LOGI("SKWSClient", "TOFU: first use, pinning leaf %s", leaf_fp.c_str());
      client->stash_pending_leaf(leaf_fp, cert_common_name(crt));
      break;
    case TofuCaptureDecision::kCaptureCa:
      ESP_LOGI("SKWSClient", "TOFU: first use, pinning issuing CA (identity %s)",
               leaf_san.c_str());
      client->set_pending_san(leaf_san);  // bind the leaf identity to the CA
      break;
    case TofuCaptureDecision::kAccept:
      // Leaf matches the stored fingerprint: keep the leaf pin and drop any
      // stray pending state (e.g. a CA stashed at depth > 0 this handshake --
      // mode is fixed at first use, so a later-presented CA is not adopted).
      client->clear_pending_tofu();
      break;
  }
  *flags = 0;
  return 0;
}
 
// Attach function installed for the TLS connection. In CA-anchor mode it
// installs the pinned CA as the mbedTLS trust anchor (esp-tls runs
// VERIFY_REQUIRED, so mbedTLS validates the chain against it); otherwise it
// selects VERIFY_OPTIONAL so the verify callback can capture / fingerprint.
static esp_err_t tofu_crt_bundle_attach(void* conf) {
  mbedtls_ssl_config* ssl_conf = static_cast<mbedtls_ssl_config*>(conf);
  SKWSClient* client = ws_client;
 
  if (client != nullptr && client->is_tofu_enabled() && client->has_tofu_ca()) {
    // Re-parse the stored CA into a static cert that outlives the handshake.
    static mbedtls_x509_crt pinned_ca;
    static bool pinned_ca_inited = false;
    if (pinned_ca_inited) {
      mbedtls_x509_crt_free(&pinned_ca);
    }
    mbedtls_x509_crt_init(&pinned_ca);
    pinned_ca_inited = true;
    const String& pem = client->get_tofu_ca();
    int r = mbedtls_x509_crt_parse(
        &pinned_ca, reinterpret_cast<const unsigned char*>(pem.c_str()),
        pem.length() + 1);
    if (r == 0) {
      mbedtls_ssl_conf_ca_chain(ssl_conf, &pinned_ca, nullptr);
      // esp-tls already set VERIFY_REQUIRED before calling us; keep it.
      ESP_LOGD("SKWSClient", "TOFU: pinned CA installed as trust anchor");
    } else {
      // Stored CA won't parse (corruption/bug). Fail closed: leave
      // VERIFY_REQUIRED with no trust anchor so the handshake is rejected and
      // surfaces as a certificate error requiring a manual reset -- never
      // silently downgrade to accept-any.
      ESP_LOGE("SKWSClient",
               "TOFU: stored CA failed to parse (-0x%x); connections will fail "
               "until reset",
               -r);
    }
  } else {
    // Capture / leaf-fingerprint mode (or TOFU disabled): the callback decides.
    mbedtls_ssl_conf_authmode(ssl_conf, MBEDTLS_SSL_VERIFY_OPTIONAL);
  }
 
  mbedtls_ssl_conf_verify(ssl_conf, tofu_verify_callback, client);
  return ESP_OK;
}
#endif  // SENSESP_SSL_SUPPORT
 
static void websocket_event_handler(void* handler_args,
                                    esp_event_base_t base,
                                    int32_t event_id, void* event_data) {
  // Drop events from a client that has been handed off for destruction: the
  // handler was registered with the generation current at build time; if that no
  // longer matches the live generation, this callback is from a reaped client.
  if (ws_client == nullptr ||
      static_cast<uint32_t>(reinterpret_cast<uintptr_t>(handler_args)) !=
          ws_client->client_generation()) {
    return;
  }
  esp_websocket_event_data_t* data = (esp_websocket_event_data_t*)event_data;
  switch (event_id) {
    case WEBSOCKET_EVENT_CONNECTED:
      ws_client->on_connected();
      break;
    case WEBSOCKET_EVENT_DISCONNECTED:
      ws_client->on_disconnected();
      break;
    case WEBSOCKET_EVENT_DATA:
      // Only process text frames (opcode 0x1) and continuation frames (0x0).
      // Control frames (ping/pong/close: 0x8-0xA) have no JSON payload.
      if (data->op_code <= 0x2) {
        ws_client->on_receive_delta((uint8_t*)data->data_ptr, data->data_len);
      }
      break;
    case WEBSOCKET_EVENT_ERROR:
      // The HTTP status of the failed upgrade (e.g. 401 for a rejected token)
      // lets us distinguish a bad token from a transient transport error. The
      // handshake status field only exists in the newer esp_websocket_client
      // component pulled for SSL builds; the version bundled with the EOL
      // espressif32 Arduino platform lacks it, so fall back to 0 (not
      // applicable), matching how on_error() treats a transient failure.
#ifdef SENSESP_SSL_SUPPORT
      ws_client->on_error(data->error_handle.esp_ws_handshake_status_code);
#else
      ws_client->on_error(0);
#endif
      break;
  }
}
 
SKWSClient::SKWSClient(const String& config_path,
                       std::shared_ptr<SKDeltaQueue> sk_delta_queue,
                       const String& server_address, uint16_t server_port,
                       bool use_mdns)
    : FileSystemSaveable{config_path},
      conf_server_address_{server_address},
      conf_server_port_{server_port},
      use_mdns_{use_mdns},
      sk_delta_queue_{sk_delta_queue} {
  // a SKWSClient object observes its own connection_state_ member
  // and simply passes through any notification it emits. As a result,
  // whenever the value of connection_state_ is updated, observers of the
  // SKWSClient object get automatically notified.
  this->connection_state_.attach(
      [this]() { this->emit(this->connection_state_.get()); });
 
  // process any received updates in the main task
  event_loop()->onRepeat(1, [this]() { this->process_received_updates(); });
 
  // set the singleton object pointer
  ws_client = this;
 
  load();
 
  // Connect the counters
  delta_tx_tick_producer_.connect_to(&delta_tx_count_producer_);
 
  event_loop()->onDelay(0, [this]() {
    ESP_LOGD(__FILENAME__, "Starting SKWSClient");
    // Run the connection lifecycle on the event loop instead of a dedicated
    // task: connect() is a non-blocking dispatcher (it spawns a worker for the
    // blocking auth legs) and send_delta() is non-blocking, so neither stalls
    // the loop. The ~100 ms cadence matches the former task's vTaskDelay(100ms).
    event_loop()->onRepeat(100, [this]() {
      // Retry a teardown whose reaper failed to spawn under OOM (no-op if none).
      reap_async(pending_teardown_.load());
      if (is_connect_due()) {
        connect();
      }
      send_delta();
    });
    MDNS.addService("signalk-sensesp", "tcp", 80);
  });
}
 
void SKWSClient::on_disconnected() {
  this->set_connection_state(SKWSConnectionState::kSKWSDisconnected);
}
 
bool should_clear_token_on_status(bool ssl_enabled, int handshake_status) {
  return ssl_enabled && handshake_status == kHttpUnauthorized;
}
 
void SKWSClient::on_error(int handshake_status) {
  this->set_connection_state(SKWSConnectionState::kSKWSDisconnected);
  if (should_clear_token_on_status(ssl_enabled_, handshake_status)) {
    // The server rejected the token on the WebSocket upgrade (e.g. the server
    // was reinstalled, or the device was moved to a different server). Clear it
    // so the next reconnect requests fresh access, and shorten the backoff so
    // re-authorization starts promptly instead of after the (possibly grown)
    // reconnect interval. A non-401 error (transport, TLS, network) leaves the
    // token intact and simply retries.
    ESP_LOGW(__FILENAME__, "Token rejected (%d), requesting new access",
             handshake_status);
    auth_token_ = NULL_AUTH_TOKEN;
    save();
    reset_reconnect_interval();
  } else {
    ESP_LOGW(__FILENAME__, "Websocket client error.");
  }
}
 
void SKWSClient::on_connected() {
  // The connection is fully established and the server proved possession of its
  // certificate's private key, so it is safe to persist any anchor stashed
  // during this attempt's handshake (first-use capture).
  this->commit_pending_tofu();
  this->set_connection_state(SKWSConnectionState::kSKWSConnected);
  this->reset_reconnect_interval();
  this->sk_delta_queue_->reset_meta_send();
  ESP_LOGI(__FILENAME__, "Subscribing to Signal K listeners...");
  this->subscribe_listeners();
}
 
void SKWSClient::subscribe_listeners() {
  bool output_available = false;
  JsonDocument subscription;
  subscription["context"] = "vessels.self";
 
  SKListener::take_semaphore();
  const std::vector<SKListener*>& listeners = SKListener::get_listeners();
 
  if (listeners.size() > 0) {
    output_available = true;
    JsonArray subscribe = subscription["subscribe"].to<JsonArray>();
 
    // Collapse listeners that share a path into one subscription entry.
    // With sendMeta=all (a connection-level flag), a single subscription
    // delivers both value and meta deltas, so a value listener and a
    // metadata listener on the same path — the common gauge pattern —
    // would otherwise emit two identical entries. Keep the smallest
    // period so the fastest listener's cadence wins.
    std::map<String, int> path_period;
    for (size_t i = 0; i < listeners.size(); i++) {
      auto* listener = listeners.at(i);
      const String& sk_path = listener->get_sk_path();
      int listen_delay = listener->get_listen_delay();
      auto it = path_period.find(sk_path);
      if (it == path_period.end() || listen_delay < it->second) {
        path_period[sk_path] = listen_delay;
      }
    }
    for (const auto& [sk_path, listen_delay] : path_period) {
      JsonObject subscribe_path = subscribe.add<JsonObject>();
      subscribe_path["path"] = sk_path;
      subscribe_path["period"] = listen_delay;
      ESP_LOGI(__FILENAME__, "Adding %s subscription with listen_delay %d\n",
               sk_path.c_str(), listen_delay);
    }
  }
  SKListener::release_semaphore();
 
  if (output_available &&
      get_connection_state() == SKWSConnectionState::kSKWSConnected) {
    String json_message;
 
    serializeJson(subscription, json_message);
    ESP_LOGI(__FILENAME__, "Subscription JSON message:\n %s",
             json_message.c_str());
    int result = esp_websocket_client_send_text(
        client_.load(), json_message.c_str(), json_message.length(),
        kWsSendTimeoutTicks);
    if (result < 0) {
      ESP_LOGE(__FILENAME__, "Subscription send failed (result=%d)", result);
    }
  }
}
 
void SKWSClient::on_receive_delta(uint8_t* payload, size_t length) {
  // Need to work on null-terminated strings
  constexpr size_t kMaxWsMessageSize = 4096;
  if (length > kMaxWsMessageSize) {
    ESP_LOGW(__FILENAME__, "WebSocket message too large (%u bytes), dropping",
             (unsigned)length);
    return;
  }
  std::unique_ptr<char[]> buf(new char[length + 1]);
  memcpy(buf.get(), payload, length);
  buf[length] = 0;
 
#ifdef SIGNALK_PRINT_RCV_DELTA
  ESP_LOGD(__FILENAME__, "Websocket payload received: %s", buf.get());
#endif
 
  JsonDocument message;
  auto error = deserializeJson(message, buf.get());
 
  if (!error) {
    if (message["updates"].is<JsonVariant>()) {
      on_receive_updates(message);
    }
 
    if (message["put"].is<JsonVariant>()) {
      on_receive_put(message);
    }
 
    // Putrequest contains also requestId Key  GA
    if (message["requestId"].is<JsonVariant>() &&
        !message["put"].is<JsonVariant>()) {
      SKRequest::handle_response(message);
    }
  } else {
    ESP_LOGE(__FILENAME__, "deserializeJson error: %s", error.c_str());
  }
}
 
void SKWSClient::on_receive_updates(JsonDocument& message) {
  // Process updates from subscriptions...
  JsonArray updates = message["updates"];
 
  // With sendMeta=all enabled by default, the server pushes meta deltas to
  // every client. Skip copying them onto the queue unless something actually
  // consumes them. Compute this before taking received_updates_semaphore_ to
  // preserve the global lock order (SKListener before received_updates).
  bool has_meta_listener = false;
  SKListener::take_semaphore();
  for (SKListener* listener : SKListener::get_listeners()) {
    if (listener->wants_meta()) {
      has_meta_listener = true;
      break;
    }
  }
  SKListener::release_semaphore();
 
  take_received_updates_semaphore();
  for (size_t i = 0; i < updates.size(); i++) {
    JsonObject update = updates[i];
 
    JsonArray values = update["values"];
 
    for (size_t vi = 0; vi < values.size(); vi++) {
      // Copy each value into an owned document for processing in the main
      // task (decoupled from `message`, which is freed when on_receive_delta
      // returns).
      ReceivedUpdate ru;
      ru.is_meta = false;
      ru.doc.set(values[vi]);
      enqueue_received_update(std::move(ru));
    }
 
    // Meta deltas only arrive when subscribed with sendMeta=all and are
    // typically one-shot per path (at subscribe + on metadata change). Copy
    // each meta entry into an owned document the same way; SKMetadataListener
    // consumers receive it (path-routed) on the main task. No user code runs
    // here, so the critical section stays free of arbitrary callbacks. Skip
    // entirely when no listener consumes meta (see has_meta_listener above).
    if (has_meta_listener) {
      JsonArray meta_entries = update["meta"];
      for (size_t mi = 0; mi < meta_entries.size(); mi++) {
        JsonObject entry = meta_entries[mi];
        if (entry["path"].isNull() || entry["value"].isNull()) continue;
        ReceivedUpdate ru;
        ru.is_meta = true;
        ru.doc.set(entry);  // {path, value: {...meta...}}
        enqueue_received_update(std::move(ru));
      }
    }
  }
  release_received_updates_semaphore();
}
 
void SKWSClient::enqueue_received_update(ReceivedUpdate&& update) {
  // Per-kind caps, tunable via build_flags (see signalk_ws_client.h). Meta and
  // value deltas are budgeted independently so a metadata burst cannot evict
  // pending values, and vice versa.
  constexpr size_t kMaxReceivedValues = SENSESP_MAX_RECEIVED_VALUE_UPDATES;
  constexpr size_t kMaxReceivedMeta = SENSESP_MAX_RECEIVED_META_UPDATES;
 
  const bool is_meta = update.is_meta;
  const size_t cap = is_meta ? kMaxReceivedMeta : kMaxReceivedValues;
 
  size_t kind_count = 0;
  for (const auto& ru : received_updates_) {
    if (ru.is_meta == is_meta) kind_count++;
  }
 
  while (kind_count >= cap) {
    // Drop the oldest entry of the SAME kind, leaving the other kind intact.
    for (auto it = received_updates_.begin(); it != received_updates_.end();
         ++it) {
      if (it->is_meta == is_meta) {
        received_updates_.erase(it);
        kind_count--;
        break;
      }
    }
    ESP_LOGW(__FILENAME__, "Dropping oldest received %s update (queue full)",
             is_meta ? "meta" : "value");
  }
 
  received_updates_.push_back(std::move(update));
}
 
void SKWSClient::process_received_updates() {
  SKListener::take_semaphore();
 
  const std::vector<SKListener*>& listeners = SKListener::get_listeners();
  const std::vector<SKPutListener*>& put_listeners =
      SKPutListener::get_listeners();
 
  take_received_updates_semaphore();
  // Count only value/put deltas toward the rx metric; meta deltas are
  // low-frequency one-shots and would otherwise inflate it.
  int num_updates = 0;
  while (!received_updates_.empty()) {
    ReceivedUpdate& ru = received_updates_.front();
 
    if (ru.is_meta) {
      // Capture the path into an owned String before moving the document: the
      // const char* would dangle once ru.doc is moved into the shared_ptr.
      String path = ru.doc["path"].as<String>();
      // Move the already-owned queue document into a refcounted, read-only
      // shared_ptr (no extra deep copy) so it safely outlives the queue entry
      // and any deferred consumer, then fan it out to matching listeners.
      std::shared_ptr<const JsonDocument> meta_doc =
          std::make_shared<const JsonDocument>(std::move(ru.doc));
      for (size_t i = 0; i < listeners.size(); i++) {
        SKListener* listener = listeners[i];
        if (listener->wants_meta() && listener->matches(path)) {
          listener->parse_meta(meta_doc);
        }
      }
    } else {
      num_updates++;
      const char* path = ru.doc["path"];
      JsonObject value = ru.doc.as<JsonObject>();
 
      for (size_t i = 0; i < listeners.size(); i++) {
        SKListener* listener = listeners[i];
        if (!listener->wants_meta() && listener->matches(path)) {
          listener->parse_value(value);
        }
      }
      //   to be able to parse values of Put Listeners    GA
      for (size_t i = 0; i < put_listeners.size(); i++) {
        SKPutListener* listener = put_listeners[i];
        if (listener->get_sk_path().equals(path)) {
          listener->parse_value(value);
        }
      }
    }
    received_updates_.pop_front();
  }
  release_received_updates_semaphore();
  delta_rx_count_producer_.set(num_updates);
 
  SKListener::release_semaphore();
}
 
void SKWSClient::on_receive_put(JsonDocument& message) {
  // Process PUT requests...
  JsonArray puts = message["put"];
  bool all_matched = true;
  for (size_t i = 0; i < puts.size(); i++) {
    JsonObject value = puts[i];
    const char* path = value["path"];
    bool matched = false;
 
    SKListener::take_semaphore();
    const std::vector<SKPutListener*>& listeners =
        SKPutListener::get_listeners();
    for (size_t j = 0; j < listeners.size(); j++) {
      SKPutListener* listener = listeners[j];
      if (listener->get_sk_path().equals(path)) {
        ReceivedUpdate ru;
        ru.is_meta = false;
        ru.doc.set(value);
        take_received_updates_semaphore();
        enqueue_received_update(std::move(ru));
        release_received_updates_semaphore();
        matched = true;
      }
    }
    SKListener::release_semaphore();
 
    if (!matched) {
      all_matched = false;
    }
  }
 
  // Send back a single request response if still connected
  if (get_connection_state() == SKWSConnectionState::kSKWSConnected) {
    JsonDocument put_response;
    put_response["requestId"] = message["requestId"];
    if (all_matched) {
      put_response["state"] = "COMPLETED";
      put_response["statusCode"] = 200;
    } else {
      put_response["state"] = "FAILED";
      put_response["statusCode"] = 405;
    }
    String response_text;
    serializeJson(put_response, response_text);
    int result = esp_websocket_client_send_text(
        client_.load(), response_text.c_str(), response_text.length(),
        kWsSendTimeoutTicks);
    if (result < 0) {
      ESP_LOGE(__FILENAME__, "PUT response send failed (result=%d)", result);
    }
  }
}
 
void SKWSClient::sendTXT(String& payload) {
  if (get_connection_state() == SKWSConnectionState::kSKWSConnected) {
    int result = esp_websocket_client_send_text(
        client_.load(), payload.c_str(), payload.length(), kWsSendTimeoutTicks);
    if (result < 0) {
      ESP_LOGE(__FILENAME__, "sendTXT failed (result=%d)", result);
    }
  }
}
 
bool SKWSClient::get_mdns_service(String& server_address,
                                  uint16_t& server_port) {
  // get IP address using an mDNS query
  // Try SSL service first, then fall back to non-SSL
  int num = MDNS.queryService("signalk-wss", "tcp");
  if (num > 0) {
    // Found SSL-enabled server
    ssl_enabled_ = true;
    ESP_LOGI(__FILENAME__, "Found Signal K server via mDNS (signalk-wss)");
  } else {
    // Try non-SSL service
    num = MDNS.queryService("signalk-ws", "tcp");
    if (num == 0) {
      // no service found
      return false;
    }
    // Found non-SSL server, disable SSL
    ssl_enabled_ = false;
    ESP_LOGI(__FILENAME__, "Found Signal K server via mDNS (signalk-ws)");
  }
 
#if ESP_ARDUINO_VERSION_MAJOR < 3
  server_address = MDNS.IP(0).toString();
#else
  server_address = MDNS.address(0).toString();
#endif
  server_port = MDNS.port(0);
  ESP_LOGI(__FILENAME__, "Found server %s (port %d)", server_address.c_str(),
           server_port);
  return true;
}
 
// Event handler for detect_ssl() to capture the Location response header
static esp_err_t detect_ssl_event_handler(esp_http_client_event_t* evt) {
  if (evt->event_id == HTTP_EVENT_ON_HEADER) {
    // Check for Location header (case-insensitive)
    if (strcasecmp(evt->header_key, "Location") == 0) {
      String* location = static_cast<String*>(evt->user_data);
      *location = evt->header_value;
    }
  }
  return ESP_OK;
}
 
bool SKWSClient::detect_ssl() {
  // Try to detect if the server requires SSL by checking for HTTP->HTTPS
  // redirects
  String url =
      String("http://") + server_address_ + ":" + server_port_ + "/signalk";
 
  ESP_LOGD(__FILENAME__, "Probing for SSL redirect at %s", url.c_str());
 
  String location;
 
  esp_http_client_config_t config = {};
  config.url = url.c_str();
  config.disable_auto_redirect = true;
  config.timeout_ms = 10000;
  config.event_handler = detect_ssl_event_handler;
  config.user_data = &location;
 
  esp_http_client_handle_t client = esp_http_client_init(&config);
  if (client == nullptr) {
    ESP_LOGE(__FILENAME__, "Failed to initialize HTTP client");
    return false;
  }
 
  esp_err_t err = esp_http_client_perform(client);
  int status_code = esp_http_client_get_status_code(client);
  esp_http_client_cleanup(client);
 
  if (err != ESP_OK) {
    ESP_LOGD(__FILENAME__, "HTTP request failed: %s", esp_err_to_name(err));
    return false;
  }
 
  if ((status_code == 301 || status_code == 302 ||
       status_code == 307 || status_code == 308) &&
      location.startsWith("https://")) {
    ESP_LOGI(__FILENAME__, "SSL redirect detected, enabling HTTPS/WSS");
    ssl_enabled_ = true;
    save();
    return true;
  }
 
  return false;
}
 
 
void SKWSClient::connect() {
  // A prior certificate rejection leaves the state in kSKWSCertificateError;
  // treat it like a disconnect for retry purposes so the device keeps trying
  // (and re-surfaces the cert error each failed attempt).
  if (get_connection_state() != SKWSConnectionState::kSKWSDisconnected &&
      get_connection_state() != SKWSConnectionState::kSKWSCertificateError) {
    return;
  }
 
  // A connect attempt is already running on a worker task; let it finish.
  if (auth_job_running_.load()) {
    return;
  }
 
  // Reap any client left from a previous attempt before starting a new one,
  // off this context. While the reap is in flight, defer bring-up so at most one
  // client ever exists; state stays Disconnected, so a later cycle retries.
  if (client_.load() != nullptr) {
    detach_teardown();
  }
  if (teardown_in_progress_.load()) {
    return;
  }
 
  // Discard any anchor candidate stashed by a previous attempt; it is only
  // committed after a fully successful connection (see on_connected). Clearing
  // here also prevents committing stale data if a resumed TLS session skips the
  // certificate callback.
  clear_pending_tofu();
 
  // Schedule next attempt with backoff in case this one fails.
  // Will be reset on successful connection.
  schedule_reconnect();
 
  // Wait for the active network provisioner (WiFi, Ethernet, …) to be
  // up before initiating the WS connection. The provisioner abstracts
  // away whether we're on WiFi, Ethernet, or some other transport.
  auto provisioner = SensESPApp::get()->get_network_provisioner();
  if (!provisioner || !provisioner->is_connected()) {
    ESP_LOGI(__FILENAME__,
             "Network is not yet up. SignalK client connection will be "
             "initiated when the link comes up.");
    return;
  }
 
  set_connection_state(SKWSConnectionState::kSKWSAuthorizing);
 
  // The rest of the attempt — mDNS resolve, SSL detect, and the access-request /
  // poll / connect_ws leg — makes blocking HTTP/mDNS calls, so run it on a
  // one-shot worker task. The SK/event-loop context stays responsive, and
  // auth_job_running_ keeps at most one attempt in flight.
  auth_job_running_.store(true);
  if (xTaskCreate(&SKWSClient::connect_worker, "SKWSConnect",
                  kWsClientTaskStackSize, this, 1, nullptr) != pdPASS) {
    ESP_LOGE(__FILENAME__, "connect worker spawn failed");
    auth_job_running_.store(false);
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
  }
}
 
void SKWSClient::connect_worker(void* arg) {
  auto* self = static_cast<SKWSClient*>(arg);
  self->run_connect_attempt();
  self->auth_job_running_.store(false);
  vTaskDelete(nullptr);
}
 
void SKWSClient::run_connect_attempt() {
  ESP_LOGI(__FILENAME__, "Initiating websocket connection with server...");
 
  if (use_mdns_) {
    if (!get_mdns_service(this->server_address_, this->server_port_)) {
      ESP_LOGE(__FILENAME__,
               "No Signal K server found in network when using mDNS service!");
    } else {
      ESP_LOGI(__FILENAME__,
               "Signal K server has been found at address %s:%d by mDNS.",
               this->server_address_.c_str(), this->server_port_);
    }
  } else {
    this->server_address_ = this->conf_server_address_;
    this->server_port_ = this->conf_server_port_;
  }
 
  if (!this->server_address_.isEmpty() && this->server_port_ > 0) {
    ESP_LOGD(__FILENAME__,
             "Websocket is connecting to Signal K server on address %s:%d",
             this->server_address_.c_str(), this->server_port_);
 
    // Detect if server requires SSL (check for HTTP->HTTPS redirects)
    if (!ssl_enabled_) {
      detect_ssl();
    }
  } else {
    // host and port not defined - don't try to connect
    ESP_LOGD(__FILENAME__,
             "Websocket is not connecting to Signal K server because host and "
             "port are not defined.");
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  if (this->polling_href_.length() > 0 && this->polling_href_.startsWith("/")) {
    // existing pending request
    this->poll_access_request(this->server_address_, this->server_port_,
                              this->polling_href_);
    return;
  }
 
  if (this->auth_token_ == NULL_AUTH_TOKEN) {
    // initiate HTTP authentication
    ESP_LOGD(__FILENAME__, "No prior authorization token present.");
    this->send_access_request(this->server_address_, this->server_port_);
    return;
  }
 
  // A token is already present. Validate it before streaming.
#ifdef SENSESP_SSL_SUPPORT
  // Connect the WebSocket directly rather than first probing the token over a
  // separate HTTPS request: on memory-constrained targets (e.g. ESP32-C3) the
  // back-to-back token-probe TLS handshake and the WebSocket TLS handshake
  // fragment the heap, and the second fails to allocate
  // (MBEDTLS_ERR_SSL_ALLOC_FAILED). The server validates the token on the
  // upgrade itself; a 401 there is handled in on_error() (clears the token and
  // re-requests access on the next reconnect).
  this->connect_ws(this->server_address_, this->server_port_);
#else
  // The bundled (non-SSL) esp_websocket_client reports no upgrade status, so
  // on_error() cannot tell a rejected token from a transient failure. Probe the
  // token over plain HTTP first -- there is no TLS handshake to fragment the
  // heap. A 401 there clears the token and re-requests access.
  this->test_token(this->server_address_, this->server_port_);
#endif
}
 
void SKWSClient::loop() {
  // No-op: esp_websocket_client handles data via event callbacks
}
 
#ifndef SENSESP_SSL_SUPPORT
void SKWSClient::test_token(const String server_address,
                            const uint16_t server_port) {
  String url = String("http://") + server_address + ":" + server_port +
               "/signalk/v1/stream";
  ESP_LOGD(__FILENAME__, "Testing token with url %s", url.c_str());
 
  const String full_token = String("Bearer ") + auth_token_;
  ESP_LOGD(__FILENAME__, "Authorization: %.8s...[redacted]", full_token.c_str());
 
  esp_http_client_config_t config = {};
  config.url = url.c_str();
  config.timeout_ms = 10000;
 
  esp_http_client_handle_t client = esp_http_client_init(&config);
  if (client == nullptr) {
    ESP_LOGE(__FILENAME__, "Failed to initialize HTTP client");
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  esp_http_client_set_header(client, "Authorization", full_token.c_str());
 
  // Use streaming API for GET request
  esp_err_t err = esp_http_client_open(client, 0);
  if (err != ESP_OK) {
    ESP_LOGE(__FILENAME__, "Failed to open HTTP connection: %s",
             esp_err_to_name(err));
    esp_http_client_cleanup(client);
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  int content_length = esp_http_client_fetch_headers(client);
  int http_code = esp_http_client_get_status_code(client);
 
  ESP_LOGD(__FILENAME__, "Testing resulted in http status %d", http_code);
 
  // Read response body
  String payload;
  if (content_length > 0 && content_length < 4096) {
    char* buffer = new char[content_length + 1];
    int read_len = esp_http_client_read(client, buffer, content_length);
    buffer[read_len > 0 ? read_len : 0] = '\0';
    payload = String(buffer);
    delete[] buffer;
  } else {
    // Chunked encoding or unknown/large content length - read in chunks
    char buffer[512];
    int read_len;
    while ((read_len = esp_http_client_read(client, buffer,
                                            sizeof(buffer) - 1)) > 0) {
      buffer[read_len] = '\0';
      payload += String(buffer);
      if (payload.length() > 4096) break;
    }
  }
 
  esp_http_client_close(client);
  esp_http_client_cleanup(client);
 
  if (payload.length() > 0) {
    ESP_LOGD(__FILENAME__, "Returned payload (%d bytes): %s", payload.length(),
             payload.c_str());
  }
 
  if (http_code == 426) {
    // HTTP status 426 is "Upgrade Required", the expected response for a
    // websocket endpoint reached over plain HTTP: the token is valid.
    ESP_LOGD(__FILENAME__, "Attempting to connect to Signal K Websocket...");
    this->connect_ws(server_address, server_port);
  } else if (http_code == kHttpUnauthorized) {
    // Token is invalid/expired - clear it and request new access.
    // Keep client_id_ so we reuse the same device identity.
    ESP_LOGW(__FILENAME__, "Token rejected (401), requesting new access");
    this->auth_token_ = NULL_AUTH_TOKEN;
    this->save();
    this->send_access_request(server_address, server_port);
  } else if (http_code > 0) {
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
  } else {
    ESP_LOGE(__FILENAME__, "HTTP request failed with code %d", http_code);
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
  }
}
#endif  // !SENSESP_SSL_SUPPORT
 
void SKWSClient::send_access_request(const String server_address,
                                     const uint16_t server_port) {
  ESP_LOGD(__FILENAME__, "Sending access request (client_id=%s, ssl=%d)",
           client_id_.c_str(), ssl_enabled_);
  if (client_id_ == "") {
    // generate a client ID
    client_id_ = generate_uuid4();
    save();
  }
 
  // create a new access request
  JsonDocument doc;
  doc["clientId"] = client_id_;
  doc["description"] =
      String("SensESP device: ") + SensESPBaseApp::get_hostname();
  doc["permissions"] = kRequestPermission;
  String json_req = "";
  serializeJson(doc, json_req);
 
  ESP_LOGD(__FILENAME__, "Access request: %s", json_req.c_str());
 
  String protocol = ssl_enabled_ ? "https://" : "http://";
  String url = protocol + server_address + ":" + server_port +
               "/signalk/v1/access/requests";
  ESP_LOGD(__FILENAME__, "Access request url: %s", url.c_str());
 
  esp_http_client_config_t config = {};
  config.url = url.c_str();
  config.method = HTTP_METHOD_POST;
  config.timeout_ms = 10000;
#ifdef SENSESP_SSL_SUPPORT
  if (ssl_enabled_) {
    config.crt_bundle_attach = tofu_crt_bundle_attach;
    config.skip_cert_common_name_check = true;
  }
#endif
 
  esp_http_client_handle_t client = esp_http_client_init(&config);
  if (client == nullptr) {
    ESP_LOGE(__FILENAME__, "Failed to initialize HTTP client");
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    // Don't clear client_id_ - keep device identity for retry
    return;
  }
 
  esp_http_client_set_header(client, "Content-Type", "application/json");
 
  // Use streaming API: open -> write request -> fetch headers -> read response
  esp_err_t err = esp_http_client_open(client, json_req.length());
  if (err != ESP_OK) {
    ESP_LOGE(__FILENAME__, "Failed to open HTTP connection: %s", esp_err_to_name(err));
    esp_http_client_cleanup(client);
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  int write_len = esp_http_client_write(client, json_req.c_str(), json_req.length());
  if (write_len < 0 || write_len != (int)json_req.length()) {
    ESP_LOGE(__FILENAME__, "Failed to write request body (wrote %d of %d bytes)",
             write_len, json_req.length());
    esp_http_client_close(client);
    esp_http_client_cleanup(client);
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  int content_length = esp_http_client_fetch_headers(client);
  int http_code = esp_http_client_get_status_code(client);
 
  ESP_LOGD(__FILENAME__, "HTTP response: code=%d, content_length=%d", http_code, content_length);
 
  // Read response body
  String payload;
  char buffer[512];
  int read_len;
  while ((read_len = esp_http_client_read(client, buffer, sizeof(buffer) - 1)) > 0) {
    buffer[read_len] = '\0';
    payload += String(buffer);
    if (payload.length() > 4096) break;
  }
  ESP_LOGD(__FILENAME__, "Response payload (%d bytes): %s",
           payload.length(), payload.c_str());
 
  esp_http_client_close(client);
  esp_http_client_cleanup(client);
 
  // Parse JSON response for both 202 and 400 status codes
  deserializeJson(doc, payload.c_str());
  String state = doc["state"].is<const char*>() ? doc["state"].as<String>() : "";
  String href = doc["href"].is<const char*>() ? doc["href"].as<String>() : "";
  String message = doc["message"].is<const char*>() ? doc["message"].as<String>() : "";
 
  ESP_LOGD(__FILENAME__, "Access request response: http=%d, state=%s, href=%s",
           http_code, state.c_str(), href.c_str());
  if (message.length() > 0) {
    ESP_LOGI(__FILENAME__, "Server message: %s", message.c_str());
  }
 
  // HTTP 400 with href means "already requested" - save href for polling on
  // next connect() cycle (after backoff)
  if (http_code == 400 && href.length() > 0 && href.startsWith("/")) {
    ESP_LOGI(__FILENAME__, "Existing request found, will poll href: %s", href.c_str());
    polling_href_ = href;
    save();
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  // HTTP 202 with href means new request pending - save href for polling on
  // next connect() cycle (after backoff)
  if (http_code == 202 && href.length() > 0 && href.startsWith("/")) {
    polling_href_ = href;
    save();
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  // HTTP 404 means the server has no security enabled — access requests are
  // not available. Connect without a token.
  if (http_code == 404) {
    ESP_LOGI(__FILENAME__,
             "Server security disabled (404 on access request) — connecting "
             "without token");
    auth_token_ = NULL_AUTH_TOKEN;
    this->connect_ws(server_address, server_port);
    return;
  }
 
  // Can't proceed - disconnect and retry later
  ESP_LOGW(__FILENAME__, "Cannot handle response: http=%d, state=%s", http_code, state.c_str());
  set_connection_state(SKWSConnectionState::kSKWSDisconnected);
}
 
void SKWSClient::poll_access_request(const String server_address,
                                     const uint16_t server_port,
                                     const String href) {
  ESP_LOGD(__FILENAME__, "Polling SK Server for authentication token");
 
  String protocol = ssl_enabled_ ? "https://" : "http://";
  String url = protocol + server_address + ":" + server_port + href;
 
  esp_http_client_config_t config = {};
  config.url = url.c_str();
  config.timeout_ms = 10000;
#ifdef SENSESP_SSL_SUPPORT
  if (ssl_enabled_) {
    config.crt_bundle_attach = tofu_crt_bundle_attach;
    config.skip_cert_common_name_check = true;
  }
#endif
 
  esp_http_client_handle_t client = esp_http_client_init(&config);
  if (client == nullptr) {
    ESP_LOGE(__FILENAME__, "Failed to initialize HTTP client");
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  // Use streaming API for GET request
  esp_err_t err = esp_http_client_open(client, 0);
  if (err != ESP_OK) {
    ESP_LOGE(__FILENAME__, "Failed to open HTTP connection: %s", esp_err_to_name(err));
    esp_http_client_cleanup(client);
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  int content_length = esp_http_client_fetch_headers(client);
  int http_code = esp_http_client_get_status_code(client);
 
  // Read response body
  String payload;
  if (content_length > 0 && content_length < 4096) {
    char* buffer = new char[content_length + 1];
    int read_len = esp_http_client_read(client, buffer, content_length);
    buffer[read_len > 0 ? read_len : 0] = '\0';
    payload = String(buffer);
    delete[] buffer;
  } else {
    // Chunked encoding or unknown/large content length - read in chunks
    char buffer[512];
    int read_len;
    while ((read_len = esp_http_client_read(client, buffer, sizeof(buffer) - 1)) > 0) {
      buffer[read_len] = '\0';
      payload += String(buffer);
      if (payload.length() > 4096) break;
    }
  }
 
  // An APPROVED poll response carries the access token in its body, so log only
  // the status and size, never the payload itself. The web log buffer exposes
  // captured log lines over HTTP, so a payload dump here would leak the token.
  ESP_LOGD(__FILENAME__, "Poll response: http=%d, %d bytes", http_code,
           static_cast<int>(payload.length()));
 
  esp_http_client_close(client);
  esp_http_client_cleanup(client);
 
  if (http_code == 200 || http_code == 202) {
    JsonDocument doc;
    auto error = deserializeJson(doc, payload.c_str());
    if (error) {
      ESP_LOGW(__FILENAME__, "WARNING: Could not deserialize http payload.");
      ESP_LOGW(__FILENAME__, "DeserializationError: %s", error.c_str());
      set_connection_state(SKWSConnectionState::kSKWSDisconnected);
      return;
    }
    String state = doc["state"];
    ESP_LOGD(__FILENAME__, "%s", state.c_str());
    if (state == "PENDING") {
      set_connection_state(SKWSConnectionState::kSKWSDisconnected);
      return;
    }
    if (state == "COMPLETED") {
      JsonObject access_req = doc["accessRequest"];
      String permission = access_req["permission"];
 
      polling_href_ = "";
      save();
 
      if (permission == "DENIED") {
        ESP_LOGW(__FILENAME__, "Permission denied");
        set_connection_state(SKWSConnectionState::kSKWSDisconnected);
        return;
      }
 
      if (permission == "APPROVED") {
        ESP_LOGI(__FILENAME__, "Permission granted");
        String token = access_req["token"];
        auth_token_ = token;
        save();
        this->connect_ws(server_address, server_port);
        return;
      }
    }
  } else {
    if (http_code == 404 || http_code == 500) {
      // Server doesn't recognize this request (stale href after
      // server restart, different server, or security disabled).
      // Clear the polling href so the next connect cycle starts
      // a fresh access-request flow.
      ESP_LOGD(__FILENAME__,
               "Got %d polling access request — clearing stale href.",
               http_code);
      polling_href_ = "";
      save();
      set_connection_state(SKWSConnectionState::kSKWSDisconnected);
      return;
    }
    // any other HTTP status code
    ESP_LOGW(__FILENAME__,
             "Can't handle response %d to pending access request.\n",
             http_code);
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
  // Catch-all: a 200/202 COMPLETED whose permission is neither APPROVED nor
  // DENIED, or any unexpected state, leaves no terminal state. With no live
  // client, no event will move us off Authorizing, so fall back to Disconnected
  // to retry rather than wedge.
  set_connection_state(SKWSConnectionState::kSKWSDisconnected);
}
 
void SKWSClient::connect_ws(const String& host, const uint16_t port) {
  // connect() reaps any prior client before dispatching here, so client_ is
  // null and no teardown is in flight. Guard defensively against a leak.
  if (client_.load() != nullptr || teardown_in_progress_.load()) {
    ESP_LOGW(__FILENAME__, "connect_ws: prior client not reaped; deferring");
    detach_teardown();
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  // Discard any anchor candidate stashed by the earlier esp_http_client legs
  // (token check / access request). Only the WebSocket handshake -- the one
  // whose success reaches on_connected and proves the server holds the leaf's
  // private key -- may populate the anchor that gets committed.
  clear_pending_tofu();
  set_connection_state(SKWSConnectionState::kSKWSConnecting);
 
  String protocol = ssl_enabled_ ? "wss" : "ws";
  String path = "/signalk/v1/stream?subscribe=none";
  if (send_meta_enabled_) path += "&sendMeta=all";
  String url = protocol + "://" + host + ":" + String(port) + path;
 
  ESP_LOGD(__FILENAME__, "Connecting WebSocket to %s", url.c_str());
 
  // Build authorization header string (must persist through init call)
  String auth_header;
  if (auth_token_ != NULL_AUTH_TOKEN) {
    auth_header = String("Authorization: Bearer ") + auth_token_ + "\r\n";
  }
 
  // Configure WebSocket client
  esp_websocket_client_config_t config = {};
  config.uri = url.c_str();
  config.task_stack = kWsTransportTaskStackSize;
  config.buffer_size = SENSESP_SK_WS_BUFFER_SIZE;
  if (auth_header.length() > 0) {
    config.headers = auth_header.c_str();
  }
 
#ifdef SENSESP_SSL_SUPPORT
  if (ssl_enabled_) {
    // Use custom crt_bundle_attach to disable SSL verification
    // This directly configures mbedTLS to skip certificate verification
    config.crt_bundle_attach = tofu_crt_bundle_attach;
    config.skip_cert_common_name_check = true;
  }
#endif
 
  esp_websocket_client_handle_t h = esp_websocket_client_init(&config);
  if (h == nullptr) {
    ESP_LOGE(__FILENAME__, "Failed to initialize WebSocket client");
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
  client_.store(h);
 
  // Register the event handler tagged with the current generation, so that any
  // late event from this client after it is later handed off for destruction is
  // dropped by websocket_event_handler (generation mismatch).
  esp_websocket_register_events(
      h, WEBSOCKET_EVENT_ANY, websocket_event_handler,
      reinterpret_cast<void*>(
          static_cast<uintptr_t>(client_generation_.load())));
 
  // Start the client
  esp_err_t err = esp_websocket_client_start(h);
  if (err != ESP_OK) {
    ESP_LOGE(__FILENAME__, "Failed to start WebSocket client: %s",
             esp_err_to_name(err));
    // Null the shared handle before freeing it so a concurrent send sees null.
    client_.store(nullptr);
    esp_websocket_client_destroy(h);
    set_connection_state(SKWSConnectionState::kSKWSDisconnected);
    return;
  }
 
  ESP_LOGD(__FILENAME__, "WebSocket client started, waiting for connection...");
}
 
bool SKWSClient::is_connected() {
  return get_connection_state() == SKWSConnectionState::kSKWSConnected;
}
 
namespace {
struct WsTeardownArg {
  esp_websocket_client_handle_t handle;
  SKWSClient* self;
};
}  // namespace
 
void SKWSClient::teardown_task(void* arg) {
  auto* a = static_cast<WsTeardownArg*>(arg);
  // Blocking: esp_websocket_client_stop() waits for the client's task to exit
  // (up to one ~1 s poll cycle), destroy() frees its buffers. Run here, on a
  // throwaway task, so it never blocks the connect/event-loop context.
  esp_websocket_client_stop(a->handle);
  esp_websocket_client_destroy(a->handle);
  a->self->teardown_in_progress_.store(false);
  delete a;
  vTaskDelete(nullptr);
}
 
void SKWSClient::reap_async(esp_websocket_client_handle_t old) {
  if (old == nullptr) {
    return;
  }
  auto* arg = new WsTeardownArg{old, this};
  if (xTaskCreate(&SKWSClient::teardown_task, "SKWSTeardown", 4096, arg, 1,
                  nullptr) == pdPASS) {
    // The task now owns `old` and clears teardown_in_progress_ when done.
    pending_teardown_.store(nullptr);
    return;
  }
  // Spawn failed (OOM). Do NOT reap synchronously: a ~1 s stop()+destroy() on
  // the event loop would stall every consumer under the exact heap pressure
  // this path exists for. Stash the handle and retry on the next loop tick;
  // teardown_in_progress_ stays set so bring-up remains deferred (at most one
  // un-reaped client, no leak beyond it).
  delete arg;
  pending_teardown_.store(old);
  ESP_LOGW(__FILENAME__, "teardown task spawn failed; will retry next cycle");
}
 
void SKWSClient::detach_teardown() {
  // Single atomic check-and-null: if two contexts race here, only one gets the
  // handle, so it is stopped/destroyed exactly once (no double-free).
  esp_websocket_client_handle_t old = client_.exchange(nullptr);
  if (old == nullptr) {
    return;
  }
  // Invalidate the old client's generation so any late event it dispatches
  // while being reaped is dropped by websocket_event_handler.
  client_generation_.fetch_add(1);
  teardown_in_progress_.store(true);
  reap_async(old);
}
 
void SKWSClient::restart() {
  // Set state first so event handler callbacks and send callsites see the
  // disconnected state and skip operations on the client being destroyed.
  set_connection_state(SKWSConnectionState::kSKWSDisconnected);
  detach_teardown();
}
 
void SKWSClient::send_delta() {
  if (get_connection_state() == SKWSConnectionState::kSKWSConnected) {
    if (sk_delta_queue_->data_available()) {
      std::vector<String> deltas;
      sk_delta_queue_->get_deltas(deltas);
      bool first = true;
      for (const auto& delta : deltas) {
        if (sk_delta_exceeds_ws_buffer(delta.length(),
                                       SENSESP_SK_WS_BUFFER_SIZE)) {
          // Drop the delta to keep the connection alive (signalk_ws_delta_size.h
          // explains why an oversize delta would otherwise abort it). Unlike the
          // transient send failure below, an oversize delta is deterministic, so
          // do NOT re-arm metadata here: get_deltas() already bundles metadata
          // into the first delta and marks it sent, and re-arming would have
          // get_deltas() rebuild the same oversize first delta every cycle --
          // dropped and re-armed forever, never delivered. Leaving it sent lets
          // the next, metadata-free first delta fit and flow; metadata waits for
          // a reconnect or a larger SENSESP_SK_WS_BUFFER_SIZE.
          uint32_t now = millis();
          if (last_oversize_log_ms_ == 0 ||
              now - last_oversize_log_ms_ >= kOversizeDropLogIntervalMs) {
            ESP_LOGW(__FILENAME__,
                     "Delta too large (%u B > %u buffer); dropped to keep the "
                     "connection alive -- raise SENSESP_SK_WS_BUFFER_SIZE",
                     (unsigned)delta.length(),
                     (unsigned)SENSESP_SK_WS_BUFFER_SIZE);
            last_oversize_log_ms_ = now;
          }
          first = false;
          continue;
        }
        int send_result = esp_websocket_client_send_text(
            client_.load(), delta.c_str(), delta.length(), kWsDeltaSendTimeoutTicks);
        if (send_result < 0) {
          // Non-blocking send (0 timeout) did not complete: either brief
          // ws-client lock contention (retry next cycle) or the transport
          // backpressured and esp_websocket_client aborted the connection
          // internally -- its disconnect/error event drives reconnect. Never
          // block or tear the connection down from here. Deltas are
          // supersedable, so drop the rest of this batch. See SignalK/SensESP#1033.
          if (first) {
            // get_deltas() builds one-shot metadata (units, zones, ...) into the
            // first delta and marks it sent before it leaves the device. The
            // first delta is the only one that can carry that metadata, so if its
            // send is the one that fails, re-arm metadata for the next batch --
            // otherwise the server runs without it until the next reconnect.
            sk_delta_queue_->reset_meta_send();
          }
          ESP_LOGW(__FILENAME__,
                   "Delta send incomplete (result=%d); dropping rest of batch",
                   send_result);
          break;
        }
        this->delta_tx_tick_producer_.set(1);
        first = false;
      }
    }
  }
}
 
bool SKWSClient::to_json(JsonObject& root) {
  root["sk_address"] = this->conf_server_address_;
  root["sk_port"] = this->conf_server_port_;
  root["use_mdns"] = this->use_mdns_;
 
  root["token"] = this->auth_token_;
  root["client_id"] = this->client_id_;
  root["polling_href"] = this->polling_href_;
 
  root["ssl_enabled"] = this->ssl_enabled_;
  root["tofu_enabled"] = this->tofu_enabled_;
  // Persisted trust anchor: leaf fingerprint (legacy / leaf mode) and/or CA PEM.
  root["tofu_fingerprint"] = this->tofu_fingerprint_;
  root["tofu_ca_pem"] = this->tofu_ca_pem_;
  root["tofu_san"] = this->tofu_san_;
  // Read-only display fields for the pinned identity.
  root["tofu_pin_cn"] = this->tofu_pin_cn_;
  root["tofu_pin_is_ca"] = this->tofu_pin_is_ca_;
  root["send_meta_enabled"] = this->send_meta_enabled_;
  return true;
}
 
bool SKWSClient::from_json(const JsonObject& config) {
  if (config["sk_address"].is<String>()) {
    this->conf_server_address_ = config["sk_address"].as<String>();
  }
  if (config["sk_port"].is<int>()) {
    this->conf_server_port_ = config["sk_port"].as<int>();
  }
  if (config["use_mdns"].is<bool>()) {
    this->use_mdns_ = config["use_mdns"].as<bool>();
  }
  if (config["token"].is<String>()) {
    this->auth_token_ = config["token"].as<String>();
  }
  if (config["client_id"].is<String>()) {
    this->client_id_ = config["client_id"].as<String>();
  }
  if (config["polling_href"].is<String>()) {
    String href = config["polling_href"].as<String>();
    // Only accept valid hrefs (must start with /)
    this->polling_href_ = href.startsWith("/") ? href : "";
  }
 
  if (config["ssl_enabled"].is<bool>()) {
    this->ssl_enabled_ = config["ssl_enabled"].as<bool>();
  }
  if (config["tofu_enabled"].is<bool>()) {
    this->tofu_enabled_ = config["tofu_enabled"].as<bool>();
  }
  // A legacy config carries only tofu_fingerprint (loads as leaf mode — the
  // migration entry point); newer configs may also carry a pinned CA and the
  // display fields. Tolerate both.
  if (config["tofu_fingerprint"].is<String>()) {
    this->tofu_fingerprint_ = config["tofu_fingerprint"].as<String>();
  }
  if (config["tofu_ca_pem"].is<String>()) {
    this->tofu_ca_pem_ = config["tofu_ca_pem"].as<String>();
  }
  if (config["tofu_san"].is<String>()) {
    this->tofu_san_ = config["tofu_san"].as<String>();
  }
  if (config["tofu_pin_cn"].is<String>()) {
    this->tofu_pin_cn_ = config["tofu_pin_cn"].as<String>();
  }
  if (config["tofu_pin_is_ca"].is<bool>()) {
    this->tofu_pin_is_ca_ = config["tofu_pin_is_ca"].as<bool>();
  }
  if (config["send_meta_enabled"].is<bool>()) {
    this->send_meta_enabled_ = config["send_meta_enabled"].as<bool>();
  }
 
  return true;
}
 
String SKWSClient::get_connection_status() {
  auto state = get_connection_state();
  switch (state) {
    case SKWSConnectionState::kSKWSAuthorizing:
      return "Authorizing with SignalK";
    case SKWSConnectionState::kSKWSConnected:
      return "Connected";
    case SKWSConnectionState::kSKWSConnecting:
      return "Connecting";
    case SKWSConnectionState::kSKWSDisconnected:
      return "Disconnected";
    case SKWSConnectionState::kSKWSCertificateError:
      return "Certificate verification failed";
  }
 
  return "Unknown";
}
 
}  // namespace sensesp