torq_core/
api.rs

1//! REST + SSE API for the daemon.
2//!
3//! Bound to 127.0.0.1 only; every route except the middleware passes requires
4//! `Authorization: Bearer <token>` (token lives in config.toml so local
5//! clients can read it).
6
7use std::convert::Infallible;
8use std::path::PathBuf;
9use std::sync::Arc;
10
11use axum::body::Body;
12use axum::extract::{Path, Query, State};
13use axum::http::{header, HeaderMap, HeaderValue, Request, StatusCode};
14use axum::middleware::Next;
15use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
16use axum::response::{IntoResponse, Response};
17use axum::routing::{delete, get, patch, post};
18use axum::{Json, Router};
19use futures::stream::{Stream, StreamExt};
20use librqbit::api::TorrentIdOrHash;
21use serde::{Deserialize, Serialize};
22use tokio::io::AsyncReadExt;
23use tokio_stream::wrappers::BroadcastStream;
24use tokio_util::io::ReaderStream;
25
26use crate::daemon::{Daemon, Event, TorrentView};
27use crate::VERSION;
28
29#[derive(Clone)]
30pub struct AppState {
31    pub daemon: Arc<Daemon>,
32    pub sources: Arc<torq_sources::Registry>,
33    pub client: reqwest::Client,
34    auth_token: String,
35}
36
37pub fn router(
38    daemon: Arc<Daemon>,
39    auth_token: String,
40    sources: Arc<torq_sources::Registry>,
41    client: reqwest::Client,
42) -> Router {
43    let state = Arc::new(AppState {
44        daemon,
45        sources,
46        client,
47        auth_token,
48    });
49    Router::new()
50        .route("/health", get(health))
51        .route("/torrents", get(list_torrents).post(add_torrent))
52        .route("/torrents/{id}", delete(remove_torrent))
53        .route("/torrents/{id}/pause", post(pause_torrent))
54        .route("/torrents/{id}/resume", post(resume_torrent))
55        .route("/torrents/{id}/files", get(torrent_files))
56        .route("/torrents/{id}/stream/{file_id}", get(stream_file))
57        .route("/search", get(search))
58        .route("/rss", get(list_rss).post(add_rss))
59        .route("/rss/{id}", delete(remove_rss))
60        .route("/library", get(library_status).post(library_scan))
61        .route("/config/limits", patch(set_limits))
62        .route("/events", get(events))
63        .route_layer(axum::middleware::from_fn_with_state(
64            state.clone(),
65            require_auth,
66        ))
67        .with_state(state)
68}
69
70async fn require_auth(
71    State(state): State<Arc<AppState>>,
72    req: Request<Body>,
73    next: Next,
74) -> Response {
75    let authed = req
76        .headers()
77        .get(header::AUTHORIZATION)
78        .and_then(|v| v.to_str().ok())
79        .and_then(|v| v.strip_prefix("Bearer "))
80        .map(|token| constant_time_eq(token.as_bytes(), state.auth_token.as_bytes()))
81        .unwrap_or(false);
82    if authed {
83        next.run(req).await
84    } else {
85        (StatusCode::UNAUTHORIZED, "missing or invalid bearer token").into_response()
86    }
87}
88
89fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
90    if a.len() != b.len() {
91        return false;
92    }
93    a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
94}
95
96// -- handlers ---------------------------------------------------------------
97
98#[derive(Serialize)]
99struct Health {
100    version: &'static str,
101    torrents: usize,
102}
103
104async fn health(State(state): State<Arc<AppState>>) -> Json<Health> {
105    Json(Health {
106        version: VERSION,
107        torrents: state.daemon.views().len(),
108    })
109}
110
111async fn list_torrents(State(state): State<Arc<AppState>>) -> Json<Vec<TorrentView>> {
112    Json(state.daemon.views())
113}
114
115#[derive(Serialize)]
116struct FileInfo {
117    id: usize,
118    name: String,
119    length: u64,
120    included: bool,
121}
122
123async fn torrent_files(
124    State(state): State<Arc<AppState>>,
125    Path(id): Path<String>,
126) -> Result<Json<Vec<FileInfo>>, ApiError> {
127    let details = state
128        .daemon
129        .engine()
130        .api()
131        .api_torrent_details(TorrentIdOrHash::parse(&id)?)?;
132    let files = details
133        .files
134        .unwrap_or_default()
135        .iter()
136        .enumerate()
137        .map(|(i, f)| FileInfo {
138            id: i,
139            name: f.components.join("/"),
140            length: f.length,
141            included: f.included,
142        })
143        .collect();
144    Ok(Json(files))
145}
146
147/// HTTP range streaming of a torrent file, works mid-download: librqbit's
148/// `FileStream` reads pieces on demand (32MB lookahead), so a player can start
149/// before the file completes — mpv/VLC over this endpoint are the target.
150async fn stream_file(
151    State(state): State<Arc<AppState>>,
152    Path((id, file_id)): Path<(String, usize)>,
153    headers: HeaderMap,
154) -> Result<Response, ApiError> {
155    let parsed = TorrentIdOrHash::parse(&id)?;
156    let api = state.daemon.engine().api();
157    let details = api.api_torrent_details(parsed)?;
158    let files = details.files.as_deref().unwrap_or_default();
159    let file = files
160        .get(file_id)
161        .ok_or_else(|| ApiError::NotFound(format!("file {file_id} not found")))?;
162    let total = file.length;
163
164    let mut stream = api.api_stream(parsed, file_id)?;
165    let range = headers
166        .get(header::RANGE)
167        .and_then(|v| v.to_str().ok())
168        .and_then(|r| parse_range(r, total));
169    let (status, start, end) = match range {
170        Some((s, e)) => (StatusCode::PARTIAL_CONTENT, s, e),
171        None => (StatusCode::OK, 0, total.saturating_sub(1)),
172    };
173    if start > 0 {
174        use tokio::io::AsyncSeekExt;
175        stream.seek(std::io::SeekFrom::Start(start)).await?;
176    }
177    let len = end - start + 1;
178
179    let mut headers = HeaderMap::new();
180    headers.insert("Accept-Ranges", HeaderValue::from_static("bytes"));
181    headers.insert(
182        header::CONTENT_TYPE,
183        HeaderValue::from_static(mime_for(&file.name)),
184    );
185    if status == StatusCode::PARTIAL_CONTENT {
186        headers.insert(
187            header::CONTENT_RANGE,
188            HeaderValue::from_str(&format!("bytes {start}-{end}/{total}")).expect("valid header"),
189        );
190    }
191    headers.insert(
192        header::CONTENT_LENGTH,
193        HeaderValue::from_str(&len.to_string()).expect("valid header"),
194    );
195
196    let body = Body::from_stream(ReaderStream::with_capacity(stream.take(len), 64 * 1024));
197    Ok((status, headers, body).into_response())
198}
199
200/// Parse a single-range `bytes=` header; returns (start, end), inclusive.
201fn parse_range(header: &str, total: u64) -> Option<(u64, u64)> {
202    if total == 0 {
203        return None;
204    }
205    let spec = header.strip_prefix("bytes=")?;
206    let (start_str, end_str) = spec.split_once('-')?;
207    if start_str.is_empty() {
208        // Suffix range: last N bytes.
209        let n = end_str.parse::<u64>().ok()?;
210        if n == 0 {
211            return None;
212        }
213        let start = total.saturating_sub(n);
214        return Some((start, total - 1));
215    }
216    let start = start_str.parse::<u64>().ok()?;
217    if start >= total {
218        return None;
219    }
220    let end = if end_str.is_empty() {
221        total - 1
222    } else {
223        end_str.parse::<u64>().ok()?.min(total - 1)
224    };
225    (end >= start).then_some((start, end))
226}
227
228fn mime_for(name: &str) -> &'static str {
229    match name
230        .rsplit('.')
231        .next()
232        .unwrap_or("")
233        .to_ascii_lowercase()
234        .as_str()
235    {
236        "mp4" | "m4v" => "video/mp4",
237        "mkv" => "video/x-matroska",
238        "webm" => "video/webm",
239        "avi" => "video/x-msvideo",
240        "mov" => "video/quicktime",
241        "ts" => "video/mp2t",
242        "wmv" => "video/x-ms-wmv",
243        "flv" => "video/x-flv",
244        "mp3" => "audio/mpeg",
245        "m4a" | "aac" => "audio/mp4",
246        "flac" => "audio/flac",
247        "ogg" | "opus" => "audio/ogg",
248        _ => "application/octet-stream",
249    }
250}
251
252#[derive(Deserialize)]
253struct SearchReq {
254    q: String,
255    /// Comma-separated source ids; empty = all.
256    #[serde(default)]
257    sources: Option<String>,
258}
259
260/// Aggregated search across enabled sources, deduped by infohash. Failing
261/// sources are reported in `offline`, never fatal.
262async fn search(
263    State(state): State<Arc<AppState>>,
264    Query(req): Query<SearchReq>,
265) -> Json<torq_sources::SearchReport> {
266    let only = req
267        .sources
268        .as_deref()
269        .map(|s| s.split(',').map(str::to_string).collect::<Vec<_>>());
270    let report = torq_sources::aggregate::search_all(
271        &state.sources.sources,
272        &state.client,
273        &req.q,
274        only.as_deref(),
275    )
276    .await;
277    Json(report)
278}
279
280#[derive(Deserialize)]
281struct AddReq {
282    magnet: String,
283    #[serde(default)]
284    paused: bool,
285}
286
287async fn add_torrent(
288    State(state): State<Arc<AppState>>,
289    Json(req): Json<AddReq>,
290) -> Result<Json<TorrentView>, ApiError> {
291    let view = state.daemon.add_magnet(&req.magnet, req.paused).await?;
292    Ok(Json(view))
293}
294
295#[derive(Deserialize, Default)]
296struct RemoveReq {
297    #[serde(default, deserialize_with = "deserialize_bool_flag")]
298    delete_files: bool,
299}
300
301/// serde_urlencoded only parses `true`/`false` for bools; scripts and curl
302/// users naturally write `?delete_files=1`, so accept both spellings.
303fn deserialize_bool_flag<'de, D>(d: D) -> Result<bool, D::Error>
304where
305    D: serde::Deserializer<'de>,
306{
307    let s = String::deserialize(d)?;
308    match s.as_str() {
309        "true" | "1" => Ok(true),
310        "false" | "0" => Ok(false),
311        other => Err(serde::de::Error::custom(format!(
312            "expected true/false/1/0, got {other:?}"
313        ))),
314    }
315}
316
317async fn remove_torrent(
318    State(state): State<Arc<AppState>>,
319    Path(id): Path<String>,
320    Query(req): Query<RemoveReq>,
321) -> Result<StatusCode, ApiError> {
322    let parsed = TorrentIdOrHash::parse(&id)?;
323    state.daemon.remove(parsed, req.delete_files).await?;
324    Ok(StatusCode::NO_CONTENT)
325}
326
327async fn pause_torrent(
328    State(state): State<Arc<AppState>>,
329    Path(id): Path<String>,
330) -> Result<StatusCode, ApiError> {
331    state.daemon.pause(TorrentIdOrHash::parse(&id)?).await?;
332    Ok(StatusCode::NO_CONTENT)
333}
334
335async fn resume_torrent(
336    State(state): State<Arc<AppState>>,
337    Path(id): Path<String>,
338) -> Result<StatusCode, ApiError> {
339    state.daemon.resume(TorrentIdOrHash::parse(&id)?).await?;
340    Ok(StatusCode::NO_CONTENT)
341}
342
343async fn list_rss(State(state): State<Arc<AppState>>) -> Json<Vec<crate::rss::Subscription>> {
344    Json(state.daemon.rss.list())
345}
346
347#[derive(Deserialize)]
348struct AddRssReq {
349    url: String,
350    #[serde(default)]
351    title_re: Option<String>,
352    #[serde(default)]
353    min_size: Option<u64>,
354    #[serde(default)]
355    max_size: Option<u64>,
356    #[serde(default = "default_sub_interval")]
357    interval_secs: u64,
358}
359
360fn default_sub_interval() -> u64 {
361    300
362}
363
364async fn add_rss(
365    State(state): State<Arc<AppState>>,
366    Json(req): Json<AddRssReq>,
367) -> Result<Json<crate::rss::Subscription>, ApiError> {
368    let sub = state.daemon.rss.add(
369        &req.url,
370        req.title_re,
371        req.min_size,
372        req.max_size,
373        req.interval_secs,
374    )?;
375    Ok(Json(sub))
376}
377
378async fn remove_rss(
379    State(state): State<Arc<AppState>>,
380    Path(id): Path<u64>,
381) -> Result<StatusCode, ApiError> {
382    if state.daemon.rss.remove(id) {
383        Ok(StatusCode::NO_CONTENT)
384    } else {
385        Err(ApiError::NotFound(format!("subscription {id} not found")))
386    }
387}
388
389#[derive(Serialize)]
390struct LibraryStatus {
391    indexed: usize,
392    dirs: Vec<PathBuf>,
393}
394
395async fn library_status(State(state): State<Arc<AppState>>) -> Json<LibraryStatus> {
396    Json(LibraryStatus {
397        indexed: state.daemon.library.count(),
398        dirs: state.daemon.library.dirs(),
399    })
400}
401
402/// Rescan library dirs; returns the number of torrents indexed.
403async fn library_scan(State(state): State<Arc<AppState>>) -> Result<Json<LibraryStatus>, ApiError> {
404    state.daemon.library.scan()?;
405    Ok(Json(LibraryStatus {
406        indexed: state.daemon.library.count(),
407        dirs: state.daemon.library.dirs(),
408    }))
409}
410
411#[derive(Deserialize)]
412struct LimitsReq {
413    upload_bps: Option<u32>,
414    download_bps: Option<u32>,
415}
416
417/// Apply rate limits live (None clears the limit). Persists to config so the
418/// daemon restarts with them.
419async fn set_limits(
420    State(state): State<Arc<AppState>>,
421    Json(req): Json<LimitsReq>,
422) -> Result<StatusCode, ApiError> {
423    state
424        .daemon
425        .engine()
426        .set_limits(req.upload_bps, req.download_bps);
427    let mut config = crate::config::Config::load()?;
428    config.upload_bps = req.upload_bps;
429    config.download_bps = req.download_bps;
430    config.save()?;
431    Ok(StatusCode::NO_CONTENT)
432}
433
434async fn events(
435    State(state): State<Arc<AppState>>,
436) -> Sse<impl Stream<Item = Result<SseEvent, Infallible>>> {
437    let stream = BroadcastStream::new(state.daemon.subscribe()).map(|item| {
438        let data = match item {
439            Ok(Event::TorrentFailed { id, error }) => {
440                serde_json::json!({"type": "torrent_failed", "id": id, "error": error}).to_string()
441            }
442            Ok(ev) => serde_json::to_string(&ev).unwrap_or_else(|_| "{}".into()),
443            Err(_) => "{}".into(), // lagged behind; client should re-poll /torrents
444        };
445        Ok(SseEvent::default().event("torq").data(data))
446    });
447    Sse::new(stream).keep_alive(KeepAlive::default())
448}
449
450// -- errors ------------------------------------------------------------------
451
452pub enum ApiError {
453    NotFound(String),
454    BadRequest(String),
455    Internal(anyhow::Error),
456}
457
458impl IntoResponse for ApiError {
459    fn into_response(self) -> Response {
460        match self {
461            Self::NotFound(m) => (StatusCode::NOT_FOUND, m).into_response(),
462            Self::BadRequest(m) => (StatusCode::BAD_REQUEST, m).into_response(),
463            Self::Internal(e) => {
464                tracing::error!("api error: {e:#}");
465                (StatusCode::INTERNAL_SERVER_ERROR, "internal error").into_response()
466            }
467        }
468    }
469}
470
471impl From<anyhow::Error> for ApiError {
472    fn from(e: anyhow::Error) -> Self {
473        let msg = e.to_string();
474        if msg.contains("not found") || msg.contains("not managed") {
475            Self::NotFound(msg)
476        } else if msg.contains("not a valid magnet") || msg.contains("failed to parse") {
477            Self::BadRequest(msg)
478        } else {
479            Self::Internal(e)
480        }
481    }
482}
483
484impl From<librqbit::ApiError> for ApiError {
485    fn from(e: librqbit::ApiError) -> Self {
486        let msg = e.to_string();
487        if msg.contains("not found") {
488            Self::NotFound(msg)
489        } else {
490            Self::Internal(anyhow::anyhow!(msg))
491        }
492    }
493}
494
495impl From<std::io::Error> for ApiError {
496    fn from(e: std::io::Error) -> Self {
497        Self::Internal(anyhow::anyhow!(e))
498    }
499}
torq_core/api.rs

torq_core/
api.rs
