#include "naut/event.h"
#include "naut/log.h"
#include "naut/metainfo.h"
#include "naut/plugin.h"
#include "naut/rpc.h"
#include "naut/script.h"
#include "naut/session.h"
#include "naut/storage.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

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

typedef struct {
    naut_event_bus *events;
    naut_rpc_registry *rpc;
    naut_plugin_manager *plugins;
    naut_script *script;
    naut_session *session;
    pthread_mutex_t move_lock;
    move_command moves[MOVE_QUEUE_CAPACITY];
    size_t move_head;
    size_t move_count;
    uint64_t moves_processed;
    pthread_mutex_t subscriber_lock;
    int subscribers[MAX_SUBSCRIBERS];
    size_t subscriber_count;
    bool stopping;
} daemon_state;

static volatile sig_atomic_t interrupted;

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

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);
    pthread_mutex_lock(&state->move_lock);
    uint64_t moves = state->moves_processed;
    size_t pending = state->move_count;
    pthread_mutex_unlock(&state->move_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(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 uint8_t *read_file(const char *path, size_t *len) {
    FILE *f = fopen(path, "rb");
    if (!f) return NULL;
    if (fseek(f, 0, SEEK_END) != 0) { fclose(f); return NULL; }
    long n = ftell(f);
    if (n < 0 || fseek(f, 0, SEEK_SET) != 0) { fclose(f); return NULL; }
    uint8_t *buf = malloc((size_t)n);
    if (!buf) { fclose(f); return NULL; }
    if (fread(buf, 1, (size_t)n, f) != (size_t)n) {
        free(buf); fclose(f); return NULL;
    }
    fclose(f);
    *len = (size_t)n;
    return buf;
}

/* add_torrent {torrent_id, torrent: <.torrent path>, root: <output dir>} opens
 * the torrent's storage and registers it so move_file can later relocate one of
 * its files. This is the control-plane seam that binds a script's move command
 * to a concrete naut_storage; it runs on the daemon owner thread. */
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; }
    json_int_t torrent_id =
        json_integer_value(json_object_get(params, "torrent_id"));
    const char *torrent_path =
        json_string_value(json_object_get(params, "torrent"));
    const char *root = json_string_value(json_object_get(params, "root"));
    if (torrent_id < 0 || !torrent_path || !root) {
        *error = NAUT_ERR_INVAL;
        return NULL;
    }
    if (naut_session_has(state->session, (uint64_t)torrent_id)) {
        *error = NAUT_ERR_INVAL;
        return NULL;
    }
    size_t len = 0;
    uint8_t *raw = read_file(torrent_path, &len);
    if (!raw) { *error = NAUT_ERR_IO; return NULL; }
    naut_metainfo mi;
    naut_err err = naut_metainfo_parse(raw, len, &mi);
    free(raw);
    if (err != NAUT_OK) { *error = err; return NULL; }
    naut_storage *storage =
        naut_storage_open(mi.files, mi.num_files, root, &err);
    if (!storage) {
        naut_metainfo_free(&mi);
        *error = err != NAUT_OK ? err : NAUT_ERR_IO;
        return NULL;
    }
    naut_metainfo_free(&mi);
    err = naut_session_add(state->session, (uint64_t)torrent_id, storage);
    if (err != NAUT_OK) {
        naut_storage_close(storage);
        *error = err;
        return NULL;
    }
    *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;
    pthread_mutex_lock(&state->move_lock);
    if (state->move_count == MOVE_QUEUE_CAPACITY) {
        pthread_mutex_unlock(&state->move_lock);
        return NAUT_ERR_FULL;
    }
    size_t tail = (state->move_head + state->move_count) %
                  MOVE_QUEUE_CAPACITY;
    state->moves[tail] = (move_command) {
        .torrent_id = torrent_id,
        .file_index = file_index,
    };
    snprintf(state->moves[tail].destination,
             sizeof state->moves[tail].destination, "%s", destination);
    state->move_count++;
    pthread_mutex_unlock(&state->move_lock);
    return NAUT_OK;
}

static void drain_moves(daemon_state *state) {
    /* Copy each pending command out under the lock, then perform the relocate
     * with the lock released (so the script thread can keep enqueuing). All
     * relocates run on this, the owner thread, as the session requires. */
    for (;;) {
        move_command command;
        pthread_mutex_lock(&state->move_lock);
        if (state->move_count == 0) {
            pthread_mutex_unlock(&state->move_lock);
            return;
        }
        command = state->moves[state->move_head];
        state->move_head = (state->move_head + 1) % MOVE_QUEUE_CAPACITY;
        state->move_count--;
        state->moves_processed++;
        pthread_mutex_unlock(&state->move_lock);

        naut_err err = naut_session_move_file(state->session,
                                              command.torrent_id,
                                              command.file_index,
                                              command.destination);
        if (err == NAUT_OK)
            NAUT_INFO("moved torrent=%llu file=%u -> %s",
                      (unsigned long long)command.torrent_id,
                      command.file_index, command.destination);
        else
            NAUT_WARN("move torrent=%llu file=%u -> %s failed: %s",
                      (unsigned long long)command.torrent_id,
                      command.file_index, command.destination,
                      naut_strerror(err));
    }
}

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, "shutdown", rpc_shutdown, state) == NAUT_OK;
}

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};
    pthread_mutex_init(&state.move_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);
    state.session = naut_session_create();
    if (!state.events || !state.rpc || !state.plugins || !state.session ||
        !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;
        }
        drain_moves(&state);
    }

    close(listener);
    unlink(socket_path);
    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_script_destroy(state.script);   /* joins the script thread */
    drain_moves(&state);                 /* flush any moves it left queued */
    naut_plugin_manager_destroy(state.plugins);
    naut_rpc_registry_destroy(state.rpc);
    naut_session_destroy(state.session);
    naut_event_bus_destroy(state.events);
    pthread_mutex_destroy(&state.subscriber_lock);
    pthread_mutex_destroy(&state.move_lock);
    return 0;
}