fix(tables): validate replace before deleting rows; ignore stale replayed import events by importId

Author: TheodoreSpeaks

apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx

@@ -346,6 +346,16 @@ export function ImportCsvDialog({
             }),
         },
         {
+          onSuccess: (data) => {
+            // Record the import id so the tracker can ignore replayed events from a prior import.
+            useImportTrayStore.getState().upsert({
+              tableId: table.id,
+              workspaceId,
+              title: table.name,
+              importId: data?.importId,
+              phase: 'importing',
+            })
+          },
           onError: (err) => {
             useImportTrayStore.getState().dismiss(table.id)
             toast.error(getErrorMessage(err, 'Failed to start import'))

apps/sim/app/workspace/[workspaceId]/tables/components/import-progress-menu/use-hydrate-import-tray.ts

@@ -37,6 +37,7 @@ export function useHydrateImportTray(workspaceId: string | undefined): void {
           tableId: table.id,
           workspaceId,
           title: table.name,
+          importId: table.importId ?? undefined,
           phase: 'importing',
           rowsProcessed: table.importRowsProcessed ?? 0,
           error: table.importError ?? undefined,

apps/sim/app/workspace/[workspaceId]/tables/components/import-progress-menu/use-import-progress-tracker.ts

@@ -43,8 +43,15 @@ export function useImportProgressTracker(): void {
           if (event?.kind !== 'import') return
           const tray = useImportTrayStore.getState()
           const existing = tray.entries[tableId]
-          const title = existing?.title ?? 'table'
+          // The stream replays from the start, so the buffer can hold a *prior* import's events
+          // for this table. Once we know this run's importId, ignore anything that doesn't match;
+          // before we know it (brief optimistic window), don't trust a replayed terminal event.
+          const lockedId = existing?.importId
+          if (lockedId && event.importId !== lockedId) return
+          if (!lockedId && (event.status === 'ready' || event.status === 'failed')) return
 
+          const importId = lockedId ?? event.importId
+          const title = existing?.title ?? 'table'
           const rows = event.progress ?? existing?.rowsProcessed ?? 0
           if (event.status === 'ready') {
             toast.success(`Imported ${rows.toLocaleString()} rows into "${title}"`)
@@ -53,6 +60,7 @@ export function useImportProgressTracker(): void {
               tableId,
               workspaceId: existing?.workspaceId ?? '',
               title,
+              importId,
               phase: 'ready',
             })
             setTimeout(() => {
@@ -69,6 +77,7 @@ export function useImportProgressTracker(): void {
             tableId,
             workspaceId: existing?.workspaceId ?? '',
             title,
+            importId,
             phase: event.status,
             rowsProcessed: rows,
             total: event.total,

apps/sim/app/workspace/[workspaceId]/tables/tables.tsx

@@ -439,6 +439,7 @@ export function Tables() {
                     tableId: result.tableId,
                     workspaceId,
                     title: file.name,
+                    importId: result.importId,
                     phase: 'importing',
                     rowsProcessed: 0,
                   })

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

@@ -78,10 +78,6 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
     // Stream the file rather than buffering it — a ~1M-row import must never be held in memory.
     const source = await downloadFileStream({ key: fileKey, context: 'workspace' })
 
-    // Delete only after the stream opens (a missing object rejects above) — otherwise a failed
-    // download would wipe the table with nothing to replace it with.
-    if (mode === 'replace') await deleteAllTableRows(tableId)
-
     // Append must continue after the existing rows; create/replace start empty. Read once up
     // front (the import is the table's sole writer) and assign contiguous positions from it.
     const basePosition = mode === 'append' ? await nextImportStartPosition(tableId) : 0
@@ -161,6 +157,11 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
       })
       schema = targetSchema
       headerToColumn = validation.effectiveMap
+
+      // Replace deletes existing rows only after schema/mapping validation passes, so an
+      // invalid or empty file fails the import with the old rows still intact (a mid-stream
+      // insert failure after this point leaves a partial replace — replace is destructive).
+      if (mode === 'replace') await deleteAllTableRows(tableId)
     }
 
     const flush = async (rows: Record<string, unknown>[]) => {

apps/sim/stores/table/import-tray/store.ts

@@ -13,6 +13,9 @@ export interface ImportTrayEntry {
   workspaceId: string
   /** Table name when known, otherwise the source file name. */
   title: string
+  /** Identifies this specific import run, so replayed SSE events from a prior import of the
+   *  same table can be ignored. Known from the kickoff result / the table's `importId`. */
+  importId?: string
   phase: ImportPhase
   rowsProcessed: number
   /** Estimated total rows for a determinate bar; absent until the first progress tick. */
@@ -59,6 +62,7 @@ export const useImportTrayStore = create<ImportTrayState>()(
             tableId: entry.tableId,
             workspaceId: entry.workspaceId,
             title: entry.title || prev?.title || 'table',
+            importId: entry.importId ?? prev?.importId,
             phase: entry.phase ?? prev?.phase ?? 'importing',
             rowsProcessed: entry.rowsProcessed ?? prev?.rowsProcessed ?? 0,
             total: entry.total ?? prev?.total,