alkblobs 0.2.1

//! The engine-owned schema tier: the bootstrap DDL (the B5 row shape)
//! and the GC schema's DDL (the fleet host's both-tables row), plus the
//! test-only catalog probe. The constructor and `ensure_gc_schema` both
//! compose through these statements; the session options themselves
//! (the `synchronous_commit` knob, pool sizing, the schema name) ride
//! [`crate::pg_common::PgOpts`] and [`crate::pg_common::build_pool`] —
//! the busy-timeout-analogue posture is the shipped driver posture there,
//! nothing engine-local to tune.

use crate::pg_common::{qualified_table, quote_identifier};

/// The engine-owned bootstrap DDL — idempotent, no migration machinery.
/// The B5 row shape under the measured `fillfactor = 90`; column names
/// quote-follow the alkstore hazard-list discipline (`"key"` quoted;
/// `value`/`size` non-reserved).
pub(super) fn bootstrap_ddl(schema: &str) -> String {
    let schema_q = quote_identifier(schema);
    let table = qualified_table(schema, "blobs");
    format!(
        "CREATE SCHEMA IF NOT EXISTS {schema_q};

CREATE TABLE IF NOT EXISTS {table} (
    \"key\" bytea PRIMARY KEY,
    value bytea NOT NULL,
    size bigint NOT NULL
) WITH (fillfactor = 90);
"
    )
}

/// The GC schema's DDL — idempotent, the fleet host's both-tables row
/// (ADR-012 §5): `gc_pins` with its `(key, owner)` row identity and
/// `gc_window` keyed by the candidate key, both timestamps `BIGINT`
/// nanos. The embedder's liveness table is never named here (the
/// schema-ownership rule).
pub(super) fn gc_ddl(schema: &str) -> String {
    let schema_q = quote_identifier(schema);
    let pins = qualified_table(schema, "gc_pins");
    let window = qualified_table(schema, "gc_window");
    format!(
        "CREATE SCHEMA IF NOT EXISTS {schema_q};

CREATE TABLE IF NOT EXISTS {pins} (
    key bytea NOT NULL,
    owner bytea NOT NULL,
    expiry bigint NOT NULL,
    PRIMARY KEY (key, owner)
);

CREATE TABLE IF NOT EXISTS {window} (
    key bytea PRIMARY KEY,
    visible_after bigint NOT NULL
);
"
    )
}

#[cfg(test)]
mod tests {
    use crate::backend::{Backend, StagingCursor};
    use crate::Error;

    use crate::pg_common::test_support::SchemaJanitor;
    use crate::pg_common::PgOpts;
    use crate::postgres::testing::{
        fresh_schema, harness_dsn, harness_schemaless, skip_note, stored_value, CONTENT,
    };
    use crate::postgres::PostgresBackend;

    /// Virgin-store no-schema-yet tolerance (POC #1 finding 2): every
    /// read path against a schema that was never bootstrapped is a
    /// no-op — absent/empty, never "table does not exist" — delete and
    /// the batch-delete helper report Missing/zero rows. The tolerance
    /// is read-only: writes surface the engine error and leave zero
    /// residue.
    #[tokio::test]
    async fn virgin_store_no_schema_yet_reads_are_noops() {
        let Some(backend) = harness_schemaless(" virgin").await else {
            return;
        };
        let key = b"\x01opaque-key".to_vec();

        assert!(
            !backend.has(&key).await.unwrap(),
            "has on a schema-less database is a no-op read"
        );
        assert!(
            backend.get(&key).await.unwrap().is_none(),
            "get is a no-op read"
        );
        assert_eq!(
            backend.size(&key).await.unwrap(),
            None,
            "size is a no-op read"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "list is empty on a schema-less database"
        );
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "delete on a schema-less database reports Missing"
        );
        assert_eq!(
            backend
                .delete_batch(std::slice::from_ref(&key))
                .await
                .unwrap(),
            0,
            "the batch-delete helper on a schema-less database removes nothing"
        );

        assert!(
            backend.put(&key, CONTENT).await.is_err(),
            "the tolerance is read-only: writes surface the engine error"
        );
        let mut staging = backend.put_staged().await.unwrap();
        staging.append(CONTENT).await.unwrap();
        let err = staging.commit(&key).await;
        assert!(
            err.is_err(),
            "a staged commit on a schema-less database surfaces the engine error"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the failed writes left zero residue"
        );
    }

    /// The bootstrap creates the B5 row shape — column types
    /// bytea/bytea/bigint and the measured `fillfactor = 90` — and is
    /// idempotent (a second open over the same engine-owned schema is a
    /// no-op; the store then keeps working).
    #[tokio::test]
    async fn bootstrap_creates_the_b5_row_shape_idempotently() {
        let Some(dsn) = harness_dsn() else {
            skip_note();
            return;
        };
        let schema = fresh_schema("bootstrap");
        let opts = PgOpts {
            schema: schema.clone(),
            ..Default::default()
        };
        let first = PostgresBackend::new(&dsn, opts.clone()).await.unwrap();
        let second = PostgresBackend::new(&dsn, opts).await.unwrap();
        let _janitor = SchemaJanitor::new(&dsn, &schema);

        let client = first.pool.get().await.unwrap();
        let columns: Vec<String> = client
            .query(
                "SELECT data_type::text FROM information_schema.columns \
                 WHERE table_schema = $1 AND table_name = 'blobs' \
                 ORDER BY ordinal_position",
                &[&schema],
            )
            .await
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>(0))
            .collect();
        assert_eq!(
            columns,
            ["bytea", "bytea", "bigint"],
            "the B5 row shape: key/value/size column types"
        );

        let reloptions: String = client
            .query_one(
                "SELECT reloptions::text FROM pg_class \
                 WHERE relname = 'blobs' \
                 AND relnamespace = (SELECT oid FROM pg_namespace WHERE nspname = $1)",
                &[&schema],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(
            reloptions, "{fillfactor=90}",
            "the measured fillfactor posture is stamped on the table"
        );

        let key = b"\x01bootstrap-key".to_vec();
        let backend = &first;
        backend.put(&key, CONTENT).await.unwrap();
        assert!(second.has(&key).await.unwrap(), "the second open sees it");
    }

    /// The quoted-identifier boundary: a hostile schema name —
    /// reserved-word-carved, containing an embedded double-quote —
    /// round-trips storage through the schema-qualified statements
    /// without ever entering SQL text unquoted.
    #[tokio::test]
    async fn hostile_schema_name_is_quoted_safely() {
        let Some(dsn) = harness_dsn() else {
            skip_note();
            return;
        };
        let schema = fresh_schema("we\"ird");
        let backend = PostgresBackend::new(
            &dsn,
            PgOpts {
                schema: schema.clone(),
                ..Default::default()
            },
        )
        .await
        .expect("open with a hostile schema name");
        let _janitor = SchemaJanitor::new(&dsn, &schema);

        let key = b"\x01hostile-schema-key".to_vec();
        backend.put(&key, CONTENT).await.unwrap();
        assert!(backend.has(&key).await.unwrap());
        assert_eq!(stored_value(&backend, &key).await.as_deref(), Some(CONTENT));
        assert_eq!(
            backend.size(&key).await.unwrap(),
            Some(CONTENT.len() as u64)
        );
        assert_eq!(backend.list().await.unwrap(), vec![key.clone()]);
        backend.delete(&key).await.unwrap();
        assert!(!backend.has(&key).await.unwrap());
    }
}