/* naut_swarm — multi-peer download driver built on the torrent-peer engine.
 *
 * The engine (external/torrent-peer, engine.h) owns all peer sockets, the wire
 * protocol, the request pipeline, transports (TCP/µTP/MSE), and piece selection.
 * This driver:
 *   - parses the .torrent / magnet and (for magnet) fetches the info dict,
 *   - discovers peers via HTTP/UDP trackers + DHT (src/discovery),
 *   - feeds discovered endpoints + a piece-priority vector to the engine,
 *   - drains delivered blocks, verifies+persists them through naut_download,
 *   - re-arms hash-failed pieces and reports progress for nautd / the web UI.
 *
 *   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/tracker.h"
#include "naut/log.h"
#include "naut/system.h"
#include "naut/swarm.h"

#include "engine.h"  /* torrent-peer multi-peer engine */

#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#define DEFAULT_TARGET_PEERS 80
#define MAX_TARGET_PEERS 512
#define TRACKER_DEFAULT_INTERVAL 1800.0
#define TRACKER_MIN_INTERVAL 60.0
#define TRACKER_FAILURE_RETRY_INTERVAL 300.0
#define DHT_REFRESH_INTERVAL 300.0
#define READY_BATCH 64

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

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

static uint32_t target_peer_count(void) {
    const char *env = getenv("NAUT_TARGET_PEERS");
    if (!env || !*env) return DEFAULT_TARGET_PEERS;
    char *end = NULL;
    unsigned long value = strtoul(env, &end, 10);
    if (!end || *end || value == 0) return DEFAULT_TARGET_PEERS;
    return (uint32_t)NAUT_MIN(value, MAX_TARGET_PEERS);
}

/* Below this many connected peers the swarm is "starved": re-announce as often
 * as the tracker's min interval allows instead of waiting the full interval. */
#define LOW_PEER_THRESHOLD 10

static double tracker_delay_seconds(int32_t interval) {
    if (interval <= 0) return TRACKER_DEFAULT_INTERVAL;
    if (interval < (int32_t)TRACKER_MIN_INTERVAL)
        return TRACKER_MIN_INTERVAL;
    return (double)interval;
}

/* Seconds to wait before the next tracker announce. Normally the tracker's full
 * advertised interval, but when the swarm is starved (< LOW_PEER_THRESHOLD
 * connected peers) we re-announce sooner — down to the tracker's min_interval,
 * never below our 60s floor — so a thin swarm can actually recover. */
static double next_announce_delay(int32_t interval, int32_t min_interval,
                                  uint32_t peers_connected) {
    double full = interval > 0 ? tracker_delay_seconds(interval)
                               : TRACKER_FAILURE_RETRY_INTERVAL;
    if (peers_connected >= LOW_PEER_THRESHOLD) return full;
    double floor_s = min_interval > 0 ? (double)min_interval : TRACKER_MIN_INTERVAL;
    if (floor_s < TRACKER_MIN_INTERVAL) floor_s = TRACKER_MIN_INTERVAL;
    return floor_s < full ? floor_s : full;
}

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

/* --- tracker stats ------------------------------------------------------- */

static uint32_t init_tracker_stats(char *const *trackers,
                                   const uint32_t *tracker_tiers,
                                   size_t num_trackers,
                                   naut_swarm_tracker_stats *stats,
                                   uint32_t capacity) {
    uint32_t count = 0;
    if (!stats) return 0;
    for (size_t i = 0; i < num_trackers && count < capacity; i++) {
        naut_swarm_tracker_stats *out = &stats[count++];
        memset(out, 0, sizeof(*out));
        snprintf(out->url, sizeof out->url, "%s", trackers[i]);
        out->tier = tracker_tiers ? (int32_t)tracker_tiers[i] : (int32_t)i;
        snprintf(out->status, sizeof out->status, "not contacted");
        out->seeds = -1;
        out->peers = -1;
        out->leeches = -1;
        out->downloaded = -1;
    }
    return count;
}

static naut_swarm_tracker_stats *tracker_stat_for(
    naut_swarm_tracker_stats *stats, uint32_t count, const char *url) {
    if (!stats || !url) return NULL;
    for (uint32_t i = 0; i < count; i++)
        if (strcmp(stats[i].url, url) == 0) return &stats[i];
    return NULL;
}

static void tracker_set_status(naut_swarm_tracker_stats *tracker,
                               const char *status, const char *message) {
    if (!tracker) return;
    snprintf(tracker->status, sizeof tracker->status, "%s",
             status ? status : "");
    snprintf(tracker->message, sizeof tracker->message, "%s",
             message ? message : "");
}

static void snapshot_tracker_stats(naut_swarm_stats *stats,
                                   const naut_swarm_tracker_stats *trackers,
                                   uint32_t tracker_count) {
    if (!stats || !trackers) return;
    if (tracker_count > NAUT_SWARM_MAX_TRACKER_STATS)
        tracker_count = NAUT_SWARM_MAX_TRACKER_STATS;
    stats->tracker_count = tracker_count;
    for (uint32_t i = 0; i < tracker_count; i++)
        stats->tracker_stats[i] = trackers[i];
}

static void snapshot_file_stats(naut_swarm_stats *stats,
                                const naut_download *download,
                                const naut_metainfo *metainfo) {
    if (!stats || !metainfo || !metainfo->files) return;
    uint64_t offset = 0;
    uint32_t count = 0;
    for (size_t i = 0;
         i < metainfo->num_files && count < NAUT_SWARM_MAX_FILE_STATS;
         i++) {
        const naut_file *file = &metainfo->files[i];
        uint64_t size = file->length > 0 ? (uint64_t)file->length : 0;
        uint64_t done = 0;
        if (download && size > 0) {
            uint64_t start = offset;
            uint64_t end = offset + size;
            uint32_t first = (uint32_t)(start / (uint64_t)metainfo->piece_length);
            uint32_t last = (uint32_t)((end - 1) /
                                       (uint64_t)metainfo->piece_length);
            for (uint32_t p = first; p <= last; p++) {
                if (!naut_download_have(download, p)) continue;
                uint64_t piece_start = (uint64_t)p *
                                       (uint64_t)metainfo->piece_length;
                uint64_t piece_end = piece_start +
                    (uint64_t)metainfo->piece_length;
                if (piece_end > (uint64_t)metainfo->total_length)
                    piece_end = (uint64_t)metainfo->total_length;
                uint64_t lo = piece_start > start ? piece_start : start;
                uint64_t hi = piece_end < end ? piece_end : end;
                if (hi > lo) done += hi - lo;
            }
        }
        naut_swarm_file_stats *out = &stats->file_stats[count++];
        memset(out, 0, sizeof(*out));
        snprintf(out->path, sizeof out->path, "%s",
                 file->path ? file->path : "");
        out->size = size;
        out->progress = size ? (double)done / (double)size : 1.0;
        if (out->progress > 1.0) out->progress = 1.0;
        out->priority = 1;
        out->availability = 1.0;
        offset += size;
    }
    stats->file_count = count;
}

/* Fill and deliver a progress snapshot from engine status + naut_download.
 * Per-peer detail collapses to the engine's aggregate counts for now; rich
 * per-peer rows are a later web-UI feature. */
static void report_progress(const naut_swarm_config *config,
                            engine *eng, uint32_t torrent_id,
                            const naut_download *download,
                            const naut_metainfo *metainfo,
                            const naut_swarm_tracker_stats *trackers,
                            uint32_t tracker_count,
                            double started_at) {
    if (!config->on_progress) return;
    torrent_status ts;
    memset(&ts, 0, sizeof ts);
    if (eng) engine_torrent_status(eng, torrent_id, &ts);
    uint32_t connecting = ts.peers > ts.peers_connected + ts.peers_failed
        ? ts.peers - ts.peers_connected - ts.peers_failed : 0;
    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 = ts.peers,
        .peers_connecting = connecting,
        .peers_active = ts.peers_connected,
        .peers_failed = ts.peers_failed,
        .stalled = ts.peers_connected == 0,
        .elapsed_seconds = now() - started_at,
    };
    snapshot_tracker_stats(&stats, trackers, tracker_count);
    snapshot_file_stats(&stats, download, metainfo);
    if (download)
        stats.piece_state_count = (uint32_t)naut_download_piece_states(
            download, stats.piece_states, NAUT_SWARM_MAX_PIECE_STATS);
    if (eng) {
        engine_peer_info peers[NAUT_SWARM_MAX_PEER_STATS];
        uint32_t pc = engine_peer_list(eng, torrent_id, peers,
                                       NAUT_SWARM_MAX_PEER_STATS);
        stats.peer_count = pc;
        for (uint32_t i = 0; i < pc; i++) {
            naut_swarm_peer_stats *o = &stats.peer_stats[i];
            memset(o, 0, sizeof *o);
            snprintf(o->ip, sizeof o->ip, "%.45s", peers[i].ip);
            o->port = peers[i].port;
            snprintf(o->connection, sizeof o->connection, "BT");
            snprintf(o->flags, sizeof o->flags, "%s",
                     peers[i].state == PEER_STATE_RUNNING
                         ? (peers[i].unchoked ? "D" : "d") : "K");
            o->progress = peers[i].num_pieces
                ? (double)peers[i].have_pieces / (double)peers[i].num_pieces
                : 0.0;
            o->relevance = o->progress;
            o->dlspeed = peers[i].rate_bps;
            o->downloaded = peers[i].bytes_received;
        }
    }
    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);
}

/* Render a full diagnostic snapshot (block assembly + engine piece selection)
 * when the caller requests one, and hand the text back through on_dump. Runs on
 * the owner thread, the only place engine + download state can be read safely. */
static void service_dump(const naut_swarm_config *config, engine *eng,
                         uint32_t torrent_id, const naut_download *download) {
    if (!config->should_dump || !config->on_dump) return;
    if (!config->should_dump(config->context)) return;

    char *buf = NULL;
    size_t len = 0;
    FILE *f = open_memstream(&buf, &len);
    if (!f) {
        config->on_dump(config->context, "dump: out of memory\n");
        return;
    }
    naut_download_dump(download, f);
    engine_dump_torrent(eng, torrent_id, f);
    fclose(f);
    config->on_dump(config->context, buf ? buf : "dump: render failed\n");
    free(buf);
}

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

/* --- endpoint collection ------------------------------------------------- */

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],
                              uint64_t downloaded, uint64_t left,
                              naut_tracker_event event,
                              endpoint_t **eps, size_t *neps, size_t *cap,
                              int32_t *announce_interval,
                              int32_t *announce_min_interval,
                              naut_swarm_tracker_stats *tracker_stats,
                              uint32_t tracker_count) {
    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.downloaded = downloaded;
    req.left = total_length;
    if (left <= total_length) req.left = left;
    req.event = event;
    req.numwant = 100;
    memcpy(&req.key, peerid + 8, sizeof req.key);
    if (announce_interval) *announce_interval = 0;
    if (announce_min_interval) *announce_min_interval = 0;

    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;
            memset(&response, 0, sizeof response);
            response.seeders = response.leechers = -1;
            naut_swarm_tracker_stats *tracker_stat =
                tracker_stat_for(tracker_stats, tracker_count, tracker);
            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 {
                tracker_set_status(tracker_stat, "unsupported",
                                   "unsupported tracker scheme");
                NAUT_WARN("tracker scheme unsupported: %s", tracker);
                continue;
            }
            if (e != NAUT_OK) {
                tracker_set_status(tracker_stat, "error",
                                   response.failure
                                       ? response.failure : "announce failed");
                NAUT_WARN("tracker announce failed: %s", tracker);
                naut_tracker_response_free(&response);
                continue;
            }
            tier_succeeded = true;
            if (announce_interval && response.interval > 0)
                *announce_interval = response.interval;
            if (announce_min_interval && response.min_interval > 0)
                *announce_min_interval = response.min_interval;
            if (tracker_stat) {
                tracker_set_status(tracker_stat, "working", "");
                tracker_stat->seeds = response.seeders;
                tracker_stat->leeches = response.leechers;
                tracker_stat->peers = response.num_peers > INT32_MAX
                    ? INT32_MAX : (int32_t)response.num_peers;
                tracker_stat->downloaded = -1;
            }
            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;
}

/* Hand every not-yet-fed endpoint to the engine, which owns the connection. */
static void feed_engine(engine *eng, uint32_t torrent_id,
                        const endpoint_t *eps, size_t neps, size_t *fed) {
    for (size_t i = *fed; i < neps; i++) {
        char ip[16];
        snprintf(ip, sizeof ip, "%u.%u.%u.%u",
                 eps[i].addr.ip[0], eps[i].addr.ip[1],
                 eps[i].addr.ip[2], eps[i].addr.ip[3]);
        engine_add_peer(eng, torrent_id, ip, eps[i].addr.port);
    }
    *fed = neps;
}

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;
    uint32_t target_peers = target_peer_count();
    int32_t tracker_interval = 0, tracker_min_interval = 0;
    naut_swarm_tracker_stats tracker_stats[NAUT_SWARM_MAX_TRACKER_STATS];
    uint32_t tracker_count = 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;
        tracker_count = init_tracker_stats(trackers, tracker_tiers,
                                           num_trackers, tracker_stats,
                                           NAUT_SWARM_MAX_TRACKER_STATS);
        if (from_magnet) {
            if (!discover_trackers(hash, 0, trackers, num_trackers,
                                   tracker_tiers, peerid, 0, 0,
                                   NAUT_TEV_STARTED, &endpoints, &neps, &epcap,
                                   &tracker_interval, &tracker_min_interval,
                                   tracker_stats, tracker_count) ||
                (neps < target_peers &&
                 !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;
            }
        }
    }

    /* Magnet: resolve the info dict from a peer before we can size the torrent.
     * Neither the engine nor torrent-tracker does BEP-9; use Naut's own fetch. */
    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) {
        if (from_magnet) {
            NAUT_ERROR("magnet metadata unavailable: no peers discovered");
            naut_metainfo_free(&mi);
            free(endpoints);
            return NAUT_ERR_NOTFOUND;
        }
        if (config->num_peers > 0)
            NAUT_WARN("no valid peer addresses; torrent stalled");
    }

    naut_err err;
    /* Reopen any files relocated on a prior run in place (no re-download). */
    const char **overrides = NULL;
    if (config->num_locations && mi.num_files) {
        overrides = calloc(mi.num_files, sizeof *overrides);
        if (overrides)
            for (size_t i = 0; i < config->num_locations; i++) {
                const naut_swarm_file_location *loc = &config->locations[i];
                if (loc->path && loc->file_index < mi.num_files)
                    overrides[loc->file_index] = loc->path;
            }
    }
    naut_storage_opts storage_opts = {
        .direct_io = getenv("NAUT_DIRECT_IO") != NULL,
        .preallocate = true,
        .overrides = overrides,
    };
    naut_storage *st = naut_storage_open_opts(
        mi.files, mi.num_files, config->output_dir, &storage_opts, &err);
    free(overrides);
    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);
    err = naut_download_resume(d);
    if (err != NAUT_OK) {
        NAUT_ERROR("resume scan: %s", naut_strerror(err));
        naut_download_destroy(d);
        naut_storage_close(st);
        naut_metainfo_free(&mi);
        free(endpoints);
        return err;
    }
    uint64_t resumed_bytes = naut_download_bytes_done(d);
    uint64_t resumed_left = (uint64_t)mi.total_length > resumed_bytes
        ? (uint64_t)mi.total_length - resumed_bytes : 0;
    NAUT_INFO("resume scan complete: %u/%u pieces correct, %llu bytes left",
              naut_download_pieces_done(d), mi.num_pieces,
              (unsigned long long)resumed_left);

    /* Check-only: the resume scan above already hash-verified every piece on
     * disk. Report the result and stop — no peers, no engine, no download. */
    if (config->check_only) {
        report_progress(config, NULL, 0, d, &mi, tracker_stats, tracker_count,
                        now());
        naut_download_destroy(d);
        naut_storage_close(st);
        naut_metainfo_free(&mi);
        free(endpoints);
        return NAUT_OK;
    }

    if (!from_magnet && config->num_peers == 0 && !naut_download_complete(d)) {
        if (!discover_trackers(mi.infohash_v1, (uint64_t)mi.total_length,
                               mi.trackers, mi.num_trackers,
                               mi.tracker_tiers, peerid,
                               resumed_bytes, resumed_left,
                               NAUT_TEV_STARTED, &endpoints, &neps, &epcap,
                               &tracker_interval, &tracker_min_interval,
                               tracker_stats, tracker_count) ||
            (neps < target_peers &&
             !discover_dht(mi.infohash_v1, &endpoints, &neps, &epcap))) {
            NAUT_ERROR("out of memory collecting discovered peers");
            naut_download_destroy(d);
            naut_storage_close(st);
            naut_metainfo_free(&mi);
            free(endpoints);
            return NAUT_ERR_NOMEM;
        }
        if (neps == 0)
            NAUT_WARN("tracker and DHT discovery returned no peers; torrent stalled");
    }

    /* Spin up the engine and register the torrent. The engine owns sockets,
     * the wire protocol, the pipeline, transports, and piece selection. */
    engine_config ecfg;
    memset(&ecfg, 0, sizeof ecfg);
    ecfg.encryption = 1;   /* offer MSE (RC4) + plaintext: most compatible */
    ecfg.fallback = 1;     /* retry transport/encryption combos per endpoint */
    engine *eng = engine_create(&ecfg);
    int32_t tid = eng ? engine_add_torrent(eng, mi.infohash_v1, peerid,
                                           (uint64_t)mi.piece_length,
                                           (uint64_t)mi.total_length,
                                           mi.num_pieces)
                      : -1;
    if (!eng || tid < 0) {
        NAUT_ERROR("unable to create download engine");
        if (eng) engine_destroy(eng);
        naut_download_destroy(d);
        naut_storage_close(st);
        naut_metainfo_free(&mi);
        free(endpoints);
        return NAUT_ERR_NOMEM;
    }
    uint32_t torrent_id = (uint32_t)tid;

    /* Priority vector: skip what resume already verified, request the rest. */
    uint8_t *prio = malloc(mi.num_pieces ? mi.num_pieces : 1);
    if (!prio) {
        engine_destroy(eng);
        naut_download_destroy(d);
        naut_storage_close(st);
        naut_metainfo_free(&mi);
        free(endpoints);
        return NAUT_ERR_NOMEM;
    }
    for (uint32_t p = 0; p < mi.num_pieces; p++)
        prio[p] = naut_download_have(d, p) ? 0 : 1;
    engine_set_priorities(eng, torrent_id, prio, mi.num_pieces);

    size_t fed = 0;
    feed_engine(eng, torrent_id, endpoints, neps, &fed);

    double t0 = now();
    /* Use the discovered endpoint count as the initial peer proxy: if we start
     * thin, schedule a quick re-announce instead of waiting the full interval. */
    double next_tracker_announce =
        t0 + next_announce_delay(tracker_interval, tracker_min_interval,
                                 (uint32_t)neps);
    double next_dht_lookup = t0 + DHT_REFRESH_INTERVAL;
    naut_err run_error = NAUT_OK;
    bool cancelled = false;

    NAUT_INFO("swarm: engine started, %zu peers queued, %u pieces, %lld bytes",
              neps, mi.num_pieces, (long long)mi.total_length);
    emit_event(config, NAUT_EVENT_TORRENT_ADDED, 0, NULL, NULL);
    report_progress(config, eng, torrent_id, d, &mi,
                    tracker_stats, tracker_count, t0);

    engine_block blocks[READY_BATCH];
    uint64_t applied_rate = UINT64_MAX;   /* force first apply */
    while (!naut_download_complete(d) && run_error == NAUT_OK) {
        service_control(config, st);
        service_dump(config, eng, torrent_id, d);
        if (stop_requested(config)) { cancelled = true; break; }

        /* Apply the live download throttle when it changes. */
        if (config->download_rate) {
            uint64_t rate = config->download_rate(config->context);
            if (rate != applied_rate) {
                engine_set_download_rate(eng, rate);
                applied_rate = rate;
            }
        }

        /* Top up the swarm from trackers / DHT when it runs thin. */
        if (config->num_peers == 0) {
            torrent_status ts;
            engine_torrent_status(eng, torrent_id, &ts);
            if (ts.peers_connected < target_peers) {
                double t = now();
                if (mi.num_trackers > 0 && t >= next_tracker_announce) {
                    uint64_t downloaded = naut_download_bytes_done(d);
                    uint64_t left = (uint64_t)mi.total_length > downloaded
                        ? (uint64_t)mi.total_length - downloaded : 0;
                    int32_t interval = 0, min_interval = 0;
                    if (!discover_trackers(mi.infohash_v1,
                                           (uint64_t)mi.total_length,
                                           mi.trackers, mi.num_trackers,
                                           mi.tracker_tiers, peerid,
                                           downloaded, left, NAUT_TEV_NONE,
                                           &endpoints, &neps, &epcap,
                                           &interval, &min_interval,
                                           tracker_stats, tracker_count)) {
                        run_error = NAUT_ERR_NOMEM;
                        break;
                    }
                    next_tracker_announce = t + next_announce_delay(
                        interval, min_interval, ts.peers_connected);
                    feed_engine(eng, torrent_id, endpoints, neps, &fed);
                }
                if (t >= next_dht_lookup) {
                    if (!discover_dht(mi.infohash_v1, &endpoints, &neps,
                                      &epcap)) {
                        run_error = NAUT_ERR_NOMEM;
                        break;
                    }
                    next_dht_lookup = t + DHT_REFRESH_INTERVAL;
                    feed_engine(eng, torrent_id, endpoints, neps, &fed);
                }
            }
        }

        engine_wait(eng, 200);
        uint32_t n;
        while ((n = engine_poll_ready(eng, blocks, READY_BATCH)) > 0) {
            for (uint32_t i = 0; i < n; i++) {
                engine_block *b = &blocks[i];
                uint8_t *data = (uint8_t *)engine_arena_base(eng, b->loop) +
                                (uint64_t)b->slot * PEER_BLOCK_SIZE;
                bool done = false;
                naut_err be = naut_download_on_block(d, b->piece, b->begin,
                                                     data, b->len, &done);
                engine_release_slot(eng, b->loop, b->slot);
                if (be == NAUT_ERR_PROTO) {
                    /* Bad/failed piece: re-arm it for another fetch. */
                    engine_request_piece(eng, torrent_id, b->piece);
                } else if (be != NAUT_OK && be != NAUT_ERR_RANGE) {
                    run_error = be;
                    break;
                } else if (done) {
                    engine_set_priority(eng, torrent_id, b->piece, 0);
                }
            }
            if (run_error != NAUT_OK) break;
        }
        report_progress(config, eng, torrent_id, d, &mi,
                        tracker_stats, tracker_count, t0);
    }

    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 in %.2fs (%.1f MB/s), all SHA-1 verified",
                  naut_download_pieces_done(d), mi.num_pieces, dt,
                  dt > 0 ? mb / dt : 0.0);
        report_progress(config, eng, torrent_id, d, &mi,
                        tracker_stats, tracker_count, t0);
        emit_event(config, NAUT_EVENT_TORRENT_FINISHED, 0, NULL, NULL);
        while (config->keep_alive && !stop_requested(config)) {
            service_control(config, st);
            service_dump(config, eng, torrent_id, d);
            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);
    }

    service_control(config, st);
    naut_storage_sync(st);
    engine_destroy(eng);
    free(prio);
    free(endpoints);
    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;
}