/* naut_swarm — Phase 4 gate: download from a SWARM of peers concurrently.
 *
 * A poll()-based multi-socket driver around the same sans-IO peer codec and the
 * multi-peer engine (rarest-first + bounded endgame duplication). Peers can be
 * supplied explicitly for deterministic testing, or discovered from the
 * torrent's HTTP/UDP trackers.
 *
 *   usage: naut_swarm <file.torrent|magnet-uri> <output-dir> [<ip:port> ...]
 */
#include "naut/dht.h"
#include "naut/extension.h"
#include "naut/metainfo.h"
#include "naut/storage.h"
#include "naut/piece.h"
#include "naut/peer.h"
#include "naut/tracker.h"
#include "naut/bitfield.h"
#include "naut/log.h"
#include "naut/pipeline.h"
#include "naut/system.h"
#include "naut/swarm.h"
#include "naut/worker.h"

#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <poll.h>
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>

#define REQUEST_TIMEOUT 15.0
#define CONNECT_TIMEOUT_MS 5000
#define EXT_RESERVED 0x0000000000100000ULL

typedef struct {
    uint32_t piece, begin, length;
    double   sent_at;
} req_t;

typedef struct {
    naut_peer_addr addr;
    char           name[32];
} endpoint_t;

typedef struct {
    int      fd;
    naut_peer_addr addr;
    char     name[40];
    uint8_t  peer_id[20];
    bool     have_peer_id;
    uint8_t *rbuf; size_t rcap, rlen;
    bool     hs_done, peer_choking;
    naut_bitfield have;
    req_t   *inflight; size_t nflight, cflight;
    uint64_t blocks_received;
    uint64_t bytes_received;
    naut_ext_handshake extensions;
    uint64_t pex_received;
    naut_pipeline pipeline;
    bool     dead, availability_removed;
} peer_t;

static double now(void) { struct timespec t; clock_gettime(CLOCK_MONOTONIC, &t); return t.tv_sec + t.tv_nsec*1e-9; }

static void random_bytes(uint8_t *output, size_t length) {
    int fd = open("/dev/urandom", O_RDONLY);
    size_t offset = 0;
    while (fd >= 0 && offset < length) {
        ssize_t count = read(fd, output + offset, length - offset);
        if (count > 0) {
            offset += (size_t)count;
        } else if (count < 0 && errno == EINTR) {
            continue;
        } else {
            break;
        }
    }
    if (fd >= 0) close(fd);
    uint64_t fallback = (uint64_t)(now() * 1e9) ^
                        (uint64_t)(uintptr_t)output ^
                        (uint64_t)getpid();
    while (offset < length) {
        fallback ^= fallback << 13;
        fallback ^= fallback >> 7;
        fallback ^= fallback << 17;
        output[offset++] = (uint8_t)fallback;
    }
}

static void emit_event(const naut_swarm_config *config, naut_event_type type,
                       uint32_t index, const char *message, const char *path) {
    if (!config->events) return;
    naut_event event = {
        .type = type,
        .torrent_id = config->torrent_id,
        .index = index,
        .message = message,
        .path = path,
    };
    naut_event_emit(config->events, &event);
}

static void on_file_complete(void *opaque, uint32_t index, const char *path) {
    const naut_swarm_config *config = opaque;
    char full_path[PATH_MAX];
    const char *event_path = path;
    if (path && path[0] != '/') {
        int length = snprintf(full_path, sizeof full_path, "%s/%s",
                              config->output_dir, path);
        if (length >= 0 && (size_t)length < sizeof full_path)
            event_path = full_path;
    }
    emit_event(config, NAUT_EVENT_FILE_COMPLETE, index, NULL, event_path);
}

static void on_piece_complete(void *opaque, uint32_t index) {
    const naut_swarm_config *config = opaque;
    emit_event(config, NAUT_EVENT_PIECE_COMPLETE, index, NULL, NULL);
}

static void peer_client_label(const peer_t *peer, char out[64]) {
    if (!peer || !peer->have_peer_id) {
        snprintf(out, 64, "Unknown");
        return;
    }
    if (peer->peer_id[0] == '-' && peer->peer_id[7] == '-') {
        char code[3] = {
            (char)peer->peer_id[1],
            (char)peer->peer_id[2],
            0,
        };
        char version[5];
        memcpy(version, peer->peer_id + 3, 4);
        version[4] = 0;
        for (size_t i = 0; i < sizeof version - 1; i++)
            if (!isprint((unsigned char)version[i])) version[i] = '?';
        snprintf(out, 64, "%s %s", code, version);
        return;
    }
    char id[21];
    memcpy(id, peer->peer_id, sizeof peer->peer_id);
    id[20] = 0;
    for (size_t i = 0; i < sizeof id - 1; i++)
        if (!isprint((unsigned char)id[i])) id[i] = '.';
    snprintf(out, 64, "%s", id);
}

static void peer_flags(const peer_t *peer, char out[16]) {
    size_t n = 0;
    if (peer && !peer->peer_choking && n + 1 < 16) out[n++] = 'D';
    if (peer && (peer->extensions.ut_pex || peer->pex_received) && n + 1 < 16)
        out[n++] = 'X';
    out[n] = 0;
}

static void snapshot_peer_stats(naut_swarm_stats *stats,
                                const peer_t *peers, int npeers,
                                uint32_t total_pieces) {
    if (!stats || !peers || npeers <= 0) return;
    for (int i = 0; i < npeers &&
                    stats->peer_count < NAUT_SWARM_MAX_PEER_STATS; i++) {
        const peer_t *peer = &peers[i];
        if (peer->dead || !peer->hs_done) continue;
        naut_swarm_peer_stats *out =
            &stats->peer_stats[stats->peer_count++];
        snprintf(out->ip, sizeof out->ip, "%u.%u.%u.%u",
                 peer->addr.ip[0], peer->addr.ip[1],
                 peer->addr.ip[2], peer->addr.ip[3]);
        out->port = peer->addr.port;
        peer_client_label(peer, out->client);
        snprintf(out->connection, sizeof out->connection, "TCP");
        peer_flags(peer, out->flags);
        size_t have = peer->have.words ? naut_bitfield_count(&peer->have) : 0;
        double ratio = total_pieces
            ? (double)have / (double)total_pieces : 0.0;
        if (ratio > 1.0) ratio = 1.0;
        out->progress = ratio;
        out->relevance = ratio;
        out->dlspeed = peer->pipeline.bytes_per_second;
        out->upspeed = 0.0;
        out->downloaded = peer->bytes_received;
        out->uploaded = 0;
    }
}

static void report_progress(const naut_swarm_config *config,
                            const naut_download *download,
                            const naut_metainfo *metainfo,
                            uint32_t peers_total, uint32_t peers_connecting,
                            uint32_t peers_active, uint32_t peers_failed,
                            const peer_t *peers, int npeers,
                            double started_at) {
    if (!config->on_progress) return;
    naut_swarm_stats stats = {
        .total_bytes = (uint64_t)metainfo->total_length,
        .bytes_done = download ? naut_download_bytes_done(download) : 0,
        .total_pieces = metainfo->num_pieces,
        .pieces_done = download ? naut_download_pieces_done(download) : 0,
        .peers_total = peers_total,
        .peers_connecting = peers_connecting,
        .peers_active = peers_active,
        .peers_failed = peers_failed,
        .elapsed_seconds = now() - started_at,
    };
    snapshot_peer_stats(&stats, peers, npeers, metainfo->num_pieces);
    if (download) {
        stats.piece_state_count = (uint32_t)naut_download_piece_states(
            download, stats.piece_states, NAUT_SWARM_MAX_PIECE_STATS);
    }
    config->on_progress(config->context, &stats);
}

static bool stop_requested(const naut_swarm_config *config) {
    return config->should_stop && config->should_stop(config->context);
}

static void service_control(const naut_swarm_config *config,
                            naut_storage *storage) {
    if (config->on_control) config->on_control(config->context, storage);
}

static uint8_t *slurp(const char *path, size_t *len) {
    FILE *f = fopen(path, "rb"); if (!f) return NULL;
    fseek(f, 0, SEEK_END); long n = ftell(f); fseek(f, 0, SEEK_SET);
    uint8_t *b = malloc(n);
    if (fread(b, 1, n, f) != (size_t)n) { fclose(f); free(b); return NULL; }
    fclose(f); *len = (size_t)n; return b;
}
static bool send_all(int fd, const void *p, size_t n) {
    const uint8_t *b = p;
    while (n) { ssize_t w = send(fd, b, n, MSG_NOSIGNAL);
        if (w < 0) { if (errno == EINTR) continue; return false; } b += w; n -= (size_t)w; }
    return true;
}

static bool endpoint_add(endpoint_t **v, size_t *n, size_t *cap,
                         const naut_peer_addr *addr) {
    if (addr->port == 0) return true;
    for (size_t i = 0; i < *n; i++)
        if ((*v)[i].addr.port == addr->port &&
            memcmp((*v)[i].addr.ip, addr->ip, sizeof addr->ip) == 0)
            return true;
    if (*n == *cap) {
        size_t newcap = *cap ? *cap * 2 : 32;
        endpoint_t *next = realloc(*v, newcap * sizeof(*next));
        if (!next) return false;
        *v = next;
        *cap = newcap;
    }
    endpoint_t *ep = &(*v)[(*n)++];
    ep->addr = *addr;
    snprintf(ep->name, sizeof ep->name, "%u.%u.%u.%u:%u",
             addr->ip[0], addr->ip[1], addr->ip[2], addr->ip[3], addr->port);
    return true;
}

static bool endpoint_parse(const char *s, naut_peer_addr *out) {
    char host[INET_ADDRSTRLEN];
    const char *colon = strrchr(s, ':');
    if (!colon || colon == s || (size_t)(colon - s) >= sizeof host) return false;
    char *end = NULL;
    unsigned long port = strtoul(colon + 1, &end, 10);
    if (!end || *end || port == 0 || port > UINT16_MAX) return false;
    memcpy(host, s, (size_t)(colon - s));
    host[colon - s] = 0;
    struct in_addr ip;
    if (inet_pton(AF_INET, host, &ip) != 1) return false;
    memcpy(out->ip, &ip, sizeof out->ip);
    out->port = (uint16_t)port;
    return true;
}

static bool parse_udp_tracker(const char *url, char *host, size_t hostsz,
                              uint16_t *port) {
    if (strncmp(url, "udp://", 6) != 0) return false;
    const char *start = url + 6;
    const char *slash = strchr(start, '/');
    const char *end = slash ? slash : start + strlen(start);
    const char *colon = memchr(start, ':', (size_t)(end - start));
    if (!colon || colon == start) return false;
    size_t hlen = (size_t)(colon - start);
    if (hlen >= hostsz) return false;
    char pbuf[16];
    size_t plen = (size_t)(end - colon - 1);
    if (plen == 0 || plen >= sizeof pbuf) return false;
    memcpy(host, start, hlen);
    host[hlen] = 0;
    memcpy(pbuf, colon + 1, plen);
    pbuf[plen] = 0;
    char *tail = NULL;
    unsigned long parsed = strtoul(pbuf, &tail, 10);
    if (!tail || *tail || parsed == 0 || parsed > UINT16_MAX) return false;
    *port = (uint16_t)parsed;
    return true;
}

static bool discover_trackers(const uint8_t info_hash[20], uint64_t total_length,
                              char *const *trackers, size_t num_trackers,
                              const uint32_t *tracker_tiers,
                              const uint8_t peerid[20],
                              endpoint_t **eps, size_t *neps, size_t *cap) {
    naut_announce_req req;
    memset(&req, 0, sizeof req);
    memcpy(req.info_hash, info_hash, sizeof req.info_hash);
    memcpy(req.peer_id, peerid, sizeof req.peer_id);
    req.port = 6881;
    req.left = total_length;
    req.event = NAUT_TEV_STARTED;
    req.numwant = 100;
    memcpy(&req.key, peerid + 8, sizeof req.key);

    size_t tier_start = 0;
    while (tier_start < num_trackers) {
        uint32_t tier = tracker_tiers
            ? tracker_tiers[tier_start] : (uint32_t)tier_start;
        size_t tier_end = tier_start + 1;
        if (tracker_tiers)
            while (tier_end < num_trackers &&
                   tracker_tiers[tier_end] == tier)
                tier_end++;

        size_t tier_count = tier_end - tier_start;
        uint32_t random = 0;
        random_bytes((uint8_t *)&random, sizeof random);
        size_t first = tier_count ? random % tier_count : 0;
        bool tier_succeeded = false;
        for (size_t n = 0; n < tier_count; n++) {
            size_t i = tier_start + (first + n) % tier_count;
            const char *tracker = trackers[i];
            naut_tracker_response response;
            naut_err e = NAUT_ERR_INVAL;
            if (strncmp(tracker, "http://", 7) == 0) {
                char url[4096];
                if (naut_tracker_http_url(tracker, &req, url,
                                          sizeof url) != 0)
                    e = naut_tracker_announce_http(url, &response);
            } else if (strncmp(tracker, "udp://", 6) == 0) {
                char host[256];
                uint16_t port;
                if (parse_udp_tracker(tracker, host, sizeof host, &port))
                    e = naut_tracker_announce_udp(host, port, &req, &response);
            } else {
                NAUT_WARN("tracker scheme unsupported: %s", tracker);
                continue;
            }
            if (e != NAUT_OK) {
                NAUT_WARN("tracker announce failed: %s", tracker);
                continue;
            }
            tier_succeeded = true;
            NAUT_INFO("tracker %s returned %zu peers",
                      tracker, response.num_peers);
            for (size_t p = 0; p < response.num_peers; p++) {
                if (!endpoint_add(eps, neps, cap, &response.peers[p])) {
                    naut_tracker_response_free(&response);
                    return false;
                }
            }
            naut_tracker_response_free(&response);
            /* Trackers within a tier are alternatives, not a fan-out set.
             * Once one accepts the announce, do not load the rest. */
            break;
        }
        if (tier_succeeded) break;
        tier_start = tier_end;
    }
    return true;
}

static bool discover_dht(const uint8_t info_hash[20],
                         endpoint_t **eps, size_t *neps, size_t *cap) {
    static const char *defaults[] = {
        "router.bittorrent.com:6881",
        "router.utorrent.com:6881",
        "dht.transmissionbt.com:6881",
    };
    const char *const *bootstrap = defaults;
    size_t num_bootstrap = NAUT_ARRAY_LEN(defaults);
    char *copy = NULL;
    const char *custom[32];

    const char *env = getenv("NAUT_DHT_BOOTSTRAP");
    if (env && *env) {
        copy = strdup(env);
        if (!copy) return false;
        num_bootstrap = 0;
        char *save = NULL;
        for (char *part = strtok_r(copy, ",", &save);
             part && num_bootstrap < NAUT_ARRAY_LEN(custom);
             part = strtok_r(NULL, ",", &save))
            custom[num_bootstrap++] = part;
        bootstrap = custom;
    }

    naut_peer_addr *peers = NULL;
    size_t num_peers = 0;
    naut_err e = num_bootstrap
        ? naut_dht_get_peers(bootstrap, num_bootstrap, info_hash,
                             &peers, &num_peers)
        : NAUT_ERR_INVAL;
    free(copy);
    if (e != NAUT_OK) {
        NAUT_WARN("DHT lookup found no peers");
        return true;
    }
    NAUT_INFO("DHT returned %zu peers", num_peers);
    for (size_t i = 0; i < num_peers; i++) {
        if (!endpoint_add(eps, neps, cap, &peers[i])) {
            free(peers);
            return false;
        }
    }
    free(peers);
    return true;
}

static bool peer_reserve_inflight(peer_t *p) {
    if (p->nflight == p->cflight) {
        size_t cap = p->cflight ? p->cflight * 2 : 64;
        req_t *v = realloc(p->inflight, cap * sizeof(*v));
        if (!v) return false;
        p->inflight = v;
        p->cflight = cap;
    }
    return true;
}

static void peer_add_inflight(peer_t *p, uint32_t piece, uint32_t begin,
                              uint32_t length) {
    p->inflight[p->nflight].piece = piece;
    p->inflight[p->nflight].begin = begin;
    p->inflight[p->nflight].length = length;
    p->inflight[p->nflight].sent_at = now();
    p->nflight++;
}
static bool peer_del_inflight(peer_t *p, uint32_t piece, uint32_t begin,
                              req_t *removed) {
    for (size_t i = 0; i < p->nflight; i++)
        if (p->inflight[i].piece == piece && p->inflight[i].begin == begin) {
            if (removed) *removed = p->inflight[i];
            p->inflight[i] = p->inflight[--p->nflight];
            return true;
        }
    return false;
}

static bool peer_has_request(void *ctx, uint32_t piece, uint32_t begin) {
    peer_t *p = ctx;
    for (size_t i = 0; i < p->nflight; i++)
        if (p->inflight[i].piece == piece && p->inflight[i].begin == begin)
            return true;
    return false;
}

/* return remaining inflight blocks to the picker (choke / disconnect) */
static void peer_release(naut_download *d, peer_t *p) {
    for (size_t i = 0; i < p->nflight; i++)
        naut_download_unrequest(d, p->inflight[i].piece, p->inflight[i].begin);
    p->nflight = 0;
}

static void peer_drop(naut_download *d, peer_t *p) {
    if (!p->availability_removed) {
        naut_download_remove_bitfield(d, &p->have);
        p->availability_removed = true;
    }
    peer_release(d, p);
    if (p->fd >= 0) close(p->fd);
    p->fd = -1;
    p->dead = true;
}

static bool refill_one(naut_download *d, peer_t *p) {
    if (p->dead || !p->hs_done || p->peer_choking ||
        p->nflight >= naut_pipeline_depth(&p->pipeline))
        return false;
    if (!peer_reserve_inflight(p)) {
        peer_drop(d, p);
        return false;
    }
    uint32_t i, b, l;
    if (!naut_download_pick_for_peer(d, &p->have, peer_has_request, p,
                                     &i, &b, &l))
        return false;
    uint8_t req[17];
    naut_peer_msg_request(req, i, b, l);
    if (!send_all(p->fd, req, sizeof req)) {
        naut_download_unrequest(d, i, b);
        peer_drop(d, p);
        return false;
    }
    peer_add_inflight(p, i, b, l);
    return true;
}

static void expire_requests(naut_download *d, peer_t *p, double t) {
    size_t i = 0;
    while (i < p->nflight) {
        if (t - p->inflight[i].sent_at < REQUEST_TIMEOUT) {
            i++;
            continue;
        }
        naut_download_unrequest(d, p->inflight[i].piece, p->inflight[i].begin);
        p->inflight[i] = p->inflight[--p->nflight];
    }
}

static int connect_start(const endpoint_t *ep, bool *connected) {
    *connected = false;
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) return -1;
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0) {
        close(fd);
        return -1;
    }
    struct sockaddr_in a; memset(&a, 0, sizeof a);
    a.sin_family = AF_INET;
    a.sin_port = htons(ep->addr.port);
    memcpy(&a.sin_addr, ep->addr.ip, sizeof ep->addr.ip);
    if (connect(fd, (struct sockaddr *)&a, sizeof a) == 0) {
        *connected = true;
    } else if (errno != EINPROGRESS) {
        close(fd);
        return -1;
    }
    return fd;
}

static bool connect_finish(int fd) {
    int error = 0;
    socklen_t length = sizeof error;
    if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &length) != 0 ||
        error != 0)
        return false;
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags < 0 || fcntl(fd, F_SETFL, flags & ~O_NONBLOCK) != 0)
        return false;
    int one = 1; setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
    return true;
}

static bool peer_start(naut_download *download, const naut_metainfo *metainfo,
                       const uint8_t peerid[20], const endpoint_t *endpoint,
                       peer_t *peer, int fd) {
    peer->fd = fd;
    peer->addr = endpoint->addr;
    snprintf(peer->name, sizeof peer->name, "%s", endpoint->name);
    peer->peer_choking = true;
    naut_pipeline_init(&peer->pipeline, NAUT_BLOCK, 4, 1024, 32);
    peer->rcap = 1 << 18;
    peer->rbuf = malloc(peer->rcap);
    if (!peer->rbuf ||
        naut_bitfield_init(&peer->have, metainfo->num_pieces) != NAUT_OK) {
        free(peer->rbuf);
        peer->rbuf = NULL;
        close(fd);
        peer->fd = -1;
        peer->dead = true;
        return false;
    }
    uint8_t handshake[NAUT_HANDSHAKE_LEN];
    naut_peer_handshake_build(handshake, metainfo->infohash_v1, peerid,
                              EXT_RESERVED);
    uint8_t interested[5];
    naut_peer_msg_simple(interested, NAUT_MSG_INTERESTED);
    uint8_t *extension = NULL;
    size_t extension_length = 0;
    naut_err extension_error = naut_ext_build_handshake(
        NAUT_EXT_UT_METADATA, NAUT_EXT_UT_PEX, 0, 0, &extension,
        &extension_length);
    bool sent = extension_error == NAUT_OK &&
                send_all(fd, handshake, sizeof handshake) &&
                send_all(fd, extension, extension_length) &&
                send_all(fd, interested, sizeof interested);
    free(extension);
    if (!sent) {
        peer_drop(download, peer);
        return false;
    }
    return true;
}

/* process all complete messages currently buffered for peer p */
static void cancel_block(naut_download *d, peer_t *peers, int npeers,
                         peer_t *source, uint32_t piece, uint32_t begin) {
    for (int i = 0; i < npeers; i++) {
        peer_t *p = &peers[i];
        if (p == source || p->dead) continue;
        req_t old;
        if (!peer_del_inflight(p, piece, begin, &old)) continue;
        uint8_t msg[17];
        naut_peer_msg_cancel(msg, old.piece, old.begin, old.length);
        if (!send_all(p->fd, msg, sizeof msg)) peer_drop(d, p);
    }
}

static void cancel_piece(naut_download *d, peer_t *peers, int npeers,
                         uint32_t piece) {
    for (int i = 0; i < npeers; i++) {
        peer_t *p = &peers[i];
        size_t j = 0;
        while (j < p->nflight) {
            req_t old = p->inflight[j];
            if (old.piece != piece) {
                j++;
                continue;
            }
            p->inflight[j] = p->inflight[--p->nflight];
            naut_download_unrequest(d, old.piece, old.begin);
            if (!p->dead) {
                uint8_t msg[17];
                naut_peer_msg_cancel(msg, old.piece, old.begin, old.length);
                if (!send_all(p->fd, msg, sizeof msg)) {
                    peer_drop(d, p);
                    break;
                }
            }
        }
    }
}

static void merge_bitfield(naut_download *d, const naut_metainfo *mi,
                           peer_t *p, const uint8_t *wire, size_t wire_len) {
    naut_bitfield incoming;
    if (naut_bitfield_init(&incoming, mi->num_pieces) != NAUT_OK) {
        peer_drop(d, p);
        return;
    }
    naut_bitfield_from_wire(&incoming, wire, wire_len);
    for (uint32_t piece = 0; piece < mi->num_pieces; piece++) {
        if (naut_bitfield_test(&incoming, piece) &&
            !naut_bitfield_test(&p->have, piece)) {
            naut_bitfield_set(&p->have, piece);
            naut_download_inc_avail(d, piece);
        }
    }
    naut_bitfield_free(&incoming);
}

static naut_err peer_process(naut_download *d, const naut_metainfo *mi,
                             peer_t *peers, int npeers, peer_t *p) {
    size_t pos = 0;
    if (!p->hs_done) {
        if (p->rlen < NAUT_HANDSHAKE_LEN) return NAUT_OK;
        uint8_t ih[20], pid[20];
        if (!naut_peer_handshake_parse(p->rbuf, ih, pid, NULL) ||
            memcmp(ih, mi->infohash_v1, 20) != 0) {
            p->dead = true;
            return NAUT_OK;
        }
        memcpy(p->peer_id, pid, sizeof p->peer_id);
        p->have_peer_id = true;
        pos = NAUT_HANDSHAKE_LEN;
        p->hs_done = true;
    }
    for (;;) {
        naut_msg m;
        int c = naut_peer_msg_parse(p->rbuf + pos, p->rlen - pos, &m);
        if (c == 0) break;
        if (c < 0) { p->dead = true; break; }
        pos += (size_t)c;
        switch (m.type) {
        case NAUT_MSG_BITFIELD:
            merge_bitfield(d, mi, p, m.payload, m.payload_len);
            if (p->dead) goto parsed;
            break;
        case NAUT_MSG_HAVE:
            if (m.index < mi->num_pieces && !naut_bitfield_test(&p->have, m.index)) {
                naut_bitfield_set(&p->have, m.index);
                naut_download_inc_avail(d, m.index);
            }
            break;
        case NAUT_MSG_UNCHOKE: p->peer_choking = false; break;
        case NAUT_MSG_CHOKE:   p->peer_choking = true; peer_release(d, p); break;
        case NAUT_MSG_EXTENDED:
            if (m.payload_len < 1) {
                peer_drop(d, p);
                goto parsed;
            }
            if (m.payload[0] == 0) {
                if (naut_ext_parse_handshake(
                        m.payload + 1, m.payload_len - 1,
                        &p->extensions) != NAUT_OK) {
                    peer_drop(d, p);
                    goto parsed;
                }
            } else if (m.payload[0] == NAUT_EXT_UT_PEX) {
                naut_pex_msg pex;
                if (naut_pex_parse(m.payload + 1, m.payload_len - 1,
                                   &pex) != NAUT_OK) {
                    peer_drop(d, p);
                    goto parsed;
                }
                p->pex_received += pex.num_added;
                naut_pex_free(&pex);
            }
            break;
        case NAUT_MSG_PIECE: {
            req_t request;
            bool expected = peer_del_inflight(p, m.index, m.begin, &request);
            if (expected) {
                naut_pipeline_on_block(&p->pipeline, request.length,
                                       request.sent_at, now());
                naut_download_unrequest(d, m.index, m.begin);
                if (request.length != m.payload_len) {
                    peer_drop(d, p);
                    goto parsed;
                }
            }
            bool pdone = false;
            naut_err e = naut_download_on_block(d, m.index, m.begin, m.payload,
                                                (uint32_t)m.payload_len, &pdone);
            if (e == NAUT_OK) {
                p->blocks_received++;
                p->bytes_received += m.payload_len;
                cancel_block(d, peers, npeers, p, m.index, m.begin);
            } else if (e == NAUT_ERR_PROTO) {
                if (expected) cancel_piece(d, peers, npeers, m.index);
                else peer_drop(d, p);
            } else {
                return e;
            }
            break;
        }
        default: break;
        }
    }
parsed:
    memmove(p->rbuf, p->rbuf + pos, p->rlen - pos);
    p->rlen -= pos;
    return NAUT_OK;
}

naut_err naut_swarm_run(const naut_swarm_config *config) {
    if (!config || !config->source || !*config->source ||
        !config->output_dir || !*config->output_dir)
        return NAUT_ERR_INVAL;

    uint8_t peerid[20];
    memcpy(peerid, "-NT0001-", 8);
    random_bytes(peerid + 8, sizeof peerid - 8);

    bool from_magnet = strncmp(config->source, "magnet:?", 8) == 0;
    naut_metainfo mi;
    memset(&mi, 0, sizeof mi);
    naut_magnet magnet;
    memset(&magnet, 0, sizeof magnet);
    if (from_magnet) {
        if (naut_magnet_parse(config->source, &magnet) != NAUT_OK ||
            !magnet.has_v1) {
            NAUT_ERROR("magnet must contain a v1 btih hash");
            naut_magnet_free(&magnet);
            return NAUT_ERR_INVAL;
        }
    } else {
        size_t tlen;
        uint8_t *tor = slurp(config->source, &tlen);
        if (!tor) { NAUT_ERROR("read torrent"); return NAUT_ERR_IO; }
        if (naut_metainfo_parse(tor, tlen, &mi) != NAUT_OK) {
            NAUT_ERROR("parse torrent");
            free(tor);
            return NAUT_ERR_PROTO;
        }
        free(tor);
    }

    endpoint_t *endpoints = NULL;
    size_t neps = 0, epcap = 0;
    if (config->num_peers > 0) {
        for (size_t i = 0; i < config->num_peers; i++) {
            naut_peer_addr addr;
            if (!endpoint_parse(config->peers[i], &addr)) {
                NAUT_WARN("invalid peer address: %s", config->peers[i]);
                continue;
            }
            if (!endpoint_add(&endpoints, &neps, &epcap, &addr)) {
                NAUT_ERROR("out of memory collecting peers");
                naut_metainfo_free(&mi);
                naut_magnet_free(&magnet);
                free(endpoints);
                return NAUT_ERR_NOMEM;
            }
        }
    } else {
        const uint8_t *hash =
            from_magnet ? magnet.infohash_v1 : mi.infohash_v1;
        char *const *trackers =
            from_magnet ? magnet.trackers : mi.trackers;
        const uint32_t *tracker_tiers =
            from_magnet ? NULL : mi.tracker_tiers;
        size_t num_trackers =
            from_magnet ? magnet.num_trackers : mi.num_trackers;
        uint64_t total = from_magnet ? 0 : (uint64_t)mi.total_length;
        if (!discover_trackers(hash, total, trackers, num_trackers,
                               tracker_tiers, peerid, &endpoints, &neps,
                               &epcap) ||
            (neps == 0 &&
             !discover_dht(hash, &endpoints, &neps, &epcap))) {
            NAUT_ERROR("out of memory collecting discovered peers");
            naut_metainfo_free(&mi);
            naut_magnet_free(&magnet);
            free(endpoints);
            return NAUT_ERR_NOMEM;
        }
    }

    if (from_magnet && neps) {
        uint8_t *info = NULL;
        size_t info_len = 0;
        naut_err metadata_error = NAUT_ERR_EMPTY;
        for (size_t i = 0; i < neps; i++) {
            metadata_error = naut_metadata_fetch(
                &endpoints[i].addr, magnet.infohash_v1, peerid,
                &info, &info_len);
            if (metadata_error == NAUT_OK) break;
            NAUT_WARN("metadata fetch from %s failed", endpoints[i].name);
        }
        if (metadata_error != NAUT_OK ||
            naut_metainfo_parse_info(
                info, info_len, (const char *const *)magnet.trackers,
                magnet.num_trackers, &mi) != NAUT_OK) {
            NAUT_ERROR("unable to fetch valid magnet metadata");
            free(info);
            naut_magnet_free(&magnet);
            free(endpoints);
            return metadata_error != NAUT_OK ? metadata_error :
                                               NAUT_ERR_PROTO;
        }
        free(info);
        NAUT_INFO("magnet metadata verified: %u pieces, %lld bytes",
                  mi.num_pieces, (long long)mi.total_length);
    }
    naut_magnet_free(&magnet);

    if (neps == 0) {
        NAUT_ERROR(config->num_peers > 0 ? "no valid peer addresses" :
                                          "tracker and DHT discovery returned no peers");
        naut_metainfo_free(&mi);
        free(endpoints);
        return NAUT_ERR_NOTFOUND;
    }

    naut_err err;
    naut_storage_opts storage_opts = {
        .direct_io = getenv("NAUT_DIRECT_IO") != NULL,
        .preallocate = true,
    };
    naut_storage *st = naut_storage_open_opts(
        mi.files, mi.num_files, config->output_dir, &storage_opts, &err);
    if (!st) {
        NAUT_ERROR("storage: %s", naut_strerror(err));
        naut_metainfo_free(&mi);
        free(endpoints);
        return err != NAUT_OK ? err : NAUT_ERR_IO;
    }
    naut_download *d = naut_download_create(&mi, st);
    if (!d) {
        naut_storage_close(st);
        naut_metainfo_free(&mi);
        free(endpoints);
        return NAUT_ERR_NOMEM;
    }
    naut_download_set_file_cb(d, on_file_complete, (void *)config);
    naut_download_set_piece_cb(d, on_piece_complete, (void *)config);
    int online_cpus = naut_online_cpus();
    uint32_t worker_count =
        (uint32_t)NAUT_MAX(1, NAUT_MIN(8, online_cpus / 2));
    if (getenv("NAUT_WORKERS")) {
        unsigned long configured = strtoul(getenv("NAUT_WORKERS"), NULL, 10);
        if (configured > 0 && configured <= 256)
            worker_count = (uint32_t)configured;
    }
    int worker_cpu_base = getenv("NAUT_WORKER_CPU_BASE")
        ? atoi(getenv("NAUT_WORKER_CPU_BASE")) : -1;
    naut_worker_pool *workers =
        naut_worker_pool_create(worker_count, 1024, worker_cpu_base);
    if (!workers) {
        NAUT_ERROR("unable to create hash worker pool");
        naut_download_destroy(d);
        naut_storage_close(st);
        naut_metainfo_free(&mi);
        free(endpoints);
        return NAUT_ERR_NOMEM;
    }
    naut_download_set_worker_pool(d, workers);

    int npeers = (int)neps;
    peer_t *peers = calloc(neps, sizeof(*peers));
    struct pollfd *pfd = calloc(neps + 1, sizeof(*pfd));
    int *idx_map = calloc(neps + 1, sizeof(*idx_map));
    if (!peers || !pfd || !idx_map) {
        NAUT_ERROR("out of memory creating swarm");
        free(peers); free(pfd); free(idx_map);
        naut_worker_pool_destroy(workers);
        naut_download_destroy(d);
        naut_storage_close(st);
        naut_metainfo_free(&mi);
        free(endpoints);
        return NAUT_ERR_NOMEM;
    }
    for (int i = 0; i < npeers; i++) peers[i].fd = -1;

    double t0 = now();
    naut_err run_error = NAUT_OK;
    bool cancelled = false;
    uint32_t connecting = 0;
    uint32_t active = 0;
    uint32_t failed = 0;
    struct pollfd *connect_fds = pfd;
    for (int i = 0; i < npeers; i++) connect_fds[i].fd = -1;
    report_progress(config, d, &mi, (uint32_t)npeers, 0, 0, 0,
                    peers, npeers, t0);
    for (int i = 0; i < npeers; i++) {
        bool connected = false;
        int fd = connect_start(&endpoints[i], &connected);
        if (fd < 0) {
            NAUT_WARN("connect %s failed", endpoints[i].name);
            peers[i].dead = true;
            failed++;
            continue;
        }
        peers[i].fd = fd;
        if (connected) {
            if (!connect_finish(fd) ||
                !peer_start(d, &mi, peerid, &endpoints[i], &peers[i], fd)) {
                NAUT_WARN("connect %s failed", endpoints[i].name);
                if (peers[i].fd >= 0) close(peers[i].fd);
                peers[i].fd = -1;
                peers[i].dead = true;
                failed++;
                continue;
            }
            active++;
            emit_event(config, NAUT_EVENT_PEER_CONNECTED, 0,
                       peers[i].name, NULL);
        } else {
            connect_fds[i].fd = fd;
            connect_fds[i].events = POLLOUT;
            connecting++;
        }
    }
    double connect_deadline = now() + CONNECT_TIMEOUT_MS / 1000.0;
    while (connecting > 0 && now() < connect_deadline &&
           !stop_requested(config)) {
        report_progress(config, d, &mi, (uint32_t)npeers, connecting,
                        active, failed, peers, npeers, t0);
        int ready = poll(connect_fds, (nfds_t)neps, 100);
        if (ready < 0) {
            if (errno == EINTR) continue;
            run_error = NAUT_ERR_IO;
            break;
        }
        if (ready == 0) continue;
        for (int i = 0; i < npeers; i++) {
            if (connect_fds[i].fd < 0 ||
                !(connect_fds[i].revents &
                  (POLLOUT | POLLERR | POLLHUP | POLLNVAL)))
                continue;
            int fd = connect_fds[i].fd;
            connect_fds[i].fd = -1;
            connecting--;
            if (!connect_finish(fd) ||
                !peer_start(d, &mi, peerid, &endpoints[i], &peers[i], fd)) {
                NAUT_WARN("connect %s failed", endpoints[i].name);
                if (peers[i].fd >= 0) close(peers[i].fd);
                peers[i].fd = -1;
                peers[i].dead = true;
                failed++;
                continue;
            }
            active++;
            emit_event(config, NAUT_EVENT_PEER_CONNECTED, 0,
                       peers[i].name, NULL);
        }
    }
    for (int i = 0; i < npeers; i++) {
        if (connect_fds[i].fd < 0) continue;
        close(connect_fds[i].fd);
        connect_fds[i].fd = -1;
        peers[i].fd = -1;
        peers[i].dead = true;
        connecting--;
        failed++;
        NAUT_WARN("connect %s timed out", endpoints[i].name);
    }
    free(endpoints);
    if (!active) {
        NAUT_ERROR("no peers reachable");
        if (run_error == NAUT_OK) run_error = NAUT_ERR_IO;
        goto done;
    }
    NAUT_INFO("swarm: %u/%d peers connected, %u pieces, %lld bytes",
              active, npeers, mi.num_pieces, (long long)mi.total_length);

    emit_event(config, NAUT_EVENT_TORRENT_ADDED, 0, NULL, NULL);
    report_progress(config, d, &mi, (uint32_t)npeers, 0, active, failed,
                    peers, npeers, t0);
    while (!naut_download_complete(d) && run_error == NAUT_OK) {
        service_control(config, st);
        if (stop_requested(config)) {
            cancelled = true;
            break;
        }
        int nf = 0;
        for (int i = 0; i < npeers; i++) {
            if (peers[i].dead) continue;
            pfd[nf].fd = peers[i].fd; pfd[nf].events = POLLIN; pfd[nf].revents = 0;
            idx_map[nf] = i; nf++;
        }
        int live_peers = nf;
        pfd[nf].fd = naut_worker_eventfd(workers);
        pfd[nf].events = POLLIN;
        pfd[nf].revents = 0;
        idx_map[nf] = -1;
        nf++;
        if (live_peers == 0) {
            uint32_t completed = 0;
            run_error = naut_download_poll(d, &completed);
            if (naut_download_complete(d)) break;
            NAUT_ERROR("all peers gone (%.0f%% done)",
                       100.0 * naut_download_pieces_done(d) / mi.num_pieces);
            run_error = NAUT_ERR_IO;
            break;
        }
        report_progress(config, d, &mi, (uint32_t)npeers, 0,
                        (uint32_t)live_peers,
                        (uint32_t)npeers - (uint32_t)live_peers,
                        peers, npeers, t0);
        int r = poll(pfd, nf, 200);
        if (r < 0) {
            if (errno == EINTR) continue;
            run_error = NAUT_ERR_IO;
            break;
        }

        for (int k = 0; k < nf; k++) {
            if (idx_map[k] < 0) {
                if (pfd[k].revents & POLLIN) {
                    uint64_t count;
                    (void)read(pfd[k].fd, &count, sizeof count);
                    uint32_t completed = 0;
                    run_error = naut_download_poll(d, &completed);
                }
                continue;
            }
            peer_t *p = &peers[idx_map[k]];
            if (!(pfd[k].revents & (POLLIN | POLLHUP | POLLERR))) continue;
            if (p->rlen == p->rcap) {
                size_t cap = p->rcap * 2;
                uint8_t *buf = realloc(p->rbuf, cap);
                if (!buf) { peer_drop(d, p); continue; }
                p->rbuf = buf;
                p->rcap = cap;
            }
            ssize_t got = recv(p->fd, p->rbuf + p->rlen, p->rcap - p->rlen, 0);
            if (got <= 0) { peer_drop(d, p); continue; }
            p->rlen += (size_t)got;
            run_error = peer_process(d, &mi, peers, npeers, p);
            if (run_error != NAUT_OK) break;
            if (p->dead) peer_drop(d, p);
        }
        if (run_error != NAUT_OK) break;
        double t = now();
        for (int i = 0; i < npeers; i++) {
            peer_t *p = &peers[i];
            if (!p->dead) expire_requests(d, p, t);
        }
        for (;;) {
            bool sent = false;
            for (int i = 0; i < npeers; i++)
                if (refill_one(d, &peers[i])) sent = true;
            if (!sent) break;
        }
    }

    double dt = now() - t0;
    bool ok = naut_download_complete(d);
    if (ok) {
        double mb = (double)mi.total_length / 1e6;
        NAUT_INFO("COMPLETE: %u/%u pieces from swarm in %.2fs (%.1f MB/s), all SHA-1 verified%s",
                  naut_download_pieces_done(d), mi.num_pieces, dt, mb/dt,
                  naut_download_in_endgame(d) ? " (passed through endgame)" : "");
        report_progress(config, d, &mi, (uint32_t)npeers, 0, 0,
                        (uint32_t)npeers, peers, npeers, t0);
        emit_event(config, NAUT_EVENT_TORRENT_FINISHED, 0, NULL, NULL);
        while (config->keep_alive && !stop_requested(config)) {
            service_control(config, st);
            usleep(100000);
        }
    } else {
        if (run_error != NAUT_OK)
            NAUT_ERROR("swarm stopped: %s", naut_strerror(run_error));
        NAUT_ERROR("INCOMPLETE: %u/%u pieces", naut_download_pieces_done(d), mi.num_pieces);
    }

done:
    ok = naut_download_complete(d);
    service_control(config, st);
    naut_storage_sync(st);
    for (int i = 0; i < npeers; i++) {
        if (peers[i].blocks_received)
            NAUT_INFO("peer %s delivered %llu blocks (pipeline %u, RTT %.1f ms, %.1f MB/s)",
                      peers[i].name,
                      (unsigned long long)peers[i].blocks_received,
                      naut_pipeline_depth(&peers[i].pipeline),
                      peers[i].pipeline.rtt_seconds * 1000.0,
                      peers[i].pipeline.bytes_per_second / 1e6);
        if (peers[i].pex_received)
            NAUT_INFO("peer %s advertised %llu peers through PEX",
                      peers[i].name,
                      (unsigned long long)peers[i].pex_received);
        if (!peers[i].dead) peer_drop(d, &peers[i]);
        free(peers[i].rbuf); free(peers[i].inflight);
        if (peers[i].have.words) naut_bitfield_free(&peers[i].have);
    }
    free(peers); free(pfd); free(idx_map);
    naut_worker_pool_destroy(workers);
    naut_download_destroy(d); naut_storage_close(st); naut_metainfo_free(&mi);
    if (ok) return NAUT_OK;
    if (cancelled) return NAUT_ERR_AGAIN;
    return run_error != NAUT_OK ? run_error : NAUT_ERR_IO;
}

#ifndef NAUT_SWARM_LIBRARY
int main(int argc, char **argv) {
    if (argc < 3) {
        fprintf(stderr,
                "usage: %s <file.torrent|magnet-uri> <out-dir> [ip:port ...]\n",
                argv[0]);
        return 2;
    }
    naut_log_set_level(NAUT_LOG_INFO);
    naut_swarm_config config = {
        .source = argv[1],
        .output_dir = argv[2],
        .peers = argc > 3 ? (const char *const *)&argv[3] : NULL,
        .num_peers = argc > 3 ? (size_t)(argc - 3) : 0,
    };
    return naut_swarm_run(&config) == NAUT_OK ? 0 : 1;
}
#endif