alkblobs 0.2.2

//! The postgres engine's shared test fixtures: the guarded harness (a
//! fresh schema per test, the janitor dropping it), the DSN/skip
//! plumbing, the read-seam and staged helpers, and the window/pin raw
//! probes the clause tests 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::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use parking_lot::Mutex;
use tokio::time::Instant as TokioInstant;
use tokio_postgres::NoTls;

use super::test_support::SchemaJanitor;
use crate::backend::{Backend, ReadCursor, StagingCursor};
use crate::gc::{EngineWindowState, PinRow, WindowCandidate};
use crate::pg_common::{build_pool, err, pg_err, qualified_table, PgOpts};
use crate::postgres::pool::DEFAULT_PIN_TTL;
use crate::Error;

use super::PostgresBackend;

/// The gc-window/pin rows' timestamp decoding — the test tunings'
/// fixture helper (the sqlite engine's helpers are the reference —
/// per-engine bodies, one encoding: [`crate::pg_common::time_to_nanos`]
/// is the shared forward direction).
pub(crate) fn nanos_to_time(n: i64) -> SystemTime {
    if n >= 0 {
        UNIX_EPOCH + std::time::Duration::from_nanos(n as u64)
    } else {
        UNIX_EPOCH - std::time::Duration::from_nanos((-n) as u64)
    }
}

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 schema is dropped when the guard drops — success, early
/// return, or unwind included (the janitor is declared after the
/// backend, so the pool drains first).
pub(crate) struct Harness {
    backend: PostgresBackend,
    _janitor: SchemaJanitor,
}

impl Deref for Harness {
    type Target = PostgresBackend;

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

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

/// A fresh engine-owned schema per test — the alkstore suite
/// factory's fresh-schema-per-open convention: every test opens its
/// own schema on the shared harness database, and the virgin-store
/// fixture reads a schema that was never bootstrapped at all.
pub(crate) static SCHEMA_SEQ: AtomicU64 = AtomicU64::new(0);

pub(crate) fn fresh_schema(tag: &str) -> String {
    format!(
        "abt_{}_{}_{}",
        tag,
        std::process::id(),
        SCHEMA_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 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 test-instrumented cancellation windows (the F-2/N-6 tests'
/// determinism instrument): `sweep_lock_acquire`/`sweep_lock_release`
/// park at their guarded seam — after the advisory lock's grant, after
/// the lease record's `take()` — exactly while this arm names their
/// backend's schema. Arming is per-schema-safe (fresh schema per test):
/// no other test's acquire/release can be caught by one test's armed
/// window, and the parked task releases only through its own abort.
pub(crate) static LEASE_CANCEL_WINDOW: Mutex<Option<String>> = Mutex::new(None);

pub(crate) fn arm_lease_cancel_window(schema: &str) {
    *LEASE_CANCEL_WINDOW.lock() = Some(schema.to_owned());
}

pub(crate) fn disarm_lease_cancel_window() {
    *LEASE_CANCEL_WINDOW.lock() = None;
}

/// The parked point each window's code reaches under its schema's arm
/// — a never-resolving future; the cancellation (a task `abort`) at
/// this exact await is the recorded window's deterministic form.
pub(crate) async fn lease_cancel_window_park(schema: &str) {
    if LEASE_CANCEL_WINDOW.lock().as_deref() == Some(schema) {
        std::future::pending::<()>().await;
    }
}

/// Open a backend on a fresh schema — or `None` (skip note printed)
/// when the harness server is not configured. The backend rides a
/// guard that drops the test's schema when the test ends.
pub(crate) async fn harness(tag: &str) -> Option<Harness> {
    harness_with(tag, PgOpts::default()).await
}

/// The TTL-tuned twin of [`harness`]: the backend opens through
/// `new_with_pin_ttl` (the fleet constructors' instrument), so the
/// renewal-body wiring tests the constructor-carried value.
pub(crate) async fn harness_with_ttl(tag: &str, pin_ttl: Duration) -> Option<Harness> {
    let Some(dsn) = harness_dsn() else {
        skip_note();
        return None;
    };
    let schema = fresh_schema(tag);
    let opts = PgOpts {
        schema: schema.clone(),
        ..PgOpts::default()
    };
    let backend = PostgresBackend::new_with_pin_ttl(&dsn, opts, pin_ttl)
        .await
        .expect("open");
    backend.ensure_gc_schema().await.expect("gc schema");
    Some(Harness {
        backend,
        _janitor: SchemaJanitor::new(&dsn, &schema),
    })
}

pub(crate) async fn harness_with(tag: &str, opts: PgOpts) -> Option<Harness> {
    let Some(dsn) = harness_dsn() else {
        skip_note();
        return None;
    };
    let schema = fresh_schema(tag);
    let opts = PgOpts {
        schema: schema.clone(),
        ..opts
    };
    let backend = PostgresBackend::new(&dsn, opts).await.expect("open");
    Some(Harness {
        backend,
        _janitor: SchemaJanitor::new(&dsn, &schema),
    })
}

/// The virgin-store fixture: a pool connected to a schema that was
/// never bootstrapped (the sqlite `new_tableless` pattern). The
/// guard drops the (table-less) schema at test end.
pub(crate) async fn harness_schemaless(tag: &str) -> Option<Harness> {
    let Some(dsn) = harness_dsn() else {
        skip_note();
        return None;
    };
    let schema = fresh_schema(tag);
    let backend = PostgresBackend::new_schemaless(&dsn, &schema)
        .await
        .expect("open");
    Some(Harness {
        backend,
        _janitor: SchemaJanitor::new(&dsn, &schema),
    })
}

/// The real stored value, observed through the read seam — the
/// mirror of the sqlite engine's helper.
pub(crate) async fn stored_value(backend: &PostgresBackend, 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: &PostgresBackend,
    key: &[u8],
    content: &[u8],
) -> Result<(), Error> {
    let mut staging = backend.put_staged().await?;
    staging.append(content).await?;
    staging.commit(key).await
}

/// 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: &PostgresBackend) -> 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:?}")
}

/// The live session's `application_name` — the per-instance tag's
/// wiring test's probe (the tag rides the startup parameters; a pooled
/// checkout must be asked).
pub(crate) async fn shown_application_name(backend: &PostgresBackend) -> String {
    let client = backend.pool.get().await.expect("checkout");
    let messages = client
        .simple_query("SHOW application_name")
        .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:?}")
}

/// 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 at(nanos: i64) -> SystemTime {
    nanos_to_time(nanos)
}

/// A constructed backend with the GC schema ensured — the clause
/// tests' fixture (store construction always wires the ensure call;
/// the backend constructor itself never creates the GC tables).
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,
    }
}

pub(crate) fn pin_row(key: &[u8], owner: &[u8], expiry: SystemTime) -> PinRow {
    PinRow {
        key: key.to_vec(),
        owner: owner.to_vec(),
        expiry,
    }
}

/// 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: &PostgresBackend) -> Vec<(Vec<u8>, i64)> {
    let client = backend.pool.get().await.unwrap();
    let sql = format!(
        "SELECT key, visible_after FROM {} ORDER BY key",
        backend.window_table
    );
    let rows = client.query(sql.as_str(), &[]).await.unwrap();
    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 gc_pins rows as the engine stores them — the raw encoding.
pub(crate) async fn pin_rows(backend: &PostgresBackend) -> Vec<(Vec<u8>, Vec<u8>, i64)> {
    let client = backend.pool.get().await.unwrap();
    let sql = format!(
        "SELECT key, owner, expiry FROM {} ORDER BY key, owner",
        backend.pins_table
    );
    let rows = client.query(sql.as_str(), &[]).await.unwrap();
    rows.iter()
        .map(|row| {
            (
                row.get::<_, Vec<u8>>(0),
                row.get::<_, Vec<u8>>(1),
                row.get::<_, i64>(2),
            )
        })
        .collect()
}

/// A standalone pg session outside the backend's pool — the fleet
/// tests' instrument for cross-session facts (advisory-lock
/// contention, session-loss reaping, and the embedder's own
/// schema work).
pub(crate) async fn raw_connection(
    dsn: &str,
) -> (tokio_postgres::Client, tokio::task::JoinHandle<()>) {
    let (client, conn) = tokio_postgres::connect(dsn, NoTls)
        .await
        .expect("raw harness connect");
    let handle = tokio::spawn(async move {
        let _ = conn.await;
    });
    (client, handle)
}

/// An `Arc`-shared instance of the harness backend's state — the
/// cancellation/panic tests' instrument: the same engine shape, shared
/// across spawned tasks (the abort/panic side always runs on its own
/// task). The backend's own instance state (the lease record, the
/// liveness registrations) is the clone's — fresh and empty.
pub(crate) fn arc_backend(backend: &PostgresBackend) -> Arc<PostgresBackend> {
    Arc::new(PostgresBackend {
        pool: backend.pool.clone(),
        table: backend.table.clone(),
        schema: backend.schema.clone(),
        pins_table: backend.pins_table.clone(),
        window_table: backend.window_table.clone(),
        pin_ttl: backend.pin_ttl,
        lease: Mutex::new(None),
        liveness_sources: backend.liveness_sources.clone(),
        #[cfg(test)]
        fail_commit: backend.fail_commit,
        #[cfg(test)]
        fail_joint_commit: backend.fail_joint_commit,
        #[cfg(test)]
        fail_window_stage: backend.fail_window_stage,
    })
}

/// Whether the 64-bit advisory key is granted to any session, read
/// from the `pg_locks` catalog (the probe is read-only — it never
/// contends with the observed acquirer; the key's catalog encoding is
/// the two 32-bit halves, `classid` high / `objid` low, compared as
/// text to dodge the `oid`-typed column's client encoding).
pub(crate) async fn advisory_held(client: &tokio_postgres::Client, key: i64) -> bool {
    let high = ((key as u64) >> 32) as u32;
    let low = ((key as u64) & 0xFFFF_FFFF) as u32;
    client
        .query_one(
            "SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND granted \
             AND classid::text = $1 AND objid::text = $2",
            &[&high.to_string(), &low.to_string()],
        )
        .await
        .expect("advisory-lock probe")
        .get::<_, i64>(0)
        != 0
}

/// Poll until the advisory lock key's holdership state reaches
/// `want` — bounded; a miss is a panic naming the missed step.
pub(crate) async fn wait_for_advisory(
    client: &tokio_postgres::Client,
    key: i64,
    want: bool,
    what: &str,
) {
    let deadline = TokioInstant::now() + std::time::Duration::from_secs(15);
    loop {
        if advisory_held(client, key).await == want {
            return;
        }
        if TokioInstant::now() >= deadline {
            panic!("{what}: the advisory lock never reached want = {want}");
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
}

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

#[cfg(test)]
impl PostgresBackend {
    /// Open a pool without schema creation — the virgin-store
    /// tolerance test's instrument (constructor-vs-lazy is a recorded
    /// design point; the read paths' no-schema-yet semantics are tested
    /// against a schema that was never bootstrapped). The sqlite
    /// `new_tableless` pattern.
    #[cfg(test)]
    pub(crate) async fn new_schemaless(config: &str, schema: &str) -> Result<Self, Error> {
        let pool = build_pool(config, &PgOpts::default())?;
        let schema = schema.to_owned();
        let pins_table = qualified_table(&schema, "gc_pins");
        let window_table = qualified_table(&schema, "gc_window");
        Ok(Self {
            pool,
            table: qualified_table(&schema, "blobs"),
            schema,
            pins_table,
            window_table,
            pin_ttl: DEFAULT_PIN_TTL,
            lease: Mutex::new(None),
            liveness_sources: Arc::new(Mutex::new(Vec::new())),
            #[cfg(test)]
            fail_commit: false,
            #[cfg(test)]
            fail_joint_commit: false,
            #[cfg(test)]
            fail_window_stage: false,
        })
    }
}

impl PostgresBackend {
    /// The catalog table-existence probe — test-only: the store
    /// construction test's instrument (mirrors the sqlite engine's
    /// `sqlite_master` probe), answering for the engine-owned schema
    /// the backend was constructed with.
    #[cfg(test)]
    pub(crate) async fn table_exists(&self, table: &str) -> Result<bool, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let row = client
            .query_one(
                "SELECT EXISTS (
                    SELECT 1 FROM pg_tables WHERE schemaname = $1 AND tablename = $2
                )",
                &[&self.schema, &table],
            )
            .await
            .map_err(pg_err)?;
        let present: bool = row.try_get(0).map_err(pg_err)?;
        Ok(present)
    }
}