feat: add unused_replication_slot lint for stale replication slots

.claude/skills/new-lint/SKILL.md

@@ -23,15 +23,17 @@ Create a SQL view in the `lint` schema. The view **must** return exactly these 1
 |--------|------|-------|
 | `name` | text | snake_case identifier, e.g. `'my_lint_name'` |
 | `title` | text | Human-readable title |
-| `level` | text | `'ERROR'`, `'WARN'`, or `'INFO'` |
+| `level` | text | `'ERROR'`, `'WARN'`, or `'INFO'` — a `case` expression is fine if severity varies per row (see `lints/0031_unused_replication_slot.sql`) |
 | `facing` | text | `'EXTERNAL'` or `'INTERNAL'` |
 | `categories` | text[] | e.g. `array['SECURITY']` or `array['PERFORMANCE']` |
 | `description` | text | What the lint checks and why it matters |
 | `detail` | text | `format()`-interpolated message naming the specific object |
 | `remediation` | text | `'https://supabase.com/docs/guides/database/database-linter?lint=XXXX_<name>'` |
-| `metadata` | jsonb | `jsonb_build_object('schema', ..., 'name', ..., 'type', ...)` |
+| `metadata` | jsonb | `jsonb_build_object('schema', ..., 'name', ..., 'type', ...)`. If the flagged object has no schema (e.g. a replication slot, a role), set `entity` instead of `name` — Studio's `getLintEntityString` needs either `schema`+`name` or `entity` to render anything. |
 | `cache_key` | text | `format('<name>_%s_%s', schema, object)` — unique per violation |
 
+Whether `level` is a literal or a `case` expression, it must sit on one physical line ending `as level`, with no trailing comment: `bin/check_lints.py`'s check is line-anchored, not SQL-aware.
+
 **Copy guidance for Advisor surfaces:**
 - Keep `description` to 1-2 short sentences: what the lint detects and the first likely action.
 - Keep `detail` concise and object-specific. It should describe the failing object, not restate the full rationale.
@@ -108,16 +110,20 @@ order by
 
 Read `bin/installcheck`. Find line 55 — the long `psql` command that loads all lint files. It currently ends with something like:
 ```
--f lints/0024*.sql -d contrib_regression
+-f lints/<highest-numbered-existing-lint>*.sql -d contrib_regression
 ```
 
 Insert the new lint **before** `-d contrib_regression`, maintaining numeric order:
 ```
--f lints/0024*.sql -f lints/XXXX*.sql -d contrib_regression
+-f lints/<highest-numbered-existing-lint>*.sql -f lints/XXXX*.sql -d contrib_regression
 ```
 
 ## Step 4 — Create `docs/XXXX_<name>.md`
 
+- Pick the level your lint always emits.
+- If level varies per row (a `case` expression in Step 2), replace the `**Level:**` line with an explicit statement of which value maps to which level instead, e.g. "`**Level:** WARN (unreserved) or ERROR (lost)`".
+- Never leave the `**Level:** WARN|ERROR|INFO` placeholder unfilled; `bin/check_lints.py` rejects it whenever it names a level set different from what the lint's own SQL emits.
+
 ```markdown
 **Level:** WARN|ERROR|INFO
 
@@ -166,7 +172,7 @@ Cases where this lint may fire when the pattern is intentional, and how to handl
 
 ## Step 5 — Create `test/sql/XXXX_<name>.sql`
 
-Structure the test file exactly as follows:
+Structure the test file exactly as follows, unless the lint touches non-transactional catalog state (replication slots, `ALTER SYSTEM`) that a `rollback` can't undo — see `test/sql/0031_unused_replication_slot.sql` for the explicit-cleanup pattern that case needs instead.
 
 ```sql
 begin;
@@ -229,14 +235,18 @@ cp results/XXXX_<name>.out test/expected/XXXX_<name>.out
 
 ## Step 7 — Update `test/sql/queries_are_unionable.sql`
 
-Read the file. Before the final semicolon (on the last `select * from lint."0024_..."` line), append:
+Read the file. Before the final semicolon (on the last `select * from lint."<highest-numbered-existing-lint>"` line), append:
 ```sql
     union all
     select * from lint."XXXX_<name>"
 ```
 
 The file ends with a semicolon after the last view reference, then `rollback;`. Add the new entry before that semicolon.
 
+## Step 7b — Add the doc page to `mkdocs.yaml`'s nav
+
+Read `mkdocs.yaml` and add `docs/XXXX_<name>.md` under `nav:` -> `Lints:`, alongside the existing entries. `bin/check_lints.py` (see the Verification Checklist below) fails the build if a doc page exists but isn't linked from the nav.
+
 ## Step 8 — Verify and promote `test/expected/queries_are_unionable.out`
 
 The Docker run from Step 6 already produced `results/queries_are_unionable.out`. **Read and verify it:**
@@ -278,11 +288,9 @@ docker rmi -f dockerfiles-test && SUPABASE_VERSION=15.1.1.13 docker-compose -f d
 Then check:
 - [ ] `results/regression.diffs` is empty (no unexpected diffs)
 - [ ] `git diff test/expected/` shows only the new files you intentionally added/changed
-- [ ] `splinter.sql` includes the new lint in the `UNION ALL`
-- [ ] `lints/XXXX_<name>.sql` exists
-- [ ] `docs/XXXX_<name>.md` exists
-- [ ] `test/sql/XXXX_<name>.sql` exists
-- [ ] `test/expected/XXXX_<name>.out` exists
+- [ ] `lints/XXXX_<name>.sql` exists — `check_lints.py` iterates this glob, so a missing file just means fewer lints get checked, not an error
+- [ ] `splinter.sql` includes the new lint in the `UNION ALL` — regenerated automatically by the `compile-script` pre-commit hook, not checked by `check_lints.py`
+- [ ] `python bin/check_lints.py` passes — covers everything else: registration, nav linkage, and that each doc's `**Level:**` line matches what its view actually emits
 
 ---
 
@@ -295,4 +303,5 @@ Then check:
 | pgrst.db_schemas API exposure check | `lints/0023_sensitive_columns_exposed.sql`, `lints/0016_materialized_view_in_api.sql` |
 | pg_graphql extension-enabled check | `lints/0014_extension_in_public.sql` (for the `pg_catalog.pg_extension` pattern) |
 | begin/savepoint/rollback test structure | `test/sql/0024_rls_policy_always_true.sql` |
+| explicit-cleanup test structure (non-transactional catalog state) | `test/sql/0031_unused_replication_slot.sql` |
 | Doc format | `docs/0024_permissive_rls_policy.md` |

.pre-commit-config.yaml

@@ -36,3 +36,10 @@ repos:
     language: python
     always_run: true
     pass_filenames: false
+
+  - id: test-check-lints
+    name: Pinning tests for check_lints.py's level extractor
+    entry: python bin/test_check_lints.py
+    language: python
+    always_run: true
+    pass_filenames: false

bin/check_lints.py

@@ -12,10 +12,14 @@
     3. `test/sql/queries_are_unionable.sql` unions it
     4. `docs/NNNN_*.md` exists
     5. `test/sql/NNNN_*.sql` and `test/expected/NNNN_*.out` exist
-    6. `mkdocs.yaml` links that doc page
 
-It also checks that no two lints share a number, and that every doc page is in
-the nav. Run via pre-commit, or directly:
+It also checks that no two lints share a number, and, for every doc page
+(including one with no matching SQL view, e.g. 0012, an auth config check):
+
+    6. it is linked from `mkdocs.yaml`'s nav
+    7. its `**Level:**` line names exactly the level(s) the lint's own SQL emits
+
+Run via pre-commit, or directly:
 
     python bin/check_lints.py
 """
@@ -38,6 +42,34 @@
 }
 
 STEM_RE = re.compile(r"^(\d{4})_[a-z0-9_]+$")
+LEVEL_LINE_RE = re.compile(r"^\*\*Level:\*\*[ \t]*(.+)$", re.MULTILINE)
+LEVEL_WORD_RE = re.compile(r"\b(?:warn|error|info)\b", re.IGNORECASE)
+# Line-anchored on purpose: an unrelated `case`/`end` elsewhere in the query (a CTE, a nested case branch) never enters the extracted set.
+LEVEL_COL_RE = re.compile(
+    r"^\s*(.+?)\s+as\s+level\s*,?\s*$", re.MULTILINE | re.IGNORECASE
+)
+
+
+def _levels_in_lint_sql(sql: str) -> set[str]:
+    """The level(s) a lint's `... as level` column emits, or an empty set if no single matching column line is found: zero matches, more than one match (e.g. a CTE's own level column shadowing the view's real one), or a `case` expression split across multiple physical lines. A genuine single-line `case`, however nested, always has as many `end`s as `case`s; a truncated tail line left behind by a split expression has more `end`s than `case`s, however deep the leftover nesting -- either way this line-anchored check treats it as unparseable rather than risk silently extracting an incomplete level set from just that tail line."""
+    matches = LEVEL_COL_RE.findall(sql)
+    if len(matches) != 1:
+        return set()
+    column = matches[0]
+    if len(re.findall(r"\bend\b", column, re.IGNORECASE)) > len(
+        re.findall(r"\bcase\b", column, re.IGNORECASE)
+    ):
+        return set()
+    return {w.upper() for w in LEVEL_WORD_RE.findall(column)}
+
+
+def _level_line_error(value: str, sql_levels: set[str]) -> str | None:
+    doc_levels = {w.upper() for w in LEVEL_WORD_RE.findall(value)}
+    if not doc_levels:
+        return "does not mention WARN, ERROR, or INFO"
+    if sql_levels and doc_levels != sql_levels:
+        return f"says {sorted(doc_levels)} but the lint's own SQL emits {sorted(sql_levels)}"
+    return None
 
 
 def find_by_number(directory: Path, number: str, suffix: str) -> list[Path]:
@@ -52,17 +84,18 @@ def check() -> list[str]:
     mkdocs = MKDOCS.read_text()
 
     seen_numbers = {}
+    levels_by_number: dict[str, set[str]] = {}
 
     for lint_path in sorted(LINTS_DIR.glob("*.sql")):
         stem = lint_path.stem
-        match = STEM_RE.match(stem)
-        if not match:
+        stem_match = STEM_RE.match(stem)
+        if not stem_match:
             errors.append(
                 f"{lint_path}: name must be NNNN_snake_case (four digits, "
                 f"underscore, lowercase)"
             )
             continue
-        number = match.group(1)
+        number = stem_match.group(1)
 
         if number in seen_numbers:
             errors.append(
@@ -71,8 +104,20 @@ def check() -> list[str]:
             )
         seen_numbers[number] = lint_path
 
+        sql = lint_path.read_text()
+
+        sql_levels = _levels_in_lint_sql(sql)
+        if sql_levels:
+            levels_by_number[number] = sql_levels
+        else:
+            errors.append(
+                f"{lint_path}: could not determine the level(s) this lint "
+                f"emits (expected `'X' as level` or `case ... end as level`, "
+                f"on one physical line with no trailing comment)"
+            )
+
         # 1. view name matches the file stem
-        if f'create view lint."{stem}"' not in lint_path.read_text().lower():
+        if f'create view lint."{stem}"' not in sql.lower():
             errors.append(
                 f'{lint_path}: must declare `create view lint."{stem}"` '
                 f"(view name has to match the file name)"
@@ -93,8 +138,7 @@ def check() -> list[str]:
             )
 
         # 4. documented
-        docs = find_by_number(DOCS_DIR, number, ".md")
-        if not docs:
+        if not find_by_number(DOCS_DIR, number, ".md"):
             errors.append(f"{lint_path}: missing docs page {DOCS_DIR}/{number}_*.md")
 
         # 5. tested
@@ -108,14 +152,25 @@ def check() -> list[str]:
                 f"{TEST_EXPECTED_DIR}/{number}_*.out"
             )
 
-    # Step 6: every doc page is reachable from the nav. Covers docs for lints
-    # that are not SQL views too (e.g. 0012, an auth config check).
+    # steps 6-7 cover every doc page, including docs with no matching SQL view (e.g. 0012, an auth config check)
     for doc in sorted(DOCS_DIR.glob("[0-9]*.md")):
+        # 6. every doc page is reachable from the nav
         if doc.name not in mkdocs:
             errors.append(
                 f"{doc}: not listed in {MKDOCS}; add it under `nav:` -> `Lints:`"
             )
 
+        # 7. its Level line names exactly the level(s) the lint's own SQL emits
+        level_match = LEVEL_LINE_RE.search(doc.read_text())
+        if not level_match:
+            errors.append(f"{doc}: missing a '**Level:**' line")
+        else:
+            # Empty set (no matching lint view, or its own level column already errored above) skips the comparison in _level_line_error rather than silently passing.
+            doc_sql_levels = levels_by_number.get(doc.stem.split("_", 1)[0], set())
+            level_error = _level_line_error(level_match.group(1), doc_sql_levels)
+            if level_error:
+                errors.append(f"{doc}: Level line {level_error}")
+
     return sorted(set(errors))
 
 

bin/installcheck

@@ -30,7 +30,7 @@ rm -rf "$TMPDIR"
 # Initialize: setting PGUSER as the owner
 initdb --no-locale --encoding=UTF8 --nosync -U "$PGUSER"
 # Start the server
-pg_ctl start -o "-F -c listen_addresses=\"\" -c log_min_messages=WARNING -k $PGDATA"
+pg_ctl start -o "-F -c listen_addresses=\"\" -c log_min_messages=WARNING -c wal_level=logical -k $PGDATA"
 # Create the test db
 createdb contrib_regression
 
@@ -52,7 +52,7 @@ else
 fi
 
 # Execute the test fixtures
-psql -v ON_ERROR_STOP= -f test/fixtures.sql -f lints/0001*.sql -f lints/0002*.sql -f lints/0003*.sql -f lints/0004*.sql -f lints/0005*.sql -f lints/0006*.sql -f lints/0007*.sql -f lints/0008*.sql -f lints/0009*.sql -f lints/0010*.sql -f lints/0011*.sql -f lints/0013*.sql -f lints/0014*.sql -f lints/0015*.sql -f lints/0016*.sql -f lints/0017*.sql -f lints/0018*.sql -f lints/0019*.sql -f lints/0020*.sql -f lints/0021*.sql -f lints/0022*.sql -f lints/0023*.sql -f lints/0024*.sql -f lints/0025*.sql -f lints/0026*.sql -f lints/0027*.sql -f lints/0028*.sql -f lints/0029*.sql -f lints/0030*.sql -d contrib_regression
+psql -v ON_ERROR_STOP= -f test/fixtures.sql -f lints/0001*.sql -f lints/0002*.sql -f lints/0003*.sql -f lints/0004*.sql -f lints/0005*.sql -f lints/0006*.sql -f lints/0007*.sql -f lints/0008*.sql -f lints/0009*.sql -f lints/0010*.sql -f lints/0011*.sql -f lints/0013*.sql -f lints/0014*.sql -f lints/0015*.sql -f lints/0016*.sql -f lints/0017*.sql -f lints/0018*.sql -f lints/0019*.sql -f lints/0020*.sql -f lints/0021*.sql -f lints/0022*.sql -f lints/0023*.sql -f lints/0024*.sql -f lints/0025*.sql -f lints/0026*.sql -f lints/0027*.sql -f lints/0028*.sql -f lints/0029*.sql -f lints/0030*.sql -f lints/0031*.sql -d contrib_regression
 
 # Run tests
 ${REGRESS} --use-existing --dbname=contrib_regression --inputdir=${TESTDIR} ${TESTS}

bin/test_check_lints.py

@@ -0,0 +1,63 @@
+"""Pinning tests for check_lints.py's level-column extractor.
+
+Run directly: python3 bin/test_check_lints.py
+
+No test framework in this repo (test/ is pg_regress only) -- assert-based,
+so a future edit that re-breaks the multi-line-case guard fails loudly here
+instead of only in a mutation test someone has to remember to write by hand.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from check_lints import _level_line_error, _levels_in_lint_sql  # noqa: E402
+
+CASES = {
+    "bare literal": ("'INFO' as level,", {"INFO"}),
+    "single-line case (0031's own shape)": (
+        "case when prs.wal_status = 'lost' then 'ERROR' else 'WARN' end as level,",
+        {"ERROR", "WARN"},
+    ),
+    "single-line nested case": (
+        "case when a then 'ERROR' else case when b then 'WARN' else 'INFO' end end as level,",
+        {"ERROR", "WARN", "INFO"},
+    ),
+    "multi-line case, bare end tail": (
+        "case\n    when a then 'ERROR'\n    else 'WARN'\nend as level,",
+        set(),
+    ),
+    "multi-line case, value inlined on tail": (
+        "case when a then 'ERROR'\n     else 'WARN' end as level,",
+        set(),
+    ),
+    "multi-line case, nested case on tail": (
+        "case when a then 'ERROR'\n     else case when b then 'INFO' else 'WARN' end end as level,",
+        set(),
+    ),
+    "trailing comment": ("'WARN' as level, -- always warn", set()),
+}
+
+failures = []
+for name, (column_line, expected) in CASES.items():
+    sql = f"select\n    {column_line}\n    y\nfrom z;"
+    actual = _levels_in_lint_sql(sql)
+    if actual != expected:
+        failures.append(f"{name}: expected {expected!r}, got {actual!r}")
+
+# An empty sql_levels set means _level_line_error skips the comparison rather than silently passing; a doc mentioning no level word is still always an error.
+if _level_line_error("WARN", set()) is not None:
+    failures.append(
+        "_level_line_error('WARN', set()) should skip comparison, not error"
+    )
+if _level_line_error("nothing here", set()) is None:
+    failures.append("_level_line_error with no level word should always error")
+if _level_line_error("WARN", {"ERROR"}) is None:
+    failures.append("_level_line_error should error on a real mismatch")
+
+if failures:
+    print(f"{len(failures)} pinning test(s) failed:\n")
+    for failure in failures:
+        print(f"  - {failure}")
+    raise SystemExit(1)
+print(f"all {len(CASES)} level-extraction pinning tests passed")

docs/0007_policy_exists_rls_disabled.md

@@ -1,5 +1,5 @@
 
-**Level:** INFO
+**Level:** ERROR
 
 **Summary:** Security policy not enforced