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

torq_core/
api.rs
