//! The signing proxy and the key server end to end, over HTTP on loopback, against a
//! licence server written here (a fixture: it answers every request with fixed bytes).
//!
//! It holds: a granted request is forwarded and its licence returned unchanged, with a
//! receipt that commits to the listed keys; a refused one is answered 403 and never reaches
//! the licence server; a replay is refused, also when the first request is still being
//! forwarded; an offline licence and its renewal; a licence server's refusal releases
//! nothing; a clear key is released only by a ticket the key server redeems once, so no key
//! byte passes through the proxy; the log is held by one process, survives a restart and a
//! torn last line; the whole log verifies under the reference verifier; and `GET /metrics` reads
//! strictly as the text format, counts the receipts by decision and delivery, tells no answer
//! from a licence server's 5xx, counts every route the routers declare by its pattern, states
//! what a restarted proxy recovered and a request at the licence server while it is there, names
//! no route, content, key, device or receipt, and is on the listener that is not public when the
//! proxy has one. A revocation is held only when it holds, is by a trusted authority and its instant
//! is after the proxy's clock; it is made durable, refuses every request under the entitlements it
//! names from its instant on, before anything is forwarded, and is held again after a restart. A
//! content's chain of lists is extended while the proxy runs, its next span's keys released from
//! then on, a list that does not extend it refused, and the chain read back after a restart. A
//! checkpoint is signed after every N receipts and on request, and the proxy goes on from it: a
//! replay is still refused, a licence before it is still renewed, across a restart, and an
//! earlier receipt is still served.

use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicU16, AtomicU64, Ordering};
use std::time::Duration;

use axum::Router;
use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use ed25519_dalek::SigningKey;
use tokio::sync::{Mutex, RwLock};
use wai::key_release::corpus::{content_key, kid};
use wai::key_release::{
    CLASS_ATTESTED, CommitmentList, Decision, Delivery, DeliveryKind, Entitlement, ListKey, LogCase, OutputStatement, Period, ProxyReceipt,
    Request, RequestKind, Revocation, RevocationScope, Rules, Scheme, b64url, cpix, hex, key_commitment, prove, sha256, unb64url, verify_log,
};
use wai_key_proxy::config::{KeyServerFile, Proxy, ProxyFile, RouteFile};
use wai_key_proxy::http::{Answer, Target, send};
use wai_key_proxy::{keyserver, proxy};

const SYSTEM: &str = "edef8ba9-79d6-4ace-a3c8-27dcd51d21ed";

fn key(n: u8) -> SigningKey {
    SigningKey::from_bytes(&[n; 32])
}

struct Upstream {
    hits: Mutex<Vec<(HeaderMap, Vec<u8>)>>,
    status: AtomicU16,
    delay_ms: AtomicU64,
    /// Held for writing, every request waits here for it to be let go.
    hold: RwLock<()>,
}

async fn licence_server(State(u): State<Arc<Upstream>>, headers: HeaderMap, body: Bytes) -> (StatusCode, Vec<u8>) {
    tokio::time::sleep(Duration::from_millis(u.delay_ms.load(Ordering::SeqCst))).await;
    drop(u.hold.read().await);
    u.hits.lock().await.push((headers, body.to_vec()));
    let mut licence = b"LICENCE:".to_vec();
    licence.extend_from_slice(&body);
    (StatusCode::from_u16(u.status.load(Ordering::SeqCst)).unwrap(), licence)
}

async fn spawn(router: Router) -> (SocketAddr, tokio::task::JoinHandle<()>) {
    let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let a = l.local_addr().unwrap();
    let h = tokio::spawn(async move {
        let _ = axum::serve(l, router).await;
    });
    (a, h)
}

struct World {
    dir: PathBuf,
    upstream: Arc<Upstream>,
    upstream_addr: SocketAddr,
    key_addr: SocketAddr,
    film_list: CommitmentList,
}

impl World {
    async fn new(name: &str) -> World {
        let dir = std::env::temp_dir().join(format!("wai-key-proxy-{}-{name}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let docs = wai::key_release::corpus::documents();
        let mut lists = Vec::new();
        for (doc, required) in [("film.cpix.xml", "svod:premium"), ("episode.cpix.xml", "svod:basic")] {
            std::fs::write(dir.join(doc), &docs[doc]).unwrap();
            let d = cpix::derive(docs[doc].as_bytes()).unwrap();
            let l = CommitmentList::seal(&key(1), "holder", &d.content_id, required, Some(d.sha256), 0, d.list_keys());
            std::fs::write(dir.join(format!("{}.wkl1", d.content_id)), l.to_bytes()).unwrap();
            lists.push(l);
        }
        std::fs::write(dir.join("proxy.key"), hex(&[3; 32])).unwrap();
        let upstream = Arc::new(Upstream { hits: Mutex::new(Vec::new()), status: AtomicU16::new(200), delay_ms: AtomicU64::new(0), hold: RwLock::new(()) });
        let (upstream_addr, _) =
            spawn(Router::new().route("/licence", axum::routing::post(licence_server)).with_state(upstream.clone())).await;
        let ks = KeyServerFile {
            listen: "127.0.0.1:0".parse().unwrap(),
            cpix: vec![dir.join("film.cpix.xml"), dir.join("episode.cpix.xml")],
            proxies: vec![hex(&key(3).verifying_key().to_bytes())],
            redeemed: dir.join("redeemed.txt"),
        };
        let (key_addr, _) = spawn(keyserver::router(keyserver::KeyApp::load(&ks).unwrap())).await;
        World { dir, upstream, upstream_addr, key_addr, film_list: lists.remove(0) }
    }

    fn config(&self) -> Proxy {
        Proxy::from_file(self.file()).unwrap()
    }

    fn file(&self) -> ProxyFile {
        let route = |name: &str, content: &str, delivery: &str, sys: Option<&str>, up: bool| RouteFile {
            name: name.into(),
            content_id: content.into(),
            delivery: delivery.into(),
            system_id: sys.map(String::from),
            upstream: up.then(|| format!("http://{}/licence", self.upstream_addr)),
            upstream_headers: BTreeMap::from([("x-licence-server-token".to_string(), "fixture".to_string())]),
            key_url: (!up).then(|| format!("http://{}/v1/key", self.key_addr)),
        };
        let mut routes = vec![
            route("film.a", "film-0001", "drm", Some(SYSTEM), true),
            route("film.ck", "film-0001", "clear-key", None, false),
            route("episode.hls", "episode-0002", "hls-key-uri", None, false),
        ];
        for r in &mut routes[1..] {
            r.upstream_headers.clear();
        }
        ProxyFile {
            listen: "127.0.0.1:0".parse().unwrap(),
            admin_listen: None,
            signer_key: self.dir.join("proxy.key"),
            signer_id: "https://keys.proxy.example".into(),
            authorities: vec![hex(&key(2).verifying_key().to_bytes())],
            attesters: vec![hex(&key(9).verifying_key().to_bytes())],
            lists: vec![self.dir.join("film-0001.wkl1"), self.dir.join("episode-0002.wkl1")],
            list_dir: Some(self.dir.join("lists.d")),
            routes,
            log: self.dir.join("receipts.jsonl"),
            revocations: Some(self.dir.join("revocations.txt")),
            checkpoint_every: 0,
            ticket_ttl_s: 60,
            upstream_timeout_ms: 60_000,
            max_body: 1 << 16,
        }
    }

    async fn start(&self) -> (Arc<proxy::App>, SocketAddr, SocketAddr, tokio::task::JoinHandle<()>, tokio::task::JoinHandle<()>) {
        // A restarted proxy waits for the old one's log to be let go, as a supervisor would.
        let mut tries = 0;
        let app = loop {
            match proxy::App::open(self.config()) {
                Ok(a) => break a,
                Err(e) if tries < 100 => {
                    tries += 1;
                    let _ = e;
                    tokio::time::sleep(Duration::from_millis(30)).await;
                }
                Err(e) => panic!("{e}"),
            }
        };
        let (a, h) = spawn(proxy::router(app.clone())).await;
        let (b, h2) = spawn(proxy::admin_router(app.clone())).await;
        (app, a, b, h, h2)
    }
}

fn entitlement(class: &str, offline_s: u64, renewals: u32) -> Entitlement {
    let now = wai_key_proxy::now();
    Entitlement::seal(&key(2), "authority", [9; 16], key(4).verifying_key().to_bytes(), class, now - 60, now + 3_600, offline_s, renewals)
}

struct Ask {
    route: &'static str,
    content: &'static str,
    delivery: Delivery,
    kids: Vec<[u8; 16]>,
    kind: RequestKind,
    renews: Option<[u8; 32]>,
    body: Vec<u8>,
}

/// The header fields of a request, signed by the device (key 4).
fn signed(a: &Ask, e: &Entitlement) -> Vec<(String, String)> {
    signed_by(a, e, 4, None)
}

/// The header fields of a request signed by the device whose key is `device`, presenting an
/// output statement (§14.3) when one is given.
fn signed_by(a: &Ask, e: &Entitlement, device: u8, output: Option<Vec<u8>>) -> Vec<(String, String)> {
    let mut kids = a.kids.clone();
    kids.sort_unstable();
    let t = wai_key_proxy::now();
    let req = Request {
        content_id: a.content.into(),
        delivery: a.delivery,
        kids: kids.clone(),
        kind: a.kind,
        renews: a.renews,
        request_sha256: sha256(&a.body),
        request_t: t,
        output: output.clone(),
    };
    let eb = e.to_bytes();
    let mut h = vec![
        ("wai-key-ids".to_string(), kids.iter().map(|k| hex(k)).collect::<Vec<_>>().join(",")),
        ("wai-key-kind".into(), a.kind.label().into()),
        ("wai-key-time".into(), t.to_string()),
        ("wai-key-entitlement".into(), b64url(&eb)),
        ("wai-key-proof".into(), b64url(&prove(&key(device), &eb, &req))),
        ("content-type".into(), "application/octet-stream".into()),
    ];
    if let Some(r) = a.renews {
        h.push(("wai-key-renews".into(), hex(&r)));
    }
    if let Some(o) = output {
        h.push(("wai-key-output".into(), b64url(&o)));
    }
    h
}

async fn call(addr: SocketAddr, method: &str, path: &str, headers: &[(String, String)], body: Vec<u8>) -> Answer {
    let t = Target { authority: addr.to_string(), path: path.into() };
    send(method, &t, headers, body, Duration::from_secs(90)).await.expect("an answer")
}

async fn receipt_of(addr: SocketAddr, a: &Answer) -> ProxyReceipt {
    let id = a.header("wai-key-receipt").expect("every decided request names its receipt");
    let got = call(addr, "GET", &format!("/v1/receipts/{id}"), &[], vec![]).await;
    assert_eq!(got.status, 200);
    let r = ProxyReceipt::from_json(&serde_json::from_slice(&got.body).unwrap()).unwrap();
    assert_eq!(hex(&r.identity()), id);
    assert_eq!(r.check(), Ok(()));
    r
}

fn film_drm(kind: RequestKind, renews: Option<[u8; 32]>, body: &str) -> Ask {
    Ask {
        route: "film.a",
        content: "film-0001",
        delivery: Delivery { kind: DeliveryKind::Drm, system_id: cpix::uuid(SYSTEM) },
        kids: vec![kid("film/video"), kid("film/audio")],
        kind,
        renews,
        body: body.as_bytes().to_vec(),
    }
}

fn path(a: &Ask) -> String {
    format!("/v1/licence/{}", a.route)
}

/// The whole log, from the listener that is not public, under the reference verifier.
async fn verify_whole_log(admin: SocketAddr, w: &World) -> usize {
    let log = call(admin, "GET", "/v1/log", &[], vec![]).await;
    assert_eq!(log.status, 200);
    let receipts: Vec<serde_json::Value> =
        String::from_utf8(log.body).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect();
    let lists = vec![std::fs::read(w.dir.join("film-0001.wkl1")).unwrap(), std::fs::read(w.dir.join("episode-0002.wkl1")).unwrap()];
    let cpix = vec![(0, std::fs::read(w.dir.join("film.cpix.xml")).unwrap()), (1, std::fs::read(w.dir.join("episode.cpix.xml")).unwrap())];
    let held = call(admin, "GET", "/v1/revocations", &[], vec![]).await;
    assert_eq!(held.status, 200);
    let revocations: Vec<Vec<u8>> = String::from_utf8(held.body).unwrap().lines().map(|l| wai::key_release::unhex_vec(l).unwrap()).collect();
    let case = LogCase {
        authorities: &[key(2).verifying_key().to_bytes()],
        attesters: &[key(9).verifying_key().to_bytes()],
        proxies: &[key(3).verifying_key().to_bytes()],
        lists: &lists,
        cpix: &cpix,
        receipts: &receipts,
        tickets: &[],
        revocations: &revocations,
    };
    verify_log(&case).unwrap_or_else(|r| panic!("the proxy's own log is refused: {} at {} {}", r.reason.code(), r.place, r.index)).len()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn releases_refusals_tickets_and_the_log() {
    let w = World::new("main").await;
    let (_app, addr, admin, _, _) = w.start().await;
    let premium = entitlement("svod:premium", 600, 1);

    // A granted request is forwarded, and its licence returned unchanged.
    let ask = film_drm(RequestKind::Streaming, None, "challenge one");
    let headers = signed(&ask, &premium);
    let a = call(addr, "POST", &path(&ask), &headers, ask.body.clone()).await;
    assert_eq!(a.status, 200);
    assert_eq!(a.body, b"LICENCE:challenge one");
    let r = receipt_of(addr, &a).await;
    assert_eq!((r.decision, r.released, r.upstream_status), (Decision::Granted, true, Some(200)));
    assert_eq!(r.response_sha256, Some(sha256(b"LICENCE:challenge one")));
    for k in &r.keys {
        let label = if k.kid == kid("film/video") { "film/video" } else { "film/audio" };
        assert_eq!(k.key_commit, Some(key_commitment(&content_key(label, 16))), "the commitment is the key's");
    }
    {
        let hits = w.upstream.hits.lock().await;
        assert_eq!(hits.len(), 1);
        let (h, body) = &hits[0];
        assert_eq!(body, b"challenge one", "the body is forwarded unchanged");
        assert!(h.get("wai-key-ids").is_some() && h.get("x-licence-server-token").is_some());
        assert!(h.get("wai-key-entitlement").is_none() && h.get("wai-key-proof").is_none(), "the entitlement stays at the proxy");
    }

    // The same request again: refused, and not forwarded.
    let again = call(addr, "POST", &path(&ask), &headers, ask.body.clone()).await;
    assert_eq!(again.status, 403);
    assert_eq!(receipt_of(addr, &again).await.decision, Decision::DeniedReplayed);
    // The wrong class: refused, and not forwarded.
    let basic = entitlement("svod:basic", 0, 0);
    let ask2 = film_drm(RequestKind::Streaming, None, "challenge two");
    let refused = call(addr, "POST", &path(&ask2), &signed(&ask2, &basic), ask2.body.clone()).await;
    assert_eq!(refused.status, 403);
    let rr: ProxyReceipt = ProxyReceipt::from_json(&serde_json::from_slice(&refused.body).unwrap()).unwrap();
    assert_eq!((rr.decision, rr.released, rr.upstream_status), (Decision::DeniedNoEntitlement, false, None));
    assert_eq!(w.upstream.hits.lock().await.len(), 1, "no refused request reaches the licence server");

    // A replay sent while the first request is still at the licence server waits, and is
    // refused after it.
    w.upstream.delay_ms.store(300, Ordering::SeqCst);
    let ask3 = film_drm(RequestKind::Streaming, None, "challenge three");
    let (h3, p3) = (signed(&ask3, &premium), path(&ask3));
    let (x, y) = tokio::join!(call(addr, "POST", &p3, &h3, ask3.body.clone()), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        call(addr, "POST", &p3, &h3, ask3.body.clone()).await
    });
    w.upstream.delay_ms.store(0, Ordering::SeqCst);
    assert_eq!((x.status, y.status), (200, 403));
    assert_eq!(receipt_of(addr, &y).await.decision, Decision::DeniedReplayed);

    // An offline licence and its renewal; the licence server is told the expiry.
    let off = film_drm(RequestKind::Offline, None, "offline");
    let o = call(addr, "POST", &path(&off), &signed(&off, &premium), off.body.clone()).await;
    assert_eq!(o.status, 200);
    let or = receipt_of(addr, &o).await;
    let expiry = or.licence_expiry.expect("an offline licence has an expiry");
    assert_eq!(expiry, (or.issued + 600).min(premium.not_after));
    let told = w.upstream.hits.lock().await.last().unwrap().0.get("wai-key-licence-expiry").map(|v| v.to_str().unwrap().to_string());
    assert_eq!(told, Some(expiry.to_string()));
    let ren = film_drm(RequestKind::Renewal, Some(or.identity()), "renewal");
    let n = call(addr, "POST", &path(&ren), &signed(&ren, &premium), ren.body.clone()).await;
    assert_eq!(n.status, 200);
    assert_eq!(receipt_of(addr, &n).await.renewal, Some(1));

    // The licence server refuses: nothing is released.
    w.upstream.status.store(403, Ordering::SeqCst);
    let ask4 = film_drm(RequestKind::Streaming, None, "challenge four");
    let up = call(addr, "POST", &path(&ask4), &signed(&ask4, &premium), ask4.body.clone()).await;
    w.upstream.status.store(200, Ordering::SeqCst);
    assert_eq!(up.status, 403);
    let ur = receipt_of(addr, &up).await;
    assert_eq!((ur.decision, ur.released, ur.upstream_status), (Decision::Granted, false, Some(403)));
    assert!(ur.keys.iter().all(|k| k.key_commit.is_none()));

    // An HLS key file: a ticket, redeemed once at the key server.
    let basic_hls = entitlement("svod:basic", 0, 0);
    let hls = Ask {
        route: "episode.hls",
        content: "episode-0002",
        delivery: Delivery { kind: DeliveryKind::HlsKeyUri, system_id: None },
        kids: vec![kid("episode/main")],
        kind: RequestKind::Streaming,
        renews: None,
        body: vec![],
    };
    let k = call(addr, "GET", &path(&hls), &signed(&hls, &basic_hls), vec![]).await;
    assert_eq!(k.status, 307);
    let kr = receipt_of(addr, &k).await;
    assert!(kr.released && kr.upstream_status.is_none());
    let loc = k.header("location").unwrap().to_string();
    let key_path = loc.split_once(&w.key_addr.to_string()).unwrap().1.to_string();
    let got = call(w.key_addr, "GET", &key_path, &[], vec![]).await;
    assert_eq!(got.status, 200);
    assert_eq!(got.body, content_key("episode/main", 16), "the key server returns the key file");
    assert_eq!(call(w.key_addr, "GET", &key_path, &[], vec![]).await.status, 409, "a ticket is redeemed once");
    let mut forged = loc.split_once("wai-ticket=").unwrap().1.to_string();
    let last = forged.pop().unwrap();
    forged.push(if last == 'A' { 'B' } else { 'A' });
    assert_eq!(call(w.key_addr, "GET", &format!("/v1/key?wai-ticket={forged}"), &[], vec![]).await.status, 403);

    // Clear Key: the licence request reaches the key server by the redirect; a JWK set back.
    let ck = Ask {
        route: "film.ck",
        content: "film-0001",
        delivery: Delivery { kind: DeliveryKind::ClearKey, system_id: None },
        kids: vec![kid("film/video")],
        kind: RequestKind::Offline,
        renews: None,
        body: format!("{{\"kids\":[\"{}\"],\"type\":\"persistent-license\"}}", b64url(&kid("film/video"))).into_bytes(),
    };
    let c = call(addr, "POST", &path(&ck), &signed(&ck, &premium), ck.body.clone()).await;
    assert_eq!(c.status, 307);
    let loc = c.header("location").unwrap().to_string();
    let jwk = call(w.key_addr, "POST", loc.split_once(&w.key_addr.to_string()).unwrap().1, &[], ck.body.clone()).await;
    assert_eq!(jwk.status, 200);
    let j: serde_json::Value = serde_json::from_slice(&jwk.body).unwrap();
    assert_eq!(j["type"], "persistent-license");
    assert_eq!(unb64url(j["keys"][0]["k"].as_str().unwrap()), Some(content_key("film/video", 16)));

    // Requests that name nothing to decide have no receipt.
    assert_eq!(call(addr, "POST", "/v1/licence/nowhere", &headers, vec![]).await.status, 404);
    assert_eq!(call(addr, "POST", &path(&ask), &[], vec![]).await.status, 400);

    // The proxy's counters: every receipt by decision and delivery, the licence server's answers,
    // the requests by route; and no route name, content, key, device or receipt among them.
    let m = call(addr, "GET", "/metrics", &[], vec![]).await;
    assert_eq!((m.status, m.content_type.as_deref()), (200, Some(wai_metrics::CONTENT_TYPE)));
    let text = String::from_utf8(m.body).unwrap();
    let mf = wai_metrics::read::parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
    let value = |name: &str, labels: &[(&str, &str)]| wai_metrics::read::value(&mf, name, labels);
    let receipts = |decision: &str, delivery: &str| value("wai_key_proxy_receipts_total", &[("decision", decision), ("delivery", delivery)]);
    assert_eq!(receipts("Granted", "drm"), Some(5.0));
    assert_eq!(receipts("DeniedReplayed", "drm"), Some(2.0));
    assert_eq!(receipts("DeniedNoEntitlement", "drm"), Some(1.0));
    assert_eq!((receipts("Granted", "hls-key-uri"), receipts("Granted", "clear-key")), (Some(1.0), Some(1.0)));
    assert_eq!(wai_metrics::read::sum(&mf, "wai_key_proxy_receipts_total", &[]), 10.0);
    assert_eq!((value("wai_key_proxy_receipts_held", &[]), value("wai_key_proxy_forwarding", &[])), (Some(10.0), Some(0.0)));
    assert_eq!(value("wai_key_proxy_proofs_in_flight", &[]), Some(0.0));
    assert_eq!(value("wai_key_proxy_upstream_total", &[("class", "2xx")]), Some(4.0));
    assert_eq!(value("wai_key_proxy_upstream_total", &[("class", "4xx")]), Some(1.0));
    let requests = |method: &str, code: &str| value("wai_key_proxy_requests_total", &[("route", "/v1/licence/{route}"), ("method", method), ("code", code)]);
    assert_eq!((requests("POST", "200"), requests("POST", "403"), requests("GET", "307"), requests("POST", "307")), (Some(4.0), Some(4.0), Some(1.0), Some(1.0)));
    assert_eq!((requests("POST", "404"), requests("POST", "400")), (Some(1.0), Some(1.0)));
    for l in wai_metrics::read::label_values(&mf) {
        assert!(!wai_metrics::read::identifier_like(l), "a label value {l:?}");
    }
    let device = hex(&key(4).verifying_key().to_bytes());
    let receipt = a.header("wai-key-receipt").unwrap().to_string();
    for id in [device.as_str(), receipt.as_str(), "film", "episode", "nowhere", "keys.proxy.example", "authority", "svod", "127.0.0.1"] {
        assert!(!text.contains(id), "{id} is in the metrics");
    }

    // No key byte, in any spelling, is in the proxy's log.
    let log = std::fs::read(w.dir.join("receipts.jsonl")).unwrap();
    for label in ["film/video", "film/audio", "episode/main"] {
        let k = content_key(label, 16);
        let text = String::from_utf8_lossy(&log);
        assert!(!text.contains(&hex(&k)) && !text.contains(&b64url(&k)) && !text.contains(&wai::key_release::b64(&k)));
        assert!(!log.windows(16).any(|w| w == k.as_slice()));
    }
    let n = verify_whole_log(admin, &w).await;
    assert_eq!(n, 10, "every decided request is in the log");
    assert!(w.film_list.keys.len() == 2);
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// The gauge `name`, as `GET /metrics` at `addr` states it.
async fn gauge(addr: SocketAddr, name: &str) -> f64 {
    let m = call(addr, "GET", "/metrics", &[], vec![]).await;
    assert_eq!(m.status, 200);
    let text = String::from_utf8(m.body).unwrap();
    let mf = wai_metrics::read::parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
    wai_metrics::read::value(&mf, name, &[]).expect("the gauge")
}

/// The requests at the licence server now.
async fn forwarding(addr: SocketAddr) -> f64 {
    gauge(addr, "wai_key_proxy_forwarding").await
}

/// A granted request held at the licence server is stated while it is there, and not after it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_request_at_the_licence_server_is_stated_while_it_is_there() {
    let w = World::new("forwarding").await;
    let (_app, addr, _, _, _) = w.start().await;
    assert_eq!(forwarding(addr).await, 0.0);
    // Through a front, the scrape is a path the proxy does not have.
    let through = [("x-real-ip".to_string(), "198.51.100.9".to_string())];
    assert_eq!(call(addr, "GET", "/metrics", &through, vec![]).await.status, 404);
    let answer = call(addr, "POST", "/metrics", &through, vec![]).await;
    assert_eq!((answer.status, answer.body.is_empty()), (404, true), "by any method, as a path the proxy does not have");
    let premium = entitlement("svod:premium", 600, 1);
    let ask = film_drm(RequestKind::Streaming, None, "a held challenge");
    let (headers, to, body) = (signed(&ask, &premium), path(&ask), ask.body.clone());
    let held = w.upstream.hold.write().await;
    let request = tokio::spawn(async move { call(addr, "POST", &to, &headers, body).await });
    // Within the proxy's `upstream_timeout_ms`, past which it gives the request up.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
    while forwarding(addr).await != 1.0 {
        assert!(tokio::time::Instant::now() < deadline, "the request reaches the licence server");
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert_eq!(gauge(addr, "wai_key_proxy_proofs_in_flight").await, 1.0);
    drop(held);
    assert_eq!(request.await.unwrap().status, 200);
    assert_eq!(forwarding(addr).await, 0.0, "and is not stated once it is answered");
    assert_eq!(gauge(addr, "wai_key_proxy_proofs_in_flight").await, 0.0);
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// A client that goes away takes its request with it: it is no longer at the licence server, and
/// `GET /metrics` no longer states it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_request_whose_client_goes_away_is_no_longer_stated() {
    let w = World::new("gone").await;
    let (_app, addr, _, _, _) = w.start().await;
    let premium = entitlement("svod:premium", 600, 1);
    let ask = film_drm(RequestKind::Streaming, None, "a challenge given up");
    let (headers, to, body) = (signed(&ask, &premium), path(&ask), ask.body.clone());
    let held = w.upstream.hold.write().await;
    let request = tokio::spawn(async move { call(addr, "POST", &to, &headers, body).await });
    // Within the proxy's `upstream_timeout_ms`, past which it gives the request up itself.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
    while forwarding(addr).await != 1.0 {
        assert!(tokio::time::Instant::now() < deadline, "the request reaches the licence server");
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    request.abort();
    let _ = request.await;
    while forwarding(addr).await != 0.0 {
        assert!(tokio::time::Instant::now() < deadline, "the request goes with its client");
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    drop(held);
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// The licence server's answers by class, as `GET /metrics` at `addr` states them.
async fn upstream(addr: SocketAddr, class: &str) -> Option<f64> {
    let m = call(addr, "GET", "/metrics", &[], vec![]).await;
    let text = String::from_utf8(m.body).unwrap();
    let mf = wai_metrics::read::parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
    wai_metrics::read::value(&mf, "wai_key_proxy_upstream_total", &[("class", class)])
}

/// A licence server that gives no answer in time is told from one that answers 5xx: an operator
/// reads the class to tell which of the two it has.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn no_answer_from_the_licence_server_is_not_a_5xx() {
    let w = World::new("no-answer").await;
    let mut cfg = w.config();
    cfg.upstream_timeout_ms = 200;
    let (addr, _) = spawn(proxy::router(proxy::App::open(cfg).unwrap())).await;
    let premium = entitlement("svod:premium", 600, 1);

    let ask = film_drm(RequestKind::Streaming, None, "a challenge held past the time limit");
    let held = w.upstream.hold.write().await;
    assert_eq!(call(addr, "POST", &path(&ask), &signed(&ask, &premium), ask.body.clone()).await.status, 502);
    drop(held);
    assert_eq!((upstream(addr, "none").await, upstream(addr, "5xx").await), (Some(1.0), None));

    w.upstream.status.store(503, Ordering::SeqCst);
    let ask = film_drm(RequestKind::Streaming, None, "a challenge answered 503");
    assert_eq!(call(addr, "POST", &path(&ask), &signed(&ask, &premium), ask.body.clone()).await.status, 503);
    assert_eq!((upstream(addr, "none").await, upstream(addr, "5xx").await), (Some(1.0), Some(1.0)));
    let _ = std::fs::remove_dir_all(&w.dir);

    // A licence server that is not there: the connection is refused, and there is no answer.
    let w = World::new("no-server").await;
    let gone = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap().local_addr().unwrap();
    let mut cfg = w.config();
    cfg.routes.get_mut("film.a").unwrap().upstream = Some(Target { authority: gone.to_string(), path: "/licence".into() });
    let (addr, _) = spawn(proxy::router(proxy::App::open(cfg).unwrap())).await;
    let ask = film_drm(RequestKind::Streaming, None, "a challenge with nobody to answer it");
    assert_eq!(call(addr, "POST", &path(&ask), &signed(&ask, &premium), ask.body.clone()).await.status, 502);
    assert_eq!((upstream(addr, "none").await, upstream(addr, "5xx").await), (Some(1.0), None));
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// Every route the routers declare, asked once, is counted by its pattern: a route left out of
/// `route_of` would be counted as `other`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn every_route_the_routers_declare_is_counted_by_its_pattern() {
    use wai_metrics::read::{declared_routes, parse, path_of, sum};
    let w = World::new("routes").await;
    let (_app, addr, _, _, _) = w.start().await;
    let routes = declared_routes(include_str!("../src/proxy.rs"));
    assert!(routes.len() >= 7, "{routes:?}");
    for r in &routes {
        call(addr, "GET", &path_of(r), &[], vec![]).await;
    }
    let head = call(addr, "HEAD", "/metrics", &[], vec![]).await;
    assert_eq!((head.status, head.content_type.as_deref(), head.body.is_empty()), (200, Some(wai_metrics::CONTENT_TYPE), true));
    let text = String::from_utf8(call(addr, "GET", "/metrics", &[], vec![]).await.body).unwrap();
    let mf = parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
    for r in &routes {
        assert!(sum(&mf, "wai_key_proxy_requests_total", &[("route", *r)]) >= 1.0, "{r} is not counted by its pattern\n{text}");
    }
    assert_eq!(sum(&mf, "wai_key_proxy_requests_total", &[("route", "other")]), 0.0, "{text}");
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// A proxy with `admin_listen`: `GET /metrics` is there, counting what the public listener
/// answered, and the public listener does not have it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn metrics_are_on_the_listener_that_is_not_public_when_there_is_one() {
    let w = World::new("admin-metrics").await;
    let mut cfg = w.config();
    cfg.admin_listen = Some("127.0.0.1:0".parse().unwrap());
    let app = proxy::App::open(cfg).unwrap();
    let (public, _) = spawn(proxy::router(app.clone())).await;
    let (admin, _) = spawn(proxy::admin_router(app.clone())).await;

    assert_eq!(call(public, "GET", "/v1/health", &[], vec![]).await.status, 200);
    assert_eq!(call(public, "GET", "/metrics", &[], vec![]).await.status, 404);
    assert_eq!(call(admin, "GET", "/v1/log", &[], vec![]).await.status, 200);
    let m = call(admin, "GET", "/metrics", &[], vec![]).await;
    assert_eq!((m.status, m.content_type.as_deref()), (200, Some(wai_metrics::CONTENT_TYPE)));
    let text = String::from_utf8(m.body).unwrap();
    let mf = wai_metrics::read::parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
    let requests = |route: &str, code: &str| wai_metrics::read::value(&mf, "wai_key_proxy_requests_total", &[("route", route), ("method", "GET"), ("code", code)]);
    assert_eq!((requests("/v1/health", "200"), requests("/metrics", "404"), requests("/v1/log", "200")), (Some(1.0), Some(1.0), Some(1.0)));
    // That listener is not public: it answers whatever fields a request carries.
    let through = [("x-forwarded-for".to_string(), "198.51.100.9".to_string())];
    assert_eq!(call(admin, "GET", "/metrics", &through, vec![]).await.status, 200);
    let _ = std::fs::remove_dir_all(&w.dir);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_log_is_held_once_and_survives_a_restart() {
    let w = World::new("restart").await;
    let premium = entitlement("svod:premium", 600, 1);
    let first = film_drm(RequestKind::Streaming, None, "before the restart");
    let first_headers = signed(&first, &premium);
    let off = film_drm(RequestKind::Offline, None, "offline before the restart");
    let off_id = {
        let (app, addr, _, h, h2) = w.start().await;
        assert_eq!(call(addr, "POST", &path(&first), &first_headers, first.body.clone()).await.status, 200);
        let o = call(addr, "POST", &path(&off), &signed(&off, &premium), off.body.clone()).await;
        let id = receipt_of(addr, &o).await.identity();
        assert!(proxy::App::open(w.config()).is_err(), "a second process does not take the log");
        h.abort();
        h2.abort();
        let _ = h.await;
        let _ = h2.await;
        drop(app);
        id
    };
    // A write that never finished, as a crash would leave it.
    use std::io::Write;
    std::fs::OpenOptions::new().append(true).open(w.dir.join("receipts.jsonl")).unwrap().write_all(b"{\"form\":2,\"cont").unwrap();
    let (app, addr, admin, _, _) = w.start().await;
    assert_eq!(app.log_len().await, 2, "the log is read back, its unfinished line cut");
    assert_eq!(gauge(addr, "wai_key_proxy_receipts_held").await, 2.0, "and stated before any request adds to it");
    let again = call(addr, "POST", &path(&first), &first_headers, first.body.clone()).await;
    assert_eq!((again.status, receipt_of(addr, &again).await.decision), (403, Decision::DeniedReplayed), "a replay across a restart");
    let ren = film_drm(RequestKind::Renewal, Some(off_id), "renewal after the restart");
    let n = call(addr, "POST", &path(&ren), &signed(&ren, &premium), ren.body.clone()).await;
    assert_eq!((n.status, receipt_of(addr, &n).await.renewal), (200, Some(1)), "a renewal of a licence released before the restart");
    assert_eq!(verify_whole_log(admin, &w).await, 4);
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// An entitlement of version 2 for the device whose key is `device`, in its own session.
fn ruled(device: u8, rules: Rules) -> Entitlement {
    let now = wai_key_proxy::now();
    Entitlement::seal(&key(2), "authority", [device; 16], key(device).verifying_key().to_bytes(), "svod:premium", now - 60, now + 3_600, 0, 0).with_rules(rules, &key(2))
}

/// The entitlement rules through the proxy (key-release §14): a limit of one session at once,
/// held by the session that asks again and refused to a second one, whose receipt names the
/// release it counted; two sessions of one group at the licence server together, of which
/// exactly one is released; an output level released on an attested statement, refused on a
/// lower one, and undecided, with nothing released, where no statement establishes one; and
/// the whole log, form 3 among it, under the reference verifier.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn stream_limit_and_output_level() {
    let w = World::new("rules").await;
    let (_app, addr, admin, _, _) = w.start().await;
    let one = Rules { group: [7; 16], streams: 1, window_s: 600, output_min: 0, output_class: 0 };
    let (a, b) = (ruled(4, one), ruled(5, one));
    let ask = |body: &str| film_drm(RequestKind::Streaming, None, body);

    // The first session is released; the second is refused and names that release.
    let q = ask("session a");
    let first = call(addr, "POST", &path(&q), &signed_by(&q, &a, 4, None), q.body.clone()).await;
    let r1 = receipt_of(addr, &first).await;
    assert_eq!((first.status, r1.decision, r1.form()), (200, Decision::Granted, 3));
    let q = ask("session b");
    let second = call(addr, "POST", &path(&q), &signed_by(&q, &b, 5, None), q.body.clone()).await;
    let r2 = receipt_of(addr, &second).await;
    assert_eq!((second.status, r2.decision, r2.released), (403, Decision::DeniedStreamLimit, false));
    assert_eq!(r2.rules.as_ref().map(|x| x.leases.clone()), Some(vec![r1.identity()]), "the refusal names the release it counted");
    assert_eq!(w.upstream.hits.lock().await.len(), 1, "a request past the limit never reaches the licence server");
    // The session that holds the place asks again and keeps it.
    let q = ask("session a again");
    let again = call(addr, "POST", &path(&q), &signed_by(&q, &a, 4, None), q.body.clone()).await;
    assert_eq!((again.status, receipt_of(addr, &again).await.decision), (200, Decision::Granted));

    // Two sessions of another group at the licence server together: one release, one refusal.
    let other = Rules { group: [8; 16], ..one };
    let (c, d) = (ruled(4, other), ruled(5, other));
    let (qc, qd) = (ask("session c"), ask("session d"));
    let (hc, hd) = (signed_by(&qc, &c, 4, None), signed_by(&qd, &d, 5, None));
    let held = w.upstream.hold.write().await;
    let tc = tokio::spawn({
        let (p, body) = (path(&qc), qc.body.clone());
        async move { call(addr, "POST", &p, &hc, body).await }
    });
    let td = tokio::spawn({
        let (p, body) = (path(&qd), qd.body.clone());
        async move { call(addr, "POST", &p, &hd, body).await }
    });
    tokio::time::sleep(Duration::from_millis(300)).await;
    drop(held);
    let mut statuses = vec![tc.await.unwrap().status, td.await.unwrap().status];
    statuses.sort_unstable();
    assert_eq!(statuses, vec![200, 403], "of two sessions under a limit of one, asked together, one is released");

    // An output level: 2.2, attested by key 9.
    let hd22 = ruled(4, Rules { group: [0; 16], streams: 0, window_s: 0, output_min: 8, output_class: CLASS_ATTESTED });
    let now = wai_key_proxy::now();
    let state = |signer: u8, level: u8| Some(OutputStatement::seal(&key(signer), key(4).verifying_key().to_bytes(), level, CLASS_ATTESTED, now - 60, now + 600).to_bytes());
    let hits = w.upstream.hits.lock().await.len();
    for (body, output, status, decision) in [
        ("no statement", None, 403, Decision::UndecidedOutput),
        ("an attester the proxy does not trust", state(6, 8), 403, Decision::UndecidedOutput),
        ("a level below", state(9, 5), 403, Decision::DeniedOutputUnmet),
        ("the level required", state(9, 8), 200, Decision::Granted),
    ] {
        let q = ask(body);
        let got = call(addr, "POST", &path(&q), &signed_by(&q, &hd22, 4, output.clone()), q.body.clone()).await;
        let r = receipt_of(addr, &got).await;
        assert_eq!((got.status, r.decision, r.released), (status, decision, status == 200), "{body}");
        assert_eq!(r.rules.as_ref().and_then(|x| x.output.clone()), output, "{body}: the receipt records the statement presented");
    }
    assert_eq!(w.upstream.hits.lock().await.len(), hits + 1, "an undecided or refused request never reaches the licence server");
    assert_eq!(verify_whole_log(admin, &w).await, 9);
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// A revocation through the proxy (key-release §16): one by an authority the proxy does not
/// trust, one that does not hold and one whose instant is not after its clock are refused; one
/// held refuses the session it names from its instant on (`DeniedRevoked`), never forwarded,
/// leaves another session alone, and is held again after a restart; the whole log verifies with
/// the revocations the proxy states.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_revocation_refuses_from_its_instant() {
    let w = World::new("revocation").await;
    let premium = entitlement("svod:premium", 0, 0);
    let now = wai_key_proxy::now();
    let other = Entitlement::seal(&key(2), "authority", [10; 16], key(4).verifying_key().to_bytes(), "svod:premium", now - 60, now + 3_600, 0, 0);
    let revocation = |by: u8, from: u64| Revocation::seal(&key(by), "authority", RevocationScope::Session, [9; 16], from).to_bytes();
    // The instant a few seconds ahead of the clock when it is posted, however long a loaded host
    // took to get there.
    let ahead = || wai_key_proxy::now() + 3;
    {
        let (app, addr, admin, h, h2) = w.start().await;
        let q = film_drm(RequestKind::Streaming, None, "before the revocation");
        assert_eq!(call(addr, "POST", &path(&q), &signed(&q, &premium), q.body.clone()).await.status, 200);
        assert_eq!(call(admin, "POST", "/v1/revocations", &[], revocation(6, ahead())).await.status, 422, "an authority the proxy does not trust");
        let mut forged = revocation(2, ahead());
        let n = forged.len();
        forged[n - 1] ^= 1;
        assert_eq!(call(admin, "POST", "/v1/revocations", &[], forged).await.status, 422, "a revocation that does not hold");
        assert_eq!(call(admin, "POST", "/v1/revocations", &[], revocation(2, now.saturating_sub(1))).await.status, 409, "an instant not after the clock");
        let from = ahead();
        assert_eq!(call(admin, "POST", "/v1/revocations", &[], revocation(2, from)).await.status, 201);
        while wai_key_proxy::now() < from {
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        let hits = w.upstream.hits.lock().await.len();
        let q = film_drm(RequestKind::Streaming, None, "after the revocation");
        let after = call(addr, "POST", &path(&q), &signed(&q, &premium), q.body.clone()).await;
        let r = receipt_of(addr, &after).await;
        assert_eq!((after.status, r.decision, r.released), (403, Decision::DeniedRevoked, false));
        assert_eq!(w.upstream.hits.lock().await.len(), hits, "a revoked request never reaches the licence server");
        let q = film_drm(RequestKind::Streaming, None, "another session");
        assert_eq!(call(addr, "POST", &path(&q), &signed(&q, &other), q.body.clone()).await.status, 200, "another session is not named");
        assert_eq!(verify_whole_log(admin, &w).await, 3);
        h.abort();
        h2.abort();
        let _ = h.await;
        let _ = h2.await;
        drop(app);
    }
    let (_app, addr, admin, _, _) = w.start().await;
    let q = film_drm(RequestKind::Streaming, None, "after the restart");
    let again = call(addr, "POST", &path(&q), &signed(&q, &premium), q.body.clone()).await;
    assert_eq!((again.status, receipt_of(addr, &again).await.decision), (403, Decision::DeniedRevoked), "the revocation is held again after a restart");
    assert_eq!(verify_whole_log(admin, &w).await, 4);
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// A content whose keys come in spans (key-release §4.6) through the proxy: the next span's key
/// is unknown until a list extending the chain is given on the listener that is not public, and
/// released after, naming that list; a list that does not extend the chain's last is refused; a
/// key of each span is released together; the chain is read back from `list_dir` after a restart;
/// and the whole log verifies with the chain.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_chain_of_lists_is_extended_while_the_proxy_runs() {
    let w = World::new("chain").await;
    let span = |keys: &[(&str, u32)]| -> Vec<ListKey> {
        keys.iter()
            .map(|(label, i)| ListKey { kid: kid(label), scheme: Scheme::Cenc, key_commit: key_commitment(&content_key(label, 16)), period: Some(Period { index: Some(*i), window: None }) })
            .collect()
    };
    let first = CommitmentList::seal(&key(1), "holder", "live-0010", "svod:premium", None, 0, span(&[("live/p1", 1), ("live/p2", 2)]));
    let next = CommitmentList::seal_extending(&key(1), "holder", &first, None, 0, span(&[("live/p3", 3), ("live/p4", 4)]));
    let fork = CommitmentList::seal_extending(&key(1), "holder", &first, None, 0, span(&[("live/p3b", 3)]));
    std::fs::write(w.dir.join("live-0010.wkl1"), first.to_bytes()).unwrap();
    let open = || {
        let mut f = w.file();
        f.lists.push(w.dir.join("live-0010.wkl1"));
        f.routes.push(RouteFile {
            name: "live.a".into(),
            content_id: "live-0010".into(),
            delivery: "drm".into(),
            system_id: Some(SYSTEM.into()),
            upstream: Some(format!("http://{}/licence", w.upstream_addr)),
            upstream_headers: BTreeMap::new(),
            key_url: None,
        });
        proxy::App::open(Proxy::from_file(f).unwrap())
    };
    let ask = |labels: &[&str], body: &str| Ask {
        route: "live.a",
        content: "live-0010",
        delivery: Delivery { kind: DeliveryKind::Drm, system_id: cpix::uuid(SYSTEM) },
        kids: labels.iter().map(|l| kid(l)).collect(),
        kind: RequestKind::Streaming,
        renews: None,
        body: body.as_bytes().to_vec(),
    };
    let premium = entitlement("svod:premium", 0, 0);
    let (app, addr, admin, h, h2) = {
        let app = open().unwrap();
        let (a, h) = spawn(proxy::router(app.clone())).await;
        let (b, h2) = spawn(proxy::admin_router(app.clone())).await;
        (app, a, b, h, h2)
    };
    let q = ask(&["live/p3"], "the next span's key, before its list");
    let got = call(addr, "POST", &path(&q), &signed(&q, &premium), q.body.clone()).await;
    assert_eq!((got.status, receipt_of(addr, &got).await.decision), (403, Decision::DeniedUnknownKey));
    assert_eq!(call(admin, "POST", "/v1/lists", &[], next.to_bytes()).await.status, 201);
    assert_eq!(call(admin, "POST", "/v1/lists", &[], fork.to_bytes()).await.status, 409, "a list that does not extend the chain's last");
    assert_eq!(call(admin, "POST", "/v1/lists", &[], b"not a list".to_vec()).await.status, 422);
    let q = ask(&["live/p3"], "the next span's key, once its list is held");
    let got = call(addr, "POST", &path(&q), &signed(&q, &premium), q.body.clone()).await;
    let r = receipt_of(addr, &got).await;
    assert_eq!((got.status, r.decision, r.list), (200, Decision::Granted, Some(next.identity())), "released, naming the chain's last list");
    let q = ask(&["live/p1", "live/p4"], "a key of each span");
    assert_eq!(call(addr, "POST", &path(&q), &signed(&q, &premium), q.body.clone()).await.status, 200);
    h.abort();
    h2.abort();
    let _ = h.await;
    let _ = h2.await;
    drop(app);
    let mut tries = 0;
    let app = loop {
        match open() {
            Ok(a) => break a,
            Err(_) if tries < 100 => {
                tries += 1;
                tokio::time::sleep(Duration::from_millis(30)).await;
            }
            Err(e) => panic!("{e}"),
        }
    };
    let (addr, _) = spawn(proxy::router(app.clone())).await;
    let (admin, _) = spawn(proxy::admin_router(app.clone())).await;
    let q = ask(&["live/p4"], "the next span's key after a restart");
    assert_eq!(call(addr, "POST", &path(&q), &signed(&q, &premium), q.body.clone()).await.status, 200, "the chain is read back from list_dir");
    let log = call(admin, "GET", "/v1/log", &[], vec![]).await;
    let receipts: Vec<serde_json::Value> = String::from_utf8(log.body).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect();
    let lists = vec![first.to_bytes(), next.to_bytes()];
    let case = LogCase {
        authorities: &[key(2).verifying_key().to_bytes()],
        attesters: &[],
        proxies: &[key(3).verifying_key().to_bytes()],
        lists: &lists,
        cpix: &[],
        receipts: &receipts,
        tickets: &[],
        revocations: &[],
    };
    assert_eq!(verify_log(&case).map(|f| f.len()).map_err(|r| (r.reason.code(), r.place, r.index)), Ok(4));
    let _ = std::fs::remove_dir_all(&w.dir);
}

/// A checkpoint through the proxy (key-release §17): signed after the configured number of
/// receipts and on request, written to the log, and gone on from: a replay of a request before it
/// is refused by the proof it keeps, an offline licence before it is renewed from it, both again
/// after a restart that reads the log from its last checkpoint, while a receipt before it is still
/// served from the file; and the whole log, checkpoints among it, verifies.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_checkpoint_is_signed_and_the_proxy_goes_on_from_it() {
    let w = World::new("checkpoint").await;
    let premium = entitlement("svod:premium", 600, 1);
    let open = || {
        let mut f = w.file();
        f.checkpoint_every = 3;
        proxy::App::open(Proxy::from_file(f).unwrap())
    };
    let started = |app: Arc<proxy::App>| async move {
        let (a, h) = spawn(proxy::router(app.clone())).await;
        let (b, h2) = spawn(proxy::admin_router(app.clone())).await;
        (app, a, b, h, h2)
    };
    let (app, addr, admin, h, h2) = started(open().unwrap()).await;
    let first = film_drm(RequestKind::Streaming, None, "before the checkpoint, one");
    let first_headers = signed(&first, &premium);
    let one = call(addr, "POST", &path(&first), &first_headers, first.body.clone()).await;
    assert_eq!(one.status, 200);
    let first_id = receipt_of(addr, &one).await.identity();
    for body in ["before the checkpoint, two", "before the checkpoint, three"] {
        let q = film_drm(RequestKind::Streaming, None, body);
        assert_eq!(call(addr, "POST", &path(&q), &signed(&q, &premium), q.body.clone()).await.status, 200);
    }
    let lines = |b: Vec<u8>| -> Vec<serde_json::Value> { String::from_utf8(b).unwrap().lines().map(|l| serde_json::from_str(l).unwrap()).collect() };
    let log = lines(call(admin, "GET", "/v1/log", &[], vec![]).await.body);
    assert_eq!((log.len(), wai::key_release::checkpoint::Checkpoint::is_one(&log[3])), (4, true), "a checkpoint after the third receipt");
    let off = film_drm(RequestKind::Offline, None, "an offline licence after the checkpoint");
    let o = call(addr, "POST", &path(&off), &signed(&off, &premium), off.body.clone()).await;
    let off_id = receipt_of(addr, &o).await.identity();
    let again = call(addr, "POST", &path(&first), &first_headers, first.body.clone()).await;
    let replay = receipt_of(addr, &again).await;
    assert_eq!((again.status, replay.decision), (403, Decision::DeniedReplayed), "the checkpoint keeps the proof");
    // The same replay at once: in the same second it is the same receipt, held once (§9.5).
    let twice = call(addr, "POST", &path(&first), &first_headers, first.body.clone()).await;
    let same_second = receipt_of(addr, &twice).await.identity() == replay.identity();
    let signed_now = call(admin, "POST", "/v1/checkpoint", &[], vec![]).await;
    assert_eq!(signed_now.status, 201, "{}", String::from_utf8_lossy(&signed_now.body));
    assert_eq!(app.log_len().await, 0, "the proxy holds only the receipts since its last checkpoint");
    h.abort();
    h2.abort();
    let _ = h.await;
    let _ = h2.await;
    drop(app);
    let mut tries = 0;
    let app = loop {
        match open() {
            Ok(a) => break a,
            Err(_) if tries < 100 => {
                tries += 1;
                tokio::time::sleep(Duration::from_millis(30)).await;
            }
            Err(e) => panic!("{e}"),
        }
    };
    let (_app, addr, admin, _, _) = started(app).await;
    // A second later than the replays before the restart, so this one is a receipt of its own.
    tokio::time::sleep(Duration::from_millis(1_100)).await;
    let got = call(addr, "GET", &format!("/v1/receipts/{}", hex(&first_id)), &[], vec![]).await;
    assert_eq!(got.status, 200, "a receipt before the checkpoint, read from the file");
    let again = call(addr, "POST", &path(&first), &first_headers, first.body.clone()).await;
    assert_eq!((again.status, receipt_of(addr, &again).await.decision), (403, Decision::DeniedReplayed), "a replay across a restart, from the checkpoint");
    let ren = film_drm(RequestKind::Renewal, Some(off_id), "a renewal from the checkpoint");
    let n = call(addr, "POST", &path(&ren), &signed(&ren, &premium), ren.body.clone()).await;
    assert_eq!((n.status, receipt_of(addr, &n).await.renewal), (200, Some(1)), "a licence the checkpoint keeps is renewed");
    assert_eq!(verify_whole_log(admin, &w).await, if same_second { 7 } else { 8 }, "the whole log, two checkpoints among it, verifies");
    let _ = std::fs::remove_dir_all(&w.dir);
}