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    api_port: u16,
35    auth_token: String,
36}
37
38pub fn router(
39    daemon: Arc<Daemon>,
40    auth_token: String,
41    sources: Arc<torq_sources::Registry>,
42    client: reqwest::Client,
43    api_port: u16,
44) -> Router {
45    let state = Arc::new(AppState {
46        daemon,
47        sources,
48        client,
49        api_port,
50        auth_token,
51    });
52    Router::new()
53        .route("/health", get(health))
54        .route("/torrents", get(list_torrents).post(add_torrent))
55        .route("/torrents/{id}", delete(remove_torrent))
56        .route("/torrents/{id}/pause", post(pause_torrent))
57        .route("/torrents/{id}/resume", post(resume_torrent))
58        .route("/torrents/{id}/files", get(torrent_files))
59        .route("/torrents/{id}/play", get(play_file))
60        .route("/torrents/{id}/stream/{file_id}", get(stream_file))
61        .route("/search", get(search))
62        .route("/rss", get(list_rss).post(add_rss))
63        .route("/rss/{id}", delete(remove_rss))
64        .route("/library", get(library_status).post(library_scan))
65        .route("/config/limits", patch(set_limits))
66        .route("/events", get(events))
67        .route_layer(axum::middleware::from_fn_with_state(
68            state.clone(),
69            require_auth,
70        ))
71        .with_state(state)
72}
73
74async fn require_auth(
75    State(state): State<Arc<AppState>>,
76    req: Request<Body>,
77    next: Next,
78) -> Response {
79    let authed = req
80        .headers()
81        .get(header::AUTHORIZATION)
82        .and_then(|v| v.to_str().ok())
83        .and_then(|v| v.strip_prefix("Bearer "))
84        .map(|token| constant_time_eq(token.as_bytes(), state.auth_token.as_bytes()))
85        .unwrap_or(false);
86    if authed {
87        next.run(req).await
88    } else {
89        (StatusCode::UNAUTHORIZED, "missing or invalid bearer token").into_response()
90    }
91}
92
93fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
94    if a.len() != b.len() {
95        return false;
96    }
97    a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
98}
99
100// -- handlers ---------------------------------------------------------------
101
102#[derive(Serialize)]
103struct Health {
104    version: &'static str,
105    torrents: usize,
106}
107
108async fn health(State(state): State<Arc<AppState>>) -> Json<Health> {
109    Json(Health {
110        version: VERSION,
111        torrents: state.daemon.views().len(),
112    })
113}
114
115async fn list_torrents(State(state): State<Arc<AppState>>) -> Json<Vec<TorrentView>> {
116    Json(state.daemon.views())
117}
118
119#[derive(Serialize, Debug, PartialEq)]
120struct FileInfo {
121    id: usize,
122    name: String,
123    length: u64,
124    included: bool,
125}
126
127fn file_list(details: &librqbit::api::TorrentDetailsResponse) -> Vec<FileInfo> {
128    details
129        .files
130        .as_deref()
131        .unwrap_or_default()
132        .iter()
133        .enumerate()
134        .map(|(i, f)| FileInfo {
135            id: i,
136            name: f.components.join("/"),
137            length: f.length,
138            included: f.included,
139        })
140        .collect()
141}
142
143async fn torrent_files(
144    State(state): State<Arc<AppState>>,
145    Path(id): Path<String>,
146) -> Result<Json<Vec<FileInfo>>, ApiError> {
147    let details = state
148        .daemon
149        .engine()
150        .api()
151        .api_torrent_details(TorrentIdOrHash::parse(&id)?)?;
152    Ok(Json(file_list(&details)))
153}
154
155const VIDEO_EXTS: &[&str] = &[
156    "mp4", "mkv", "webm", "avi", "mov", "m4v", "ts", "wmv", "flv",
157];
158
159/// Largest video file, else the largest file overall — what a player wants.
160fn pick_play_file(files: &[FileInfo]) -> Option<&FileInfo> {
161    let is_video = |f: &FileInfo| {
162        f.name
163            .rsplit('.')
164            .next()
165            .is_some_and(|e| VIDEO_EXTS.contains(&e.to_ascii_lowercase().as_str()))
166    };
167    files
168        .iter()
169        .filter(|f| is_video(f))
170        .max_by_key(|f| f.length)
171        .or_else(|| files.iter().max_by_key(|f| f.length))
172}
173
174#[derive(Serialize)]
175struct PlayResponse {
176    url: String,
177    name: String,
178    file_id: usize,
179    length: u64,
180}
181
182/// Resolve the playable stream URL for a torrent: the largest video file
183/// (fallback: largest file), served by the range endpoint. One implementation
184/// shared by `torq play` and the TUI's `P` key.
185async fn play_file(
186    State(state): State<Arc<AppState>>,
187    Path(id): Path<String>,
188) -> Result<Json<PlayResponse>, ApiError> {
189    let details = state
190        .daemon
191        .engine()
192        .api()
193        .api_torrent_details(TorrentIdOrHash::parse(&id)?)?;
194    let files = file_list(&details);
195    let file =
196        pick_play_file(&files).ok_or_else(|| ApiError::NotFound("torrent has no files".into()))?;
197    let url = format!(
198        "http://127.0.0.1:{}/torrents/{id}/stream/{}",
199        state.api_port, file.id
200    );
201    Ok(Json(PlayResponse {
202        url,
203        name: file.name.clone(),
204        file_id: file.id,
205        length: file.length,
206    }))
207}
208
209/// HTTP range streaming of a torrent file, works mid-download: librqbit's
210/// `FileStream` reads pieces on demand (32MB lookahead), so a player can start
211/// before the file completes — mpv/VLC over this endpoint are the target.
212async fn stream_file(
213    State(state): State<Arc<AppState>>,
214    Path((id, file_id)): Path<(String, usize)>,
215    headers: HeaderMap,
216) -> Result<Response, ApiError> {
217    let parsed = TorrentIdOrHash::parse(&id)?;
218    let api = state.daemon.engine().api();
219    let details = api.api_torrent_details(parsed)?;
220    let files = details.files.as_deref().unwrap_or_default();
221    let file = files
222        .get(file_id)
223        .ok_or_else(|| ApiError::NotFound(format!("file {file_id} not found")))?;
224    let total = file.length;
225
226    let mut stream = api.api_stream(parsed, file_id)?;
227    let range = headers
228        .get(header::RANGE)
229        .and_then(|v| v.to_str().ok())
230        .and_then(|r| parse_range(r, total));
231    let (status, start, end) = match range {
232        Some((s, e)) => (StatusCode::PARTIAL_CONTENT, s, e),
233        None => (StatusCode::OK, 0, total.saturating_sub(1)),
234    };
235    if start > 0 {
236        use tokio::io::AsyncSeekExt;
237        stream.seek(std::io::SeekFrom::Start(start)).await?;
238    }
239    let len = end - start + 1;
240
241    let mut headers = HeaderMap::new();
242    headers.insert("Accept-Ranges", HeaderValue::from_static("bytes"));
243    headers.insert(
244        header::CONTENT_TYPE,
245        HeaderValue::from_static(mime_for(&file.name)),
246    );
247    if status == StatusCode::PARTIAL_CONTENT {
248        headers.insert(
249            header::CONTENT_RANGE,
250            HeaderValue::from_str(&format!("bytes {start}-{end}/{total}")).expect("valid header"),
251        );
252    }
253    headers.insert(
254        header::CONTENT_LENGTH,
255        HeaderValue::from_str(&len.to_string()).expect("valid header"),
256    );
257
258    let body = Body::from_stream(ReaderStream::with_capacity(stream.take(len), 64 * 1024));
259    Ok((status, headers, body).into_response())
260}
261
262/// Parse a single-range `bytes=` header; returns (start, end), inclusive.
263fn parse_range(header: &str, total: u64) -> Option<(u64, u64)> {
264    if total == 0 {
265        return None;
266    }
267    let spec = header.strip_prefix("bytes=")?;
268    let (start_str, end_str) = spec.split_once('-')?;
269    if start_str.is_empty() {
270        // Suffix range: last N bytes.
271        let n = end_str.parse::<u64>().ok()?;
272        if n == 0 {
273            return None;
274        }
275        let start = total.saturating_sub(n);
276        return Some((start, total - 1));
277    }
278    let start = start_str.parse::<u64>().ok()?;
279    if start >= total {
280        return None;
281    }
282    let end = if end_str.is_empty() {
283        total - 1
284    } else {
285        end_str.parse::<u64>().ok()?.min(total - 1)
286    };
287    (end >= start).then_some((start, end))
288}
289
290fn mime_for(name: &str) -> &'static str {
291    match name
292        .rsplit('.')
293        .next()
294        .unwrap_or("")
295        .to_ascii_lowercase()
296        .as_str()
297    {
298        "mp4" | "m4v" => "video/mp4",
299        "mkv" => "video/x-matroska",
300        "webm" => "video/webm",
301        "avi" => "video/x-msvideo",
302        "mov" => "video/quicktime",
303        "ts" => "video/mp2t",
304        "wmv" => "video/x-ms-wmv",
305        "flv" => "video/x-flv",
306        "mp3" => "audio/mpeg",
307        "m4a" | "aac" => "audio/mp4",
308        "flac" => "audio/flac",
309        "ogg" | "opus" => "audio/ogg",
310        _ => "application/octet-stream",
311    }
312}
313
314#[derive(Deserialize)]
315struct SearchReq {
316    q: String,
317    /// Comma-separated source ids; empty = all.
318    #[serde(default)]
319    sources: Option<String>,
320}
321
322/// Aggregated search across enabled sources, deduped by infohash. Failing
323/// sources are reported in `offline`, never fatal.
324async fn search(
325    State(state): State<Arc<AppState>>,
326    Query(req): Query<SearchReq>,
327) -> Json<torq_sources::SearchReport> {
328    let only = req
329        .sources
330        .as_deref()
331        .map(|s| s.split(',').map(str::to_string).collect::<Vec<_>>());
332    let report = torq_sources::aggregate::search_all(
333        &state.sources.sources,
334        &state.client,
335        &req.q,
336        only.as_deref(),
337    )
338    .await;
339    Json(report)
340}
341
342#[derive(Deserialize)]
343struct AddReq {
344    #[serde(default)]
345    magnet: String,
346    #[serde(default)]
347    paused: bool,
348    /// Base64-encoded .torrent bytes (mutually exclusive with magnet).
349    #[serde(default)]
350    torrent_b64: Option<String>,
351}
352
353async fn add_torrent(
354    State(state): State<Arc<AppState>>,
355    Json(req): Json<AddReq>,
356) -> Result<Json<TorrentView>, ApiError> {
357    let view = match req.torrent_b64 {
358        Some(b64) => {
359            use base64::Engine;
360            let bytes = base64::engine::general_purpose::STANDARD
361                .decode(b64.trim())
362                .map_err(|e| ApiError::BadRequest(format!("invalid torrent_b64: {e}")))?;
363            state.daemon.add_torrent_bytes(bytes, req.paused).await?
364        }
365        None if req.magnet.trim().is_empty() => {
366            return Err(ApiError::BadRequest(
367                "provide a magnet or torrent_b64".into(),
368            ));
369        }
370        None => state.daemon.add_magnet(&req.magnet, req.paused).await?,
371    };
372    Ok(Json(view))
373}
374
375#[derive(Deserialize, Default)]
376struct RemoveReq {
377    #[serde(default, deserialize_with = "deserialize_bool_flag")]
378    delete_files: bool,
379}
380
381/// serde_urlencoded only parses `true`/`false` for bools; scripts and curl
382/// users naturally write `?delete_files=1`, so accept both spellings.
383fn deserialize_bool_flag<'de, D>(d: D) -> Result<bool, D::Error>
384where
385    D: serde::Deserializer<'de>,
386{
387    let s = String::deserialize(d)?;
388    match s.as_str() {
389        "true" | "1" => Ok(true),
390        "false" | "0" => Ok(false),
391        other => Err(serde::de::Error::custom(format!(
392            "expected true/false/1/0, got {other:?}"
393        ))),
394    }
395}
396
397async fn remove_torrent(
398    State(state): State<Arc<AppState>>,
399    Path(id): Path<String>,
400    Query(req): Query<RemoveReq>,
401) -> Result<StatusCode, ApiError> {
402    let parsed = TorrentIdOrHash::parse(&id)?;
403    state.daemon.remove(parsed, req.delete_files).await?;
404    Ok(StatusCode::NO_CONTENT)
405}
406
407async fn pause_torrent(
408    State(state): State<Arc<AppState>>,
409    Path(id): Path<String>,
410) -> Result<StatusCode, ApiError> {
411    state.daemon.pause(TorrentIdOrHash::parse(&id)?).await?;
412    Ok(StatusCode::NO_CONTENT)
413}
414
415async fn resume_torrent(
416    State(state): State<Arc<AppState>>,
417    Path(id): Path<String>,
418) -> Result<StatusCode, ApiError> {
419    state.daemon.resume(TorrentIdOrHash::parse(&id)?).await?;
420    Ok(StatusCode::NO_CONTENT)
421}
422
423async fn list_rss(State(state): State<Arc<AppState>>) -> Json<Vec<crate::rss::Subscription>> {
424    Json(state.daemon.rss.list())
425}
426
427#[derive(Deserialize)]
428struct AddRssReq {
429    url: String,
430    #[serde(default)]
431    title_re: Option<String>,
432    #[serde(default)]
433    min_size: Option<u64>,
434    #[serde(default)]
435    max_size: Option<u64>,
436    #[serde(default = "default_sub_interval")]
437    interval_secs: u64,
438}
439
440fn default_sub_interval() -> u64 {
441    300
442}
443
444async fn add_rss(
445    State(state): State<Arc<AppState>>,
446    Json(req): Json<AddRssReq>,
447) -> Result<Json<crate::rss::Subscription>, ApiError> {
448    let sub = state.daemon.rss.add(
449        &req.url,
450        req.title_re,
451        req.min_size,
452        req.max_size,
453        req.interval_secs,
454    )?;
455    Ok(Json(sub))
456}
457
458async fn remove_rss(
459    State(state): State<Arc<AppState>>,
460    Path(id): Path<u64>,
461) -> Result<StatusCode, ApiError> {
462    if state.daemon.rss.remove(id) {
463        Ok(StatusCode::NO_CONTENT)
464    } else {
465        Err(ApiError::NotFound(format!("subscription {id} not found")))
466    }
467}
468
469#[derive(Serialize)]
470struct LibraryStatus {
471    indexed: usize,
472    dirs: Vec<PathBuf>,
473}
474
475async fn library_status(State(state): State<Arc<AppState>>) -> Json<LibraryStatus> {
476    Json(LibraryStatus {
477        indexed: state.daemon.library.count(),
478        dirs: state.daemon.library.dirs(),
479    })
480}
481
482/// Rescan library dirs; returns the number of torrents indexed.
483async fn library_scan(State(state): State<Arc<AppState>>) -> Result<Json<LibraryStatus>, ApiError> {
484    state.daemon.library.scan()?;
485    Ok(Json(LibraryStatus {
486        indexed: state.daemon.library.count(),
487        dirs: state.daemon.library.dirs(),
488    }))
489}
490
491#[derive(Deserialize)]
492struct LimitsReq {
493    upload_bps: Option<u32>,
494    download_bps: Option<u32>,
495}
496
497/// Apply rate limits live (None clears the limit). Persists to config so the
498/// daemon restarts with them.
499async fn set_limits(
500    State(state): State<Arc<AppState>>,
501    Json(req): Json<LimitsReq>,
502) -> Result<StatusCode, ApiError> {
503    state
504        .daemon
505        .engine()
506        .set_limits(req.upload_bps, req.download_bps);
507    let mut config = crate::config::Config::load()?;
508    config.upload_bps = req.upload_bps;
509    config.download_bps = req.download_bps;
510    config.save()?;
511    Ok(StatusCode::NO_CONTENT)
512}
513
514async fn events(
515    State(state): State<Arc<AppState>>,
516) -> Sse<impl Stream<Item = Result<SseEvent, Infallible>>> {
517    let stream = BroadcastStream::new(state.daemon.subscribe()).map(|item| {
518        let data = match item {
519            Ok(Event::TorrentFailed { id, error }) => {
520                serde_json::json!({"type": "torrent_failed", "id": id, "error": error}).to_string()
521            }
522            Ok(ev) => serde_json::to_string(&ev).unwrap_or_else(|_| "{}".into()),
523            Err(_) => "{}".into(), // lagged behind; client should re-poll /torrents
524        };
525        Ok(SseEvent::default().event("torq").data(data))
526    });
527    Sse::new(stream).keep_alive(KeepAlive::default())
528}
529
530// -- errors ------------------------------------------------------------------
531
532pub enum ApiError {
533    NotFound(String),
534    BadRequest(String),
535    Internal(anyhow::Error),
536}
537
538impl IntoResponse for ApiError {
539    fn into_response(self) -> Response {
540        match self {
541            Self::NotFound(m) => (StatusCode::NOT_FOUND, m).into_response(),
542            Self::BadRequest(m) => (StatusCode::BAD_REQUEST, m).into_response(),
543            Self::Internal(e) => {
544                tracing::error!("api error: {e:#}");
545                (StatusCode::INTERNAL_SERVER_ERROR, "internal error").into_response()
546            }
547        }
548    }
549}
550
551impl From<anyhow::Error> for ApiError {
552    fn from(e: anyhow::Error) -> Self {
553        let msg = e.to_string();
554        if msg.contains("not found") || msg.contains("not managed") {
555            Self::NotFound(msg)
556        } else if msg.contains("not a valid magnet") || msg.contains("failed to parse") {
557            Self::BadRequest(msg)
558        } else {
559            Self::Internal(e)
560        }
561    }
562}
563
564impl From<librqbit::ApiError> for ApiError {
565    fn from(e: librqbit::ApiError) -> Self {
566        let msg = e.to_string();
567        if msg.contains("not found") {
568            Self::NotFound(msg)
569        } else {
570            Self::Internal(anyhow::anyhow!(msg))
571        }
572    }
573}
574
575impl From<std::io::Error> for ApiError {
576    fn from(e: std::io::Error) -> Self {
577        Self::Internal(anyhow::anyhow!(e))
578    }
579}
580
581#[cfg(test)]
582mod tests {
583    use super::*;
584
585    fn file(id: usize, name: &str, length: u64) -> FileInfo {
586        FileInfo {
587            id,
588            name: name.into(),
589            length,
590            included: true,
591        }
592    }
593
594    #[test]
595    fn play_picks_largest_video_else_largest_file() {
596        let files = vec![
597            file(0, "cover.jpg", 500_000),
598            file(1, "movie.mkv", 5_000_000_000),
599            file(2, "sample.mp4", 200_000_000),
600        ];
601        assert_eq!(pick_play_file(&files).unwrap().id, 1);
602        // No video: largest file wins.
603        let only_audio = vec![file(0, "song.mp3", 10_000_000), file(1, "notes.txt", 1)];
604        assert_eq!(pick_play_file(&only_audio).unwrap().id, 0);
605        assert_eq!(pick_play_file(&[]), None);
606    }
607}
torq_core/api.rs

torq_core/
api.rs
