fix(tables): gate multi-batch CSV create + initial rows against the plan, harden limits cache bound

Author: TheodoreSpeaks

apps/sim/app/api/table/import-csv/route.ts

@@ -105,12 +105,18 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
       headerToColumn: Map<string, string>
     }
 
-    const insertRows = async (rows: Record<string, unknown>[], state: ImportState) => {
+    const insertRows = async (
+      rows: Record<string, unknown>[],
+      state: ImportState,
+      currentRowCount: number
+    ) => {
       if (rows.length === 0) return 0
       const coerced = coerceRowsForTable(rows, state.schema, state.headerToColumn)
       const result = await batchInsertRows(
         { tableId: state.table.id, rows: coerced, workspaceId, userId },
-        state.table,
+        // The created table's rowCount is frozen at 0; pass the running total so the
+        // per-batch capacity check sees cumulative rows, not an always-empty table.
+        { ...state.table, rowCount: currentRowCount },
         generateId().slice(0, 8)
       )
       return result.length
@@ -152,13 +158,13 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
           sample.push(record)
           if (sample.length >= CSV_SCHEMA_SAMPLE_SIZE) {
             state = await buildTable(sample)
-            inserted += await insertRows(sample, state)
+            inserted += await insertRows(sample, state, inserted)
           }
           continue
         }
         batch.push(record)
         if (batch.length >= CSV_MAX_BATCH_SIZE) {
-          inserted += await insertRows(batch, state)
+          inserted += await insertRows(batch, state, inserted)
           batch = []
         }
       }
@@ -168,9 +174,9 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
           return NextResponse.json({ error: 'CSV file has no data rows' }, { status: 400 })
         }
         state = await buildTable(sample)
-        inserted += await insertRows(sample, state)
+        inserted += await insertRows(sample, state, inserted)
       } else {
-        inserted += await insertRows(batch, state)
+        inserted += await insertRows(batch, state, inserted)
       }
     } catch (streamError) {
       if (state) await deleteTable(state.table.id, requestId).catch(() => {})

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

@@ -80,6 +80,18 @@ describe('getWorkspaceTableLimits', () => {
     // The fallback is never cached, so the next call re-attempts and resolves the real plan.
     expect(await getWorkspaceTableLimits(ws)).toEqual(LIMITS.pro)
   })
+
+  it('stays bounded under a burst of distinct all-fresh workspaces', async () => {
+    // Far more distinct workspaces than the cap, all within one TTL window. The Map
+    // must not grow without limit; eviction keeps it at/under the ceiling.
+    for (let i = 0; i < 6_000; i++) {
+      await getWorkspaceTableLimits(`burst-${i}`)
+    }
+    // Re-resolving an early (evicted) workspace must re-hit the billing lookup.
+    mockGetWorkspaceBilledAccountUserId.mockClear()
+    await getWorkspaceTableLimits('burst-0')
+    expect(mockGetWorkspaceBilledAccountUserId).toHaveBeenCalledTimes(1)
+  })
 })
 
 describe('getMaxRowsPerTable', () => {

apps/sim/lib/table/billing.ts

@@ -74,13 +74,19 @@ export async function getWorkspaceTableLimits(workspaceId: string): Promise<Tabl
 }
 
 function cacheLimits(workspaceId: string, limits: TablePlanLimits): void {
-  // Sweep expired entries before growing past the cap so workspaces queried once
-  // (never re-accessed, so never deleted on read) can't accumulate forever.
-  if (limitsCache.size >= LIMITS_CACHE_MAX_ENTRIES) {
+  // Keep the Map bounded for a new key: sweep expired entries, then (if a burst of
+  // all-fresh entries still sits at the cap) evict oldest-inserted ones. Map iteration
+  // is insertion order, so the first key is the oldest. Net: size never exceeds the cap.
+  if (limitsCache.size >= LIMITS_CACHE_MAX_ENTRIES && !limitsCache.has(workspaceId)) {
     const now = Date.now()
     for (const [key, entry] of limitsCache) {
       if (entry.expiresAt <= now) limitsCache.delete(key)
     }
+    while (limitsCache.size >= LIMITS_CACHE_MAX_ENTRIES) {
+      const oldest = limitsCache.keys().next().value
+      if (oldest === undefined) break
+      limitsCache.delete(oldest)
+    }
   }
   limitsCache.set(workspaceId, { limits, expiresAt: Date.now() + LIMITS_CACHE_TTL_MS })
 }

apps/sim/lib/table/service.ts

@@ -15,6 +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 { 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'
@@ -281,6 +282,17 @@ export async function createTable(
       ? { id: data.jobId, type: data.jobType ?? 'import', startedAt: now }
       : null
 
+  // 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
+  if (initialRowCount > 0) {
+    await assertRowCapacity({
+      workspaceId: data.workspaceId,
+      currentRowCount: 0,
+      addedRows: initialRowCount,
+    })
+  }
+
   // Wrap count check, duplicate check, and insert in a transaction with FOR UPDATE
   // to prevent TOCTOU race on the table count limit
   try {
@@ -332,7 +344,6 @@ export async function createTable(
         })
       }
 
-      const initialRowCount = data.initialRowCount ?? 0
       if (initialRowCount > 0) {
         const orderKeys = nKeysBetween(null, null, initialRowCount)
         const rowsToInsert = Array.from({ length: initialRowCount }, (_, i) => ({