Skip to main content

bramley_engine/
runtime.rs

1use std::cell::RefCell;
2use std::collections::{BTreeMap, HashMap, HashSet};
3use std::rc::Rc;
4use std::sync::mpsc::{Receiver, Sender, channel};
5use std::time::{Duration, Instant};
6
7use anyhow::{Result, anyhow};
8use rquickjs::{
9    ArrayBuffer, CatchResultExt, Context, Ctx, Function, Object, Runtime, Value,
10    context::EvalOptions, function::IntoArgs,
11};
12use serde_json::{Value as J, json};
13
14use crate::EngineConfig;
15use crate::audio::{AudioSink, PulseSink};
16use crate::buffer::{MediaSourceNative, SOURCE_BUFFER_QUOTA_BYTES, SourceBufferNative};
17use crate::cdm::{CdmBackend, CdmEvent, Widevine};
18use crate::eme::{audio_type_supported, can_play_type, eme_check};
19use crate::media::{MediaEngine, SinkFactory};
20use crate::mp4;
21use crate::net::{self, Completion, HttpRequest};
22
23const PLATFORM: &str = include_str!("../js/platform.js");
24
25pub type CdmFactory = Box<dyn Fn(bool) -> Result<Box<dyn CdmBackend>>>;
26
27struct Native {
28    start: Instant,
29    next_id: u32,
30    mses: HashMap<u32, MediaSourceNative>,
31    sbs: HashMap<u32, SourceBufferNative>,
32    medias: HashMap<u32, MediaEngine>,
33    cdms: HashMap<u32, Box<dyn CdmBackend>>,
34    cdm_factory: CdmFactory,
35    sink_factory: SinkFactory,
36    cdm_error: Option<String>,
37    net_tx: Sender<Completion>,
38    aborted: HashSet<u32>,
39    op_counts: BTreeMap<&'static str, u64>,
40    eme_requests: Vec<J>,
41    script_log: Vec<J>,
42    http_count: u64,
43    http_bytes: u64,
44    redact: Vec<String>,
45    truncate_scripts: bool,
46    retired_output: OutputCounts,
47    in_flight: usize,
48}
49
50impl Native {
51    fn id(&mut self) -> u32 {
52        self.next_id += 1;
53        self.next_id
54    }
55    fn count(&mut self, op: &'static str) {
56        *self.op_counts.entry(op).or_default() += 1;
57    }
58    fn primary_media(&self) -> Option<&MediaEngine> {
59        self.medias
60            .iter()
61            .filter(|(_, m)| m.keys_id.is_some())
62            .min_by_key(|(id, _)| **id)
63            .map(|(_, m)| m)
64    }
65    fn position_for_buffer(&self, sb_id: u32) -> i64 {
66        self.medias
67            .values()
68            .find(|m| {
69                m.ms_id
70                    .and_then(|id| self.mses.get(&id))
71                    .is_some_and(|ms| ms.sbs.contains(&sb_id))
72            })
73            .map(|m| m.current_time_us())
74            .unwrap_or(0)
75    }
76}
77
78type N = Rc<RefCell<Native>>;
79
80pub struct Engine {
81    ctx: Context,
82    rt: Runtime,
83    n: N,
84    net_rx: Receiver<Completion>,
85    last_gc: Instant,
86}
87
88const GC_INTERVAL: Duration = Duration::from_secs(2);
89const ACTIVE_WAIT_MS: f64 = 5.0;
90const IDLE_WAIT: Duration = Duration::from_secs(60);
91
92fn opts(name: &str) -> EvalOptions {
93    let mut o = EvalOptions::default();
94    o.strict = false;
95    o.global = true;
96    o.filename = Some(name.into());
97    o
98}
99
100fn bytes_of(v: &Value) -> Option<Vec<u8>> {
101    let ab = ArrayBuffer::from_value(v.clone())?;
102    unsafe { ab.as_bytes() }.map(|b| b.to_vec())
103}
104
105fn media_state(m: &MediaEngine) -> String {
106    json!({"currentTime": m.current_time(), "readyState": m.ready_state, "paused": m.paused, "seeking": m.seeking, "ended": m.ended}).to_string()
107}
108
109fn install_ops<'js>(ctx: &Ctx<'js>, n: &N) -> rquickjs::Result<()> {
110    let g = ctx.globals();
111    macro_rules! op {
112        ($name:literal, $f:expr) => {
113            g.set($name, Function::new(ctx.clone(), $f)?)?;
114        };
115    }
116    macro_rules! with_media {
117        ($name:literal, $count:literal, |$m:ident $(, $arg:ident : $ty:ty)*| $body:expr) => {
118            let s = n.clone();
119            op!($name, move |id: u32 $(, $arg: $ty)*| {
120                let mut st = s.borrow_mut();
121                st.count($count);
122                if let Some($m) = st.medias.get_mut(&id) {
123                    $body;
124                }
125            });
126        };
127    }
128    macro_rules! with_cdm {
129        ($name:literal, $count:literal, |$c:ident, $p:ident $(, $arg:ident : $ty:ty)*| $body:expr) => {
130            let s = n.clone();
131            op!($name, move |keys_id: u32, $p: u32 $(, $arg: $ty)*| {
132                let mut st = s.borrow_mut();
133                st.count($count);
134                if let Some($c) = st.cdms.get_mut(&keys_id) {
135                    $body;
136                }
137            });
138        };
139    }
140    let s = n.clone();
141    op!("__op_now", move || -> f64 {
142        s.borrow().start.elapsed().as_secs_f64() * 1000.0
143    });
144    let s = n.clone();
145    op!("__op_print", move |level: String, msg: String| {
146        eprintln!("JS[{level}] {}", crate::redact(&msg, &s.borrow().redact));
147    });
148    op!("__op_user_agent", crate::user_agent);
149    op!("__op_random", move |ctx: Ctx<'js>,
150                             len: u32|
151          -> rquickjs::Result<
152        ArrayBuffer<'js>,
153    > {
154        let mut buf = vec![0u8; len as usize];
155        if let Err(e) = getrandom::getrandom(&mut buf) {
156            return Err(rquickjs::Exception::throw_internal(
157                &ctx,
158                &format!("crypto.getRandomValues: no system randomness: {e}"),
159            ));
160        }
161        ArrayBuffer::new(ctx, buf)
162    });
163    let s = n.clone();
164    op!("__op_http", move |id: u32,
165                           method: String,
166                           url: String,
167                           headers: String,
168                           body: Value<'js>,
169                           timeout: f64,
170                           cache: String| {
171        let mut st = s.borrow_mut();
172        st.count("http");
173        st.http_count += 1;
174        st.in_flight += 1;
175        let headers: Vec<(String, String)> = serde_json::from_str(&headers).unwrap_or_default();
176        let req = HttpRequest {
177            id,
178            method,
179            url,
180            headers,
181            body: bytes_of(&body),
182            timeout_ms: if timeout > 0.0 {
183                timeout as u64
184            } else {
185                60_000
186            },
187            use_cache: cache == "default",
188        };
189        net::spawn(req, st.net_tx.clone());
190    });
191    let s = n.clone();
192    op!("__op_http_abort", move |id: u32| {
193        let mut st = s.borrow_mut();
194        st.count("http_abort");
195        st.aborted.insert(id);
196    });
197    let s = n.clone();
198    op!("__op_load_script", move |id: u32, url: String| {
199        let mut st = s.borrow_mut();
200        st.count("load_script");
201        st.in_flight += 1;
202        st.script_log
203            .push(json!({"id": id, "url": url, "phase": "fetching"}));
204        net::spawn_script(id, url, st.net_tx.clone());
205    });
206    let s = n.clone();
207    op!("__op_ms_create", move || -> u32 {
208        let mut st = s.borrow_mut();
209        st.count("ms_create");
210        let id = st.id();
211        st.mses.insert(
212            id,
213            MediaSourceNative {
214                duration: f64::NAN,
215                ended: false,
216                sbs: Vec::new(),
217            },
218        );
219        id
220    });
221    let s = n.clone();
222    op!("__op_sb_add", move |ms_id: u32, mime: String| -> i32 {
223        let mut st = s.borrow_mut();
224        st.count("sb_add");
225        if !audio_type_supported(&mime) {
226            return -1;
227        }
228        let id = st.id();
229        st.sbs
230            .insert(id, SourceBufferNative::new(mime, SOURCE_BUFFER_QUOTA_BYTES));
231        if let Some(ms) = st.mses.get_mut(&ms_id) {
232            ms.sbs.push(id);
233        }
234        id as i32
235    });
236    let s = n.clone();
237    op!("__op_sb_remove_buffer", move |ms_id: u32, sb_id: u32| {
238        let mut st = s.borrow_mut();
239        st.count("sb_remove_buffer");
240        st.sbs.remove(&sb_id);
241        if let Some(ms) = st.mses.get_mut(&ms_id) {
242            ms.sbs.retain(|x| *x != sb_id);
243        }
244    });
245    let s = n.clone();
246    op!("__op_sb_prepare_append", move |sb_id: u32,
247                                        len: u32|
248          -> bool {
249        let mut st = s.borrow_mut();
250        st.count("sb_prepare_append");
251        let position = st.position_for_buffer(sb_id);
252        st.sbs
253            .get_mut(&sb_id)
254            .is_some_and(|sb| sb.make_room(len as usize, position))
255    });
256    let s = n.clone();
257    op!("__op_sb_append", move |sb_id: u32,
258                                data: Value<'js>,
259                                ts_offset: f64|
260          -> String {
261        let mut st = s.borrow_mut();
262        st.count("sb_append");
263        let Some(bytes) = bytes_of(&data) else {
264            return json!({"error": "not an ArrayBuffer"}).to_string();
265        };
266        let Some(sb) = st.sbs.get_mut(&sb_id) else {
267            return json!({"error": "no such SourceBuffer"}).to_string();
268        };
269        match sb.append(&bytes, if ts_offset.is_finite() { ts_offset } else { 0.0 }) {
270            Ok(r) => json!({"init": r.init_segment, "frames": r.media_frames, "hasInitData": r.init_segment && sb.init_data.is_some(), "initDuration": sb.init.as_ref().and_then(|i| i.duration_secs), "end": sb.end_us().map(|e| e as f64 / 1e6)}).to_string(),
271            Err(e) => json!({"error": format!("{e:#}")}).to_string(),
272        }
273    });
274    let s = n.clone();
275    op!(
276        "__op_sb_set_append_window",
277        move |sb_id: u32, start: f64, end: f64| {
278            if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
279                let us = |t: f64| (t * 1e6).round() as i64;
280                sb.append_window_us = (us(start), us(end));
281            }
282        }
283    );
284    let s = n.clone();
285    op!("__op_sb_init_data", move |ctx: Ctx<'js>,
286                                   sb_id: u32|
287          -> rquickjs::Result<
288        Value<'js>,
289    > {
290        match s
291            .borrow()
292            .sbs
293            .get(&sb_id)
294            .and_then(|sb| sb.init_data.clone())
295        {
296            Some(b) => Ok(ArrayBuffer::new(ctx, b)?.into_value()),
297            None => Ok(Value::new_null(ctx)),
298        }
299    });
300    let s = n.clone();
301    op!("__op_sb_remove", move |sb_id: u32, start: f64, end: f64| {
302        let mut st = s.borrow_mut();
303        st.count("sb_remove");
304        if let Some(sb) = st.sbs.get_mut(&sb_id) {
305            sb.remove(start, end);
306        }
307    });
308    let s = n.clone();
309    op!("__op_sb_abort", move |sb_id: u32| {
310        if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
311            sb.abort();
312        }
313    });
314    let s = n.clone();
315    op!("__op_sb_change_type", move |sb_id: u32,
316                                     mime: String|
317          -> bool {
318        if !audio_type_supported(&mime) {
319            return false;
320        }
321        if let Some(sb) = s.borrow_mut().sbs.get_mut(&sb_id) {
322            sb.mime = mime;
323        }
324        true
325    });
326    let s = n.clone();
327    op!("__op_sb_buffered", move |sb_id: u32| -> String {
328        let r: Vec<[f64; 2]> = s
329            .borrow()
330            .sbs
331            .get(&sb_id)
332            .map(|sb| sb.buffered().into_iter().map(|(a, b)| [a, b]).collect())
333            .unwrap_or_default();
334        serde_json::to_string(&r).unwrap()
335    });
336    let s = n.clone();
337    op!("__op_ms_set_duration", move |ms_id: u32, d: f64| {
338        if let Some(ms) = s.borrow_mut().mses.get_mut(&ms_id) {
339            ms.duration = d;
340        }
341    });
342    let s = n.clone();
343    op!("__op_ms_end", move |ms_id: u32, ended: bool| {
344        if let Some(ms) = s.borrow_mut().mses.get_mut(&ms_id) {
345            ms.ended = ended;
346        }
347    });
348    let s = n.clone();
349    op!("__op_media_create", move || -> u32 {
350        let mut st = s.borrow_mut();
351        st.count("media_create");
352        let id = st.id();
353        let m = MediaEngine::new(st.sink_factory.clone());
354        st.medias.insert(id, m);
355        id
356    });
357    with_media!("__op_media_attach", "media_attach", |m, ms_id: u32| m
358        .attach(ms_id));
359    with_media!("__op_media_detach", "media_detach", |m| m.detach());
360    with_media!("__op_media_play", "media_play", |m| m.play());
361    with_media!("__op_media_pause", "media_pause", |m| m.pause());
362    with_media!("__op_media_seek", "media_seek", |m, t: f64| m
363        .seek((t * 1e6).round() as i64));
364    with_media!(
365        "__op_media_set_loop",
366        "media_set_loop",
367        |m, looping: bool| m.looping = looping
368    );
369    with_media!("__op_media_set_gain", "media_set_gain", |m, gain: f64| m
370        .set_gain(gain as f32));
371    with_media!(
372        "__op_media_set_keys",
373        "media_set_keys",
374        |m, keys_id: u32| m.set_keys(Some(keys_id).filter(|k| *k != 0))
375    );
376    let s = n.clone();
377    op!("__op_media_state", move |id: u32| -> String {
378        s.borrow()
379            .medias
380            .get(&id)
381            .map(media_state)
382            .unwrap_or_else(|| "{}".into())
383    });
384    let s = n.clone();
385    op!("__op_destroy", move |kind: String, id: u32| {
386        let mut st = s.borrow_mut();
387        st.count("destroy");
388        match kind.as_str() {
389            "ms" => {
390                for sb in st.mses.remove(&id).map(|ms| ms.sbs).unwrap_or_default() {
391                    st.sbs.remove(&sb);
392                }
393            }
394            "media" => {
395                if let Some(mut m) = st.medias.remove(&id) {
396                    m.collect_underflows();
397                    let counts = output_counts(&m.counters);
398                    st.retired_output = add(st.retired_output, counts);
399                }
400            }
401            "cdm" => drop(st.cdms.remove(&id)),
402            _ => {}
403        }
404    });
405    op!("__op_can_play_type", |mime: String| can_play_type(&mime));
406    op!("__op_type_supported", |mime: String| audio_type_supported(
407        &mime
408    ));
409    let s = n.clone();
410    op!("__op_eme_check", move |key_system: String,
411                                configs: String|
412          -> String {
413        let mut st = s.borrow_mut();
414        st.count("eme_check");
415        let r = eme_check(&key_system, &configs);
416        st.eme_requests.push(json!({"keySystem": key_system, "requested": serde_json::from_str::<J>(&configs).unwrap_or(J::Null), "granted": r.clone()}));
417        r.map(|x| x.to_string()).unwrap_or_default()
418    });
419    let s = n.clone();
420    op!(
421        "__op_cdm_create",
422        move |allow_persistent_state: bool| -> i32 {
423            let mut st = s.borrow_mut();
424            st.count("cdm_create");
425            match (st.cdm_factory)(allow_persistent_state) {
426                Ok(c) => {
427                    let id = st.id();
428                    st.cdms.insert(id, c);
429                    id as i32
430                }
431                Err(e) => {
432                    eprintln!("ENGINE cdm create failed: {e:#}");
433                    st.cdm_error = Some(format!("{e:#}"));
434                    -1
435                }
436            }
437        }
438    );
439    with_cdm!(
440        "__op_cdm_set_cert",
441        "cdm_set_cert",
442        |c, promise, data: Value<'js>| c
443            .set_server_certificate(promise, &bytes_of(&data).unwrap_or_default())
444    );
445    with_cdm!(
446        "__op_cdm_generate",
447        "cdm_generate",
448        |c, promise, init_type: u32, data: Value<'js>| c.create_session(
449            promise,
450            init_type,
451            &bytes_of(&data).unwrap_or_default()
452        )
453    );
454    with_cdm!(
455        "__op_cdm_update",
456        "cdm_update",
457        |c, promise, sid: String, data: Value<'js>| c.update_session(
458            promise,
459            &sid,
460            &bytes_of(&data).unwrap_or_default()
461        )
462    );
463    with_cdm!("__op_cdm_close", "cdm_close", |c, promise, sid: String| c
464        .close_session(promise, &sid));
465    with_cdm!(
466        "__op_cdm_remove",
467        "cdm_remove",
468        |c, promise, sid: String| c.remove_session(promise, &sid)
469    );
470    Ok(())
471}
472
473fn call<'js, A: IntoArgs<'js>>(
474    ctx: &Ctx<'js>,
475    n: &N,
476    obj: &str,
477    name: &str,
478    args: A,
479) -> Option<Value<'js>> {
480    let r = (|| -> rquickjs::Result<Value<'js>> {
481        let p: Object = ctx.globals().get(obj)?;
482        let f: Function = p.get(name)?;
483        f.call::<_, Value>(args)
484    })();
485    match r.catch(ctx) {
486        Ok(v) => Some(v),
487        Err(e) => {
488            let msg = crate::redact(&e.to_string(), &n.borrow().redact);
489            eprintln!("ENGINE call {obj}.{name} failed: {msg}");
490            None
491        }
492    }
493}
494
495fn buffer_or_null<'js>(ctx: &Ctx<'js>, body: Option<Vec<u8>>) -> Value<'js> {
496    body.and_then(|b| ArrayBuffer::new(ctx.clone(), b).ok())
497        .map(|b| b.into_value())
498        .unwrap_or_else(|| Value::new_null(ctx.clone()))
499}
500
501fn sha256_hex(bytes: &[u8]) -> String {
502    use sha2::{Digest, Sha256};
503    mp4::hex(&Sha256::digest(bytes))
504}
505
506fn handle_completion(ctx: &Ctx, n: &N, c: Completion) {
507    n.borrow_mut().in_flight -= 1;
508    match c {
509        Completion::Http(r) => {
510            {
511                let mut st = n.borrow_mut();
512                if st.aborted.remove(&r.id) {
513                    return;
514                }
515                st.http_bytes += r.body.len() as u64;
516            }
517            let meta = json!({"status": r.status, "statusText": r.status_text, "url": r.url, "headers": r.headers, "error": r.error}).to_string();
518            call(
519                ctx,
520                n,
521                "__platform",
522                "dispatchHttp",
523                (r.id, meta, buffer_or_null(ctx, Some(r.body))),
524            );
525        }
526        Completion::Script { id, url, result } => {
527            let mut entry = json!({"id": id, "url": url});
528            let outcome = result.and_then(|mut bytes| {
529                entry["fetchedBytes"] = json!(bytes.len());
530                entry["fetchedSha256"] = json!(sha256_hex(&bytes));
531                if n.borrow().truncate_scripts {
532                    bytes.truncate(bytes.len() / 2);
533                }
534                entry["evaluatedBytes"] = json!(bytes.len());
535                entry["evaluatedSha256"] = json!(sha256_hex(&bytes));
536                let t = Instant::now();
537                ctx.eval_with_options::<(), _>(bytes, opts(&url))
538                    .catch(ctx)
539                    .map(|_| t.elapsed())
540                    .map_err(|e| e.to_string())
541            });
542            let (ok, err) = match outcome {
543                Ok(took) => {
544                    entry["phase"] = json!("evaluated");
545                    entry["evalMs"] = json!(took.as_secs_f64() * 1000.0);
546                    (true, String::new())
547                }
548                Err(e) => {
549                    entry["phase"] = json!("error");
550                    entry["error"] = json!(e);
551                    (false, e)
552                }
553            };
554            n.borrow_mut().script_log.push(entry);
555            call(ctx, n, "__platform", "dispatchScript", (id, ok, err));
556        }
557    }
558}
559
560#[cfg(feature = "trace")]
561pub(crate) fn trace_meta(meta: &J) -> J {
562    let mut shown = meta.clone();
563    if let Some(count) = shown["keys"].as_array().map(Vec::len) {
564        shown["keys"] = json!(count);
565    }
566    if let Some(len) = shown["message"].as_str().map(str::len) {
567        shown["message"] = json!(len);
568    }
569    shown
570}
571
572fn dispatch_cdm(ctx: &Ctx, n: &N, keys_id: u32, e: CdmEvent) {
573    #[cfg(feature = "trace")]
574    let traced = format!("{:.3}", n.borrow().start.elapsed().as_secs_f64() * 1000.0);
575    let (meta, body) = match e {
576        CdmEvent::Initialized(ok) => (json!({"type": "initialized", "ok": ok}), None),
577        CdmEvent::NewSession { promise, sid } => (
578            json!({"type": "newSession", "promise": promise, "sid": sid}),
579            None,
580        ),
581        CdmEvent::Resolve(p) => (json!({"type": "resolve", "promise": p}), None),
582        CdmEvent::Reject {
583            promise,
584            exception,
585            system_code,
586            message,
587        } => (
588            json!({"type": "reject", "promise": promise, "exception": exception, "code": system_code, "message": message}),
589            None,
590        ),
591        CdmEvent::Message { sid, kind, body } => (
592            json!({"type": "message", "sid": sid, "kind": kind}),
593            Some(body),
594        ),
595        CdmEvent::Keys { sid, keys } => (
596            json!({"type": "keys", "sid": sid, "keys": keys.iter().map(|(k, s)| json!({"kid": mp4::hex(k), "status": s})).collect::<Vec<_>>()}),
597            None,
598        ),
599        CdmEvent::Expiration { sid, expiry } => (
600            json!({"type": "expiration", "sid": sid, "expiry": expiry}),
601            None,
602        ),
603        CdmEvent::Closed { sid } => (json!({"type": "closed", "sid": sid}), None),
604    };
605    call(
606        ctx,
607        n,
608        "__platform",
609        "dispatchCdm",
610        (
611            keys_id,
612            {
613                #[cfg(feature = "trace")]
614                eprintln!(
615                    "TRACE {traced} cdm-event keys={keys_id} {}",
616                    trace_meta(&meta)
617                );
618                meta.to_string()
619            },
620            buffer_or_null(ctx, body),
621        ),
622    );
623}
624
625impl Engine {
626    pub fn new(config: EngineConfig) -> Result<Engine> {
627        let EngineConfig {
628            cdm_path,
629            audio_sink_name,
630            output_device,
631            redact,
632        } = config;
633        let cdm: CdmFactory = Box::new(move |persistent| Widevine::create(&cdm_path, persistent));
634        let sink: SinkFactory = Rc::new(move || {
635            Box::new(PulseSink::start(
636                audio_sink_name.clone(),
637                output_device.clone(),
638            )) as Box<dyn AudioSink>
639        });
640        crate::net::clear_cache();
641        let engine = Engine::with_backends(cdm, sink)?;
642        {
643            let mut st = engine.n.borrow_mut();
644            st.redact = redact;
645        }
646        Ok(engine)
647    }
648
649    pub(crate) fn with_backends(
650        cdm_factory: CdmFactory,
651        sink_factory: SinkFactory,
652    ) -> Result<Engine> {
653        let (net_tx, net_rx) = channel();
654        let n: N = Rc::new(RefCell::new(Native {
655            start: Instant::now(),
656            next_id: 0,
657            mses: HashMap::new(),
658            sbs: HashMap::new(),
659            medias: HashMap::new(),
660            cdms: HashMap::new(),
661            cdm_factory,
662            sink_factory,
663            cdm_error: None,
664            net_tx,
665            aborted: HashSet::new(),
666            op_counts: BTreeMap::new(),
667            eme_requests: Vec::new(),
668            script_log: Vec::new(),
669            http_count: 0,
670            http_bytes: 0,
671            redact: Vec::new(),
672            truncate_scripts: false,
673            retired_output: [0; OUTPUT_COUNTS],
674            in_flight: 0,
675        }));
676        let rt = Runtime::new()?;
677        rt.set_max_stack_size(4 * 1024 * 1024);
678        let ctx = Context::full(&rt)?;
679        ctx.with(|ctx| install_ops(&ctx, &n))
680            .map_err(|e| anyhow!("install ops: {e}"))?;
681        let engine = Engine {
682            ctx,
683            rt,
684            n,
685            net_rx,
686            last_gc: Instant::now(),
687        };
688        engine.eval_script("platform.js", PLATFORM)?;
689        #[cfg(feature = "trace")]
690        engine.eval_script("trace.js", include_str!("../js/trace.js"))?;
691        Ok(engine)
692    }
693
694    fn drain(&self) {
695        for _ in 0..100_000 {
696            match self.rt.execute_pending_job() {
697                Ok(true) => continue,
698                Ok(false) => break,
699                Err(e) => eprintln!("ENGINE job error: {e:?}"),
700            }
701        }
702    }
703
704    #[cfg(any(test, feature = "sabotage"))]
705    pub fn sabotage_truncate_scripts(&self) {
706        self.n.borrow_mut().truncate_scripts = true;
707    }
708
709    #[cfg(any(test, feature = "diagnostics"))]
710    pub fn enable_diagnostics(&self) -> Result<()> {
711        self.eval_script("diagnostics.js", include_str!("../js/diagnostics.js"))
712    }
713
714    pub fn eval_script(&self, name: &str, source: &str) -> Result<()> {
715        let r = self.ctx.with(|ctx| {
716            ctx.eval_with_options::<(), _>(source, opts(name))
717                .catch(&ctx)
718                .map_err(|e| anyhow!("{name} failed: {e}"))
719        });
720        self.drain();
721        r
722    }
723
724    pub fn eval_json(&self, expression: &str) -> Result<J> {
725        let src = format!("JSON.stringify(({expression}))");
726        let r = self.ctx.with(|ctx| {
727            ctx.eval_with_options::<Option<String>, _>(src, opts("eval"))
728                .catch(&ctx)
729                .map_err(|e| anyhow!("eval failed: {e}"))
730        });
731        self.drain();
732        Ok(r?
733            .map(|s| serde_json::from_str(&s))
734            .transpose()?
735            .unwrap_or(J::Null))
736    }
737
738    pub(crate) fn global_number(&self, object: &str, key: &str) -> Option<f64> {
739        self.ctx
740            .with(|ctx| ctx.globals().get::<_, Object>(object)?.get::<_, f64>(key))
741            .ok()
742    }
743
744    pub fn run_gc(&mut self) {
745        self.rt.run_gc();
746        self.last_gc = Instant::now();
747        self.drain();
748    }
749
750    fn run_tasks(&self) -> (bool, Option<f64>) {
751        for ran in 0..500 {
752            let r: f64 = self.ctx.with(|ctx| {
753                call(&ctx, &self.n, "__platform", "runTimers", ())
754                    .and_then(|v| v.as_number())
755                    .unwrap_or(-1.0)
756            });
757            self.drain();
758            if r != -2.0 {
759                return (ran > 0, (r >= 0.0).then_some(r));
760            }
761        }
762        (true, None)
763    }
764
765    pub fn tick(&mut self) -> Duration {
766        if self.last_gc.elapsed() >= GC_INTERVAL {
767            self.run_gc();
768        }
769        self.drain();
770        let mut busy = false;
771        while let Ok(c) = self.net_rx.try_recv() {
772            busy = true;
773            self.ctx.with(|ctx| handle_completion(&ctx, &self.n, c));
774            self.drain();
775            self.run_tasks();
776        }
777        let cdm_events: Vec<(u32, CdmEvent)> = self
778            .n
779            .borrow_mut()
780            .cdms
781            .iter_mut()
782            .flat_map(|(id, c)| c.poll().into_iter().map(move |e| (*id, e)))
783            .collect();
784        busy |= !cdm_events.is_empty();
785        for (id, e) in cdm_events {
786            self.ctx.with(|ctx| dispatch_cdm(&ctx, &self.n, id, e));
787            self.drain();
788        }
789        let (ran, next_ms) = self.run_tasks();
790        busy |= ran;
791        let now = Instant::now();
792        let media_events: Vec<(u32, String)> = {
793            let mut guard = self.n.borrow_mut();
794            let st = &mut *guard;
795            let mut out = Vec::new();
796            for (id, m) in st.medias.iter_mut() {
797                m.pump(now, &st.sbs, &st.mses, &mut st.cdms);
798                if !m.events.is_empty() {
799                    let events = std::mem::take(&mut m.events);
800                    out.push((
801                        *id,
802                        format!(
803                            "{{\"events\":{},\"state\":{}}}",
804                            json!(events),
805                            media_state(m)
806                        ),
807                    ));
808                }
809            }
810            out
811        };
812        busy |= !media_events.is_empty();
813        for (id, payload) in media_events {
814            self.ctx.with(|ctx| {
815                call(&ctx, &self.n, "__platform", "dispatchMedia", (id, payload));
816            });
817            self.drain();
818            self.run_tasks();
819        }
820        let st = self.n.borrow();
821        let active = busy
822            || st.in_flight > st.aborted.len()
823            || self.rt.is_job_pending()
824            || st.medias.values().any(|m| !m.idle());
825        if active {
826            let ms = next_ms.unwrap_or(ACTIVE_WAIT_MS).clamp(1.0, ACTIVE_WAIT_MS);
827            return Duration::from_micros((ms * 1000.0) as u64);
828        }
829        let js = next_ms.map(|ms| Duration::from_micros((ms.max(1.0) * 1000.0) as u64));
830        let cdm = st.cdms.values().filter_map(|c| c.next_timer()).min();
831        js.into_iter().chain(cdm).min().unwrap_or(IDLE_WAIT)
832    }
833
834    pub fn run_for(&mut self, d: Duration) {
835        let end = Instant::now() + d;
836        while Instant::now() < end {
837            let wait = self.tick();
838            std::thread::sleep(wait.min(end.saturating_duration_since(Instant::now())));
839        }
840    }
841
842    pub fn current_time(&self) -> Option<f64> {
843        self.n.borrow().primary_media().map(|m| m.current_time())
844    }
845
846    pub fn item_time(&self) -> Option<f64> {
847        let st = self.n.borrow();
848        let m = st.primary_media()?;
849        let t = m.current_time_us();
850        let offset = m
851            .ms_id
852            .and_then(|id| st.mses.get(&id))
853            .and_then(|ms| ms.sbs.first())
854            .and_then(|id| st.sbs.get(id))
855            .and_then(|sb| sb.offset_us_at(t))
856            .unwrap_or(0);
857        Some((t - offset).max(0) as f64 / 1e6)
858    }
859
860    pub fn output_failed(&self) -> bool {
861        self.n.borrow().medias.values().any(|m| m.output_failed())
862    }
863
864    pub fn output_counts(&self) -> OutputCounts {
865        let st = self.n.borrow();
866        let live = st.medias.values().map(|m| output_counts(&m.counters));
867        live.fold(st.retired_output, add)
868    }
869
870    pub fn status(&self) -> J {
871        let st = self.n.borrow();
872        let medias: serde_json::Map<String, J> = st
873            .medias
874            .iter()
875            .map(|(id, m)| {
876                let c = &m.counters;
877                let rate = if m.rate > 0 { m.rate as f64 } else { 48000.0 };
878                let v = json!({
879                    "currentTime": m.current_time(), "maxCurrentTime": m.max_current_time_us as f64 / 1e6, "outputPosition": m.output_position(),
880                    "readyState": m.ready_state, "paused": m.paused, "seeking": m.seeking, "ended": m.ended, "waitingForKey": m.waiting_for_key, "keysId": m.keys_id,
881                    "rate": m.rate, "channels": m.channels, "clearSamples": c.clear_samples, "encryptedSamples": c.encrypted_samples, "decryptOk": c.decrypt_ok,
882                    "decryptFail": c.decrypt_fail.iter().map(|(k, v)| (k.to_string(), json!(v))).collect::<serde_json::Map<_, _>>(),
883                    "decodedClearSeconds": c.decoded_clear_frames as f64 / rate, "decodedDecryptedSeconds": c.decoded_decrypted_frames as f64 / rate,
884                    "decodeErr": c.decode_err, "keyWaits": c.key_waits, "underruns": c.underruns, "seeks": c.seeks, "loops": c.loops,
885                    "drainsAcked": c.drains_acked, "drainTimeouts": c.drain_timeouts, "sinkRestarts": c.sink_restarts,
886                });
887                (id.to_string(), v)
888            })
889            .collect();
890        let sbs: serde_json::Map<String, J> = st
891            .sbs
892            .iter()
893            .map(|(id, sb)| {
894                let init = sb.init.as_ref().map(|i| json!({"codec": i.codec, "timescale": i.timescale, "channels": i.channels, "sampleRate": i.sample_rate, "protected": i.protection.as_ref().map(|p| json!({"scheme": p.scheme, "kid": mp4::hex(&p.kid), "ivSize": p.iv_size})), "psshBytes": i.pssh.len()}));
895                (id.to_string(), json!({"mime": sb.mime, "appendedBytes": sb.appended_bytes, "bytes": sb.bytes, "evictedFrames": sb.evicted_frames, "frames": sb.frames.len(), "buffered": sb.buffered(), "init": init}))
896            })
897            .collect();
898        let platform = self
899            .ctx
900            .with(|ctx| {
901                call(&ctx, &self.n, "__platform", "statusJson", ())
902                    .and_then(|v| v.as_string().and_then(|s| s.to_string().ok()))
903            })
904            .and_then(|s| serde_json::from_str(&s).ok())
905            .unwrap_or(J::Null);
906        json!({
907            "pid": std::process::id(),
908            "elapsedSec": st.start.elapsed().as_secs_f64(),
909            "opCounts": st.op_counts,
910            "live": {"medias": st.medias.len(), "mediaSources": st.mses.len(), "sourceBuffers": st.sbs.len(), "cdms": st.cdms.len()},
911            "medias": medias,
912            "sourceBuffers": sbs,
913            "cdm": st.cdm_error.as_ref().map(|e| json!({"error": e})).or_else(|| st.cdms.values().next().map(|c| c.describe())),
914            "emeRequests": st.eme_requests,
915            "scripts": st.script_log,
916            "http": {"count": st.http_count, "bytes": st.http_bytes},
917            "platform": platform,
918        })
919    }
920}
921
922pub const OUTPUT_COUNTS: usize = 8;
923pub type OutputCounts = [u64; OUTPUT_COUNTS];
924
925pub(crate) fn output_counts(c: &crate::media::Counters) -> OutputCounts {
926    let decrypt_fails = c.decrypt_fail.values().sum();
927    [
928        c.drains_acked,
929        c.drain_timeouts,
930        c.sent_us,
931        c.skipped_us,
932        c.decrypted_us,
933        c.underruns,
934        decrypt_fails,
935        c.output_underflows,
936    ]
937}
938
939fn add(a: OutputCounts, b: OutputCounts) -> OutputCounts {
940    std::array::from_fn(|i| a[i] + b[i])
941}