alkblobs 0.2.1

//! The window family (the fleet host's both-tables row): the durable
//! staged delete window — `window_stage`'s transactional upsert discipline
//! (forward-only deadlines) and `window_take`'s bounded atomic
//! `DELETE … RETURNING` — plus `ensure_gc_schema` (the GC
//! `gc_pins`/`gc_window` bootstrap the fleet host runs unconditionally at
//! construction).

use std::time::SystemTime;

use super::schema::gc_ddl;
use super::PostgresBackend;
use crate::gc::{EngineWindowState, WindowCandidate, WINDOW_TAKE_BATCH};
use crate::pg_common::{err, is_undefined_table, pg_err, time_to_nanos};
use crate::Error;

/// The window family (ADR-012 §5's postgres row): the fleet host's
/// both-tables schema and the durable staged delete window. Same
/// semantics as the sqlite engine's reference implementation —
/// different dialect.
impl EngineWindowState for PostgresBackend {
    /// Create `gc_pins (key, owner, expiry)` + `gc_window (key,
    /// visible_after)` — unconditionally, both tables (the applicability
    /// row does not condition on the fleet declaration); idempotent, so
    /// the twice-at-open call form is a no-op. The embedder's liveness
    /// table is never created here (the schema-ownership rule).
    async fn ensure_gc_schema(&self) -> Result<(), Error> {
        let client = self.pool.get().await.map_err(err)?;
        client
            .batch_execute(&gc_ddl(&self.schema))
            .await
            .map_err(pg_err)
    }

    /// Stage candidates into the durable window — **one transaction per
    /// call** (review-001 N-2): the loop's upserts commit together, and a
    /// failure anywhere inside the loop rolls the session-level
    /// transaction back, leaving zero staged rows (the failure-path
    /// discipline is `pin_commit`'s: every error arm closes the tx with an
    /// explicit `ROLLBACK` before surfacing). The upsert discipline: a
    /// re-stage of the same key at the same deadline is idempotent (one
    /// row, value unchanged), a re-stage at a new deadline updates the
    /// row's `visible_after` rather than duplicating it — **forward-only**
    /// (review-001 N-4): an earlier re-stage never moves `visible_after`
    /// backwards (`GREATEST` semantics; the target table rides the `w`
    /// alias, the schema-qualified insert's DO-UPDATE reference form).
    async fn window_stage(&self, candidates: &[WindowCandidate]) -> Result<(), Error> {
        if candidates.is_empty() {
            return Ok(());
        }
        #[cfg(test)]
        let fail_window_stage = self.fail_window_stage;
        let client = self.pool.get().await.map_err(err)?;
        client.batch_execute("BEGIN").await.map_err(pg_err)?;
        let sql = format!(
            "INSERT INTO {} AS w (key, visible_after) VALUES ($1, $2) \
             ON CONFLICT (key) DO UPDATE SET \
             visible_after = GREATEST(w.visible_after, excluded.visible_after)",
            self.window_table
        );
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(pg_err(e));
            }
        };
        for candidate in candidates.iter() {
            #[cfg(test)]
            if fail_window_stage
                && candidates
                    .last()
                    .is_some_and(|last| last.key == candidate.key)
            {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(Error::Io(
                    "postgres: window staging failure (mid-loop, injected)".to_owned(),
                ));
            }
            let visible_after = time_to_nanos(candidate.visible_after);
            match client
                .execute(&stmt, &[&candidate.key, &visible_after])
                .await
            {
                Ok(_) => {}
                Err(e) => {
                    let _ = client.batch_execute("ROLLBACK").await;
                    return Err(pg_err(e));
                }
            }
        }
        match client.batch_execute("COMMIT").await {
            Ok(_) => Ok(()),
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                Err(pg_err(e))
            }
        }
    }

    /// Take overdue candidates (`visible_after <= now` — the `<=`
    /// boundary is the seam doc's pin): bounded to [`WINDOW_TAKE_BATCH`]
    /// rows (review-001 N-3 — oldest deadlines first, `ORDER BY
    /// visible_after`; the `LIMIT` rides the inner candidate query), the
    /// calling sweep executor consuming the take in a loop until drained.
    /// One atomic step, unchanged: the single `DELETE … RETURNING`
    /// statement selecting and removing the rows together, so a row is
    /// never handed out twice and an aborted body leaves no partial take.
    /// A missing table maps to empty (the sqlite twin's virgin take; the
    /// read-tolerance split).
    async fn window_take(&self, now: SystemTime) -> Result<Vec<Vec<u8>>, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let horizon = time_to_nanos(now);
        let sql = format!(
            "DELETE FROM {} WHERE key IN ( \
                 SELECT key FROM {} WHERE visible_after <= $1 \
                 ORDER BY visible_after LIMIT $2 \
             ) RETURNING key",
            self.window_table, self.window_table
        );
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) if is_undefined_table(&e) => return Ok(Vec::new()),
            Err(e) => return Err(err(e)),
        };
        let rows = client
            .query(&stmt, &[&horizon, &(WINDOW_TAKE_BATCH as i64)])
            .await
            .map_err(pg_err)?;
        let mut keys = rows
            .iter()
            .map(|row| row.try_get::<_, Vec<u8>>(0).map_err(pg_err))
            .collect::<Result<Vec<Vec<u8>>, _>>()?;
        keys.sort();
        Ok(keys)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::postgres::testing::{
        at, fresh_schema, harness, harness_dsn, harness_gc, schema_tables, skip_note, stages,
        window_rows,
    };

    /// Schema contract (the fleet host's both-tables row): `ensure_gc_schema`
    /// creates `gc_pins` (the `(key, owner)` identity) + `gc_window`, the
    /// call is idempotent — and the embedder's liveness table is never
    /// created by the engine (the schema-ownership rule: exactly the
    /// engine's three tables exist, no `alive_keys`).
    #[tokio::test]
    async fn gc_schema_creates_both_tables_and_never_the_embedders() {
        let Some(backend) = harness("gc-schema").await else {
            return;
        };
        backend.ensure_gc_schema().await.unwrap();
        backend.ensure_gc_schema().await.unwrap();
        assert_eq!(
            schema_tables(&backend).await,
            vec!["blobs", "gc_pins", "gc_window"],
            "the engine's three tables exactly — no embedder liveness table"
        );

        let client = backend.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 = 'gc_pins' \
                 ORDER BY ordinal_position",
                &[&backend.schema],
            )
            .await
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>(0))
            .collect();
        assert_eq!(
            columns,
            ["bytea", "bytea", "bigint"],
            "the gc_pins row shape: key/owner/expiry"
        );
        let columns: Vec<String> = client
            .query(
                "SELECT data_type::text FROM information_schema.columns \
                 WHERE table_schema = $1 AND table_name = 'gc_window' \
                 ORDER BY ordinal_position",
                &[&backend.schema],
            )
            .await
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>(0))
            .collect();
        assert_eq!(
            columns,
            ["bytea", "bigint"],
            "the gc_window row shape: key/visible_after"
        );
    }

    /// Staging contract: idempotent per `(key, deadline)` — the same
    /// candidate staged twice leaves exactly one row — and the
    /// re-stage's deadline conflict update is **forward-only** (review-001
    /// N-4): a re-stage at a later deadline moves the row forward, a
    /// re-stage at an earlier deadline never moves `visible_after`
    /// backwards.
    #[tokio::test]
    async fn window_stage_is_idempotent_and_restage_moves_the_deadline_forward_only() {
        let Some(backend) = harness_gc("window").await else {
            return;
        };
        let key = b"\x01window-key".to_vec();
        let deadline = at(1_000_000_000);

        backend
            .window_stage(&[stages(&key, deadline)])
            .await
            .unwrap();
        backend
            .window_stage(&[stages(&key, deadline)])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await,
            vec![(key.clone(), 1_000_000_000)],
            "re-stage at the same deadline is one idempotent row"
        );

        backend
            .window_stage(&[stages(&key, at(2_000_000_000))])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await,
            vec![(key.clone(), 2_000_000_000)],
            "a re-stage at a later deadline moves the row forward"
        );

        backend
            .window_stage(&[stages(&key, at(1_000_000_000))])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await,
            vec![(key.clone(), 2_000_000_000)],
            "a re-stage at an earlier deadline never moves visible_after backwards"
        );

        let other = b"\x01window-other".to_vec();
        backend
            .window_stage(&[stages(&other, deadline)])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await.len(),
            2,
            "distinct keys are distinct rows"
        );
    }

    /// The staging is transactional (review-001 N-2): a mid-loop failure
    /// (injected after the first candidate's upsert has run inside the
    /// open transaction) rolls the whole call back — zero staged rows,
    /// prior rows untouched — and the backend stages cleanly again once
    /// the failure clears.
    #[tokio::test]
    async fn window_stage_mid_loop_failure_stages_nothing() {
        let Some(mut backend) = harness_gc("window-abort").await else {
            return;
        };
        let prior = b"\x01window-prior".to_vec();
        backend
            .window_stage(&[stages(&prior, at(1_000_000_000))])
            .await
            .unwrap();

        backend.fail_window_stage = true;
        let aborted_a = b"\x01window-aborted-a".to_vec();
        let aborted_b = b"\x01window-aborted-b".to_vec();
        let err = backend
            .window_stage(&[
                stages(&aborted_a, at(2_000_000_000)),
                stages(&aborted_b, at(2_000_000_000)),
            ])
            .await
            .unwrap_err();
        assert!(
            matches!(err, Error::Io(ref m) if m.contains("mid-loop")),
            "the injected stage failure surfaces as Io: {err}"
        );
        assert_eq!(
            window_rows(&backend).await,
            vec![(prior.clone(), 1_000_000_000)],
            "the aborted call staged nothing — the prior row is the only residue"
        );

        backend.fail_window_stage = false;
        backend
            .window_stage(&[
                stages(&aborted_a, at(2_000_000_000)),
                stages(&aborted_b, at(2_000_000_000)),
            ])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await.len(),
            3,
            "the cleared backend stages cleanly again"
        );
    }

    /// The take is bounded (review-001 N-3): a backlog larger than
    /// [`WINDOW_TAKE_BATCH`] hands out at most that many overdue rows per
    /// call — oldest deadlines first — draining across repeated calls,
    /// and the drained window takes empty.
    #[tokio::test]
    async fn window_take_is_bounded_across_the_overdue_backlog() {
        let Some(backend) = harness_gc("take-bounded").await else {
            return;
        };
        let total = WINDOW_TAKE_BATCH + 7;
        let candidates: Vec<WindowCandidate> = (0..total)
            .map(|i| {
                stages(
                    format!("\x01win-backlog-{i:05}").as_bytes(),
                    at(i as i64 * 1_000_000),
                )
            })
            .collect();
        backend.window_stage(&candidates).await.unwrap();

        let first = backend
            .window_take(at(total as i64 * 1_000_000))
            .await
            .unwrap();
        let mut expected_first: Vec<Vec<u8>> = (0..WINDOW_TAKE_BATCH)
            .map(|i| format!("\x01win-backlog-{i:05}").into_bytes())
            .collect();
        expected_first.sort();
        assert_eq!(
            first, expected_first,
            "the first take hands out exactly the batch, oldest deadlines first"
        );

        let second = backend
            .window_take(at(total as i64 * 1_000_000))
            .await
            .unwrap();
        let mut expected_rest: Vec<Vec<u8>> = (WINDOW_TAKE_BATCH..total)
            .map(|i| format!("\x01win-backlog-{i:05}").into_bytes())
            .collect();
        expected_rest.sort();
        assert_eq!(second, expected_rest, "the next take drains the remainder");

        assert_eq!(
            backend
                .window_take(at(total as i64 * 1_000_000))
                .await
                .unwrap(),
            Vec::<Vec<u8>>::new(),
            "the drained window takes empty"
        );
    }

    /// Take contract: exactly the overdue set (`visible_after <= now`),
    /// removed atomically — a second take returns nothing — with future
    /// rows left in place, and the key round-trips through the raw
    /// encoding (signed nanos, so a pre-epoch deadline rides along).
    #[tokio::test]
    async fn window_take_returns_the_overdue_set_and_leaves_future_rows() {
        let Some(backend) = harness_gc("take").await else {
            return;
        };
        let past_pre_epoch = b"\x01win-past-pre-epoch".to_vec();
        let past = b"\x01win-past".to_vec();
        let future = b"\x01win-future".to_vec();
        let far_future = b"\x01win-far-future".to_vec();

        backend
            .window_stage(&[
                stages(&past_pre_epoch, at(-1_000_000_000)),
                stages(&past, at(1_000_000_000)),
                stages(&future, at(2_000_000_000)),
                stages(&far_future, at(3_600_000_000_000)),
            ])
            .await
            .unwrap();

        let taken = backend.window_take(at(1_000_000_001)).await.unwrap();
        let mut expected = vec![past_pre_epoch.clone(), past.clone()];
        expected.sort();
        assert_eq!(taken, expected, "exactly the overdue set");
        assert_eq!(
            backend.window_take(at(1_000_000_001)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "the rows left the table: never handed out twice"
        );
        assert_eq!(
            window_rows(&backend).await,
            vec![
                (far_future.clone(), 3_600_000_000_000),
                (future.clone(), 2_000_000_000)
            ],
            "future rows left in place"
        );
    }

    /// The take boundary is `visible_after == now` (the seam doc's
    /// `<=`): a candidate visible exactly at the horizon is overdue;
    /// one nanosecond later it is not.
    #[tokio::test]
    async fn window_take_boundary_visible_after_equal_now_is_overdue() {
        let Some(backend) = harness_gc("boundary").await else {
            return;
        };
        let equal = b"\x01win-equal-boundary".to_vec();
        let just_later = b"\x01win-nanosecond-later".to_vec();

        backend
            .window_stage(&[
                stages(&equal, at(1_500_000_000)),
                stages(&just_later, at(1_500_000_001)),
            ])
            .await
            .unwrap();

        assert_eq!(
            backend.window_take(at(1_500_000_000)).await.unwrap(),
            vec![equal.clone()],
            "visible_after == now is overdue (the `<=` boundary)"
        );
        assert_eq!(
            backend.window_take(at(1_500_000_000)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "the not-yet-overdue row stays (one nanosecond of slack)"
        );
        assert_eq!(
            backend.window_take(at(1_500_000_001)).await.unwrap(),
            vec![just_later.clone()],
            "the horizon itself moved, so the row is now overdue"
        );
    }

    /// Virgin-store shapes: on a schema-less database `window_take` maps
    /// a missing table to empty (the read-tolerance parity with the
    /// sqlite twin), while `window_stage` surfaces the engine error
    /// without silently creating schema. A freshly constructed backend
    /// with an empty window table also takes empty.
    #[tokio::test]
    async fn window_virgin_store_shapes() {
        let Some(dsn) = harness_dsn() else {
            skip_note();
            return;
        };
        let tableless = PostgresBackend::new_schemaless(&dsn, &fresh_schema("window-virgin"))
            .await
            .unwrap();
        let key = b"\x01virgin-window-key".to_vec();

        assert_eq!(
            tableless.window_take(at(1_000_000_000)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "schema-less take is a no-op read"
        );
        assert_eq!(
            tableless.window_take(at(1_000_000_000)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "still empty on a repeat"
        );
        match tableless.window_stage(&[stages(&key, at(1))]).await {
            Err(Error::Io(message)) => {
                assert!(
                    message.contains("does not exist"),
                    "the write surfaces the engine error, not a silent schema create: {message}"
                )
            }
            other => panic!("schema-less stage is the engine error, got {other:?}"),
        }

        let Some(backend) = harness_gc("window-empty").await else {
            return;
        };
        assert_eq!(
            backend.window_take(SystemTime::now()).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "empty window table takes empty"
        );
    }
}