#!/usr/bin/env bash
set -euo pipefail

PG_VERSION="${PG_VERSION:-pg17}"
PG_FEATURE="${PG_FEATURE:-pg17}"
PG_CONFIG="${PG_CONFIG:-/opt/homebrew/opt/postgresql@17/bin/pg_config}"
PGHOST="${PGHOST:-localhost}"
PGPORT="${PGPORT:-28817}"
DBNAME="${DBNAME:-pgcontext_hnsw_vacuum_check}"

if [[ ! "${DBNAME}" =~ ^[A-Za-z_][A-Za-z0-9_]*$ ]]; then
    echo "DBNAME must be a simple SQL identifier" >&2
    exit 2
fi

cargo pgrx start "${PG_VERSION}"
cargo pgrx install -p context-pg --features "${PG_FEATURE}" --pg-config "${PG_CONFIG}"

psql -h "${PGHOST}" -p "${PGPORT}" -d postgres -v ON_ERROR_STOP=1 \
    -c "DROP DATABASE IF EXISTS ${DBNAME}" \
    -c "CREATE DATABASE ${DBNAME}"

psql -h "${PGHOST}" -p "${PGPORT}" -d "${DBNAME}" -v ON_ERROR_STOP=1 <<'SQL'
CREATE EXTENSION pgcontext;
SET search_path = pgcontext, public;

CREATE TABLE hnsw_vacuum_items (
    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    embedding vector NOT NULL
);

INSERT INTO hnsw_vacuum_items (embedding)
VALUES ('[1,1,1]'::vector), ('[2,2,2]'::vector), ('[3,3,3]'::vector);

CREATE INDEX hnsw_vacuum_items_embedding_idx
    ON hnsw_vacuum_items USING pgcontext_hnsw (embedding);

DELETE FROM hnsw_vacuum_items WHERE id = 2;
VACUUM hnsw_vacuum_items;

SELECT * FROM compact('hnsw_vacuum_items_embedding_idx'::regclass);

DO $$
BEGIN
    IF (SELECT count(*)::bigint FROM hnsw_vacuum_items) <> 2 THEN
        RAISE EXCEPTION 'unexpected visible row count after HNSW VACUUM';
    END IF;
    IF (SELECT immutable_rows FROM hnsw_segment_stats(
            'hnsw_vacuum_items_embedding_idx'::regclass)) <> 2 THEN
        RAISE EXCEPTION 'full compaction resurrected or retained a VACUUM-dead row';
    END IF;
    IF (SELECT frozen_mutation_records FROM hnsw_segment_stats(
            'hnsw_vacuum_items_embedding_idx'::regclass)) <> 0 THEN
        RAISE EXCEPTION 'full compaction did not drain frozen tombstones';
    END IF;
END
$$;

SELECT true AS vacuum_kept_visible_rows;

-- Delta-region tombstoning (P2-S3). A row inserted after CREATE INDEX is
-- absorbed by the delta region rather than spliced into the graph, so
-- deleting it exercises a different VACUUM path than the base-graph delete
-- above. This lives here rather than in a #[pg_test] because those run
-- inside a transaction and VACUUM cannot.
CREATE TABLE hnsw_vacuum_delta_items (
    id bigint PRIMARY KEY,
    embedding vector(3) NOT NULL
);

INSERT INTO hnsw_vacuum_delta_items (id, embedding)
VALUES (1, '[1,0,0]'::vector), (2, '[0,1,0]'::vector), (3, '[0,0,1]'::vector);

CREATE INDEX hnsw_vacuum_delta_items_embedding_idx
    ON hnsw_vacuum_delta_items USING pgcontext_hnsw (embedding);

SET pgcontext.hnsw_delta_segment_limit = 1;

-- Only ever lives in the delta region: inserted post-build, deleted before
-- any compaction.
INSERT INTO hnsw_vacuum_delta_items (id, embedding) VALUES (998, '[0.9,0.1,0]'::vector);
INSERT INTO hnsw_vacuum_delta_items (id, embedding) VALUES (999, '[1,0,0]'::vector);

SET enable_seqscan = off;

DO $$
DECLARE
    found_ids bigint[];
BEGIN
    SELECT array_agg(id ORDER BY id) INTO found_ids
      FROM (
        SELECT id FROM hnsw_vacuum_delta_items
         ORDER BY embedding OPERATOR(pgcontext.<=>) '[1,0,0]'::vector
         LIMIT 4
      ) ranked;
    IF NOT (999 = ANY(found_ids)) THEN
        RAISE EXCEPTION
            'delta-region row 999 must be served before VACUUM, saw %', found_ids;
    END IF;
END
$$;

DELETE FROM hnsw_vacuum_delta_items WHERE id = 999;
VACUUM hnsw_vacuum_delta_items;

DO $$
DECLARE
    found_ids bigint[];
BEGIN
    SELECT array_agg(id ORDER BY id) INTO found_ids
      FROM (
        SELECT id FROM hnsw_vacuum_delta_items
         ORDER BY embedding OPERATOR(pgcontext.<=>) '[1,0,0]'::vector
         LIMIT 4
      ) ranked;
    IF 999 = ANY(found_ids) THEN
        RAISE EXCEPTION
            'VACUUM must tombstone the delta-only deleted row, saw %', found_ids;
    END IF;
    IF (SELECT count(*)::bigint FROM hnsw_vacuum_delta_items) <> 4 THEN
        RAISE EXCEPTION 'unexpected visible row count after delta-region VACUUM';
    END IF;
END
$$;

-- Repeated bounded pair compaction must incorporate the tombstone exactly
-- once and must not make the deleted delta-only row reappear.
DO $$
DECLARE
    segments integer;
    epoch bigint;
BEGIN
    LOOP
        SELECT segment_count, directory_epoch INTO segments, epoch
          FROM hnsw_segment_stats('hnsw_vacuum_delta_items_embedding_idx'::regclass);
        EXIT WHEN segments < 2;
        PERFORM _compact_hnsw_segment_pair(
            'hnsw_vacuum_delta_items_embedding_idx'::regclass, epoch);
    END LOOP;
    IF (SELECT immutable_rows FROM hnsw_segment_stats(
            'hnsw_vacuum_delta_items_embedding_idx'::regclass)) <> 4 THEN
        RAISE EXCEPTION 'bounded compaction retained a VACUUM-dead delta row';
    END IF;
END
$$;

RESET enable_seqscan;
RESET pgcontext.hnsw_delta_segment_limit;

SELECT true AS vacuum_tombstoned_delta_only_row;
SQL
