/**
 * aviation_event_bus.c — Thread-safe event pub/sub implementation.
 *
 * Uses pthread_mutex for thread safety and snapshot-id-then-fire pattern
 * to prevent deadlocks from re-entrant listeners while still respecting
 * unsubscriptions that happen before a listener is invoked.
 */

#include "aviation_event_bus.h"
#include <pthread.h>
#include <stdlib.h>
#include <string.h>

/* ─── Internal types ──────────────────────────────────────────────── */

typedef struct {
    uint32_t id;
    AviationCommandListener listener;
    void *user_data;
    bool active;
    int in_flight;
} CommandListenerEntry;

typedef struct {
    uint32_t id;
    AviationDomainListener listener;
    void *user_data;
    bool active;
    int in_flight;
} DomainListenerEntry;

struct AviationEventBus {
    pthread_mutex_t mutex;
    pthread_cond_t idle_cond;
    uint32_t next_id;

    CommandListenerEntry command_listeners[AVIATION_MAX_LISTENERS];
    int command_count;

    DomainListenerEntry domain_listeners[AVIATION_MAX_LISTENERS];
    int domain_count;
};

typedef enum {
    AVIATION_CALLBACK_NONE,
    AVIATION_CALLBACK_COMMAND,
    AVIATION_CALLBACK_DOMAIN,
} AviationCallbackKind;

static _Thread_local AviationEventBus *current_callback_bus = NULL;
static _Thread_local AviationCallbackKind current_callback_kind = AVIATION_CALLBACK_NONE;
static _Thread_local uint32_t current_callback_subscription_id = 0;

/* ─── Internal helpers ────────────────────────────────────────────── */

static uint32_t next_subscription_id(AviationEventBus *bus) {
    uint32_t id = bus->next_id++;
    if (bus->next_id == 0) {
        bus->next_id = 1;
    }
    return id;
}

static void compact_command_listeners(AviationEventBus *bus) {
    int write = 0;
    for (int read = 0; read < bus->command_count; read++) {
        CommandListenerEntry *entry = &bus->command_listeners[read];
        if (entry->active || entry->in_flight > 0) {
            if (write != read) {
                bus->command_listeners[write] = bus->command_listeners[read];
            }
            write++;
        }
    }
    bus->command_count = write;
}

static void compact_domain_listeners(AviationEventBus *bus) {
    int write = 0;
    for (int read = 0; read < bus->domain_count; read++) {
        DomainListenerEntry *entry = &bus->domain_listeners[read];
        if (entry->active || entry->in_flight > 0) {
            if (write != read) {
                bus->domain_listeners[write] = bus->domain_listeners[read];
            }
            write++;
        }
    }
    bus->domain_count = write;
}

static int find_command_listener(AviationEventBus *bus, uint32_t subscription_id) {
    for (int i = 0; i < bus->command_count; i++) {
        if (bus->command_listeners[i].id == subscription_id) {
            return i;
        }
    }
    return -1;
}

static int find_domain_listener(AviationEventBus *bus, uint32_t subscription_id) {
    for (int i = 0; i < bus->domain_count; i++) {
        if (bus->domain_listeners[i].id == subscription_id) {
            return i;
        }
    }
    return -1;
}

static int in_flight_wait_threshold(AviationEventBus *bus, AviationCallbackKind kind, uint32_t subscription_id) {
    return current_callback_bus == bus && current_callback_kind == kind &&
                   current_callback_subscription_id == subscription_id
               ? 1
               : 0;
}

/* ─── Lifecycle ───────────────────────────────────────────────────── */

AviationEventBus *aviation_event_bus_create(void) {
    AviationEventBus *bus = (AviationEventBus *)calloc(1, sizeof(AviationEventBus));
    if (!bus)
        return NULL;

    pthread_mutex_init(&bus->mutex, NULL);
    pthread_cond_init(&bus->idle_cond, NULL);
    bus->next_id = 1;
    bus->command_count = 0;
    bus->domain_count = 0;
    return bus;
}

void aviation_event_bus_destroy(AviationEventBus *bus) {
    if (!bus)
        return;
    aviation_event_bus_remove_all_listeners(bus);
    pthread_mutex_destroy(&bus->mutex);
    pthread_cond_destroy(&bus->idle_cond);
    free(bus);
}

/* ─── Command event channel ───────────────────────────────────────── */

uint32_t aviation_event_bus_subscribe_commands(AviationEventBus *bus, AviationCommandListener listener,
                                               void *user_data) {
    if (!bus || !listener)
        return 0;

    pthread_mutex_lock(&bus->mutex);

    compact_command_listeners(bus);
    if (bus->command_count >= AVIATION_MAX_LISTENERS) {
        pthread_mutex_unlock(&bus->mutex);
        return 0;
    }

    uint32_t id = next_subscription_id(bus);
    CommandListenerEntry *entry = &bus->command_listeners[bus->command_count];
    entry->id = id;
    entry->listener = listener;
    entry->user_data = user_data;
    entry->active = true;
    entry->in_flight = 0;
    bus->command_count++;

    pthread_mutex_unlock(&bus->mutex);
    return id;
}

void *aviation_event_bus_unsubscribe_commands(AviationEventBus *bus, uint32_t subscription_id) {
    if (!bus || subscription_id == 0)
        return NULL;

    void *user_data = NULL;
    pthread_mutex_lock(&bus->mutex);

    int index = find_command_listener(bus, subscription_id);
    if (index >= 0) {
        CommandListenerEntry *entry = &bus->command_listeners[index];
        user_data = entry->user_data;
        entry->active = false;

        while (entry->in_flight > in_flight_wait_threshold(bus, AVIATION_CALLBACK_COMMAND, subscription_id)) {
            pthread_cond_wait(&bus->idle_cond, &bus->mutex);
            index = find_command_listener(bus, subscription_id);
            if (index < 0)
                break;
            entry = &bus->command_listeners[index];
        }

        compact_command_listeners(bus);
    }

    pthread_mutex_unlock(&bus->mutex);
    return user_data;
}

void aviation_event_bus_emit_command(AviationEventBus *bus, const AviationCommandEvent *event) {
    if (!bus || !event)
        return;

    /* Snapshot IDs, then revalidate each subscription before invoking it. */
    uint32_t ids[AVIATION_MAX_LISTENERS];
    int count;

    pthread_mutex_lock(&bus->mutex);
    count = bus->command_count;
    for (int i = 0; i < count; i++) {
        ids[i] = bus->command_listeners[i].active ? bus->command_listeners[i].id : 0;
    }
    pthread_mutex_unlock(&bus->mutex);

    for (int i = 0; i < count; i++) {
        if (ids[i] == 0)
            continue;

        AviationCommandListener listener = NULL;
        void *user_data = NULL;

        pthread_mutex_lock(&bus->mutex);
        int index = find_command_listener(bus, ids[i]);
        if (index >= 0 && bus->command_listeners[index].active && bus->command_listeners[index].listener) {
            bus->command_listeners[index].in_flight++;
            listener = bus->command_listeners[index].listener;
            user_data = bus->command_listeners[index].user_data;
        }
        pthread_mutex_unlock(&bus->mutex);

        if (!listener)
            continue;

        AviationEventBus *previous_bus = current_callback_bus;
        AviationCallbackKind previous_kind = current_callback_kind;
        uint32_t previous_id = current_callback_subscription_id;
        current_callback_bus = bus;
        current_callback_kind = AVIATION_CALLBACK_COMMAND;
        current_callback_subscription_id = ids[i];

        listener(event, user_data);

        current_callback_bus = previous_bus;
        current_callback_kind = previous_kind;
        current_callback_subscription_id = previous_id;

        pthread_mutex_lock(&bus->mutex);
        index = find_command_listener(bus, ids[i]);
        if (index >= 0) {
            bus->command_listeners[index].in_flight--;
            if (bus->command_listeners[index].in_flight == 0) {
                pthread_cond_broadcast(&bus->idle_cond);
            }
        }
        pthread_mutex_unlock(&bus->mutex);
    }
}

/* ─── Domain event channel ────────────────────────────────────────── */

uint32_t aviation_event_bus_subscribe_domain(AviationEventBus *bus, AviationDomainListener listener, void *user_data) {
    if (!bus || !listener)
        return 0;

    pthread_mutex_lock(&bus->mutex);

    compact_domain_listeners(bus);
    if (bus->domain_count >= AVIATION_MAX_LISTENERS) {
        pthread_mutex_unlock(&bus->mutex);
        return 0;
    }

    uint32_t id = next_subscription_id(bus);
    DomainListenerEntry *entry = &bus->domain_listeners[bus->domain_count];
    entry->id = id;
    entry->listener = listener;
    entry->user_data = user_data;
    entry->active = true;
    entry->in_flight = 0;
    bus->domain_count++;

    pthread_mutex_unlock(&bus->mutex);
    return id;
}

void *aviation_event_bus_unsubscribe_domain(AviationEventBus *bus, uint32_t subscription_id) {
    if (!bus || subscription_id == 0)
        return NULL;

    void *user_data = NULL;
    pthread_mutex_lock(&bus->mutex);

    int index = find_domain_listener(bus, subscription_id);
    if (index >= 0) {
        DomainListenerEntry *entry = &bus->domain_listeners[index];
        user_data = entry->user_data;
        entry->active = false;

        while (entry->in_flight > in_flight_wait_threshold(bus, AVIATION_CALLBACK_DOMAIN, subscription_id)) {
            pthread_cond_wait(&bus->idle_cond, &bus->mutex);
            index = find_domain_listener(bus, subscription_id);
            if (index < 0)
                break;
            entry = &bus->domain_listeners[index];
        }

        compact_domain_listeners(bus);
    }

    pthread_mutex_unlock(&bus->mutex);
    return user_data;
}

void aviation_event_bus_emit_domain(AviationEventBus *bus, const AviationDomainEvent *event) {
    if (!bus || !event)
        return;

    uint32_t ids[AVIATION_MAX_LISTENERS];
    int count;

    pthread_mutex_lock(&bus->mutex);
    count = bus->domain_count;
    for (int i = 0; i < count; i++) {
        ids[i] = bus->domain_listeners[i].active ? bus->domain_listeners[i].id : 0;
    }
    pthread_mutex_unlock(&bus->mutex);

    for (int i = 0; i < count; i++) {
        if (ids[i] == 0)
            continue;

        AviationDomainListener listener = NULL;
        void *user_data = NULL;

        pthread_mutex_lock(&bus->mutex);
        int index = find_domain_listener(bus, ids[i]);
        if (index >= 0 && bus->domain_listeners[index].active && bus->domain_listeners[index].listener) {
            bus->domain_listeners[index].in_flight++;
            listener = bus->domain_listeners[index].listener;
            user_data = bus->domain_listeners[index].user_data;
        }
        pthread_mutex_unlock(&bus->mutex);

        if (!listener)
            continue;

        AviationEventBus *previous_bus = current_callback_bus;
        AviationCallbackKind previous_kind = current_callback_kind;
        uint32_t previous_id = current_callback_subscription_id;
        current_callback_bus = bus;
        current_callback_kind = AVIATION_CALLBACK_DOMAIN;
        current_callback_subscription_id = ids[i];

        listener(event, user_data);

        current_callback_bus = previous_bus;
        current_callback_kind = previous_kind;
        current_callback_subscription_id = previous_id;

        pthread_mutex_lock(&bus->mutex);
        index = find_domain_listener(bus, ids[i]);
        if (index >= 0) {
            bus->domain_listeners[index].in_flight--;
            if (bus->domain_listeners[index].in_flight == 0) {
                pthread_cond_broadcast(&bus->idle_cond);
            }
        }
        pthread_mutex_unlock(&bus->mutex);
    }
}

/* ─── Teardown ────────────────────────────────────────────────────── */

void aviation_event_bus_remove_all_listeners(AviationEventBus *bus) {
    if (!bus)
        return;

    pthread_mutex_lock(&bus->mutex);

    for (int i = 0; i < bus->command_count; i++) {
        bus->command_listeners[i].active = false;
    }
    for (int i = 0; i < bus->domain_count; i++) {
        bus->domain_listeners[i].active = false;
    }

    bool waiting;
    do {
        waiting = false;
        for (int i = 0; i < bus->command_count; i++) {
            int threshold = in_flight_wait_threshold(bus, AVIATION_CALLBACK_COMMAND, bus->command_listeners[i].id);
            if (bus->command_listeners[i].in_flight > threshold) {
                waiting = true;
                break;
            }
        }
        for (int i = 0; !waiting && i < bus->domain_count; i++) {
            int threshold = in_flight_wait_threshold(bus, AVIATION_CALLBACK_DOMAIN, bus->domain_listeners[i].id);
            if (bus->domain_listeners[i].in_flight > threshold) {
                waiting = true;
                break;
            }
        }
        if (waiting) {
            pthread_cond_wait(&bus->idle_cond, &bus->mutex);
        }
    } while (waiting);

    compact_command_listeners(bus);
    compact_domain_listeners(bus);
    pthread_mutex_unlock(&bus->mutex);
}