#include "naut/event.h"
#include "naut/log.h"
#include "naut/plugin.h"
#include "naut/rpc.h"
#include "naut/script.h"
#include "naut/storage.h"
#include "naut/swarm.h"

#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <poll.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>

#define DEFAULT_SOCKET "/tmp/nautd.sock"
#define MOVE_QUEUE_CAPACITY 64
#define MAX_SUBSCRIBERS 64
#define MAX_TORRENTS 128

typedef struct {
    uint32_t file_index;
    char destination[PATH_MAX];
} move_command;

typedef enum {
    TORRENT_QUEUED,
    TORRENT_RUNNING,
    TORRENT_COMPLETE,
    TORRENT_STOPPING,
    TORRENT_STOPPED,
    TORRENT_ERROR,
} torrent_state;

typedef struct daemon_state daemon_state;

typedef struct {
    daemon_state *daemon;
    uint64_t id;
    char *source;
    bool source_is_temp;   /* source is a daemon-owned upload; unlink on destroy */
    char *output_dir;
    char **peers;
    size_t num_peers;
    pthread_t thread;
    bool thread_started;
    bool thread_done;
    pthread_mutex_t lock;
    torrent_state state;
    naut_err result;
    bool stop_requested;
    bool remove_requested;
    naut_swarm_stats stats;
    move_command moves[MOVE_QUEUE_CAPACITY];
    size_t move_head;
    size_t move_count;
    uint64_t moves_processed;
    uint64_t moves_failed;
} torrent_task;

struct daemon_state {
    naut_event_bus *events;
    naut_rpc_registry *rpc;
    naut_plugin_manager *plugins;
    naut_script *script;
    pthread_mutex_t torrent_lock;
    torrent_task *torrents[MAX_TORRENTS];
    size_t torrent_count;
    uint64_t next_torrent_id;
    pthread_mutex_t subscriber_lock;
    int subscribers[MAX_SUBSCRIBERS];
    size_t subscriber_count;
    bool stopping;
};

static volatile sig_atomic_t interrupted;

static naut_err queue_move(void *opaque, uint64_t torrent_id,
                           uint32_t file_index, const char *destination);

static void on_signal(int signal_number) {
    (void)signal_number;
    interrupted = 1;
}

static const char *torrent_state_name(torrent_state state) {
    static const char *names[] = {
        [TORRENT_QUEUED] = "queued",
        [TORRENT_RUNNING] = "downloading",
        [TORRENT_COMPLETE] = "complete",
        [TORRENT_STOPPING] = "stopping",
        [TORRENT_STOPPED] = "stopped",
        [TORRENT_ERROR] = "error",
    };
    return (size_t)state < NAUT_ARRAY_LEN(names) ? names[state] : "unknown";
}

static torrent_task *find_torrent_locked(daemon_state *state, uint64_t id) {
    for (size_t i = 0; i < state->torrent_count; i++)
        if (state->torrents[i]->id == id) return state->torrents[i];
    return NULL;
}

static void torrent_progress(void *opaque, const naut_swarm_stats *stats) {
    torrent_task *task = opaque;
    pthread_mutex_lock(&task->lock);
    task->stats = *stats;
    if (stats->total_pieces > 0 &&
        stats->pieces_done == stats->total_pieces)
        task->state = TORRENT_COMPLETE;
    else if (task->state == TORRENT_QUEUED)
        task->state = TORRENT_RUNNING;
    pthread_mutex_unlock(&task->lock);
}

static bool torrent_should_stop(void *opaque) {
    torrent_task *task = opaque;
    pthread_mutex_lock(&task->lock);
    bool stop = task->stop_requested;
    pthread_mutex_unlock(&task->lock);
    return stop;
}

static void torrent_control(void *opaque, naut_storage *storage) {
    torrent_task *task = opaque;
    for (;;) {
        move_command command;
        pthread_mutex_lock(&task->lock);
        if (task->move_count == 0) {
            pthread_mutex_unlock(&task->lock);
            return;
        }
        command = task->moves[task->move_head];
        task->move_head = (task->move_head + 1) % MOVE_QUEUE_CAPACITY;
        task->move_count--;
        pthread_mutex_unlock(&task->lock);

        naut_err error = naut_storage_relocate(
            storage, command.file_index, command.destination);
        pthread_mutex_lock(&task->lock);
        if (error == NAUT_OK)
            task->moves_processed++;
        else
            task->moves_failed++;
        pthread_mutex_unlock(&task->lock);
        if (error == NAUT_OK)
            NAUT_INFO("moved torrent=%llu file=%u -> %s",
                      (unsigned long long)task->id, command.file_index,
                      command.destination);
        else
            NAUT_WARN("move torrent=%llu file=%u -> %s failed: %s",
                      (unsigned long long)task->id, command.file_index,
                      command.destination, naut_strerror(error));
    }
}

static void *torrent_worker(void *opaque) {
    torrent_task *task = opaque;
    pthread_mutex_lock(&task->lock);
    task->state = TORRENT_RUNNING;
    pthread_mutex_unlock(&task->lock);

    naut_swarm_config config = {
        .source = task->source,
        .output_dir = task->output_dir,
        .peers = (const char *const *)task->peers,
        .num_peers = task->num_peers,
        .torrent_id = task->id,
        .events = task->daemon->events,
        .keep_alive = true,
        .on_progress = torrent_progress,
        .on_control = torrent_control,
        .should_stop = torrent_should_stop,
        .context = task,
    };
    naut_err result = naut_swarm_run(&config);

    pthread_mutex_lock(&task->lock);
    task->result = result;
    if (result == NAUT_OK)
        task->state = TORRENT_COMPLETE;
    else if (task->stop_requested)
        task->state = TORRENT_STOPPED;
    else
        task->state = TORRENT_ERROR;
    task->thread_done = true;
    pthread_mutex_unlock(&task->lock);
    return NULL;
}

static json_t *torrent_json(torrent_task *task) {
    pthread_mutex_lock(&task->lock);
    json_t *result = json_object();
    if (result) {
        json_object_set_new(result, "torrent_id",
                            json_integer((json_int_t)task->id));
        json_object_set_new(result, "source", json_string(task->source));
        json_object_set_new(result, "output",
                            json_string(task->output_dir));
        json_object_set_new(result, "state",
                            json_string(torrent_state_name(task->state)));
        json_object_set_new(result, "bytes_done",
                            json_integer((json_int_t)task->stats.bytes_done));
        json_object_set_new(result, "total_bytes",
                            json_integer((json_int_t)task->stats.total_bytes));
        json_object_set_new(result, "pieces_done",
                            json_integer(task->stats.pieces_done));
        json_object_set_new(result, "total_pieces",
                            json_integer(task->stats.total_pieces));
        json_object_set_new(result, "peers",
                            json_integer(task->stats.peers_active));
        json_object_set_new(result, "peers_discovered",
                            json_integer(task->stats.peers_total));
        json_object_set_new(result, "peers_connecting",
                            json_integer(task->stats.peers_connecting));
        json_object_set_new(result, "peers_failed",
                            json_integer(task->stats.peers_failed));
        json_object_set_new(result, "elapsed_seconds",
                            json_real(task->stats.elapsed_seconds));
        json_object_set_new(result, "pending_moves",
                            json_integer((json_int_t)task->move_count));
        json_object_set_new(result, "moves_processed",
                            json_integer((json_int_t)task->moves_processed));
        json_object_set_new(result, "moves_failed",
                            json_integer((json_int_t)task->moves_failed));
        if (task->state == TORRENT_ERROR)
            json_object_set_new(result, "error",
                                json_string(naut_strerror(task->result)));
    }
    pthread_mutex_unlock(&task->lock);
    return result;
}

static json_t *rpc_ping(void *opaque, const json_t *params, naut_err *error) {
    (void)opaque;
    (void)params;
    json_t *result = json_object();
    if (!result) {
        *error = NAUT_ERR_NOMEM;
        return NULL;
    }
    json_object_set_new(result, "protocol", json_integer(NAUT_RPC_VERSION));
    json_object_set_new(result, "service", json_string("nautd"));
    *error = NAUT_OK;
    return result;
}

static json_t *rpc_status(void *opaque, const json_t *params,
                          naut_err *error) {
    (void)params;
    daemon_state *state = opaque;
    naut_script_stats stats = {0};
    if (state->script) naut_script_get_stats(state->script, &stats);
    size_t torrent_count;
    size_t active = 0;
    uint64_t moves = 0;
    size_t pending = 0;
    pthread_mutex_lock(&state->torrent_lock);
    torrent_count = state->torrent_count;
    for (size_t i = 0; i < torrent_count; i++) {
        torrent_task *task = state->torrents[i];
        pthread_mutex_lock(&task->lock);
        if (task->state == TORRENT_RUNNING ||
            task->state == TORRENT_STOPPING)
            active++;
        moves += task->moves_processed;
        pending += task->move_count;
        pthread_mutex_unlock(&task->lock);
    }
    pthread_mutex_unlock(&state->torrent_lock);
    json_t *result = json_object();
    json_t *script = json_object();
    if (!result || !script) {
        json_decref(result);
        json_decref(script);
        *error = NAUT_ERR_NOMEM;
        return NULL;
    }
    json_object_set_new(result, "protocol", json_integer(NAUT_RPC_VERSION));
    json_object_set_new(result, "plugins",
                        json_integer((json_int_t)naut_plugin_count(
                            state->plugins)));
    json_object_set_new(result, "storage_backends",
                        json_integer((json_int_t)naut_plugin_storage_count(
                            state->plugins)));
    json_object_set_new(result, "torrents",
                        json_integer((json_int_t)torrent_count));
    json_object_set_new(result, "active_torrents",
                        json_integer((json_int_t)active));
    json_object_set_new(result, "script_loaded",
                        json_boolean(state->script != NULL));
    json_object_set_new(script, "queued", json_integer(stats.queued));
    json_object_set_new(script, "handled", json_integer(stats.handled));
    json_object_set_new(script, "dropped", json_integer(stats.dropped));
    json_object_set_new(script, "errors", json_integer(stats.errors));
    json_object_set_new(script, "move_requests",
                        json_integer(stats.move_requests));
    json_object_set_new(result, "script", script);
    json_object_set_new(result, "move_commands", json_integer(moves));
    json_object_set_new(result, "pending_move_commands",
                        json_integer((json_int_t)pending));
    *error = NAUT_OK;
    return result;
}

static json_t *rpc_plugins(void *opaque, const json_t *params,
                           naut_err *error) {
    (void)params;
    daemon_state *state = opaque;
    json_t *plugins = json_array();
    json_t *storage = json_array();
    if (!plugins || !storage) {
        json_decref(plugins);
        json_decref(storage);
        *error = NAUT_ERR_NOMEM;
        return NULL;
    }
    for (size_t i = 0; i < naut_plugin_count(state->plugins); i++)
        json_array_append_new(plugins,
            json_string(naut_plugin_name(state->plugins, i)));
    for (size_t i = 0; i < naut_plugin_storage_count(state->plugins); i++)
        json_array_append_new(storage,
            json_string(naut_plugin_storage_name(state->plugins, i)));
    json_t *result = json_object();
    if (!result) {
        json_decref(plugins);
        json_decref(storage);
        *error = NAUT_ERR_NOMEM;
        return NULL;
    }
    json_object_set_new(result, "plugins", plugins);
    json_object_set_new(result, "storage_backends", storage);
    *error = NAUT_OK;
    return result;
}

static json_t *rpc_emit(void *opaque, const json_t *params, naut_err *error) {
    daemon_state *state = opaque;
    if (!json_is_object(params)) {
        *error = NAUT_ERR_INVAL;
        return NULL;
    }
    const char *type_name = json_string_value(json_object_get(params, "type"));
    naut_event_type type;
    if (!type_name || !naut_event_type_parse(type_name, &type)) {
        *error = NAUT_ERR_INVAL;
        return NULL;
    }
    json_int_t torrent_id =
        json_integer_value(json_object_get(params, "torrent_id"));
    json_int_t index = json_integer_value(json_object_get(params, "index"));
    if (torrent_id < 0 || index < 0 || (uint64_t)index > UINT32_MAX) {
        *error = NAUT_ERR_RANGE;
        return NULL;
    }
    naut_event event = {
        .type = type,
        .torrent_id = (uint64_t)torrent_id,
        .index = (uint32_t)index,
        .message = json_string_value(json_object_get(params, "message")),
        .path = json_string_value(json_object_get(params, "path")),
    };
    naut_event_emit(state->events, &event);
    *error = NAUT_OK;
    return json_true();
}

static json_t *rpc_shutdown(void *opaque, const json_t *params,
                            naut_err *error) {
    (void)params;
    daemon_state *state = opaque;
    state->stopping = true;
    *error = NAUT_OK;
    return json_true();
}

static int b64_val(int c) {
    if (c >= 'A' && c <= 'Z') return c - 'A';
    if (c >= 'a' && c <= 'z') return c - 'a' + 26;
    if (c >= '0' && c <= '9') return c - '0' + 52;
    if (c == '+') return 62;
    if (c == '/') return 63;
    return -1;   /* padding / whitespace / invalid -> skipped */
}

/* Decode standard base64 (padding optional, whitespace ignored). */
static unsigned char *b64_decode(const char *in, size_t *out_len) {
    size_t cap = strlen(in) / 4 * 3 + 4;
    unsigned char *out = malloc(cap);
    if (!out) return NULL;
    size_t o = 0;
    int acc = 0, bits = 0;
    for (const char *p = in; *p; p++) {
        if (*p == '=') break;
        int v = b64_val((unsigned char)*p);
        if (v < 0) continue;
        acc = (acc << 6) | v;
        bits += 6;
        if (bits >= 8) { bits -= 8; out[o++] = (unsigned char)((acc >> bits) & 0xff); }
    }
    *out_len = o;
    return out;
}

static bool write_all_fd(int fd, const void *buf, size_t len) {
    const char *p = buf;
    while (len) {
        ssize_t n = write(fd, p, len);
        if (n < 0) { if (errno == EINTR) continue; return false; }
        p += n;
        len -= (size_t)n;
    }
    return true;
}

/* A browser uploads a .torrent's bytes as base64 in "data"; the daemon writes
 * its own temp file and owns its lifecycle (no shared path with the client).
 * Returns the temp path in `out` (size cap) or sets *error. */
static bool add_torrent_write_upload(const char *data_b64, char *out,
                                     size_t cap, naut_err *error) {
    size_t raw_len = 0;
    unsigned char *raw = b64_decode(data_b64, &raw_len);
    if (!raw || raw_len == 0) { free(raw); *error = NAUT_ERR_INVAL; return false; }
    char tmpl[] = "/tmp/naut-upload-XXXXXX";
    int fd = mkstemp(tmpl);
    if (fd < 0) { free(raw); *error = NAUT_ERR_IO; return false; }
    bool ok = write_all_fd(fd, raw, raw_len);
    close(fd);
    free(raw);
    if (!ok) { unlink(tmpl); *error = NAUT_ERR_IO; return false; }
    snprintf(out, cap, "%s", tmpl);
    return true;
}

static json_t *rpc_add_torrent(void *opaque, const json_t *params,
                               naut_err *error) {
    daemon_state *state = opaque;
    if (!json_is_object(params)) { *error = NAUT_ERR_INVAL; return NULL; }

    const char *source =
        json_string_value(json_object_get(params, "source"));
    if (!source)
        source = json_string_value(json_object_get(params, "torrent"));
    const char *data_b64 =
        json_string_value(json_object_get(params, "data"));
    const char *output =
        json_string_value(json_object_get(params, "output"));
    if (!output)
        output = json_string_value(json_object_get(params, "root"));
    json_t *peers_json = json_object_get(params, "peers");
    /* need an output and either a source (path/magnet) or uploaded bytes */
    if (!output || !*output ||
        ((!source || !*source) && (!data_b64 || !*data_b64)) ||
        (peers_json && !json_is_array(peers_json))) {
        *error = NAUT_ERR_INVAL;
        return NULL;
    }

    /* materialize an upload into a daemon-owned temp .torrent */
    char temp_source[PATH_MAX];
    bool is_temp = false;
    if ((!source || !*source) && data_b64 && *data_b64) {
        if (!add_torrent_write_upload(data_b64, temp_source, sizeof temp_source,
                                      error))
            return NULL;
        source = temp_source;
        is_temp = true;
    }

    torrent_task *task = calloc(1, sizeof(*task));
    if (!task) {
        if (is_temp) unlink(source);
        *error = NAUT_ERR_NOMEM;
        return NULL;
    }
    task->daemon = state;
    task->state = TORRENT_QUEUED;
    task->result = NAUT_ERR_AGAIN;
    task->source = strdup(source);
    task->source_is_temp = is_temp;
    task->output_dir = strdup(output);
    if (!task->source || !task->output_dir) {
        if (is_temp) unlink(temp_source);
        free(task->source);
        free(task->output_dir);
        free(task);
        *error = NAUT_ERR_NOMEM;
        return NULL;
    }
    if (pthread_mutex_init(&task->lock, NULL) != 0) {
        if (is_temp) unlink(temp_source);
        free(task->source);
        free(task->output_dir);
        free(task);
        *error = NAUT_ERR_NOMEM;
        return NULL;
    }

    task->num_peers = peers_json ? json_array_size(peers_json) : 0;
    if (task->num_peers) {
        task->peers = calloc(task->num_peers, sizeof(*task->peers));
        if (!task->peers) { *error = NAUT_ERR_NOMEM; goto fail_task; }
        for (size_t i = 0; i < task->num_peers; i++) {
            const char *peer =
                json_string_value(json_array_get(peers_json, i));
            if (!peer || !*peer) { *error = NAUT_ERR_INVAL; goto fail_task; }
            task->peers[i] = strdup(peer);
            if (!task->peers[i]) { *error = NAUT_ERR_NOMEM; goto fail_task; }
        }
    }

    json_t *id_json = json_object_get(params, "torrent_id");
    pthread_mutex_lock(&state->torrent_lock);
    if (state->torrent_count == MAX_TORRENTS) {
        pthread_mutex_unlock(&state->torrent_lock);
        *error = NAUT_ERR_FULL;
        goto fail_task;
    }
    if (id_json) {
        if (!json_is_integer(id_json) || json_integer_value(id_json) < 0) {
            pthread_mutex_unlock(&state->torrent_lock);
            *error = NAUT_ERR_INVAL;
            goto fail_task;
        }
        task->id = (uint64_t)json_integer_value(id_json);
        if (task->id < (uint64_t)INT64_MAX &&
            task->id >= state->next_torrent_id)
            state->next_torrent_id = task->id + 1;
    } else {
        while (find_torrent_locked(state, state->next_torrent_id))
            state->next_torrent_id++;
        if (state->next_torrent_id > (uint64_t)INT64_MAX) {
            pthread_mutex_unlock(&state->torrent_lock);
            *error = NAUT_ERR_FULL;
            goto fail_task;
        }
        task->id = state->next_torrent_id++;
    }
    if (find_torrent_locked(state, task->id)) {
        pthread_mutex_unlock(&state->torrent_lock);
        *error = NAUT_ERR_INVAL;
        goto fail_task;
    }
    state->torrents[state->torrent_count++] = task;
    pthread_mutex_unlock(&state->torrent_lock);

    if (pthread_create(&task->thread, NULL, torrent_worker, task) != 0) {
        pthread_mutex_lock(&state->torrent_lock);
        state->torrent_count--;
        pthread_mutex_unlock(&state->torrent_lock);
        *error = NAUT_ERR_NOMEM;
        goto fail_task;
    }
    task->thread_started = true;
    *error = NAUT_OK;
    return torrent_json(task);

fail_task:
    if (is_temp) unlink(temp_source);
    for (size_t i = 0; i < task->num_peers; i++) free(task->peers[i]);
    free(task->peers);
    free(task->source);
    free(task->output_dir);
    pthread_mutex_destroy(&task->lock);
    free(task);
    return NULL;
}

static json_t *rpc_torrents(void *opaque, const json_t *params,
                            naut_err *error) {
    (void)params;
    daemon_state *state = opaque;
    json_t *result = json_array();
    if (!result) { *error = NAUT_ERR_NOMEM; return NULL; }
    pthread_mutex_lock(&state->torrent_lock);
    for (size_t i = 0; i < state->torrent_count; i++) {
        json_t *item = torrent_json(state->torrents[i]);
        if (!item || json_array_append_new(result, item) != 0) {
            json_decref(item);
            json_decref(result);
            pthread_mutex_unlock(&state->torrent_lock);
            *error = NAUT_ERR_NOMEM;
            return NULL;
        }
    }
    pthread_mutex_unlock(&state->torrent_lock);
    *error = NAUT_OK;
    return result;
}

static bool parse_torrent_id(const json_t *params, uint64_t *id) {
    if (!json_is_object(params)) return false;
    json_t *value = json_object_get(params, "torrent_id");
    if (!json_is_integer(value) || json_integer_value(value) < 0) return false;
    *id = (uint64_t)json_integer_value(value);
    return true;
}

static json_t *rpc_torrent(void *opaque, const json_t *params,
                           naut_err *error) {
    daemon_state *state = opaque;
    uint64_t id;
    if (!parse_torrent_id(params, &id)) {
        *error = NAUT_ERR_INVAL;
        return NULL;
    }
    pthread_mutex_lock(&state->torrent_lock);
    torrent_task *task = find_torrent_locked(state, id);
    json_t *result = task ? torrent_json(task) : NULL;
    pthread_mutex_unlock(&state->torrent_lock);
    *error = task ? (result ? NAUT_OK : NAUT_ERR_NOMEM) : NAUT_ERR_NOTFOUND;
    return result;
}

static json_t *rpc_remove_torrent(void *opaque, const json_t *params,
                                  naut_err *error) {
    daemon_state *state = opaque;
    uint64_t id;
    if (!parse_torrent_id(params, &id)) {
        *error = NAUT_ERR_INVAL;
        return NULL;
    }
    pthread_mutex_lock(&state->torrent_lock);
    torrent_task *task = find_torrent_locked(state, id);
    if (task) {
        pthread_mutex_lock(&task->lock);
        task->stop_requested = true;
        task->remove_requested = true;
        if (!task->thread_done)
            task->state = TORRENT_STOPPING;
        pthread_mutex_unlock(&task->lock);
    }
    pthread_mutex_unlock(&state->torrent_lock);
    if (!task) {
        *error = NAUT_ERR_NOTFOUND;
        return NULL;
    }
    *error = NAUT_OK;
    return torrent_json(task);
}

static json_t *rpc_load_script(void *opaque, const json_t *params,
                               naut_err *error) {
    daemon_state *state = opaque;
    const char *path = json_is_object(params)
        ? json_string_value(json_object_get(params, "path")) : NULL;
    if (!path || !*path) { *error = NAUT_ERR_INVAL; return NULL; }
    naut_script *script = naut_script_create(
        state->events, path, 256, queue_move, state, error);
    if (!script) return NULL;
    naut_script *old = state->script;
    state->script = script;
    naut_script_destroy(old);
    *error = NAUT_OK;
    return json_string(path);
}

static json_t *rpc_unload_script(void *opaque, const json_t *params,
                                 naut_err *error) {
    (void)params;
    daemon_state *state = opaque;
    naut_script *old = state->script;
    state->script = NULL;
    naut_script_destroy(old);
    *error = NAUT_OK;
    return json_true();
}

static naut_err queue_move(void *opaque, uint64_t torrent_id,
                           uint32_t file_index, const char *destination) {
    daemon_state *state = opaque;
    if (!destination || strlen(destination) >= PATH_MAX)
        return NAUT_ERR_RANGE;
    pthread_mutex_lock(&state->torrent_lock);
    torrent_task *task = find_torrent_locked(state, torrent_id);
    if (!task) {
        pthread_mutex_unlock(&state->torrent_lock);
        return NAUT_ERR_NOTFOUND;
    }
    pthread_mutex_lock(&task->lock);
    pthread_mutex_unlock(&state->torrent_lock);
    if (task->remove_requested) {
        pthread_mutex_unlock(&task->lock);
        return NAUT_ERR_NOTFOUND;
    }
    if (task->move_count == MOVE_QUEUE_CAPACITY) {
        pthread_mutex_unlock(&task->lock);
        return NAUT_ERR_FULL;
    }
    size_t tail = (task->move_head + task->move_count) % MOVE_QUEUE_CAPACITY;
    task->moves[tail].file_index = file_index;
    snprintf(task->moves[tail].destination,
             sizeof task->moves[tail].destination, "%s", destination);
    task->move_count++;
    pthread_mutex_unlock(&task->lock);
    return NAUT_OK;
}

static void broadcast_event(void *opaque, const naut_event *event) {
    daemon_state *state = opaque;
    json_t *payload = naut_rpc_event_json(event);
    if (!payload) return;
    pthread_mutex_lock(&state->subscriber_lock);
    for (size_t i = 0; i < state->subscriber_count;) {
        if (naut_rpc_send_json(state->subscribers[i], NAUT_RPC_EVENT,
                               payload) == NAUT_OK) {
            i++;
            continue;
        }
        close(state->subscribers[i]);
        state->subscribers[i] =
            state->subscribers[--state->subscriber_count];
    }
    pthread_mutex_unlock(&state->subscriber_lock);
    json_decref(payload);
}

static int listen_unix(const char *path) {
    int fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (fd < 0) return -1;
    struct sockaddr_un address;
    memset(&address, 0, sizeof address);
    address.sun_family = AF_UNIX;
    if (strlen(path) >= sizeof address.sun_path) {
        close(fd);
        return -1;
    }
    strcpy(address.sun_path, path);
    unlink(path);
    if (bind(fd, (struct sockaddr *)&address, sizeof address) != 0 ||
        listen(fd, 32) != 0) {
        close(fd);
        return -1;
    }
    return fd;
}

static json_t *response(bool ok, json_t *result, naut_err error) {
    json_t *reply = json_object();
    if (!reply) return NULL;
    json_object_set_new(reply, "ok", json_boolean(ok));
    if (ok) {
        json_object_set(reply, "result", result ? result : json_null());
    } else {
        json_object_set_new(reply, "code", json_integer(error));
        json_object_set_new(reply, "error",
                            json_string(naut_strerror(error)));
    }
    return reply;
}

static void add_subscriber(daemon_state *state, int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags >= 0) fcntl(fd, F_SETFL, flags | O_NONBLOCK);
    pthread_mutex_lock(&state->subscriber_lock);
    if (state->subscriber_count < MAX_SUBSCRIBERS) {
        state->subscribers[state->subscriber_count++] = fd;
        fd = -1;
    }
    pthread_mutex_unlock(&state->subscriber_lock);
    if (fd >= 0) close(fd);
}

static void handle_client(daemon_state *state, int fd) {
    struct timeval timeout = {.tv_sec = 2};
    setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout);
    naut_rpc_frame_type type;
    json_t *request = NULL;
    naut_err error = naut_rpc_recv_json(fd, &type, &request);
    if (error != NAUT_OK || type != NAUT_RPC_REQUEST ||
        !json_is_object(request)) {
        json_decref(request);
        close(fd);
        return;
    }
    const char *method =
        json_string_value(json_object_get(request, "method"));
    json_t *params = json_object_get(request, "params");
    if (method && strcmp(method, "subscribe") == 0) {
        json_t *subscribed = json_string("subscribed");
        json_t *reply = response(true, subscribed, NAUT_OK);
        json_decref(subscribed);
        if (reply &&
            naut_rpc_send_json(fd, NAUT_RPC_RESPONSE, reply) == NAUT_OK)
            add_subscriber(state, fd);
        else
            close(fd);
        json_decref(reply);
        json_decref(request);
        return;
    }
    if (!method) error = NAUT_ERR_INVAL;
    json_t *result = method
        ? naut_rpc_dispatch(state->rpc, method, params, &error) : NULL;
    json_t *reply = response(error == NAUT_OK && result, result, error);
    json_decref(result);
    if (reply) {
        naut_rpc_send_json(fd, NAUT_RPC_RESPONSE, reply);
        json_decref(reply);
    }
    json_decref(request);
    close(fd);
}

static bool register_commands(daemon_state *state) {
    return naut_rpc_register(state->rpc, "ping", rpc_ping, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "status", rpc_status, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "plugins", rpc_plugins, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "emit", rpc_emit, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "add_torrent", rpc_add_torrent, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "torrents", rpc_torrents, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "torrent", rpc_torrent, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "remove_torrent", rpc_remove_torrent, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "load_script", rpc_load_script, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "unload_script", rpc_unload_script, state) == NAUT_OK &&
           naut_rpc_register(state->rpc, "shutdown", rpc_shutdown, state) == NAUT_OK;
}

static void stop_torrents(daemon_state *state) {
    pthread_mutex_lock(&state->torrent_lock);
    for (size_t i = 0; i < state->torrent_count; i++) {
        torrent_task *task = state->torrents[i];
        pthread_mutex_lock(&task->lock);
        task->stop_requested = true;
        if (task->state == TORRENT_RUNNING)
            task->state = TORRENT_STOPPING;
        pthread_mutex_unlock(&task->lock);
    }
    pthread_mutex_unlock(&state->torrent_lock);
}

static void destroy_torrent(torrent_task *task) {
    if (task->thread_started) pthread_join(task->thread, NULL);
    /* uploaded torrents live in a daemon-owned temp file; remove it now that
     * the worker has finished reading it */
    if (task->source_is_temp && task->source) unlink(task->source);
    for (size_t p = 0; p < task->num_peers; p++) free(task->peers[p]);
    free(task->peers);
    free(task->source);
    free(task->output_dir);
    pthread_mutex_destroy(&task->lock);
    free(task);
}

static void reap_torrents(daemon_state *state) {
    for (;;) {
        torrent_task *task = NULL;
        pthread_mutex_lock(&state->torrent_lock);
        for (size_t i = 0; i < state->torrent_count; i++) {
            torrent_task *candidate = state->torrents[i];
            pthread_mutex_lock(&candidate->lock);
            bool reap = candidate->remove_requested &&
                        candidate->thread_done;
            pthread_mutex_unlock(&candidate->lock);
            if (!reap) continue;
            task = candidate;
            state->torrents[i] =
                state->torrents[--state->torrent_count];
            break;
        }
        pthread_mutex_unlock(&state->torrent_lock);
        if (!task) return;
        destroy_torrent(task);
    }
}

static void destroy_torrents(daemon_state *state) {
    for (size_t i = 0; i < state->torrent_count; i++)
        destroy_torrent(state->torrents[i]);
    state->torrent_count = 0;
}

static void usage(const char *program) {
    fprintf(stderr,
        "usage: %s [--socket PATH] [--plugin PATH]... [--script PATH]\n",
        program);
}

int main(int argc, char **argv) {
    const char *socket_path = DEFAULT_SOCKET;
    const char *script_path = NULL;
    const char *plugin_paths[64];
    size_t plugin_count = 0;
    for (int i = 1; i < argc; i++) {
        if (strcmp(argv[i], "--socket") == 0 && i + 1 < argc)
            socket_path = argv[++i];
        else if (strcmp(argv[i], "--plugin") == 0 && i + 1 < argc &&
                 plugin_count < NAUT_ARRAY_LEN(plugin_paths))
            plugin_paths[plugin_count++] = argv[++i];
        else if (strcmp(argv[i], "--script") == 0 && i + 1 < argc)
            script_path = argv[++i];
        else {
            usage(argv[0]);
            return 2;
        }
    }

    signal(SIGINT, on_signal);
    signal(SIGTERM, on_signal);
    signal(SIGPIPE, SIG_IGN);
    daemon_state state = {0};
    state.next_torrent_id = 1;
    pthread_mutex_init(&state.torrent_lock, NULL);
    pthread_mutex_init(&state.subscriber_lock, NULL);
    state.events = naut_event_bus_create();
    state.rpc = naut_rpc_registry_create();
    state.plugins = naut_plugin_manager_create(state.rpc, state.events);
    if (!state.events || !state.rpc || !state.plugins ||
        !register_commands(&state)) {
        fprintf(stderr, "nautd: failed to initialize control plane\n");
        return 1;
    }
    for (size_t i = 0; i < plugin_count; i++) {
        if (naut_plugin_load(state.plugins, plugin_paths[i]) != NAUT_OK) {
            fprintf(stderr, "nautd: failed to load plugin %s\n",
                    plugin_paths[i]);
            return 1;
        }
    }
    if (script_path) {
        naut_err error;
        state.script = naut_script_create(state.events, script_path, 256,
                                          queue_move, &state, &error);
        if (!state.script) {
            fprintf(stderr, "nautd: failed to load script %s: %s\n",
                    script_path, naut_strerror(error));
            return 1;
        }
    }
    uint64_t event_subscription;
    if (naut_event_subscribe(state.events, broadcast_event, &state,
                             &event_subscription) != NAUT_OK)
        return 1;
    int listener = listen_unix(socket_path);
    if (listener < 0) {
        perror("nautd: listen");
        return 1;
    }
    NAUT_INFO("nautd listening on %s", socket_path);
    while (!state.stopping && !interrupted) {
        struct pollfd pollfd = {.fd = listener, .events = POLLIN};
        int ready = poll(&pollfd, 1, 100);
        if (ready > 0 && (pollfd.revents & POLLIN)) {
            int client = accept(listener, NULL, NULL);
            if (client >= 0) handle_client(&state, client);
        } else if (ready < 0 && errno != EINTR) {
            break;
        }
        reap_torrents(&state);
    }

    close(listener);
    unlink(socket_path);
    /* Tear down plugins first: a plugin like webui runs its own threads that
     * call back into the daemon via RPC, so it must be stopped (and its
     * threads joined) before we free the torrent tasks those calls touch. */
    naut_plugin_manager_destroy(state.plugins);
    state.plugins = NULL;
    naut_script_destroy(state.script);
    state.script = NULL;
    stop_torrents(&state);
    destroy_torrents(&state);
    naut_event_unsubscribe(state.events, event_subscription);
    pthread_mutex_lock(&state.subscriber_lock);
    for (size_t i = 0; i < state.subscriber_count; i++)
        close(state.subscribers[i]);
    pthread_mutex_unlock(&state.subscriber_lock);
    naut_rpc_registry_destroy(state.rpc);
    naut_event_bus_destroy(state.events);
    pthread_mutex_destroy(&state.subscriber_lock);
    pthread_mutex_destroy(&state.torrent_lock);
    return 0;
}