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::{HeaderMap, HeaderValue, Request, StatusCode, header};
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::VERSION;
27use crate::daemon::{Daemon, Event, TorrentView};
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    #[serde(default)]
283    magnet: String,
284    #[serde(default)]
285    paused: bool,
286    /// Base64-encoded .torrent bytes (mutually exclusive with magnet).
287    #[serde(default)]
288    torrent_b64: Option<String>,
289}
290
291async fn add_torrent(
292    State(state): State<Arc<AppState>>,
293    Json(req): Json<AddReq>,
294) -> Result<Json<TorrentView>, ApiError> {
295    let view = match req.torrent_b64 {
296        Some(b64) => {
297            use base64::Engine;
298            let bytes = base64::engine::general_purpose::STANDARD
299                .decode(b64.trim())
300                .map_err(|e| ApiError::BadRequest(format!("invalid torrent_b64: {e}")))?;
301            state.daemon.add_torrent_bytes(bytes, req.paused).await?
302        }
303        None if req.magnet.trim().is_empty() => {
304            return Err(ApiError::BadRequest(
305                "provide a magnet or torrent_b64".into(),
306            ));
307        }
308        None => state.daemon.add_magnet(&req.magnet, req.paused).await?,
309    };
310    Ok(Json(view))
311}
312
313#[derive(Deserialize, Default)]
314struct RemoveReq {
315    #[serde(default, deserialize_with = "deserialize_bool_flag")]
316    delete_files: bool,
317}
318
319/// serde_urlencoded only parses `true`/`false` for bools; scripts and curl
320/// users naturally write `?delete_files=1`, so accept both spellings.
321fn deserialize_bool_flag<'de, D>(d: D) -> Result<bool, D::Error>
322where
323    D: serde::Deserializer<'de>,
324{
325    let s = String::deserialize(d)?;
326    match s.as_str() {
327        "true" | "1" => Ok(true),
328        "false" | "0" => Ok(false),
329        other => Err(serde::de::Error::custom(format!(
330            "expected true/false/1/0, got {other:?}"
331        ))),
332    }
333}
334
335async fn remove_torrent(
336    State(state): State<Arc<AppState>>,
337    Path(id): Path<String>,
338    Query(req): Query<RemoveReq>,
339) -> Result<StatusCode, ApiError> {
340    let parsed = TorrentIdOrHash::parse(&id)?;
341    state.daemon.remove(parsed, req.delete_files).await?;
342    Ok(StatusCode::NO_CONTENT)
343}
344
345async fn pause_torrent(
346    State(state): State<Arc<AppState>>,
347    Path(id): Path<String>,
348) -> Result<StatusCode, ApiError> {
349    state.daemon.pause(TorrentIdOrHash::parse(&id)?).await?;
350    Ok(StatusCode::NO_CONTENT)
351}
352
353async fn resume_torrent(
354    State(state): State<Arc<AppState>>,
355    Path(id): Path<String>,
356) -> Result<StatusCode, ApiError> {
357    state.daemon.resume(TorrentIdOrHash::parse(&id)?).await?;
358    Ok(StatusCode::NO_CONTENT)
359}
360
361async fn list_rss(State(state): State<Arc<AppState>>) -> Json<Vec<crate::rss::Subscription>> {
362    Json(state.daemon.rss.list())
363}
364
365#[derive(Deserialize)]
366struct AddRssReq {
367    url: String,
368    #[serde(default)]
369    title_re: Option<String>,
370    #[serde(default)]
371    min_size: Option<u64>,
372    #[serde(default)]
373    max_size: Option<u64>,
374    #[serde(default = "default_sub_interval")]
375    interval_secs: u64,
376}
377
378fn default_sub_interval() -> u64 {
379    300
380}
381
382async fn add_rss(
383    State(state): State<Arc<AppState>>,
384    Json(req): Json<AddRssReq>,
385) -> Result<Json<crate::rss::Subscription>, ApiError> {
386    let sub = state.daemon.rss.add(
387        &req.url,
388        req.title_re,
389        req.min_size,
390        req.max_size,
391        req.interval_secs,
392    )?;
393    Ok(Json(sub))
394}
395
396async fn remove_rss(
397    State(state): State<Arc<AppState>>,
398    Path(id): Path<u64>,
399) -> Result<StatusCode, ApiError> {
400    if state.daemon.rss.remove(id) {
401        Ok(StatusCode::NO_CONTENT)
402    } else {
403        Err(ApiError::NotFound(format!("subscription {id} not found")))
404    }
405}
406
407#[derive(Serialize)]
408struct LibraryStatus {
409    indexed: usize,
410    dirs: Vec<PathBuf>,
411}
412
413async fn library_status(State(state): State<Arc<AppState>>) -> Json<LibraryStatus> {
414    Json(LibraryStatus {
415        indexed: state.daemon.library.count(),
416        dirs: state.daemon.library.dirs(),
417    })
418}
419
420/// Rescan library dirs; returns the number of torrents indexed.
421async fn library_scan(State(state): State<Arc<AppState>>) -> Result<Json<LibraryStatus>, ApiError> {
422    state.daemon.library.scan()?;
423    Ok(Json(LibraryStatus {
424        indexed: state.daemon.library.count(),
425        dirs: state.daemon.library.dirs(),
426    }))
427}
428
429#[derive(Deserialize)]
430struct LimitsReq {
431    upload_bps: Option<u32>,
432    download_bps: Option<u32>,
433}
434
435/// Apply rate limits live (None clears the limit). Persists to config so the
436/// daemon restarts with them.
437async fn set_limits(
438    State(state): State<Arc<AppState>>,
439    Json(req): Json<LimitsReq>,
440) -> Result<StatusCode, ApiError> {
441    state
442        .daemon
443        .engine()
444        .set_limits(req.upload_bps, req.download_bps);
445    let mut config = crate::config::Config::load()?;
446    config.upload_bps = req.upload_bps;
447    config.download_bps = req.download_bps;
448    config.save()?;
449    Ok(StatusCode::NO_CONTENT)
450}
451
452async fn events(
453    State(state): State<Arc<AppState>>,
454) -> Sse<impl Stream<Item = Result<SseEvent, Infallible>>> {
455    let stream = BroadcastStream::new(state.daemon.subscribe()).map(|item| {
456        let data = match item {
457            Ok(Event::TorrentFailed { id, error }) => {
458                serde_json::json!({"type": "torrent_failed", "id": id, "error": error}).to_string()
459            }
460            Ok(ev) => serde_json::to_string(&ev).unwrap_or_else(|_| "{}".into()),
461            Err(_) => "{}".into(), // lagged behind; client should re-poll /torrents
462        };
463        Ok(SseEvent::default().event("torq").data(data))
464    });
465    Sse::new(stream).keep_alive(KeepAlive::default())
466}
467
468// -- errors ------------------------------------------------------------------
469
470pub enum ApiError {
471    NotFound(String),
472    BadRequest(String),
473    Internal(anyhow::Error),
474}
475
476impl IntoResponse for ApiError {
477    fn into_response(self) -> Response {
478        match self {
479            Self::NotFound(m) => (StatusCode::NOT_FOUND, m).into_response(),
480            Self::BadRequest(m) => (StatusCode::BAD_REQUEST, m).into_response(),
481            Self::Internal(e) => {
482                tracing::error!("api error: {e:#}");
483                (StatusCode::INTERNAL_SERVER_ERROR, "internal error").into_response()
484            }
485        }
486    }
487}
488
489impl From<anyhow::Error> for ApiError {
490    fn from(e: anyhow::Error) -> Self {
491        let msg = e.to_string();
492        if msg.contains("not found") || msg.contains("not managed") {
493            Self::NotFound(msg)
494        } else if msg.contains("not a valid magnet") || msg.contains("failed to parse") {
495            Self::BadRequest(msg)
496        } else {
497            Self::Internal(e)
498        }
499    }
500}
501
502impl From<librqbit::ApiError> for ApiError {
503    fn from(e: librqbit::ApiError) -> Self {
504        let msg = e.to_string();
505        if msg.contains("not found") {
506            Self::NotFound(msg)
507        } else {
508            Self::Internal(anyhow::anyhow!(msg))
509        }
510    }
511}
512
513impl From<std::io::Error> for ApiError {
514    fn from(e: std::io::Error) -> Self {
515        Self::Internal(anyhow::anyhow!(e))
516    }
517}
torq_core/api.rs

torq_core/
api.rs
