#include "session_broker.h"

#include <string.h>

#include "esp_heap_caps.h"
#include "freertos/FreeRTOS.h"
#include "freertos/queue.h"
#include "freertos/semphr.h"
#include "freertos/stream_buffer.h"
#include "freertos/task.h"
#include "serial_service.h"

#define SESSION_BROKER_RX_CHUNK_SIZE 256U
#define SESSION_BROKER_ACTIVE_BURST_BYTES (8U * SESSION_BROKER_RX_CHUNK_SIZE)
#define SESSION_BROKER_TASK_STACK_SIZE 4096U
#define SESSION_BROKER_TASK_PRIORITY 9U
#define SESSION_BROKER_IDLE_POLL_MS 5U
#define SESSION_BROKER_SLOT_BITS 3U
#define SESSION_BROKER_SLOT_MASK ((1U << SESSION_BROKER_SLOT_BITS) - 1U)
#define SESSION_BROKER_MAX_GENERATION (UINT32_MAX >> SESSION_BROKER_SLOT_BITS)
#define SESSION_BROKER_OUTPUT_STORAGE_SIZE (SESSION_BROKER_OUTPUT_SIZE + 1U)
#define SESSION_BROKER_EVENT_STORAGE_SIZE                                      \
    (SESSION_BROKER_EVENT_QUEUE_LENGTH * sizeof(session_broker_event_t))

_Static_assert(
    SESSION_BROKER_MAX_CLIENTS == (1U << SESSION_BROKER_SLOT_BITS),
    "Client slot bit width must exactly represent SESSION_BROKER_MAX_CLIENTS");

typedef struct {
    StreamBufferHandle_t output;
    QueueHandle_t events;
    StaticStreamBuffer_t output_control;
    StaticQueue_t events_control;
    uint8_t *output_storage;
    uint8_t *event_storage;
    session_broker_client_id_t id;
    uint32_t generation;
    session_broker_client_type_t type;
    char name[SESSION_BROKER_CLIENT_NAME_MAX + 1U];
    bool connected;
    session_broker_client_counters_t counters;
} session_broker_slot_t;

static SemaphoreHandle_t s_mutex;
static TaskHandle_t s_broker_task;
static session_broker_slot_t s_slots[SESSION_BROKER_MAX_CLIENTS];
static session_broker_client_id_t s_writer_id;
static uint32_t s_connected_clients;
static uint64_t s_event_sequence;
static session_broker_global_counters_t s_counters;
static bool s_initialized;

static TickType_t milliseconds_to_ticks(uint32_t milliseconds)
{
    TickType_t ticks = pdMS_TO_TICKS(milliseconds);
    return (milliseconds > 0U && ticks == 0U) ? 1U : ticks;
}

static bool client_type_valid(session_broker_client_type_t type)
{
    return type >= SESSION_BROKER_CLIENT_CONSOLE &&
           type <= SESSION_BROKER_CLIENT_INTERNAL;
}

static session_broker_client_id_t make_client_id(size_t slot_index, uint32_t generation)
{
    return (session_broker_client_id_t)((generation << SESSION_BROKER_SLOT_BITS) |
                                        (uint32_t)slot_index);
}

static session_broker_slot_t *find_slot_locked(session_broker_client_id_t client_id)
{
    if (client_id == SESSION_BROKER_NO_CLIENT) {
        return NULL;
    }

    size_t slot_index = client_id & SESSION_BROKER_SLOT_MASK;
    if (slot_index >= SESSION_BROKER_MAX_CLIENTS) {
        return NULL;
    }

    session_broker_slot_t *slot = &s_slots[slot_index];
    if (!slot->connected || slot->id != client_id) {
        return NULL;
    }
    return slot;
}

static void fill_client_snapshot_locked(const session_broker_slot_t *slot,
                                        session_broker_client_snapshot_t *snapshot)
{
    snapshot->id = slot->id;
    snapshot->type = slot->type;
    memcpy(snapshot->name, slot->name, sizeof(snapshot->name));
    snapshot->is_writer = slot->id == s_writer_id;
    snapshot->output_bytes_pending = xStreamBufferBytesAvailable(slot->output);
    snapshot->events_pending = (uint32_t)uxQueueMessagesWaiting(slot->events);
    snapshot->counters = slot->counters;
}

static void broadcast_event_locked(session_broker_event_type_t type,
                                   session_broker_client_id_t client_id,
                                   session_broker_client_id_t writer_id)
{
    session_broker_event_t event = {
        .sequence = ++s_event_sequence,
        .type = type,
        .client_id = client_id,
        .writer_id = writer_id,
    };
    ++s_counters.events_generated;

    for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
        session_broker_slot_t *slot = &s_slots[i];
        if (!slot->connected) {
            continue;
        }

        if (xQueueSend(slot->events, &event, 0) == pdTRUE) {
            ++slot->counters.events_queued;
            ++s_counters.events_queued;
        } else {
            ++slot->counters.event_drops;
            ++s_counters.event_drops;
        }
    }
}

static void fan_out_rx_locked(const uint8_t *data, size_t size)
{
    s_counters.uart_rx_bytes += size;

    if (s_connected_clients == 0U) {
        /* UART RX is always drained, even when nobody can observe the bytes. */
        s_counters.unobserved_rx_bytes += size;
        return;
    }

    /*
     * The broker task is the sole output-stream writer. Every send is
     * zero-time, so a slow observer only fills its own bounded stream and
     * accrues drops; it can never stall UART draining or another observer.
     */
    for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
        session_broker_slot_t *slot = &s_slots[i];
        if (!slot->connected) {
            continue;
        }

        slot->counters.uart_rx_bytes += size;
        size_t queued = xStreamBufferSend(slot->output, data, size, 0);
        size_t dropped = size - queued;
        slot->counters.output_queued_bytes += queued;
        slot->counters.output_dropped_bytes += dropped;
        s_counters.output_queued_bytes += queued;
        s_counters.output_dropped_bytes += dropped;
    }
}

static void broker_task(void *context)
{
    (void)context;
    uint8_t data[SESSION_BROKER_RX_CHUNK_SIZE];
    size_t active_burst_bytes = 0U;

    for (;;) {
        /*
         * Hold the broker mutex across the serial read and fan-out. This gives
         * every byte batch an exact order relative to connect/disconnect and
         * prevents data crossing client-generation boundaries.
         */
        xSemaphoreTake(s_mutex, portMAX_DELAY);
        size_t received = serial_service_read(data, sizeof(data));
        if (received > 0U) {
            fan_out_rx_locked(data, received);
        }
        xSemaphoreGive(s_mutex);

        if (received == 0U) {
            active_burst_bytes = 0U;
            vTaskDelay(milliseconds_to_ticks(SESSION_BROKER_IDLE_POLL_MS));
        } else {
            active_burst_bytes += received;
            if (active_burst_bytes >= SESSION_BROKER_ACTIVE_BURST_BYTES) {
                active_burst_bytes = 0U;
                /*
                 * A continuously readable UART must not make this priority-9
                 * task permanently runnable. One tick after each bounded burst
                 * preserves 1 Mbaud headroom while allowing idle and transports
                 * to run on a loaded dual-core system.
                 */
                vTaskDelay(1U);
            }
        }
    }
}

static void cleanup_allocations(void)
{
    for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
        if (s_slots[i].events != NULL) {
            vQueueDelete(s_slots[i].events);
            s_slots[i].events = NULL;
        }
        if (s_slots[i].output != NULL) {
            vStreamBufferDelete(s_slots[i].output);
            s_slots[i].output = NULL;
        }
        heap_caps_free(s_slots[i].event_storage);
        s_slots[i].event_storage = NULL;
        heap_caps_free(s_slots[i].output_storage);
        s_slots[i].output_storage = NULL;
    }
    if (s_mutex != NULL) {
        vSemaphoreDelete(s_mutex);
        s_mutex = NULL;
    }
}

esp_err_t session_broker_init(void)
{
    if (s_initialized || s_mutex != NULL) {
        return ESP_ERR_INVALID_STATE;
    }

    memset(s_slots, 0, sizeof(s_slots));
    memset(&s_counters, 0, sizeof(s_counters));
    s_writer_id = SESSION_BROKER_NO_CLIENT;
    s_connected_clients = 0U;
    s_event_sequence = 0U;
    s_broker_task = NULL;

    s_mutex = xSemaphoreCreateMutex();
    if (s_mutex == NULL) {
        return ESP_ERR_NO_MEM;
    }

    for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
        s_slots[i].output_storage = heap_caps_calloc_prefer(
            1U, SESSION_BROKER_OUTPUT_STORAGE_SIZE, 2,
            MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT,
            MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT);
        s_slots[i].event_storage = heap_caps_calloc_prefer(
            1U, SESSION_BROKER_EVENT_STORAGE_SIZE, 2,
            MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT,
            MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT);
        if (s_slots[i].output_storage != NULL) {
            s_slots[i].output = xStreamBufferCreateStatic(
                SESSION_BROKER_OUTPUT_STORAGE_SIZE, 1U,
                s_slots[i].output_storage, &s_slots[i].output_control);
        }
        if (s_slots[i].event_storage != NULL) {
            s_slots[i].events = xQueueCreateStatic(
                SESSION_BROKER_EVENT_QUEUE_LENGTH,
                sizeof(session_broker_event_t),
                s_slots[i].event_storage, &s_slots[i].events_control);
        }
        if (s_slots[i].output == NULL || s_slots[i].events == NULL) {
            cleanup_allocations();
            return ESP_ERR_NO_MEM;
        }
    }

    if (xTaskCreate(broker_task,
                    "session_broker",
                    SESSION_BROKER_TASK_STACK_SIZE,
                    NULL,
                    SESSION_BROKER_TASK_PRIORITY,
                    &s_broker_task) != pdPASS) {
        s_broker_task = NULL;
        cleanup_allocations();
        return ESP_ERR_NO_MEM;
    }

    s_initialized = true;
    return ESP_OK;
}

esp_err_t session_broker_connect(session_broker_client_type_t type,
                                 const char *name,
                                 session_broker_client_id_t *client_id)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }
    if (!client_type_valid(type) || name == NULL || client_id == NULL) {
        return ESP_ERR_INVALID_ARG;
    }
    size_t name_length = strnlen(name, SESSION_BROKER_CLIENT_NAME_MAX + 1U);
    if (name_length > SESSION_BROKER_CLIENT_NAME_MAX) {
        return ESP_ERR_INVALID_ARG;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    session_broker_slot_t *slot = NULL;
    size_t slot_index = 0U;
    for (; slot_index < SESSION_BROKER_MAX_CLIENTS; ++slot_index) {
        if (!s_slots[slot_index].connected) {
            slot = &s_slots[slot_index];
            break;
        }
    }
    if (slot == NULL) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NO_MEM;
    }

    /* DTR on-connect must succeed before this connection becomes observable. */
    if (s_connected_clients == 0U) {
        esp_err_t result = serial_service_set_session_active(true);
        if (result != ESP_OK) {
            xSemaphoreGive(s_mutex);
            return result;
        }
    }

    uint32_t generation = slot->generation + 1U;
    if (generation == 0U || generation > SESSION_BROKER_MAX_GENERATION) {
        generation = 1U;
    }

    xStreamBufferReset(slot->output);
    xQueueReset(slot->events);
    memset(&slot->counters, 0, sizeof(slot->counters));
    slot->generation = generation;
    slot->id = make_client_id(slot_index, generation);
    slot->type = type;
    memset(slot->name, 0, sizeof(slot->name));
    memcpy(slot->name, name, name_length);
    slot->connected = true;

    ++s_connected_clients;
    ++slot->counters.connections;
    ++s_counters.connections;
    *client_id = slot->id;
    broadcast_event_locked(SESSION_BROKER_EVENT_CONNECT, slot->id, s_writer_id);
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_disconnect(session_broker_client_id_t client_id)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    session_broker_slot_t *slot = find_slot_locked(client_id);
    if (slot == NULL) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NOT_FOUND;
    }

    /* A failed last-client DTR transition leaves the session fully intact. */
    if (s_connected_clients == 1U) {
        esp_err_t dtr_error = serial_service_set_session_active(false);
        if (dtr_error != ESP_OK) {
            xSemaphoreGive(s_mutex);
            return dtr_error;
        }
    }

    bool released_writer = s_writer_id == client_id;
    slot->connected = false;
    --s_connected_clients;

    if (released_writer) {
        s_writer_id = SESSION_BROKER_NO_CLIENT;
        ++slot->counters.writer_releases;
        ++slot->counters.writer_changes;
        ++s_counters.writer_releases;
        ++s_counters.writer_changes;
        broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_RELEASED,
                               client_id,
                               s_writer_id);
    }

    ++s_counters.disconnections;
    broadcast_event_locked(SESSION_BROKER_EVENT_DISCONNECT, client_id, s_writer_id);

    /* Account for accepted output/events that the disconnect discards. */
    size_t unread_output = xStreamBufferBytesAvailable(slot->output);
    UBaseType_t unpopped_events = uxQueueMessagesWaiting(slot->events);
    slot->counters.output_dropped_bytes += unread_output;
    slot->counters.event_drops += unpopped_events;
    s_counters.output_dropped_bytes += unread_output;
    s_counters.event_drops += unpopped_events;

    /* Zero-time resets are mutex-protected against the broker writer/reader. */
    xStreamBufferReset(slot->output);
    xQueueReset(slot->events);
    slot->id = SESSION_BROKER_NO_CLIENT;
    slot->name[0] = '\0';

    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_request_writer(session_broker_client_id_t client_id)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    session_broker_slot_t *slot = find_slot_locked(client_id);
    if (slot == NULL) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NOT_FOUND;
    }

    ++slot->counters.writer_requests;
    ++s_counters.writer_requests;
    if (s_writer_id == client_id) {
        xSemaphoreGive(s_mutex);
        return ESP_OK;
    }
    if (s_writer_id != SESSION_BROKER_NO_CLIENT) {
        ++slot->counters.writer_denials;
        ++s_counters.writer_denials;
        broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_DENIED,
                               client_id,
                               s_writer_id);
        xSemaphoreGive(s_mutex);
        return ESP_ERR_INVALID_STATE;
    }

    s_writer_id = client_id;
    ++slot->counters.writer_grants;
    ++slot->counters.writer_changes;
    ++s_counters.writer_grants;
    ++s_counters.writer_changes;
    broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_GRANTED,
                           client_id,
                           s_writer_id);
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_release_writer(session_broker_client_id_t client_id)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    session_broker_slot_t *slot = find_slot_locked(client_id);
    if (slot == NULL) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NOT_FOUND;
    }
    if (s_writer_id != client_id) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_INVALID_STATE;
    }

    s_writer_id = SESSION_BROKER_NO_CLIENT;
    ++slot->counters.writer_releases;
    ++slot->counters.writer_changes;
    ++s_counters.writer_releases;
    ++s_counters.writer_changes;
    broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_RELEASED,
                           client_id,
                           s_writer_id);
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_force_writer(session_broker_client_id_t client_id)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    session_broker_slot_t *new_writer = NULL;
    if (client_id != SESSION_BROKER_NO_CLIENT) {
        new_writer = find_slot_locked(client_id);
        if (new_writer == NULL) {
            xSemaphoreGive(s_mutex);
            return ESP_ERR_NOT_FOUND;
        }
    }
    if (s_writer_id == client_id) {
        xSemaphoreGive(s_mutex);
        return ESP_OK;
    }

    session_broker_client_id_t old_writer_id = s_writer_id;
    session_broker_slot_t *old_writer = find_slot_locked(old_writer_id);

    /* Every emitted event carries the final atomic ownership state. */
    s_writer_id = client_id;
    if (old_writer != NULL) {
        ++old_writer->counters.writer_revocations;
        ++old_writer->counters.writer_changes;
        ++s_counters.writer_revocations;
    }
    if (new_writer != NULL) {
        ++new_writer->counters.writer_grants;
        ++new_writer->counters.writer_changes;
        ++s_counters.writer_grants;
    }
    ++s_counters.writer_changes;

    if (old_writer != NULL) {
        broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_REVOKED,
                               old_writer_id,
                               s_writer_id);
    }
    if (new_writer != NULL) {
        broadcast_event_locked(SESSION_BROKER_EVENT_WRITER_GRANTED,
                               new_writer->id,
                               s_writer_id);
    }

    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

session_broker_client_id_t session_broker_get_writer_id(void)
{
    if (!s_initialized) {
        return SESSION_BROKER_NO_CLIENT;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    session_broker_client_id_t writer_id = s_writer_id;
    xSemaphoreGive(s_mutex);
    return writer_id;
}

esp_err_t session_broker_write(session_broker_client_id_t client_id,
                               const uint8_t *data,
                               size_t size,
                               size_t *accepted)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }
    if (accepted == NULL || (data == NULL && size > 0U)) {
        return ESP_ERR_INVALID_ARG;
    }
    *accepted = 0U;

    if (xSemaphoreTake(s_mutex, 0) != pdTRUE) {
        return ESP_ERR_TIMEOUT;
    }
    session_broker_slot_t *slot = find_slot_locked(client_id);
    if (slot == NULL) {
        s_counters.tx_rejected_bytes += size;
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NOT_FOUND;
    }
    if (s_writer_id != client_id) {
        slot->counters.tx_rejected_bytes += size;
        s_counters.tx_rejected_bytes += size;
        xSemaphoreGive(s_mutex);
        return ESP_ERR_INVALID_STATE;
    }

    size_t queued = serial_service_write(data, size);
    size_t rejected = size - queued;
    slot->counters.tx_accepted_bytes += queued;
    slot->counters.tx_rejected_bytes += rejected;
    s_counters.tx_accepted_bytes += queued;
    s_counters.tx_rejected_bytes += rejected;
    *accepted = queued;
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_read(session_broker_client_id_t client_id,
                              uint8_t *data,
                              size_t size,
                              size_t *received)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }
    if (received == NULL || (data == NULL && size > 0U)) {
        return ESP_ERR_INVALID_ARG;
    }
    *received = 0U;

    if (xSemaphoreTake(s_mutex, 0) != pdTRUE) {
        return ESP_ERR_TIMEOUT;
    }
    session_broker_slot_t *slot = find_slot_locked(client_id);
    if (slot == NULL) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NOT_FOUND;
    }
    if (size == 0U) {
        xSemaphoreGive(s_mutex);
        return ESP_OK;
    }

    size_t count = xStreamBufferReceive(slot->output, data, size, 0);
    slot->counters.output_read_bytes += count;
    s_counters.output_read_bytes += count;
    *received = count;
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_pop_event(session_broker_client_id_t client_id,
                                   session_broker_event_t *event)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }
    if (event == NULL) {
        return ESP_ERR_INVALID_ARG;
    }

    if (xSemaphoreTake(s_mutex, 0) != pdTRUE) {
        return ESP_ERR_TIMEOUT;
    }
    session_broker_slot_t *slot = find_slot_locked(client_id);
    if (slot == NULL) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NOT_FOUND;
    }
    if (xQueueReceive(slot->events, event, 0) != pdTRUE) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_TIMEOUT;
    }

    ++slot->counters.events_popped;
    ++s_counters.events_popped;
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_get_client_snapshot(session_broker_client_id_t client_id,
                                             session_broker_client_snapshot_t *snapshot)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }
    if (snapshot == NULL) {
        return ESP_ERR_INVALID_ARG;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    session_broker_slot_t *slot = find_slot_locked(client_id);
    if (slot == NULL) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NOT_FOUND;
    }
    fill_client_snapshot_locked(slot, snapshot);
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_get_global_snapshot(session_broker_global_snapshot_t *snapshot)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }
    if (snapshot == NULL) {
        return ESP_ERR_INVALID_ARG;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    snapshot->writer_id = s_writer_id;
    snapshot->connected_clients = s_connected_clients;
    snapshot->latest_event_sequence = s_event_sequence;
    snapshot->counters = s_counters;
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

size_t session_broker_list_clients(session_broker_client_snapshot_t *clients,
                                   size_t capacity)
{
    if (!s_initialized || (clients == NULL && capacity > 0U)) {
        return 0U;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    if (clients == NULL) {
        size_t count = s_connected_clients;
        xSemaphoreGive(s_mutex);
        return count;
    }

    size_t copied = 0U;
    for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS && copied < capacity; ++i) {
        if (!s_slots[i].connected) {
            continue;
        }
        fill_client_snapshot_locked(&s_slots[i], &clients[copied]);
        ++copied;
    }
    xSemaphoreGive(s_mutex);
    return copied;
}

esp_err_t session_broker_clear_counters(void)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    memset(&s_counters, 0, sizeof(s_counters));
    for (size_t i = 0; i < SESSION_BROKER_MAX_CLIENTS; ++i) {
        memset(&s_slots[i].counters, 0, sizeof(s_slots[i].counters));
    }
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}

esp_err_t session_broker_clear_client_counters(session_broker_client_id_t client_id)
{
    if (!s_initialized) {
        return ESP_ERR_INVALID_STATE;
    }

    xSemaphoreTake(s_mutex, portMAX_DELAY);
    session_broker_slot_t *slot = find_slot_locked(client_id);
    if (slot == NULL) {
        xSemaphoreGive(s_mutex);
        return ESP_ERR_NOT_FOUND;
    }
    memset(&slot->counters, 0, sizeof(slot->counters));
    xSemaphoreGive(s_mutex);
    return ESP_OK;
}