use ratdb_core::{file::FileStorage, Config, Database};
use ratdb_server::{
    auth::{credential_path, Credentials},
    protocol::*,
    transport::{Client, Server},
};
use serde_json::json;
use std::{
    io::{Read, Write},
    net::{SocketAddr, TcpStream},
    path::PathBuf,
    sync::{
        atomic::{AtomicBool, AtomicUsize, Ordering},
        Arc,
    },
    thread,
};
static NEXT: AtomicUsize = AtomicUsize::new(0);
const PASSWORD: &str = "test password only 123";
struct Fixture {
    path: PathBuf,
    stop: Arc<AtomicBool>,
    handle: Option<thread::JoinHandle<()>>,
    addr: SocketAddr,
}
impl Fixture {
    fn new(max: usize) -> Self {
        Self::limited(max, Default::default())
    }
    fn limited(max: usize, limits: ratdb_server::transport::Limits) -> Self {
        let path = std::env::temp_dir().join(format!(
            "ratdb-server-{}-{}.db",
            std::process::id(),
            NEXT.fetch_add(1, Ordering::Relaxed)
        ));
        drop(Database::open(FileStorage::open(&path).unwrap(), Config::default()).unwrap());
        Credentials::new("admin", PASSWORD)
            .unwrap()
            .save_new(&credential_path(&path))
            .unwrap();
        let server =
            Server::bind_with_limits(&path, "127.0.0.1:0".parse().unwrap(), max, limits).unwrap();
        let addr = server.local_addr().unwrap();
        let stop = Arc::new(AtomicBool::new(false));
        let flag = stop.clone();
        let handle = Some(thread::spawn(move || server.run(flag).unwrap()));
        Self {
            path,
            addr,
            stop,
            handle,
        }
    }
    fn client(&self) -> Client {
        let mut c = Client::connect(self.addr).unwrap();
        let r = c.authenticate("admin".into(), PASSWORD.into()).unwrap();
        assert!(r.ok, "{r:?}");
        c
    }
    fn shutdown(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
        if let Some(h) = self.handle.take() {
            h.join().unwrap();
        }
    }
}
impl Drop for Fixture {
    fn drop(&mut self) {
        self.shutdown();
        std::fs::remove_file(&self.path).unwrap();
        std::fs::remove_file(credential_path(&self.path)).unwrap();
    }
}
#[test]
fn persistent_crud_restart_and_ownership() {
    let mut f = Fixture::new(16);
    assert!(FileStorage::open(&f.path).is_err());
    let mut c = f.client();
    assert!(!c.command(Command::Get { id: "1".into() }).unwrap().ok);
    assert!(
        c.command(Command::Use {
            collection: "cars".into()
        })
        .unwrap()
        .ok
    );
    assert!(
        c.command(Command::Put {
            id: "1".into(),
            document: json!({"make":"Ford","text":"hello world"})
        })
        .unwrap()
        .ok
    );
    assert_eq!(
        c.command(Command::Get { id: "1".into() }).unwrap().data["make"],
        "Ford"
    );
    assert_eq!(
        c.command(Command::Find {
            field: "make".into(),
            value: json!("Ford")
        })
        .unwrap()
        .data["documents"]
            .as_array()
            .unwrap()
            .len(),
        1
    );
    assert_eq!(
        c.command(Command::Collections).unwrap().data["collections"],
        json!(["cars"])
    );
    assert!(c.command(Command::Check).unwrap().ok);
    assert!(
        c.command(Command::Use {
            collection: "users".into()
        })
        .unwrap()
        .ok
    );
    assert!(c
        .command(Command::Get { id: "1".into() })
        .unwrap()
        .data
        .is_null());
    assert!(c.command(Command::Quit).unwrap().ok);
    assert!(c.command(Command::Stats).is_err());
    drop(c);
    f.shutdown();
    let server = Server::bind(&f.path, f.addr, 16).unwrap();
    f.stop = Arc::new(AtomicBool::new(false));
    let flag = f.stop.clone();
    f.handle = Some(thread::spawn(move || server.run(flag).unwrap()));
    let mut c = f.client();
    c.command(Command::Use {
        collection: "cars".into(),
    })
    .unwrap();
    assert_eq!(
        c.command(Command::Get { id: "1".into() }).unwrap().data["make"],
        "Ford"
    );
    assert!(c.command(Command::Delete { id: "1".into() }).unwrap().ok);
    assert!(c
        .command(Command::Get { id: "1".into() })
        .unwrap()
        .data
        .is_null());
}
#[test]
fn authentication_and_versions() {
    let f = Fixture::new(16);
    for (user, password) in [("admin", "incorrect"), ("missing", PASSWORD)] {
        let mut c = Client::connect(f.addr).unwrap();
        assert!(!c.authenticate(user.into(), password.into()).unwrap().ok);
        assert!(c.command(Command::Stats).is_err());
    }
    let mut s = TcpStream::connect(f.addr).unwrap();
    write_frame(
        &mut s,
        &Request {
            version: 99,
            command: Command::Stats,
        },
    )
    .unwrap();
    let r: Response = read_frame(&mut s, MAX_FRAME).unwrap();
    assert_eq!(r.error.unwrap().code, "version");
    let mut c = Client::connect(f.addr).unwrap();
    assert!(!c.command(Command::Stats).unwrap().ok);
    let mut c = f.client();
    assert!(!c.authenticate("admin".into(), PASSWORD.into()).unwrap().ok);
    assert!(c.command(Command::Stats).unwrap().ok);
}
#[test]
fn concurrent_clients_serialize_writes() {
    let f = Fixture::new(16);
    let mut handles = Vec::new();
    for i in 0..4 {
        let addr = f.addr;
        handles.push(thread::spawn(move || {
            let mut c = Client::connect(addr).unwrap();
            assert!(c.authenticate("admin".into(), PASSWORD.into()).unwrap().ok);
            c.command(Command::Use {
                collection: format!("c{i}"),
            })
            .unwrap();
            for n in 0..8 {
                assert!(
                    c.command(Command::Put {
                        id: n.to_string(),
                        document: json!({"client":i})
                    })
                    .unwrap()
                    .ok
                );
            }
        }));
    }
    for h in handles {
        h.join().unwrap();
    }
    let mut c = f.client();
    assert_eq!(c.command(Command::Stats).unwrap().data["records"], 32);
    assert!(c.command(Command::Check).unwrap().ok);
}
#[test]
fn malformed_and_oversized_frames() {
    let f = Fixture::new(16);
    for bytes in [
        vec![0, 0, 0, 0],
        vec![0, 16, 0, 1],
        vec![0, 0, 0, 1, b'{'],
        vec![0, 0, 0, 2, b'[', b']'],
    ] {
        let mut s = TcpStream::connect(f.addr).unwrap();
        s.write_all(&bytes).unwrap();
        let r: Response = read_frame(&mut s, MAX_FRAME).unwrap();
        assert_eq!(r.error.unwrap().code, "bad_frame");
        let mut b = [0];
        assert_eq!(s.read(&mut b).unwrap(), 0);
    }
    // Fragmented header and body are supported by exact reads.
    let mut s = TcpStream::connect(f.addr).unwrap();
    let mut wire = Vec::new();
    write_frame(
        &mut wire,
        &Request {
            version: VERSION,
            command: Command::Auth {
                username: "admin".into(),
                password: PASSWORD.into(),
            },
        },
    )
    .unwrap();
    for byte in wire {
        s.write_all(&[byte]).unwrap();
    }
    let r: Response = read_frame(&mut s, MAX_FRAME).unwrap();
    assert!(r.ok);
    // Authenticated oversized frames are also rejected before allocating a body.
    s.write_all(&((MAX_FRAME + 1) as u32).to_be_bytes())
        .unwrap();
    let r: Response = read_frame(&mut s, MAX_FRAME).unwrap();
    assert_eq!(r.error.unwrap().code, "bad_frame");
    let mut truncated = TcpStream::connect(f.addr).unwrap();
    truncated.write_all(&[0, 0]).unwrap();
    truncated.shutdown(std::net::Shutdown::Write).unwrap();
    let r: Response = read_frame(&mut truncated, MAX_FRAME).unwrap();
    assert!(!r.ok);
}
#[test]
fn limits_credentials_and_bind() {
    let mut f = Fixture::new(1);
    let mut c = f.client();
    let mut extra = TcpStream::connect(f.addr).unwrap();
    extra
        .set_read_timeout(Some(std::time::Duration::from_secs(5)))
        .unwrap();
    let r: Response = read_frame(&mut extra, MAX_FRAME).unwrap();
    assert_eq!(r.error.unwrap().code, "server_busy");
    assert!(Server::bind(&f.path, "0.0.0.0:0".parse().unwrap(), 1).is_err());
    assert!(Client::connect("192.0.2.1:7337".parse().unwrap()).is_err());
    assert!(Credentials::new("admin", "short").is_err());
    assert!(Credentials::new("admin", PASSWORD)
        .unwrap()
        .save_new(&credential_path(&f.path))
        .is_err());
    f.shutdown();
    assert!(c.command(Command::Stats).is_err());
    std::fs::write(credential_path(&f.path), b"corrupted").unwrap();
    assert!(Server::bind(&f.path, f.addr, 1).is_err());
    std::fs::remove_file(credential_path(&f.path)).unwrap();
    assert!(Server::bind(&f.path, f.addr, 1).is_err());
    Credentials::new("admin", PASSWORD)
        .unwrap()
        .save_new(&credential_path(&f.path))
        .unwrap();
}

#[test]
fn strict_envelope_and_authenticated_version() {
    let f = Fixture::new(16);
    for request in [
        json!({"version":1,"command":{"op":"auth","username":"admin","password":PASSWORD,"extra":true}}),
        json!({"version":1,"command":{"op":"stats"},"extra":true}),
        json!({"version":1,"command":{"op":"unknown"}}),
    ] {
        let mut s = TcpStream::connect(f.addr).unwrap();
        write_frame(&mut s, &request).unwrap();
        let r: Response = read_frame(&mut s, MAX_FRAME).unwrap();
        assert_eq!(r.error.unwrap().code, "bad_frame");
    }
    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();
    let r: Response = read_frame(&mut s, MAX_FRAME).unwrap();
    assert!(r.ok);
    write_frame(
        &mut s,
        &Request {
            version: 2,
            command: Command::Stats,
        },
    )
    .unwrap();
    let r: Response = read_frame(&mut s, MAX_FRAME).unwrap();
    assert_eq!(r.error.unwrap().code, "version");
}

#[test]
fn warren_navigation_isolation_and_persistence() {
    let mut f = Fixture::new(16);
    let mut c = f.client();
    let nav = |name: &str| Command::Navigate { name: name.into() };
    assert!(!c.command(Command::Dens).unwrap().ok);
    for (w, d, n) in [
        ("garage", "cars", 1),
        ("garage", "trucks", 2),
        ("work", "cars", 3),
    ] {
        assert!(c.command(nav(&format!("/{w}"))).unwrap().ok);
        assert!(c.command(nav(d)).unwrap().ok);
        assert!(
            c.command(Command::Put {
                id: "same".into(),
                document: json!({"n":n})
            })
            .unwrap()
            .ok
        );
    }
    c.command(nav("/garage")).unwrap();
    assert_eq!(
        c.command(Command::Dens).unwrap().data["names"],
        json!(["cars", "trucks"])
    );
    c.command(nav("cars")).unwrap();
    assert_eq!(
        c.command(Command::Get { id: "same".into() }).unwrap().data["n"],
        1
    );
    assert_eq!(
        c.command(Command::Shinies).unwrap().data["names"],
        json!(["same"])
    );
    let r = c.command(nav("..")).unwrap();
    assert_eq!(r.data["warren"], "garage");
    assert!(r.data["den"].is_null());
    c.command(nav("..")).unwrap();
    assert!(!c.command(Command::Get { id: "same".into() }).unwrap().ok);
    assert_eq!(
        c.command(Command::Warrens).unwrap().data["names"],
        json!(["garage", "work"])
    );
    c.command(Command::Quit).unwrap();
    drop(c);
    f.shutdown();
    let server = Server::bind(&f.path, f.addr, 16).unwrap();
    f.stop = Arc::new(AtomicBool::new(false));
    let flag = f.stop.clone();
    f.handle = Some(thread::spawn(move || server.run(flag).unwrap()));
    let mut c = f.client();
    c.command(nav("work")).unwrap();
    c.command(nav("cars")).unwrap();
    assert_eq!(
        c.command(Command::Get { id: "same".into() }).unwrap().data["n"],
        3
    );
}

#[test]
fn application_commands_over_persistent_connections() {
    let f = Fixture::new(4);
    let mut first = f.client();
    let mut second = f.client();
    for client in [&mut first, &mut second] {
        assert!(
            client
                .command(Command::Navigate {
                    name: "crm/contacts".into()
                })
                .unwrap()
                .ok
        );
        assert!(
            !client
                .command(Command::Navigate {
                    name: "work/../x".into()
                })
                .unwrap()
                .ok
        );
    }
    for id in ["b", "a", "c"] {
        assert!(
            first
                .command(Command::Insert {
                    id: id.into(),
                    document: json!({"active":id != "b"})
                })
                .unwrap()
                .ok
        );
    }
    assert!(
        !second
            .command(Command::Insert {
                id: "a".into(),
                document: json!({})
            })
            .unwrap()
            .ok
    );
    let page = |after: Option<String>| Command::Page {
        limit: 1,
        after,
        filter: Some(Equality {
            field: "active".into(),
            value: json!(true),
        }),
    };
    let first_page = second.command(page(None)).unwrap();
    assert_eq!(
        first_page.data["documents"],
        json!([{"id":"a","active":true}])
    );
    let cursor = first_page.data["next_after"].as_str().unwrap().to_owned();
    assert!(
        first
            .command(Command::Delete { id: cursor.clone() })
            .unwrap()
            .ok
    );
    let second_page = second.command(page(Some(cursor))).unwrap();
    assert_eq!(
        second_page.data["documents"],
        json!([{"id":"c","active":true}])
    );
    assert!(second_page.data["next_after"].is_null());
    assert!(
        !second
            .command(Command::Page {
                limit: 101,
                after: None,
                filter: None
            })
            .unwrap()
            .ok
    );
    thread::scope(|s| {
        for (client, fields) in [
            (&mut first, json!({"one":1})),
            (&mut second, json!({"two":2})),
        ] {
            s.spawn(move || {
                assert!(
                    client
                        .command(Command::Update {
                            id: "c".into(),
                            fields
                        })
                        .unwrap()
                        .ok
                )
            });
        }
    });
    assert!(
        !first
            .command(Command::Update {
                id: "c".into(),
                fields: json!({"id":"x"})
            })
            .unwrap()
            .ok
    );
    assert_eq!(
        first.command(Command::Get { id: "c".into() }).unwrap().data,
        json!({"id":"c","active":true,"one":1,"two":2})
    );
}

#[path = "support/client_facade.rs"]
mod client_facade;

#[test]
fn configured_limits_reject_oversize_before_commit_and_frame_allocation() {
    use ratdb_server::transport::Limits;
    let f = Fixture::limited(
        2,
        Limits {
            request_bytes: 256,
            document_bytes: 64,
            index_entries: 0,
        },
    );
    let mut c = f.client();
    assert!(
        c.command(Command::Use {
            collection: "cars".into()
        })
        .unwrap()
        .ok
    );
    assert!(
        c.command(Command::Put {
            id: "a".into(),
            document: json!({"x":"kept"})
        })
        .unwrap()
        .ok
    );
    assert!(
        !c.command(Command::Put {
            id: "a".into(),
            document: json!({"x":"q".repeat(80)})
        })
        .unwrap()
        .ok
    );
    assert_eq!(
        c.command(Command::Get { id: "a".into() }).unwrap().data["x"],
        "kept"
    );
    let r = c
        .command(Command::Put {
            id: "a".into(),
            document: json!({"x":"q".repeat(512)}),
        })
        .unwrap();
    assert_eq!(r.error.unwrap().code, "bad_frame");
}