import { readdirSync, readFileSync } from 'fs' import { dirname, join } from 'path' import { fileURLToPath } from 'url' import pg from 'pg' import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest' // Focused integration test for the effect_outbox trigger (migrations 047, 048), the // generic transactional outbox that TLFileEffectProcessor (sync-worker) drains. The // design-critical case is #5 below: cleanup_deleted_group_trigger (023_groups.sql) // soft-deletes group-owned files from inside a plpgsql trigger when a group is // deleted, and file_effect_outbox_after_change must still fire for that UPDATE — // a trigger-on-trigger cascade that only a real Postgres can exercise. // // This talks to a real postgres (the triggers are plpgsql, so fakes can't exercise // them). It is opt-in: set ZERO_CACHE_TEST_POSTGRES_URL (local dev stack: // postgres://user:password@localhost:6543/postgres) to run it. Without a connection // string the suite is skipped so CI stays green. // // Isolation differs from stamp_comment_created_at.test.ts / delete_file_states.test.ts: // those pin `search_path` to a throwaway schema because their migration SQL uses // unqualified table names. 047/048 hardcode `public.` (CREATE TABLE public.effect_outbox, // CREATE TRIGGER ... ON public.file), so search_path can't redirect them — running them // verbatim always targets the connection's `public` schema. Since ZERO_CACHE_TEST_POSTGRES_URL // points at a real local dev stack whose `public` schema holds real dev data, this suite // instead creates a throwaway database with CREATE DATABASE, and drops it in afterAll. // // Owning a whole database also means the REAL migration chain (000 onward) can run // verbatim, exactly like migrate.ts applies it. Every trigger involved — the outbox // trigger, the group-delete cascade, the updatedAt bumper (005), the owner-details // denormalizers — is the shipped one, not a test copy, so a behavior change in any // migration is caught here. const CONNECTION_STRING = process.env.ZERO_CACHE_TEST_POSTGRES_URL const MIGRATIONS_DIR = join(dirname(fileURLToPath(import.meta.url)), 'migrations') // The full, shipped migration chain in filename order — the same ordering migrate.ts // uses (readdirSync().sort()). const MIGRATION_FILES = readdirSync(MIGRATIONS_DIR) .filter((f) => f.endsWith('.sql')) .sort() const dbName = `tldraw_test_outbox_${process.pid}` const describeMaybe = CONNECTION_STRING ? describe : describe.skip if (!CONNECTION_STRING) { // eslint-disable-next-line no-console console.warn( 'effect_outbox.test.ts: skipping — TLFileEffectProcessor depends on this trigger; set ' + 'ZERO_CACHE_TEST_POSTGRES_URL to verify it against a real postgres.' ) } describeMaybe('effect_outbox trigger (file changes + group-delete cascade)', () => { // One client on the admin connection (for CREATE/DROP DATABASE, which cannot run // inside a transaction) and one on the throwaway database (where the migration // chain and all DML run un-namespaced, matching production). let adminClient: pg.Client let client: pg.Client beforeAll(async () => { adminClient = new pg.Client({ connectionString: CONNECTION_STRING }) await adminClient.connect() await adminClient.query(`DROP DATABASE IF EXISTS "${dbName}"`) await adminClient.query(`CREATE DATABASE "${dbName}"`) const url = new URL(CONNECTION_STRING!) url.pathname = `/${dbName}` client = new pg.Client({ connectionString: url.toString() }) await client.connect() // Apply the real migration chain, in order, verbatim. for (const filename of MIGRATION_FILES) { const sql = readFileSync(join(MIGRATIONS_DIR, filename), 'utf8') try { await client.query(sql) } catch (err) { throw new Error(`Migration ${filename} failed: ${(err as Error).message}`) } } }) afterAll(async () => { if (client) await client.end() if (adminClient) { await adminClient.query(`DROP DATABASE IF EXISTS "${dbName}"`) await adminClient.end() } }) beforeEach(async () => { // TRUNCATE fires no row-level triggers, so resetting state writes no outbox rows. // CASCADE clears referencing tables (file_state, group_file, group_user, ...). await client.query(`TRUNCATE "user", "file", "group", effect_outbox CASCADE`) }) // Also creates the user's home workspace, mirroring production, where a home group's id is // the user's own id. `file."owningGroupId"` is NOT NULL since 050, so every file needs one. async function seedUser(id: string) { await client.query( `INSERT INTO "user" ("id", "name", "email", "avatar", "color", "exportFormat", "exportTheme", "exportBackground", "exportPadding", "createdAt", "updatedAt", "flags") VALUES ($1, $1, $1, '', '', 'png', 'auto', false, false, 0, 0, '')`, [id] ) await client.query(`INSERT INTO "group" ("id", "name") VALUES ($1, $1)`, [id]) } async function seedFile(id: string, overrides: { owningGroupId?: string } = {}) { const owningGroupId = overrides.owningGroupId ?? 'u1' await client.query( `INSERT INTO "file" ("id", "name", "owningGroupId", "thumbnail", "shared", "sharedLinkType", "published", "lastPublished", "publishedSlug", "createdAt", "updatedAt", "isEmpty") VALUES ($1, $1, $2, '', false, 'view', false, 0, $1, 0, 0, false)`, [id, owningGroupId] ) } interface OutboxRow { tableName: string entityId: string command: string payload: any prevPayload: any attempts: number nextRetryAt: string | null lastError: string | null } async function outboxRows(entityId: string): Promise { const res = await client.query( `SELECT "tableName", "entityId", command, payload, "prevPayload", attempts, "nextRetryAt", "lastError" FROM effect_outbox WHERE "entityId" = $1 ORDER BY id`, [entityId] ) return res.rows } it('file INSERT produces one outbox row with the new row as payload and no prevPayload', async () => { await seedUser('u1') await seedFile('f1') const rows = await outboxRows('f1') expect(rows).toHaveLength(1) expect(rows[0].tableName).toBe('file') expect(rows[0].command).toBe('insert') expect(rows[0].payload.id).toBe('f1') expect(rows[0].prevPayload).toBeNull() }) it('a no-op update (bumping only updatedAt) produces no new outbox row', async () => { await seedUser('u1') await seedFile('f1') // This is the room-persist path: sync-worker bumps updatedAt every few seconds // per active room. The 005 trigger rewrites updatedAt to now() on any distinct // update, so only updatedAt differs between OLD and NEW — the outbox filter // must treat that as a no-op. await client.query(`UPDATE "file" SET "updatedAt" = 12345 WHERE "id" = 'f1'`) const rows = await outboxRows('f1') // only the insert row from seedFile expect(rows).toHaveLength(1) expect(rows[0].command).toBe('insert') }) it('an effect-relevant update (publish) produces an update row carrying the full OLD row as prevPayload', async () => { await seedUser('u1') await seedFile('f1') await client.query( `UPDATE "file" SET "published" = true, "lastPublished" = 999 WHERE "id" = 'f1'` ) const rows = await outboxRows('f1') expect(rows).toHaveLength(2) const updateRow = rows[1] expect(updateRow.command).toBe('update') expect(updateRow.payload.published).toBe(true) expect(updateRow.payload.lastPublished).toBe(999) // prevPayload is the full OLD row: changed columns show their old values, // unchanged columns are present and equal expect(updateRow.prevPayload.published).toBe(false) expect(updateRow.prevPayload.lastPublished).toBe(0) expect(updateRow.prevPayload.id).toBe('f1') expect(updateRow.prevPayload.name).toBe('f1') expect(updateRow.prevPayload.owningGroupId).toBe('u1') }) it('file DELETE produces a delete row with the OLD row as payload and no prevPayload', async () => { await seedUser('u1') await seedFile('f1') await client.query(`DELETE FROM "file" WHERE "id" = 'f1'`) const rows = await outboxRows('f1') expect(rows).toHaveLength(2) const deleteRow = rows[1] expect(deleteRow.command).toBe('delete') expect(deleteRow.payload.id).toBe('f1') expect(deleteRow.prevPayload).toBeNull() }) it('deleting a group cascades to a soft-delete update on its files, and the outbox trigger fires for it', async () => { await seedUser('u1') await client.query(`INSERT INTO "group" ("id", "name") VALUES ('g1', 'g1')`) await seedFile('fGroup', { owningGroupId: 'g1' }) // The real cleanup_deleted_group_trigger (023_groups.sql) fires inside this // UPDATE and soft-deletes fGroup, which in turn must fire // file_effect_outbox_after_change for that nested UPDATE. await client.query(`UPDATE "group" SET "isDeleted" = true WHERE "id" = 'g1'`) const rows = await outboxRows('fGroup') expect(rows).toHaveLength(2) const cascadeRow = rows[1] expect(cascadeRow.tableName).toBe('file') expect(cascadeRow.command).toBe('update') expect(cascadeRow.payload.isDeleted).toBe(true) expect(cascadeRow.prevPayload.isDeleted).toBe(false) }) it('new rows start with attempts = 0 and nextRetryAt IS NULL (048 applied)', async () => { await seedUser('u1') await seedFile('f1') const rows = await outboxRows('f1') expect(rows).toHaveLength(1) expect(rows[0].attempts).toBe(0) expect(rows[0].nextRetryAt).toBeNull() }) it('new rows start with lastError IS NULL and a failed attempt can record it (051 applied)', async () => { await seedUser('u1') await seedFile('f1') expect((await outboxRows('f1'))[0].lastError).toBeNull() // The same statement shape the DO's bumpAttempts issues: attempts + 1 and lastError in one // UPDATE. A later attempt overwrites; an admin retry (attempts = 0) leaves it in place. const bump = (error: string) => client.query( `UPDATE effect_outbox SET attempts = attempts + 1, "lastError" = $2 WHERE "entityId" = $1`, ['f1', error] ) await bump('EffectTimeoutError: timed out after 30000ms') await bump('Error: proxy request failed') let [row] = await outboxRows('f1') expect(row.attempts).toBe(2) expect(row.lastError).toBe('Error: proxy request failed') await client.query( `UPDATE effect_outbox SET attempts = 0, "nextRetryAt" = NULL WHERE "entityId" = $1`, ['f1'] ) ;[row] = await outboxRows('f1') expect(row.attempts).toBe(0) expect(row.lastError).toBe('Error: proxy request failed') }) })