feat: Phase 1.5 — health prober, Oracle's Memory, off-thread EPG, PWA

Resilience + scale pass on top of the v1 core.

Stream health prober:
- New SSRF-validated /probe endpoint (HEAD -> ranged-GET fallback, 6s timeout,
  5-min TTL cache, concurrency-capped, no-store) + orb_probe_total metrics
- Front-end health.js: lazy bounded-concurrency probing (current + sampled
  channels, not the whole catalog), LIVE/DEGRADED/DEAD scoring, guide skull
  glyph, instant no-signal on zap to a known-dead channel

Oracle's Memory (memory.js):
- Session resurrection (last channel, volume, mute, theater/dim/ambient)
- Exponential-decay per-channel uptime history; pinning; dead-channel auto-skip

Off-thread XMLTV:
- lib/xmltv-parse.js (pure, hand-rolled scanner, no DOMParser/regex backtracking)
  run in epg-worker.js module worker; wired into the iptv-org provider
- Unit-tested in node: entity decode, TZ offsets, 20k programmes off-thread

Service worker / PWA:
- manifest.webmanifest + orb icons; sw.js (cache-first shell + hls.js,
  stale-while-revalidate for JSON-over-relay, never caches streams)
- CSP gains worker-src + manifest-src; relay sends no-store on /relay + /probe
  so the SW is the sole metadata freshness authority; webmanifest MIME fix

Docs: README de-emojified (title only) and reframed as a peer parallel build.

Verified: SSRF probes blocked, legit probe + cache, new headers present,
PWA assets served, parser tests 8/8, JS syntax clean, zero Rust warnings.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

server/src/main.rs

@@ -7,12 +7,14 @@
 mod relay;
 mod ssrf;
 
+use std::collections::HashMap;
 use std::convert::Infallible;
+use std::hash::{Hash, Hasher};
 use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::Arc;
-use std::time::Instant;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
 
 use bytes::Bytes;
 use http_body_util::Full;
@@ -27,11 +29,13 @@ use tokio_rustls::TlsConnector;
 use url::Url;
 
 use relay::{
-    build_tls_connector, fetch_validated, looks_like_manifest, max_for, rewrite_manifest,
+    build_tls_connector, fetch_validated, looks_like_manifest, max_for, probe, rewrite_manifest,
     RelayError,
 };
 
 const MAX_INFLIGHT_RELAY: usize = 32;
+const PROBE_TIMEOUT: Duration = Duration::from_secs(6);
+const PROBE_TTL: Duration = Duration::from_secs(300);
 
 #[derive(Default)]
 struct Metrics {
@@ -40,13 +44,19 @@ struct Metrics {
     relay_upstream_fail: AtomicU64,
     relay_ratelimit: AtomicU64,
     relay_bad_request: AtomicU64,
+    probe_total: AtomicU64,
+    probe_dead: AtomicU64,
 }
 
+/// Cached probe result: (recorded_at, ok, status, latency_ms).
+type ProbeCacheEntry = (Instant, bool, u16, u64);
+
 struct AppState {
     web_root: PathBuf,
     tls: TlsConnector,
     metrics: Metrics,
     relay_gate: Semaphore,
+    probe_cache: Mutex<HashMap<u64, ProbeCacheEntry>>,
 }
 
 #[tokio::main]
@@ -64,6 +74,7 @@ async fn main() {
         tls: build_tls_connector(),
         metrics: Metrics::default(),
         relay_gate: Semaphore::new(MAX_INFLIGHT_RELAY),
+        probe_cache: Mutex::new(HashMap::new()),
     });
 
     let listener = TcpListener::bind(bind).await.expect("bind");
@@ -109,6 +120,7 @@ async fn handle(
     let path = req.uri().path().to_string();
     let resp = match path.as_str() {
         "/relay" => serve_relay(&state, &req).await,
+        "/probe" => serve_probe(&state, &req).await,
         "/metrics" => serve_metrics(&state),
         "/healthz" => text(StatusCode::OK, "ok"),
         _ => serve_static(&state, &path).await,
@@ -190,6 +202,58 @@ async fn serve_relay(state: &AppState, req: &Request<Incoming>) -> Response<Full
     }
 }
 
+// ----------------------------- /probe ------------------------------------
+
+async fn serve_probe(state: &AppState, req: &Request<Incoming>) -> Response<Full<Bytes>> {
+    let query = req.uri().query().unwrap_or("");
+    let raw = url::form_urlencoded::parse(query.as_bytes())
+        .find(|(k, _)| k == "url")
+        .map(|(_, v)| v.into_owned());
+    let Some(raw) = raw else {
+        return json(StatusCode::BAD_REQUEST, "{\"ok\":false,\"status\":0,\"latency_ms\":0}");
+    };
+    let parsed = match Url::parse(&raw) {
+        Ok(u) if u.scheme() == "https" => u,
+        _ => return json(StatusCode::BAD_REQUEST, "{\"ok\":false,\"status\":0,\"latency_ms\":0}"),
+    };
+
+    // hash the URL for the cache key (never store the raw URL — token hygiene)
+    let mut hasher = std::collections::hash_map::DefaultHasher::new();
+    raw.hash(&mut hasher);
+    let key = hasher.finish();
+
+    if let Ok(cache) = state.probe_cache.lock() {
+        if let Some((ts, ok, status, ms)) = cache.get(&key) {
+            if ts.elapsed() < PROBE_TTL {
+                return json(StatusCode::OK, &probe_json(*ok, *status, *ms));
+            }
+        }
+    }
+
+    let _permit = match state.relay_gate.try_acquire() {
+        Ok(p) => p,
+        Err(_) => {
+            state.metrics.relay_ratelimit.fetch_add(1, Ordering::Relaxed);
+            return json(StatusCode::SERVICE_UNAVAILABLE, "{\"ok\":false,\"status\":0,\"latency_ms\":0}");
+        }
+    };
+
+    let outcome = probe(&state.tls, &parsed, PROBE_TIMEOUT).await;
+    state.metrics.probe_total.fetch_add(1, Ordering::Relaxed);
+    if !outcome.ok {
+        state.metrics.probe_dead.fetch_add(1, Ordering::Relaxed);
+    }
+    if let Ok(mut cache) = state.probe_cache.lock() {
+        cache.insert(key, (Instant::now(), outcome.ok, outcome.status, outcome.latency_ms));
+    }
+    log_line("probe", if outcome.ok { "live" } else { "dead" }, parsed.host_str(), Some(outcome.latency_ms));
+    json(StatusCode::OK, &probe_json(outcome.ok, outcome.status, outcome.latency_ms))
+}
+
+fn probe_json(ok: bool, status: u16, ms: u64) -> String {
+    format!("{{\"ok\":{ok},\"status\":{status},\"latency_ms\":{ms}}}")
+}
+
 // ----------------------------- /metrics ----------------------------------
 
 fn serve_metrics(state: &AppState) -> Response<Full<Bytes>> {
@@ -201,12 +265,18 @@ fn serve_metrics(state: &AppState) -> Response<Full<Bytes>> {
          orb_relay_total{{event=\"ssrf_block\"}} {}\n\
          orb_relay_total{{event=\"upstream_fail\"}} {}\n\
          orb_relay_total{{event=\"ratelimit\"}} {}\n\
-         orb_relay_total{{event=\"bad_request\"}} {}\n",
+         orb_relay_total{{event=\"bad_request\"}} {}\n\
+         # HELP orb_probe_total Health probes performed.\n\
+         # TYPE orb_probe_total counter\n\
+         orb_probe_total{{result=\"all\"}} {}\n\
+         orb_probe_total{{result=\"dead\"}} {}\n",
         m.relay_ok.load(Ordering::Relaxed),
         m.relay_ssrf_block.load(Ordering::Relaxed),
         m.relay_upstream_fail.load(Ordering::Relaxed),
         m.relay_ratelimit.load(Ordering::Relaxed),
         m.relay_bad_request.load(Ordering::Relaxed),
+        m.probe_total.load(Ordering::Relaxed),
+        m.probe_dead.load(Ordering::Relaxed),
     );
     bytes_resp(StatusCode::OK, "text/plain; version=0.0.4", Bytes::from(body))
 }
@@ -250,6 +320,7 @@ fn content_type_for(p: &Path) -> &'static str {
         Some("js") | Some("mjs") => "text/javascript; charset=utf-8",
         Some("css") => "text/css; charset=utf-8",
         Some("json") => "application/json",
+        Some("webmanifest") => "application/manifest+json",
         Some("svg") => "image/svg+xml",
         Some("png") => "image/png",
         Some("jpg") | Some("jpeg") => "image/jpeg",
@@ -273,8 +344,9 @@ fn with_security_headers(mut resp: Response<Full<Bytes>>, path: &str) -> Respons
         "Content-Security-Policy",
         hyper::header::HeaderValue::from_static(
             "default-src 'none'; script-src 'self'; style-src 'self'; img-src 'self' https: data:; \
-             media-src 'self' blob:; connect-src 'self'; font-src 'self'; object-src 'none'; \
-             base-uri 'none'; frame-src 'none'; form-action 'none'; frame-ancestors 'none'",
+             media-src 'self' blob:; connect-src 'self'; font-src 'self'; worker-src 'self'; \
+             manifest-src 'self'; object-src 'none'; base-uri 'none'; frame-src 'none'; \
+             form-action 'none'; frame-ancestors 'none'",
         ),
     );
     h.insert("X-Content-Type-Options", hyper::header::HeaderValue::from_static("nosniff"));
@@ -289,7 +361,14 @@ fn with_security_headers(mut resp: Response<Full<Bytes>>, path: &str) -> Respons
         "Strict-Transport-Security",
         hyper::header::HeaderValue::from_static("max-age=31536000"),
     );
-    let _ = path;
+    // Live stream + probe traffic is never cached: the service worker is the sole
+    // freshness authority, and it only caches static shell + JSON metadata.
+    if path == "/relay" || path == "/probe" {
+        h.insert(
+            hyper::header::CACHE_CONTROL,
+            hyper::header::HeaderValue::from_static("no-store"),
+        );
+    }
     resp
 }
 
@@ -305,6 +384,10 @@ fn text(status: StatusCode, msg: &str) -> Response<Full<Bytes>> {
     bytes_resp(status, "text/plain; charset=utf-8", Bytes::from(msg.to_string()))
 }
 
+fn json(status: StatusCode, body: &str) -> Response<Full<Bytes>> {
+    bytes_resp(status, "application/json", Bytes::from(body.to_string()))
+}
+
 /// Structured one-line JSON log. We log host (not full URL) + outcome — query
 /// strings can carry stream tokens, so the raw URL is never recorded.
 fn log_line(event: &str, detail: &str, host: Option<&str>, latency_ms: Option<u64>) {

server/src/relay.rs

@@ -63,6 +63,13 @@ pub struct Upstream {
     pub redirects: u32,
 }
 
+/// Lightweight liveness result for the health prober.
+pub struct ProbeOutcome {
+    pub ok: bool,
+    pub status: u16,
+    pub latency_ms: u64,
+}
+
 /// Build the shared rustls/TLS connector (webpki roots, explicit ring provider).
 pub fn build_tls_connector() -> TlsConnector {
     let mut roots = rustls::RootCertStore::empty();
@@ -201,6 +208,94 @@ pub async fn fetch_validated(
     }
 }
 
+/// Open an SSRF-validated HTTPS/1.1 connection and send one request, returning
+/// the response status (body is not read — caller decides). Used by the prober.
+async fn send_method(
+    connector: &TlsConnector,
+    url: &Url,
+    method: hyper::Method,
+    range: bool,
+) -> Result<StatusCode, RelayError> {
+    if !is_https(url) {
+        return Err(RelayError::NotHttps);
+    }
+    let host = url.host_str().ok_or(RelayError::BadUrl)?.to_string();
+    let port = url.port_or_known_default().unwrap_or(443);
+    let addr = resolve_and_validate(&host, port)
+        .await
+        .map_err(RelayError::Ssrf)?;
+    let tcp = TcpStream::connect(addr)
+        .await
+        .map_err(|e| RelayError::Connect(e.to_string()))?;
+    let _ = tcp.set_nodelay(true);
+    let server_name = rustls::pki_types::ServerName::try_from(host.clone())
+        .map_err(|_| RelayError::Tls("invalid sni".into()))?;
+    let tls = connector
+        .connect(server_name, tcp)
+        .await
+        .map_err(|e| RelayError::Tls(e.to_string()))?;
+    let io = TokioIo::new(tls);
+    let (mut sender, conn) = http1::handshake(io)
+        .await
+        .map_err(|e| RelayError::Http(e.to_string()))?;
+    tokio::spawn(async move {
+        let _ = conn.await;
+    });
+
+    let mut pq = url.path().to_string();
+    if let Some(q) = url.query() {
+        pq.push('?');
+        pq.push_str(q);
+    }
+    let mut builder = Request::builder()
+        .method(method)
+        .uri(pq)
+        .header(hyper::header::HOST, host.as_str())
+        .header(hyper::header::USER_AGENT, UA)
+        .header(hyper::header::ACCEPT, "*/*");
+    if range {
+        builder = builder.header(hyper::header::RANGE, "bytes=0-1");
+    }
+    let req = builder
+        .body(Empty::<Bytes>::new())
+        .map_err(|e| RelayError::Http(e.to_string()))?;
+    let resp = sender
+        .send_request(req)
+        .await
+        .map_err(|e| RelayError::Http(e.to_string()))?;
+    Ok(resp.status())
+}
+
+/// Probe a stream URL for liveness: HEAD first, falling back to a ranged GET
+/// (many CDNs reject HEAD). Whole thing is time-boxed.
+pub async fn probe(connector: &TlsConnector, url: &Url, timeout: std::time::Duration) -> ProbeOutcome {
+    let started = std::time::Instant::now();
+    let result = tokio::time::timeout(timeout, async {
+        // HEAD
+        if let Ok(status) = send_method(connector, url, hyper::Method::HEAD, false).await {
+            if status.is_success() || status.is_redirection() {
+                return Some(status);
+            }
+        }
+        // ranged GET fallback
+        if let Ok(status) = send_method(connector, url, hyper::Method::GET, true).await {
+            return Some(status);
+        }
+        None
+    })
+    .await;
+
+    let latency_ms = started.elapsed().as_millis() as u64;
+    match result {
+        Ok(Some(status)) => ProbeOutcome {
+            ok: status.is_success() || status.is_redirection(),
+            status: status.as_u16(),
+            latency_ms,
+        },
+        Ok(None) | Err(_) => ProbeOutcome { ok: false, status: 0, latency_ms },
+    }
+}
+
 /// Heuristic: is this an HLS manifest we should rewrite?
 pub fn looks_like_manifest(url: &Url, content_type: &str) -> bool {
     let ct = content_type.to_ascii_lowercase();