#include "naut/event.h" #include "naut/log.h" #include "naut/metainfo.h" #include "naut/plugin.h" #include "naut/rpc.h" #include "naut/script.h" #include "naut/storage.h" #include "naut/swarm.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #define DEFAULT_SOCKET "/tmp/nautd.sock" #define MOVE_QUEUE_CAPACITY 64 #define MAX_SUBSCRIBERS 64 #define MAX_TORRENTS 128 typedef struct { uint32_t file_index; char destination[PATH_MAX]; } move_command; /* Last known on-disk location of a file after a relocate, persisted so a moved * file is reopened in place across restarts instead of re-downloaded. */ typedef struct { uint32_t file_index; char *path; } file_location; typedef enum { TORRENT_QUEUED, TORRENT_RUNNING, TORRENT_STALLED, TORRENT_COMPLETE, TORRENT_STOPPING, TORRENT_STOPPED, TORRENT_ERROR, TORRENT_PAUSED, TORRENT_CHECKING, } torrent_state; typedef struct daemon_state daemon_state; typedef struct { daemon_state *daemon; uint64_t id; char *source; bool source_is_temp; /* ephemeral /tmp upload; unlink on any destroy */ bool source_managed; /* durable upload under state_dir/uploads; unlink only * when the torrent is removed (not on shutdown) */ char *name; /* optional display name (persisted for the UI) */ char *output_dir; char **peers; size_t num_peers; pthread_t thread; bool thread_started; bool thread_done; pthread_mutex_t lock; torrent_state state; naut_err result; bool stop_requested; bool remove_requested; bool paused; /* user-paused: never auto-activated (persisted) */ bool force_start; /* bypass the queue cap (persisted) */ bool restart_requested; /* one-shot stop->start (recheck) */ bool needs_check; /* run a one-shot hash check (paused add/recheck) */ bool checking; /* a check-only worker is currently running */ int queue_pos; /* ordering within the download queue (persisted) */ uint64_t rate_share; /* engine download cap for this torrent, bytes/sec */ naut_swarm_stats stats; move_command moves[MOVE_QUEUE_CAPACITY]; size_t move_head; size_t move_count; uint64_t moves_processed; uint64_t moves_failed; file_location *locations; /* last known location of each relocated file */ size_t num_locations; char *pending_save_path; /* "Set location" target; the worker moves the * torrent's files there on its next control pass */ bool pending_save_path_reset; /* true: reset every file to its original * download relpath; false: keep per-file moves */ char *category; /* single category (qBittorrent-style), may be ""*/ char **tags; /* user tags (multiple) */ size_t num_tags; /* category + tags are the flat labels Lua sees */ char **files; /* torrent's file relpaths (for overlap checks) */ size_t num_files; /* parsed at add; empty for magnets pre-metadata */ uint64_t dump_seq; /* bumped by a dump RPC; > dump_done_seq => pending */ uint64_t dump_done_seq; /* highest dump_seq the worker has rendered */ char *dump_text; /* latest rendered dump (owner: task) */ } torrent_task; struct daemon_state { naut_event_bus *events; naut_rpc_registry *rpc; naut_plugin_manager *plugins; naut_script *script; char *script_path; pthread_mutex_t script_lock; /* Script settings: the loaded script declares a schema via * naut.define_settings; the user edits values in the web UI. The schema is * rebuilt on each (re)load; values persist independently. */ pthread_mutex_t settings_lock; json_t *script_settings_schema; /* array of {key,label,type,default} */ json_t *script_settings; /* object: key -> value string (user-set)*/ char settings_file[PATH_MAX]; /* /script_settings.json */ /* Label taxonomy: the web UI's full category + tag lists (including ones * created but not yet assigned). Web-layer schema; the daemon just persists * it so they survive restarts. */ pthread_mutex_t taxonomy_lock; json_t *label_categories; /* array of {name, savePath} */ json_t *label_tags; /* array of tag name strings */ char taxonomy_file[PATH_MAX]; /* /labels.json */ char data_dir[PATH_MAX]; /* the resolved state dir (for blobs) */ pthread_mutex_t blob_lock; /* guards webui_.json blob files */ pthread_mutex_t torrent_lock; torrent_task *torrents[MAX_TORRENTS]; size_t torrent_count; uint64_t next_torrent_id; pthread_mutex_t subscriber_lock; int subscribers[MAX_SUBSCRIBERS]; size_t subscriber_count; bool stopping; bool persist_enabled; char state_file[PATH_MAX]; /* /torrents.json */ char uploads_dir[PATH_MAX]; /* /uploads */ char prefs_file[PATH_MAX]; /* /prefs.json */ /* Daemon preferences (queue + throttle). Upload limits are stored but inert: * the engine is leech-only (no seeding) so only download limits take effect. */ uint32_t max_active; /* max concurrent downloading torrents */ uint64_t dl_limit; /* global download cap, bytes/sec (0=off)*/ uint64_t alt_dl_limit; /* alt download cap, bytes/sec */ uint64_t up_limit; /* stored, inert */ uint64_t alt_up_limit; /* stored, inert */ bool alt_speed_enabled; /* use alt_* limits when true */ }; #define DEFAULT_MAX_ACTIVE 5 static volatile sig_atomic_t interrupted; static naut_err queue_move(void *opaque, uint64_t torrent_id, uint32_t file_index, const char *destination); static void service_lifecycle(daemon_state *state); static void persist_torrents(daemon_state *state); static json_t *script_settings_json(daemon_state *state); static void on_signal(int signal_number) { (void)signal_number; interrupted = 1; } static const char *torrent_state_name(torrent_state state) { static const char *names[] = { [TORRENT_QUEUED] = "queued", [TORRENT_RUNNING] = "downloading", [TORRENT_STALLED] = "stalled", [TORRENT_COMPLETE] = "complete", [TORRENT_STOPPING] = "stopping", [TORRENT_STOPPED] = "stopped", [TORRENT_ERROR] = "error", [TORRENT_PAUSED] = "paused", [TORRENT_CHECKING] = "checking", }; return (size_t)state < NAUT_ARRAY_LEN(names) ? names[state] : "unknown"; } static torrent_task *find_torrent_locked(daemon_state *state, uint64_t id) { for (size_t i = 0; i < state->torrent_count; i++) if (state->torrents[i]->id == id) return state->torrents[i]; return NULL; } static void torrent_progress(void *opaque, const naut_swarm_stats *stats) { torrent_task *task = opaque; pthread_mutex_lock(&task->lock); task->stats = *stats; /* A check-only run just reports verified progress; the worker decides the * final state (it stays paused), so don't flip it to complete/running here. */ if (!task->checking) { if (stats->total_pieces > 0 && stats->pieces_done == stats->total_pieces) task->state = TORRENT_COMPLETE; else if (stats->stalled) task->state = TORRENT_STALLED; else if (task->state == TORRENT_QUEUED || task->state == TORRENT_STALLED) task->state = TORRENT_RUNNING; } pthread_mutex_unlock(&task->lock); } static bool torrent_should_stop(void *opaque) { torrent_task *task = opaque; pthread_mutex_lock(&task->lock); bool stop = task->stop_requested; pthread_mutex_unlock(&task->lock); return stop; } static bool torrent_should_dump(void *opaque) { torrent_task *task = opaque; pthread_mutex_lock(&task->lock); bool pending = task->dump_seq != task->dump_done_seq; pthread_mutex_unlock(&task->lock); return pending; } static void torrent_on_dump(void *opaque, const char *text) { torrent_task *task = opaque; char *copy = text ? strdup(text) : NULL; pthread_mutex_lock(&task->lock); free(task->dump_text); task->dump_text = copy; task->dump_done_seq = task->dump_seq; pthread_mutex_unlock(&task->lock); } /* The download throttle the reconciler computed for this torrent (its share of * the global limit). 0 = unlimited. */ static uint64_t torrent_download_rate(void *opaque) { torrent_task *task = opaque; pthread_mutex_lock(&task->lock); uint64_t rate = task->rate_share; pthread_mutex_unlock(&task->lock); return rate; } /* Record (or update) the last known location of a relocated file. Caller holds * task->lock. */ static void task_set_location(torrent_task *task, uint32_t file_index, const char *path) { char *copy = strdup(path); if (!copy) return; for (size_t i = 0; i < task->num_locations; i++) { if (task->locations[i].file_index == file_index) { free(task->locations[i].path); task->locations[i].path = copy; return; } } file_location *grown = realloc(task->locations, (task->num_locations + 1) * sizeof *grown); if (!grown) { free(copy); return; } task->locations = grown; task->locations[task->num_locations].file_index = file_index; task->locations[task->num_locations].path = copy; task->num_locations++; } /* Replace the task's tag set from a JSON array of strings (empty/duplicate * entries dropped). NULL leaves the tags unchanged. Caller holds task->lock. */ static void task_set_tags(torrent_task *task, const json_t *tags_json) { if (!json_is_array(tags_json)) return; size_t n = json_array_size(tags_json); char **next = n ? calloc(n, sizeof *next) : NULL; size_t count = 0; if (next) { for (size_t i = 0; i < n; i++) { const char *s = json_string_value(json_array_get(tags_json, i)); if (!s || !*s) continue; bool dup = false; for (size_t j = 0; j < count; j++) if (strcmp(next[j], s) == 0) { dup = true; break; } if (dup) continue; char *copy = strdup(s); if (copy) next[count++] = copy; } } for (size_t i = 0; i < task->num_tags; i++) free(task->tags[i]); free(task->tags); task->tags = next; task->num_tags = count; } /* Set the task's category. NULL leaves it unchanged. Caller holds task->lock. */ static void task_set_category(torrent_task *task, const char *category) { if (!category) return; char *copy = strdup(category); if (!copy) return; free(task->category); task->category = copy; } /* JSON array of the task's tags. Caller holds task->lock. */ static json_t *task_tags_json(const torrent_task *task) { json_t *tags = json_array(); if (tags) for (size_t i = 0; i < task->num_tags; i++) json_array_append_new(tags, json_string(task->tags[i])); return tags; } /* Copy a torrent's labels (category + tags, flattened) out for the Lua * `naut.get_labels` accessor. Runs on the script worker thread. */ static char **script_labels(void *opaque, uint64_t torrent_id, size_t *count) { daemon_state *state = opaque; *count = 0; char **out = NULL; pthread_mutex_lock(&state->torrent_lock); torrent_task *task = find_torrent_locked(state, torrent_id); if (task) { pthread_mutex_lock(&task->lock); bool has_cat = task->category && *task->category; size_t cap = task->num_tags + (has_cat ? 1 : 0); if (cap && (out = calloc(cap, sizeof *out))) { size_t c = 0; if (has_cat) { char *copy = strdup(task->category); if (copy) out[c++] = copy; } for (size_t i = 0; i < task->num_tags; i++) { char *copy = strdup(task->tags[i]); if (copy) out[c++] = copy; } *count = c; if (c == 0) { free(out); out = NULL; } } pthread_mutex_unlock(&task->lock); } pthread_mutex_unlock(&state->torrent_lock); return out; } /* Remove directories left empty after moving a file out, walking up from the * file's old parent but never reaching or passing `base` (the download root, * which may be shared). rmdir only deletes empty dirs, so this is safe. */ static void prune_empty_dirs(const char *old_file_path, const char *base, size_t base_len) { char dir[PATH_MAX]; if ((size_t)snprintf(dir, sizeof dir, "%s", old_file_path) >= sizeof dir) return; char *slash = strrchr(dir, '/'); if (!slash) return; *slash = '\0'; /* dir = the file's parent directory */ while (strlen(dir) > base_len && strncmp(dir, base, base_len) == 0 && dir[base_len] == '/') { if (rmdir(dir) != 0) break; /* non-empty / busy: stop pruning */ slash = strrchr(dir, '/'); if (!slash) break; *slash = '\0'; } } /* Apply a pending "Set location": move the torrent's files under the new base * directory, then adopt it as the output dir. Runs on the worker thread (the * sole owner of `storage`), so it never races engine writes. A one-time move, * with no rule that re-locates later. * * Per file (relative to the old base): * - reset: -> new_base/ * - kept, under old base: -> new_base/ (preserve moves) * - kept, separate dir: left exactly where it is. * Empty residual folders under the old base are pruned. */ static void apply_pending_save_path(torrent_task *task, naut_storage *storage) { pthread_mutex_lock(&task->lock); if (!task->pending_save_path || task->stats.file_count == 0) { pthread_mutex_unlock(&task->lock); return; /* nothing to do, or file list not known yet — retry next pass */ } char *target = task->pending_save_path; /* take ownership */ task->pending_save_path = NULL; bool reset = task->pending_save_path_reset; char *old_base = strdup(task->output_dir ? task->output_dir : ""); size_t nfiles = task->stats.file_count; if (nfiles > NAUT_SWARM_MAX_FILE_STATS) nfiles = NAUT_SWARM_MAX_FILE_STATS; char **orig_rel = calloc(nfiles, sizeof *orig_rel); char **cur = calloc(nfiles, sizeof *cur); /* each file's current abs path */ bool ok = old_base && orig_rel && cur; for (size_t i = 0; ok && i < nfiles; i++) { orig_rel[i] = strdup(task->stats.file_stats[i].path); const char *ov = NULL; for (size_t j = 0; j < task->num_locations; j++) if (task->locations[j].file_index == i) { ov = task->locations[j].path; break; } char tmp[PATH_MAX]; if (ov) cur[i] = strdup(ov); else if (orig_rel[i] && (size_t)snprintf(tmp, sizeof tmp, "%s/%s", old_base, orig_rel[i]) < sizeof tmp) cur[i] = strdup(tmp); if (!orig_rel[i] || !cur[i]) ok = false; } pthread_mutex_unlock(&task->lock); if (!ok) { for (size_t i = 0; i < nfiles; i++) { free(orig_rel[i]); free(cur[i]); } free(orig_rel); free(cur); free(old_base); free(target); return; } /* Normalize trailing slashes for clean prefix comparisons. */ size_t blen = strlen(old_base); while (blen > 1 && old_base[blen - 1] == '/') old_base[--blen] = '\0'; size_t tlen = strlen(target); while (tlen > 1 && target[tlen - 1] == '/') target[--tlen] = '\0'; file_location *newloc = NULL; size_t nnew = 0, moved = 0; for (size_t i = 0; i < nfiles; i++) { char def[PATH_MAX], final[PATH_MAX]; snprintf(def, sizeof def, "%s/%s", target, orig_rel[i]); bool under = strlen(cur[i]) > blen && strncmp(cur[i], old_base, blen) == 0 && cur[i][blen] == '/'; if (reset) snprintf(final, sizeof final, "%s", def); else if (under) snprintf(final, sizeof final, "%s/%s", target, cur[i] + blen + 1); else snprintf(final, sizeof final, "%s", cur[i]); /* separate dir: leave */ bool did_move = false; if (strcmp(final, cur[i]) != 0) { naut_err e = naut_storage_relocate(storage, (size_t)i, final); if (e == NAUT_OK) { did_move = true; moved++; } else { NAUT_WARN("set-location torrent=%llu file=%zu -> %s: %s", (unsigned long long)task->id, i, final, naut_strerror(e)); snprintf(final, sizeof final, "%s", cur[i]); /* stayed put */ } } if (did_move && under) prune_empty_dirs(cur[i], old_base, blen); if (strcmp(final, def) != 0) { /* not at the default path -> track it */ file_location *grown = realloc(newloc, (nnew + 1) * sizeof *grown); char *p = strdup(final); if (grown && p) { newloc = grown; newloc[nnew].file_index = (uint32_t)i; newloc[nnew].path = p; nnew++; } else { free(p); if (grown) newloc = grown; } } free(orig_rel[i]); free(cur[i]); } free(orig_rel); free(cur); pthread_mutex_lock(&task->lock); free(task->output_dir); task->output_dir = target; /* take ownership */ for (size_t i = 0; i < task->num_locations; i++) free(task->locations[i].path); free(task->locations); task->locations = newloc; task->num_locations = nnew; pthread_mutex_unlock(&task->lock); NAUT_INFO("set-location torrent=%llu -> %s (%zu/%zu files moved%s)", (unsigned long long)task->id, target, moved, nfiles, reset ? ", reset to original paths" : ""); free(old_base); persist_torrents(task->daemon); } static void torrent_control(void *opaque, naut_storage *storage) { torrent_task *task = opaque; bool moved = false; for (;;) { move_command command; pthread_mutex_lock(&task->lock); if (task->move_count == 0) { pthread_mutex_unlock(&task->lock); break; } command = task->moves[task->move_head]; task->move_head = (task->move_head + 1) % MOVE_QUEUE_CAPACITY; task->move_count--; pthread_mutex_unlock(&task->lock); naut_err error = naut_storage_relocate( storage, command.file_index, command.destination); pthread_mutex_lock(&task->lock); if (error == NAUT_OK) { task->moves_processed++; task_set_location(task, command.file_index, command.destination); moved = true; } else { task->moves_failed++; } pthread_mutex_unlock(&task->lock); if (error == NAUT_OK) NAUT_INFO("moved torrent=%llu file=%u -> %s", (unsigned long long)task->id, command.file_index, command.destination); else NAUT_WARN("move torrent=%llu file=%u -> %s failed: %s", (unsigned long long)task->id, command.file_index, command.destination, naut_strerror(error)); } /* Persist the new locations so a restart reopens the files in place. */ if (moved) persist_torrents(task->daemon); apply_pending_save_path(task, storage); } static void *torrent_worker(void *opaque) { torrent_task *task = opaque; /* Snapshot the saved moved-file locations so the swarm reopens them in * place instead of re-downloading. Copied so a later move (processed on this * same thread) reallocating task->locations can't invalidate them. */ naut_swarm_file_location *locations = NULL; size_t num_locations = 0; pthread_mutex_lock(&task->lock); bool check_only = task->checking; task->state = check_only ? TORRENT_CHECKING : TORRENT_RUNNING; if (task->num_locations && (locations = calloc(task->num_locations, sizeof *locations))) { for (size_t i = 0; i < task->num_locations; i++) { char *path = strdup(task->locations[i].path); if (!path) continue; locations[num_locations].file_index = task->locations[i].file_index; locations[num_locations].path = path; num_locations++; } } pthread_mutex_unlock(&task->lock); naut_swarm_config config = { .source = task->source, .output_dir = task->output_dir, .peers = (const char *const *)task->peers, .num_peers = task->num_peers, .locations = locations, .num_locations = num_locations, .torrent_id = task->id, .events = task->daemon->events, .keep_alive = true, .check_only = check_only, .on_progress = torrent_progress, .on_control = torrent_control, .should_stop = torrent_should_stop, .download_rate = torrent_download_rate, .should_dump = torrent_should_dump, .on_dump = torrent_on_dump, .context = task, }; naut_err result = naut_swarm_run(&config); for (size_t i = 0; i < num_locations; i++) free((char *)locations[i].path); free(locations); pthread_mutex_lock(&task->lock); task->result = result; if (check_only) { /* One-shot hash check finished: progress is recorded; return to the * paused state (or queued if the user resumed mid-check). */ task->checking = false; task->needs_check = false; task->state = task->paused ? TORRENT_PAUSED : TORRENT_QUEUED; } else if (task->stop_requested) { /* A requested stop wins over the run result: a completed torrent returns * NAUT_OK even when paused/stopped, and marking it COMPLETE would make * the lifecycle reconciler immediately relaunch its keep-alive worker * (clearing `paused`) — i.e. pause wouldn't stick for seeding torrents. */ task->state = task->paused ? TORRENT_PAUSED : TORRENT_STOPPED; } else if (result == NAUT_OK) { task->state = TORRENT_COMPLETE; } else { task->state = TORRENT_ERROR; } task->thread_done = true; pthread_mutex_unlock(&task->lock); return NULL; } static json_t *torrent_json(torrent_task *task) { pthread_mutex_lock(&task->lock); json_t *result = json_object(); if (result) { json_object_set_new(result, "torrent_id", json_integer((json_int_t)task->id)); json_object_set_new(result, "source", json_string(task->source)); json_object_set_new(result, "output", json_string(task->output_dir)); if (task->name) json_object_set_new(result, "name", json_string(task->name)); json_object_set_new(result, "state", json_string(torrent_state_name(task->state))); json_object_set_new(result, "paused", json_boolean(task->paused)); json_object_set_new(result, "force_start", json_boolean(task->force_start)); json_object_set_new(result, "queue_pos", json_integer(task->queue_pos)); json_object_set_new(result, "bytes_done", json_integer((json_int_t)task->stats.bytes_done)); json_object_set_new(result, "total_bytes", json_integer((json_int_t)task->stats.total_bytes)); json_object_set_new(result, "pieces_done", json_integer(task->stats.pieces_done)); json_object_set_new(result, "total_pieces", json_integer(task->stats.total_pieces)); json_t *piece_states = json_array(); if (piece_states) { uint32_t state_count = task->stats.piece_state_count; if (state_count > NAUT_SWARM_MAX_PIECE_STATS) state_count = NAUT_SWARM_MAX_PIECE_STATS; for (uint32_t i = 0; i < state_count; i++) json_array_append_new(piece_states, json_integer(task->stats.piece_states[i])); json_object_set_new(result, "piece_states", piece_states); } json_object_set_new(result, "peers", json_integer(task->stats.peers_active)); json_object_set_new(result, "peers_discovered", json_integer(task->stats.peers_total)); json_object_set_new(result, "peers_connecting", json_integer(task->stats.peers_connecting)); json_object_set_new(result, "peers_failed", json_integer(task->stats.peers_failed)); json_t *peer_list = json_array(); if (peer_list) { uint32_t peer_count = task->stats.peer_count; if (peer_count > NAUT_SWARM_MAX_PEER_STATS) peer_count = NAUT_SWARM_MAX_PEER_STATS; for (uint32_t i = 0; i < peer_count; i++) { const naut_swarm_peer_stats *peer = &task->stats.peer_stats[i]; json_t *item = json_pack( "{s:s,s:i,s:s,s:s,s:s,s:f,s:f,s:I,s:I,s:I,s:I}", "ip", peer->ip, "port", (int)peer->port, "client", peer->client, "connection", peer->connection, "flags", peer->flags, "progress", peer->progress, "relevance", peer->relevance, "downloaded", (json_int_t)peer->downloaded, "uploaded", (json_int_t)peer->uploaded, "dlspeed", (json_int_t)peer->dlspeed, "upspeed", (json_int_t)peer->upspeed); if (item) json_array_append_new(peer_list, item); } json_object_set_new(result, "peer_list", peer_list); } json_t *trackers = json_array(); if (trackers) { uint32_t tracker_count = task->stats.tracker_count; if (tracker_count > NAUT_SWARM_MAX_TRACKER_STATS) tracker_count = NAUT_SWARM_MAX_TRACKER_STATS; for (uint32_t i = 0; i < tracker_count; i++) { const naut_swarm_tracker_stats *tracker = &task->stats.tracker_stats[i]; json_t *item = json_pack( "{s:s,s:i,s:s,s:i,s:i,s:i,s:i,s:s}", "url", tracker->url, "tier", tracker->tier, "status", tracker->status, "seeds", tracker->seeds, "peers", tracker->peers, "leeches", tracker->leeches, "downloaded", tracker->downloaded, "message", tracker->message); if (item) json_array_append_new(trackers, item); } json_object_set_new(result, "trackers", trackers); } json_t *files = json_array(); if (files) { uint32_t file_count = task->stats.file_count; if (file_count > NAUT_SWARM_MAX_FILE_STATS) file_count = NAUT_SWARM_MAX_FILE_STATS; for (uint32_t i = 0; i < file_count; i++) { const naut_swarm_file_stats *file = &task->stats.file_stats[i]; json_t *item = json_pack( "{s:s,s:I,s:f,s:i,s:f}", "name", file->path, "size", (json_int_t)file->size, "progress", file->progress, "priority", file->priority, "availability", file->availability); if (item) json_array_append_new(files, item); } json_object_set_new(result, "files", files); } json_object_set_new(result, "elapsed_seconds", json_real(task->stats.elapsed_seconds)); json_object_set_new(result, "pending_moves", json_integer((json_int_t)task->move_count)); json_object_set_new(result, "moves_processed", json_integer((json_int_t)task->moves_processed)); json_object_set_new(result, "moves_failed", json_integer((json_int_t)task->moves_failed)); if (task->num_locations) { json_t *locations = json_array(); if (locations) { for (size_t i = 0; i < task->num_locations; i++) json_array_append_new(locations, json_pack( "{s:i,s:s}", "file", (int)task->locations[i].file_index, "path", task->locations[i].path)); json_object_set_new(result, "locations", locations); } } json_object_set_new(result, "category", json_string(task->category ? task->category : "")); json_object_set_new(result, "tags", task_tags_json(task)); if (task->pending_save_path) json_object_set_new(result, "pending_save_path", json_string(task->pending_save_path)); if (task->state == TORRENT_ERROR) json_object_set_new(result, "error", json_string(naut_strerror(task->result))); } pthread_mutex_unlock(&task->lock); return result; } static json_t *rpc_ping(void *opaque, const json_t *params, naut_err *error) { (void)opaque; (void)params; json_t *result = json_object(); if (!result) { *error = NAUT_ERR_NOMEM; return NULL; } json_object_set_new(result, "protocol", json_integer(NAUT_RPC_VERSION)); json_object_set_new(result, "service", json_string("nautd")); *error = NAUT_OK; return result; } static char *read_text_file_limited(const char *path, size_t max_bytes) { FILE *file = fopen(path, "rb"); if (!file) return NULL; char *buf = malloc(max_bytes + 1); if (!buf) { fclose(file); return NULL; } size_t n = fread(buf, 1, max_bytes, file); bool too_large = !feof(file); bool error = ferror(file); fclose(file); if (error) { free(buf); return NULL; } buf[n] = 0; if (too_large) { const char suffix[] = "\n-- truncated --\n"; size_t suffix_len = sizeof suffix - 1; if (max_bytes >= suffix_len) { memcpy(buf + max_bytes - suffix_len, suffix, suffix_len + 1); } } return buf; } static json_t *script_status_json(daemon_state *state) { naut_script_stats stats = {0}; char last_error[256] = {0}; char *path = NULL; bool loaded = false; pthread_mutex_lock(&state->script_lock); loaded = state->script != NULL; if (state->script) { naut_script_get_stats(state->script, &stats); snprintf(last_error, sizeof last_error, "%s", naut_script_last_error(state->script)); } if (state->script_path) path = strdup(state->script_path); pthread_mutex_unlock(&state->script_lock); char *source = path ? read_text_file_limited(path, 256 * 1024) : NULL; json_t *script = json_pack( "{s:b,s:s,s:s,s:I,s:I,s:I,s:I,s:I,s:s}", "loaded", loaded, "path", path ? path : "", "source", source ? source : "", "queued", (json_int_t)stats.queued, "handled", (json_int_t)stats.handled, "dropped", (json_int_t)stats.dropped, "errors", (json_int_t)stats.errors, "move_requests", (json_int_t)stats.move_requests, "last_error", last_error); if (script) { json_t *settings = script_settings_json(state); json_object_set_new(script, "settings", settings ? settings : json_array()); } free(source); free(path); return script; } static json_t *rpc_status(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; size_t torrent_count; size_t active = 0; uint64_t moves = 0; size_t pending = 0; pthread_mutex_lock(&state->torrent_lock); torrent_count = state->torrent_count; for (size_t i = 0; i < torrent_count; i++) { torrent_task *task = state->torrents[i]; pthread_mutex_lock(&task->lock); if (task->state == TORRENT_RUNNING || task->state == TORRENT_STALLED || task->state == TORRENT_STOPPING) active++; moves += task->moves_processed; pending += task->move_count; pthread_mutex_unlock(&task->lock); } pthread_mutex_unlock(&state->torrent_lock); json_t *result = json_object(); json_t *script = script_status_json(state); if (!result || !script) { json_decref(result); json_decref(script); *error = NAUT_ERR_NOMEM; return NULL; } json_object_set_new(result, "protocol", json_integer(NAUT_RPC_VERSION)); json_object_set_new(result, "plugins", json_integer((json_int_t)naut_plugin_count( state->plugins))); json_object_set_new(result, "storage_backends", json_integer((json_int_t)naut_plugin_storage_count( state->plugins))); json_object_set_new(result, "torrents", json_integer((json_int_t)torrent_count)); json_object_set_new(result, "active_torrents", json_integer((json_int_t)active)); json_object_set_new(result, "script_loaded", json_boolean(json_boolean_value( json_object_get(script, "loaded")))); json_object_set_new(result, "script", script); json_object_set_new(result, "move_commands", json_integer(moves)); json_object_set_new(result, "pending_move_commands", json_integer((json_int_t)pending)); *error = NAUT_OK; return result; } static json_t *rpc_script_status(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; json_t *script = script_status_json(state); *error = script ? NAUT_OK : NAUT_ERR_NOMEM; return script; } static json_t *rpc_plugins(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; json_t *plugins = json_array(); json_t *storage = json_array(); if (!plugins || !storage) { json_decref(plugins); json_decref(storage); *error = NAUT_ERR_NOMEM; return NULL; } for (size_t i = 0; i < naut_plugin_count(state->plugins); i++) json_array_append_new(plugins, json_string(naut_plugin_name(state->plugins, i))); for (size_t i = 0; i < naut_plugin_storage_count(state->plugins); i++) json_array_append_new(storage, json_string(naut_plugin_storage_name(state->plugins, i))); json_t *result = json_object(); if (!result) { json_decref(plugins); json_decref(storage); *error = NAUT_ERR_NOMEM; return NULL; } json_object_set_new(result, "plugins", plugins); json_object_set_new(result, "storage_backends", storage); *error = NAUT_OK; return result; } static json_t *rpc_emit(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; if (!json_is_object(params)) { *error = NAUT_ERR_INVAL; return NULL; } const char *type_name = json_string_value(json_object_get(params, "type")); naut_event_type type; if (!type_name || !naut_event_type_parse(type_name, &type)) { *error = NAUT_ERR_INVAL; return NULL; } json_int_t torrent_id = json_integer_value(json_object_get(params, "torrent_id")); json_int_t index = json_integer_value(json_object_get(params, "index")); if (torrent_id < 0 || index < 0 || (uint64_t)index > UINT32_MAX) { *error = NAUT_ERR_RANGE; return NULL; } naut_event event = { .type = type, .torrent_id = (uint64_t)torrent_id, .index = (uint32_t)index, .message = json_string_value(json_object_get(params, "message")), .path = json_string_value(json_object_get(params, "path")), }; naut_event_emit(state->events, &event); *error = NAUT_OK; return json_true(); } static json_t *rpc_shutdown(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; state->stopping = true; *error = NAUT_OK; return json_true(); } static int b64_val(int c) { if (c >= 'A' && c <= 'Z') return c - 'A'; if (c >= 'a' && c <= 'z') return c - 'a' + 26; if (c >= '0' && c <= '9') return c - '0' + 52; if (c == '+') return 62; if (c == '/') return 63; return -1; /* padding / whitespace / invalid -> skipped */ } /* Decode standard base64 (padding optional, whitespace ignored). */ static unsigned char *b64_decode(const char *in, size_t *out_len) { size_t cap = strlen(in) / 4 * 3 + 4; unsigned char *out = malloc(cap); if (!out) return NULL; size_t o = 0; int acc = 0, bits = 0; for (const char *p = in; *p; p++) { if (*p == '=') break; int v = b64_val((unsigned char)*p); if (v < 0) continue; acc = (acc << 6) | v; bits += 6; if (bits >= 8) { bits -= 8; out[o++] = (unsigned char)((acc >> bits) & 0xff); } } *out_len = o; return out; } static bool write_all_fd(int fd, const void *buf, size_t len) { const char *p = buf; while (len) { ssize_t n = write(fd, p, len); if (n < 0) { if (errno == EINTR) continue; return false; } p += n; len -= (size_t)n; } return true; } /* A browser uploads a .torrent's bytes as base64 in "data"; the daemon writes * its own file and owns its lifecycle (no shared path with the client). When * persistence is enabled the file goes under /uploads so it survives * a restart (*managed = true, unlink only on remove); otherwise it lands in /tmp * (*managed = false, unlink on any destroy). Returns the path in `out`. */ static bool add_torrent_write_upload(daemon_state *state, const char *data_b64, char *out, size_t cap, bool *managed, naut_err *error) { size_t raw_len = 0; unsigned char *raw = b64_decode(data_b64, &raw_len); if (!raw || raw_len == 0) { free(raw); *error = NAUT_ERR_INVAL; return false; } char tmpl[PATH_MAX + 32]; if (state->persist_enabled) snprintf(tmpl, sizeof tmpl, "%s/upload-XXXXXX", state->uploads_dir); else snprintf(tmpl, sizeof tmpl, "/tmp/naut-upload-XXXXXX"); int fd = mkstemp(tmpl); if (fd < 0) { free(raw); *error = NAUT_ERR_IO; return false; } bool ok = write_all_fd(fd, raw, raw_len); close(fd); free(raw); if (!ok) { unlink(tmpl); *error = NAUT_ERR_IO; return false; } size_t path_len = strlen(tmpl); if (path_len + 1 > cap) { unlink(tmpl); *error = NAUT_ERR_INVAL; return false; } memcpy(out, tmpl, path_len + 1); *managed = state->persist_enabled; return true; } /* --- persistence --------------------------------------------------------- */ /* On-disk record for one torrent (caller holds task->lock). */ static json_t *torrent_record(const torrent_task *task) { json_t *rec = json_object(); if (!rec) return NULL; json_object_set_new(rec, "torrent_id", json_integer((json_int_t)task->id)); json_object_set_new(rec, "source", json_string(task->source)); json_object_set_new(rec, "output", json_string(task->output_dir)); json_object_set_new(rec, "source_managed", json_boolean(task->source_managed)); if (task->name) json_object_set_new(rec, "name", json_string(task->name)); json_object_set_new(rec, "paused", json_boolean(task->paused)); json_object_set_new(rec, "force_start", json_boolean(task->force_start)); json_object_set_new(rec, "queue_pos", json_integer(task->queue_pos)); json_t *peers = json_array(); if (peers) { for (size_t i = 0; i < task->num_peers; i++) json_array_append_new(peers, json_string(task->peers[i])); json_object_set_new(rec, "peers", peers); } if (task->num_locations) { json_t *locations = json_array(); if (locations) { for (size_t i = 0; i < task->num_locations; i++) json_array_append_new(locations, json_pack( "{s:i,s:s}", "file", (int)task->locations[i].file_index, "path", task->locations[i].path)); json_object_set_new(rec, "locations", locations); } } if (task->category && *task->category) json_object_set_new(rec, "category", json_string(task->category)); if (task->num_tags) json_object_set_new(rec, "tags", task_tags_json(task)); if (task->pending_save_path) { json_object_set_new(rec, "pending_save_path", json_string(task->pending_save_path)); json_object_set_new(rec, "pending_save_path_reset", json_boolean(task->pending_save_path_reset)); } return rec; } /* Atomically write the current (non-removed) torrent set to state_file. */ static void persist_torrents(daemon_state *state) { if (!state->persist_enabled) return; json_t *array = json_array(); if (!array) return; pthread_mutex_lock(&state->torrent_lock); for (size_t i = 0; i < state->torrent_count; i++) { torrent_task *task = state->torrents[i]; pthread_mutex_lock(&task->lock); json_t *rec = task->remove_requested ? NULL : torrent_record(task); pthread_mutex_unlock(&task->lock); if (rec) json_array_append_new(array, rec); } pthread_mutex_unlock(&state->torrent_lock); char tmp[PATH_MAX + 8]; snprintf(tmp, sizeof tmp, "%s.tmp", state->state_file); if (json_dump_file(array, tmp, JSON_INDENT(2)) != 0) { NAUT_WARN("persist: write %s failed", tmp); unlink(tmp); } else if (rename(tmp, state->state_file) != 0) { NAUT_WARN("persist: rename to %s failed: %s", state->state_file, strerror(errno)); unlink(tmp); } json_decref(array); } /* --- daemon preferences (queue limit + throttle) ------------------------ */ static void persist_prefs(daemon_state *state) { if (!state->persist_enabled) return; json_t *p = json_pack( "{s:i,s:I,s:I,s:I,s:I,s:b}", "max_active", (json_int_t)state->max_active, "dl_limit", (json_int_t)state->dl_limit, "alt_dl_limit", (json_int_t)state->alt_dl_limit, "up_limit", (json_int_t)state->up_limit, "alt_up_limit", (json_int_t)state->alt_up_limit, "alt_speed_enabled", state->alt_speed_enabled); if (!p) return; char tmp[PATH_MAX + 8]; snprintf(tmp, sizeof tmp, "%s.tmp", state->prefs_file); if (json_dump_file(p, tmp, JSON_INDENT(2)) != 0 || rename(tmp, state->prefs_file) != 0) { NAUT_WARN("persist: write %s failed", state->prefs_file); unlink(tmp); } json_decref(p); } static void load_prefs(daemon_state *state) { if (!state->persist_enabled) return; json_error_t jerr; json_t *p = json_load_file(state->prefs_file, 0, &jerr); if (!p) return; json_t *v; if ((v = json_object_get(p, "max_active")) && json_is_integer(v) && json_integer_value(v) > 0) state->max_active = (uint32_t)json_integer_value(v); if ((v = json_object_get(p, "dl_limit")) && json_is_integer(v)) state->dl_limit = (uint64_t)json_integer_value(v); if ((v = json_object_get(p, "alt_dl_limit")) && json_is_integer(v)) state->alt_dl_limit = (uint64_t)json_integer_value(v); if ((v = json_object_get(p, "up_limit")) && json_is_integer(v)) state->up_limit = (uint64_t)json_integer_value(v); if ((v = json_object_get(p, "alt_up_limit")) && json_is_integer(v)) state->alt_up_limit = (uint64_t)json_integer_value(v); if ((v = json_object_get(p, "alt_speed_enabled"))) state->alt_speed_enabled = json_boolean_value(v); json_decref(p); } static json_t *prefs_json(daemon_state *state) { return json_pack( "{s:i,s:I,s:I,s:I,s:I,s:b}", "max_active", (json_int_t)state->max_active, "dl_limit", (json_int_t)state->dl_limit, "alt_dl_limit", (json_int_t)state->alt_dl_limit, "up_limit", (json_int_t)state->up_limit, "alt_up_limit", (json_int_t)state->alt_up_limit, "alt_speed_enabled", state->alt_speed_enabled); } /* --- script settings (schema declared by the script, values set by the UI) - */ static const char *jstr(const json_t *obj, const char *key, const char *fallback) { const char *v = json_string_value(json_object_get(obj, key)); return v ? v : fallback; } /* Coerce any JSON scalar to a freshly allocated string ("true"/"false" for * bools, plain digits for numbers). Returns NULL for non-scalars. */ static char *json_scalar_to_string(const json_t *v) { if (json_is_string(v)) return strdup(json_string_value(v)); if (json_is_true(v)) return strdup("true"); if (json_is_false(v)) return strdup("false"); if (json_is_integer(v)) { char buf[32]; snprintf(buf, sizeof buf, "%lld", (long long)json_integer_value(v)); return strdup(buf); } if (json_is_real(v)) { char buf[32]; snprintf(buf, sizeof buf, "%g", json_real_value(v)); return strdup(buf); } return NULL; } /* Find a schema entry by key (caller holds settings_lock). */ static json_t *settings_schema_entry(daemon_state *state, const char *key) { if (!state->script_settings_schema) return NULL; size_t i; json_t *entry; json_array_foreach(state->script_settings_schema, i, entry) if (strcmp(jstr(entry, "key", ""), key) == 0) return entry; return NULL; } static void persist_script_settings(daemon_state *state) { if (!state->persist_enabled) return; pthread_mutex_lock(&state->settings_lock); json_t *copy = state->script_settings ? json_deep_copy(state->script_settings) : json_object(); pthread_mutex_unlock(&state->settings_lock); if (!copy) return; char tmp[PATH_MAX + 8]; snprintf(tmp, sizeof tmp, "%s.tmp", state->settings_file); if (json_dump_file(copy, tmp, JSON_INDENT(2)) != 0 || rename(tmp, state->settings_file) != 0) { NAUT_WARN("persist: write %s failed", state->settings_file); unlink(tmp); } json_decref(copy); } static void load_script_settings(daemon_state *state) { if (!state->persist_enabled) return; json_error_t jerr; json_t *v = json_load_file(state->settings_file, 0, &jerr); if (!v) return; if (json_is_object(v)) { pthread_mutex_lock(&state->settings_lock); json_decref(state->script_settings); state->script_settings = v; pthread_mutex_unlock(&state->settings_lock); } else { json_decref(v); } } /* Host callback: the script (re)declared its settings schema. */ static void daemon_define_settings(void *opaque, const naut_script_setting_def *defs, size_t count) { daemon_state *state = opaque; json_t *schema = json_array(); if (!schema) return; for (size_t i = 0; i < count; i++) { const char *type = defs[i].type ? defs[i].type : "string"; json_t *entry = json_pack( "{s:s,s:s,s:s,s:s}", "key", defs[i].key, "label", defs[i].label ? defs[i].label : defs[i].key, "type", type, "default", defs[i].default_value ? defs[i].default_value : ""); if (entry) json_array_append_new(schema, entry); } pthread_mutex_lock(&state->settings_lock); json_decref(state->script_settings_schema); state->script_settings_schema = schema; pthread_mutex_unlock(&state->settings_lock); } /* Host callback: resolve a setting (user value, else declared default). */ static char *daemon_get_setting(void *opaque, const char *key, naut_setting_type *type) { daemon_state *state = opaque; char *out = NULL; *type = NAUT_SETTING_STRING; pthread_mutex_lock(&state->settings_lock); json_t *entry = settings_schema_entry(state, key); const char *tname = entry ? jstr(entry, "type", "string") : "string"; if (strcmp(tname, "bool") == 0) *type = NAUT_SETTING_BOOL; else if (strcmp(tname, "number") == 0) *type = NAUT_SETTING_NUMBER; json_t *value = state->script_settings ? json_object_get(state->script_settings, key) : NULL; if (value) out = json_scalar_to_string(value); else if (entry) out = strdup(jstr(entry, "default", "")); pthread_mutex_unlock(&state->settings_lock); return out; } /* The settings block for script_status: schema fields plus the effective value * (user-set if present, otherwise the declared default). */ static json_t *script_settings_json(daemon_state *state) { json_t *out = json_array(); if (!out) return NULL; pthread_mutex_lock(&state->settings_lock); if (state->script_settings_schema) { size_t i; json_t *entry; json_array_foreach(state->script_settings_schema, i, entry) { const char *key = jstr(entry, "key", ""); const char *def = jstr(entry, "default", ""); json_t *uv = state->script_settings ? json_object_get(state->script_settings, key) : NULL; char *vs = uv ? json_scalar_to_string(uv) : NULL; json_t *item = json_pack( "{s:s,s:s,s:s,s:s,s:s}", "key", key, "label", jstr(entry, "label", key), "type", jstr(entry, "type", "string"), "default", def, "value", vs ? vs : def); free(vs); if (item) json_array_append_new(out, item); } } pthread_mutex_unlock(&state->settings_lock); return out; } static naut_script_host script_host(daemon_state *state) { naut_script_host host = { .move_file = queue_move, .labels = script_labels, .define_settings = daemon_define_settings, .get_setting = daemon_get_setting, .context = state, }; return host; } /* Merge user-supplied values (keys must exist in the schema) and persist. */ static json_t *rpc_set_script_settings(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; json_t *settings = json_is_object(params) ? json_object_get(params, "settings") : NULL; if (!json_is_object(settings)) { *error = NAUT_ERR_INVAL; return NULL; } pthread_mutex_lock(&state->settings_lock); if (!state->script_settings) state->script_settings = json_object(); if (state->script_settings) { const char *key; json_t *value; json_object_foreach(settings, key, value) { if (!settings_schema_entry(state, key)) continue; /* unknown key */ char *vs = json_scalar_to_string(value); if (vs) { json_object_set_new(state->script_settings, key, json_string(vs)); free(vs); } } } pthread_mutex_unlock(&state->settings_lock); persist_script_settings(state); *error = NAUT_OK; return script_status_json(state); } /* --- label taxonomy (web-layer category + tag lists, persisted here) ------- */ static void persist_taxonomy(daemon_state *state) { if (!state->persist_enabled) return; pthread_mutex_lock(&state->taxonomy_lock); json_t *cats = state->label_categories ? json_deep_copy(state->label_categories) : json_array(); json_t *tags = state->label_tags ? json_deep_copy(state->label_tags) : json_array(); pthread_mutex_unlock(&state->taxonomy_lock); json_t *doc = json_object(); if (!doc) { json_decref(cats); json_decref(tags); return; } json_object_set_new(doc, "categories", cats ? cats : json_array()); json_object_set_new(doc, "tags", tags ? tags : json_array()); char tmp[PATH_MAX + 8]; snprintf(tmp, sizeof tmp, "%s.tmp", state->taxonomy_file); if (json_dump_file(doc, tmp, JSON_INDENT(2)) != 0 || rename(tmp, state->taxonomy_file) != 0) { NAUT_WARN("persist: write %s failed", state->taxonomy_file); unlink(tmp); } json_decref(doc); } static void load_taxonomy(daemon_state *state) { if (!state->persist_enabled) return; json_error_t jerr; json_t *doc = json_load_file(state->taxonomy_file, 0, &jerr); if (!doc) return; json_t *cats = json_object_get(doc, "categories"); json_t *tags = json_object_get(doc, "tags"); pthread_mutex_lock(&state->taxonomy_lock); if (json_is_array(cats)) { json_decref(state->label_categories); state->label_categories = json_deep_copy(cats); } if (json_is_array(tags)) { json_decref(state->label_tags); state->label_tags = json_deep_copy(tags); } pthread_mutex_unlock(&state->taxonomy_lock); json_decref(doc); } static json_t *rpc_get_label_taxonomy(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; pthread_mutex_lock(&state->taxonomy_lock); json_t *cats = state->label_categories ? json_deep_copy(state->label_categories) : json_array(); json_t *tags = state->label_tags ? json_deep_copy(state->label_tags) : json_array(); pthread_mutex_unlock(&state->taxonomy_lock); json_t *out = json_object(); if (!out) { json_decref(cats); json_decref(tags); *error = NAUT_ERR_NOMEM; return NULL; } json_object_set_new(out, "categories", cats ? cats : json_array()); json_object_set_new(out, "tags", tags ? tags : json_array()); *error = NAUT_OK; return out; } static json_t *rpc_set_label_taxonomy(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; if (!json_is_object(params)) { *error = NAUT_ERR_INVAL; return NULL; } json_t *cats = json_object_get(params, "categories"); json_t *tags = json_object_get(params, "tags"); pthread_mutex_lock(&state->taxonomy_lock); if (json_is_array(cats)) { json_decref(state->label_categories); state->label_categories = json_deep_copy(cats); } if (json_is_array(tags)) { json_decref(state->label_tags); state->label_tags = json_deep_copy(tags); } pthread_mutex_unlock(&state->taxonomy_lock); persist_taxonomy(state); *error = NAUT_OK; return json_object(); } /* --- generic web-UI blob store (RSS feeds, indexer config, ...) ------------ * * The web layer owns these schemas; the daemon only persists them, one JSON * document per key, under /webui_.json. Keys are sanitized to a * safe filename charset so a key can never escape the state directory. */ static bool blob_path(daemon_state *state, const char *key, char *out, size_t n) { if (!key || !*key || !state->data_dir[0]) return false; char safe[64]; size_t j = 0; for (size_t i = 0; key[i] && j + 1 < sizeof safe; i++) { char c = key[i]; if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' || c == '-') safe[j++] = c; } safe[j] = 0; if (j == 0) return false; return (size_t)snprintf(out, n, "%s/webui_%s.json", state->data_dir, safe) < n; } static json_t *rpc_get_webui_blob(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; const char *key = json_string_value(json_object_get(params, "key")); char path[PATH_MAX]; if (!blob_path(state, key, path, sizeof path)) { *error = NAUT_ERR_INVAL; return NULL; } pthread_mutex_lock(&state->blob_lock); json_error_t jerr; json_t *value = json_load_file(path, 0, &jerr); pthread_mutex_unlock(&state->blob_lock); json_t *out = json_object(); if (!out) { json_decref(value); *error = NAUT_ERR_NOMEM; return NULL; } json_object_set_new(out, "value", value ? value : json_null()); *error = NAUT_OK; return out; } static json_t *rpc_set_webui_blob(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; const char *key = json_string_value(json_object_get(params, "key")); json_t *value = json_object_get(params, "value"); char path[PATH_MAX]; if (!value || !blob_path(state, key, path, sizeof path)) { *error = NAUT_ERR_INVAL; return NULL; } *error = NAUT_OK; if (!state->persist_enabled) return json_object(); pthread_mutex_lock(&state->blob_lock); char tmp[PATH_MAX + 8]; snprintf(tmp, sizeof tmp, "%s.tmp", path); if (json_dump_file(value, tmp, JSON_INDENT(2)) != 0 || rename(tmp, path) != 0) { NAUT_WARN("persist: write %s failed", path); unlink(tmp); } pthread_mutex_unlock(&state->blob_lock); return json_object(); } /* --- data-overlap guard (block torrents that would write the same files) --- */ static uint8_t *slurp_file(const char *path, size_t *len) { FILE *f = fopen(path, "rb"); if (!f) return NULL; if (fseek(f, 0, SEEK_END) != 0) { fclose(f); return NULL; } long n = ftell(f); if (n < 0) { fclose(f); return NULL; } rewind(f); uint8_t *buf = malloc((size_t)n + 1); if (!buf) { fclose(f); return NULL; } size_t got = fread(buf, 1, (size_t)n, f); fclose(f); if (got != (size_t)n) { free(buf); return NULL; } if (len) *len = got; return buf; } /* Parse a .torrent file's file list into relative paths. Returns the count (0 * for magnets, which have no metadata yet, or on failure). Caller frees. */ static size_t parse_torrent_files(const char *source, char ***out) { *out = NULL; if (!source || strncmp(source, "magnet:", 7) == 0) return 0; size_t len = 0; uint8_t *bytes = slurp_file(source, &len); if (!bytes) return 0; naut_metainfo mi; memset(&mi, 0, sizeof mi); if (naut_metainfo_parse(bytes, len, &mi) != NAUT_OK) { free(bytes); return 0; } free(bytes); char **files = mi.num_files ? calloc(mi.num_files, sizeof *files) : NULL; size_t c = 0; if (files) for (size_t i = 0; i < mi.num_files; i++) if (mi.files[i].path) { char *p = strdup(mi.files[i].path); if (p) files[c++] = p; } naut_metainfo_free(&mi); *out = files; return c; } /* Two absolute paths conflict if equal, or one is a directory-prefix of the * other (a file vs a folder that would contain it). */ static bool paths_conflict(const char *a, const char *b) { size_t la = strlen(a), lb = strlen(b); if (la == lb) return strcmp(a, b) == 0; const char *shorter = la < lb ? a : b, *longer = la < lb ? b : a; size_t sl = la < lb ? la : lb; return strncmp(shorter, longer, sl) == 0 && longer[sl] == '/'; } /* Current absolute path of file `fi` (relpath `rel`) for `task`: its override if * individually moved, else output_dir/rel. Caller holds task->lock. */ static void task_file_abs(const torrent_task *task, size_t fi, const char *rel, char *out, size_t outsz) { for (size_t i = 0; i < task->num_locations; i++) if (task->locations[i].file_index == fi) { snprintf(out, outsz, "%s", task->locations[i].path); return; } snprintf(out, outsz, "%s/%s", task->output_dir ? task->output_dir : "", rel); } /* True if any of `news` (absolute paths) would write where an already-registered * torrent's file lives. Caller holds state->torrent_lock; locks each task. */ static bool data_overlaps_locked(daemon_state *state, char *const *news, size_t nnew, char *conflict, size_t csz) { for (size_t t = 0; t < state->torrent_count; t++) { torrent_task *o = state->torrents[t]; pthread_mutex_lock(&o->lock); bool removed = o->remove_requested; for (size_t fi = 0; !removed && fi < o->num_files; fi++) { char have[PATH_MAX]; task_file_abs(o, fi, o->files[fi], have, sizeof have); for (size_t k = 0; k < nnew; k++) if (paths_conflict(have, news[k])) { snprintf(conflict, csz, "%s", have); pthread_mutex_unlock(&o->lock); return true; } } pthread_mutex_unlock(&o->lock); } return false; } /* Build a torrent_task from validated inputs, register it under an id, and start * its worker. Returns the task (added to state->torrents) or NULL + *error. All * inputs are copied; the source file is never unlinked here (the caller owns * that decision so a failed restore does not delete a durable upload). When * `check_overlap` is set, an add whose files would land on an existing torrent's * data is refused with NAUT_ERR_EXIST. */ static torrent_task *spawn_torrent(daemon_state *state, const char *source, bool source_managed, bool source_is_temp, const char *output, const json_t *peers_json, const json_t *locations_json, const json_t *meta_json, const json_t *id_opt, const char *name, bool start_paused, bool force_start, int queue_pos, bool check_overlap, naut_err *error) { torrent_task *task = calloc(1, sizeof(*task)); if (!task) { *error = NAUT_ERR_NOMEM; return NULL; } task->daemon = state; task->state = start_paused ? TORRENT_PAUSED : TORRENT_QUEUED; task->result = NAUT_ERR_AGAIN; task->paused = start_paused; /* A paused torrent never runs a download worker, so hash-check its data once * (via a check-only worker) to report accurate progress. */ task->needs_check = start_paused; task->force_start = force_start; task->source = strdup(source); task->source_is_temp = source_is_temp; task->source_managed = source_managed; task->output_dir = strdup(output); task->name = (name && *name) ? strdup(name) : NULL; if (!task->source || !task->output_dir || (name && *name && !task->name)) { *error = NAUT_ERR_NOMEM; goto fail_early; } if (pthread_mutex_init(&task->lock, NULL) != 0) { *error = NAUT_ERR_NOMEM; goto fail_early; } task->num_peers = peers_json ? json_array_size(peers_json) : 0; if (task->num_peers) { task->peers = calloc(task->num_peers, sizeof(*task->peers)); if (!task->peers) { *error = NAUT_ERR_NOMEM; goto fail_task; } for (size_t i = 0; i < task->num_peers; i++) { const char *peer = json_string_value(json_array_get(peers_json, i)); if (!peer || !*peer) { *error = NAUT_ERR_INVAL; goto fail_task; } task->peers[i] = strdup(peer); if (!task->peers[i]) { *error = NAUT_ERR_NOMEM; goto fail_task; } } } size_t nloc = locations_json ? json_array_size(locations_json) : 0; for (size_t i = 0; i < nloc; i++) { json_t *entry = json_array_get(locations_json, i); json_t *fidx = json_object_get(entry, "file"); const char *path = json_string_value(json_object_get(entry, "path")); if (!json_is_integer(fidx) || json_integer_value(fidx) < 0 || !path) continue; /* skip malformed entries rather than fail the restore */ task_set_location(task, (uint32_t)json_integer_value(fidx), path); } if (meta_json) { task_set_category(task, json_string_value(json_object_get(meta_json, "category"))); json_t *tags = json_object_get(meta_json, "tags"); /* Migrate the old flat "labels" record into tags. */ if (!json_is_array(tags)) tags = json_object_get(meta_json, "labels"); task_set_tags(task, tags); const char *pending = json_string_value(json_object_get(meta_json, "pending_save_path")); if (pending && *pending) { task->pending_save_path = strdup(pending); task->pending_save_path_reset = json_boolean_value(json_object_get(meta_json, "pending_save_path_reset")); } } /* File list (for overlap detection); empty for magnets until metadata. */ task->num_files = parse_torrent_files(source, &task->files); pthread_mutex_lock(&state->torrent_lock); if (state->torrent_count == MAX_TORRENTS) { pthread_mutex_unlock(&state->torrent_lock); *error = NAUT_ERR_FULL; goto fail_task; } if (check_overlap && task->num_files) { char **news = calloc(task->num_files, sizeof *news); bool ok = news != NULL; for (size_t i = 0; ok && i < task->num_files; i++) { char tmp[PATH_MAX]; snprintf(tmp, sizeof tmp, "%s/%s", output, task->files[i]); news[i] = strdup(tmp); if (!news[i]) ok = false; } char conflict[PATH_MAX] = {0}; bool overlap = ok && data_overlaps_locked(state, news, task->num_files, conflict, sizeof conflict); for (size_t i = 0; news && i < task->num_files; i++) free(news[i]); free(news); if (!ok || overlap) { pthread_mutex_unlock(&state->torrent_lock); if (overlap) NAUT_WARN("add blocked: data overlaps existing torrent at %s", conflict); *error = overlap ? NAUT_ERR_EXIST : NAUT_ERR_NOMEM; goto fail_task; } } if (id_opt) { if (!json_is_integer(id_opt) || json_integer_value(id_opt) < 0) { pthread_mutex_unlock(&state->torrent_lock); *error = NAUT_ERR_INVAL; goto fail_task; } task->id = (uint64_t)json_integer_value(id_opt); if (task->id < (uint64_t)INT64_MAX && task->id >= state->next_torrent_id) state->next_torrent_id = task->id + 1; } else { while (find_torrent_locked(state, state->next_torrent_id)) state->next_torrent_id++; if (state->next_torrent_id > (uint64_t)INT64_MAX) { pthread_mutex_unlock(&state->torrent_lock); *error = NAUT_ERR_FULL; goto fail_task; } task->id = state->next_torrent_id++; } if (find_torrent_locked(state, task->id)) { pthread_mutex_unlock(&state->torrent_lock); *error = NAUT_ERR_INVAL; goto fail_task; } if (queue_pos >= 0) { task->queue_pos = queue_pos; } else { /* append to the tail of the queue */ int max_pos = 0; for (size_t i = 0; i < state->torrent_count; i++) if (state->torrents[i]->queue_pos > max_pos) max_pos = state->torrents[i]->queue_pos; task->queue_pos = max_pos + 1; } state->torrents[state->torrent_count++] = task; pthread_mutex_unlock(&state->torrent_lock); /* Register without a worker; the lifecycle reconciler starts it when it is * within the active-download budget (and not paused). thread_done=true marks * it as cleanly (re)startable. */ task->thread_done = true; *error = NAUT_OK; return task; fail_task: for (size_t i = 0; i < task->num_peers; i++) free(task->peers[i]); free(task->peers); for (size_t i = 0; i < task->num_locations; i++) free(task->locations[i].path); free(task->locations); for (size_t i = 0; i < task->num_tags; i++) free(task->tags[i]); free(task->tags); for (size_t i = 0; i < task->num_files; i++) free(task->files[i]); free(task->files); free(task->category); free(task->pending_save_path); pthread_mutex_destroy(&task->lock); fail_early: free(task->name); free(task->source); free(task->output_dir); free(task); return NULL; } static json_t *rpc_add_torrent(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; if (!json_is_object(params)) { *error = NAUT_ERR_INVAL; return NULL; } const char *source = json_string_value(json_object_get(params, "source")); if (!source) source = json_string_value(json_object_get(params, "torrent")); const char *data_b64 = json_string_value(json_object_get(params, "data")); const char *output = json_string_value(json_object_get(params, "output")); if (!output) output = json_string_value(json_object_get(params, "root")); json_t *peers_json = json_object_get(params, "peers"); /* need an output and either a source (path/magnet) or uploaded bytes */ if (!output || !*output || ((!source || !*source) && (!data_b64 || !*data_b64)) || (peers_json && !json_is_array(peers_json))) { *error = NAUT_ERR_INVAL; return NULL; } const char *name = json_string_value(json_object_get(params, "name")); bool start_paused = json_boolean_value(json_object_get(params, "paused")); /* materialize an upload into a daemon-owned .torrent (durable under * state_dir/uploads when persistence is on, else an ephemeral /tmp file). */ char temp_source[PATH_MAX]; bool managed = false, is_temp = false; if ((!source || !*source) && data_b64 && *data_b64) { if (!add_torrent_write_upload(state, data_b64, temp_source, sizeof temp_source, &managed, error)) return NULL; source = temp_source; is_temp = !managed; /* /tmp fallback when persistence is disabled */ } torrent_task *task = spawn_torrent( state, source, managed, is_temp, output, peers_json, /*locations_json=*/NULL, /*meta_json=*/params, json_object_get(params, "torrent_id"), name, start_paused, /*force_start=*/false, /*queue_pos=*/-1, /*check_overlap=*/true, error); if (!task) { if (managed || is_temp) unlink(source); return NULL; } persist_torrents(state); service_lifecycle(state); /* start it now if within the active budget */ return torrent_json(task); } static json_t *rpc_torrents(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; json_t *result = json_array(); if (!result) { *error = NAUT_ERR_NOMEM; return NULL; } pthread_mutex_lock(&state->torrent_lock); for (size_t i = 0; i < state->torrent_count; i++) { json_t *item = torrent_json(state->torrents[i]); if (!item || json_array_append_new(result, item) != 0) { json_decref(item); json_decref(result); pthread_mutex_unlock(&state->torrent_lock); *error = NAUT_ERR_NOMEM; return NULL; } } pthread_mutex_unlock(&state->torrent_lock); *error = NAUT_OK; return result; } static bool parse_torrent_id(const json_t *params, uint64_t *id) { if (!json_is_object(params)) return false; json_t *value = json_object_get(params, "torrent_id"); if (!json_is_integer(value) || json_integer_value(value) < 0) return false; *id = (uint64_t)json_integer_value(value); return true; } static json_t *rpc_torrent(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; uint64_t id; if (!parse_torrent_id(params, &id)) { *error = NAUT_ERR_INVAL; return NULL; } pthread_mutex_lock(&state->torrent_lock); torrent_task *task = find_torrent_locked(state, id); json_t *result = task ? torrent_json(task) : NULL; pthread_mutex_unlock(&state->torrent_lock); *error = task ? (result ? NAUT_OK : NAUT_ERR_NOMEM) : NAUT_ERR_NOTFOUND; return result; } static json_t *rpc_remove_torrent(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; uint64_t id; if (!parse_torrent_id(params, &id)) { *error = NAUT_ERR_INVAL; return NULL; } pthread_mutex_lock(&state->torrent_lock); torrent_task *task = find_torrent_locked(state, id); if (task) { pthread_mutex_lock(&task->lock); task->stop_requested = true; task->remove_requested = true; if (!task->thread_done) task->state = TORRENT_STOPPING; pthread_mutex_unlock(&task->lock); } pthread_mutex_unlock(&state->torrent_lock); if (!task) { *error = NAUT_ERR_NOTFOUND; return NULL; } persist_torrents(state); /* drop the removed torrent from disk now */ *error = NAUT_OK; return torrent_json(task); } /* Find a torrent by id, run `apply` under its lock, persist, and return its * json. Shared by pause/resume/recheck. */ static json_t *torrent_flag_op(daemon_state *state, const json_t *params, void (*apply)(torrent_task *, const json_t *), naut_err *error) { uint64_t id; if (!parse_torrent_id(params, &id)) { *error = NAUT_ERR_INVAL; return NULL; } pthread_mutex_lock(&state->torrent_lock); torrent_task *task = find_torrent_locked(state, id); if (task) { pthread_mutex_lock(&task->lock); apply(task, params); pthread_mutex_unlock(&task->lock); } pthread_mutex_unlock(&state->torrent_lock); if (!task) { *error = NAUT_ERR_NOTFOUND; return NULL; } persist_torrents(state); service_lifecycle(state); /* apply the desired-state change immediately */ *error = NAUT_OK; return torrent_json(task); } static void apply_pause(torrent_task *task, const json_t *params) { (void)params; task->paused = true; task->force_start = false; if (!task->thread_done) { task->stop_requested = true; task->state = TORRENT_STOPPING; } else { task->state = TORRENT_PAUSED; } } static void apply_resume(torrent_task *task, const json_t *params) { task->paused = false; task->force_start = json_boolean_value(json_object_get(params, "force")); /* The reconciler activates it (subject to the queue, or immediately if * forced); leaving the flags is enough. */ } static void apply_recheck(torrent_task *task, const json_t *params) { (void)params; if (task->paused) { /* Recheck a paused torrent: hash-verify in place and stay paused (a * check-only worker runs via the reconciler). */ task->needs_check = true; return; } /* Active torrent: stop->start so the fresh run re-hashes via the resume scan. */ task->restart_requested = true; if (!task->thread_done) { task->stop_requested = true; task->state = TORRENT_STOPPING; } } static void apply_set_labels(torrent_task *task, const json_t *params) { /* category and/or tags; absent fields leave that part unchanged. */ task_set_category(task, json_string_value(json_object_get(params, "category"))); json_t *tags = json_object_get(params, "tags"); if (!json_is_array(tags)) tags = json_object_get(params, "labels"); task_set_tags(task, tags); } static json_t *rpc_pause_torrent(void *opaque, const json_t *params, naut_err *error) { return torrent_flag_op(opaque, params, apply_pause, error); } static json_t *rpc_set_labels(void *opaque, const json_t *params, naut_err *error) { return torrent_flag_op(opaque, params, apply_set_labels, error); } /* "Set location": queue a one-time move of the torrent's files to a new base. * The worker performs it on its next control pass (see apply_pending_save_path); * a stopped torrent applies it when it next runs. */ static void apply_set_save_path(torrent_task *task, const json_t *params) { const char *path = json_string_value(json_object_get(params, "savePath")); if (!path || !*path || strlen(path) >= PATH_MAX) return; bool reset = json_boolean_value(json_object_get(params, "reset")); /* Same base with nothing to reset is a no-op; with reset it still pulls any * individually-moved files back to their original relpaths. */ if (!reset && task->output_dir && strcmp(task->output_dir, path) == 0) return; free(task->pending_save_path); task->pending_save_path = strdup(path); task->pending_save_path_reset = reset; } static json_t *rpc_set_save_path(void *opaque, const json_t *params, naut_err *error) { return torrent_flag_op(opaque, params, apply_set_save_path, error); } static json_t *rpc_resume_torrent(void *opaque, const json_t *params, naut_err *error) { return torrent_flag_op(opaque, params, apply_resume, error); } static json_t *rpc_recheck_torrent(void *opaque, const json_t *params, naut_err *error) { return torrent_flag_op(opaque, params, apply_recheck, error); } /* Render a diagnostic state dump for one torrent. The rich engine + piece state * lives on the swarm worker thread, so we bump the task's dump request and wait * for the worker to render it (via torrent_should_dump/torrent_on_dump), then * return the text. Re-resolves the task by id on every poll so a concurrently * reaped torrent is detected rather than dereferenced. */ static json_t *dump_result(const char *text) { json_t *result = json_object(); if (result) json_object_set_new(result, "dump", json_string(text)); return result; } static json_t *rpc_dump_torrent(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; uint64_t id; if (!parse_torrent_id(params, &id)) { *error = NAUT_ERR_INVAL; return NULL; } pthread_mutex_lock(&state->torrent_lock); torrent_task *task = find_torrent_locked(state, id); bool active = false; uint64_t want = 0; if (task) { pthread_mutex_lock(&task->lock); active = task->thread_started && !task->thread_done; want = ++task->dump_seq; pthread_mutex_unlock(&task->lock); } pthread_mutex_unlock(&state->torrent_lock); if (!task) { *error = NAUT_ERR_NOTFOUND; return NULL; } if (!active) { *error = NAUT_OK; return dump_result("torrent is not running; no live engine state to dump\n"); } /* Wait for the worker thread to service the request (~3s budget). */ char *text = NULL; bool vanished = false; for (int i = 0; i < 300 && !text && !vanished; i++) { usleep(10000); /* 10 ms */ pthread_mutex_lock(&state->torrent_lock); torrent_task *t = find_torrent_locked(state, id); if (!t) { vanished = true; } else { pthread_mutex_lock(&t->lock); if (t->dump_done_seq >= want && t->dump_text) text = strdup(t->dump_text); pthread_mutex_unlock(&t->lock); } pthread_mutex_unlock(&state->torrent_lock); } if (!text) { *error = NAUT_OK; return dump_result(vanished ? "torrent was removed before the dump completed\n" : "dump timed out: worker did not respond\n"); } json_t *result = dump_result(text); free(text); *error = result ? NAUT_OK : NAUT_ERR_NOMEM; return result; } /* Reorder the download queue: op = top | bottom | up | down. */ static json_t *rpc_queue_move(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; uint64_t id; const char *op = json_is_object(params) ? json_string_value(json_object_get(params, "op")) : NULL; if (!parse_torrent_id(params, &id) || !op) { *error = NAUT_ERR_INVAL; return NULL; } pthread_mutex_lock(&state->torrent_lock); torrent_task *task = find_torrent_locked(state, id); if (!task) { pthread_mutex_unlock(&state->torrent_lock); *error = NAUT_ERR_NOTFOUND; return NULL; } int self = task->queue_pos, lo = self, hi = self; torrent_task *prev = NULL, *next = NULL; /* nearest neighbors by position */ for (size_t i = 0; i < state->torrent_count; i++) { torrent_task *o = state->torrents[i]; if (o == task) continue; if (o->queue_pos < lo) lo = o->queue_pos; if (o->queue_pos > hi) hi = o->queue_pos; if (o->queue_pos < self && (!prev || o->queue_pos > prev->queue_pos)) prev = o; if (o->queue_pos > self && (!next || o->queue_pos < next->queue_pos)) next = o; } if (strcmp(op, "top") == 0) { task->queue_pos = lo - 1; } else if (strcmp(op, "bottom") == 0) { task->queue_pos = hi + 1; } else if (strcmp(op, "up") == 0 && prev) { int tmp = task->queue_pos; task->queue_pos = prev->queue_pos; prev->queue_pos = tmp; } else if (strcmp(op, "down") == 0 && next) { int tmp = task->queue_pos; task->queue_pos = next->queue_pos; next->queue_pos = tmp; } pthread_mutex_unlock(&state->torrent_lock); persist_torrents(state); service_lifecycle(state); /* reordering may change the active set */ *error = NAUT_OK; return torrent_json(task); } static json_t *rpc_get_preferences(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; json_t *result = prefs_json(state); *error = result ? NAUT_OK : NAUT_ERR_NOMEM; return result; } static json_t *rpc_set_preferences(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; if (!json_is_object(params)) { *error = NAUT_ERR_INVAL; return NULL; } json_t *v; if ((v = json_object_get(params, "max_active")) && json_is_integer(v) && json_integer_value(v) > 0) state->max_active = (uint32_t)json_integer_value(v); if ((v = json_object_get(params, "dl_limit")) && json_is_integer(v) && json_integer_value(v) >= 0) state->dl_limit = (uint64_t)json_integer_value(v); if ((v = json_object_get(params, "alt_dl_limit")) && json_is_integer(v) && json_integer_value(v) >= 0) state->alt_dl_limit = (uint64_t)json_integer_value(v); if ((v = json_object_get(params, "up_limit")) && json_is_integer(v) && json_integer_value(v) >= 0) state->up_limit = (uint64_t)json_integer_value(v); if ((v = json_object_get(params, "alt_up_limit")) && json_is_integer(v) && json_integer_value(v) >= 0) state->alt_up_limit = (uint64_t)json_integer_value(v); if ((v = json_object_get(params, "alt_speed_enabled"))) state->alt_speed_enabled = json_boolean_value(v); persist_prefs(state); service_lifecycle(state); /* apply new budget / throttle shares now */ json_t *result = prefs_json(state); *error = result ? NAUT_OK : NAUT_ERR_NOMEM; return result; } static json_t *rpc_toggle_altspeed(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; state->alt_speed_enabled = !state->alt_speed_enabled; persist_prefs(state); service_lifecycle(state); /* switch the active throttle immediately */ *error = NAUT_OK; return json_pack("{s:b}", "alt_speed_enabled", state->alt_speed_enabled); } static json_t *rpc_load_script(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; const char *path = json_is_object(params) ? json_string_value(json_object_get(params, "path")) : NULL; if (!path || !*path) { *error = NAUT_ERR_INVAL; return NULL; } char *path_copy = strdup(path); if (!path_copy) { *error = NAUT_ERR_NOMEM; return NULL; } naut_script_host host = script_host(state); naut_script *script = naut_script_create( state->events, path, 256, &host, error); if (!script) { free(path_copy); return NULL; } pthread_mutex_lock(&state->script_lock); naut_script *old = state->script; char *old_path = state->script_path; state->script = script; state->script_path = path_copy; pthread_mutex_unlock(&state->script_lock); *error = NAUT_OK; naut_script_destroy(old); free(old_path); return script_status_json(state); } static json_t *rpc_unload_script(void *opaque, const json_t *params, naut_err *error) { (void)params; daemon_state *state = opaque; pthread_mutex_lock(&state->script_lock); naut_script *old = state->script; char *old_path = state->script_path; state->script = NULL; state->script_path = NULL; pthread_mutex_unlock(&state->script_lock); naut_script_destroy(old); free(old_path); *error = NAUT_OK; return script_status_json(state); } static json_t *rpc_update_script(void *opaque, const json_t *params, naut_err *error) { daemon_state *state = opaque; json_t *source_json = json_is_object(params) ? json_object_get(params, "source") : NULL; if (!json_is_string(source_json)) { *error = NAUT_ERR_INVAL; return NULL; } const char *source = json_string_value(source_json); size_t source_len = json_string_length(source_json); pthread_mutex_lock(&state->script_lock); char *path = state->script_path ? strdup(state->script_path) : NULL; pthread_mutex_unlock(&state->script_lock); if (!path) { *error = NAUT_ERR_NOTFOUND; return NULL; } char tmp_path[PATH_MAX + 32]; int n = snprintf(tmp_path, sizeof tmp_path, "%s.update-XXXXXX", path); if (n < 0 || (size_t)n >= sizeof tmp_path) { free(path); *error = NAUT_ERR_RANGE; return NULL; } int fd = mkstemp(tmp_path); if (fd < 0) { free(path); *error = NAUT_ERR_IO; return NULL; } bool ok = write_all_fd(fd, source, source_len); if (close(fd) != 0) ok = false; if (!ok) { unlink(tmp_path); free(path); *error = NAUT_ERR_IO; return NULL; } naut_script_host host = script_host(state); naut_script *script = naut_script_create( state->events, tmp_path, 256, &host, error); if (!script) { unlink(tmp_path); free(path); return NULL; } if (rename(tmp_path, path) != 0) { naut_script_destroy(script); unlink(tmp_path); free(path); *error = NAUT_ERR_IO; return NULL; } pthread_mutex_lock(&state->script_lock); naut_script *old = state->script; state->script = script; pthread_mutex_unlock(&state->script_lock); naut_script_destroy(old); free(path); *error = NAUT_OK; return script_status_json(state); } static naut_err queue_move(void *opaque, uint64_t torrent_id, uint32_t file_index, const char *destination) { daemon_state *state = opaque; if (!destination || strlen(destination) >= PATH_MAX) return NAUT_ERR_RANGE; pthread_mutex_lock(&state->torrent_lock); torrent_task *task = find_torrent_locked(state, torrent_id); if (!task) { pthread_mutex_unlock(&state->torrent_lock); return NAUT_ERR_NOTFOUND; } pthread_mutex_lock(&task->lock); pthread_mutex_unlock(&state->torrent_lock); if (task->remove_requested) { pthread_mutex_unlock(&task->lock); return NAUT_ERR_NOTFOUND; } if (task->move_count == MOVE_QUEUE_CAPACITY) { pthread_mutex_unlock(&task->lock); return NAUT_ERR_FULL; } size_t tail = (task->move_head + task->move_count) % MOVE_QUEUE_CAPACITY; task->moves[tail].file_index = file_index; snprintf(task->moves[tail].destination, sizeof task->moves[tail].destination, "%s", destination); task->move_count++; pthread_mutex_unlock(&task->lock); return NAUT_OK; } static void broadcast_event(void *opaque, const naut_event *event) { daemon_state *state = opaque; json_t *payload = naut_rpc_event_json(event); if (!payload) return; pthread_mutex_lock(&state->subscriber_lock); for (size_t i = 0; i < state->subscriber_count;) { if (naut_rpc_send_json(state->subscribers[i], NAUT_RPC_EVENT, payload) == NAUT_OK) { i++; continue; } close(state->subscribers[i]); state->subscribers[i] = state->subscribers[--state->subscriber_count]; } pthread_mutex_unlock(&state->subscriber_lock); json_decref(payload); } static int listen_unix(const char *path) { int fd = socket(AF_UNIX, SOCK_STREAM, 0); if (fd < 0) return -1; struct sockaddr_un address; memset(&address, 0, sizeof address); address.sun_family = AF_UNIX; if (strlen(path) >= sizeof address.sun_path) { close(fd); return -1; } strcpy(address.sun_path, path); unlink(path); if (bind(fd, (struct sockaddr *)&address, sizeof address) != 0 || listen(fd, 32) != 0) { close(fd); return -1; } return fd; } static json_t *response(bool ok, json_t *result, naut_err error) { json_t *reply = json_object(); if (!reply) return NULL; json_object_set_new(reply, "ok", json_boolean(ok)); if (ok) { json_object_set(reply, "result", result ? result : json_null()); } else { json_object_set_new(reply, "code", json_integer(error)); json_object_set_new(reply, "error", json_string(naut_strerror(error))); } return reply; } static void add_subscriber(daemon_state *state, int fd) { int flags = fcntl(fd, F_GETFL, 0); if (flags >= 0) fcntl(fd, F_SETFL, flags | O_NONBLOCK); pthread_mutex_lock(&state->subscriber_lock); if (state->subscriber_count < MAX_SUBSCRIBERS) { state->subscribers[state->subscriber_count++] = fd; fd = -1; } pthread_mutex_unlock(&state->subscriber_lock); if (fd >= 0) close(fd); } static void handle_client(daemon_state *state, int fd) { struct timeval timeout = {.tv_sec = 2}; setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout); naut_rpc_frame_type type; json_t *request = NULL; naut_err error = naut_rpc_recv_json(fd, &type, &request); if (error != NAUT_OK || type != NAUT_RPC_REQUEST || !json_is_object(request)) { json_decref(request); close(fd); return; } const char *method = json_string_value(json_object_get(request, "method")); json_t *params = json_object_get(request, "params"); if (method && strcmp(method, "subscribe") == 0) { json_t *subscribed = json_string("subscribed"); json_t *reply = response(true, subscribed, NAUT_OK); json_decref(subscribed); if (reply && naut_rpc_send_json(fd, NAUT_RPC_RESPONSE, reply) == NAUT_OK) add_subscriber(state, fd); else close(fd); json_decref(reply); json_decref(request); return; } if (!method) error = NAUT_ERR_INVAL; json_t *result = method ? naut_rpc_dispatch(state->rpc, method, params, &error) : NULL; json_t *reply = response(error == NAUT_OK && result, result, error); json_decref(result); if (reply) { naut_rpc_send_json(fd, NAUT_RPC_RESPONSE, reply); json_decref(reply); } json_decref(request); close(fd); } static bool register_commands(daemon_state *state) { return naut_rpc_register(state->rpc, "ping", rpc_ping, state) == NAUT_OK && naut_rpc_register(state->rpc, "status", rpc_status, state) == NAUT_OK && naut_rpc_register(state->rpc, "plugins", rpc_plugins, state) == NAUT_OK && naut_rpc_register(state->rpc, "emit", rpc_emit, state) == NAUT_OK && naut_rpc_register(state->rpc, "add_torrent", rpc_add_torrent, state) == NAUT_OK && naut_rpc_register(state->rpc, "torrents", rpc_torrents, state) == NAUT_OK && naut_rpc_register(state->rpc, "torrent", rpc_torrent, state) == NAUT_OK && naut_rpc_register(state->rpc, "remove_torrent", rpc_remove_torrent, state) == NAUT_OK && naut_rpc_register(state->rpc, "pause_torrent", rpc_pause_torrent, state) == NAUT_OK && naut_rpc_register(state->rpc, "resume_torrent", rpc_resume_torrent, state) == NAUT_OK && naut_rpc_register(state->rpc, "recheck_torrent", rpc_recheck_torrent, state) == NAUT_OK && naut_rpc_register(state->rpc, "set_labels", rpc_set_labels, state) == NAUT_OK && naut_rpc_register(state->rpc, "set_save_path", rpc_set_save_path, state) == NAUT_OK && naut_rpc_register(state->rpc, "dump_torrent", rpc_dump_torrent, state) == NAUT_OK && naut_rpc_register(state->rpc, "queue_move", rpc_queue_move, state) == NAUT_OK && naut_rpc_register(state->rpc, "get_preferences", rpc_get_preferences, state) == NAUT_OK && naut_rpc_register(state->rpc, "set_preferences", rpc_set_preferences, state) == NAUT_OK && naut_rpc_register(state->rpc, "toggle_altspeed", rpc_toggle_altspeed, state) == NAUT_OK && naut_rpc_register(state->rpc, "load_script", rpc_load_script, state) == NAUT_OK && naut_rpc_register(state->rpc, "script_status", rpc_script_status, state) == NAUT_OK && naut_rpc_register(state->rpc, "update_script", rpc_update_script, state) == NAUT_OK && naut_rpc_register(state->rpc, "set_script_settings", rpc_set_script_settings, state) == NAUT_OK && naut_rpc_register(state->rpc, "get_label_taxonomy", rpc_get_label_taxonomy, state) == NAUT_OK && naut_rpc_register(state->rpc, "set_label_taxonomy", rpc_set_label_taxonomy, state) == NAUT_OK && naut_rpc_register(state->rpc, "get_webui_blob", rpc_get_webui_blob, state) == NAUT_OK && naut_rpc_register(state->rpc, "set_webui_blob", rpc_set_webui_blob, state) == NAUT_OK && naut_rpc_register(state->rpc, "unload_script", rpc_unload_script, state) == NAUT_OK && naut_rpc_register(state->rpc, "shutdown", rpc_shutdown, state) == NAUT_OK; } /* --- queue / lifecycle reconciler (main thread only) -------------------- */ /* (Re)start a torrent's worker. The task must have no live worker * (thread_done). Joins any prior thread first. Main thread only. */ /* Start the worker. `check` => a one-shot hash-check pass that keeps the torrent * paused (no download); otherwise a normal download run (clears paused). */ static bool start_worker_ex(torrent_task *task, bool check) { if (task->thread_started) { pthread_join(task->thread, NULL); task->thread_started = false; } pthread_mutex_lock(&task->lock); task->stop_requested = false; task->restart_requested = false; task->thread_done = false; task->checking = check; if (!check) task->paused = false; task->result = NAUT_ERR_AGAIN; task->state = check ? TORRENT_CHECKING : TORRENT_RUNNING; pthread_mutex_unlock(&task->lock); if (pthread_create(&task->thread, NULL, torrent_worker, task) != 0) { pthread_mutex_lock(&task->lock); task->state = TORRENT_ERROR; task->thread_done = true; task->checking = false; pthread_mutex_unlock(&task->lock); return false; } task->thread_started = true; return true; } static bool start_worker(torrent_task *task) { return start_worker_ex(task, false); } /* Ask a running worker to stop; it exits asynchronously (revisited next tick). */ static void request_stop(torrent_task *task) { pthread_mutex_lock(&task->lock); if (!task->thread_done) { task->stop_requested = true; task->state = TORRENT_STOPPING; } pthread_mutex_unlock(&task->lock); } typedef enum { ACT_NONE, ACT_START, ACT_STOP, ACT_SETSTATE, ACT_CHECK } lifecycle_act; /* Reconcile desired vs actual run-state for every torrent: enforce the * max-active download queue, honor pause/force-start, run recheck restarts, and * recompute each active torrent's share of the global download limit. Drives * worker start/stop to match. Main thread only (beside reap_torrents). */ static void service_lifecycle(daemon_state *state) { torrent_task *tasks[MAX_TORRENTS]; bool want_run[MAX_TORRENTS], complete[MAX_TORRENTS], forced[MAX_TORRENTS]; bool eligible[MAX_TORRENTS]; int qpos[MAX_TORRENTS]; lifecycle_act act[MAX_TORRENTS]; int target[MAX_TORRENTS]; pthread_mutex_lock(&state->torrent_lock); size_t n = state->torrent_count; for (size_t i = 0; i < n; i++) { torrent_task *t = state->torrents[i]; tasks[i] = t; pthread_mutex_lock(&t->lock); bool removed = t->remove_requested; complete[i] = (t->state == TORRENT_COMPLETE); forced[i] = t->force_start; qpos[i] = t->queue_pos; bool paused = t->paused; pthread_mutex_unlock(&t->lock); eligible[i] = !removed && !paused && !complete[i]; /* Completed torrents keep their keep-alive worker but never occupy an * active download slot. */ want_run[i] = (!removed && complete[i]); act[i] = ACT_NONE; target[i] = 0; if (removed) eligible[i] = false; /* reap owns removed tasks */ } /* Choose the active set: forced torrents always run; otherwise the * lowest-queue_pos eligible torrents up to max_active. */ size_t order[MAX_TORRENTS], ec = 0; for (size_t i = 0; i < n; i++) if (eligible[i]) order[ec++] = i; for (size_t a = 1; a < ec; a++) { /* insertion sort by queue_pos */ size_t v = order[a]; size_t b = a; while (b > 0 && qpos[order[b - 1]] > qpos[v]) { order[b] = order[b - 1]; b--; } order[b] = v; } uint32_t budget = state->max_active ? state->max_active : DEFAULT_MAX_ACTIVE; uint32_t chosen = 0; for (size_t k = 0; k < ec; k++) { size_t i = order[k]; if (forced[i]) { want_run[i] = true; } else if (chosen < budget) { want_run[i] = true; chosen++; } } /* Split the global download limit across torrents that will actually run. */ size_t active_dl = 0; for (size_t i = 0; i < n; i++) if (want_run[i] && !complete[i]) active_dl++; uint64_t eff = state->alt_speed_enabled ? state->alt_dl_limit : state->dl_limit; uint64_t share = eff == 0 ? 0 : eff / (active_dl ? active_dl : 1); for (size_t i = 0; i < n; i++) { torrent_task *t = tasks[i]; pthread_mutex_lock(&t->lock); t->rate_share = (want_run[i] && !complete[i]) ? share : 0; bool removed = t->remove_requested; bool started = t->thread_started, done = t->thread_done; bool stopping = t->stop_requested, restart = t->restart_requested; bool paused = t->paused, checking = t->checking; bool needs_check = t->needs_check; torrent_state st = t->state; pthread_mutex_unlock(&t->lock); if (removed) continue; bool running = started && !done; if (checking && running) { act[i] = ACT_NONE; /* let the one-shot hash check finish */ } else if (restart) { if (running && !stopping) act[i] = ACT_STOP; else if (done) act[i] = ACT_START; } else if (want_run[i]) { if (!running && done) act[i] = ACT_START; } else { if (running && !stopping) { act[i] = ACT_STOP; } else if (done && paused && needs_check) { act[i] = ACT_CHECK; /* verify a paused torrent's data */ } else if (done) { int want = paused ? TORRENT_PAUSED : TORRENT_QUEUED; if ((int)st != want) { act[i] = ACT_SETSTATE; target[i] = want; } } } } pthread_mutex_unlock(&state->torrent_lock); /* Apply outside torrent_lock (start_worker joins/creates threads). */ for (size_t i = 0; i < n; i++) { switch (act[i]) { case ACT_START: start_worker(tasks[i]); break; case ACT_CHECK: start_worker_ex(tasks[i], true); break; case ACT_STOP: request_stop(tasks[i]); break; case ACT_SETSTATE: pthread_mutex_lock(&tasks[i]->lock); tasks[i]->state = (torrent_state)target[i]; pthread_mutex_unlock(&tasks[i]->lock); break; case ACT_NONE: break; } } } static void stop_torrents(daemon_state *state) { pthread_mutex_lock(&state->torrent_lock); for (size_t i = 0; i < state->torrent_count; i++) { torrent_task *task = state->torrents[i]; pthread_mutex_lock(&task->lock); task->stop_requested = true; if (task->state == TORRENT_RUNNING) task->state = TORRENT_STOPPING; pthread_mutex_unlock(&task->lock); } pthread_mutex_unlock(&state->torrent_lock); } static void destroy_torrent(torrent_task *task) { if (task->thread_started) pthread_join(task->thread, NULL); /* Ephemeral /tmp uploads always go. Durable managed uploads are kept across * a normal shutdown (for restore) and removed only when the torrent was * explicitly removed. */ if (task->source && (task->source_is_temp || (task->source_managed && task->remove_requested))) unlink(task->source); for (size_t p = 0; p < task->num_peers; p++) free(task->peers[p]); free(task->peers); for (size_t i = 0; i < task->num_locations; i++) free(task->locations[i].path); free(task->locations); for (size_t i = 0; i < task->num_tags; i++) free(task->tags[i]); free(task->tags); for (size_t i = 0; i < task->num_files; i++) free(task->files[i]); free(task->files); free(task->category); free(task->pending_save_path); free(task->name); free(task->source); free(task->output_dir); free(task->dump_text); pthread_mutex_destroy(&task->lock); free(task); } static void reap_torrents(daemon_state *state) { bool reaped = false; for (;;) { torrent_task *task = NULL; pthread_mutex_lock(&state->torrent_lock); for (size_t i = 0; i < state->torrent_count; i++) { torrent_task *candidate = state->torrents[i]; pthread_mutex_lock(&candidate->lock); bool reap = candidate->remove_requested && candidate->thread_done; pthread_mutex_unlock(&candidate->lock); if (!reap) continue; task = candidate; state->torrents[i] = state->torrents[--state->torrent_count]; break; } pthread_mutex_unlock(&state->torrent_lock); if (!task) break; destroy_torrent(task); reaped = true; } if (reaped) persist_torrents(state); } static void destroy_torrents(daemon_state *state) { for (size_t i = 0; i < state->torrent_count; i++) destroy_torrent(state->torrents[i]); state->torrent_count = 0; } /* Recursively create a directory path (like `mkdir -p`). */ static int mkdir_p(const char *path, mode_t mode) { char tmp[PATH_MAX]; size_t len = snprintf(tmp, sizeof tmp, "%s", path); if (len == 0 || len >= sizeof tmp) return -1; if (tmp[len - 1] == '/') tmp[len - 1] = 0; for (char *p = tmp + 1; *p; p++) { if (*p != '/') continue; *p = 0; if (mkdir(tmp, mode) != 0 && errno != EEXIST) return -1; *p = '/'; } return (mkdir(tmp, mode) != 0 && errno != EEXIST) ? -1 : 0; } /* Resolve and prepare the persistence directory; fills state->state_file and * state->uploads_dir. Returns true if persistence can be used. */ static bool resolve_state_dir(daemon_state *state, const char *override) { char dir[PATH_MAX]; if (override && *override) { if ((size_t)snprintf(dir, sizeof dir, "%s", override) >= sizeof dir) return false; } else { const char *xdg = getenv("XDG_DATA_HOME"); const char *home = getenv("HOME"); int n; if (xdg && *xdg) n = snprintf(dir, sizeof dir, "%s/naut", xdg); else if (home && *home) n = snprintf(dir, sizeof dir, "%s/.local/share/naut", home); else return false; if (n < 0 || (size_t)n >= sizeof dir) return false; } if (mkdir_p(dir, 0700) != 0) { NAUT_WARN("state dir %s: %s", dir, strerror(errno)); return false; } if ((size_t)snprintf(state->uploads_dir, sizeof state->uploads_dir, "%s/uploads", dir) >= sizeof state->uploads_dir) return false; if (mkdir(state->uploads_dir, 0700) != 0 && errno != EEXIST) { NAUT_WARN("uploads dir %s: %s", state->uploads_dir, strerror(errno)); return false; } if ((size_t)snprintf(state->state_file, sizeof state->state_file, "%s/torrents.json", dir) >= sizeof state->state_file) return false; if ((size_t)snprintf(state->prefs_file, sizeof state->prefs_file, "%s/prefs.json", dir) >= sizeof state->prefs_file) return false; if ((size_t)snprintf(state->settings_file, sizeof state->settings_file, "%s/script_settings.json", dir) >= sizeof state->settings_file) return false; if ((size_t)snprintf(state->taxonomy_file, sizeof state->taxonomy_file, "%s/labels.json", dir) >= sizeof state->taxonomy_file) return false; if ((size_t)snprintf(state->data_dir, sizeof state->data_dir, "%s", dir) >= sizeof state->data_dir) return false; return true; } /* Re-create torrents recorded in state_file (called once at startup). */ static void restore_torrents(daemon_state *state) { if (!state->persist_enabled) return; json_error_t jerr; json_t *array = json_load_file(state->state_file, 0, &jerr); if (!array) return; /* no prior state, or unreadable */ if (!json_is_array(array)) { NAUT_WARN("state file %s is not a torrent array; ignoring", state->state_file); json_decref(array); return; } size_t restored = 0, dropped = 0, index; json_t *rec; json_array_foreach(array, index, rec) { const char *source = json_string_value(json_object_get(rec, "source")); const char *output = json_string_value(json_object_get(rec, "output")); if (!source || !output) { dropped++; continue; } bool managed = json_boolean_value(json_object_get(rec, "source_managed")); /* A path/upload source that no longer exists can't be restored; magnets * carry no file to check. */ if (strncmp(source, "magnet:", 7) != 0 && access(source, R_OK) != 0) { NAUT_WARN("restore: source missing, dropping: %s", source); dropped++; continue; } bool paused = json_boolean_value(json_object_get(rec, "paused")); bool force_start = json_boolean_value(json_object_get(rec, "force_start")); json_t *qp = json_object_get(rec, "queue_pos"); int queue_pos = json_is_integer(qp) ? (int)json_integer_value(qp) : -1; naut_err error = NAUT_OK; if (spawn_torrent(state, source, managed, false, output, json_object_get(rec, "peers"), json_object_get(rec, "locations"), /*meta_json=*/rec, json_object_get(rec, "torrent_id"), json_string_value(json_object_get(rec, "name")), paused, force_start, queue_pos, /*check_overlap=*/false, &error)) restored++; else { NAUT_WARN("restore: %s failed: %s", source, naut_strerror(error)); dropped++; } } json_decref(array); if (restored || dropped) NAUT_INFO("restore: %zu torrents restored, %zu dropped", restored, dropped); if (dropped) persist_torrents(state); /* prune dropped records */ } static void usage(const char *program) { fprintf(stderr, "usage: %s [--socket PATH] [--plugin PATH]... [--script PATH] " "[--state-dir PATH] [--max-active N]\n", program); } int main(int argc, char **argv) { const char *socket_path = DEFAULT_SOCKET; const char *script_path = NULL; const char *state_dir = NULL; long max_active_arg = 0; /* 0 => use default/persisted */ const char *plugin_paths[64]; size_t plugin_count = 0; for (int i = 1; i < argc; i++) { if (strcmp(argv[i], "--socket") == 0 && i + 1 < argc) socket_path = argv[++i]; else if (strcmp(argv[i], "--plugin") == 0 && i + 1 < argc && plugin_count < NAUT_ARRAY_LEN(plugin_paths)) plugin_paths[plugin_count++] = argv[++i]; else if (strcmp(argv[i], "--script") == 0 && i + 1 < argc) script_path = argv[++i]; else if (strcmp(argv[i], "--state-dir") == 0 && i + 1 < argc) state_dir = argv[++i]; else if (strcmp(argv[i], "--max-active") == 0 && i + 1 < argc) max_active_arg = strtol(argv[++i], NULL, 10); else { usage(argv[0]); return 2; } } signal(SIGINT, on_signal); signal(SIGTERM, on_signal); signal(SIGPIPE, SIG_IGN); daemon_state state = {0}; state.next_torrent_id = 1; state.max_active = DEFAULT_MAX_ACTIVE; state.persist_enabled = resolve_state_dir(&state, state_dir); if (!state.persist_enabled) NAUT_WARN("persistence disabled: torrents will not survive a restart"); load_prefs(&state); /* override defaults with any saved prefs */ if (max_active_arg > 0) state.max_active = (uint32_t)max_active_arg; pthread_mutex_init(&state.torrent_lock, NULL); pthread_mutex_init(&state.subscriber_lock, NULL); pthread_mutex_init(&state.script_lock, NULL); pthread_mutex_init(&state.settings_lock, NULL); load_script_settings(&state); /* user-set values; schema comes from the script */ pthread_mutex_init(&state.taxonomy_lock, NULL); load_taxonomy(&state); /* persisted category + tag lists for the web UI */ pthread_mutex_init(&state.blob_lock, NULL); state.events = naut_event_bus_create(); state.rpc = naut_rpc_registry_create(); state.plugins = naut_plugin_manager_create(state.rpc, state.events); if (!state.events || !state.rpc || !state.plugins || !register_commands(&state)) { fprintf(stderr, "nautd: failed to initialize control plane\n"); return 1; } if (script_path) { naut_err error = NAUT_OK; json_t *params = json_pack("{s:s}", "path", script_path); json_t *loaded = params ? rpc_load_script(&state, params, &error) : (error = NAUT_ERR_NOMEM, NULL); json_decref(params); json_decref(loaded); if (error != NAUT_OK) { fprintf(stderr, "nautd: failed to load script %s: %s\n", script_path, naut_strerror(error)); return 1; } } for (size_t i = 0; i < plugin_count; i++) { if (naut_plugin_load(state.plugins, plugin_paths[i]) != NAUT_OK) { fprintf(stderr, "nautd: failed to load plugin %s\n", plugin_paths[i]); return 1; } } uint64_t event_subscription; if (naut_event_subscribe(state.events, broadcast_event, &state, &event_subscription) != NAUT_OK) return 1; int listener = listen_unix(socket_path); if (listener < 0) { perror("nautd: listen"); return 1; } restore_torrents(&state); /* re-load torrents saved by a previous run */ NAUT_INFO("nautd listening on %s", socket_path); while (!state.stopping && !interrupted) { struct pollfd pollfd = {.fd = listener, .events = POLLIN}; int ready = poll(&pollfd, 1, 100); if (ready > 0 && (pollfd.revents & POLLIN)) { int client = accept(listener, NULL, NULL); if (client >= 0) handle_client(&state, client); } else if (ready < 0 && errno != EINTR) { break; } service_lifecycle(&state); /* enforce queue, pause/resume, throttle */ reap_torrents(&state); } close(listener); unlink(socket_path); /* Tear down plugins first: a plugin like webui runs its own threads that * call back into the daemon via RPC, so it must be stopped (and its * threads joined) before we free the torrent tasks those calls touch. */ naut_plugin_manager_destroy(state.plugins); state.plugins = NULL; pthread_mutex_lock(&state.script_lock); naut_script *script = state.script; char *loaded_script_path = state.script_path; state.script = NULL; state.script_path = NULL; pthread_mutex_unlock(&state.script_lock); naut_script_destroy(script); free(loaded_script_path); stop_torrents(&state); destroy_torrents(&state); naut_event_unsubscribe(state.events, event_subscription); pthread_mutex_lock(&state.subscriber_lock); for (size_t i = 0; i < state.subscriber_count; i++) close(state.subscribers[i]); pthread_mutex_unlock(&state.subscriber_lock); naut_rpc_registry_destroy(state.rpc); naut_event_bus_destroy(state.events); pthread_mutex_destroy(&state.subscriber_lock); pthread_mutex_destroy(&state.torrent_lock); pthread_mutex_destroy(&state.script_lock); json_decref(state.script_settings); json_decref(state.script_settings_schema); pthread_mutex_destroy(&state.settings_lock); json_decref(state.label_categories); json_decref(state.label_tags); pthread_mutex_destroy(&state.taxonomy_lock); pthread_mutex_destroy(&state.blob_lock); return 0; }