ppoppo_schema_constrained/

lib.rs

//! **NOT a stable public API.** Engine-tier binding primitive — published to
//! crates.io only because the SDK closure requires it on the registry; 3rd
//! parties never name this crate. They meet the value-sets it binds through an
//! SDK product facade or a wire contract, never here.
//!
//! # Schema-Constrained Value-Sets (the enum ↔ `CHECK` anchor)
//!
//! A *schema-constrained* enum is a domain value-set whose members are also
//! enumerated by a PostgreSQL `CHECK (col IN (…))` constraint. The enum and the
//! constraint are two reifications of one fact ("the legal values of this
//! column"); left unbound they drift independently — a migration widens the
//! `CHECK`, or a variant is added, and the other side silently goes stale.
//!
//! This crate is the single seam that binds them. The same drift class exists
//! on both sides of the monorepo (PAS `scaccounts`, PCS `scchat`) and neither
//! core may depend on the other, so the binding primitive is hoisted out of
//! both — a pure, dependency-free trait + macro (`std::BTreeSet` only), which
//! both cores (that ban IO/transport crates) can depend on.
//!
//! ## Why engine tier and not `crates/shared/`
//!
//! It sat in `crates/shared/` (`publish = false`) until `RFC_202607252223`
//! T-03, which is when the placement was first *tested* rather than assumed:
//! `ppoppo-identity` needs to enroll its own `EntityType`, and `engine →
//! shared` is forbidden by the crate lattice (`xtask::policy::rules::taxonomy`)
//! — so the enrollment was unreachable, and the vocabulary had to keep a
//! second PAS-local enum alive just to carry it.
//!
//! The fix was to notice that the folder was wrong, not the lattice. Engine
//! tier means *published substrate that no 3rd party names* — a dependency-free
//! trait + macro consumed by two service cores and one vocabulary crate is
//! exactly that. The move corrected a misfile that predates the tier; it did
//! not trade a principle for convenience.
//!
//! ## The anchor triple (per `STS_SSOT_GOVERNANCE`)
//!
//! - **Owner**: the domain enum in `accounts-core` / `accounts-api` /
//!   `chat-core` (the closest reified form of the value-set decision).
//! - **Anchor**: domain-specific — these are tuned domain vocabularies with no
//!   external standard, so *this crate doc-comment is the anchor of record*
//!   (governance §4). (Formerly PAS
//!   `ADR_202605242324_schema-constrained-value-sets.md`, folded into
//!   `accounts-core` on its retirement, then hoisted here when PCS adopted the
//!   same gate.)
//! - **Verification**: [`bindings`](SchemaConstrained::bindings) feeds each
//!   service's `schema_check_drift.rs` DB test
//!   (`accounts-api/tests/` for `scaccounts`, `chat-api/tests/` for `scchat`),
//!   which reads each `CHECK` from the *materialized* schema
//!   (`pg_get_constraintdef`) and asserts set-equality with `ALL`. The
//!   compile-time half lives in the [`impl_schema_constrained!`] macro: it
//!   emits an exhaustive `match`, so adding a variant without listing it fails
//!   to build.
//!
//! ## Why a DB test and not a file parse
//!
//! The value-sets evolve through `ALTER … DROP/ADD CONSTRAINT` migrations (e.g.
//! `lifecycle_state` gained `tombstoned`; `oauth_audit_events.event_type`
//! gained `otp_issue`/`otp_verify`). The authoritative set is therefore the
//! *result of applying every migration*, which only the database knows —
//! parsing the baseline `.sql` would report phantom drift. Asking Postgres via
//! `pg_get_constraintdef` is the only correct anchor (and avoids the brittle
//! bespoke-parser ops-tax rejected in `STS_RATE_LIMITS_PPOPPO` §Anchor
//! "Rationale" option A).
//!
//! ## Two more rejected alternatives
//!
//! - **`#[sqlx::Type]` alone.** Binds the column *type* (text), not the
//!   `CHECK`'s value-set — a typo in the enum still compiles and the set still
//!   drifts. Complementary at the query boundary, not a substitute for the
//!   verification.
//! - **Accept drift under human review.** Leaves security-adjacent value-sets
//!   (audit taxonomy, lifecycle, step-up purpose) under governance §2's
//!   "aspiration, not enforcement" gate. Rejected.
//!
//! ## Caveat — the value-equality half is integration-tier
//!
//! The compile-time exhaustiveness guard covers the Rust side on every build.
//! The `ALL`-vs-`CHECK` set-equality half needs a live database, and there is
//! no CI job running DB-backed tests — so it bites via each service's
//! `just test-integration` and the `/deploy-ppoppo` pre-flight, not on a plain
//! `cargo test`.

#![deny(rust_2018_idioms)]
#![warn(missing_debug_implementations)]

use std::collections::BTreeSet;

/// One `(constraint, allowed-value-set)` pair, erased of the originating enum
/// type so the drift test can iterate heterogeneous value-sets.
#[derive(Debug, Clone)]
pub struct SchemaBinding {
    /// The `pg_constraint.conname` (e.g. `"ck_ppnums_entity_type_enum"`).
    pub constraint: &'static str,
    /// The DB-text values the owning enum permits — must equal the constraint's
    /// `IN (…)` set in the materialized schema.
    pub allowed: BTreeSet<&'static str>,
}

/// A domain value-set whose members are mirrored by one or more SQL `CHECK`
/// constraints.
///
/// Implemented via [`impl_schema_constrained!`]; never hand-written, so the
/// compile-time exhaustiveness guard is always emitted alongside.
pub trait SchemaConstrained: Sized + 'static {
    /// Every variant, in any order. The macro hand-lists these (no `strum`);
    /// the exhaustiveness guard makes an omission a build error and the DB test
    /// makes a stale list a pre-flight failure.
    const ALL: &'static [Self];

    /// The `CHECK` constraint(s) whose `IN (…)` set must equal
    /// `{ ALL.map(db_value) }`. More than one when several columns share the
    /// value-set (e.g. `lifecycle_state` on three columns).
    const CHECK_CONSTRAINTS: &'static [&'static str];

    /// The DB-text form of a variant — the literal stored in the column and
    /// named in the `CHECK`. Delegates to the enum's inherent `as_str` /
    /// `as_wire`.
    fn db_value(&self) -> &'static str;

    /// Erased `(constraint, allowed)` pairs for the drift test — one per entry
    /// in [`CHECK_CONSTRAINTS`](Self::CHECK_CONSTRAINTS).
    fn bindings() -> Vec<SchemaBinding> {
        let allowed: BTreeSet<&'static str> = Self::ALL.iter().map(Self::db_value).collect();
        Self::CHECK_CONSTRAINTS
            .iter()
            .map(|&constraint| SchemaBinding {
                constraint,
                allowed: allowed.clone(),
            })
            .collect()
    }
}

/// Implement [`SchemaConstrained`] for a unit-variant enum and emit a
/// compile-time exhaustiveness guard from the same variant list.
///
/// ```ignore
/// ppoppo_schema_constrained::impl_schema_constrained!(EntityType via as_str {
///     all: [Human, AiAgent, Enterprise, Programmable, Mask],
///     constraints: ["ck_ppnums_entity_type_enum"],
/// });
/// ```
///
/// `via $method` is the enum's inherent value accessor (`as_str` for most,
/// `as_db_str` / `as_wire` for others). Place the invocation next to the enum
/// so the constraint name lives *on the fact* (governance §6 carrier).
///
/// `non_stored` lists render-only variants that exist in the enum but are never
/// persisted (and so never appear in the `CHECK`) — e.g. `EntityType::Delegated`.
/// They are excluded from `ALL` yet still covered by the exhaustiveness guard,
/// so the asymmetry is declared, not hidden.
#[macro_export]
macro_rules! impl_schema_constrained {
    (
        $ty:ident via $method:ident {
            all: [ $( $variant:ident ),+ $(,)? ],
            $( non_stored: [ $( $ns:ident ),+ $(,)? ], )?
            constraints: [ $( $constraint:literal ),+ $(,)? ] $(,)?
        }
    ) => {
        impl $crate::SchemaConstrained for $ty {
            const ALL: &'static [Self] = &[ $( $ty::$variant ),+ ];
            const CHECK_CONSTRAINTS: &'static [&'static str] = &[ $( $constraint ),+ ];
            fn db_value(&self) -> &'static str {
                self.$method()
            }
        }

        // Compile-time half: if a variant is added to the enum but listed in
        // neither `all` nor `non_stored`, this match is non-exhaustive and the
        // build fails — pointing the author at the enrollment.
        const _: fn($ty) = |x| match x {
            $( $ty::$variant => () ),+
            $( , $( $ty::$ns => () ),+ )?
        };
    };
}

/// Finding value-set `CHECK` constraints in migration SQL.
///
/// The enrollment registries answer "which enums claim a constraint". This
/// answers the opposite question — "which constraints exist" — so a service
/// can assert that every value-set in its schema is either enrolled or
/// explicitly declared unbound. Both organs need it, and a parser duplicated
/// per organ is a parser that gets fixed in one of them.
pub mod migration_scan {
    use std::collections::BTreeSet;

    /// Names every **single-column value-set** CHECK in one SQL text.
    ///
    /// Both spellings count, because both occur: `pg_dump` writes
    /// `CHECK ((col = ANY (ARRAY['a'::text])))` and a hand-written migration
    /// writes `CHECK (col IN ('a', 'b'))`. Postgres treats them as the same
    /// constraint; so does this.
    ///
    /// Out of scope, and deliberately so — these are not value-sets and have no
    /// enum to bind to: range checks, regex checks, `IS NULL` checks, and any
    /// multi-clause CHECK such as `CHECK ((plan = 'enterprise') = (limit IS
    /// NULL))`. They are excluded by *structure* (the expression must open on a
    /// bare column followed immediately by the membership operator), not by a
    /// blocklist that would need maintaining.
    pub fn value_set_constraints_in(sql: &str) -> BTreeSet<String> {
        let flat = flatten(sql);
        value_set_events(&flat)
            .into_iter()
            .map(|(_, name)| name)
            .collect()
    }

    /// Every `RENAME CONSTRAINT <old> TO <new>` in one SQL text, in text order.
    ///
    /// # Why a text scan of migrations needs this at all
    ///
    /// [`value_set_constraints_in`] sees a constraint at the name the migration
    /// that *created* it used. A later migration is free to rename it, and the
    /// text of the first file never changes — so a corpus scan without this
    /// reports a constraint that no longer exists, and the guard demands an
    /// enum bound to a name nothing will ever match while the enum correctly
    /// names the new one. That is a false positive on every rename, and it
    /// reads as a missing enrolment, which sends the reader to the wrong file.
    ///
    /// Lives here for the same reason the parser does: both organs need it, and
    /// a resolver duplicated per organ is a resolver that gets fixed in one.
    pub fn constraint_renames_in(sql: &str) -> Vec<(String, String)> {
        let flat = flatten(sql);
        rename_events(&flat)
            .into_iter()
            .map(|(_, rename)| rename)
            .collect()
    }

    /// Every value-set CHECK across an ordered migration corpus, **at its final
    /// name** — the set the materialized schema actually holds.
    ///
    /// `sources` must be in chronological order — for this repo's stamped
    /// filenames that is lexicographic order — and the order is load-bearing:
    /// this is a **fold over statements**, applied exactly as Postgres applies
    /// them. An `ADD CONSTRAINT` inserts a name; a `RENAME CONSTRAINT` moves a
    /// name that exists *at that moment* and is a no-op on one that does not;
    /// a `DROP CONSTRAINT` removes one.
    /// Nothing is resolved after the fact, so a rename never reaches back to a
    /// constraint that had not been created yet, and never reaches forward to
    /// one created later under the retired name.
    ///
    /// The earlier shape — collect every rename from the whole corpus, then
    /// resolve every constraint name through all of them — violated exactly
    /// that (`RFC_202609080357` T-19): a CHECK re-created under a name a
    /// previous migration had renamed away was resolved through that old
    /// rename and dropped out of the demanded set, so an unenrolled enum
    /// escaped `schema_check_coverage`. The fold also makes a rename cycle
    /// harmless by construction — each statement applies once, and the result
    /// is whatever name the cycle leaves the constraint on, which is what the
    /// database ends with too.
    ///
    /// This composition lives here rather than in each service's guard on
    /// purpose: two organs that each assemble "scan, then resolve" for
    /// themselves are two places for the second half to be forgotten, which is
    /// exactly how PCS's guard shipped without it.
    pub fn value_set_constraints_across<'a>(
        sources: impl IntoIterator<Item = &'a str>,
    ) -> BTreeSet<String> {
        let mut live = BTreeSet::new();
        for sql in sources {
            let flat = flatten(sql);
            for event in events_in(&flat) {
                match event {
                    Event::Add(name) => {
                        live.insert(name);
                    }
                    Event::Rename(from, to) => {
                        if live.remove(&from) {
                            live.insert(to);
                        }
                    }
                    Event::Drop(name) => {
                        live.remove(&name);
                    }
                }
            }
        }
        live
    }

    /// One statement the fold cares about, in the order it appears.
    enum Event {
        Add(String),
        Rename(String, String),
        Drop(String),
    }

    /// The value-set adds and the constraint renames and drops of one
    /// flattened text, merged by text position — the only order in which they
    /// mean anything.
    fn events_in(flat: &str) -> Vec<Event> {
        let mut events: Vec<(usize, Event)> = value_set_events(flat)
            .into_iter()
            .map(|(at, name)| (at, Event::Add(name)))
            .chain(
                rename_events(flat)
                    .into_iter()
                    .map(|(at, (from, to))| (at, Event::Rename(from, to))),
            )
            .chain(
                drop_events(flat)
                    .into_iter()
                    .map(|(at, name)| (at, Event::Drop(name))),
            )
            .collect();
        events.sort_by_key(|(at, _)| *at);
        events.into_iter().map(|(_, event)| event).collect()
    }

    /// `(position, constraint name)` for every single-column value-set CHECK
    /// in a flattened text.
    fn value_set_events(flat: &str) -> Vec<(usize, String)> {
        let mut found = Vec::new();
        for (at, _) in flat.match_indices("CONSTRAINT ") {
            let rest = &flat[at + "CONSTRAINT ".len()..];
            let Some((name, tail)) = rest.split_once(' ') else {
                continue;
            };
            let Some(body) = tail.strip_prefix("CHECK ") else {
                continue;
            };
            let Some(body) = balanced(body) else {
                continue;
            };
            if is_pure_value_set(body) {
                found.push((at, name.to_string()));
            }
        }
        found
    }

    /// `(position, (old, new))` for every `RENAME CONSTRAINT` in a flattened
    /// text.
    fn rename_events(flat: &str) -> Vec<(usize, (String, String))> {
        let mut renames = Vec::new();
        for (at, _) in flat.match_indices("RENAME CONSTRAINT ") {
            let rest = &flat[at + "RENAME CONSTRAINT ".len()..];
            let mut parts = rest.split_whitespace();
            let (Some(from), Some(to_kw), Some(to)) = (parts.next(), parts.next(), parts.next())
            else {
                continue;
            };
            if !to_kw.eq_ignore_ascii_case("TO") {
                continue;
            }
            renames.push((
                at,
                (
                    from.trim_end_matches(';').to_string(),
                    to.trim_end_matches(';').to_string(),
                ),
            ));
        }
        renames
    }

    /// `(position, name)` for every `DROP CONSTRAINT [IF EXISTS] <name>` in a
    /// flattened text.
    ///
    /// A value-set CHECK that leaves only because its column is dropped is not
    /// seen here — the fold does not track columns. The migration that drops
    /// such a column names the CHECK in a `DROP CONSTRAINT` first, which also
    /// states in the file what the drop removes.
    fn drop_events(flat: &str) -> Vec<(usize, String)> {
        let mut drops = Vec::new();
        for (at, _) in flat.match_indices("DROP CONSTRAINT ") {
            let rest = &flat[at + "DROP CONSTRAINT ".len()..];
            let rest = rest.strip_prefix("IF EXISTS ").unwrap_or(rest);
            let Some(name) = rest.split_whitespace().next() else {
                continue;
            };
            let name = name.trim_end_matches([';', ',']);
            if !name.is_empty() {
                drops.push((at, name.to_string()));
            }
        }
        drops
    }

    /// Drops `--` comments and collapses runs of whitespace, so `find` can
    /// match across the line breaks migrations put between a constraint name
    /// and its clause — and cannot match a statement the author commented out.
    ///
    /// Comments are stripped **per line, before the lines are joined**: a
    /// `--` runs to the end of its own line and no further, and stripping
    /// after the join would delete the whole remainder of the file past the
    /// first comment. A `--` inside a single-quoted literal is text, not a
    /// comment, and is kept.
    fn flatten(sql: &str) -> String {
        let mut out = String::with_capacity(sql.len());
        let mut in_space = false;
        for line in sql.lines() {
            for ch in strip_line_comment(line)
                .chars()
                .chain(std::iter::once('\n'))
            {
                if ch.is_whitespace() {
                    if !in_space {
                        out.push(' ');
                    }
                    in_space = true;
                } else {
                    out.push(ch);
                    in_space = false;
                }
            }
        }
        out
    }

    /// The part of one line before its `--` comment, if any, ignoring a `--`
    /// that sits inside a single-quoted literal.
    fn strip_line_comment(line: &str) -> &str {
        let mut in_quote = false;
        let mut prev_dash = false;
        for (i, ch) in line.char_indices() {
            match ch {
                '\'' => {
                    in_quote = !in_quote;
                    prev_dash = false;
                }
                '-' if !in_quote => {
                    if prev_dash {
                        return &line[..i - 1];
                    }
                    prev_dash = true;
                }
                _ => prev_dash = false,
            }
        }
        line
    }

    /// The contents of the leading `(...)` group, without its outer parens.
    fn balanced(s: &str) -> Option<&str> {
        let s = s.strip_prefix('(')?;
        let mut depth = 1usize;
        for (i, ch) in s.char_indices() {
            match ch {
                '(' => depth += 1,
                ')' => {
                    depth -= 1;
                    if depth == 0 {
                        return Some(&s[..i]);
                    }
                }
                _ => {}
            }
        }
        None
    }

    /// Is this expression *nothing but* a membership test on one column?
    ///
    /// The "nothing but" is the whole difficulty. An earlier version stripped
    /// leading parens and checked only that the expression *began* with
    /// `col IN (` — which accepts
    /// `((state = ANY (…)) AND (x IS NOT NULL)) OR (…)`, a compound predicate
    /// whose first clause happens to be a membership test. PAS has exactly one
    /// of those (`ck_signup_sessions_reservation_pair`) and it was misreported
    /// as an unbound value-set, i.e. the guard demanded an enum for a
    /// constraint that can never have one.
    ///
    /// So the test is structural on both ends: unwrap redundant parens, then
    /// require the membership group to consume the entire remainder.
    fn is_pure_value_set(expr: &str) -> bool {
        let mut expr = expr.trim();
        // Unwrap only parens that wrap the *whole* expression.
        while let Some(inner) = balanced(expr) {
            if inner.len() + 2 == expr.len() {
                expr = inner.trim();
            } else {
                break;
            }
        }

        // The column is either bare (`kind`) or parenthesised, which is how
        // `pg_dump` writes a cast of a varchar column: `(kind)::text`.
        let mut after = if expr.starts_with('(') {
            match balanced(expr) {
                Some(inner)
                    if !inner.is_empty()
                        && inner.chars().all(|c| c.is_alphanumeric() || c == '_') =>
                {
                    expr[inner.len() + 2..].trim_start()
                }
                _ => return false,
            }
        } else {
            let column_len = expr
                .find(|c: char| !(c.is_alphanumeric() || c == '_'))
                .unwrap_or(expr.len());
            if column_len == 0 {
                return false;
            }
            expr[column_len..].trim_start()
        };
        loop {
            if let Some(r) = after.strip_prefix(')') {
                after = r.trim_start();
                continue;
            }
            if let Some(r) = after.strip_prefix("::") {
                after = r
                    .trim_start_matches(|ch: char| ch.is_alphanumeric() || ch == '_')
                    .trim_start();
                continue;
            }
            break;
        }

        let list = if let Some(r) = after.strip_prefix("IN ") {
            r.trim_start()
        } else if let Some(r) = after.strip_prefix("= ANY ") {
            r.trim_start()
        } else {
            return false;
        };
        // The membership group must be the last thing in the expression — no
        // trailing ` AND …`, no ` OR …`.
        match balanced(list) {
            Some(inner) => list.len() == inner.len() + 2,
            None => false,
        }
    }

    #[cfg(test)]
    #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
    mod tests {
        use super::{constraint_renames_in, value_set_constraints_across};

        // ─── rename resolution ────────────────────────────────────────────
        //
        // The rename half is half the guard, so it is tested like one: a fold
        // that silently saw no renames would restore exactly the false
        // positive it exists to remove, and would do it quietly.

        #[test]
        fn renames_parse_across_the_line_breaks_migrations_use() {
            let sql = "ALTER TABLE scchat.t\n    RENAME CONSTRAINT one_check\n                  TO two_check;";
            assert_eq!(
                constraint_renames_in(sql),
                vec![("one_check".to_string(), "two_check".to_string())],
            );
        }

        #[test]
        fn a_rename_chain_lands_on_its_final_name() {
            let corpus = value_set_constraints_across([
                "ALTER TABLE t ADD CONSTRAINT a_check CHECK (kind IN ('a')); \
                 ALTER TABLE t ADD CONSTRAINT untouched_check CHECK (mode IN ('m'));",
                "ALTER TABLE t RENAME CONSTRAINT a_check TO b_check;",
                "ALTER TABLE t RENAME CONSTRAINT b_check TO c_check;",
            ]);
            assert!(corpus.contains("c_check"), "{corpus:?}");
            assert!(
                !corpus.contains("a_check") && !corpus.contains("b_check"),
                "neither retired name is demanded: {corpus:?}"
            );
            assert!(
                corpus.contains("untouched_check"),
                "a constraint nobody renamed keeps its name: {corpus:?}"
            );
        }

        #[test]
        fn a_cyclic_rename_ends_where_the_database_ends() {
            // A cycle is a migration bug, but it is one Postgres executes
            // happily — each statement applies once — so the guard must
            // terminate AND report the name the schema is actually left with.
            let corpus = value_set_constraints_across([
                "ALTER TABLE t ADD CONSTRAINT a_check CHECK (kind IN ('a')); \
                 ALTER TABLE t RENAME CONSTRAINT a_check TO b_check; \
                 ALTER TABLE t RENAME CONSTRAINT b_check TO a_check;",
            ]);
            assert_eq!(
                corpus.into_iter().collect::<Vec<_>>(),
                vec!["a_check".to_string()]
            );
        }

        #[test]
        fn a_corpus_reports_the_check_at_its_final_name_only() {
            // The whole point: the creating migration's text never changes, so
            // without the second pass the guard demands an enum bound to a name
            // that no longer exists — and that reads as a missing enrolment,
            // which sends the reader to the wrong file entirely.
            let created =
                "ALTER TABLE scchat.t ADD CONSTRAINT old_check CHECK (kind IN ('a', 'b'));";
            let renamed = "ALTER TABLE scchat.t RENAME CONSTRAINT old_check TO new_check;";

            let one_file = value_set_constraints_across([created]);
            assert!(
                one_file.contains("old_check"),
                "control: the scan finds it before any rename"
            );

            let corpus = value_set_constraints_across([created, renamed]);
            assert!(
                corpus.contains("new_check"),
                "the final name is what the schema has"
            );
            assert!(
                !corpus.contains("old_check"),
                "the retired name must not be demanded"
            );
        }

        // ─── drops ────────────────────────────────────────────────────────

        #[test]
        fn a_dropped_check_is_no_longer_demanded() {
            let corpus = value_set_constraints_across([
                "ALTER TABLE t ADD CONSTRAINT a_check CHECK (kind IN ('a')); \
                 ALTER TABLE t ADD CONSTRAINT kept_check CHECK (mode IN ('m'));",
                "ALTER TABLE t\n    DROP CONSTRAINT a_check;\nALTER TABLE t DROP COLUMN kind;",
            ]);
            assert_eq!(
                corpus.into_iter().collect::<Vec<_>>(),
                vec!["kept_check".to_string()]
            );
        }

        #[test]
        fn a_drop_then_readd_in_one_statement_keeps_the_check() {
            // The value-set widening shape: `DROP CONSTRAINT x, ADD CONSTRAINT
            // x CHECK (...)`. Text order decides, so the name survives.
            let corpus = value_set_constraints_across([
                "ALTER TABLE t ADD CONSTRAINT a_check CHECK (kind IN ('a'));",
                "ALTER TABLE t DROP CONSTRAINT IF EXISTS a_check, \
                 ADD CONSTRAINT a_check CHECK (kind IN ('a', 'b'));",
            ]);
            assert!(corpus.contains("a_check"), "{corpus:?}");
        }

        #[test]
        fn a_commented_out_drop_is_not_applied() {
            let corpus = value_set_constraints_across([
                "ALTER TABLE t ADD CONSTRAINT a_check CHECK (kind IN ('a'));",
                "-- ALTER TABLE t DROP CONSTRAINT a_check;",
            ]);
            assert!(corpus.contains("a_check"), "{corpus:?}");
        }

        #[test]
        fn a_check_recreated_under_a_renamed_away_name_is_demanded_again() {
            // RFC_202609080357 T-19. Three migrations: the first creates
            // `a_check`, the second renames it to `b_check`, the third creates
            // a NEW `a_check`. The schema ends with both. A scan that collects
            // every rename first and only then resolves every name sends the
            // third file's `a_check` through the second file's rename — so the
            // new constraint vanishes from the demanded set, and an enum that
            // forgot to enrol it is never asked to.
            let first = "ALTER TABLE t ADD CONSTRAINT a_check CHECK (kind IN ('a'));";
            let second = "ALTER TABLE t RENAME CONSTRAINT a_check TO b_check;";
            let third = "ALTER TABLE t ADD CONSTRAINT a_check CHECK (mode IN ('x'));";

            let corpus = value_set_constraints_across([first, second, third]);
            assert!(
                corpus.contains("b_check"),
                "the renamed original: {corpus:?}"
            );
            assert!(
                corpus.contains("a_check"),
                "the re-created constraint must be demanded — a rename that \
                 happened before it existed does not apply to it: {corpus:?}"
            );
        }

        #[test]
        fn a_rename_applies_only_to_what_exists_when_it_runs_within_one_file() {
            // Same rule inside a single file: statements apply in text order.
            let sql = "ALTER TABLE t RENAME CONSTRAINT a_check TO b_check; \
                       ALTER TABLE t ADD CONSTRAINT a_check CHECK (kind IN ('a'));";
            let corpus = value_set_constraints_across([sql]);
            assert!(corpus.contains("a_check"), "{corpus:?}");
            assert!(
                !corpus.contains("b_check"),
                "nothing named b_check was ever a value-set CHECK: {corpus:?}"
            );
        }

        #[test]
        fn a_rename_inside_a_comment_is_not_a_rename() {
            // A well-formed statement that happens to be commented out — the
            // shape a migration author leaves behind when they reconsider —
            // must mint nothing. The comment-stripping happens before parsing,
            // per line, so the statement after the comment is still seen.
            let sql = "-- ALTER TABLE t RENAME CONSTRAINT real_check TO phantom_check;\n\
                       ALTER TABLE t ADD CONSTRAINT real_check CHECK (kind IN ('a'));";
            assert!(
                constraint_renames_in(sql).is_empty(),
                "a commented-out rename was parsed: {:?}",
                constraint_renames_in(sql)
            );
            let corpus = value_set_constraints_across([sql]);
            assert_eq!(
                corpus.into_iter().collect::<Vec<_>>(),
                vec!["real_check".to_string()],
                "the live statement after the comment is still parsed"
            );
        }

        #[test]
        fn a_commented_out_check_is_not_demanded() {
            let sql = "-- ADD CONSTRAINT ck_dead CHECK (kind IN ('a'));\n\
                       ADD CONSTRAINT ck_live CHECK (kind IN ('a'));";
            let found = super::value_set_constraints_in(sql);
            assert!(!found.contains("ck_dead"), "{found:?}");
            assert!(found.contains("ck_live"), "{found:?}");
        }

        #[test]
        fn a_non_rename_use_of_the_word_is_not_read_as_one() {
            // `RENAME TO` on a table is not `RENAME CONSTRAINT`, and prose that
            // merely says the word is not a statement.
            let renames = constraint_renames_in(
                "-- we RENAME CONSTRAINT names when a table moves\n\
                 ALTER TABLE scchat.old_t RENAME TO new_t;",
            );
            assert!(
                renames.iter().all(|(from, _)| from != "names"),
                "a comment must not be parsed as a rename: {renames:?}"
            );
        }

        /// The control for every guard built on this parser.
        ///
        /// Each input below is a shape a migration in this repo is or could be written
        /// in. The old matcher found only the last one; the four before it are the
        /// ones a PR would actually add, and every one of them was invisible.
        #[test]
        fn the_parser_sees_the_shapes_a_migration_is_actually_written_in() {
            let must_find = [
                (
                    "hand-written ALTER, name and CHECK on separate lines, IN spelling",
                    "ALTER TABLE scchat.t\n    ADD CONSTRAINT ck_target\n    CHECK (kind IN ('a', 'b'));",
                ),
                (
                    "hand-written ALTER, separate lines, pg_dump spelling",
                    "ALTER TABLE scchat.t\n    ADD CONSTRAINT ck_target\n    CHECK ((kind = ANY (ARRAY['a'::text])));",
                ),
                (
                    "inline in CREATE TABLE, IN spelling",
                    "CREATE TABLE scchat.t (\n  kind text,\n  CONSTRAINT ck_target CHECK (kind IN ('a', 'b'))\n);",
                ),
                (
                    "cast between column and operator",
                    "ADD CONSTRAINT ck_target CHECK (((kind)::text = ANY (ARRAY['a'::text])));",
                ),
                (
                    "single-line pg_dump form (the only one the first matcher caught)",
                    "ADD CONSTRAINT ck_target CHECK ((kind = ANY (ARRAY['a'::text, 'b'::text])));",
                ),
            ];
            for (label, sql) in must_find {
                assert!(
                    super::value_set_constraints_in(sql).contains("ck_target"),
                    "parser missed a real value-set CHECK — {label}\n  input: {sql}"
                );
            }

            // The exclusions must stay exclusions, or the guard starts demanding an
            // enum for constraints that can never have one.
            let must_ignore = [
                (
                    "multi-clause pairing check (this repo has one)",
                    "ADD CONSTRAINT ck_pairing CHECK ((plan = 'enterprise') = (monthly_message_limit IS NULL));",
                ),
                (
                    "range check",
                    "ADD CONSTRAINT ck_range CHECK ((retention_days > 0));",
                ),
                (
                    "regex check",
                    "ADD CONSTRAINT ck_regex CHECK ((ppnum ~ '^[0-9]{4}$'));",
                ),
                (
                    "membership nested inside a compound predicate is not a value-set",
                    "ADD CONSTRAINT ck_compound CHECK ((a IS NULL) OR (kind IN ('a', 'b')));",
                ),
                (
                    "membership as the FIRST clause of a compound predicate — PAS has\
                     exactly this shape in ck_signup_sessions_reservation_pair, and a\
                     parser that only checks the start of the expression accepts it",
                    "ADD CONSTRAINT ck_pair CHECK ((((state = ANY (ARRAY['a'::text])) AND\
                     (r IS NOT NULL)) OR ((state = 'b'::text) AND (r IS NULL))));",
                ),
            ];
            for (label, sql) in must_ignore {
                assert!(
                    super::value_set_constraints_in(sql).is_empty(),
                    "parser claimed a non-value-set CHECK — {label}\n  input: {sql}"
                );
            }
        }
    }
}