//! A WHEP session (`draft-ietf-wish-whep-05`): a WebRTC player of one owner's tracks of a
//! room. The gateway subscribes to the tracks and their receipts tracks at the relay,
//! verifies every object against its receipt, and hands the player the frames that
//! verify, packetised back to RTP by the WebRTC stack.
//!
//! **What a WHEP player cannot see.** WebRTC carries media, not receipts: a WHEP player
//! receives frames and nothing that proves where they came from. The gateway is the
//! verifier on its behalf: an object whose receipt does not verify, whose receipt never
//! comes, or that does not open as an envelope of the track's codec, is dropped and
//! counted, and never reaches the player. The counts go into the session's signed record.
//! A WAI-aware player gets the receipts itself by subscribing to the same tracks and their
//! receipts tracks over MoQ (the relay's WebTransport and WebSocket paths), and verifying
//! each object as this module does (`extensions/live-ingest.md` §9).

use crate::catalog::{self, Declared};
use crate::frame::{self, Codec};
use crate::gateway::{check_if_match, forbidden, problem, random_hex, Cmd, Gateway, Handle, Reply, SDP, SDPFRAG};
use crate::record::{Energy, LedgerEntry, SessionRecord, SignedRecord, TrackEntry, FORMAT};
use crate::relay::STEP;
use crate::room::{media, Admitted, Room};
use crate::rtc;
use crate::sdp;
use hyper::StatusCode;
use joule_moq::carriage::ReceiptBundle;
use joule_moq::draft22::data::FetchEntry;
use joule_moq::draft22::{ControlMessage, LocationFilter, Parameter, Parameters, StreamHeader, TrackRequest, UniEvent};
use joule_moq::receipt::{ReceiptVerifier, VerifiedGroup};
use joule_moq::{GroupId, ObjectId, TrackAlias, TrackName, TrackNamespace};
use joule_moq_relay::client::Client;
use joule_moq_relay::wire::{ReqRx, ReqTx};
use std::collections::{BTreeMap, HashMap, VecDeque};
use std::sync::Arc;
use std::time::{Duration, Instant};
use str0m::change::SdpOffer;
use str0m::media::{Frequency, MediaKind, MediaTime, Mid};
use str0m::{Candidate, Event, IceConnectionState, Rtc};
use tokio::net::UdpSocket;
use tokio::sync::mpsc;

/// Frames held for a player that has not yet connected.
const HELD_FRAMES: usize = 2048;
const DISCONNECT_GRACE: Duration = Duration::from_secs(5);

/// One object from the relay: on a subgroup stream, by its downstream Track Alias, or on
/// a fill fetch stream, by the Request ID of the SUBSCRIBE it fills.
#[derive(Debug)]
enum In {
    Object { alias: u64, group: u64, object: u64, payload: Vec<u8> },
    Fill { request: u64, group: u64, object: u64, payload: Vec<u8> },
}

/// Read every data stream of `client`, each concurrently and every read bounded, until
/// the session closes.
async fn read_streams(client: Arc<Client>, tx: mpsc::UnboundedSender<In>) {
    loop {
        if tx.is_closed() {
            return;
        }
        let Some(mut r) = client.next_stream(Duration::from_secs(1)).await else { continue };
        let tx = tx.clone();
        tokio::spawn(async move {
            let Ok(Ok(h)) = tokio::time::timeout(STEP, r.start()).await else { return };
            loop {
                match tokio::time::timeout(STEP, r.next()).await {
                    Ok(Ok(Some(UniEvent::SubgroupObject(o)))) => {
                        if let StreamHeader::Subgroup(sh) = &h
                            && !o.payload.is_empty()
                        {
                            let _ = tx.send(In::Object { alias: sh.track_alias, group: sh.group_id, object: o.object_id, payload: o.payload });
                        }
                    }
                    Ok(Ok(Some(UniEvent::FetchEntry(FetchEntry::Object(o))))) => {
                        if let StreamHeader::Fetch(fh) = &h
                            && !o.payload.is_empty()
                        {
                            let _ = tx.send(In::Fill { request: fh.request_id, group: o.group_id, object: o.object_id, payload: o.payload });
                        }
                    }
                    Ok(Ok(Some(_))) => {}
                    _ => return,
                }
            }
        });
    }
}

/// SUBSCRIBE from the start of the track's latest group: the live part from the Next
/// Object, and a fill of the latest group from its first object (MOQT §3.5, §9.20.9), so a
/// player starts at a group's first object, a keyframe for video. Returns the Request ID,
/// the downstream Track Alias, and the request's stream.
async fn subscribe_joining(client: &Client, ns: TrackNamespace, name: TrackName) -> Result<(u64, u64, ReqTx, ReqRx), String> {
    let fill = Parameters::new().with(Parameter::LocationFilter(LocationFilter::RelativeStart { start_group: 1 }));
    let params = Parameters::new().with(Parameter::LocationFilter(LocationFilter::NextObject)).with(Parameter::FillParameters(fill));
    let id = client.next_request_id();
    let (tx, mut rx) = client.request(&ControlMessage::Subscribe(TrackRequest { request_id: id, namespace: ns, name, params })).await?;
    match Client::response(&mut rx).await? {
        ControlMessage::SubscribeOk(ok) => Ok((id, ok.track_alias, tx, rx)),
        ControlMessage::RequestError(e) => Err(format!("REQUEST_ERROR {}: {}", e.code, String::from_utf8_lossy(&e.reason))),
        other => Err(format!("an unexpected {}", other.name())),
    }
}

/// Counts of one track a WHEP session receives.
#[derive(Clone, Debug, Default, serde::Serialize)]
pub struct InCounts {
    pub objects_in: u64,
    pub bytes_in: u64,
    pub receipts_in: u64,
    pub groups_verified: u64,
    pub objects_verified: u64,
    /// Refused: the group receipt does not verify under a trusted key.
    pub refused_signature: u64,
    /// Refused: the object does not match its receipt or its proof.
    pub refused_inclusion: u64,
    /// Refused: the receipt names another track, group or object (moq-streaming-format
    /// §4.1, `moqt_mismatch`).
    pub refused_moqt: u64,
    /// Refused: the group receipt does not chain to the group before it.
    pub refused_chain: u64,
    /// Refused: the receipts object does not read as a bundle.
    pub refused_bundle: u64,
    /// The objects of the groups refused above (those counts are of groups).
    pub refused_objects: u64,
    /// Dropped: no receipt came within `hold_ms`.
    pub receipt_missing: u64,
    /// Receipted, not delivered: an object the receipts name that did not arrive.
    pub object_missing: u64,
    /// Dropped: it verified but does not open as an envelope of the track's codec.
    pub unopenable: u64,
    pub late: u64,
    pub duplicate: u64,
    pub frames_out: u64,
    pub bytes_out: u64,
}

struct Pending {
    last_seen: Instant,
    objects: BTreeMap<u64, Vec<u8>>,
    receipts: Option<(ReceiptBundle, VerifiedGroup)>,
    refused: bool,
    next: u64,
}

impl Pending {
    fn new(now: Instant) -> Self {
        Pending { last_seen: now, objects: BTreeMap::new(), receipts: None, refused: false, next: 0 }
    }
}

/// A frame that verified, for the player.
#[derive(Debug)]
pub struct Frame {
    pub kind: u8,
    pub codec: Codec,
    pub t: u64,
    pub data: Vec<u8>,
}

/// One track a WHEP session receives, and its verifier.
pub struct InTrack {
    pub kind: u8,
    pub declared: Declared,
    pub ns: TrackNamespace,
    pub media_alias: u64,
    pub receipts_alias: u64,
    pub media_request: u64,
    pub receipts_request: u64,
    groups: BTreeMap<u64, Pending>,
    links: BTreeMap<u64, [u8; 32]>,
    released: Option<u64>,
    pub counts: InCounts,
}

impl InTrack {
    fn object(&mut self, group: u64, object: u64, payload: Vec<u8>, now: Instant) {
        self.counts.objects_in += 1;
        self.counts.bytes_in += payload.len() as u64;
        if self.released.is_some_and(|r| group <= r) {
            self.counts.late += 1;
            return;
        }
        let p = self.groups.entry(group).or_insert_with(|| Pending::new(now));
        p.last_seen = now;
        if object < p.next || p.objects.insert(object, payload).is_some() {
            self.counts.duplicate += 1;
        }
    }

    fn receipts(&mut self, group: u64, payload: &[u8], verifier: &ReceiptVerifier, now: Instant) {
        self.counts.receipts_in += 1;
        if self.released.is_some_and(|r| group <= r) {
            return;
        }
        let p = self.groups.entry(group).or_insert_with(|| Pending::new(now));
        if p.receipts.is_some() || p.refused {
            return;
        }
        let Ok(b) = ReceiptBundle::from_payload(payload) else {
            self.counts.refused_bundle += 1;
            p.refused = true;
            return;
        };
        // A receipt names its MOQT group and object, and the track's receipt chain, before
        // its signature is checked (moq-streaming-format §4.1).
        let alias = TrackAlias(self.declared.receipt_alias);
        if b.group.group_id != GroupId(group) || b.group.track_alias != alias || b.objects.iter().enumerate().any(|(i, r)| r.check_moqt(alias, GroupId(group), ObjectId(i as u64)).is_err()) {
            self.counts.refused_moqt += 1;
            p.refused = true;
            return;
        }
        let vg = match verifier.verify_group(&b.group) {
            Ok(v) => v,
            Err(_) => {
                self.counts.refused_signature += 1;
                p.refused = true;
                return;
            }
        };
        if let Some(prev) = group.checked_sub(1).and_then(|g| self.links.get(&g))
            && b.group.parent_receipt_hash != Some(*prev)
        {
            self.counts.refused_chain += 1;
            p.refused = true;
            return;
        }
        self.links.insert(group, b.group.link_hash());
        self.counts.groups_verified += 1;
        p.receipts = Some((b, vg));
    }

    /// Hand on, in group order and object order, every object whose receipt verified;
    /// drop and count what fails, or what waited past `hold`.
    fn release(&mut self, verifier: &ReceiptVerifier, hold: Duration, now: Instant, out: &mut Vec<Frame>) {
        let alias = TrackAlias(self.declared.receipt_alias);
        while let Some((&g, p)) = self.groups.iter_mut().next() {
            let expired = now.duration_since(p.last_seen) > hold;
            let done = if p.refused {
                // Its group was refused by its receipt: every object of it is dropped.
                self.counts.refused_objects += p.objects.len() as u64;
                true
            } else if let Some((b, vg)) = &p.receipts {
                let n = b.group.leaf_count;
                while p.next < n {
                    let Some(payload) = p.objects.remove(&p.next) else { break };
                    let r = &b.objects[p.next as usize];
                    let ok = r.check_moqt(alias, GroupId(g), ObjectId(p.next)).is_ok() && verifier.verify_payload(&payload, r, vg).is_ok();
                    if !ok {
                        self.counts.refused_inclusion += 1;
                    } else {
                        self.counts.objects_verified += 1;
                        match frame::open(&payload).ok().filter(|o| o.codec == self.declared.codec).and_then(|o| frame::unframe(o.codec, &o.payload).map(|d| (o.t, d))) {
                            Some((t, data)) => out.push(Frame { kind: self.kind, codec: self.declared.codec, t, data }),
                            None => self.counts.unopenable += 1,
                        }
                    }
                    p.next += 1;
                }
                if p.next >= n {
                    true
                } else if expired {
                    // Receipted and not delivered, where the session held any of the
                    // group: a group whose objects all came before it joined is not.
                    if p.next > 0 || !p.objects.is_empty() {
                        self.counts.object_missing += n - p.next;
                    }
                    true
                } else {
                    false
                }
            } else if expired {
                self.counts.receipt_missing += p.objects.len() as u64;
                true
            } else {
                false
            };
            if !done {
                break;
            }
            self.groups.remove(&g);
            self.released = Some(g);
        }
    }
}

pub async fn create(gw: Arc<Gateway>, room: Arc<Room>, ad: Admitted, token_hash: [u8; 32], owner: String, layer: u8, offer: String) -> Reply {
    if !ad.grant.caps.subscribe {
        return forbidden("subscribe_not_permitted", "the grant does not let its principal subscribe");
    }
    if layer > ad.grant.caps.max_layer {
        return forbidden("layer_not_permitted", "the grant's max_layer is below the layer asked for");
    }
    for s in sdp::sections(&offer) {
        if (s.media == "audio" || s.media == "video") && matches!(s.direction.as_deref(), Some("sendonly") | Some("inactive")) {
            return problem(StatusCode::UNPROCESSABLE_ENTITY, "direction", "a WHEP offer's sections are recvonly or sendrecv (WHEP §4.3.6)");
        }
    }
    if !gw.reserve(true) {
        return gw.busy();
    }
    match open(&gw, &room, &ad, token_hash, &owner, layer, &offer).await {
        Ok(r) => r,
        Err(r) => {
            gw.release(true);
            r
        }
    }
}

fn not_live(gw: &Gateway, detail: &str) -> Reply {
    problem(StatusCode::CONFLICT, "not_live", detail).header(hyper::header::RETRY_AFTER, gw.cfg.whep.retry_after_s.to_string())
}

async fn open(gw: &Arc<Gateway>, room: &Arc<Room>, ad: &Admitted, token_hash: [u8; 32], owner: &str, layer: u8, offer: &str) -> Result<Reply, Reply> {
    let (client, endpoint) = gw.link.connect().await.map_err(|e| problem(StatusCode::SERVICE_UNAVAILABLE, "relay_unreachable", &e))?;
    let client = Arc::new(client);
    let (in_tx, mut in_rx) = mpsc::unbounded_channel();
    tokio::spawn(read_streams(client.clone(), in_tx.clone()));
    let close = |c: &Client| c.close(0, "WHEP session refused");

    // The catalog: which tracks the owner publishes, in which codec, under which receipt
    // chain (moq-streaming-format §4), read only once its own receipt verifies under a
    // trusted first hop. A player of an owner with none is told to come back (WHEP
    // §4.3.9).
    let verifier = ReceiptVerifier::with_anchors(gw.trust.clone());
    let cat_ns = catalog::catalog_namespace(&room.id, owner);
    let subscribed = async {
        let c = subscribe_joining(&client, cat_ns.clone(), TrackName::from_str(catalog::CATALOG_NAME)).await?;
        let r = subscribe_joining(&client, cat_ns.clone(), catalog::receipts_name(catalog::CATALOG_NAME)).await?;
        Ok::<_, String>((c, r))
    };
    let (cat, cat_r) = match subscribed.await {
        Ok(x) => x,
        Err(e) => {
            close(&client);
            return Err(not_live(gw, &format!("no catalog for this owner: {e}")));
        }
    };
    let deadline = tokio::time::Instant::now() + Duration::from_millis(gw.cfg.whep.catalog_wait_ms);
    let (mut bodies, mut bundles): (BTreeMap<u64, Vec<u8>>, BTreeMap<u64, Vec<u8>>) = (BTreeMap::new(), BTreeMap::new());
    let mut declared = None;
    let mut early = Vec::new();
    while declared.is_none() {
        match tokio::time::timeout_at(deadline, in_rx.recv()).await {
            Ok(Some(In::Object { alias, group, payload, .. })) if alias == cat.1 => drop(bodies.insert(group, payload)),
            Ok(Some(In::Fill { request, group, payload, .. })) if request == cat.0 => drop(bodies.insert(group, payload)),
            Ok(Some(In::Object { alias, group, payload, .. })) if alias == cat_r.1 => drop(bundles.insert(group, payload)),
            Ok(Some(In::Fill { request, group, payload, .. })) if request == cat_r.0 => drop(bundles.insert(group, payload)),
            Ok(Some(other)) => early.push(other),
            _ => break,
        }
        // The latest group whose catalog and receipt are both held, and verify.
        for (g, body) in bodies.iter().rev() {
            if let Some(b) = bundles.get(g).and_then(|b| ReceiptBundle::from_payload(b).ok())
                && b.group.group_id == GroupId(*g)
                && b.objects.len() == 1
                && b.objects[0].group_id == GroupId(*g)
                && b.objects[0].object_id == ObjectId(0)
                && let Ok(vg) = verifier.verify_group(&b.group)
                && verifier.verify_payload(body, &b.objects[0], &vg).is_ok()
            {
                declared = catalog::read(body);
                break;
            }
        }
    }
    let Some(declared) = declared else {
        close(&client);
        return Err(not_live(gw, "no catalog of the owner's, with a receipt that verifies, arrived"));
    };
    let mut want: Vec<(u8, Declared)> = Vec::new();
    for m in [media::AUDIO, media::VIDEO] {
        let ns = catalog::render(&catalog::media_namespace(&room.id, owner, m));
        if let Some(d) = declared.iter().find(|d| d.namespace == ns && d.name == layer.to_string() && d.codec.is_video() == (m == media::VIDEO)) {
            want.push((m, d.clone()));
        }
    }
    if want.is_empty() {
        close(&client);
        return Err(problem(StatusCode::NOT_FOUND, "layer_unknown", "the owner's catalog declares no track of this layer"));
    }

    // The answer: only the codecs the tracks carry, so a player that offers none of a
    // kind's has that section rejected (WHEP §4.3.4).
    let socket = UdpSocket::bind((gw.cfg.media.bind_ip, 0)).await.map_err(|e| problem(StatusCode::SERVICE_UNAVAILABLE, "media_socket", &e.to_string()))?;
    let local = socket.local_addr().expect("a bound socket");
    let codecs: Vec<Codec> = want.iter().map(|(_, d)| d.codec).collect();
    let mut rtc = rtc::config(gw.cfg.media.ice_lite, &codecs).build(Instant::now());
    let cand = Candidate::host(local, "udp").map_err(|e| problem(StatusCode::SERVICE_UNAVAILABLE, "media_candidate", &e.to_string()))?;
    rtc.add_local_candidate(cand);
    let parsed = SdpOffer::from_sdp_string(offer).map_err(|e| problem(StatusCode::BAD_REQUEST, "offer_malformed", &e.to_string()))?;
    let answer = rtc.sdp_api().accept_offer(parsed).map_err(|e| problem(StatusCode::UNPROCESSABLE_ENTITY, "offer_refused", &e.to_string()))?.to_sdp_string();
    let accepted: Vec<u8> = sdp::sections(&answer)
        .iter()
        .filter(|s| s.port != 0)
        .filter_map(|s| match s.media.as_str() {
            "audio" => Some(media::AUDIO),
            "video" => Some(media::VIDEO),
            _ => None,
        })
        .collect();
    let mut tracks = Vec::new();
    let mut done_watch = Vec::new();
    for (m, d) in want.into_iter().filter(|(m, _)| accepted.contains(m)) {
        let ns = catalog::media_namespace(&room.id, owner, m);
        let (mr, ma, mtx, mrx) = subscribe_joining(&client, ns.clone(), TrackName::from_str(&d.name)).await.map_err(|e| not_live(gw, &e))?;
        let (rr, ra, rtx, rrx) = subscribe_joining(&client, ns.clone(), TrackName::from_str(&d.receipts_track)).await.map_err(|e| not_live(gw, &e))?;
        done_watch.push((mtx, mrx));
        done_watch.push((rtx, rrx));
        tracks.push(InTrack {
            kind: m,
            declared: d,
            ns,
            media_alias: ma,
            receipts_alias: ra,
            media_request: mr,
            receipts_request: rr,
            groups: BTreeMap::new(),
            links: BTreeMap::new(),
            released: None,
            counts: InCounts::default(),
        });
    }
    if tracks.is_empty() {
        close(&client);
        return Err(problem(StatusCode::UNPROCESSABLE_ENTITY, "no_common_codec", "the offer names none of the codecs the owner's tracks carry"));
    }
    // A track's end at the relay (PUBLISH_DONE, MOQT §9.9) ends the session.
    let (end_tx, end_rx) = mpsc::unbounded_channel::<String>();
    for (i, (_tx, mut rx)) in done_watch.into_iter().enumerate() {
        let end_tx = end_tx.clone();
        tokio::spawn(async move {
            let _keep = _tx;
            loop {
                match rx.next().await {
                    Ok(Some(ControlMessage::PublishDone(d))) => {
                        if i % 2 == 0 {
                            let _ = end_tx.send(format!("track_ended ({:?})", d.status));
                        }
                        return;
                    }
                    Ok(Some(_)) => {}
                    _ => return,
                }
            }
        });
    }
    for e in early {
        let _ = in_tx.send(e);
    }
    drop(in_tx);

    let id = random_hex(16);
    let etag = random_hex(16);
    let (tx, rx) = mpsc::channel(16);
    let frames = Arc::new(std::sync::atomic::AtomicU64::new(0));
    gw.whep.lock().unwrap().insert(id.clone(), Handle { tx, token_hash, frames: frames.clone() });
    gw.stats.whep_opened.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let s = Session {
        gw: gw.clone(),
        id: id.clone(),
        room: room.clone(),
        ad: ad.clone(),
        layer,
        rtc,
        offer: offer.to_string(),
        etag: etag.clone(),
        client,
        _endpoint: endpoint,
        verifier,
        tracks,
        mids: HashMap::new(),
        held: VecDeque::new(),
        held_dropped: 0,
        base: None,
        connected: false,
        created: Instant::now(),
        disconnected_since: None,
        start_us: 0,
        frames,
    };
    tokio::spawn(s.run(socket, local, rx, in_rx, end_rx));
    Ok(Reply::new(StatusCode::CREATED).header(hyper::header::LOCATION, format!("/whep/s/{id}")).header(hyper::header::ETAG, format!("\"{etag}\"")).body(SDP, answer))
}

struct Session {
    gw: Arc<Gateway>,
    id: String,
    room: Arc<Room>,
    ad: Admitted,
    layer: u8,
    rtc: Rtc,
    offer: String,
    etag: String,
    client: Arc<Client>,
    _endpoint: quinn::Endpoint,
    verifier: ReceiptVerifier,
    tracks: Vec<InTrack>,
    mids: HashMap<u8, Mid>,
    held: VecDeque<Frame>,
    held_dropped: u64,
    /// The instant and room-clock time of the first frame written: every frame's
    /// wallclock is placed from it, so its tracks keep their relative timing.
    base: Option<(Instant, u64)>,
    connected: bool,
    created: Instant,
    disconnected_since: Option<Instant>,
    start_us: u64,
    frames: Arc<std::sync::atomic::AtomicU64>,
}

impl Session {
    fn room_now(&self) -> u64 {
        self.gw.clock.unix_us().saturating_sub(self.room.descriptor.t0_us)
    }

    async fn run(mut self, socket: UdpSocket, local: std::net::SocketAddr, mut rx: mpsc::Receiver<Cmd>, mut in_rx: mpsc::UnboundedReceiver<In>, mut end_rx: mpsc::UnboundedReceiver<String>) {
        self.start_us = self.room_now();
        let mut buf = vec![0u8; 2048];
        let mut events = Vec::new();
        let mut tick = tokio::time::interval(Duration::from_millis(50));
        let mut delete_reply = None;
        let hold = Duration::from_millis(self.gw.cfg.whep.hold_ms);
        let ended: String = 'run: loop {
            let next = match rtc::drain(&mut self.rtc, &socket, &mut events) {
                Ok(t) => t,
                Err(e) => break 'run format!("webrtc: {e}"),
            };
            if !events.is_empty() {
                let evs: Vec<Event> = std::mem::take(&mut events);
                for e in evs {
                    self.on_event(e);
                }
                continue;
            }
            let sleep = tokio::time::sleep_until(tokio::time::Instant::from_std(next.max(Instant::now())));
            tokio::select! {
                r = socket.recv_from(&mut buf) => {
                    if let Ok((n, from)) = r
                        && let Err(e) = rtc::input(&mut self.rtc, local, from, &buf[..n])
                    {
                        break 'run format!("webrtc: {e}");
                    }
                }
                _ = sleep => {
                    if let Err(e) = self.rtc.handle_input(str0m::Input::Timeout(Instant::now())) {
                        break 'run format!("webrtc: {e}");
                    }
                }
                i = in_rx.recv() => {
                    let Some(i) = i else { break 'run "relay_lost".into() };
                    self.on_in(i);
                    while let Ok(i) = in_rx.try_recv() {
                        self.on_in(i);
                    }
                    self.release(hold);
                }
                why = end_rx.recv() => {
                    // Whatever verified is handed on before the session ends.
                    while let Ok(i) = in_rx.try_recv() {
                        self.on_in(i);
                    }
                    self.release(Duration::ZERO);
                    break 'run why.unwrap_or_else(|| "track_ended".into());
                }
                c = rx.recv() => match c {
                    None => break 'run "gateway_stopped".into(),
                    Some(Cmd::Delete { reply }) => {
                        delete_reply = Some(reply);
                        break 'run "deleted".into();
                    }
                    Some(Cmd::Patch { frag, if_match, reply }) => {
                        let r = self.patch(frag, if_match.as_deref());
                        let _ = reply.send(r);
                    }
                },
                _ = tick.tick() => {
                    self.release(hold);
                    let m = &self.gw.cfg.media;
                    if !self.connected && self.created.elapsed() > Duration::from_millis(m.connect_timeout_ms) {
                        break 'run "connect_timeout".into();
                    }
                    if self.disconnected_since.is_some_and(|t| t.elapsed() > DISCONNECT_GRACE) {
                        break 'run "ice_disconnected".into();
                    }
                }
            }
        };
        self.end(&ended).await;
        if let Some(r) = delete_reply {
            let _ = r.send(());
        }
    }

    fn on_event(&mut self, e: Event) {
        match e {
            Event::Connected => {
                self.connected = true;
                while let Some(f) = self.held.pop_front() {
                    self.write(f);
                }
            }
            Event::IceConnectionStateChange(s) => match s {
                IceConnectionState::Disconnected => {
                    self.disconnected_since.get_or_insert_with(Instant::now);
                }
                IceConnectionState::Connected | IceConnectionState::Completed => self.disconnected_since = None,
                _ => {}
            },
            Event::MediaAdded(m) => {
                let k = if m.kind == MediaKind::Audio { media::AUDIO } else { media::VIDEO };
                self.mids.insert(k, m.mid);
            }
            _ => {}
        }
    }

    fn on_in(&mut self, i: In) {
        let now = Instant::now();
        for t in self.tracks.iter_mut() {
            match &i {
                In::Object { alias, group, object, payload } if *alias == t.media_alias => return t.object(*group, *object, payload.clone(), now),
                In::Fill { request, group, object, payload } if *request == t.media_request => return t.object(*group, *object, payload.clone(), now),
                In::Object { alias, group, payload, .. } if *alias == t.receipts_alias => return t.receipts(*group, payload, &self.verifier, now),
                In::Fill { request, group, payload, .. } if *request == t.receipts_request => return t.receipts(*group, payload, &self.verifier, now),
                _ => {}
            }
        }
    }

    fn release(&mut self, hold: Duration) {
        let now = Instant::now();
        let mut out = Vec::new();
        for t in self.tracks.iter_mut() {
            t.release(&self.verifier, hold, now, &mut out);
        }
        for f in out {
            if self.connected {
                self.write(f);
            } else {
                if self.held.len() >= HELD_FRAMES {
                    self.held.pop_front();
                    self.held_dropped += 1;
                }
                self.held.push_back(f);
            }
        }
    }

    /// One verified frame to the player: its RTP time from its room-clock time, its
    /// wallclock placed from the session's first frame.
    fn write(&mut self, f: Frame) {
        let Some(&mid) = self.mids.get(&f.kind) else { return };
        let Some(pt) = self.rtc.writer(mid).and_then(|w| w.payload_params().find(|p| p.spec().codec == f.codec.to_str0m()).map(|p| p.pt())) else { return };
        let (base_i, base_t) = *self.base.get_or_insert((Instant::now(), f.t));
        let wallclock = if f.t >= base_t { base_i + Duration::from_micros(f.t - base_t) } else { base_i.checked_sub(Duration::from_micros(base_t - f.t)).unwrap_or(base_i) };
        let (ticks, freq) = if f.codec.is_video() { (f.t as u128 * 90_000 / 1_000_000, Frequency::NINETY_KHZ) } else { (f.t as u128 * 48_000 / 1_000_000, Frequency::FORTY_EIGHT_KHZ) };
        let n = f.data.len() as u64;
        if let Some(w) = self.rtc.writer(mid)
            && w.write(pt, wallclock, MediaTime::new(ticks as u64, freq), f.data).is_ok()
            && let Some(t) = self.tracks.iter_mut().find(|t| t.kind == f.kind)
        {
            t.counts.frames_out += 1;
            t.counts.bytes_out += n;
            self.frames.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
        // A write changes the session: the caller's loop drains it before it waits.
    }

    fn patch(&mut self, frag: sdp::Frag, if_match: Option<&str>) -> Reply {
        if let Err(r) = check_if_match(if_match, &self.etag) {
            return r;
        }
        let current = (sdp::attr(&self.offer, "ice-ufrag").map(str::to_string), sdp::attr(&self.offer, "ice-pwd").map(str::to_string));
        let restart = matches!((&frag.ufrag, &frag.pwd), (Some(u), Some(p)) if (Some(u.clone()), Some(p.clone())) != current);
        if restart {
            let (u, p) = (frag.ufrag.as_deref().unwrap(), frag.pwd.as_deref().unwrap());
            let offer = sdp::restart_offer(&self.offer, u, p, &frag.candidates, frag.end_of_candidates);
            let parsed = match SdpOffer::from_sdp_string(&offer) {
                Ok(o) => o,
                Err(e) => return problem(StatusCode::BAD_REQUEST, "sdpfrag_malformed", &e.to_string()),
            };
            match self.rtc.sdp_api().accept_offer(parsed) {
                Ok(a) => {
                    self.offer = offer;
                    self.etag = random_hex(16);
                    Reply::new(StatusCode::OK).header(hyper::header::ETAG, format!("\"{}\"", self.etag)).body(SDPFRAG, sdp::restart_frag(&a.to_sdp_string()))
                }
                Err(e) => problem(StatusCode::UNPROCESSABLE_ENTITY, "ice_restart_refused", &e.to_string()),
            }
        } else {
            for c in &frag.candidates {
                if let Ok(c) = Candidate::from_sdp_string(c) {
                    self.rtc.add_remote_candidate(c);
                }
            }
            Reply::new(StatusCode::NO_CONTENT)
        }
    }

    async fn end(mut self, why: &str) {
        self.client.close(0, "the WHEP session ended");
        self.rtc.disconnect();
        let end_us = self.room_now();
        let record = self.record(why, end_us);
        self.gw.sink.record(&SignedRecord::sign(record, &self.gw.signer.key));
        self.gw.forget(true, &self.id);
        self.gw.release(true);
        tracing::info!(session = %self.id, "WHEP session ended: {why}");
    }

    fn record(&self, why: &str, end: u64) -> SessionRecord {
        let mut tracks = Vec::new();
        let mut detail = serde_json::Map::new();
        let (mut bytes_in, mut bytes_out) = (0, 0);
        for t in &self.tracks {
            let digest = hex::encode(catalog::track_digest(&t.ns, &t.declared.name));
            let base = TrackEntry { kind: 1, track: digest, namespace: catalog::render(&t.ns), name: t.declared.name.clone(), ..Default::default() };
            let c = &t.counts;
            let refused = c.refused_objects + c.refused_inclusion + c.receipt_missing + c.unopenable;
            tracks.push(TrackEntry { dir: 1, objects: c.objects_in, bytes: c.bytes_in, dropped_refused: refused, dropped_duplicate: c.duplicate, ..base.clone() });
            tracks.push(TrackEntry { dir: 2, objects: c.frames_out, bytes: c.bytes_out, receipted: c.object_missing, ..base });
            bytes_in += c.bytes_in;
            bytes_out += c.bytes_out;
            detail.insert(crate::room::media_name(t.kind).unwrap_or("media").into(), serde_json::to_value(c).unwrap());
        }
        tracks.sort_by(|a, b| (a.kind, &a.track, a.dir).cmp(&(b.kind, &b.track, b.dir)));
        detail.insert("held_dropped".into(), self.held_dropped.into());
        let duration = end.saturating_sub(self.start_us);
        let mut subscriber_us = [0u64; 9];
        subscriber_us[(self.layer as usize).min(7)] = duration;
        let g = &self.ad.grant;
        SessionRecord {
            format: FORMAT.into(),
            hop: hex::encode(self.gw.public_key()),
            role: 3,
            protocol: "whep".into(),
            session: self.id.clone(),
            room: hex::encode(self.room.id),
            principal: hex::encode(g.principal),
            start: self.start_us,
            end: end.max(self.start_us + 1),
            energy: Energy::unmetered(),
            tracks,
            sessions: 1,
            subscriber_us,
            ledger: vec![LedgerEntry {
                grant: hex::encode(self.ad.digest),
                joules_ceiling: g.joules_ceiling,
                funds_ceiling: g.funds_ceiling,
                currency: String::from_utf8_lossy(&g.currency).trim_matches('\0').to_string(),
                sessions: 1,
                bytes_in,
                bytes_out,
                subscriber_us,
            }],
            ended: why.into(),
            detail: serde_json::Value::Object(detail),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::frame::GrantRef;
    use crate::package::{Emit, Packager, Signer, TrackOut};
    use ed25519_dalek::SigningKey;

    fn declared(alias: u64) -> Declared {
        Declared { namespace: "n".into(), name: "0".into(), codec: Codec::Opus, receipt_alias: alias, receipts_track: "0_receipts".into() }
    }

    fn track(alias: u64) -> InTrack {
        InTrack {
            kind: media::AUDIO,
            declared: declared(alias),
            ns: TrackNamespace::from_strs(["n"]),
            media_alias: 2,
            receipts_alias: 3,
            media_request: 10,
            receipts_request: 12,
            groups: BTreeMap::new(),
            links: BTreeMap::new(),
            released: None,
            counts: InCounts::default(),
        }
    }

    /// Two audio groups as a WHIP session packages and signs them.
    fn signed(key: &SigningKey, alias: u64) -> Vec<Emit> {
        let signer = Signer { key: Arc::new(key.clone()), signer_id: "x".into(), origin_label: "wai-ingest".into() };
        let out = TrackOut { media: media::AUDIO, media_alias: 2, receipts_alias: 3, receipt_alias: alias, priority: 0 };
        let mut p = Packager::new(out, 48_000, 250_000, GrantRef { grant_hash: [1; 32], joule_ceiling_micro: u64::MAX, funds_ceiling: None }, signer, 0, 200_000, 2_000_000, 1 << 20);
        let mut e = Vec::new();
        for i in 0..20u64 {
            p.on_frame(Codec::Opus, &[0xfc, i as u8, 7], 960 * i, 1_000_000 + 20_000 * i, false, &mut e);
        }
        p.end_group(&mut e);
        e
    }

    fn feed(t: &mut InTrack, emits: &[Emit], v: &ReceiptVerifier, tamper: Option<(u64, u64)>) -> Vec<Frame> {
        let now = Instant::now();
        for e in emits {
            match e {
                Emit::Object { group, object, bytes, .. } => {
                    let mut b = bytes.clone();
                    if tamper == Some((*group, *object)) {
                        *b.last_mut().unwrap() ^= 1;
                    }
                    t.object(*group, *object, b, now);
                }
                Emit::Single { group, bytes, .. } => t.receipts(*group, bytes, v, now),
                Emit::EndGroup { .. } => {}
            }
        }
        let mut out = Vec::new();
        t.release(v, Duration::from_secs(5), now, &mut out);
        out
    }

    #[test]
    fn every_object_is_verified_before_it_is_handed_on() {
        let key = SigningKey::from_bytes(&[4; 32]);
        let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
        let e = signed(&key, 41);
        let mut t = track(41);
        let out = feed(&mut t, &e, &v, None);
        assert_eq!(out.len(), 20);
        assert!(out.iter().enumerate().all(|(i, f)| f.data == vec![0xfc, i as u8, 7]));
        assert_eq!(t.counts.objects_verified, 20);
        assert_eq!(t.counts.groups_verified, 2);
    }

    #[test]
    fn what_fails_is_dropped_and_counted() {
        let key = SigningKey::from_bytes(&[4; 32]);
        let e = signed(&key, 41);
        // A changed object.
        let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
        let first = e.iter().find_map(|x| if let Emit::Object { group, .. } = x { Some(*group) } else { None }).unwrap();
        let mut t = track(41);
        let out = feed(&mut t, &e, &v, Some((first, 3)));
        assert_eq!((out.len(), t.counts.refused_inclusion), (19, 1));
        // An untrusted signer: every object of both groups refused.
        let other = ReceiptVerifier::with_anchors(vec![SigningKey::from_bytes(&[5; 32]).verifying_key()]);
        let mut t = track(41);
        assert!(feed(&mut t, &e, &other, None).is_empty());
        assert_eq!(t.counts.refused_signature, 2);
        // Receipts for another chain.
        let mut t = track(42);
        assert!(feed(&mut t, &e, &v, None).is_empty());
        assert_eq!(t.counts.refused_moqt, 2);
        // No receipts at all: held, then dropped past the hold.
        let mut t = track(41);
        let now = Instant::now();
        for x in &e {
            if let Emit::Object { group, object, bytes, .. } = x {
                t.object(*group, *object, bytes.clone(), now);
            }
        }
        let mut out = Vec::new();
        t.release(&v, Duration::from_secs(5), now, &mut out);
        assert!(out.is_empty());
        t.release(&v, Duration::from_millis(0), now + Duration::from_millis(10), &mut out);
        assert!(out.is_empty());
        assert_eq!(t.counts.receipt_missing, 20);
    }
}