//! Concurrency spike for the store read pool: one turso `Database` with several connections.
//! Checks read-after-commit visibility, isolation from an open writer transaction, and that
//! concurrent readers next to a busy writer neither error nor report "database is locked".

use std::sync::Arc;
use std::time::Duration;

use turso::{Builder, Connection, Database};

const BUSY_TIMEOUT: Duration = Duration::from_millis(5000);

async fn open_db(dir: &tempfile::TempDir) -> Database {
    let path = dir.path().join("spike.db");
    let db = Builder::new_local(path.to_str().expect("utf-8 path"))
        .build()
        .await
        .expect("build database");
    let writer = db.connect().expect("writer connect");
    let mut rows = writer
        .query("PRAGMA journal_mode = WAL", ())
        .await
        .expect("wal");
    while rows.next().await.expect("wal row").is_some() {}
    writer
        .execute(
            "CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, v INTEGER NOT NULL)",
            (),
        )
        .await
        .expect("create table");
    db
}

/// A connection configured the way the store's readers will be: busy timeout and query_only.
async fn reader_conn(db: &Database) -> Connection {
    let conn = db.connect().expect("reader connect");
    conn.busy_timeout(BUSY_TIMEOUT).expect("busy timeout");
    let mut rows = conn
        .query("PRAGMA query_only = 1", ())
        .await
        .expect("query_only");
    while rows.next().await.expect("query_only row").is_some() {}
    conn
}

fn writer_conn(db: &Database) -> Connection {
    let conn = db.connect().expect("writer connect");
    conn.busy_timeout(BUSY_TIMEOUT).expect("busy timeout");
    conn
}

async fn count(conn: &Connection) -> i64 {
    let mut rows = conn
        .query("SELECT COUNT(*) FROM t", ())
        .await
        .expect("count query");
    let row = rows.next().await.expect("count row").expect("one row");
    row.get::<i64>(0).expect("count value")
}

#[tokio::test]
async fn reader_sees_committed_rows_immediately() {
    let dir = tempfile::tempdir().unwrap();
    let db = open_db(&dir).await;
    let writer = writer_conn(&db);
    let reader = reader_conn(&db).await;

    assert_eq!(count(&reader).await, 0);
    writer
        .execute("INSERT INTO t (v) VALUES (1)", ())
        .await
        .expect("insert");
    assert_eq!(
        count(&reader).await,
        1,
        "reader must see the autocommitted row"
    );
}

#[tokio::test]
async fn reader_does_not_see_uncommitted_rows_and_does_not_block() {
    let dir = tempfile::tempdir().unwrap();
    let db = open_db(&dir).await;
    let writer = writer_conn(&db);
    let reader = reader_conn(&db).await;

    writer.execute("BEGIN", ()).await.expect("begin");
    writer
        .execute("INSERT INTO t (v) VALUES (2)", ())
        .await
        .expect("insert in tx");
    writer
        .execute("INSERT INTO t (v) VALUES (3)", ())
        .await
        .expect("insert in tx");

    // The writer still holds its transaction open here.
    let seen = tokio::time::timeout(BUSY_TIMEOUT, count(&reader))
        .await
        .expect("reader must not block on an open writer transaction");
    assert_eq!(seen, 0, "reader must not see uncommitted rows");

    writer.execute("COMMIT", ()).await.expect("commit");
    assert_eq!(count(&reader).await, 2, "reader must see rows after commit");
}

#[tokio::test]
async fn eight_readers_with_busy_writer_all_succeed() {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(open_db(&dir).await);
    let writer = writer_conn(&db);

    let mut readers = Vec::new();
    for _ in 0..8 {
        let db = Arc::clone(&db);
        readers.push(tokio::spawn(async move {
            let conn = reader_conn(&db).await;
            let mut last = 0;
            for _ in 0..200 {
                let n = count(&conn).await;
                assert!(n >= last, "reader saw count go backwards: {last} -> {n}");
                last = n;
            }
            last
        }));
    }

    for i in 0..200 {
        writer
            .execute("INSERT INTO t (v) VALUES (?1)", (i as i64,))
            .await
            .expect("writer insert must not fail");
    }

    for handle in readers {
        handle.await.expect("reader task panicked");
    }
    assert_eq!(count(&reader_conn(&db).await).await, 200);
}

#[tokio::test]
async fn reader_rejects_writes_when_query_only() {
    let dir = tempfile::tempdir().unwrap();
    let db = open_db(&dir).await;
    let reader = reader_conn(&db).await;
    let result = reader.execute("INSERT INTO t (v) VALUES (9)", ()).await;
    assert!(result.is_err(), "query_only reader must refuse writes");
    assert_eq!(count(&reader).await, 0);
}