feat(bulldozer): add verifyDataIntegrity to all Table types

apps/backend/scripts/verify-data-integrity/index.ts

@@ -1,4 +1,7 @@
+import { toQueryableSqlQuery } from "@/lib/bulldozer/db/index";
+import { tableIdToDebugString } from "@/lib/bulldozer/db/utilities";
 import { syncExternalDatabases } from "@/lib/external-db-sync";
+import { createPaymentsSchema } from "@/lib/payments/schema/index";
 import { DEFAULT_BRANCH_ID, getSoleTenancyFromProjectBranch } from "@/lib/tenancies";
 import { getPrismaClientForTenancy, globalPrismaClient } from "@/prisma-client";
 import type { OrganizationRenderedConfig } from "@stackframe/stack-shared/dist/config/schema";
@@ -168,6 +171,21 @@ async function main() {
     console.log(`Will check at most ${maxUsersPerProject} users per project.`);
   }
 
+  await recurse(`[bulldozer] verifying data integrity across all payments tables`, async () => {
+    const schema = createPaymentsSchema();
+    for (const table of schema._allTables) {
+      const label = tableIdToDebugString(table.tableId);
+      await recurse(`[bulldozer table] ${label}`, async () => {
+        const errors = await prismaClient.$queryRawUnsafe<unknown[]>(toQueryableSqlQuery(table.verifyDataIntegrity()));
+        if (errors.length > 0) {
+          throw new StackAssertionError(deindent`
+            Bulldozer data integrity violation in table ${label}: found ${errors.length} error row(s).
+          `, { errors });
+        }
+      });
+    }
+  });
+
   const endAt = Math.min(startAt + count, projects.length);
   for (let i = startAt; i < endAt; i++) {
     const projectId = projects[i].id;

apps/backend/src/lib/bulldozer/db/index.fuzz.test.ts

@@ -1,7 +1,48 @@
 import { stringCompare } from "@stackframe/stack-shared/dist/utils/strings";
 import postgres from "postgres";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, test } from "vitest";
-import { declareCompactTable, declareConcatTable, declareFilterTable, declareFlatMapTable, declareGroupByTable, declareLeftJoinTable, declareLFoldTable, declareLimitTable, declareMapTable, declareReduceTable, declareSortTable, declareStoredTable, declareTimeFoldTable, toExecutableSqlTransaction, toQueryableSqlQuery } from "./index";
+import type { Table } from "./index";
+import {
+  declareCompactTable as _declareCompactTable,
+  declareConcatTable as _declareConcatTable,
+  declareFilterTable as _declareFilterTable,
+  declareFlatMapTable as _declareFlatMapTable,
+  declareGroupByTable as _declareGroupByTable,
+  declareLeftJoinTable as _declareLeftJoinTable,
+  declareLFoldTable as _declareLFoldTable,
+  declareLimitTable as _declareLimitTable,
+  declareMapTable as _declareMapTable,
+  declareReduceTable as _declareReduceTable,
+  declareSortTable as _declareSortTable,
+  declareStoredTable as _declareStoredTable,
+  declareTimeFoldTable as _declareTimeFoldTable,
+  toExecutableSqlTransaction,
+  toQueryableSqlQuery,
+} from "./index";
+
+// any is used here because the verifier works with heterogeneous table types
+const allInitializedTables: Table<any, any, any>[] = [];
+function trackTable<T extends Table<any, any, any>>(table: T): T {
+  allInitializedTables.push(table);
+  return table;
+}
+function tracked<Fn extends (...args: any[]) => Table<any, any, any>>(fn: Fn): Fn {
+  return ((...args: unknown[]) => trackTable(fn(...args))) as Fn;
+}
+
+const declareCompactTable = tracked(_declareCompactTable);
+const declareConcatTable = tracked(_declareConcatTable);
+const declareFilterTable = tracked(_declareFilterTable);
+const declareFlatMapTable = tracked(_declareFlatMapTable);
+const declareGroupByTable = tracked(_declareGroupByTable);
+const declareLeftJoinTable = tracked(_declareLeftJoinTable);
+const declareLFoldTable = tracked(_declareLFoldTable);
+const declareLimitTable = tracked(_declareLimitTable);
+const declareMapTable = tracked(_declareMapTable);
+const declareReduceTable = tracked(_declareReduceTable);
+const declareSortTable = tracked(_declareSortTable);
+const declareStoredTable = tracked(_declareStoredTable);
+const declareTimeFoldTable = tracked(_declareTimeFoldTable);
 
 type TestDb = { full: string, base: string };
 
@@ -675,7 +716,13 @@ describe.sequential("bulldozer db fuzz composition (real postgres)", () => {
     `;
   });
 
-  afterEach(() => {
+  afterEach(async () => {
+    for (const table of allInitializedTables) {
+      const errors = await readRows(table.verifyDataIntegrity(), "afterEach.verifyDataIntegrity");
+      expect(errors).toEqual([]);
+    }
+    allInitializedTables.length = 0;
+
     if (!FUZZ_TRACE_ENABLED) return;
     const testName = getCurrentTestNameForTrace();
     const bucket = tracesByTest.get(testName);

apps/backend/src/lib/bulldozer/db/index.perf.test.ts

@@ -1,7 +1,48 @@
 import { stringCompare } from "@stackframe/stack-shared/dist/utils/strings";
 import postgres from "postgres";
-import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
-import { declareCompactTable, declareConcatTable, declareFilterTable, declareFlatMapTable, declareGroupByTable, declareLeftJoinTable, declareLFoldTable, declareLimitTable, declareMapTable, declareReduceTable, declareSortTable, declareStoredTable, declareTimeFoldTable, toExecutableSqlTransaction, toQueryableSqlQuery } from "./index";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import type { Table } from "./index";
+import {
+  declareCompactTable as _declareCompactTable,
+  declareConcatTable as _declareConcatTable,
+  declareFilterTable as _declareFilterTable,
+  declareFlatMapTable as _declareFlatMapTable,
+  declareGroupByTable as _declareGroupByTable,
+  declareLeftJoinTable as _declareLeftJoinTable,
+  declareLFoldTable as _declareLFoldTable,
+  declareLimitTable as _declareLimitTable,
+  declareMapTable as _declareMapTable,
+  declareReduceTable as _declareReduceTable,
+  declareSortTable as _declareSortTable,
+  declareStoredTable as _declareStoredTable,
+  declareTimeFoldTable as _declareTimeFoldTable,
+  toExecutableSqlTransaction,
+  toQueryableSqlQuery,
+} from "./index";
+
+// any is used here because the verifier works with heterogeneous table types
+const allInitializedTables: Table<any, any, any>[] = [];
+function trackTable<T extends Table<any, any, any>>(table: T): T {
+  allInitializedTables.push(table);
+  return table;
+}
+function tracked<Fn extends (...args: any[]) => Table<any, any, any>>(fn: Fn): Fn {
+  return ((...args: unknown[]) => trackTable(fn(...args))) as Fn;
+}
+
+const declareCompactTable = tracked(_declareCompactTable);
+const declareConcatTable = tracked(_declareConcatTable);
+const declareFilterTable = tracked(_declareFilterTable);
+const declareFlatMapTable = tracked(_declareFlatMapTable);
+const declareGroupByTable = tracked(_declareGroupByTable);
+const declareLeftJoinTable = tracked(_declareLeftJoinTable);
+const declareLFoldTable = tracked(_declareLFoldTable);
+const declareLimitTable = tracked(_declareLimitTable);
+const declareMapTable = tracked(_declareMapTable);
+const declareReduceTable = tracked(_declareReduceTable);
+const declareSortTable = tracked(_declareSortTable);
+const declareStoredTable = tracked(_declareStoredTable);
+const declareTimeFoldTable = tracked(_declareTimeFoldTable);
 
 type TestDb = { full: string, base: string };
 type SqlExpression<T> = { type: "expression", sql: string };
@@ -315,6 +356,14 @@
     `;
   });
 
+  afterEach(async () => {
+    for (const table of allInitializedTables) {
+      const errors = await readRows(table.verifyDataIntegrity());
+      expect(errors).toEqual([]);
+    }
+    allInitializedTables.length = 0;
+  });
+
   afterAll(async () => {
     await sql.end();
     await adminSql.unsafe(`