//! Frames into MoQ groups and objects, and each group's receipts signed as the first hop
//! (`extensions/live-ingest.md` §5 to §7).
//!
//! - **Objects.** Each frame is one object, one `WAI1` envelope ([`crate::frame`]).
//! - **Groups.** Video: a group begins at each keyframe, and holds it and the frames up
//!   to the next. Audio: a group holds the frames whose mapped time falls in one window
//!   of `audio_group_ms`, the windows laid end to end from the track's first frame. Group
//!   IDs ascend by one; a track's first group's ID is its first frame's room-clock time in
//!   microseconds, so a publisher that reconnects continues above the IDs it used before.
//! - **Receipts.** When a group ends, the gateway signs its group receipt over every
//!   object of it and publishes the group receipt and the object receipts as group *g* of
//!   the track's receipts track (moq-streaming-format §4.1): after the group's last object,
//!   since the root covers every object. Each receipt chains to the track's previous group.
//! - **Time.** [`Timeline`]: the RTP timestamp to room clock mapping.

use crate::frame::{self, Codec, FrameInfo, GrantRef, Header, Live};
use ed25519_dalek::SigningKey;
use joule_moq::carriage::ReceiptBundle;
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder};
use joule_moq::{GroupId, ObjectId, TrackAlias};
use std::sync::Arc;

/// The RTP timestamp to room clock mapping of one track (`extensions/live-ingest.md` §7).
///
/// The first frame anchors it: its room-clock time is its arrival at the gateway, `A`,
/// and its extended RTP timestamp is `R0`. A later frame with timestamp `R` maps to
/// `A + ⌊(R − R0) · 10^6 / rate⌋` µs. A frame whose mapped time lies more than
/// `max_skew` from its own arrival, or whose timestamp is below `R0`, re-anchors the
/// mapping at itself. Mapped times never decrease on a track. So every frame's time lies
/// within `max_skew` of its arrival at the gateway.
#[derive(Clone, Debug)]
pub struct Timeline {
    pub rate: u64,
    pub max_skew_us: u64,
    anchor: Option<(u64, u64)>,
    last: u64,
    pub reanchors: u64,
}

impl Timeline {
    pub fn new(rate: u64, max_skew_us: u64) -> Self {
        Timeline { rate, max_skew_us, anchor: None, last: 0, reanchors: 0 }
    }

    /// The room-clock time of a frame with extended RTP timestamp `rtp` that arrived at
    /// room-clock `arrival`.
    pub fn map(&mut self, rtp: u64, arrival: u64) -> u64 {
        let t = match self.anchor {
            Some((a, r0)) if rtp >= r0 => {
                let t = a as u128 + (rtp - r0) as u128 * 1_000_000 / self.rate as u128;
                let t = t.min(u64::MAX as u128) as u64;
                if t.abs_diff(arrival) > self.max_skew_us {
                    self.reanchors += 1;
                    self.anchor = Some((arrival, rtp));
                    arrival
                } else {
                    t
                }
            }
            Some(_) => {
                self.reanchors += 1;
                self.anchor = Some((arrival, rtp));
                arrival
            }
            None => {
                self.anchor = Some((arrival, rtp));
                arrival
            }
        };
        self.last = self.last.max(t);
        self.last
    }
}

/// Where one media track and its receipts track are published.
#[derive(Clone, Debug)]
pub struct TrackOut {
    /// A media kind of WAI live-rooms §6.1.
    pub media: u8,
    pub media_alias: u64,
    pub receipts_alias: u64,
    /// The `track_alias` the receipts sign, the catalog's `wai_receipt_alias`.
    pub receipt_alias: u64,
    pub priority: u8,
}

/// What a packager asks of its publisher.
#[derive(Clone, Debug)]
pub enum Emit {
    /// One object of a media track's group. `first` opens the group's subgroup stream.
    Object { alias: u64, group: u64, object: u64, first: bool, priority: u8, bytes: Vec<u8> },
    /// The group's last object was sent: its stream ends with a FIN.
    EndGroup { alias: u64, group: u64 },
    /// A group of one object on a track of its own stream: a receipts bundle, or a catalog.
    Single { alias: u64, group: u64, priority: u8, bytes: Vec<u8> },
}

/// What a frame did, beside what it emitted.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Verdict {
    Carried,
    /// Dropped: no group is open and the frame is not a keyframe. Ask for one.
    NeedKeyframe,
    /// Carried, and its group has run past `max_video_group_ms`: ask for a keyframe.
    GroupLong,
    /// Dropped: larger than `max_frame_bytes`.
    TooLarge,
    /// Dropped: a keyframe whose header gives no picture size, which its framing needs.
    Unreadable,
}

/// Counts of one track, in the shape of a hop record's track entries (`crate::record`).
#[derive(Clone, Debug, Default, serde::Serialize)]
pub struct TrackCounts {
    pub frames_in: u64,
    pub bytes_in: u64,
    pub objects_out: u64,
    pub bytes_out: u64,
    pub groups: u64,
    pub dropped_before_keyframe: u64,
    pub dropped_too_large: u64,
    pub dropped_unreadable: u64,
    pub reanchors: u64,
}

struct OpenGroup {
    id: u64,
    hashes: Vec<[u8; 32]>,
    start_t: u64,
    last_t: u64,
}

/// What signs: the gateway's key and the labels its receipts carry.
#[derive(Clone)]
pub struct Signer {
    pub key: Arc<SigningKey>,
    pub signer_id: String,
    pub origin_label: String,
}

/// One media track of a WHIP session.
pub struct Packager {
    pub out: TrackOut,
    pub codec: Option<Codec>,
    pub header: Option<Header>,
    pub timeline: Timeline,
    pub counts: TrackCounts,
    group: Option<OpenGroup>,
    next_group: Option<u64>,
    prev_link: Option<[u8; 32]>,
    last_ticks: Option<u64>,
    grant: GrantRef,
    signer: Signer,
    t0_us: u64,
    audio_group_us: u64,
    max_video_group_us: u64,
    max_frame_bytes: usize,
}

impl Packager {
    #[allow(clippy::too_many_arguments)]
    pub fn new(out: TrackOut, rate: u64, max_skew_us: u64, grant: GrantRef, signer: Signer, t0_us: u64, audio_group_us: u64, max_video_group_us: u64, max_frame_bytes: usize) -> Self {
        Packager {
            out,
            codec: None,
            header: None,
            timeline: Timeline::new(rate, max_skew_us),
            counts: TrackCounts::default(),
            group: None,
            next_group: None,
            prev_link: None,
            last_ticks: None,
            grant,
            signer,
            t0_us,
            audio_group_us: audio_group_us.max(1),
            max_video_group_us,
            max_frame_bytes,
        }
    }

    pub fn is_video(&self) -> bool {
        self.out.media != crate::room::media::AUDIO
    }

    /// One frame, depacketised: its codec, bytes, extended RTP timestamp and room-clock
    /// arrival, and whether the WebRTC stack read it as a keyframe. Returns whether the
    /// catalog must be published again (the track's codec or picture size changed).
    pub fn on_frame(&mut self, codec: Codec, data: &[u8], rtp: u64, arrival: u64, keyframe: bool, emits: &mut Vec<Emit>) -> (Verdict, bool) {
        self.counts.frames_in += 1;
        self.counts.bytes_in += data.len() as u64;
        if data.len() > self.max_frame_bytes {
            self.counts.dropped_too_large += 1;
            return (Verdict::TooLarge, false);
        }
        let mut catalog = false;
        if self.codec != Some(codec) {
            // A new codec begins a new group, at its first keyframe.
            self.end_group(emits);
            self.codec = Some(codec);
            self.header = None;
            catalog = true;
        }
        let t = self.timeline.map(rtp, arrival);
        self.counts.reanchors = self.timeline.reanchors;
        let key = match codec {
            Codec::Avc => keyframe || frame::avc_is_idr(data),
            Codec::Opus => false,
            _ => keyframe,
        };

        // The group this frame belongs to.
        if self.is_video() {
            if key {
                let h = match codec {
                    Codec::Avc => frame::avc_header(data).or_else(|| self.header.clone()),
                    Codec::Vp8 => frame::vp8_header(data),
                    Codec::Vp9 => frame::vp9_header(data),
                    Codec::Av1 => frame::av1_header(data).or_else(|| self.header.clone()),
                    Codec::Opus => None,
                };
                match h {
                    Some(h) => {
                        if self.header.as_ref() != Some(&h) {
                            catalog = true;
                            self.header = Some(h);
                        }
                    }
                    None if codec != Codec::Avc => {
                        self.counts.dropped_unreadable += 1;
                        return (Verdict::Unreadable, catalog);
                    }
                    None => {}
                }
                self.end_group(emits);
                self.open_group(t);
            } else if self.group.is_none() {
                self.counts.dropped_before_keyframe += 1;
                return (Verdict::NeedKeyframe, catalog);
            }
        } else {
            if self.header.is_none() {
                let ch = frame::opus_channels(data).unwrap_or(2);
                self.header = Some(Header { width: 0, height: 0, codec: Some("opus".into()), channels: ch });
                catalog = true;
            }
            let window_ends = self.group.as_ref().map(|g| g.start_t + self.audio_group_us);
            if window_ends.is_none_or(|e| t >= e) {
                self.end_group(emits);
                // Windows are laid end to end: the next one starts where the last ended,
                // or at this frame after a pause longer than a window.
                let start = match window_ends {
                    Some(e) if t < e + self.audio_group_us => e,
                    _ => t,
                };
                self.open_group(start);
            }
        }

        let ticks = t as u128 * 90_000 / 1_000_000;
        let ticks = ticks as u64;
        let delta = self.last_ticks.map_or(3_000, |p| ticks.saturating_sub(p));
        self.last_ticks = Some(ticks);
        let (w, h) = self.header.as_ref().map_or((0, 0), |h| (h.width, h.height));
        let video = self.is_video();
        let payload = frame::payload(codec, data, FrameInfo { width: w, height: h, ticks_90k: ticks, delta_90k: delta });
        let g = self.group.as_mut().expect("a group is open");
        let first = g.hashes.is_empty();
        let live = Live {
            t,
            rtp,
            rate: codec.clock_rate(),
            key: codec.is_video().then_some(key),
            grant: first.then(|| self.grant.clone()),
        };
        let envelope = frame::envelope(&frame::manifest(codec, &live), &payload);
        g.hashes.push(ObjectReceipt::hash_payload(&envelope));
        g.last_t = t;
        let object = g.hashes.len() as u64 - 1;
        self.counts.objects_out += 1;
        self.counts.bytes_out += envelope.len() as u64;
        emits.push(Emit::Object { alias: self.out.media_alias, group: g.id, object, first, priority: self.out.priority, bytes: envelope });
        let long = video && t.saturating_sub(g.start_t) > self.max_video_group_us;
        (if long { Verdict::GroupLong } else { Verdict::Carried }, catalog)
    }

    fn open_group(&mut self, start_t: u64) {
        let id = self.next_group.unwrap_or(start_t);
        self.next_group = Some(id + 1);
        self.group = Some(OpenGroup { id, hashes: Vec::new(), start_t, last_t: start_t });
        self.counts.groups += 1;
    }

    /// End the open group, if any: its stream's FIN, then its receipts, signed.
    pub fn end_group(&mut self, emits: &mut Vec<Emit>) {
        let Some(g) = self.group.take() else { return };
        emits.push(Emit::EndGroup { alias: self.out.media_alias, group: g.id });
        if g.hashes.is_empty() {
            return;
        }
        let bundle = self.sign(&g);
        emits.push(Emit::Single { alias: self.out.receipts_alias, group: g.id, priority: self.out.priority, bytes: bundle });
    }

    fn sign(&mut self, g: &OpenGroup) -> Vec<u8> {
        let start = rfc3339(self.t0_us.saturating_add(g.start_t));
        let end = rfc3339(self.t0_us.saturating_add(g.last_t));
        let (bundle, link) = sign_group(&self.signer, self.out.receipt_alias, g.id, &g.hashes, &start, &end, self.prev_link);
        self.prev_link = Some(link);
        bundle
    }
}

/// The receipts of one group, as group `group` of its receipts track carries them: the
/// group receipt, signed by `signer` over every object's labelled leaf and chained to
/// `parent`, and each object's receipt with its Merkle proof. Returns the bundle and the
/// group receipt's link hash, the next group's parent.
#[allow(clippy::too_many_arguments)]
pub fn sign_group(signer: &Signer, receipt_alias: u64, group: u64, hashes: &[[u8; 32]], start: &str, end: &str, parent: Option<[u8; 32]>) -> (Vec<u8>, [u8; 32]) {
    let alias = TrackAlias(receipt_alias);
    let mut b = ReceiptBuilder::new(alias, GroupId(group));
    for h in hashes {
        // No meter backs the gateway: every figure is the unmetered marker, 0 with class
        // tag 0 (WAI energy-measurement §2), never a measurement of zero joules.
        b.add_labelled_object(*h, 0, signer.origin_label.clone(), start.to_string(), vec![0]).expect("the unmetered marker is a class");
    }
    let objects: Vec<ObjectReceipt> = hashes
        .iter()
        .enumerate()
        .map(|(i, h)| ObjectReceipt {
            track_alias: alias,
            group_id: GroupId(group),
            object_id: ObjectId(i as u64),
            content_hash: *h,
            joules_micro: 0,
            origin: signer.origin_label.clone(),
            merkle_proof: b.proof_for(i).expect("an index of the group"),
            energy_class: Some(vec![0]),
        })
        .collect();
    let g = b.finalize_labelled(&signer.key, signer.signer_id.clone(), end.to_string(), parent).expect("the unmetered marker composes");
    let link = g.link_hash();
    (ReceiptBundle { group: g, objects }.to_payload().expect("a bundle serialises"), link)
}

/// An instant in µs since 1970 as RFC 3339 UTC with microseconds.
pub fn rfc3339(unix_us: u64) -> String {
    let secs = unix_us / 1_000_000;
    let micros = unix_us % 1_000_000;
    let days = (secs / 86_400) as i64;
    let rem = secs % 86_400;
    // Civil from days (Howard Hinnant's algorithm), 1970-01-01 = day 0.
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    format!("{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}.{micros:06}Z", rem / 3600, rem / 60 % 60, rem % 60)
}

#[cfg(test)]
mod tests {
    use super::*;
    use joule_moq::receipt::ReceiptVerifier;

    #[test]
    fn rfc3339_dates() {
        assert_eq!(rfc3339(0), "1970-01-01T00:00:00.000000Z");
        assert_eq!(rfc3339(1_792_281_600_123_456), "2026-10-18T00:00:00.123456Z");
        assert_eq!(rfc3339(951_782_400_000_000), "2000-02-29T00:00:00.000000Z");
    }

    #[test]
    fn the_timeline_holds_its_bound() {
        let mut tl = Timeline::new(90_000, 250_000);
        assert_eq!(tl.map(1_000, 5_000_000), 5_000_000);
        // 90 000 ticks later: one second later, arriving 40 ms late.
        assert_eq!(tl.map(91_000, 6_040_000), 6_000_000);
        // A frame whose mapping would stray 300 ms from its arrival re-anchors.
        assert_eq!(tl.map(181_000, 7_300_000), 7_300_000);
        assert_eq!(tl.reanchors, 1);
        // Mapped times never decrease.
        assert_eq!(tl.map(100, 7_000_000), 7_300_000);
    }

    fn packager(media: u8) -> Packager {
        let signer = Signer { key: Arc::new(SigningKey::from_bytes(&[3; 32])), signer_id: "test".into(), origin_label: "wai-ingest".into() };
        let out = TrackOut { media, media_alias: 2, receipts_alias: 3, receipt_alias: 77, priority: 1 };
        let rate = if media == crate::room::media::AUDIO { 48_000 } else { 90_000 };
        Packager::new(out, rate, 250_000, GrantRef { grant_hash: [9; 32], joule_ceiling_micro: u64::MAX, funds_ceiling: None }, signer, 1_000_000_000, 200_000, 2_000_000, 1 << 20)
    }

    fn avc(idr: bool, n: u8) -> Vec<u8> {
        let mut f = vec![0, 0, 0, 1, 0x67, 0x42, 0xe0, 0x1f, 0, 0, 0, 1, 0x68, 0xce];
        f.truncate(if idr { 14 } else { 0 });
        f.extend_from_slice(&[0, 0, 0, 1, if idr { 0x65 } else { 0x41 }, n, n, n]);
        f
    }

    #[test]
    fn video_groups_begin_at_keyframes_and_their_receipts_verify() {
        let mut p = packager(crate::room::media::VIDEO);
        let mut e = Vec::new();
        assert_eq!(p.on_frame(Codec::Avc, &avc(false, 1), 0, 10, false, &mut e).0, Verdict::NeedKeyframe);
        assert!(e.is_empty());
        for i in 0..6u64 {
            let idr = i % 3 == 0;
            let (v, _) = p.on_frame(Codec::Avc, &avc(idr, i as u8 + 2), 3000 * i, 1_000 + 33_333 * i, idr, &mut e);
            assert_eq!(v, Verdict::Carried);
        }
        p.end_group(&mut e);
        let objects: Vec<_> = e.iter().filter_map(|x| if let Emit::Object { group, object, bytes, .. } = x { Some((*group, *object, bytes.clone())) } else { None }).collect();
        let bundles: Vec<_> = e.iter().filter_map(|x| if let Emit::Single { group, bytes, .. } = x { Some((*group, ReceiptBundle::from_payload(bytes).unwrap())) } else { None }).collect();
        assert_eq!(objects.len(), 6);
        assert_eq!(bundles.len(), 2, "two groups, each ended by the keyframe after it or by the end");
        let first = objects[0].0;
        // The dropped frame anchored the timeline at its arrival, 10 µs; the first IDR's RTP
        // time maps there too.
        assert_eq!(first, 10, "the first group's ID is its first frame's room-clock time");
        assert_eq!(bundles[1].0, first + 1);
        let v = ReceiptVerifier::with_anchors(vec![SigningKey::from_bytes(&[3; 32]).verifying_key()]);
        for (g, b) in &bundles {
            let vg = v.verify_group(&b.group).unwrap();
            assert_eq!(b.group.track_alias, TrackAlias(77));
            for (og, o, bytes) in objects.iter().filter(|o| o.0 == *g) {
                let r = &b.objects[*o as usize];
                r.check_moqt(TrackAlias(77), GroupId(*og), ObjectId(*o)).unwrap();
                v.verify_payload(bytes, r, &vg).unwrap();
            }
        }
        assert_eq!(bundles[1].1.group.parent_receipt_hash, Some(bundles[0].1.group.link_hash()), "each group chains to the one before");
        // The first object of a group names the grant; the others do not.
        let o0 = frame::open(&objects[0].2).unwrap();
        assert!(o0.manifest["live"]["grant"].is_object() && o0.key);
        let o1 = frame::open(&objects[1].2).unwrap();
        assert!(o1.manifest["live"].get("grant").is_none() && !o1.key);
    }

    #[test]
    fn audio_groups_are_fixed_windows() {
        let mut p = packager(crate::room::media::AUDIO);
        let mut e = Vec::new();
        // 20 ms packets for 1 s: five windows of 200 ms.
        for i in 0..50u64 {
            p.on_frame(Codec::Opus, &[0xfc, i as u8, 1, 2], 960 * i, 2_000_000 + 20_000 * i, false, &mut e);
        }
        p.end_group(&mut e);
        let groups: Vec<u64> = e.iter().filter_map(|x| if let Emit::Single { group, .. } = x { Some(*group) } else { None }).collect();
        assert_eq!(groups.len(), 5);
        assert!(groups.windows(2).all(|w| w[1] == w[0] + 1));
        let per_group = e.iter().filter(|x| matches!(x, Emit::Object { .. })).count() / 5;
        assert_eq!(per_group, 10);
    }
}