fix(billing): notify table limits post-commit so a rolled-back insert never emails or burns the claim

Author: waleedlatif1

apps/sim/lib/table/__tests__/update-row.test.ts

@@ -19,7 +19,8 @@ vi.mock('@sim/db', () => dbChainMock)
 // Capacity is exercised in billing.test.ts; here it's a no-op so the timeout-scaling
 // suites can use large synthetic row counts without tripping the plan limit.
 vi.mock('@/lib/table/billing', () => ({
-  assertRowCapacity: vi.fn().mockResolvedValue(undefined),
+  assertRowCapacity: vi.fn().mockResolvedValue(1_000_000),
+  notifyTableRowUsage: vi.fn(),
   getMaxRowsPerTable: vi.fn().mockResolvedValue(1_000_000),
   wouldExceedRowLimit: () => false,
   TableRowLimitError: class TableRowLimitError extends Error {},

apps/sim/lib/table/billing.test.ts

@@ -120,16 +120,16 @@ describe('wouldExceedRowLimit', () => {
 })
 
 describe('assertRowCapacity', () => {
-  it('passes when the write stays under the plan limit', async () => {
+  it('returns the resolved limit when the write stays under it', async () => {
     await expect(
       assertRowCapacity({ workspaceId: nextWorkspaceId(), currentRowCount: 10, addedRows: 5 })
-    ).resolves.toBeUndefined()
+    ).resolves.toBe(5000)
   })
 
-  it('allows reaching the limit exactly', async () => {
+  it('allows reaching the limit exactly and returns it', async () => {
     await expect(
       assertRowCapacity({ workspaceId: nextWorkspaceId(), currentRowCount: 4999, addedRows: 1 })
-    ).resolves.toBeUndefined()
+    ).resolves.toBe(5000)
   })
 
   it('throws TableRowLimitError when the write would exceed the limit', async () => {
@@ -155,6 +155,6 @@ describe('assertRowCapacity', () => {
         currentRowCount: 10_000_000,
         addedRows: 1,
       })
-    ).resolves.toBeUndefined()
+    ).resolves.toBe(-1)
   })
 })

apps/sim/lib/table/billing.ts

@@ -191,25 +191,23 @@ export function wouldExceedRowLimit(
  * connection (and locks) risks pool starvation. Callers already inside a tx should
  * fetch the limit up front and use {@link wouldExceedRowLimit} instead.
  *
+ * Pure check (no side effects): returns the resolved limit so callers can fire
+ * {@link notifyTableRowUsage} AFTER their insert commits — a pre-commit notify
+ * would email (and burn the dedup claim) for a write that later rolls back.
+ *
+ * @returns the resolved plan row limit (-1 for unlimited)
  * @throws {TableRowLimitError} if `currentRowCount + addedRows` exceeds the limit
  */
 export async function assertRowCapacity(params: {
   workspaceId: string
   currentRowCount: number
   addedRows: number
-}): Promise<void> {
+}): Promise<number> {
   const limit = await getMaxRowsPerTable(params.workspaceId)
   if (wouldExceedRowLimit(limit, params.currentRowCount, params.addedRows)) {
     throw new TableRowLimitError(limit)
   }
-
-  // Accepted write: warn (best-effort) once the table crosses the notify band.
-  notifyTableRowUsage({
-    workspaceId: params.workspaceId,
-    currentRowCount: params.currentRowCount,
-    addedRows: params.addedRows,
-    limit,
-  })
+  return limit
 }
 
 /**

apps/sim/lib/table/import-data.ts

@@ -9,7 +9,7 @@ import { userTableDefinitions, userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { generateId } from '@sim/utils/id'
 import { eq } from 'drizzle-orm'
-import { assertRowCapacity } from '@/lib/table/billing'
+import { assertRowCapacity, notifyTableRowUsage } from '@/lib/table/billing'
 import { CSV_MAX_BATCH_SIZE } from '@/lib/table/import'
 import { nKeysBetween } from '@/lib/table/order-key'
 import { acquireRowOrderLock } from '@/lib/table/rows/ordering'
@@ -150,12 +150,12 @@ export async function importAppendRows(
   ctx: { workspaceId: string; userId?: string; requestId: string }
 ): Promise<{ inserted: TableRow[]; table: TableDefinition }> {
   // Gate capacity before opening the tx — the lookup is a separate pool read.
-  await assertRowCapacity({
+  const rowLimit = await assertRowCapacity({
     workspaceId: ctx.workspaceId,
     currentRowCount: table.rowCount,
     addedRows: rows.length,
   })
-  return db.transaction(async (trx) => {
+  const result = await db.transaction(async (trx) => {
     let working = table
     if (additions.length > 0) {
       // Take the row-order lock before creating columns so this path uses the
@@ -179,6 +179,14 @@ export async function importAppendRows(
     }
     return { inserted, table: working }
   })
+  // Post-commit: a pre-commit notify would email/burn the claim for a rolled-back import.
+  notifyTableRowUsage({
+    workspaceId: ctx.workspaceId,
+    currentRowCount: table.rowCount,
+    addedRows: result.inserted.length,
+    limit: rowLimit,
+  })
+  return result
 }
 
 /**
@@ -193,12 +201,12 @@ export async function importReplaceRows(
 ): Promise<ReplaceRowsResult> {
   // Replace deletes all existing rows, so the footprint is just the new set. Gate
   // before opening the tx — the plan lookup is a separate pool read.
-  await assertRowCapacity({
+  const rowLimit = await assertRowCapacity({
     workspaceId: data.workspaceId,
     currentRowCount: 0,
     addedRows: data.rows.length,
   })
-  return db.transaction(async (trx) => {
+  const result = await db.transaction(async (trx) => {
     let working = table
     if (additions.length > 0) {
       await acquireRowOrderLock(trx, table.id)
@@ -211,4 +219,12 @@ export async function importReplaceRows(
       requestId
     )
   })
+  // Post-commit: footprint is the new set (prior rows deleted → prior count 0).
+  notifyTableRowUsage({
+    workspaceId: data.workspaceId,
+    currentRowCount: 0,
+    addedRows: result.insertedCount,
+    limit: rowLimit,
+  })
+  return result
 }

apps/sim/lib/table/import-runner.ts

@@ -17,7 +17,7 @@ import {
   type TableSchema,
   validateMapping,
 } from '@/lib/table'
-import { assertRowCapacity } from '@/lib/table/billing'
+import { assertRowCapacity, notifyTableRowUsage } from '@/lib/table/billing'
 import { withGeneratedColumnIds } from '@/lib/table/column-keys'
 import { appendTableEvent } from '@/lib/table/events'
 import {
@@ -200,7 +200,7 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
       const owns = await updateJobProgress(tableId, inserted, importId)
       if (!owns) throw new ImportSupersededError()
       const coerced = coerceRowsForTable(rows, schema, headerToColumn)
-      await assertRowCapacity({
+      const rowLimit = await assertRowCapacity({
         workspaceId,
         currentRowCount: existingRowCount + inserted,
         addedRows: coerced.length,
@@ -217,6 +217,13 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
         { ...table, schema },
         requestId
       )
+      // Post-commit (per batch): pre-batch count as prior, actual inserted as the delta.
+      notifyTableRowUsage({
+        workspaceId,
+        currentRowCount: existingRowCount + inserted,
+        addedRows: result.inserted,
+        limit: rowLimit,
+      })
       inserted += result.inserted
       lastOrderKey = result.lastOrderKey
       // Emit after the first batch, then every interval, so the bar appears early without flooding.

apps/sim/lib/table/rows/service.ts

@@ -123,7 +123,7 @@ export async function insertRow(
   }
 
   // Best-effort capacity check against the workspace's current plan limit.
-  await assertRowCapacity({
+  const rowLimit = await assertRowCapacity({
     workspaceId: table.workspaceId,
     currentRowCount: table.rowCount,
     addedRows: 1,
@@ -144,6 +144,14 @@ export async function insertRow(
     now,
   })
 
+  // Post-commit: a pre-commit notify would email/burn the dedup claim for a rolled-back insert.
+  notifyTableRowUsage({
+    workspaceId: table.workspaceId,
+    currentRowCount: table.rowCount,
+    addedRows: 1,
+    limit: rowLimit,
+  })
+
   logger.info(`[${requestId}] Inserted row ${rowId} into table ${data.tableId}`)
 
   const insertedRow: TableRow = {
@@ -194,13 +202,20 @@ export async function batchInsertRows(
 ): Promise<TableRow[]> {
   // Best-effort capacity check against the workspace's current plan limit. Import
   // paths call `batchInsertRowsWithTx` directly and gate capacity up front instead.
-  await assertRowCapacity({
+  const rowLimit = await assertRowCapacity({
     workspaceId: table.workspaceId,
     currentRowCount: table.rowCount,
     addedRows: data.rows.length,
   })
 
   const result = await db.transaction((trx) => batchInsertRowsWithTx(trx, data, table, requestId))
+  // Post-commit: notify with the actual inserted count, so a rolled-back batch never emails.
+  notifyTableRowUsage({
+    workspaceId: table.workspaceId,
+    currentRowCount: table.rowCount,
+    addedRows: result.length,
+    limit: rowLimit,
+  })
   dispatchAfterBatchInsert(table, result, requestId, data.userId)
   return result
 }
@@ -349,12 +364,20 @@ export async function replaceTableRows(
 ): Promise<ReplaceRowsResult> {
   // All existing rows are deleted, so the footprint is just the new set. Checked
   // before the tx opens — never inside it (the plan lookup is a separate pool read).
-  await assertRowCapacity({
+  const rowLimit = await assertRowCapacity({
     workspaceId: table.workspaceId,
     currentRowCount: 0,
     addedRows: data.rows.length,
   })
-  return db.transaction((trx) => replaceTableRowsWithTx(trx, data, table, requestId))
+  const result = await db.transaction((trx) => replaceTableRowsWithTx(trx, data, table, requestId))
+  // Post-commit: footprint is the new set (prior rows deleted → prior count 0).
+  notifyTableRowUsage({
+    workspaceId: table.workspaceId,
+    currentRowCount: 0,
+    addedRows: result.insertedCount,
+    limit: rowLimit,
+  })
+  return result
 }
 
 /**

apps/sim/lib/table/service.ts

@@ -15,7 +15,7 @@ import { generateId } from '@sim/utils/id'
 import { and, count, eq, isNull, sql } from 'drizzle-orm'
 import { generateRestoreName } from '@/lib/core/utils/restore-name'
 import type { DbOrTx } from '@/lib/db/types'
-import { assertRowCapacity } from '@/lib/table/billing'
+import { assertRowCapacity, notifyTableRowUsage } from '@/lib/table/billing'
 import { generateColumnId, getColumnId, withGeneratedColumnIds } from '@/lib/table/column-keys'
 import { COLUMN_TYPES, NAME_PATTERN, TABLE_LIMITS } from '@/lib/table/constants'
 import { EMPTY_JOB_FIELDS, latestJobForTable, latestJobsForTables } from '@/lib/table/jobs/service'
@@ -285,8 +285,9 @@ export async function createTable(
   // Starter rows count against the plan too. Checked before the tx (the lookup is a
   // separate pool read) — a new table starts empty, so the footprint is just these.
   const initialRowCount = data.initialRowCount ?? 0
+  let rowLimit: number | undefined
   if (initialRowCount > 0) {
-    await assertRowCapacity({
+    rowLimit = await assertRowCapacity({
       workspaceId: data.workspaceId,
       currentRowCount: 0,
       addedRows: initialRowCount,
@@ -369,6 +370,16 @@ export async function createTable(
     throw error
   }
 
+  // Post-commit: starter rows landed, so notify here (a rolled-back create never emails).
+  if (initialRowCount > 0 && rowLimit !== undefined) {
+    notifyTableRowUsage({
+      workspaceId: data.workspaceId,
+      currentRowCount: 0,
+      addedRows: initialRowCount,
+      limit: rowLimit,
+    })
+  }
+
   logger.info(`[${requestId}] Created table ${tableId} in workspace ${data.workspaceId}`)
 
   return {