//! An appview fed by Jetstream: a stand-in server speaks subscribeEvents
//! frames, and the index fills from the records they carry without asking
//! any PDS.

use futures::SinkExt;
use knotless_serve::App;
use knotless_serve::index::Index;
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};

const KIND: &str = "network.bsky.jetstream.subscribeEvents#";
const TIME: &str = "2026-10-20T00:00:00.000001Z";
const REPO: &str = "at://did:plc:owner/dev.knotless.repo/proj";
const KEY: &str = "3l3qo2vuowo2a";

fn message(payload: serde_json::Value) -> Message {
    Message::Text(
        serde_json::json!({"$type": "message", "payload": payload})
            .to_string()
            .into(),
    )
}

fn commit(
    seq: i64,
    did: &str,
    op: &str,
    collection: &str,
    rkey: &str,
    record: Option<serde_json::Value>,
) -> Message {
    let mut payload = serde_json::json!({
        "$type": format!("{KIND}commit"), "seq": seq, "did": did, "time": TIME,
        "rev": "3l3qo2vutsw2b", "operation": op, "collection": collection, "rkey": rkey,
    });
    if let Some(record) = record {
        payload["record"] = record;
        payload["cid"] = "bafyreia4w4bbvc5lqzkvexqlbwbagz5yfjfzzrqhfwqkhkk2xgekh4c6ai".into();
    }
    message(payload)
}

fn wait_until(mut f: impl FnMut() -> bool) {
    for _ in 0..200 {
        if f() {
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    panic!("condition never held");
}

// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
#[test]
fn the_index_fills_from_jetstream() {
    let dir = tempfile::tempdir().unwrap();
    let rt = tokio::runtime::Runtime::new().unwrap();
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let addr = listener.local_addr().unwrap();
    let (asked, mut asks) = tokio::sync::mpsc::unbounded_channel();
    let (go, mut gone) = tokio::sync::mpsc::unbounded_channel::<()>();
    rt.spawn(async move {
        for connection in 0..=u32::MAX {
            let (tcp, _) = listener.accept().await.unwrap();
            let asked = asked.clone();
            let mut ws = tokio_tungstenite::accept_hdr_async(tcp, |req: &Request, mut res: Response| {
                let protocol = req.headers()["sec-websocket-protocol"].clone();
                asked.send((req.uri().to_string(), protocol.to_str().unwrap().to_string())).unwrap();
                res.headers_mut().insert("sec-websocket-protocol", protocol);
                Ok(res)
            })
            .await
            .unwrap();
            let frames = match connection {
                0 => vec![
                    message(serde_json::json!({"$type": format!("{KIND}info"), "name": "OutdatedCursor", "message": "starting at seq 10"})),
                    Message::Text("not json".into()),
                    Message::Binary(vec![1].into()),
                    commit(10, "did:plc:owner", "create", "dev.knotless.repo", "proj", Some(serde_json::json!({
                        "$type": "dev.knotless.repo", "key": KEY, "description": "a project", "createdAt": TIME,
                    }))),
                    commit(11, "did:plc:other", "create", "dev.knotless.issue", "3l3qo2vuowo2b", Some(serde_json::json!({
                        "$type": "dev.knotless.issue", "repo": {"uri": REPO, "key": KEY}, "title": "It breaks", "createdAt": TIME,
                    }))),
                    commit(12, "did:plc:other", "create", "app.bsky.feed.post", "3l3qo2vuowo2c", Some(serde_json::json!({
                        "$type": "app.bsky.feed.post", "text": "elsewhere", "createdAt": TIME,
                    }))),
                    message(serde_json::json!({
                        "$type": format!("{KIND}identity"), "seq": 13, "did": "did:plc:moved", "time": TIME,
                        "identity": {"did": "did:plc:moved", "handle": "moved.test", "seq": 99, "time": TIME},
                    })),
                    Message::Text(r#"{"$type":"error","error":"ConsumerTooSlow"}"#.into()),
                ],
                // Its author edits it, keeping what the index does not know.
                1 => vec![commit(14, "did:plc:other", "update", "dev.knotless.issue", "3l3qo2vuowo2b", Some(serde_json::json!({
                    "$type": "dev.knotless.issue", "repo": {"uri": REPO, "key": KEY}, "title": "It breaks badly",
                    "dev.example.labels": ["bug"], "createdAt": TIME, "updatedAt": "2026-10-21T00:00:00.000Z",
                })))],
                // Once the test has seen the edit.
                2 => {
                    gone.recv().await;
                    vec![commit(15, "did:plc:other", "delete", "dev.knotless.issue", "3l3qo2vuowo2b", None)]
                }
                _ => vec![],
            };
            for frame in frames {
                ws.send(frame).await.unwrap();
            }
            ws.close(None).await.unwrap();
        }
    });

    let env = knotless::env::Env::from_vars(|_| None);
    let index = Index::open(&dir.path().join("index.db")).unwrap();
    let app = App::build(
        knotless::env::Ctx::new(env),
        Duration::ZERO,
        None,
        Some(index),
    );
    rt.spawn(knotless_serve::jetstream::watch(
        app.clone(),
        format!("ws://{addr}"),
    ));
    let index = app.index.clone().unwrap();

    // It asks for knotless commits, blocks, and every account's identity, offering
    // the one framing Jetstream speaks.
    let (first, protocol) = rt.block_on(asks.recv()).unwrap();
    assert_eq!(protocol, "xrpc.v1.json");
    assert!(
        first.starts_with(
            "/xrpc/network.bsky.jetstream.subscribeEvents?collections=dev.knotless.*&kinds=commit"
        ),
        "{first}"
    );
    assert!(!first.contains("cursor"), "{first}");

    // The records it carried are indexed as they came.
    wait_until(|| index.forge("did:plc:owner", KEY).unwrap().len() == 1);
    assert_eq!(
        index.repo(REPO).unwrap().unwrap()["description"],
        "a project"
    );
    assert_eq!(
        index.forge("did:plc:owner", KEY).unwrap()[0].value["title"],
        "It breaks"
    );
    // The identity came after the issue, so it can still be on its way.
    wait_until(|| app.events_for("did:plc:moved") == 1);
    assert_eq!(app.events_for("did:plc:owner"), 1);
    assert_eq!(app.events_for("did:plc:other"), 0);

    // After a drop it resumes past the last event it saw. An update
    // replaces what the index had, whole.
    let (second, _) = rt.block_on(asks.recv()).unwrap();
    assert!(second.ends_with("&cursor=14"), "{second}");
    wait_until(|| {
        index.forge("did:plc:owner", KEY).unwrap()[0].value["title"] == "It breaks badly"
    });
    let edited = index.forge("did:plc:owner", KEY).unwrap().remove(0).value;
    assert_eq!(edited["updatedAt"], "2026-10-21T00:00:00.000Z");
    assert_eq!(edited["dev.example.labels"][0], "bug");
    let (third, _) = rt.block_on(asks.recv()).unwrap();
    assert!(third.ends_with("&cursor=15"), "{third}");
    go.send(()).unwrap();
    wait_until(|| index.forge("did:plc:owner", KEY).unwrap().is_empty());
    let (fourth, _) = rt.block_on(asks.recv()).unwrap();
    assert!(fourth.ends_with("&cursor=16"), "{fourth}");

    // The index keeps where it was fed to, for a restart to resume from.
    assert_eq!(index.stream_seq(&format!("ws://{addr}")).unwrap(), Some(16));

    // A Jetstream it cannot reach, or cannot even name, leaves the TTL in
    // charge.
    for url in ["ws://127.0.0.1:9", "not a url"] {
        let lost = App::new(
            knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None)),
            Duration::ZERO,
        );
        lost.watching.store(true, Ordering::Relaxed);
        rt.spawn(knotless_serve::jetstream::watch(lost.clone(), url.into()));
        wait_until(|| !lost.watching.load(Ordering::Relaxed));
    }
}

// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
#[test]
fn a_silent_jetstream_is_left() {
    let dir = tempfile::tempdir().unwrap();
    let rt = tokio::runtime::Runtime::new().unwrap();
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let addr = listener.local_addr().unwrap();
    let (asked, mut asks) = tokio::sync::mpsc::unbounded_channel();
    rt.spawn(async move {
        let mut held = Vec::new();
        loop {
            let (tcp, _) = listener.accept().await.unwrap();
            let asked = asked.clone();
            let mut ws =
                tokio_tungstenite::accept_hdr_async(tcp, |req: &Request, mut res: Response| {
                    asked.send(req.uri().to_string()).unwrap();
                    let protocol = req.headers()["sec-websocket-protocol"].clone();
                    res.headers_mut().insert("sec-websocket-protocol", protocol);
                    Ok(res)
                })
                .await
                .unwrap();
            ws.send(commit(
                7,
                "did:plc:owner",
                "delete",
                "dev.knotless.issue",
                "3l3qo2vuowo2b",
                None,
            ))
            .await
            .unwrap();
            // Never read again, so its pings go unanswered: a connection
            // gone half-open.
            held.push(ws);
        }
    });

    // Restarted, it starts where the index was last fed to.
    let index = Index::open(&dir.path().join("index.db")).unwrap();
    index.set_stream_seq(&format!("ws://{addr}"), 5).unwrap();
    index.set_stream_seq(&format!("ws://{addr}"), 6).unwrap();
    let app = App::build(
        knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None)),
        Duration::ZERO,
        None,
        Some(index),
    );
    rt.spawn(knotless_serve::jetstream::follow(
        app.clone(),
        format!("ws://{addr}"),
        Duration::from_millis(100),
    ));
    let first = rt.block_on(asks.recv()).unwrap();
    assert!(first.ends_with("&cursor=6"), "{first}");
    // Silent through a ping, it is given up on and followed again from
    // past what it said.
    let second = rt.block_on(asks.recv()).unwrap();
    assert!(second.ends_with("&cursor=8"), "{second}");
    assert_eq!(
        app.index
            .as_ref()
            .unwrap()
            .stream_seq(&format!("ws://{addr}"))
            .unwrap(),
        Some(8)
    );
}

/// A stand-in Jetstream that sends `frames` once to the first client.
// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
fn sends(rt: &tokio::runtime::Runtime, frames: Vec<Message>) -> String {
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let url = format!("ws://{}", listener.local_addr().unwrap());
    rt.spawn(async move {
        let (tcp, _) = listener.accept().await.unwrap();
        let mut ws =
            tokio_tungstenite::accept_hdr_async(tcp, |req: &Request, mut res: Response| {
                let protocol = req.headers()["sec-websocket-protocol"].clone();
                res.headers_mut().insert("sec-websocket-protocol", protocol);
                Ok(res)
            })
            .await
            .unwrap();
        for frame in frames {
            ws.send(frame).await.unwrap();
        }
        // Held open, so the server does not ask again.
        std::future::pending::<()>().await;
    });
    url
}

#[test]
fn checks_by_the_runners_a_repository_names() {
    let dir = tempfile::tempdir().unwrap();
    let rt = tokio::runtime::Runtime::new().unwrap();
    let check = |seq: i64, runner: &str, rkey: &str, r: &str, sha: &str, at: &str| {
        commit(
            seq,
            runner,
            "create",
            "dev.knotless.check",
            rkey,
            Some(serde_json::json!({
                "$type": "dev.knotless.check", "repo": {"uri": REPO, "key": KEY}, "ref": r,
                "commit": sha, "workflow": "ci", "status": "success", "startedAt": at,
            })),
        )
    };
    let main = "refs/heads/main";
    let (a, b) = ("a".repeat(40), "b".repeat(40));
    let url = sends(
        &rt,
        vec![
            commit(
                1,
                "did:plc:owner",
                "create",
                "dev.knotless.repo",
                "proj",
                Some(serde_json::json!({
                    "$type": "dev.knotless.repo", "key": KEY, "name": "proj", "createdAt": TIME,
                    "runners": [{"did": "did:plc:ci"}],
                })),
            ),
            check(2, "did:plc:ci", "c1", main, &a, "2026-10-20T00:00:01.000Z"),
            check(
                3,
                "did:plc:ci",
                "c2",
                "refs/heads/x",
                &b,
                "2026-10-20T00:00:02.000Z",
            ),
            // Anyone can write a check; the repository does not name this one.
            check(
                4,
                "did:plc:stranger",
                "c3",
                main,
                &a,
                "2026-10-20T00:00:03.000Z",
            ),
        ],
    );
    let index = Index::open(&dir.path().join("index.db")).unwrap();
    let app = App::build(
        knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None)),
        Duration::ZERO,
        None,
        Some(index),
    );
    rt.spawn(knotless_serve::jetstream::watch(app.clone(), url));
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let router = knotless_serve::router(app);
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });

    let list = |query: &str| -> serde_json::Value {
        reqwest::blocking::get(format!(
            "{base}/xrpc/dev.knotless.listChecks?repo={REPO}&key={KEY}{query}"
        ))
        .unwrap()
        .json()
        .unwrap()
    };
    wait_until(|| list("")["checks"].as_array().unwrap().len() == 2);
    let all = list("");
    assert_eq!(
        all["checks"][0]["uri"],
        "at://did:plc:ci/dev.knotless.check/c2"
    );
    assert_eq!(all["checks"][0]["runner"], "did:plc:ci");
    assert_eq!(all["checks"][1]["value"]["commit"], a.as_str());
    assert!(all.get("cursor").is_none());
    assert_eq!(
        list(&format!("&ref={main}"))["checks"]
            .as_array()
            .unwrap()
            .len(),
        1
    );
    assert_eq!(
        list(&format!("&commit={b}"))["checks"][0]["value"]["ref"],
        "refs/heads/x"
    );
    let first = list("&limit=1");
    let cursor = first["cursor"].as_str().unwrap();
    let second = list(&format!("&limit=1&cursor={}", cursor.replace(' ', "%20")));
    assert_eq!(
        second["checks"][0]["uri"],
        "at://did:plc:ci/dev.knotless.check/c1"
    );
    // A repository the index has never seen names no runners.
    let unknown = reqwest::blocking::get(format!(
        "{base}/xrpc/dev.knotless.listChecks?repo=at://did:plc:nobody/dev.knotless.repo/x&key={KEY}"
    ))
    .unwrap()
    .json::<serde_json::Value>()
    .unwrap();
    assert_eq!(unknown["checks"], serde_json::json!([]));
    let bad = reqwest::blocking::get(format!("{base}/xrpc/dev.knotless.listChecks?repo=x&key=k"))
        .unwrap();
    assert_eq!(bad.status().as_u16(), 400);
}