torq_core/
api.rs

1//! REST + SSE API for the daemon.
2//!
3//! Bound to 127.0.0.1 only; routes require `Authorization: Bearer <token>`
4//! (token lives in config.toml so local clients can read it), except the
5//! stream route, which also accepts the short-lived capability token that
6//! `/play` embeds in the URL — real players can't send headers, so the URL
7//! itself is the ticket.
8
9use std::collections::HashMap;
10use std::convert::Infallible;
11use std::path::PathBuf;
12use std::sync::{Arc, Mutex};
13use std::time::{Duration, Instant};
14
15use axum::body::Body;
16use axum::extract::{Path, Query, State};
17use axum::http::{HeaderMap, HeaderValue, Request, StatusCode, header};
18use axum::middleware::Next;
19use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
20use axum::response::{IntoResponse, Response};
21use axum::routing::{delete, get, patch, post};
22use axum::{Json, Router};
23use futures::stream::{Stream, StreamExt};
24use librqbit::api::TorrentIdOrHash;
25use serde::{Deserialize, Serialize};
26use tokio::io::AsyncReadExt;
27use tokio_stream::wrappers::BroadcastStream;
28use tokio_util::io::ReaderStream;
29
30use crate::VERSION;
31use crate::daemon::{Daemon, Event, TorrentView};
32
33#[derive(Clone)]
34pub struct AppState {
35    pub daemon: Arc<Daemon>,
36    pub sources: Arc<torq_sources::Registry>,
37    pub client: reqwest::Client,
38    api_port: u16,
39    auth_token: String,
40    /// Capability tokens minted by `/play`, keyed to their issue time. The
41    /// stream route accepts one of these (in the URL) in place of the API
42    /// bearer header, since players can't send headers.
43    stream_tokens: Arc<Mutex<HashMap<String, Instant>>>,
44}
45
46/// How long a stream URL stays valid. Long enough for a player to pause and
47/// reconnect mid-session; `/play` sweeps expired entries as it mints new
48/// ones, so the map stays small.
49const STREAM_TOKEN_TTL: Duration = Duration::from_secs(60 * 60);
50
51pub fn router(
52    daemon: Arc<Daemon>,
53    auth_token: String,
54    sources: Arc<torq_sources::Registry>,
55    client: reqwest::Client,
56    api_port: u16,
57) -> Router {
58    let state = Arc::new(AppState {
59        daemon,
60        sources,
61        client,
62        api_port,
63        auth_token,
64        stream_tokens: Arc::new(Mutex::new(HashMap::new())),
65    });
66    Router::new()
67        .route("/health", get(health))
68        .route("/torrents", get(list_torrents).post(add_torrent))
69        .route("/torrents/{id}", delete(remove_torrent))
70        .route("/torrents/{id}/pause", post(pause_torrent))
71        .route("/torrents/{id}/resume", post(resume_torrent))
72        .route("/torrents/{id}/files", get(torrent_files))
73        .route("/torrents/{id}/play", get(play_file))
74        .route("/torrents/{id}/stream/{file_id}", get(stream_file))
75        .route("/search", get(search))
76        .route("/rss", get(list_rss).post(add_rss))
77        .route("/rss/{id}", delete(remove_rss))
78        .route("/library", get(library_status).post(library_scan))
79        .route("/config", get(get_config))
80        .route("/config/limits", patch(set_limits))
81        .route("/events", get(events))
82        .route_layer(axum::middleware::from_fn_with_state(
83            state.clone(),
84            require_auth,
85        ))
86        .with_state(state)
87}
88
89async fn require_auth(
90    State(state): State<Arc<AppState>>,
91    req: Request<Body>,
92    next: Next,
93) -> Response {
94    let header_ok = req
95        .headers()
96        .get(header::AUTHORIZATION)
97        .and_then(|v| v.to_str().ok())
98        .and_then(|v| v.strip_prefix("Bearer "))
99        .map(|token| constant_time_eq(token.as_bytes(), state.auth_token.as_bytes()))
100        .unwrap_or(false);
101    // The stream route is also reachable with the capability token `/play`
102    // embeds in the URL — players (VLC, IINA, mpv) can't send headers.
103    let token_ok = req
104        .uri()
105        .query()
106        .and_then(|q| {
107            q.split('&')
108                .find_map(|kv| kv.strip_prefix("token="))
109                .map(|t| stream_token_valid(&state.stream_tokens, t))
110        })
111        .unwrap_or(false);
112    if header_ok || token_ok {
113        next.run(req).await
114    } else {
115        (StatusCode::UNAUTHORIZED, "missing or invalid bearer token").into_response()
116    }
117}
118
119fn stream_token_valid(map: &Mutex<HashMap<String, Instant>>, token: &str) -> bool {
120    map.lock()
121        .ok()
122        .and_then(|m| m.get(token).copied())
123        .is_some_and(|issued| issued.elapsed() < STREAM_TOKEN_TTL)
124}
125
126/// Fresh capability token for a stream URL (16 random bytes, hex).
127fn new_stream_token() -> String {
128    let mut b = [0u8; 16];
129    getrandom::getrandom(&mut b).expect("os rng");
130    b.iter().map(|x| format!("{x:02x}")).collect()
131}
132
133fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
134    if a.len() != b.len() {
135        return false;
136    }
137    a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
138}
139
140// -- handlers ---------------------------------------------------------------
141
142#[derive(Serialize)]
143struct Health {
144    version: &'static str,
145    torrents: usize,
146}
147
148async fn health(State(state): State<Arc<AppState>>) -> Json<Health> {
149    Json(Health {
150        version: VERSION,
151        torrents: state.daemon.views().len(),
152    })
153}
154
155async fn list_torrents(State(state): State<Arc<AppState>>) -> Json<Vec<TorrentView>> {
156    Json(state.daemon.views())
157}
158
159#[derive(Serialize)]
160struct ConfigInfo {
161    /// Concurrent transfer slots; torrents beyond this wait in queue.
162    max_active: usize,
163    /// Active flat rate caps in bytes/sec (None = unlimited). The schedule
164    /// can override these at runtime, so this is best-effort for display.
165    upload_bps: Option<u32>,
166    download_bps: Option<u32>,
167}
168
169async fn get_config(State(state): State<Arc<AppState>>) -> Json<ConfigInfo> {
170    let cfg = crate::config::Config::load().unwrap_or_default();
171    Json(ConfigInfo {
172        max_active: state.daemon.max_active(),
173        upload_bps: cfg.upload_bps,
174        download_bps: cfg.download_bps,
175    })
176}
177
178#[derive(Serialize, Debug, PartialEq)]
179struct FileInfo {
180    id: usize,
181    name: String,
182    length: u64,
183    included: bool,
184}
185
186fn file_list(details: &librqbit::api::TorrentDetailsResponse) -> Vec<FileInfo> {
187    details
188        .files
189        .as_deref()
190        .unwrap_or_default()
191        .iter()
192        .enumerate()
193        .map(|(i, f)| FileInfo {
194            id: i,
195            name: f.components.join("/"),
196            length: f.length,
197            included: f.included,
198        })
199        .collect()
200}
201
202async fn torrent_files(
203    State(state): State<Arc<AppState>>,
204    Path(id): Path<String>,
205) -> Result<Json<Vec<FileInfo>>, ApiError> {
206    let details = state
207        .daemon
208        .engine()
209        .api()
210        .api_torrent_details(TorrentIdOrHash::parse(&id)?)?;
211    Ok(Json(file_list(&details)))
212}
213
214const VIDEO_EXTS: &[&str] = &[
215    "mp4", "mkv", "webm", "avi", "mov", "m4v", "ts", "wmv", "flv",
216];
217
218/// Largest video file, else the largest file overall — what a player wants.
219fn pick_play_file(files: &[FileInfo]) -> Option<&FileInfo> {
220    let is_video = |f: &FileInfo| {
221        f.name
222            .rsplit('.')
223            .next()
224            .is_some_and(|e| VIDEO_EXTS.contains(&e.to_ascii_lowercase().as_str()))
225    };
226    files
227        .iter()
228        .filter(|f| is_video(f))
229        .max_by_key(|f| f.length)
230        .or_else(|| files.iter().max_by_key(|f| f.length))
231}
232
233#[derive(Serialize)]
234struct PlayResponse {
235    url: String,
236    name: String,
237    file_id: usize,
238    length: u64,
239}
240
241/// Resolve the playable stream URL for a torrent: the largest video file
242/// (fallback: largest file), served by the range endpoint. One implementation
243/// shared by `torq play` and the TUI's `P` key.
244async fn play_file(
245    State(state): State<Arc<AppState>>,
246    Path(id): Path<String>,
247) -> Result<Json<PlayResponse>, ApiError> {
248    let details = state
249        .daemon
250        .engine()
251        .api()
252        .api_torrent_details(TorrentIdOrHash::parse(&id)?)?;
253    let files = file_list(&details);
254    let file =
255        pick_play_file(&files).ok_or_else(|| ApiError::NotFound("torrent has no files".into()))?;
256    let mut tokens = state.stream_tokens.lock().expect("stream tokens");
257    let now = Instant::now();
258    tokens.retain(|_, issued| now.duration_since(*issued) < STREAM_TOKEN_TTL);
259    let token = new_stream_token();
260    tokens.insert(token.clone(), now);
261    drop(tokens);
262    let url = format!(
263        "http://127.0.0.1:{}/torrents/{id}/stream/{}?token={token}",
264        state.api_port, file.id
265    );
266    Ok(Json(PlayResponse {
267        url,
268        name: file.name.clone(),
269        file_id: file.id,
270        length: file.length,
271    }))
272}
273
274/// HTTP range streaming of a torrent file, works mid-download: librqbit's
275/// `FileStream` reads pieces on demand (32MB lookahead), so a player can start
276/// before the file completes — mpv/VLC over this endpoint are the target.
277/// How long a ranged stream request may wait for the piece at its start
278/// offset before failing with 416. Long enough for a piece that is already
279/// mid-flight to land; short enough that VLC's MKV index seeks (which jump
280/// to the still-un-downloaded end of the file) fail fast instead of hanging
281/// and nuking the whole playback ("cannot seek / damaged file").
282const STREAM_PROBE_TIMEOUT: Duration = Duration::from_secs(3);
283
284async fn stream_file(
285    State(state): State<Arc<AppState>>,
286    Path((id, file_id)): Path<(String, usize)>,
287    headers: HeaderMap,
288) -> Result<Response, ApiError> {
289    let parsed = TorrentIdOrHash::parse(&id)?;
290    let api = state.daemon.engine().api();
291    let details = api.api_torrent_details(parsed)?;
292    let files = details.files.as_deref().unwrap_or_default();
293    let file = files
294        .get(file_id)
295        .ok_or_else(|| ApiError::NotFound(format!("file {file_id} not found")))?;
296    let total = file.length;
297
298    // Mid-download the pieces are fetched first/last-then-sequential, so the
299    // downloaded region is *not* contiguous: arbitrary seeks can stall on a
300    // missing piece. Until the file completes we don't advertise seekability
301    // (players then read linearly and wait, which is what makes MKV play),
302    // and ranged requests must start on an available piece or fail fast.
303    let stats = api.api_stats_v1(parsed)?;
304    let file_complete =
305        stats.file_progress.get(file_id).copied().unwrap_or(0) >= total;
306
307    let mut stream = api.api_stream(parsed, file_id)?;
308    let range = headers
309        .get(header::RANGE)
310        .and_then(|v| v.to_str().ok())
311        .and_then(|r| parse_range(r, total));
312
313    let mut headers = HeaderMap::new();
314    if file_complete {
315        headers.insert("Accept-Ranges", HeaderValue::from_static("bytes"));
316    }
317    headers.insert(
318        header::CONTENT_TYPE,
319        HeaderValue::from_static(mime_for(&file.name)),
320    );
321
322    let Some((start, end)) = range else {
323        let len = total;
324        headers.insert(
325            header::CONTENT_LENGTH,
326            HeaderValue::from_str(&len.to_string()).expect("valid header"),
327        );
328        let body = Body::from_stream(ReaderStream::with_capacity(
329            stream.take(len),
330            64 * 1024,
331        ));
332        return Ok((StatusCode::OK, headers, body).into_response());
333    };
334
335    use tokio::io::AsyncSeekExt;
336    stream.seek(std::io::SeekFrom::Start(start)).await?;
337    // Probe: the piece at `start` must be complete, else respond 416 now
338    // instead of leaving the client's seek hanging forever.
339    let mut probe = [0u8; 4096];
340    let n = match tokio::time::timeout(STREAM_PROBE_TIMEOUT, stream.read(&mut probe)).await {
341        Ok(Ok(n)) if n > 0 => n,
342        _ => {
343            let mut resp = Response::builder()
344                .status(StatusCode::RANGE_NOT_SATISFIABLE)
345                .header(
346                    header::CONTENT_RANGE,
347                    format!("bytes */{total}"),
348                )
349                .body(Body::empty())
350                .expect("valid response");
351            resp.headers_mut().insert(
352                "Accept-Ranges",
353                HeaderValue::from_static("bytes"),
354            );
355            return Ok(resp);
356        }
357    };
358
359    let len = end - start + 1;
360    headers.insert(
361        header::CONTENT_RANGE,
362        HeaderValue::from_str(&format!("bytes {start}-{end}/{total}")).expect("valid header"),
363    );
364    headers.insert(
365        header::CONTENT_LENGTH,
366        HeaderValue::from_str(&len.to_string()).expect("valid header"),
367    );
368
369    // The probed bytes head the body; the rest streams on as pieces land.
370    let head = futures::stream::once(async move {
371        Ok::<_, std::io::Error>(bytes::Bytes::copy_from_slice(&probe[..n]))
372    });
373    let rest = ReaderStream::with_capacity(
374        stream.take(len.saturating_sub(n as u64)),
375        64 * 1024,
376    );
377    let body = Body::from_stream(head.chain(rest));
378    Ok((StatusCode::PARTIAL_CONTENT, headers, body).into_response())
379}
380
381/// Parse a single-range `bytes=` header; returns (start, end), inclusive.
382fn parse_range(header: &str, total: u64) -> Option<(u64, u64)> {
383    if total == 0 {
384        return None;
385    }
386    let spec = header.strip_prefix("bytes=")?;
387    let (start_str, end_str) = spec.split_once('-')?;
388    if start_str.is_empty() {
389        // Suffix range: last N bytes.
390        let n = end_str.parse::<u64>().ok()?;
391        if n == 0 {
392            return None;
393        }
394        let start = total.saturating_sub(n);
395        return Some((start, total - 1));
396    }
397    let start = start_str.parse::<u64>().ok()?;
398    if start >= total {
399        return None;
400    }
401    let end = if end_str.is_empty() {
402        total - 1
403    } else {
404        end_str.parse::<u64>().ok()?.min(total - 1)
405    };
406    (end >= start).then_some((start, end))
407}
408
409fn mime_for(name: &str) -> &'static str {
410    match name
411        .rsplit('.')
412        .next()
413        .unwrap_or("")
414        .to_ascii_lowercase()
415        .as_str()
416    {
417        "mp4" | "m4v" => "video/mp4",
418        "mkv" => "video/x-matroska",
419        "webm" => "video/webm",
420        "avi" => "video/x-msvideo",
421        "mov" => "video/quicktime",
422        "ts" => "video/mp2t",
423        "wmv" => "video/x-ms-wmv",
424        "flv" => "video/x-flv",
425        "mp3" => "audio/mpeg",
426        "m4a" | "aac" => "audio/mp4",
427        "flac" => "audio/flac",
428        "ogg" | "opus" => "audio/ogg",
429        _ => "application/octet-stream",
430    }
431}
432
433#[derive(Deserialize)]
434struct SearchReq {
435    q: String,
436    /// Comma-separated source ids; empty = all.
437    #[serde(default)]
438    sources: Option<String>,
439}
440
441/// Aggregated search across enabled sources, deduped by infohash. Failing
442/// sources are reported in `offline`, never fatal.
443async fn search(
444    State(state): State<Arc<AppState>>,
445    Query(req): Query<SearchReq>,
446) -> Json<torq_sources::SearchReport> {
447    let only = req
448        .sources
449        .as_deref()
450        .map(|s| s.split(',').map(str::to_string).collect::<Vec<_>>());
451    let report = torq_sources::aggregate::search_all(
452        &state.sources.sources,
453        &state.client,
454        &req.q,
455        only.as_deref(),
456    )
457    .await;
458    Json(report)
459}
460
461#[derive(Deserialize)]
462struct AddReq {
463    #[serde(default)]
464    magnet: String,
465    #[serde(default)]
466    paused: bool,
467    /// Base64-encoded .torrent bytes (mutually exclusive with magnet).
468    #[serde(default)]
469    torrent_b64: Option<String>,
470}
471
472async fn add_torrent(
473    State(state): State<Arc<AppState>>,
474    Json(req): Json<AddReq>,
475) -> Result<Json<TorrentView>, ApiError> {
476    let view = match req.torrent_b64 {
477        Some(b64) => {
478            use base64::Engine;
479            let bytes = base64::engine::general_purpose::STANDARD
480                .decode(b64.trim())
481                .map_err(|e| ApiError::BadRequest(format!("invalid torrent_b64: {e}")))?;
482            state.daemon.add_torrent_bytes(bytes, req.paused).await?
483        }
484        None if req.magnet.trim().is_empty() => {
485            return Err(ApiError::BadRequest(
486                "provide a magnet or torrent_b64".into(),
487            ));
488        }
489        None => state.daemon.add_magnet(&req.magnet, req.paused).await?,
490    };
491    Ok(Json(view))
492}
493
494#[derive(Deserialize, Default)]
495struct RemoveReq {
496    #[serde(default, deserialize_with = "deserialize_bool_flag")]
497    delete_files: bool,
498}
499
500/// serde_urlencoded only parses `true`/`false` for bools; scripts and curl
501/// users naturally write `?delete_files=1`, so accept both spellings.
502fn deserialize_bool_flag<'de, D>(d: D) -> Result<bool, D::Error>
503where
504    D: serde::Deserializer<'de>,
505{
506    let s = String::deserialize(d)?;
507    match s.as_str() {
508        "true" | "1" => Ok(true),
509        "false" | "0" => Ok(false),
510        other => Err(serde::de::Error::custom(format!(
511            "expected true/false/1/0, got {other:?}"
512        ))),
513    }
514}
515
516async fn remove_torrent(
517    State(state): State<Arc<AppState>>,
518    Path(id): Path<String>,
519    Query(req): Query<RemoveReq>,
520) -> Result<StatusCode, ApiError> {
521    let parsed = TorrentIdOrHash::parse(&id)?;
522    state.daemon.remove(parsed, req.delete_files).await?;
523    Ok(StatusCode::NO_CONTENT)
524}
525
526async fn pause_torrent(
527    State(state): State<Arc<AppState>>,
528    Path(id): Path<String>,
529) -> Result<StatusCode, ApiError> {
530    state.daemon.pause(TorrentIdOrHash::parse(&id)?).await?;
531    Ok(StatusCode::NO_CONTENT)
532}
533
534async fn resume_torrent(
535    State(state): State<Arc<AppState>>,
536    Path(id): Path<String>,
537) -> Result<StatusCode, ApiError> {
538    state.daemon.resume(TorrentIdOrHash::parse(&id)?).await?;
539    Ok(StatusCode::NO_CONTENT)
540}
541
542async fn list_rss(State(state): State<Arc<AppState>>) -> Json<Vec<crate::rss::Subscription>> {
543    Json(state.daemon.rss.list())
544}
545
546#[derive(Deserialize)]
547struct AddRssReq {
548    url: String,
549    #[serde(default)]
550    title_re: Option<String>,
551    #[serde(default)]
552    min_size: Option<u64>,
553    #[serde(default)]
554    max_size: Option<u64>,
555    #[serde(default = "default_sub_interval")]
556    interval_secs: u64,
557}
558
559fn default_sub_interval() -> u64 {
560    300
561}
562
563async fn add_rss(
564    State(state): State<Arc<AppState>>,
565    Json(req): Json<AddRssReq>,
566) -> Result<Json<crate::rss::Subscription>, ApiError> {
567    let sub = state.daemon.rss.add(
568        &req.url,
569        req.title_re,
570        req.min_size,
571        req.max_size,
572        req.interval_secs,
573    )?;
574    Ok(Json(sub))
575}
576
577async fn remove_rss(
578    State(state): State<Arc<AppState>>,
579    Path(id): Path<u64>,
580) -> Result<StatusCode, ApiError> {
581    if state.daemon.rss.remove(id) {
582        Ok(StatusCode::NO_CONTENT)
583    } else {
584        Err(ApiError::NotFound(format!("subscription {id} not found")))
585    }
586}
587
588#[derive(Serialize)]
589struct LibraryStatus {
590    indexed: usize,
591    dirs: Vec<PathBuf>,
592}
593
594async fn library_status(State(state): State<Arc<AppState>>) -> Json<LibraryStatus> {
595    Json(LibraryStatus {
596        indexed: state.daemon.library.count(),
597        dirs: state.daemon.library.dirs(),
598    })
599}
600
601/// Rescan library dirs; returns the number of torrents indexed.
602async fn library_scan(State(state): State<Arc<AppState>>) -> Result<Json<LibraryStatus>, ApiError> {
603    state.daemon.library.scan()?;
604    Ok(Json(LibraryStatus {
605        indexed: state.daemon.library.count(),
606        dirs: state.daemon.library.dirs(),
607    }))
608}
609
610#[derive(Deserialize)]
611struct LimitsReq {
612    upload_bps: Option<u32>,
613    download_bps: Option<u32>,
614}
615
616/// Apply rate limits live (None clears the limit). Persists to config so the
617/// daemon restarts with them.
618async fn set_limits(
619    State(state): State<Arc<AppState>>,
620    Json(req): Json<LimitsReq>,
621) -> Result<StatusCode, ApiError> {
622    state
623        .daemon
624        .engine()
625        .set_limits(req.upload_bps, req.download_bps);
626    let mut config = crate::config::Config::load()?;
627    config.upload_bps = req.upload_bps;
628    config.download_bps = req.download_bps;
629    config.save()?;
630    Ok(StatusCode::NO_CONTENT)
631}
632
633async fn events(
634    State(state): State<Arc<AppState>>,
635) -> Sse<impl Stream<Item = Result<SseEvent, Infallible>>> {
636    let stream = BroadcastStream::new(state.daemon.subscribe()).map(|item| {
637        let data = match item {
638            Ok(Event::TorrentFailed { id, error }) => {
639                serde_json::json!({"type": "torrent_failed", "id": id, "error": error}).to_string()
640            }
641            Ok(ev) => serde_json::to_string(&ev).unwrap_or_else(|_| "{}".into()),
642            Err(_) => "{}".into(), // lagged behind; client should re-poll /torrents
643        };
644        Ok(SseEvent::default().event("torq").data(data))
645    });
646    Sse::new(stream).keep_alive(KeepAlive::default())
647}
648
649// -- errors ------------------------------------------------------------------
650
651pub enum ApiError {
652    NotFound(String),
653    BadRequest(String),
654    Internal(anyhow::Error),
655}
656
657impl IntoResponse for ApiError {
658    fn into_response(self) -> Response {
659        match self {
660            Self::NotFound(m) => (StatusCode::NOT_FOUND, m).into_response(),
661            Self::BadRequest(m) => (StatusCode::BAD_REQUEST, m).into_response(),
662            Self::Internal(e) => {
663                tracing::error!("api error: {e:#}");
664                (StatusCode::INTERNAL_SERVER_ERROR, "internal error").into_response()
665            }
666        }
667    }
668}
669
670impl From<anyhow::Error> for ApiError {
671    fn from(e: anyhow::Error) -> Self {
672        let msg = e.to_string();
673        if msg.contains("not found") || msg.contains("not managed") {
674            Self::NotFound(msg)
675        } else if msg.contains("not a valid magnet") || msg.contains("failed to parse") {
676            Self::BadRequest(msg)
677        } else {
678            Self::Internal(e)
679        }
680    }
681}
682
683impl From<librqbit::ApiError> for ApiError {
684    fn from(e: librqbit::ApiError) -> Self {
685        let msg = e.to_string();
686        if msg.contains("not found") {
687            Self::NotFound(msg)
688        } else {
689            Self::Internal(anyhow::anyhow!(msg))
690        }
691    }
692}
693
694impl From<std::io::Error> for ApiError {
695    fn from(e: std::io::Error) -> Self {
696        Self::Internal(anyhow::anyhow!(e))
697    }
698}
699
700#[cfg(test)]
701mod tests {
702    use super::*;
703
704    fn file(id: usize, name: &str, length: u64) -> FileInfo {
705        FileInfo {
706            id,
707            name: name.into(),
708            length,
709            included: true,
710        }
711    }
712
713    #[test]
714    fn stream_tokens_expire_and_reject_unknown() {
715        let map = Mutex::new(HashMap::new());
716        assert!(!stream_token_valid(&map, "nope"));
717        map.lock().unwrap().insert("fresh".into(), Instant::now());
718        assert!(stream_token_valid(&map, "fresh"));
719        map.lock()
720            .unwrap()
721            .insert("stale".into(), Instant::now() - STREAM_TOKEN_TTL - Duration::from_secs(1));
722        assert!(!stream_token_valid(&map, "stale"));
723        // A fresh token survives alongside the stale one.
724        assert!(stream_token_valid(&map, "fresh"));
725    }
726
727    #[test]
728    fn play_picks_largest_video_else_largest_file() {
729        let files = vec![
730            file(0, "cover.jpg", 500_000),
731            file(1, "movie.mkv", 5_000_000_000),
732            file(2, "sample.mp4", 200_000_000),
733        ];
734        assert_eq!(pick_play_file(&files).unwrap().id, 1);
735        // No video: largest file wins.
736        let only_audio = vec![file(0, "song.mp3", 10_000_000), file(1, "notes.txt", 1)];
737        assert_eq!(pick_play_file(&only_audio).unwrap().id, 0);
738        assert_eq!(pick_play_file(&[]), None);
739    }
740}
torq_core/api.rs

torq_core/
api.rs
