use super::*;
use ratdb_server::{client, protocol::TransactionWrite};
use std::{
    sync::Barrier,
    time::{Duration, Instant},
};

fn put(den: &str, id: &str, document: serde_json::Value) -> TransactionWrite {
    TransactionWrite::Put {
        warren: "app".into(),
        den: den.into(),
        id: id.into(),
        document,
    }
}
fn generation(c: &mut Client) -> u64 {
    c.command(Command::Stats).unwrap().data["generation"]
        .as_u64()
        .unwrap()
}

#[test]
fn racing_transactions_publish_one_complete_write_set() {
    let f = Fixture::new(8);
    let barrier = Arc::new(Barrier::new(2));
    let mut threads = Vec::new();
    for name in ["alice", "bob"] {
        let addr = f.addr;
        let ready = barrier.clone();
        threads.push(thread::spawn(move || {
            let mut c = client::Client::connect(addr, "admin", PASSWORD.into()).unwrap();
            let (g, value) = c
                .warren("app")
                .unwrap()
                .den("users")
                .unwrap()
                .get_versioned("active")
                .unwrap();
            assert!(value.is_none());
            ready.wait();
            let result = c.transaction(
                g,
                vec![
                    put("users", "active", json!({"owner":name})),
                    put("audit", "event", json!({"owner":name})),
                ],
            );
            let won = match result {
                Ok(after) => {
                    assert_eq!(after, g + 1);
                    true
                }
                Err(client::Error::Server(error)) => {
                    assert_eq!(error.code, "generation_conflict");
                    false
                }
                Err(error) => panic!("unexpected transaction error: {error}"),
            };
            let (after, doc) = c
                .warren("app")
                .unwrap()
                .den("audit")
                .unwrap()
                .get_versioned("event")
                .unwrap();
            assert_eq!(after, g + 1);
            (name, won, doc.unwrap()["owner"].clone())
        }));
    }
    let results: Vec<_> = threads.into_iter().map(|h| h.join().unwrap()).collect();
    let winners: Vec<_> = results.iter().filter(|(_, won, _)| *won).collect();
    assert_eq!(winners.len(), 1);
    for (_, _, seen) in &results {
        assert_eq!(seen, winners[0].0);
    }
    let mut c = f.client();
    assert!(
        c.command(Command::Navigate {
            name: "app/users".into()
        })
        .unwrap()
        .ok
    );
    assert_eq!(
        c.command(Command::Get {
            id: "active".into()
        })
        .unwrap()
        .data["owner"],
        winners[0].0
    );
}

#[test]
fn invalid_and_oversized_transactions_preserve_generation_and_documents() {
    let f = Fixture::new(4);
    let mut c = f.client();
    let g = generation(&mut c);
    for invalid in [
        vec![],
        vec![
            put("users", "a", json!({})),
            put("users", "a", json!({"other":true})),
        ],
        vec![
            put("users", "a", json!({})),
            put("events", "x", json!({"id":"wrong"})),
        ],
        vec![
            put("users", "a", json!({})),
            put("events", "x", serde_json::Value::Null),
        ],
        vec![
            put("users", "a", json!({})),
            put("bad/name", "x", json!({})),
        ],
        vec![
            put("users", "a", json!({})),
            put("users", "large", json!({"pad":"x".repeat(256*1024)})),
        ],
        (0..129)
            .map(|n| put("users", &n.to_string(), json!({})))
            .collect(),
    ] {
        let r = c
            .command(Command::Transaction {
                expected_generation: g,
                changes: invalid,
            })
            .unwrap();
        assert_eq!(r.error.unwrap().code, "operation_failed");
        assert_eq!(generation(&mut c), g);
    }
    let mut s = TcpStream::connect(f.addr).unwrap();
    write_frame(
        &mut s,
        &Request {
            version: 1,
            command: Command::Auth {
                username: "admin".into(),
                password: PASSWORD.into(),
            },
        },
    )
    .unwrap();
    assert!(read_frame::<_, Response>(&mut s, MAX_FRAME).unwrap().ok);
    write_frame(
        &mut s,
        &json!({"version":1,"command":{"op":"transaction","expected_generation":g,"changes":[
            {"op":"put","warren":"app","den":"users","id":"a","document":{}},
            {"op":"delete","warren":"app","den":"users","id":"b","document":{}}
        ]}}),
    )
    .unwrap();
    assert_eq!(
        read_frame::<_, Response>(&mut s, MAX_FRAME)
            .unwrap()
            .error
            .unwrap()
            .code,
        "bad_frame"
    );
    assert_eq!(generation(&mut c), g);
    assert_eq!(c.command(Command::Stats).unwrap().data["records"], 0);
}

#[test]
fn transaction_reconciles_lost_reply_after_restart() {
    let mut f = Fixture::new(4);
    let mut observer = f.client();
    assert!(
        observer
            .command(Command::Navigate {
                name: "app/users".into()
            })
            .unwrap()
            .ok
    );
    assert!(
        observer
            .command(Command::Put {
                id: "retire".into(),
                document: json!({})
            })
            .unwrap()
            .ok
    );
    let g = generation(&mut observer);
    let mut raw = TcpStream::connect(f.addr).unwrap();
    write_frame(
        &mut raw,
        &Request {
            version: 1,
            command: Command::Auth {
                username: "admin".into(),
                password: PASSWORD.into(),
            },
        },
    )
    .unwrap();
    assert!(read_frame::<_, Response>(&mut raw, MAX_FRAME).unwrap().ok);
    write_frame(
        &mut raw,
        &Request {
            version: 1,
            command: Command::Transaction {
                expected_generation: g,
                changes: vec![
                    put("markers", "request-7", json!({"action":"retired"})),
                    TransactionWrite::Delete {
                        warren: "app".into(),
                        den: "users".into(),
                        id: "retire".into(),
                    },
                ],
            },
        },
    )
    .unwrap();
    // Send FIN after the complete frame and discard any reply; never replay the change.
    raw.shutdown(std::net::Shutdown::Write).unwrap();
    drop(raw);
    let deadline = Instant::now() + Duration::from_secs(5);
    while generation(&mut observer) == g {
        assert!(
            Instant::now() < deadline,
            "server never published the test request"
        );
        thread::yield_now();
    }
    drop(observer);
    f.shutdown();
    let server = Server::bind(&f.path, "127.0.0.1:0".parse().unwrap(), 4).unwrap();
    f.addr = server.local_addr().unwrap();
    f.stop = Arc::new(AtomicBool::new(false));
    let stop = f.stop.clone();
    f.handle = Some(thread::spawn(move || server.run(stop).unwrap()));
    let mut c = f.client();
    assert_eq!(generation(&mut c), g + 1);
    for (den, id, expected) in [
        ("users", "retire", serde_json::Value::Null),
        (
            "markers",
            "request-7",
            json!({"id":"request-7","action":"retired"}),
        ),
    ] {
        assert!(
            c.command(Command::Navigate {
                name: format!("app/{den}")
            })
            .unwrap()
            .ok
        );
        let r = c.command(Command::GetVersioned { id: id.into() }).unwrap();
        assert_eq!(r.data, json!({"generation":g+1,"document":expected}));
    }
}