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

torq_core/
api.rs
