use super::{Fixture, PASSWORD};
use ratdb_server::client::{Client, Error};
use serde_json::json;

#[test]
fn typed_session_crud_and_pagination() {
    let f = Fixture::new(4);
    assert!(matches!(
        Client::connect(f.addr, "admin", "incorrect".into()),
        Err(Error::Server(_))
    ));
    let mut client = Client::connect(f.addr, "admin", PASSWORD.into()).unwrap();
    assert!(matches!(client.warren(".."), Err(Error::InvalidName)));
    let mut cars = client.warren("garage").unwrap().den("cars").unwrap();
    cars.insert("a", json!({"make":"Ford"})).unwrap();
    assert!(matches!(cars.insert("a", json!({})), Err(Error::Server(_))));
    cars.put("b", json!({"make":"Ford"})).unwrap();
    cars.update("a", json!({"year":2021})).unwrap();
    assert_eq!(cars.get("a").unwrap().unwrap()["year"], 2021);
    assert_eq!(
        cars.find("make", json!("Ford"), 10)
            .unwrap()
            .documents
            .len(),
        2
    );
    let first = cars.page(1, None, None).unwrap();
    assert_eq!(first.documents[0]["id"], "a");
    let second = cars.page(1, first.next_after.as_deref(), None).unwrap();
    assert_eq!(second.documents[0]["id"], "b");
    cars.delete("a").unwrap();
    assert!(cars.get("a").unwrap().is_none());
    assert!(client
        .warren("other")
        .unwrap()
        .den("cars")
        .unwrap()
        .get("b")
        .unwrap()
        .is_none());
}

#[test]
fn short_frame_deadlines_reclaim_slots_and_shutdown_partial_requests() {
    use ratdb_server::{
        protocol::{read_frame, write_frame, Command, Request, Response, MAX_FRAME, VERSION},
        transport::Server,
    };
    use std::{
        io::Write,
        net::TcpStream,
        sync::{atomic::AtomicBool, Arc},
        thread,
        time::Duration,
    };

    let mut f = Fixture::new(3);
    f.shutdown();
    let server = Server::bind(&f.path, f.addr, 3)
        .unwrap()
        .with_frame_timeouts(Duration::from_millis(200), Duration::from_millis(400))
        .unwrap();
    f.stop = Arc::new(AtomicBool::new(false));
    let stop = f.stop.clone();
    f.handle = Some(thread::spawn(move || server.run(stop).unwrap()));

    let connect = || {
        let s = TcpStream::connect(f.addr).unwrap();
        s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
        s
    };
    let mut idle = connect();
    idle.write_all(&[0]).unwrap();
    let reply: Response = read_frame(&mut idle, MAX_FRAME).unwrap();
    assert_eq!(reply.error.unwrap().code, "bad_frame");

    let mut slow = connect();
    write_frame(
        &mut slow,
        &Request {
            version: VERSION,
            command: Command::Auth {
                username: "admin".into(),
                password: PASSWORD.into(),
            },
        },
    )
    .unwrap();
    assert!(read_frame::<_, Response>(&mut slow, MAX_FRAME).unwrap().ok);
    slow.write_all(&1024u32.to_be_bytes()).unwrap();
    let mut writer = slow.try_clone().unwrap();
    let started = std::time::Instant::now();
    let dribble = thread::spawn(move || {
        // Every body byte arrives well within the configured socket timeout.
        // Resetting a per-read timeout would keep this incomplete frame alive.
        for _ in 0..40 {
            if writer.write_all(b" ").is_err() {
                break;
            }
            thread::sleep(Duration::from_millis(50));
        }
    });
    let reply: Response = read_frame(&mut slow, MAX_FRAME).unwrap();
    assert_eq!(reply.error.unwrap().code, "bad_frame");
    assert!(
        started.elapsed() < Duration::from_secs(2),
        "dribbled bytes reset the deadline"
    );
    dribble.join().unwrap();

    let mut active = f.client();
    assert!(active.command(Command::Check).unwrap().ok);
    let mut partial = connect();
    partial.write_all(&[0, 0]).unwrap();
    f.shutdown();
    assert!(active.command(Command::Stats).is_err());
}

#[test]
fn invalid_peer_replies_close_rust_client() {
    use ratdb_server::protocol::{read_frame, write_frame, Request, Response, MAX_FRAME};
    use std::{io::Write, net::TcpListener, thread};
    let mut invalid_envelope = Vec::new();
    write_frame(
        &mut invalid_envelope,
        &json!({"version":1,"ok":false,"error":null,"data":null}),
    )
    .unwrap();
    for frame in [
        invalid_envelope,
        ((MAX_FRAME + 1) as u32).to_be_bytes().to_vec(),
        vec![0, 0, 0, 2, b'{'],
    ] {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let peer = thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            stream
                .set_read_timeout(Some(std::time::Duration::from_secs(5)))
                .unwrap();
            read_frame::<_, Request>(&mut stream, MAX_FRAME).unwrap();
            write_frame(&mut stream, &Response::ok(json!({"authenticated":true}))).unwrap();
            read_frame::<_, Request>(&mut stream, MAX_FRAME).unwrap();
            stream.write_all(&frame).unwrap();
        });
        let mut client = Client::connect(addr, "fixture", "fixture".into()).unwrap();
        assert!(client.warren("garage").is_err());
        assert!(
            matches!(client.warren("other"), Err(Error::Connection(err)) if err.kind() == std::io::ErrorKind::NotConnected)
        );
        peer.join().unwrap();
    }
}