/* 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/worker.h"

#include <arpa/inet.h>
#include <errno.h>
#include <poll.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 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;
    char     name[40];
    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;
    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 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 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;
    req.key = (uint32_t)rand();

    for (size_t i = 0; i < num_trackers; i++) {
        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;
        }
        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);
    }
    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_to(const endpoint_t *ep) {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) 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) { close(fd); return -1; }
    int one = 1; setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
    return fd;
}

/* 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;
        }
        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++;
                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;
}

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);

    uint8_t peerid[20]; memcpy(peerid, "-NT0001-", 8);
    srand((unsigned)time(NULL) ^ (unsigned)getpid());
    for (int i = 8; i < 20; i++) peerid[i] = (uint8_t)(rand() & 0xff);

    bool from_magnet = strncmp(argv[1], "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(argv[1], &magnet) != NAUT_OK ||
            !magnet.has_v1) {
            NAUT_ERROR("magnet must contain a v1 btih hash");
            naut_magnet_free(&magnet);
            return 1;
        }
    } else {
        size_t tlen;
        uint8_t *tor = slurp(argv[1], &tlen);
        if (!tor) { NAUT_ERROR("read torrent"); return 1; }
        if (naut_metainfo_parse(tor, tlen, &mi) != NAUT_OK) {
            NAUT_ERROR("parse torrent");
            free(tor);
            return 1;
        }
        free(tor);
    }

    endpoint_t *endpoints = NULL;
    size_t neps = 0, epcap = 0;
    if (argc > 3) {
        for (int i = 3; i < argc; i++) {
            naut_peer_addr addr;
            if (!endpoint_parse(argv[i], &addr)) {
                NAUT_WARN("invalid peer address: %s", argv[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 1;
            }
        }
    } else {
        const uint8_t *hash =
            from_magnet ? magnet.infohash_v1 : mi.infohash_v1;
        char *const *trackers =
            from_magnet ? magnet.trackers : mi.trackers;
        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, 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 1;
        }
    }

    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 1;
        }
        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(argc > 3 ? "no valid peer addresses" :
                              "tracker and DHT discovery returned no peers");
        naut_metainfo_free(&mi);
        free(endpoints);
        return 1;
    }

    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, argv[2], &storage_opts, &err);
    if (!st) {
        NAUT_ERROR("storage: %s", naut_strerror(err));
        naut_metainfo_free(&mi);
        free(endpoints);
        return 1;
    }
    naut_download *d = naut_download_create(&mi, st);
    if (!d) {
        naut_storage_close(st);
        naut_metainfo_free(&mi);
        free(endpoints);
        return 1;
    }
    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 1;
    }
    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 1;
    }
    for (int i = 0; i < npeers; i++) peers[i].fd = -1;

    int active = 0;
    for (int i = 0; i < npeers; i++) {
        int fd = connect_to(&endpoints[i]);
        if (fd < 0) {
            NAUT_WARN("connect %s failed", endpoints[i].name);
            peers[i].dead = true;
            continue;
        }
        peer_t *p = &peers[i];
        p->fd = fd;
        snprintf(p->name, sizeof p->name, "%s", endpoints[i].name);
        p->peer_choking = true;
        naut_pipeline_init(&p->pipeline, NAUT_BLOCK, 4, 1024, 32);
        p->rcap = 1 << 18;
        p->rbuf = malloc(p->rcap);
        if (!p->rbuf || naut_bitfield_init(&p->have, mi.num_pieces) != NAUT_OK) {
            free(p->rbuf);
            p->rbuf = NULL;
            close(fd);
            p->fd = -1;
            p->dead = true;
            continue;
        }
        uint8_t hs[NAUT_HANDSHAKE_LEN];
        naut_peer_handshake_build(hs, mi.infohash_v1, peerid, EXT_RESERVED);
        uint8_t intr[5]; naut_peer_msg_simple(intr, NAUT_MSG_INTERESTED);
        uint8_t *ext = NULL;
        size_t ext_len = 0;
        naut_err ext_error = naut_ext_build_handshake(
            NAUT_EXT_UT_METADATA, NAUT_EXT_UT_PEX, 0, 0, &ext, &ext_len);
        bool sent = ext_error == NAUT_OK &&
                    send_all(fd, hs, sizeof hs) &&
                    send_all(fd, ext, ext_len) &&
                    send_all(fd, intr, 5);
        free(ext);
        if (!sent) {
            peer_drop(d, p);
            continue;
        }
        active++;
    }
    free(endpoints);
    if (!active) {
        NAUT_ERROR("no peers reachable");
        goto done;
    }
    NAUT_INFO("swarm: %d peers, %u pieces, %lld bytes", active, mi.num_pieces, (long long)mi.total_length);

    double t0 = now();
    naut_err run_error = NAUT_OK;
    while (!naut_download_complete(d) && run_error == NAUT_OK) {
        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);
            break;
        }
        int r = poll(pfd, nf, 2000);
        if (r < 0) { if (errno == EINTR) continue; 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)" : "");
    } 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);
    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);
    return ok ? 0 : 1;
}