//! 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();
    rt.spawn(async move {
        for connection in 0.. {
            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()),
                ],
                1 => vec![commit(14, "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),
        false,
    );
    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"
    );
    assert_eq!(app.events_for("did:plc:owner"), 1);
    assert_eq!(app.events_for("did:plc:other"), 0);
    assert_eq!(app.events_for("did:plc:moved"), 1);

    // After a drop it resumes past the last event it saw.
    let (second, _) = rt.block_on(asks.recv()).unwrap();
    assert!(second.ends_with("&cursor=14"), "{second}");
    wait_until(|| index.forge("did:plc:owner", KEY).unwrap().is_empty());
    let (third, _) = rt.block_on(asks.recv()).unwrap();
    assert!(third.ends_with("&cursor=15"), "{third}");

    // 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));
    }
}