feat: queues, partition sets, fix hazards, data-checked --prove, stale-migration guard, schema files as a source

README.md

@@ -26,7 +26,7 @@ Built by **[Akal](https://github.com/akalsoftware)**.
   - RLS policies and RLS coverage;
   - storage buckets and their policies;
   - auth config and Edge Functions;
-  - cron jobs, webhooks and Realtime;
+  - cron jobs, queues, webhooks and Realtime;
   - Vault, extensions, roles and grants;
   - reference data and migration history.
 - **Fixes, not just reports.** Fixes run in dependency order, in one

packages/cli/README.md

@@ -9,13 +9,16 @@ order, in one transaction, and can be proved on a throwaway copy before
 anything changes. With a single project, it snapshots, restores and clones it.
 
 - **The whole project.** It covers schema, RLS, storage, auth, Edge Functions,
-  cron, webhooks, Realtime, Vault, extensions, grants, reference data and
-  migration history.
+  cron, webhooks, Realtime, queues, partition sets, Vault, extensions, grants,
+  reference data and migration history.
 - **Safe by default.** Every write needs `--apply`, and anything that loses
   data or opens up access also needs `--allow-destructive`. A failed apply
   leaves the target exactly as it was.
+- **Production-aware.** Each fix says what it does to a live database: the
+  lock it holds and on which table, whether it may rewrite the table or scan
+  every row, whether it can't run in a transaction.
 - **Provable.** `--prove` rehearses the migration and refuses to apply one
-  that wouldn't reproduce the source.
+  that wouldn't reproduce the source or would lose rows.
 - **Anywhere.** It works with Supabase Cloud, self-hosted stacks, plain
   PostgreSQL targets, CI pipelines and AI agents (MCP).
 
@@ -63,6 +66,8 @@ Docker image needs nothing at all, and carries the PostgreSQL client tools
 | Storage | Buckets and their policies (objects are never copied) | API and SQL |
 | Auth config | Management API, or GoTrue on self-hosted | API (hosted) |
 | Cron jobs | `cron.job` | SQL |
+| Queues | Supabase Queues (`pgmq.meta`), created and dropped through pgmq | SQL |
+| Partition sets | pg_partman registrations (`part_config`), made with `create_parent` | SQL |
 | Webhooks | Triggers calling `supabase_functions.http_request` | SQL |
 | Realtime | Publications, and policies on `realtime.messages` | SQL |
 | Vault | Secret names and descriptions (values can't be read) | Command to run |
@@ -71,10 +76,17 @@ Docker image needs nothing at all, and carries the PostgreSQL client tools
 | Roles and grants | Custom roles, and grants on tables, views, sequences and functions, including `anon`, `authenticated`, `service_role` | SQL |
 
 The schema check covers tables, columns, keys, indexes, partitions, views,
-materialized views, functions, triggers, sequences, enums, composite types,
-domains, policies, extensions and comments. It runs in every schema except
-Supabase's own (`auth`, `storage`, and so on) and those in `ignoreSchemas`.
-Objects owned by an extension belong to it and aren't compared.
+materialized views, functions, aggregates, triggers, event triggers, rules,
+sequences, enums, composite and range types, domains, casts, extended
+statistics, text search configurations and dictionaries, foreign servers,
+user mappings and foreign tables, policies, default privileges, replica
+identity, statistics targets, extensions and comments. A renamed column,
+table or sequence is renamed, keeping its rows, rather than dropped and
+created again. It runs in every schema except Supabase's own (`auth`,
+`storage`, and so on) and those in `ignoreSchemas`. Objects owned by an
+extension belong to it and aren't compared. Objects of a kind it doesn't
+compare, such as operator classes, are named under the check so you can look
+at them by hand; they aren't counted as drift.
 
 A check that can't run, for example without API credentials, says so and why.
 It's never reported as a clean pass.
@@ -109,6 +121,7 @@ Useful `diff` flags:
 | `--only=<id,...>` | Apply only these findings (ids come from `--json`, globs allowed) |
 | `--dry-run` | Print the fixes in the order they'd run, run nothing |
 | `--prove` | Rehearse the fixes on a throwaway copy first |
+| `--source-dir <dir>` | Compare a directory of schema files (such as `supabase/schemas`) with the target instead of a source database |
 | `--allow-destructive` | Also apply drops and deletes |
 | `--apply-posture` | Also apply the target-only checks' fixes |
 | `--no-transaction` | Apply fix by fix, keeping what works |
@@ -129,11 +142,13 @@ Useful `diff` flags:
   is unchanged. A fix the target can't hold, because it needs an extension or
   a Supabase schema the target lacks, is skipped with the reason, and the rest
   still applies.
-- **`--prove`.** The fixes are replayed on a structure-only copy of the target,
-  made on the target's own server, the way the apply will run them. Nothing is
-  applied unless the result matches the source, and nothing is applied if the
-  proof can't run. Changes a copy can't hold, such as cron jobs (pg_cron lives
-  in one database per server), are listed as not proved.
+- **`--prove`.** The fixes are replayed on a copy of the target's structure,
+  made on the target's own server, the way the apply will run them, with a
+  sample of each table's rows. Nothing is applied unless the result matches
+  the source and the rows survive, and nothing is applied if the proof can't
+  run. It also names the tables the fixes rewrite. Changes a copy can't hold,
+  such as cron jobs (pg_cron lives in one database per server), are listed as
+  not proved.
 
 ```bash
 supaforge diff --dry-run                 # the plan, in order
@@ -260,6 +275,7 @@ supaforge migrate run --env=prod --apply
 | `SUPAFORGE_DBDIFF_MEMORY` | `1G` | Memory limit for `@dbdiff/cli` (`2G`, `-1` for none) |
 | `SUPAFORGE_CONNECT_TIMEOUT` | `15` | Seconds to wait for a connection |
 | `SUPAFORGE_CHECK_CONCURRENCY` | `4` | Checks run at once. Lower it behind a tight pooler |
+| `SUPAFORGE_PROVE_ROWS` | `1000` | Rows of each table `--prove` copies to check they survive; `0` checks structure only |
 | `SUPAFORGE_PG_BIN` | — | A directory of PostgreSQL client tools (`pg_dump`, `psql`), used ahead of `PATH` — for a server newer than the tools on `PATH` |
 
 ## Self-hosted Supabase

packages/cli/src/checks/base.ts

@@ -18,6 +18,8 @@ export interface CheckContext {
    * fine-grained to report simply ignore it.
    */
   onDetail?: (detail: string) => void
+  /** Sink for what the check could not compare — see CheckResult.notices. */
+  onNotice?: (notice: string) => void
   source: EnvironmentConfig
   target: EnvironmentConfig
   config: SupaForgeConfig

packages/cli/src/checks/extensions.ts

@@ -22,7 +22,8 @@ export class ExtensionsCheck extends Check {
       this.fetchExtensions(ctx.source.dbUrl),
       this.fetchExtensions(ctx.target.dbUrl),
     ])
-    return diffExtensions(source, target)
+    const ignored = new Set(ctx.config.ignoreExtensions ?? [])
+    return diffExtensions(source, target.filter(e => !ignored.has(e.name) || source.some(s => s.name === e.name)))
   }
 
   private async fetchExtensions(dbUrl: string): Promise<PgExtension[]> {

packages/cli/src/checks/index.ts

@@ -13,6 +13,8 @@ import { VaultCheck } from './vault.js'
 import { ExtensionsCheck } from './extensions.js'
 import { MigrationsCheck } from './migrations.js'
 import { RolesCheck } from './roles.js'
+import { QueuesCheck } from './queues.js'
+import { PartitionsCheck } from './partitions.js'
 
 export interface RegistryOptions {
   /** Enable file-level drift detection in the storage check. */
@@ -35,6 +37,8 @@ export function createDefaultRegistry(options: RegistryOptions = {}): CheckRegis
   registry.register(new ExtensionsCheck())
   registry.register(new MigrationsCheck())
   registry.register(new RolesCheck())
+  registry.register(new QueuesCheck())
+  registry.register(new PartitionsCheck())
   return registry
 }
 
@@ -54,3 +58,5 @@ export { VaultCheck } from './vault.js'
 export { ExtensionsCheck } from './extensions.js'
 export { MigrationsCheck } from './migrations.js'
 export { RolesCheck } from './roles.js'
+export { QueuesCheck } from './queues.js'
+export { PartitionsCheck } from './partitions.js'

packages/cli/src/checks/partitions.ts

@@ -0,0 +1,179 @@
+import type { QueryFn } from '../db'
+import { pgQuery } from '../db'
+import type { DriftIssue } from '../types/drift'
+import { quoteIdent, quoteLiteral } from '../utils/sql'
+import { Check, CheckSkipped, type CheckContext } from './base'
+
+/** A pg_partman partition set, as its part_config row records it. */
+export interface PartitionSet {
+  parent_table: string
+  control: string
+  partition_interval: string
+  partition_type: string
+  epoch: string
+  premake: number
+  automatic_maintenance: string
+  constraint_cols: string[] | null
+  jobmon: boolean
+  date_trunc_interval: string | null
+  time_encoder: string | null
+  time_decoder: string | null
+  has_default: boolean
+  /** The settings part_config takes by UPDATE, by column. */
+  settings: Record<string, unknown>
+}
+
+/** What decides how the set is partitioned: changed, it has to be rebuilt. */
+const SHAPE = ['control', 'partition_interval', 'partition_type', 'epoch', 'time_encoder', 'time_decoder'] as const
+
+/** part_config columns pg_partman documents setting with UPDATE. */
+const SETTINGS = [
+  'premake', 'automatic_maintenance', 'retention', 'retention_schema', 'retention_keep_index', 'retention_keep_table',
+  'optimize_constraint', 'infinite_time_partitions', 'inherit_privileges', 'constraint_valid', 'ignore_default_data',
+  'maintenance_order', 'retention_keep_publication', 'detach_before_drop', 'jobmon', 'date_trunc_interval',
+]
+
+/**
+ * pg_partman partition sets, compared by their registration.
+ *
+ * The partitions themselves are pg_partman's to make and drop, so the schema
+ * diff leaves them alone; what makes a set is its row in part_config, written
+ * by create_parent(). A set the target lacks is registered through that
+ * function, and its settings put as the source has them; a setting that
+ * differs is updated where it stands. How the set is partitioned — its
+ * control column, interval or type — cannot be changed in place, so a
+ * difference there, and a set only the target has, are left to the user.
+ */
+export class PartitionsCheck extends Check {
+  readonly name = 'partitions' as const
+
+  constructor(private queryFn: QueryFn = pgQuery) {
+    super()
+  }
+
+  async scan(ctx: CheckContext): Promise<DriftIssue[]> {
+    const [source, target] = await Promise.all([this.fetch(ctx.source.dbUrl), this.fetch(ctx.target.dbUrl)])
+    if (source.schema === null && target.schema === null) {
+      throw new CheckSkipped('pg_partman extension not installed')
+    }
+    return diffPartitionSets(source.sets, target.sets, target.schema)
+  }
+
+  private async fetch(dbUrl: string): Promise<{ schema: string | null; sets: PartitionSet[] }> {
+    const rows = await this.queryFn(dbUrl, `SELECT n.nspname AS schema FROM pg_extension e
+      JOIN pg_namespace n ON n.oid = e.extnamespace WHERE e.extname = 'pg_partman'`) as Array<{ schema: string }>
+    const schema = rows[0]?.schema ?? null
+    if (schema === null) return { schema, sets: [] }
+    const sets = await this.queryFn(dbUrl, `
+      SELECT (SELECT n.nspname || '.' || c.relname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
+               WHERE c.oid = to_regclass(pc.parent_table)) AS parent_table, pc.control, pc.partition_interval, pc.partition_type,
+             pc.epoch, pc.premake, pc.automatic_maintenance, pc.constraint_cols, pc.jobmon, pc.date_trunc_interval,
+             pc.time_encoder, pc.time_decoder,
+             to_regclass(pc.parent_table || '_default') IS NOT NULL AS has_default,
+             (to_jsonb(pc) - ARRAY(SELECT k FROM jsonb_object_keys(to_jsonb(pc)) k WHERE k <> ALL($1::text[]))) AS settings
+        FROM ${quoteIdent(schema)}.part_config pc
+       WHERE to_regclass(pc.parent_table) IS NOT NULL
+       ORDER BY 1`, [SETTINGS]) as unknown as PartitionSet[]
+    return { schema, sets }
+  }
+}
+
+const literal = (v: unknown): string =>
+  v === null || v === undefined ? 'NULL'
+    : typeof v === 'boolean' || typeof v === 'number' ? String(v)
+      : Array.isArray(v) ? `ARRAY[${v.map(x => quoteLiteral(String(x))).join(', ')}]::text[]`
+        : quoteLiteral(String(v))
+
+function registerSql(set: PartitionSet, schema: string): string {
+  const args = [
+    `p_parent_table => ${quoteLiteral(set.parent_table)}`,
+    `p_control => ${quoteLiteral(set.control)}`,
+    `p_interval => ${quoteLiteral(set.partition_interval)}`,
+    `p_type => ${quoteLiteral(set.partition_type)}`,
+    `p_epoch => ${quoteLiteral(set.epoch)}`,
+    `p_premake => ${set.premake}`,
+    `p_default_table => ${set.has_default}`,
+    `p_automatic_maintenance => ${quoteLiteral(set.automatic_maintenance)}`,
+    `p_jobmon => ${set.jobmon}`,
+    ...(set.constraint_cols?.length ? [`p_constraint_cols => ${literal(set.constraint_cols)}`] : []),
+    ...(set.date_trunc_interval ? [`p_date_trunc_interval => ${quoteLiteral(set.date_trunc_interval)}`] : []),
+    ...(set.time_encoder ? [`p_time_encoder => ${quoteLiteral(set.time_encoder)}`] : []),
+    ...(set.time_decoder ? [`p_time_decoder => ${quoteLiteral(set.time_decoder)}`] : []),
+  ]
+  return `SELECT ${quoteIdent(schema)}.create_parent(${args.join(', ')});`
+}
+
+function updateSql(schema: string, parent: string, settings: Record<string, unknown>): string {
+  const sets = Object.entries(settings).map(([k, v]) => `${quoteIdent(k)} = ${literal(v)}`)
+  return `UPDATE ${quoteIdent(schema)}.part_config SET ${sets.join(', ')} WHERE parent_table = ${quoteLiteral(parent)};`
+}
+
+function undoSql(schema: string, parent: string): string {
+  return `DELETE FROM ${quoteIdent(schema)}.part_config WHERE parent_table = ${quoteLiteral(parent)};`
+}
+
+export function diffPartitionSets(source: PartitionSet[], target: PartitionSet[], targetSchema: string | null): DriftIssue[] {
+  const issues: DriftIssue[] = []
+  const targetMap = new Map(target.map(s => [s.parent_table, s]))
+  const sourceNames = new Set(source.map(s => s.parent_table))
+
+  for (const s of source) {
+    const t = targetMap.get(s.parent_table)
+    if (!t) {
+      issues.push({
+        id: `partitions-missing-${s.parent_table}`,
+        check: 'partitions',
+        severity: 'warning',
+        title: `Partition set not registered: ${s.parent_table}`,
+        description: `pg_partman manages ${s.parent_table} in source (${s.partition_type} by ${s.control}, every ${s.partition_interval}) but not in target.`,
+        sourceValue: s,
+        ...(targetSchema
+          ? { sql: { up: `${registerSql(s, targetSchema)}\n${updateSql(targetSchema, s.parent_table, s.settings)}`, down: undoSql(targetSchema, s.parent_table) } }
+          : { manualOnly: 'the target does not have pg_partman; install it, then re-run to register the partition set' }),
+      })
+      continue
+    }
+    const reshaped = SHAPE.filter(k => String(s[k] ?? '') !== String(t[k] ?? ''))
+    if (reshaped.length > 0) {
+      issues.push({
+        id: `partitions-shape-${s.parent_table}`,
+        check: 'partitions',
+        severity: 'warning',
+        title: `Partition set differs: ${s.parent_table}`,
+        description: `${reshaped.join(', ')} differ${reshaped.length === 1 ? 's' : ''} for ${s.parent_table}. pg_partman cannot change that in place.`,
+        sourceValue: s,
+        targetValue: t,
+        manualOnly: 'changing how a set is partitioned means undoing and recreating it with pg_partman (undo_partition, then create_parent)',
+      })
+      continue
+    }
+    const changed = Object.fromEntries(Object.entries(s.settings).filter(([k, v]) => JSON.stringify(v) !== JSON.stringify(t.settings[k])))
+    if (Object.keys(changed).length > 0 && targetSchema) {
+      const previous = Object.fromEntries(Object.keys(changed).map(k => [k, t.settings[k]]))
+      issues.push({
+        id: `partitions-settings-${s.parent_table}`,
+        check: 'partitions',
+        severity: 'warning',
+        title: `Partition set settings differ: ${s.parent_table}`,
+        description: `${Object.keys(changed).join(', ')} differ for ${s.parent_table}.`,
+        sourceValue: s.settings,
+        targetValue: t.settings,
+        sql: { up: updateSql(targetSchema, s.parent_table, changed), down: updateSql(targetSchema, s.parent_table, previous) },
+      })
+    }
+  }
+
+  for (const t of target) {
+    if (sourceNames.has(t.parent_table)) continue
+    issues.push({
+      id: `partitions-extra-${t.parent_table}`,
+      check: 'partitions',
+      severity: 'info',
+      title: `Extra partition set: ${t.parent_table}`,
+      description: `pg_partman manages ${t.parent_table} in target but not in source.`,
+      targetValue: t,
+      manualOnly: 'unregistering a set leaves its partitions as ordinary tables; use pg_partman\'s undo_partition if that is what you want',
+    })
+  }
+  return issues
+}