feat(tables): trigger workflows on row deletes

Author: j15z

apps/sim/background/cleanup-table-row-ttl.test.ts

@@ -10,13 +10,15 @@ const {
   mockSignalTableRowsChanged,
   mockTask,
   mockWithLockedTable,
+  mockFireTableTrigger,
 } = vi.hoisted(() => ({
   mockDeleteExecute: vi.fn(),
   mockListExecute: vi.fn(),
   mockIsTableRowTtlEnabled: vi.fn(),
   mockSignalTableRowsChanged: vi.fn(),
   mockTask: vi.fn((config: unknown) => config),
   mockWithLockedTable: vi.fn(),
+  mockFireTableTrigger: vi.fn(),
 }))
 
 vi.mock('@sim/db', () => ({
@@ -29,16 +31,25 @@ vi.mock('@/lib/table/service', () => ({ withLockedTable: mockWithLockedTable }))
 vi.mock('@/lib/table/ttl-availability', () => ({
   isTableRowTtlEnabled: mockIsTableRowTtlEnabled,
 }))
+vi.mock('@/lib/table/trigger', () => ({ fireTableTrigger: mockFireTableTrigger }))
 
 import { cleanupTableRowTtlTask, runCleanupTableRowTtl } from '@/background/cleanup-table-row-ttl'
 
 const table = {
   id: 'table-1',
+  name: 'Expiring rows',
   workspaceId: 'workspace-1',
   schema: { columns: [{ id: 'col-ttl', name: 'expires_at', type: 'ttl' }] },
   locks: { insertLocked: false, updateLocked: false, deleteLocked: false, schemaLocked: false },
 }
 
+function deletedRows(count: number, start = 1) {
+  return Array.from({ length: count }, (_, index) => {
+    const number = start + index
+    return { id: `row-${number}`, data: { value: number } }
+  })
+}
+
 describe('table row TTL cleanup', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -57,8 +68,8 @@ describe('table row TTL cleanup', () => {
 
   it('deletes expired rows in locked, keyset batches and signals the table', async () => {
     mockDeleteExecute
-      .mockResolvedValueOnce([{ count: 500, lastId: 'row-500' }])
-      .mockResolvedValueOnce([{ count: 12, lastId: 'row-512' }])
+      .mockResolvedValueOnce(deletedRows(500))
+      .mockResolvedValueOnce(deletedRows(12, 501))
 
     await expect(runCleanupTableRowTtl()).resolves.toEqual({
       batches: 2,
@@ -68,13 +79,24 @@ describe('table row TTL cleanup', () => {
     expect(mockWithLockedTable).toHaveBeenCalledTimes(2)
     expect(mockDeleteExecute).toHaveBeenCalledTimes(2)
     expect(mockSignalTableRowsChanged).toHaveBeenCalledWith(table.id)
+    expect(mockFireTableTrigger).toHaveBeenCalledTimes(2)
+    expect(mockFireTableTrigger).toHaveBeenNthCalledWith(
+      1,
+      table.id,
+      table.name,
+      'delete',
+      deletedRows(500),
+      null,
+      table.schema,
+      'ttl-cleanup'
+    )
   })
 
   it('compares TTL values with whole Date.now epoch seconds', async () => {
     const nowEpochMilliseconds = 1_700_000_000_999
     const nowEpochSeconds = 1_700_000_000
     const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(nowEpochMilliseconds)
-    mockDeleteExecute.mockResolvedValue([{ count: 0, lastId: null }])
+    mockDeleteExecute.mockResolvedValue([])
 
     try {
       await runCleanupTableRowTtl()
@@ -132,7 +154,7 @@ describe('table row TTL cleanup', () => {
   })
 
   it('stops after one hundred full batches', async () => {
-    mockDeleteExecute.mockResolvedValue([{ count: 500, lastId: 'row-cursor' }])
+    mockDeleteExecute.mockResolvedValue(deletedRows(500))
 
     await expect(runCleanupTableRowTtl()).resolves.toEqual({
       batches: 100,
@@ -162,12 +184,12 @@ describe('table row TTL cleanup', () => {
           const attempt = (tableAttempts.get(tableId) ?? 0) + 1
           tableAttempts.set(tableId, attempt)
           if (tableId === table.id && attempt === 1) {
-            return [{ count: 500, lastId: 'row-500' }]
+            return deletedRows(500)
           }
           if (tableId === secondTable.id) {
-            return [{ count: 1, lastId: 'row-1' }]
+            return deletedRows(1)
           }
-          return [{ count: 0, lastId: null }]
+          return []
         }),
       })
     })

apps/sim/background/cleanup-table-row-ttl.ts

@@ -8,8 +8,11 @@ import { getColumnId } from '@/lib/table/column-keys'
 import { signalTableRowsChanged } from '@/lib/table/events'
 import { assertRowDelete, TableLockedError } from '@/lib/table/mutation-locks'
 import type { DbTransaction } from '@/lib/table/planner'
+import type { DeletedTableRow } from '@/lib/table/rows/ordering'
 import { withLockedTable } from '@/lib/table/service'
+import { fireTableTrigger } from '@/lib/table/trigger'
 import { isTableRowTtlEnabled } from '@/lib/table/ttl-availability'
+import type { RowData, TableSchema } from '@/lib/table/types'
 
 const logger = createLogger('CleanupTableRowTtl')
 const cleanupDb = dbFor('cleanup')
@@ -23,12 +26,20 @@ interface ExpiredTtlTableRef {
   workspaceId: string
 }
 
-interface DeletedTtlBatch {
-  attempted: boolean
+interface DeletedTtlRows {
   deleted: number
   lastId: string | null
+  rows: DeletedTableRow[]
 }
 
+type DeletedTtlBatch =
+  | { attempted: false; deleted: 0; lastId: null }
+  | (DeletedTtlRows & {
+      attempted: true
+      tableName: string
+      schema: TableSchema
+    })
+
 interface TtlTableCleanupState {
   ref: ExpiredTtlTableRef
   afterId?: string
@@ -80,20 +91,25 @@ async function listExpiredTtlTables(nowEpochSeconds: number): Promise<ExpiredTtl
   return Array.isArray(rows) ? rows : []
 }
 
-function parseDeletedBatch(rows: unknown): Omit<DeletedTtlBatch, 'attempted'> {
-  const [row] = Array.isArray(rows)
-    ? (rows as Array<{ count?: number | string; lastId?: string | null }>)
-    : []
-  if (!row) throw new Error('Table row TTL cleanup did not return a deleted count')
-
-  const deleted = Number(row.count)
-  if (!Number.isSafeInteger(deleted) || deleted < 0 || deleted > TTL_CLEANUP_BATCH_SIZE) {
+function parseDeletedBatch(rows: unknown): DeletedTtlRows {
+  if (!Array.isArray(rows)) {
+    throw new Error('Table row TTL cleanup did not return deleted rows')
+  }
+  const deletedRows = rows as Array<{ id?: unknown; data?: unknown }>
+  if (deletedRows.length > TTL_CLEANUP_BATCH_SIZE) {
     throw new Error('Table row TTL cleanup returned an invalid deleted count')
   }
-  if (deleted > 0 && typeof row.lastId !== 'string') {
-    throw new Error('Table row TTL cleanup did not return a row cursor')
+  const parsed = deletedRows.map((row) => {
+    if (typeof row.id !== 'string') {
+      throw new Error('Table row TTL cleanup did not return a row cursor')
+    }
+    return { id: row.id, data: row.data as RowData }
+  })
+  return {
+    deleted: parsed.length,
+    lastId: parsed[parsed.length - 1]?.id ?? null,
+    rows: parsed,
   }
-  return { deleted, lastId: row.lastId ?? null }
 }
 
 async function deleteExpiredTableRowBatch(
@@ -103,8 +119,8 @@ async function deleteExpiredTableRowBatch(
   columnKey: string,
   nowEpochSeconds: number,
   afterId?: string
-): Promise<Omit<DeletedTtlBatch, 'attempted'>> {
-  const rows = await trx.execute<{ count: number | string; lastId: string | null }>(sql`
+): Promise<DeletedTtlRows> {
+  const rows = await trx.execute<{ id: string; data: RowData }>(sql`
     WITH candidates AS MATERIALIZED (
       SELECT table_row.id
       FROM ${userTableRows} AS table_row
@@ -120,12 +136,11 @@ async function deleteExpiredTableRowBatch(
       DELETE FROM ${userTableRows} AS table_row
       USING candidates
       WHERE table_row.id = candidates.id
-      RETURNING table_row.id
+      RETURNING table_row.id, table_row.data
     )
-    SELECT
-      count(*)::integer AS count,
-      max(id) AS "lastId"
+    SELECT id, data
     FROM deleted
+    ORDER BY id
   `)
   return parseDeletedBatch(rows)
 }
@@ -136,9 +151,9 @@ async function deleteExpiredRowsForTable(
   afterId?: string
 ): Promise<DeletedTtlBatch> {
   try {
-    return await withLockedTable(
+    const batch = await withLockedTable(
       ref.id,
-      async (table, trx) => {
+      async (table, trx): Promise<DeletedTtlBatch> => {
         try {
           assertRowDelete(table)
         } catch (error) {
@@ -159,10 +174,27 @@ async function deleteExpiredRowsForTable(
           nowEpochSeconds,
           afterId
         )
-        return { attempted: true, ...batch }
+        return {
+          attempted: true,
+          ...batch,
+          tableName: table.name,
+          schema: table.schema,
+        } satisfies DeletedTtlBatch
       },
       { expectedWorkspaceId: ref.workspaceId }
     )
+    if (batch.attempted && batch.rows.length > 0) {
+      void fireTableTrigger(
+        ref.id,
+        batch.tableName,
+        'delete',
+        batch.rows,
+        null,
+        batch.schema,
+        'ttl-cleanup'
+      )
+    }
+    return batch
   } catch (error) {
     if (asOrchestrationError(error)?.code === 'not_found') {
       return { attempted: false, deleted: 0, lastId: null }

apps/sim/lib/table/__tests__/service-filter-threading.test.ts

@@ -15,6 +15,8 @@ import { decodeCursor } from '@/lib/table/rows/cursor'
 import { buildFilterClause, buildSortClause } from '@/lib/table/sql'
 import type { ColumnDefinition, TableDefinition } from '@/lib/table/types'
 
+const { mockFireTableTrigger } = vi.hoisted(() => ({ mockFireTableTrigger: vi.fn() }))
+
 vi.mock('@/lib/table/sql', () => ({
   buildFilterClause: vi.fn(() => sql`true`),
   buildSortClause: vi.fn(() => sql`true`),
@@ -23,7 +25,7 @@ vi.mock('@/lib/table/sql', () => ({
 }))
 
 vi.mock('@/lib/table/trigger', () => ({
-  fireTableTrigger: vi.fn(),
+  fireTableTrigger: mockFireTableTrigger,
 }))
 
 vi.mock('@/lib/table/workflow-group-deps', () => ({
@@ -64,7 +66,9 @@ vi.mock('@/lib/table/validation', () => ({
 }))
 
 import {
+  deleteRow,
   deleteRowsByFilter,
+  deleteRowsByIds,
   queryRows,
   requireTableRowIds,
   updateRowsByFilter,
@@ -183,6 +187,55 @@ describe('service filter threading', () => {
   })
 })
 
+describe('delete trigger dispatch', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('fires with the committed snapshot after deleting one row', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'row-1', data: { name: 'Ada' } }])
+
+    await deleteRow(TABLE, 'row-1', 'req-delete-one')
+
+    expect(mockFireTableTrigger).toHaveBeenCalledWith(
+      TABLE.id,
+      TABLE.name,
+      'delete',
+      [{ id: 'row-1', data: { name: 'Ada' } }],
+      null,
+      TABLE.schema,
+      'req-delete-one'
+    )
+  })
+
+  it('fires once with every committed snapshot in an ID batch', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([
+      { id: 'row-1', data: { name: 'Ada' } },
+      { id: 'row-2', data: { name: 'Grace' } },
+    ])
+
+    await deleteRowsByIds(
+      TABLE,
+      { tableId: TABLE.id, workspaceId: TABLE.workspaceId, rowIds: ['row-1', 'row-2'] },
+      'req-delete-many'
+    )
+
+    expect(mockFireTableTrigger).toHaveBeenCalledWith(
+      TABLE.id,
+      TABLE.name,
+      'delete',
+      [
+        { id: 'row-1', data: { name: 'Ada' } },
+        { id: 'row-2', data: { name: 'Grace' } },
+      ],
+      null,
+      TABLE.schema,
+      'req-delete-many'
+    )
+  })
+})
+
 describe('bulk update/delete limited-subset ordering', () => {
   beforeEach(() => {
     vi.clearAllMocks()

apps/sim/lib/table/__tests__/trigger.test.ts

@@ -69,7 +69,7 @@ function firedPayloads(): Payload[] {
 }
 
 async function fire(
-  eventType: 'insert' | 'update',
+  eventType: 'insert' | 'update' | 'delete',
   data: RowData,
   oldRows: Map<string, RowData> | null = null
 ) {
@@ -166,6 +166,19 @@ describe('fireTableTrigger — payload shape', () => {
     const [payload] = firedPayloads()
     expect(payload.changedColumns).toEqual(['Status'])
   })
+
+  it('emits the deleted row snapshot as the event row and previous row', async () => {
+    mockFetchActiveWebhooks.mockResolvedValue([webhookEntry({ eventType: 'delete' })])
+
+    await fire('delete', { col_title: 'Removed issue', col_status: 'opt_closed' })
+
+    const [payload] = firedPayloads()
+    const deletedRow = { Title: 'Removed issue', Status: 'Closed' }
+    expect(payload.rawRow).toEqual(deletedRow)
+    expect(payload.row).toEqual({ ...deletedRow, Tags: null })
+    expect(payload.previousRow).toEqual(deletedRow)
+    expect(payload.changedColumns).toEqual([])
+  })
 })
 
 describe('fireTableTrigger — gating', () => {