alkblobs 0.2.1

//! The pg-lo engine's shared test fixtures: the per-test dedicated
//! database harness (LOs are database-level — the C7 lesson), the DSN
//! plumbing, the read-seam/staged helpers, and the catalog probes the
//! zero-residue gates share. `new_schemaless` (the virgin-store
//! constructor instrument) lives here too — it is test-only.

use std::ops::{Deref, DerefMut};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::SystemTime;

use tokio_postgres::NoTls;

use crate::backend::{Backend, ReadCursor, StagingCursor};
use crate::gc::{EngineWindowState, WindowCandidate};
use crate::pg_common::test_support::DatabaseJanitor;
use crate::pg_common::{build_pool, qualified_table, quote_identifier, PgOpts};
use crate::Error;

use super::PgLoBackend;

pub(crate) const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";

/// The guarded harness backend: derefs to the engine, and the test's
/// own database is dropped when the guard drops — success, early
/// return, or unwind included (the janitor is declared after the
/// backend, so the pool drains before the database drops).
pub(crate) struct Harness {
    /// The per-test database's DSN (test-only probes: e.g. opening a
    /// second engine over the same database with hostile opts).
    pub(crate) dsn: String,
    backend: PgLoBackend,
    _janitor: DatabaseJanitor,
}

impl Deref for Harness {
    type Target = PgLoBackend;

    fn deref(&self) -> &PgLoBackend {
        &self.backend
    }
}

impl DerefMut for Harness {
    fn deref_mut(&mut self) -> &mut PgLoBackend {
        &mut self.backend
    }
}

/// A unique per-test database name — LOs are database-level (the C7
/// lesson), so every test owns a dedicated database on the shared
/// harness server; the pid + sequence + wall-clock components make
/// the name collision-free across parallel tests and reruns.
pub(crate) static DB_SEQ: AtomicU64 = AtomicU64::new(0);

pub(crate) fn fresh_db(tag: &str) -> String {
    let nanos = SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    format!(
        "abtlo_{}_{}_{}_{nanos:x}",
        tag,
        std::process::id(),
        DB_SEQ.fetch_add(1, Ordering::Relaxed)
    )
}

pub(crate) fn env_nonempty(name: &str) -> Option<String> {
    std::env::var(name).ok().filter(|v| !v.is_empty())
}

/// The harness connection settings ride the environment
/// (`ALKBLOBS_PG_HOST` required; port/user/password/db default to
/// the harness server's values). Absent → the caller skips.
pub(crate) fn harness_dsn() -> Option<String> {
    let host = env_nonempty("ALKBLOBS_PG_HOST")?;
    Some(format!(
        "host={host} port={} user={} password={} dbname={}",
        env_nonempty("ALKBLOBS_PG_PORT").unwrap_or_else(|| "15432".to_owned()),
        env_nonempty("ALKBLOBS_PG_USER").unwrap_or_else(|| "postgres".to_owned()),
        env_nonempty("ALKBLOBS_PG_PASSWORD").unwrap_or_else(|| "poc".to_owned()),
        env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()),
    ))
}

/// The maintenance-database DSN the tests connect *through* to
/// create/drop the per-test databases (the same server credentials).
pub(crate) fn harness_admin_dsn() -> Option<String> {
    let dsn = harness_dsn()?;
    Some(format!("{dsn} dbname=postgres"))
}

pub(crate) fn dsn_for_db(db: &str) -> String {
    let dsn = harness_dsn().expect("harness dsn");
    format!(
        "{} dbname={db}",
        dsn.rsplit_once("dbname=").expect("dbname").0
    )
}

/// The skip note — printed exactly once per skipped test.
pub(crate) fn skip_note() {
    eprintln!("skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB");
}

/// The POC's proven `open_fresh` shape: create the test's dedicated
/// database (the LO-level isolation decision), then open the engine
/// inside it. No engine state is left on the shared harness
/// database; the janitor in [`Harness`] removes the database.
pub(crate) async fn create_db(admin_dsn: &str, db: &str) {
    let (client, conn) = tokio_postgres::connect(admin_dsn, NoTls)
        .await
        .expect("admin connect");
    tokio::spawn(async move {
        let _ = conn.await;
    });
    client
        .execute(&format!("CREATE DATABASE {}", quote_identifier(db)), &[])
        .await
        .expect("create test database");
}

pub(crate) async fn harness(tag: &str) -> Option<Harness> {
    harness_with(tag, PgOpts::default()).await
}

pub(crate) async fn harness_with(tag: &str, opts: PgOpts) -> Option<Harness> {
    let admin = harness_admin_dsn()?;
    let db = fresh_db(tag);
    create_db(&admin, &db).await;
    let dsn = dsn_for_db(&db);
    let backend = PgLoBackend::new(&dsn, opts).await.expect("open backend");
    Some(Harness {
        dsn,
        backend,
        _janitor: DatabaseJanitor::new(&admin, &db),
    })
}

/// The virgin-store fixture: a dedicated database whose schema was
/// never bootstrapped (the postgres engine's `new_schemaless`
/// pattern). The janitor still owns the whole database.
pub(crate) async fn harness_schemaless(tag: &str) -> Option<Harness> {
    let admin = harness_admin_dsn()?;
    let db = fresh_db(tag);
    create_db(&admin, &db).await;
    let dsn = dsn_for_db(&db);
    let backend = PgLoBackend::new_schemaless(&dsn, "alkblobs")
        .await
        .expect("open schemaless backend");
    Some(Harness {
        dsn,
        backend,
        _janitor: DatabaseJanitor::new(&admin, &db),
    })
}

/// The real stored value, observed through the read seam — the
/// mirror of the postgres engine's helper.
pub(crate) async fn stored_value(backend: &PgLoBackend, key: &[u8]) -> Option<Vec<u8>> {
    let cursor = backend.get(key).await.expect("get succeeds")?;
    let (bytes, _) = cursor.read_range(0, cursor.len()).await.expect("read");
    Some(bytes)
}

pub(crate) async fn staged(backend: &PgLoBackend, key: &[u8], content: &[u8]) -> Result<(), Error> {
    let mut staging = backend.put_staged().await?;
    staging.append(content).await?;
    staging.commit(key).await
}

pub(crate) async fn staged_in_64k_windows(
    backend: &PgLoBackend,
    key: &[u8],
    content: &[u8],
) -> Result<(), Error> {
    let mut staging = backend.put_staged().await?;
    for chunk in content.chunks(64 * 1024) {
        staging.append(chunk).await?;
    }
    staging.commit(key).await
}

/// The LO catalog's distinct oids — the tests' instrumentation for
/// the zero-residue gates and the catalog-agreement checks. The
/// engine never reads this for addressing; tests may.
pub(crate) async fn catalog_loids(backend: &PgLoBackend) -> Vec<u32> {
    let client = backend.pool.get().await.expect("checkout");
    let rows = client
        .query("SELECT DISTINCT loid FROM pg_largeobject", &[])
        .await
        .expect("catalog loids");
    let mut loids: Vec<u32> = rows.iter().map(|row| row.get::<_, u32>(0)).collect();
    loids.sort();
    loids
}

/// The LO catalog's page-row count — the exact-count residue gate
/// (no LO pages, no residue).
pub(crate) async fn catalog_rows(backend: &PgLoBackend) -> i64 {
    let client = backend.pool.get().await.expect("checkout");
    client
        .query_one("SELECT count(*) FROM pg_largeobject", &[])
        .await
        .expect("catalog count")
        .get::<_, i64>(0)
}

/// The companion table's oids, sorted — the truth side of the
/// catalog-agreement check.
pub(crate) async fn companion_loids_sorted(backend: &PgLoBackend) -> Vec<u32> {
    let client = backend.pool.get().await.expect("checkout");
    let sql = format!("SELECT loid FROM {} ORDER BY loid", backend.table);
    let rows = client
        .query(sql.as_str(), &[])
        .await
        .expect("companion loids");
    rows.iter().map(|row| row.get::<_, u32>(0)).collect()
}

/// The live session's `synchronous_commit` — the parity-knob
/// posture test's probe (the knob wires per session; a pooled
/// checkout must be asked).
pub(crate) async fn shown_synchronous_commit(backend: &PgLoBackend) -> String {
    let client = backend.pool.get().await.expect("checkout");
    let messages = client
        .simple_query("SHOW synchronous_commit")
        .await
        .expect("SHOW");
    for message in &messages {
        if let tokio_postgres::SimpleQueryMessage::Row(row) = message {
            return row.get(0).expect("SHOW row").to_owned();
        }
    }
    panic!("SHOW returned no row: {messages:?}")
}

/// Patterned content large enough to span multiple 512 KiB windows.
pub(crate) fn pattern(len: usize) -> Vec<u8> {
    (0..len).map(|i| (i % 251) as u8).collect()
}

/// The LO catalog's page-row count for a content of `len` bytes —
/// ceil(len/2048), C2's catalog math.
pub(crate) fn pages_for(len: usize) -> i64 {
    (len as i64 + 2047) / 2048
}

/// The window-family helpers (the postgres engine's suite is the
/// reference — same semantics, this dialect's harness). A
/// fixed-instant deadline family for the window tests (epoch-based
/// signed nanos, so pre-epoch coverage rides the same encoding):
/// -10s (past, pre-epoch), +1s (past), +2s (future), +1h (far
/// future).
pub(crate) fn nanos_to_time(n: i64) -> SystemTime {
    if n >= 0 {
        std::time::UNIX_EPOCH + std::time::Duration::from_nanos(n as u64)
    } else {
        std::time::UNIX_EPOCH - std::time::Duration::from_nanos((-n) as u64)
    }
}

pub(crate) fn at(nanos: i64) -> SystemTime {
    nanos_to_time(nanos)
}

/// A constructed backend with the GC schema ensured — the clause
/// tests' fixture (store construction wires the ensure call where
/// the one-window-host rule makes the window active; the engine
/// constructor itself never creates the GC table).
pub(crate) async fn harness_gc(tag: &str) -> Option<Harness> {
    let backend = harness(tag).await?;
    backend.ensure_gc_schema().await.unwrap();
    Some(backend)
}

pub(crate) fn stages(key: &[u8], visible_after: SystemTime) -> WindowCandidate {
    WindowCandidate {
        key: key.to_vec(),
        visible_after,
    }
}

/// The gc_window rows as the engine stores them — the raw encoding,
/// read back through the engine-owned table itself.
pub(crate) async fn window_rows(backend: &PgLoBackend) -> Vec<(Vec<u8>, i64)> {
    let client = backend.pool.get().await.expect("checkout");
    let sql = format!(
        "SELECT key, visible_after FROM {} ORDER BY key",
        backend.window_table
    );
    let rows = client.query(sql.as_str(), &[]).await.expect("window rows");
    let mut out: Vec<(Vec<u8>, i64)> = rows
        .iter()
        .map(|row| (row.get::<_, Vec<u8>>(0), row.get::<_, i64>(1)))
        .collect();
    out.sort();
    out
}

/// The engine-owned schema's full table list — the schema-ownership
/// test's instrument (exactly the engine's tables, nothing more).
pub(crate) async fn schema_tables(backend: &PgLoBackend) -> Vec<String> {
    let client = backend.pool.get().await.expect("checkout");
    let rows = client
        .query(
            "SELECT table_name::text FROM information_schema.tables \
                 WHERE table_schema = $1 ORDER BY table_name",
            &[&backend.schema],
        )
        .await
        .expect("schema tables");
    rows.iter().map(|row| row.get::<_, String>(0)).collect()
}

#[cfg(test)]
impl PgLoBackend {
    /// Open a pool without schema creation — the virgin-store tolerance
    /// test's instrument (the postgres engine's `new_schemaless`
    /// pattern): the read paths' no-schema-yet semantics are tested
    /// against a schema that was never bootstrapped.
    #[cfg(test)]
    pub(crate) async fn new_schemaless(config: &str, schema: &str) -> Result<Self, Error> {
        let pool = build_pool(config, &PgOpts::default())?;
        Ok(Self {
            pool,
            table: qualified_table(schema, "lo_entries"),
            window_table: qualified_table(schema, "gc_window"),
            schema: schema.to_owned(),
            orphan_reap_permitted: false,
            #[cfg(test)]
            fail_commit: false,
            #[cfg(test)]
            fail_window_stage: false,
        })
    }
}