alkblobs 0.1.0

//! The crate-internal pg-common module — the shared-helper design point
//! resolved (engine-pg-lo): the pieces both postgres-attached engines
//! (`postgres` kv, `pg-lo` large) share, gated
//! `cfg(any(feature = "postgres", feature = "pg-lo"))` so each engine's
//! pure build (ADR-013 §4: `pg-lo` does not imply `postgres`; the
//! pg-lo-pure build) compiles without duplication. Held here: the
//! deadpool pool construction (the shipped driver posture),
//! [`quote_identifier`] (the quoted-identifier boundary), the SQLSTATE
//! 42P01 (undefined table) virgin-store discriminant, the error mapping,
//! the metadata-size clamp, the i64-nanos timestamp encoding, and
//! [`PgOpts`] — the shared construction-opts type whose public
//! re-export gate is widened to `any(postgres, pg-lo)` with the
//! definition's move here (the type is unchanged; `alkblobs::PgOpts`
//! now exists in the pg-lo-pure build too).
//!
//! The error mapping keeps the `postgres:` prefix for both engines: each
//! attaches to a PostgreSQL server, so family attribution reads the same
//! from either engine's failures.

use std::time::{SystemTime, UNIX_EPOCH};

use deadpool_postgres::{Manager, ManagerConfig, Object, Pool, RecyclingMethod};
use tokio_postgres::NoTls;

use crate::Error;

/// The error mapping's base form (the postgres engine's): a failure
/// surfaces as [`Error::Io`] with the engine family prefix.
pub(crate) fn err(e: impl std::fmt::Display) -> Error {
    Error::Io(format!("postgres: {e}"))
}

/// The engine's error mapping for statement bodies: a SQL-level failure
/// surfaces with its server-side message and SQLSTATE (the `DbError`
/// detail `Display` drops), anything else through [`err`].
pub(crate) fn pg_err(e: tokio_postgres::Error) -> Error {
    match e.as_db_error() {
        Some(db) => Error::Io(format!("postgres: {db}")),
        None => err(e),
    }
}

/// The virgin-store tolerance's discriminant (POC #1 finding 2): a
/// statement naming a table that does not exist maps on the read paths
/// to absent/empty (and delete to Missing), never an error. The stable
/// surface PostgreSQL carries is the SQLSTATE class+code (42P01,
/// undefined table); a missing *schema* reads identically.
pub(crate) fn is_undefined_table(e: &tokio_postgres::Error) -> bool {
    e.code() == Some(&tokio_postgres::error::SqlState::UNDEFINED_TABLE)
}

/// Quote a SQL identifier (the Postgres rule: double the embedded `"`).
///
/// The engines compose schema-qualified SQL from consumer-supplied
/// schema names; this is the quoted-identifier boundary — an identifier
/// is never interpolated into SQL text unquoted (the alkstore
/// `quote_identifier` pattern).
pub(crate) fn quote_identifier(name: &str) -> String {
    let mut quoted = String::with_capacity(name.len() + 2);
    quoted.push('"');
    for c in name.chars() {
        if c == '"' {
            quoted.push('"');
        }
        quoted.push(c);
    }
    quoted.push('"');
    quoted
}

/// A schema-qualified table reference: `"schema"."table"`.
pub(crate) fn qualified_table(schema: &str, table: &str) -> String {
    format!("{}.{}", quote_identifier(schema), quote_identifier(table))
}

/// The default engine-owned schema name (the recorded design point's
/// default). Both postgres-attached engines default to it — the
/// consolidation shape (ADR-012 §4's same-pg-instance rule) holds both
/// engines' tables in one engine-owned schema without name collision
/// (`blobs`, `lo_entries`, and the GC tables are distinct names).
pub(crate) const DEFAULT_SCHEMA: &str = "alkblobs";

/// The shipped default pool size ([`PgOpts::max_size`]) — the alkstore
/// `DEFAULT_MAX_SIZE = 8` precedent.
pub(crate) const DEFAULT_MAX_SIZE: usize = 8;

/// The `size` metadata's integer, clamped to the domain the trait probes
/// in.
pub(crate) fn size_u64(raw: i64) -> u64 {
    raw.max(0) as u64
}

/// The crate-wide timestamp encoding (the recorded design point): signed
/// i64 nanoseconds from the Unix epoch — sub-second precision for the
/// test tunings, negative for pre-epoch instants, i64 across the full
/// range a `BIGINT` stores (±292 years around the epoch).
pub(crate) fn time_to_nanos(t: SystemTime) -> i64 {
    match t.duration_since(UNIX_EPOCH) {
        Ok(d) => d.as_nanos() as i64,
        Err(e) => -(e.duration().as_nanos() as i64),
    }
}

/// Build the deadpool pool with the shipped posture: per-session
/// `synchronous_commit` merged into every connection's connect options
/// (appended to the consumer's own `options=` DSN values — postgres
/// parses repeated `-c` flags), the per-instance
/// [`PgOpts::application_name`] tag on the pool's startup parameters,
/// `RecyclingMethod::Fast`, the configured `max_size`. `max_size = 0`
/// fails at entry, before any round trip.
// consumer: the two SQL engines' constructors (postgres pool, pg-lo
// `new`) and the engines' test support — a pg-lo-only build compiles
// it dead with the engines' construction entry points.
#[cfg_attr(
    not(any(feature = "postgres", all(feature = "pg-lo", feature = "mem"))),
    allow(dead_code)
)]
pub(crate) fn build_pool(config: &str, opts: &PgOpts) -> Result<Pool, Error> {
    if opts.max_size == 0 {
        return Err(Error::Io(format!(
            "postgres: max_size must be positive, got {}",
            opts.max_size
        )));
    }
    let mut cfg: tokio_postgres::Config = config.parse().map_err(err)?;
    let engine_opt = format!(
        "-c synchronous_commit={}",
        if opts.synchronous_commit { "on" } else { "off" }
    );
    let merged = match cfg.get_options() {
        Some(existing) => format!("{existing} {engine_opt}"),
        None => engine_opt,
    };
    cfg.options(merged);
    cfg.application_name(&opts.application_name);
    let manager = Manager::from_config(
        cfg,
        NoTls,
        ManagerConfig {
            recycling_method: RecyclingMethod::Fast,
        },
    );
    Pool::builder(manager)
        .max_size(opts.max_size)
        .build()
        .map_err(err)
}

/// The best-effort cleanup a dropped held checkout owes — the shared
/// defense both engines spill to: a checkout released while it still
/// owes a SQL termination (a staged cursor's ROLLBACK, a joint tx's
/// ROLLBACK, an advisory unlock) cannot run it on the dropping path (a
/// `Drop` body is synchronous) — the statement is handed to a task
/// spawned on the current runtime, or, with no runtime live (a Drop
/// after the executor is gone), to a one-shot thread running its own
/// current-thread runtime (the POC's `release_client` shape;
/// deadpool's `Fast` recycle never resets transaction state — the
/// recorded pitfall the shape exists for). The result is ignored by
/// contract: the cleanup is best-effort, and a failing session is the
/// pool's discard case.
pub(crate) fn detached_batch_execute(client: Object, sql: String) {
    match tokio::runtime::Handle::try_current() {
        Ok(handle) => {
            handle.spawn(async move {
                let _ = client.batch_execute(&sql).await;
            });
        }
        Err(_) => {
            let _ = std::thread::spawn(move || {
                let Ok(rt) = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                else {
                    return;
                };
                rt.block_on(async move {
                    let _ = client.batch_execute(&sql).await;
                });
            })
            .join();
        }
    }
}

/// Construction options shared by the crate's postgres-attached engines
/// (`postgres` kv, `pg-lo` large) — the PgOpts design point resolved:
/// one type, its public re-export gate widened to
/// `any(postgres, pg-lo)` alongside the definition's move to this
/// module (ADR-013 §4's pg-lo-pure build carries the same opts surface).
/// Durability knobs, pool sizing, and the engine-owned schema name are
/// engine-configuration concerns — they ride the store constructors'
/// parameters, never invented defaults (the alkstore `PgOpts`
/// precedent; `open(config, PgOpts)` adapted to the named-constructor
/// shape — the constructors are [`crate::Store::dual_tier_postgres_local`]
/// and [`crate::Store::kv_only_postgres`]).
///
/// The connection settings themselves do not ride this struct — they
/// are the constructor's separate connection-settings argument (the
/// tokio-postgres [`tokio_postgres::Config`]-parseable form); the
/// engine never invents defaults for them. TLS: the engines hardwire
/// `NoTls` on every connection path regardless of the config's
/// sslmode — v1 TLS is a post-v1 deployment concern, mirroring the
/// alkstore posture.
#[derive(Debug, Clone)]
pub struct PgOpts {
    /// The engine-owned PostgreSQL schema holding the engine's tables.
    /// Both engines default to `DEFAULT_SCHEMA` (the consolidation
    /// shape holds both engines' tables in one schema); an empty string
    /// selects the default. The name is a quoted-identifier boundary
    /// (`quote_identifier`) — reserved-word and hostile names are
    /// safe.
    pub schema: String,
    /// Upper bound on pooled connections (deadpool's `max_size`).
    /// Default `DEFAULT_MAX_SIZE` = 8. A value of `0` fails
    /// construction immediately (validated at entry, no round trip —
    /// deadpool cannot grant checkouts from an empty pool; the
    /// alkstore precedent).
    pub max_size: usize,
    /// The per-session `synchronous_commit` durability knob, wired as a
    /// connect option (`-c synchronous_commit=…`) on every pooled
    /// connection (POC-verified per-session SET mechanics; the
    /// consumer's own `options=` DSN values are preserved alongside —
    /// postgres parses repeated `-c` flags). Default `true` — the ship
    /// config; `false` is the documented sqlite-NORMAL-alike latency
    /// trade (B5 ~2-3x). Both engines carry the knob; the pg-lo POC ran
    /// its server with `synchronous_commit=off` to measure LO shape
    /// without disk fsync dominating (POC #7), and the shipped knob
    /// posture is the kv engine's parity default `on`.
    pub synchronous_commit: bool,
    /// The pg-lo engine's orphan-reap declaration (ADR-016): the
    /// operator's explicit assertion "this database hosts no other
    /// large-object consumer" — the deployment-verified-invariant
    /// pattern (ADR-012 §3's `fleet: true` class), the cross-database
    /// truth the constructor cannot prove and must therefore never
    /// probe for. `pg_largeobject` is database-level, not schema-level,
    /// while the orphan-recovery sweep's claim probe is
    /// companion-table-scoped, so an undeclared sweep can unlink
    /// another large-object consumer's committed LOs co-tenning the
    /// same database (a second pg-lo pool under a different schema, or
    /// an arbitrary `pg_largeobject` user); the sweep therefore refuses
    /// with a named error while the declaration is absent. Default
    /// `false` — refuse. Deployment shapes needing two large-object
    /// consumers use two databases (or two instances): a topology
    /// fact, not a configuration knob; two consumers sharing *one*
    /// pool (two apps' liveness sources on one store instance) are
    /// unaffected. Read only by the pg-lo engine — the field is dormant
    /// on the postgres kv engine.
    pub orphan_reap_permitted: bool,
    /// The per-instance `application_name` tag carried by every pooled
    /// connection's startup parameters (the connect-config parameter,
    /// so it survives the pool's `Fast` recycle verbatim — never a
    /// session-state reset). Default `"alkblobs"` — the stable
    /// crate-identifying name an operator's `pg_stat_activity` sees.
    /// Tests pass a unique tag per instance (their fresh
    /// schema/database name): any server-global observability probe
    /// (`pg_stat_activity` and friends) filters on it, so one test's
    /// dirty-session zero-asserts are exact instead of counting every
    /// concurrently-running test's transient sessions. The pool's own
    /// observability surface — never a security or scope boundary.
    pub application_name: String,
}

impl Default for PgOpts {
    fn default() -> Self {
        Self {
            schema: DEFAULT_SCHEMA.to_owned(),
            max_size: DEFAULT_MAX_SIZE,
            synchronous_commit: true,
            orphan_reap_permitted: false,
            application_name: "alkblobs".to_owned(),
        }
    }
}

/// Test-support helpers shared with the crate's postgres-attached test
/// modules (`src/postgres/`'s and `src/pg_lo/`'s test suites,
/// `src/store.rs`'s construction tests) — crate-internal and compiled
/// out of every non-test build. Each janitor is gated by the feature
/// whose tests consume it.
#[cfg(test)]
pub(crate) mod test_support {
    use super::quote_identifier;
    use tokio_postgres::NoTls;

    /// Drops a test's engine-owned schema when the test ends — success,
    /// early return, or unwind (Drop runs in all three), so a fresh
    /// schema per test never accumulates on the shared harness
    /// database. No async teardown exists at Drop, so the drop executes
    /// synchronously from a std thread over a one-shot current-thread
    /// runtime; an unreachable harness server is tolerated (the residue
    /// stays; nothing else breaks).
    ///
    /// Construct it (or a wrapper holding it) *before* the store or
    /// backend it outlives: drop order is reverse of declaration, so
    /// the backend's pool drains before the schema drops.
    #[cfg(feature = "postgres")]
    pub(crate) struct SchemaJanitor {
        dsn: String,
        schema: String,
    }

    #[cfg(feature = "postgres")]
    impl SchemaJanitor {
        pub(crate) fn new(dsn: &str, schema: &str) -> SchemaJanitor {
            SchemaJanitor {
                dsn: dsn.to_owned(),
                schema: schema.to_owned(),
            }
        }
    }

    #[cfg(feature = "postgres")]
    impl Drop for SchemaJanitor {
        fn drop(&mut self) {
            let dsn = self.dsn.clone();
            let schema = self.schema.clone();
            let _ = std::thread::spawn(move || {
                let Ok(rt) = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                else {
                    return;
                };
                rt.block_on(async move {
                    let Ok((client, conn)) = tokio_postgres::connect(&dsn, NoTls).await else {
                        return;
                    };
                    tokio::spawn(conn);
                    let sql = format!(
                        "DROP SCHEMA IF EXISTS {} CASCADE",
                        quote_identifier(&schema)
                    );
                    let _ = client.execute(sql.as_str(), &[]).await;
                });
            })
            .join();
        }
    }

    /// Drops a test's dedicated database when the test ends — the
    /// pg-lo engine tests' janitor. Large Objects are **database-level**
    /// (the C7 lesson, recorded design point): the schema-per-test
    /// convention does not isolate LOs across tests sharing a database,
    /// and a `DROP SCHEMA` leaves the LO's pages in `pg_largeobject` —
    /// so the pg-lo tests isolate per dedicated database (the POC's
    /// proven `open_fresh` shape) and this guard removes it, zero LO
    /// residue on the shared harness database by construction.
    /// Same Drop mechanics as [`SchemaJanitor`]; `WITH (FORCE)` reaps
    /// any connection the pool has not yet drained (postgres ≥ 13).
    #[cfg(feature = "pg-lo")]
    pub(crate) struct DatabaseJanitor {
        admin_dsn: String,
        db: String,
    }

    #[cfg(feature = "pg-lo")]
    impl DatabaseJanitor {
        /// `admin_dsn` must name a database to connect *through* (the
        /// `postgres` maintenance database); `db` is the test's
        /// dedicated database to drop.
        pub(crate) fn new(admin_dsn: &str, db: &str) -> DatabaseJanitor {
            DatabaseJanitor {
                admin_dsn: admin_dsn.to_owned(),
                db: db.to_owned(),
            }
        }
    }

    #[cfg(feature = "pg-lo")]
    impl Drop for DatabaseJanitor {
        fn drop(&mut self) {
            let admin_dsn = self.admin_dsn.clone();
            let db = self.db.clone();
            let _ = std::thread::spawn(move || {
                let Ok(rt) = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                else {
                    return;
                };
                rt.block_on(async move {
                    let Ok((client, conn)) = tokio_postgres::connect(&admin_dsn, NoTls).await
                    else {
                        return;
                    };
                    tokio::spawn(conn);
                    let sql = format!(
                        "DROP DATABASE IF EXISTS {} WITH (FORCE)",
                        quote_identifier(&db)
                    );
                    let _ = client.execute(sql.as_str(), &[]).await;
                });
            })
            .join();
        }
    }
}