nusadb 0.1.0

//! Async integration tests (feature `async`) against a real `nusadb-server`. Skipped if the binary
//! is not found. Run with `cargo test --features async`.
#![cfg(feature = "async")]
#![allow(clippy::unwrap_used, clippy::expect_used)]

use std::net::{TcpListener, TcpStream};
use std::path::PathBuf;
use std::process::{Child, Command};
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::{Duration, Instant};

use nusadb::{AsyncConnection, AsyncPool, Config};

static UNIQ: AtomicU32 = AtomicU32::new(1000);

fn server_binary() -> Option<PathBuf> {
    let exe = if cfg!(windows) {
        "nusadb-server.exe"
    } else {
        "nusadb-server"
    };
    let mut bases: Vec<PathBuf> = Vec::new();
    if let Ok(td) = std::env::var("CARGO_TARGET_DIR") {
        bases.push(PathBuf::from(td));
    }
    bases.push(PathBuf::from("../../target"));
    for base in bases {
        for profile in ["debug", "release"] {
            let candidate = base.join(profile).join(exe);
            if candidate.exists() {
                return Some(candidate);
            }
        }
    }
    None
}

fn free_port() -> u16 {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    drop(listener);
    port
}

struct Server {
    child: Child,
    port: u16,
    data_dir: PathBuf,
}

impl Server {
    fn start(bin: &PathBuf) -> Self {
        Self::start_with(bin, &[])
    }

    /// Boot the server with extra CLI args (e.g. `--auth-user USER:PASSWORD` to require SCRAM).
    fn start_with(bin: &PathBuf, extra_args: &[&str]) -> Self {
        let port = free_port();
        let uniq = UNIQ.fetch_add(1, Ordering::SeqCst);
        let data_dir =
            std::env::temp_dir().join(format!("turbo_ait_{}_{}", std::process::id(), uniq));
        std::fs::create_dir_all(&data_dir).unwrap();
        let child = Command::new(bin)
            .args([
                "--listen",
                &format!("127.0.0.1:{port}"),
                "--data-dir",
                data_dir.to_str().unwrap(),
            ])
            .args(extra_args)
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .expect("spawn nusadb-server");
        let server = Self {
            child,
            port,
            data_dir,
        };
        let deadline = Instant::now() + Duration::from_secs(15);
        while Instant::now() < deadline {
            if TcpStream::connect(("127.0.0.1", server.port)).is_ok() {
                std::thread::sleep(Duration::from_millis(150));
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }
        server
    }

    fn config(&self) -> Config {
        Config {
            host: "127.0.0.1".to_owned(),
            port: self.port,
            user: "turbo".to_owned(),
            password: None,
            database: "nusadb".to_owned(),
            connect_timeout: Some(Duration::from_secs(5)),
            tls: None,
        }
    }
}

impl Drop for Server {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
        let _ = std::fs::remove_dir_all(&self.data_dir);
    }
}

#[tokio::test]
async fn async_end_to_end() {
    let Some(bin) = server_binary() else {
        eprintln!("SKIP: nusadb-server binary not found; run `cargo build -p nusadb-server` first");
        return;
    };
    let server = Server::start(&bin);
    let mut conn = AsyncConnection::connect_config(&server.config())
        .await
        .expect("connect");

    conn.execute("CREATE TABLE t (id INT NOT NULL, name TEXT, PRIMARY KEY (id))")
        .await
        .unwrap();

    let n = conn
        .query_params("INSERT INTO t VALUES ($1, $2)", &[&1_i64, &"alice"])
        .await
        .unwrap()
        .affected();
    assert_eq!(n, 1);

    let result = conn
        .query("SELECT id, name FROM t ORDER BY id")
        .await
        .unwrap();
    assert_eq!(result.rows.len(), 1);
    assert_eq!(result.rows[0].get::<i64>(0).unwrap(), 1);
    assert_eq!(result.rows[0].get::<String>(1).unwrap(), "alice");

    // prepared, reused
    let stmt = conn
        .prepare("SELECT name FROM t WHERE id = $1")
        .await
        .unwrap();
    let r = conn.query_prepared(&stmt, &[&1_i64]).await.unwrap();
    assert_eq!(r.rows[0].get::<String>(0).unwrap(), "alice");

    // transaction: commit then rollback
    conn.begin().await.unwrap();
    conn.query_params("INSERT INTO t VALUES ($1, $2)", &[&2_i64, &"bob"])
        .await
        .unwrap();
    conn.commit().await.unwrap();
    assert_eq!(
        conn.query("SELECT id FROM t WHERE id = 2")
            .await
            .unwrap()
            .rows
            .len(),
        1
    );

    conn.begin().await.unwrap();
    conn.query_params("INSERT INTO t VALUES ($1, $2)", &[&3_i64, &"ghost"])
        .await
        .unwrap();
    conn.rollback().await.unwrap();
    assert_eq!(
        conn.query("SELECT id FROM t WHERE id = 3")
            .await
            .unwrap()
            .rows
            .len(),
        0
    );

    // savepoints: partial rollback inside a transaction
    conn.execute("CREATE TABLE asp (id INT NOT NULL)")
        .await
        .unwrap();
    conn.begin().await.unwrap();
    conn.execute("INSERT INTO asp VALUES (1)").await.unwrap();
    conn.savepoint("sp1").await.unwrap();
    conn.execute("INSERT INTO asp VALUES (2)").await.unwrap();
    conn.rollback_to_savepoint("sp1").await.unwrap(); // undoes (2), keeps (1)
    conn.execute("INSERT INTO asp VALUES (3)").await.unwrap();
    conn.savepoint("sp2").await.unwrap();
    conn.execute("INSERT INTO asp VALUES (4)").await.unwrap();
    conn.release_savepoint("sp2").await.unwrap(); // keeps (4)
    conn.commit().await.unwrap();
    assert_eq!(
        conn.query("SELECT id FROM asp ORDER BY id")
            .await
            .unwrap()
            .rows
            .len(),
        3
    );

    // LISTEN/NOTIFY: self-delivery of an async notification
    conn.listen("aturbo_chan").await.unwrap();
    conn.notify("aturbo_chan", Some("hi")).await.unwrap();
    let note = conn
        .poll_notification(Some(Duration::from_secs(5)))
        .await
        .unwrap()
        .expect("expected a self-delivered notification");
    assert_eq!(note.channel, "aturbo_chan");
    assert_eq!(note.payload, "hi");

    // server error surfaces, connection survives
    assert!(conn.query("SELECT * FROM nope").await.is_err());
    assert_eq!(conn.query("SELECT 1").await.unwrap().rows.len(), 1);

    // advanced shapes transport fine
    conn.query("WITH x AS (SELECT id FROM t) SELECT count(*) FROM x")
        .await
        .unwrap();

    conn.close().await;

    // pool
    let pool = AsyncPool::new(server.config(), 4).unwrap();
    let mut c = pool.get().await.unwrap();
    assert_eq!(c.execute("SELECT 1").await.unwrap(), 1);
    drop(c);
    let mut c2 = pool.get().await.unwrap();
    assert_eq!(
        c2.query("SELECT count(*) FROM t").await.unwrap().rows.len(),
        1
    );
}

/// Async counterpart of the SCRAM auth regression (R&I user-QA D2): the async auth loop must also
/// `break` once the SASL exchange completes instead of reading the post-auth `BackendKeyData` ('K').
#[tokio::test]
async fn async_scram_auth() {
    let Some(bin) = server_binary() else {
        eprintln!("SKIP: nusadb-server binary not found; run `cargo build -p nusadb-server` first");
        return;
    };
    let server = Server::start_with(&bin, &["--auth-user", "turbo:s3cret"]);

    let mut good = server.config();
    good.password = Some("s3cret".to_owned());
    let mut conn = AsyncConnection::connect_config(&good)
        .await
        .expect("SCRAM auth should succeed");
    assert_eq!(conn.query("SELECT 1").await.unwrap().rows.len(), 1);
    conn.close().await;

    let mut bad = server.config();
    bad.password = Some("wrong".to_owned());
    assert!(
        AsyncConnection::connect_config(&bad).await.is_err(),
        "wrong password must not authenticate"
    );
}

#[tokio::test]
async fn async_copy_in_and_out() {
    let Some(bin) = server_binary() else {
        eprintln!("SKIP: nusadb-server binary not found; run `cargo build -p nusadb-server` first");
        return;
    };
    let server = Server::start(&bin);
    let mut conn = AsyncConnection::connect_config(&server.config())
        .await
        .expect("connect");

    conn.execute("CREATE TABLE copy_t (id INT NOT NULL, name TEXT, PRIMARY KEY (id))")
        .await
        .unwrap();

    // COPY FROM STDIN: tab-delimited rows, `\N` for NULL.
    let mut src: &[u8] = b"1\talice\n2\t\\N\n3\tcarol\n";
    let loaded = conn
        .copy_in("COPY copy_t FROM STDIN", &mut src)
        .await
        .expect("copy in");
    assert_eq!(loaded, 3, "COPY FROM loaded three rows");

    let rows = conn
        .query("SELECT id, name FROM copy_t ORDER BY id")
        .await
        .unwrap();
    assert_eq!(rows.rows.len(), 3);
    assert_eq!(
        rows.rows[1].get::<Option<String>>(1).unwrap(),
        None,
        "row 2's name is NULL"
    );

    // COPY TO STDOUT: bytes come back in the same text format.
    let mut out: Vec<u8> = Vec::new();
    let exported = conn
        .copy_out("COPY copy_t TO STDOUT", &mut out)
        .await
        .expect("copy out");
    assert_eq!(exported, 3, "COPY TO exported three rows");
    let text = String::from_utf8(out).unwrap();
    assert_eq!(
        text.lines().collect::<Vec<_>>(),
        vec!["1\talice", "2\t\\N", "3\tcarol"]
    );

    // A refused COPY errors and the connection stays usable.
    let mut empty: &[u8] = b"";
    assert!(
        conn.copy_in("COPY no_such_table FROM STDIN", &mut empty)
            .await
            .is_err(),
        "COPY into a missing table errors"
    );
    assert_eq!(
        conn.query("SELECT count(*) FROM copy_t")
            .await
            .unwrap()
            .rows
            .len(),
        1
    );

    conn.close().await;
}