//! A recorded room as MoQ files (`draft-jennings-moq-file-05`; WAI live-recording §8.7), end
//! to end: a publisher sends a room's video and its receipts through the relay, the DVR store
//! keeps them, the store writes its window as `.moq` and `.dat` files, and a reader that holds
//! only those files and the publisher's key verifies every object against its group's signed
//! receipt. What the files state is what a FETCH of the same window brings, byte for byte.
//!
//! Planted, each refused by the reader with its own reason and nothing of the track
//! delivered: a data file cut short, a `.moq` file cut short, and a `.moq` file naming a data
//! file outside its directory; and a payload changed in the data file, which reads (the files
//! carry no check of their own) and is caught by the receipt.

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq::carriage::ReceiptBundle;
use joule_moq::draft22::data::{FetchEntry, SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::Parameters;
use joule_moq::file::{self, FileError, FileObject, FileRefusal};
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder, ReceiptVerifier};
use joule_moq::{GroupId, ObjectId, TrackAlias};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{DvrConfig, DvrWindow};
use std::path::Path;
use std::time::Duration;
use tokio::time::timeout;

const GROUPS: u64 = 6;
const OBJECTS: u64 = 3;

fn payload_of(g: u64, o: u64) -> Vec<u8> {
    let mut v = format!("rec:{g}:{o}:").into_bytes();
    v.resize(700 + (g * 13 + o) as usize, (g * 7 + o) as u8);
    v
}

fn bundle_of(key: &SigningKey, receipt_alias: u64, g: u64, payloads: &[Vec<u8>]) -> ReceiptBundle {
    let mut b = ReceiptBuilder::new(TrackAlias(receipt_alias), GroupId(g));
    for p in payloads {
        b.add_labelled_object(ObjectReceipt::hash_payload(p), 0, "test-publisher", "2026-11-10T00:00:00Z", vec![0]).unwrap();
    }
    let objects = payloads
        .iter()
        .enumerate()
        .map(|(i, p)| ObjectReceipt {
            track_alias: TrackAlias(receipt_alias),
            group_id: GroupId(g),
            object_id: ObjectId(i as u64),
            content_hash: ObjectReceipt::hash_payload(p),
            joules_micro: 0,
            origin: "test-publisher".into(),
            merkle_proof: b.proof_for(i).unwrap(),
            energy_class: Some(vec![0]),
        })
        .collect();
    let group = b.finalize_labelled(key, "test-publisher", "2026-11-10T00:00:01Z", None).unwrap();
    ReceiptBundle { group, objects }
}

async fn publish(c: &Client, key: &SigningKey, t: &Tracks, g: u64) {
    let payloads: Vec<Vec<u8>> = (0..OBJECTS).map(|o| payload_of(g, o)).collect();
    let bundle = bundle_of(key, t.receipt_alias, g, &payloads);
    let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.expect("the receipts");
    let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
    let objects: Vec<SubgroupObject> = payloads.into_iter().enumerate().map(|(o, p)| SubgroupObject::normal(o as u64, p)).collect();
    c.send_subgroup(mh, &objects).await.expect("the objects");
}

/// Verify a directory's media track against its receipts track, from the files alone: the
/// objects verified, or why not.
fn verify_files(dir: &Path, t: &Tracks, publisher: &SigningKey) -> Result<usize, String> {
    let objects = file::read_dir(dir).map_err(|e| e.to_string())?;
    let verifier = ReceiptVerifier::with_anchors(vec![publisher.verifying_key()]);
    let of = |name: &joule_moq::TrackName| -> Vec<&FileObject> { objects.iter().filter(|o| o.namespace == t.ns && &o.name == name).collect() };
    let mut verified = 0;
    for m in of(&t.media) {
        let r = of(&t.receipts).into_iter().find(|r| r.group_id == m.group_id).ok_or(format!("no receipts for group {}", m.group_id))?;
        let b = ReceiptBundle::from_payload(&r.payload).map_err(|e| e.to_string())?;
        let vg = verifier.verify_group(&b.group).map_err(|e| e.to_string())?;
        let receipt = b.objects.get(m.object_id as usize).ok_or("no receipt at the object's position")?;
        verifier.verify_payload(&m.payload, receipt, &vg).map_err(|e| format!("group {} object {}: {e}", m.group_id, m.object_id))?;
        verified += 1;
    }
    Ok(verified)
}

fn copy_dir(from: &Path, name: &str) -> std::path::PathBuf {
    let to = store_dir(name);
    std::fs::create_dir_all(&to).unwrap();
    for e in std::fs::read_dir(from).unwrap() {
        let e = e.unwrap();
        std::fs::copy(e.path(), to.join(e.file_name())).unwrap();
    }
    to
}

fn one_with(dir: &Path, ext: &str, holding: &[u8]) -> std::path::PathBuf {
    std::fs::read_dir(dir)
        .unwrap()
        .map(|e| e.unwrap().path())
        .filter(|p| p.extension().is_some_and(|x| x == ext))
        .find(|p| {
            let b = std::fs::read(p).unwrap();
            holding.is_empty() || b.windows(holding.len()).any(|w| w == holding)
        })
        .expect("a file of the take")
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_recorded_room_is_written_as_moq_files_and_verifies_from_them() {
    timeout(Duration::from_secs(180), run()).await.expect("the test finishes in time");
}

async fn run() {
    let store = store_dir("files-store");
    let room = [0x61; 32];
    let publisher = SigningKey::from_bytes(&[0x62; 32]);
    let t = Tracks::new(ns(&[&hex::encode(room), &hex::encode(publisher.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
    let cfg = DvrConfig { path: store.clone(), window: vec![DvrWindow { prefix: vec!["*".into()], max_age_s: 900, max_bytes: 1 << 24 }], room_keys: vec![], takedown_keys: vec![] };
    let server = relay(|c| c.dvr = Some(cfg.clone())).await;
    let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _pub = publish_tracks(&p, &t, None).await;
    for g in 0..GROUPS {
        publish(&p, &publisher, &t, g).await;
    }
    let d = server.relay().dvr().expect("a DVR store").clone();
    assert!(until(STEP, || d.counts().groups_written >= 2 * GROUPS).await, "every group stored: {:?}", d.counts());

    // The recorder writes the room as files.
    let take = store_dir("files-take");
    let (d2, take2) = (d.clone(), take.clone());
    let written = tokio::task::spawn_blocking(move || d2.write_files(&[], &take2)).await.unwrap().expect("the files");
    assert_eq!(written.len(), 2, "a .moq file for the track and one for its receipts track");

    // Read back from the files alone: every object, verified against its group's receipt.
    assert_eq!(verify_files(&take, &t, &publisher), Ok((GROUPS * OBJECTS) as usize));
    let read = file::read_dir(&take).expect("the take reads");
    let media: Vec<&FileObject> = read.iter().filter(|o| o.name == t.media).collect();
    assert_eq!(media.len() as u64, GROUPS * OBJECTS);
    for o in &media {
        assert_eq!(o.payload, payload_of(o.group_id, o.object_id), "the payload as published");
        assert_eq!(o.forwarding, file::Forwarding::Subgroup(0));
        assert!(o.receive_time_ms.is_some(), "when the relay received its group");
    }

    // And what the files state is what a FETCH of the window brings.
    let (c, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    for (name, per_group) in [(&t.media, OBJECTS), (&t.receipts, 1)] {
        let (_ok, entries) = c.fetch(t.ns.clone(), name.clone(), Parameters::new()).await.expect("a FETCH of the window");
        let fetched: Vec<(u64, u64, Vec<u8>)> = entries.into_iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some((o.group_id, o.object_id, o.payload)) } else { None }).collect();
        let filed: Vec<(u64, u64, Vec<u8>)> = read.iter().filter(|o| &o.name == name).map(|o| (o.group_id, o.object_id, o.payload.clone())).collect();
        assert_eq!(fetched.len() as u64, GROUPS * per_group);
        assert_eq!(fetched, filed, "the files and the FETCH agree, byte for byte");
    }

    // ── Planted faults ──
    let marker = payload_of(2, 1);
    // A data file cut short.
    let cut = copy_dir(&take, "files-cut-dat");
    let dat = one_with(&cut, "dat", &marker[..12]);
    let bytes = std::fs::read(&dat).unwrap();
    std::fs::write(&dat, &bytes[..bytes.len() - 1]).unwrap();
    assert!(matches!(file::read_dir(&cut), Err(FileError::Refused(FileRefusal::DataTruncated))), "{:?}", file::read_dir(&cut).map(|o| o.len()));
    // A .moq file cut short.
    let cut = copy_dir(&take, "files-cut-moq");
    let moq = one_with(&cut, "moq", b"");
    let text = std::fs::read(&moq).unwrap();
    std::fs::write(&moq, &text[..text.len() - 7]).unwrap();
    assert!(matches!(file::read_dir(&cut), Err(FileError::Refused(FileRefusal::MetaMalformed))), "{:?}", file::read_dir(&cut).map(|o| o.len()));
    // A .moq file naming a data file outside its directory.
    let out = copy_dir(&take, "files-outside");
    let moq = one_with(&out, "moq", b"");
    let dat_name = one_with(&out, "dat", b"").file_name().unwrap().to_string_lossy().into_owned();
    let other_dat = std::fs::read_dir(&out).unwrap().map(|e| e.unwrap().file_name().to_string_lossy().into_owned()).filter(|n| n.ends_with(".dat")).collect::<Vec<_>>();
    let text = std::fs::read_to_string(&moq).unwrap();
    let named = other_dat.iter().find(|n| text.contains(n.as_str())).unwrap_or(&dat_name);
    std::fs::write(&moq, text.replace(named.as_str(), &format!("../{named}"))).unwrap();
    assert!(matches!(file::read_dir(&out), Err(FileError::Refused(FileRefusal::DataPathRefused))), "{:?}", file::read_dir(&out).map(|o| o.len()));
    // A payload changed in the data file: it reads, and the receipt refuses it.
    let changed = copy_dir(&take, "files-changed");
    let dat = one_with(&changed, "dat", &marker[..12]);
    let mut bytes = std::fs::read(&dat).unwrap();
    let at = bytes.windows(12).position(|w| w == &marker[..12]).unwrap();
    bytes[at + 20] ^= 1;
    std::fs::write(&dat, &bytes).unwrap();
    assert!(file::read_dir(&changed).is_ok(), "the files carry no check of their own");
    let refused = verify_files(&changed, &t, &publisher).unwrap_err();
    assert!(refused.starts_with("group 2 object 1"), "{refused}");

    c.close(0, "done");
    p.close(0, "done");
    server.shutdown().await;
}

/// One request sent with the program, as an operator sends it from a shell: its exit status
/// and its answer.
#[cfg(unix)]
async fn operator(socket: &Path, request: serde_json::Value) -> (i32, serde_json::Value) {
    let (socket, request) = (socket.to_path_buf(), request.to_string());
    let run = move || std::process::Command::new(env!("CARGO_BIN_EXE_wai-relay")).arg("--control").arg(&socket).arg(&request).output().expect("wai-relay runs");
    let o = timeout(STEP * 2, tokio::task::spawn_blocking(run)).await.expect("in time").expect("it ran");
    (o.status.code().expect("it exited"), serde_json::from_str(String::from_utf8_lossy(&o.stdout).trim()).unwrap_or(serde_json::Value::Null))
}

/// The operator's request (`files` on the control surface, sent with `wai-relay --control`):
/// two rooms recorded, one room asked for, written into a directory the relay makes, and read
/// back from the files alone. Then every track, into another. Planted, each refused with its
/// reason and nothing written: a directory that holds files, a relative path, a room the
/// store holds nothing of, and a relay that keeps no store.
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_operator_asks_for_a_room_as_files() {
    timeout(Duration::from_secs(180), async {
        let store = store_dir("files-op-store");
        let socket = std::env::temp_dir().join(format!("wai-relay-files-{}.sock", std::process::id()));
        let publisher = SigningKey::from_bytes(&[0x72; 32]);
        let pk = hex::encode(publisher.verifying_key().to_bytes());
        let (room_a, room_b) = (hex::encode([0x71; 32]), hex::encode([0x73; 32]));
        let ta = Tracks::new(ns(&[&room_a, &pk, "video"]), "0", 10, 5, 128);
        let tb = Tracks::new(ns(&[&room_b, &pk, "video"]), "0", 20, 15, 128);
        let cfg = DvrConfig { path: store.clone(), window: vec![DvrWindow { prefix: vec!["*".into()], max_age_s: 900, max_bytes: 1 << 24 }], room_keys: vec![], takedown_keys: vec![] };
        let server = relay(|c| {
            c.dvr = Some(cfg.clone());
            c.control.socket = Some(socket.clone());
        })
        .await;
        let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
        let _pubs = (publish_tracks(&p, &ta, None).await, publish_tracks(&p, &tb, None).await);
        for g in 0..GROUPS {
            publish(&p, &publisher, &ta, g).await;
            publish(&p, &publisher, &tb, g).await;
        }
        let d = server.relay().dvr().expect("a DVR store").clone();
        assert!(until(STEP, || d.counts().groups_written >= 4 * GROUPS).await, "every group stored: {:?}", d.counts());

        // One room, into a directory that does not exist yet.
        let take = store_dir("files-op-take").join("room-a");
        let (code, a) = operator(&socket, serde_json::json!({"op": "files", "dir": take, "room": room_a})).await;
        assert_eq!((code, &a["ok"], &a["tracks"], a["files"].as_array().map(Vec::len)), (0, &serde_json::json!(true), &serde_json::json!(2), Some(2)), "{a}");
        assert_eq!(verify_files(&take, &ta, &publisher), Ok((GROUPS * OBJECTS) as usize));
        let read = file::read_dir(&take).expect("the take reads");
        assert!(read.iter().all(|o| o.namespace == ta.ns), "nothing of the other room");
        for f in a["files"].as_array().unwrap() {
            let f = Path::new(f.as_str().expect("a path"));
            assert!(f.starts_with(&take) && f.is_file() && f.with_extension("dat").is_file(), "{}", f.display());
        }
        // Every track the store holds.
        let all = store_dir("files-op-all").join("all");
        let (code, a) = operator(&socket, serde_json::json!({"op": "files", "dir": all})).await;
        assert_eq!((code, &a["tracks"], a["files"].as_array().map(Vec::len)), (0, &serde_json::json!(4), Some(4)), "{a}");
        assert_eq!((verify_files(&all, &ta, &publisher), verify_files(&all, &tb, &publisher)), (Ok((GROUPS * OBJECTS) as usize), Ok((GROUPS * OBJECTS) as usize)));

        // ── Planted ──
        let listing = |dir: &Path| {
            let mut v: Vec<(std::ffi::OsString, Vec<u8>)> = std::fs::read_dir(dir).unwrap().map(|e| e.unwrap()).map(|e| (e.file_name(), std::fs::read(e.path()).unwrap())).collect();
            v.sort();
            v
        };
        let before = listing(&take);
        let refused = async |request: serde_json::Value, why: &str| {
            let (code, a) = operator(&socket, request.clone()).await;
            assert_eq!((code, &a["ok"]), (1, &serde_json::json!(false)), "{request}: {a}");
            assert!(a["error"].as_str().is_some_and(|e| e.starts_with(why)), "{request}: {a}");
        };
        // A directory that holds files: nothing of it is overwritten.
        refused(serde_json::json!({"op": "files", "dir": take, "room": room_b}), "dir_not_empty").await;
        assert_eq!(listing(&take), before, "the earlier take is as it was");
        // A relative path.
        refused(serde_json::json!({"op": "files", "dir": "take-here"}), "dir_not_absolute").await;
        assert!(!Path::new("take-here").exists());
        // A room the store holds nothing of: no directory is made for it.
        let none = store_dir("files-op-none").join("none");
        refused(serde_json::json!({"op": "files", "dir": none, "room": hex::encode([0x7f; 32])}), "no_tracks").await;
        assert!(!none.exists(), "nothing was written");
        // A room that is not one.
        refused(serde_json::json!({"op": "files", "dir": none, "room": "zz"}), "").await;
        p.close(0, "done");
        server.shutdown().await;

        // A relay that keeps no store.
        let socket2 = std::env::temp_dir().join(format!("wai-relay-files-none-{}.sock", std::process::id()));
        let bare = relay(|c| c.control.socket = Some(socket2.clone())).await;
        let (code, a) = operator(&socket2, serde_json::json!({"op": "files", "dir": none})).await;
        assert_eq!(code, 1, "{a}");
        assert!(a["error"].as_str().is_some_and(|e| e.starts_with("no_store")), "{a}");
        bare.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}