sheathe_package/
origin.rs

1//! JIT / origin mode — package on HTTP request (Phase 5).
2//!
3//! A minimal pure-`std` HTTP/1.1 origin that serves:
4//! - `GET /health` → `ok`
5//! - `GET /package?input=<path>&…` → runs [`crate::package`] into a temp dir and
6//!   returns `master.m3u8` or `manifest.mpd` based on `Accept` / `format=`
7//! - `GET /out/<rel>` → serves a previously packaged object from the origin cache
8//!
9//! Not a production CDN; it demonstrates JIT packaging without external deps.
10
11use crate::{PackageOptions, PresentationMode, package};
12use anyhow::{Context, Result, bail};
13use std::collections::HashMap;
14use std::fs;
15use std::io::{Read, Write};
16use std::net::{TcpListener, TcpStream};
17use std::path::{Path, PathBuf};
18use std::sync::{Arc, Mutex};
19use std::thread;
20
21/// Configuration for [`serve`].
22#[derive(Debug, Clone)]
23pub struct OriginConfig {
24    /// Bind address, e.g. `127.0.0.1:8787`.
25    pub bind: String,
26    /// Directory holding source media (path sandbox for `input=`).
27    pub media_root: PathBuf,
28    /// Cache directory for packaged outputs.
29    pub cache_dir: PathBuf,
30    /// Default segment duration seconds.
31    pub segment_duration: f64,
32}
33
34impl Default for OriginConfig {
35    fn default() -> Self {
36        Self {
37            bind: "127.0.0.1:8787".into(),
38            media_root: PathBuf::from("."),
39            cache_dir: PathBuf::from("/tmp/sheathe-origin"),
40            segment_duration: 6.0,
41        }
42    }
43}
44
45/// Run the origin until the process is killed. Spawns one thread per connection.
46pub fn serve(cfg: OriginConfig) -> Result<()> {
47    fs::create_dir_all(&cfg.cache_dir)?;
48    let listener = TcpListener::bind(&cfg.bind).with_context(|| format!("bind {}", cfg.bind))?;
49    eprintln!("sheathe origin listening on http://{}/", cfg.bind);
50    let cfg = Arc::new(cfg);
51    let lock = Arc::new(Mutex::new(()));
52    for conn in listener.incoming() {
53        let stream = conn.context("accept")?;
54        let cfg = Arc::clone(&cfg);
55        let lock = Arc::clone(&lock);
56        thread::spawn(move || {
57            if let Err(e) = handle_client(stream, &cfg, &lock) {
58                eprintln!("origin: {e:#}");
59            }
60        });
61    }
62    Ok(())
63}
64
65fn handle_client(mut stream: TcpStream, cfg: &OriginConfig, lock: &Mutex<()>) -> Result<()> {
66    let mut buf = [0u8; 8192];
67    let n = stream.read(&mut buf)?;
68    let req = String::from_utf8_lossy(&buf[..n]);
69    let line = req.lines().next().unwrap_or("");
70    let mut parts = line.split_whitespace();
71    let method = parts.next().unwrap_or("");
72    let target = parts.next().unwrap_or("/");
73    if method != "GET" && method != "HEAD" {
74        return respond(&mut stream, 405, "text/plain", b"method not allowed");
75    }
76
77    let (path, query) = target.split_once('?').unwrap_or((target, ""));
78    let q = parse_query(query);
79
80    match path {
81        "/health" | "/healthz" => respond(&mut stream, 200, "text/plain", b"ok\n"),
82        "/package" => {
83            let input = q.get("input").map(String::as_str).context("missing input=")?;
84            let media = resolve_media(&cfg.media_root, input)?;
85            let format = q.get("format").map(String::as_str).unwrap_or("hls");
86            let key = cache_key(&media, format, cfg.segment_duration);
87            let out_dir = cfg.cache_dir.join(&key);
88            {
89                let _g = lock.lock().unwrap_or_else(|e| e.into_inner());
90                if !out_dir.join("master.m3u8").exists() && !out_dir.join("manifest.mpd").exists() {
91                    let opts = PackageOptions {
92                        out_dir: out_dir.clone(),
93                        segment_duration: cfg.segment_duration,
94                        dash: format == "dash" || format == "both",
95                        hls: format == "hls" || format == "both",
96                        presentation: PresentationMode::Vod,
97                        ..PackageOptions::default()
98                    };
99                    package(&[media], &opts)?;
100                }
101            }
102            let body_path = if format == "dash" {
103                out_dir.join("manifest.mpd")
104            } else {
105                out_dir.join("master.m3u8")
106            };
107            let body =
108                fs::read(&body_path).with_context(|| format!("read {}", body_path.display()))?;
109            let ctype = if format == "dash" {
110                "application/dash+xml"
111            } else {
112                "application/vnd.apple.mpegurl"
113            };
114            respond(&mut stream, 200, ctype, &body)
115        }
116        p if p.starts_with("/out/") => {
117            let rel = &p["/out/".len()..];
118            let path = cfg.cache_dir.join(rel);
119            // Prevent path escape.
120            let canon_cache =
121                cfg.cache_dir.canonicalize().unwrap_or_else(|_| cfg.cache_dir.clone());
122            let canon = path.canonicalize().with_context(|| format!("missing {rel}"))?;
123            if !canon.starts_with(&canon_cache) {
124                bail!("path escape");
125            }
126            let body = fs::read(&canon)?;
127            respond(&mut stream, 200, guess_ctype(rel), &body)
128        }
129        _ => respond(
130            &mut stream,
131            404,
132            "text/plain",
133            b"not found\nGET /health | /package?input=FILE&format=hls|dash | /out/<cache-rel>\n",
134        ),
135    }
136}
137
138fn resolve_media(root: &Path, input: &str) -> Result<PathBuf> {
139    let p = Path::new(input);
140    let full = if p.is_absolute() { p.to_path_buf() } else { root.join(p) };
141    let canon = full.canonicalize().with_context(|| format!("input not found: {input}"))?;
142    let root_canon = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
143    if !canon.starts_with(&root_canon) && !Path::new(input).is_absolute() {
144        bail!("input escapes media_root");
145    }
146    Ok(canon)
147}
148
149fn cache_key(media: &Path, format: &str, seg: f64) -> String {
150    use std::collections::hash_map::DefaultHasher;
151    use std::hash::{Hash, Hasher};
152    let mut h = DefaultHasher::new();
153    media.hash(&mut h);
154    format.hash(&mut h);
155    seg.to_bits().hash(&mut h);
156    format!("{:016x}", h.finish())
157}
158
159fn parse_query(q: &str) -> HashMap<String, String> {
160    let mut m = HashMap::new();
161    for pair in q.split('&').filter(|s| !s.is_empty()) {
162        if let Some((k, v)) = pair.split_once('=') {
163            m.insert(url_decode(k), url_decode(v));
164        }
165    }
166    m
167}
168
169fn url_decode(s: &str) -> String {
170    let mut out = String::new();
171    let b = s.as_bytes();
172    let mut i = 0;
173    while i < b.len() {
174        match b[i] {
175            b'+' => {
176                out.push(' ');
177                i += 1;
178            }
179            b'%' if i + 2 < b.len() => {
180                let hex = &s[i + 1..i + 3];
181                if let Ok(v) = u8::from_str_radix(hex, 16) {
182                    out.push(v as char);
183                    i += 3;
184                } else {
185                    out.push('%');
186                    i += 1;
187                }
188            }
189            c => {
190                out.push(c as char);
191                i += 1;
192            }
193        }
194    }
195    out
196}
197
198fn guess_ctype(path: &str) -> &'static str {
199    if path.ends_with(".m3u8") {
200        "application/vnd.apple.mpegurl"
201    } else if path.ends_with(".mpd") {
202        "application/dash+xml"
203    } else if path.ends_with(".mp4") || path.ends_with(".m4s") {
204        "video/mp4"
205    } else if path.ends_with(".ts") {
206        "video/mp2t"
207    } else {
208        "application/octet-stream"
209    }
210}
211
212fn respond(stream: &mut TcpStream, status: u16, ctype: &str, body: &[u8]) -> Result<()> {
213    let reason = match status {
214        200 => "OK",
215        404 => "Not Found",
216        405 => "Method Not Allowed",
217        _ => "Error",
218    };
219    let header = format!(
220        "HTTP/1.1 {status} {reason}\r\nContent-Type: {ctype}\r\nContent-Length: {}\r\nConnection: close\r\nAccess-Control-Allow-Origin: *\r\n\r\n",
221        body.len()
222    );
223    stream.write_all(header.as_bytes())?;
224    stream.write_all(body)?;
225    Ok(())
226}
sheathe_package/origin.rs

sheathe_package/
origin.rs
