feat(tables): byte-based import progress, cancel support, and a start toast that opens the import view

Author: TheodoreSpeaks

apps/sim/app/api/table/[tableId]/import/cancel/route.test.ts

@@ -0,0 +1,110 @@
+/**
+ * @vitest-environment node
+ */
+import { hybridAuthMockFns } from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import type { TableDefinition } from '@/lib/table'
+
+const { mockCheckAccess, mockMarkImportCanceled, mockAppendTableEvent } = vi.hoisted(() => ({
+  mockCheckAccess: vi.fn(),
+  mockMarkImportCanceled: vi.fn(),
+  mockAppendTableEvent: vi.fn(),
+}))
+
+vi.mock('@/lib/table/service', () => ({ markImportCanceled: mockMarkImportCanceled }))
+vi.mock('@/lib/table/events', () => ({ appendTableEvent: mockAppendTableEvent }))
+vi.mock('@/app/api/table/utils', async () => {
+  const { NextResponse } = await import('next/server')
+  return {
+    checkAccess: mockCheckAccess,
+    accessError: (result: { status: number }) =>
+      NextResponse.json({ error: 'denied' }, { status: result.status }),
+  }
+})
+
+import { POST } from '@/app/api/table/[tableId]/import/cancel/route'
+
+function buildTable(overrides: Partial<TableDefinition> = {}): TableDefinition {
+  return {
+    id: 'tbl_1',
+    name: 'People',
+    description: null,
+    schema: { columns: [{ name: 'name', type: 'string' }] },
+    metadata: null,
+    rowCount: 0,
+    maxRows: 1_000_000,
+    workspaceId: 'workspace-1',
+    createdBy: 'user-1',
+    archivedAt: null,
+    createdAt: new Date(),
+    updatedAt: new Date(),
+    ...overrides,
+  }
+}
+
+function makeRequest(body: unknown, tableId = 'tbl_1') {
+  const req = new NextRequest(`http://localhost:3000/api/table/${tableId}/import/cancel`, {
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    body: JSON.stringify(body),
+  })
+  return POST(req, { params: Promise.resolve({ tableId }) })
+}
+
+const validBody = { workspaceId: 'workspace-1', importId: 'import-id-xyz' }
+
+describe('POST /api/table/[tableId]/import/cancel', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({
+      success: true,
+      userId: 'user-1',
+      authType: 'session',
+    })
+    mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable() })
+    mockMarkImportCanceled.mockResolvedValue(true)
+  })
+
+  it('cancels the import and emits a canceled event', async () => {
+    const response = await makeRequest(validBody)
+    const data = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(data.data).toEqual({ canceled: true })
+    expect(mockMarkImportCanceled).toHaveBeenCalledWith('tbl_1', 'import-id-xyz')
+    expect(mockAppendTableEvent).toHaveBeenCalledWith(
+      expect.objectContaining({ kind: 'import', status: 'canceled', importId: 'import-id-xyz' })
+    )
+  })
+
+  it('does not emit an event when nothing was importing', async () => {
+    mockMarkImportCanceled.mockResolvedValue(false)
+    const response = await makeRequest(validBody)
+    const data = await response.json()
+
+    expect(response.status).toBe(200)
+    expect(data.data).toEqual({ canceled: false })
+    expect(mockAppendTableEvent).not.toHaveBeenCalled()
+  })
+
+  it('returns 401 when unauthenticated', async () => {
+    hybridAuthMockFns.mockCheckSessionOrInternalAuth.mockResolvedValue({ success: false })
+    const response = await makeRequest(validBody)
+    expect(response.status).toBe(401)
+    expect(mockMarkImportCanceled).not.toHaveBeenCalled()
+  })
+
+  it('returns the access error status when access is denied', async () => {
+    mockCheckAccess.mockResolvedValue({ ok: false, status: 403 })
+    const response = await makeRequest(validBody)
+    expect(response.status).toBe(403)
+  })
+
+  it('returns 400 on workspace mismatch', async () => {
+    mockCheckAccess.mockResolvedValue({ ok: true, table: buildTable({ workspaceId: 'other-ws' }) })
+    const response = await makeRequest(validBody)
+    expect(response.status).toBe(400)
+    expect(mockMarkImportCanceled).not.toHaveBeenCalled()
+  })
+})

apps/sim/app/api/table/[tableId]/import/cancel/route.ts

@@ -0,0 +1,54 @@
+import { createLogger } from '@sim/logger'
+import { type NextRequest, NextResponse } from 'next/server'
+import { cancelTableImportContract } from '@/lib/api/contracts/tables'
+import { parseRequest } from '@/lib/api/server'
+import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
+import { generateRequestId } from '@/lib/core/utils/request'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { appendTableEvent } from '@/lib/table/events'
+import { markImportCanceled } from '@/lib/table/service'
+import { accessError, checkAccess } from '@/app/api/table/utils'
+
+const logger = createLogger('TableImportCancelAPI')
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+interface RouteParams {
+  params: Promise<{ tableId: string }>
+}
+
+/**
+ * POST /api/table/[tableId]/import/cancel
+ *
+ * Cancels an in-flight async CSV import. Flips the table's import status to `canceled`, which makes
+ * the detached worker's next ownership check fail so it stops inserting. Committed rows are left in
+ * place (no rollback) — the user can delete the table. No-op if the import already finished.
+ */
+export const POST = withRouteHandler(async (request: NextRequest, { params }: RouteParams) => {
+  const requestId = generateRequestId()
+
+  const authResult = await checkSessionOrInternalAuth(request, { requireWorkflowId: false })
+  if (!authResult.success || !authResult.userId) {
+    return NextResponse.json({ error: 'Authentication required' }, { status: 401 })
+  }
+
+  const parsed = await parseRequest(cancelTableImportContract, request, { params })
+  if (!parsed.success) return parsed.response
+  const { tableId } = parsed.data.params
+  const { workspaceId, importId } = parsed.data.body
+
+  const access = await checkAccess(tableId, authResult.userId, 'write')
+  if (!access.ok) return accessError(access, requestId, tableId)
+  if (access.table.workspaceId !== workspaceId) {
+    return NextResponse.json({ error: 'Invalid workspace ID' }, { status: 400 })
+  }
+
+  const canceled = await markImportCanceled(tableId, importId)
+  if (canceled) {
+    void appendTableEvent({ kind: 'import', tableId, importId, status: 'canceled' })
+  }
+  logger.info(`[${requestId}] Import cancel requested`, { tableId, importId, canceled })
+
+  return NextResponse.json({ success: true, data: { canceled } })
+})

apps/sim/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-event-stream.ts

@@ -241,7 +241,7 @@ export function useTableEventStream({
       // Live-fill: rows are real as each batch commits. Coalesce the per-tick row
       // refetches via a debounce; on the terminal event refetch rows + the
       // definition immediately (the worker may have rewritten the schema).
-      if (status === 'ready' || status === 'failed') {
+      if (status === 'ready' || status === 'failed' || status === 'canceled') {
         if (importInvalidateTimer !== null) clearTimeout(importInvalidateTimer)
         void queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(tableId) })
         void queryClient.invalidateQueries({ queryKey: tableKeys.detail(tableId) })

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

@@ -328,6 +328,13 @@ export function ImportCsvDialog({
         rowsProcessed: 0,
       })
       onOpenChange(false)
+      toast({
+        message: `Importing "${parsed.file.name}" into "${table.name}"…`,
+        action: {
+          label: 'View',
+          onClick: () => useImportTrayStore.getState().setMenuOpen(true),
+        },
+      })
       importAsyncMutation.mutate(
         {
           workspaceId,
@@ -342,7 +349,7 @@ export function ImportCsvDialog({
               workspaceId,
               title: table.name,
               phase: 'importing',
-              uploadPercent: percent,
+              percent,
             }),
         },
         {

apps/sim/app/workspace/[workspaceId]/tables/components/import-progress-menu/import-progress-menu.tsx

@@ -9,6 +9,7 @@ import {
   ProgressItem,
 } from '@/components/emcn'
 import { Upload } from '@/components/emcn/icons'
+import { cancelTableImport } from '@/hooks/queries/tables'
 import { selectWorkspaceImports, useImportTrayStore } from '@/stores/table/import-tray/store'
 import { getImportStage } from './import-stage'
 import { useHydrateImportTray } from './use-hydrate-import-tray'
@@ -38,6 +39,8 @@ export function ImportProgressMenu({ workspaceId, tableId }: ImportProgressMenuP
     useShallow((state) => selectWorkspaceImports(state, workspaceId))
   )
   const dismiss = useImportTrayStore((state) => state.dismiss)
+  const menuOpen = useImportTrayStore((state) => state.menuOpen)
+  const setMenuOpen = useImportTrayStore((state) => state.setMenuOpen)
 
   // Inside a table, scope the indicator to that table's import only; on the list view show
   // every active import in the workspace.
@@ -48,8 +51,16 @@ export function ImportProgressMenu({ workspaceId, tableId }: ImportProgressMenuP
   const total = imports.length
   const done = imports.filter((e) => e.phase === 'ready').length
 
+  const cancel = (entry: (typeof imports)[number]) => {
+    // Optimistically clear it; the server flips status → the SSE `canceled` event also dismisses.
+    dismiss(entry.tableId)
+    if (entry.importId) {
+      void cancelTableImport(entry.workspaceId, entry.tableId, entry.importId).catch(() => {})
+    }
+  }
+
   return (
-    <DropdownMenu>
+    <DropdownMenu open={menuOpen} onOpenChange={setMenuOpen}>
       <DropdownMenuTrigger asChild>
         <Button variant='subtle' className='px-2 py-1 text-caption'>
           <Upload className='mr-1.5 size-[14px] text-[var(--text-icon)]' />
@@ -68,6 +79,7 @@ export function ImportProgressMenu({ workspaceId, tableId }: ImportProgressMenuP
               title={stage.title}
               meta={stage.meta}
               detail={stage.detail}
+              onCancel={entry.phase === 'importing' ? () => cancel(entry) : undefined}
               onDismiss={stage.dismissible ? () => dismiss(entry.tableId) : undefined}
             />
           )

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

@@ -9,20 +9,22 @@ export interface ImportStageView {
   title: string
   /** Right-aligned on the title row: the percent (when known). */
   meta?: string
-  /** Secondary line: the row progress, or the error message on failure. */
+  /** Secondary line: the row count, or the error message on failure. */
   detail?: string
   dismissible: boolean
 }
 
 /**
  * Maps a tray entry to the stage shown in the import dropdown. The single place the import
  * stages (Uploading → Processing → Imported / Failed) are defined; the row component just
- * renders the returned slots, so every stage looks consistent: `{status} {name}` with the
- * percent on the right and the row count underneath.
+ * renders the returned slots, so every stage looks consistent: `{status} {name}` with a
+ * byte-based percent on the right and the row count underneath. The percent comes straight from
+ * `entry.percent` (exact, monotonic) rather than an estimated row fraction.
  */
 export function getImportStage(entry: ImportTrayEntry): ImportStageView {
   const rows = entry.rowsProcessed.toLocaleString()
   const name = entry.title
+  const meta = typeof entry.percent === 'number' ? `${entry.percent}%` : undefined
 
   if (entry.phase === 'failed') {
     return {
@@ -42,22 +44,15 @@ export function getImportStage(entry: ImportTrayEntry): ImportStageView {
     }
   }
 
-  // importing: processing once the worker reports rows/total, otherwise still uploading.
-  if (entry.total && entry.total > 0) {
-    const percent = Math.min(99, Math.round((entry.rowsProcessed / entry.total) * 100))
+  // importing: rows only start arriving once the worker is processing; before that it's the upload.
+  if (entry.rowsProcessed > 0) {
     return {
       status: 'pending',
       title: `Processing ${name}`,
-      meta: `${percent}%`,
-      detail: `${rows} / ${entry.total.toLocaleString()} rows`,
+      meta,
+      detail: `${rows} rows`,
       dismissible: false,
     }
   }
-
-  return {
-    status: 'pending',
-    title: `Uploading ${name}`,
-    meta: typeof entry.uploadPercent === 'number' ? `${entry.uploadPercent}%` : undefined,
-    dismissible: false,
-  }
+  return { status: 'pending', title: `Uploading ${name}`, meta, dismissible: false }
 }

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

@@ -44,8 +44,8 @@ export function useHydrateImportTray(workspaceId: string | undefined): void {
         })
       } else if (tray.entries[table.id]?.phase === 'importing') {
         // A tracked import finished while we weren't watching (missed SSE terminal event).
-        // `ready` → clear the spinner; `failed` → surface the failure instead of spinning forever.
-        if (table.importStatus === 'ready') {
+        // `ready`/`canceled` → clear the spinner; `failed` → surface the failure.
+        if (table.importStatus === 'ready' || table.importStatus === 'canceled') {
           tray.dismiss(table.id)
         } else if (table.importStatus === 'failed') {
           tray.upsert({

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

@@ -48,11 +48,18 @@ export function useImportProgressTracker(): void {
           // 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 isTerminal =
+            event.status === 'ready' || event.status === 'failed' || event.status === 'canceled'
+          if (!lockedId && isTerminal) return
 
           const importId = lockedId ?? event.importId
           const title = existing?.title ?? 'table'
           const rows = event.progress ?? existing?.rowsProcessed ?? 0
+          if (event.status === 'canceled') {
+            // The user stopped it — just clear the tray entry (no toast, they initiated it).
+            tray.dismiss(tableId)
+            return
+          }
           if (event.status === 'ready') {
             toast.success(`Imported ${rows.toLocaleString()} rows into "${title}"`)
             // Keep it briefly so the count reads `1/1`, then clear (if still ready).
@@ -80,7 +87,7 @@ export function useImportProgressTracker(): void {
             importId,
             phase: event.status,
             rowsProcessed: rows,
-            total: event.total,
+            percent: event.percent,
             error: event.error ?? undefined,
           })
         } catch (err) {

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

@@ -420,6 +420,13 @@ export function Tables() {
                 phase: 'importing',
                 rowsProcessed: 0,
               })
+              toast({
+                message: `Importing "${file.name}"…`,
+                action: {
+                  label: 'View',
+                  onClick: () => useImportTrayStore.getState().setMenuOpen(true),
+                },
+              })
               try {
                 const result = await importCsvAsync.mutateAsync({
                   workspaceId,
@@ -430,7 +437,7 @@ export function Tables() {
                       workspaceId,
                       title: file.name,
                       phase: 'importing',
-                      uploadPercent: percent,
+                      percent,
                     }),
                 })
                 useImportTrayStore.getState().dismiss(pendingId)