Function runTableImport

apps/sim/lib/table/import-runner.ts:81–385

(payload: TableImportPayload)

 * fails. Progress and the terminal state are surfaced via the table-events SSE stream.
 */
export async function runTableImport(payload: TableImportPayload): Promise<void> {
  const { importId, tableId, workspaceId, userId, fileKey, fileName, delimiter, mode } = payload
  const requestId = generateId().slice(0, 8)
  // Hoisted so `finally` can destroy it on any failure — otherwise the storage HTTP body leaks
  // open until it times out.
  let source: Readable | undefined

  try {
    const loaded = await getTableById(tableId, { includeArchived: true })
    if (!loaded) throw new Error(`Import target table ${tableId} not found`)
    const table = loaded

    // Total byte size for the progress estimate — a cheap HEAD, no download. May be null on
    // the local dev provider, in which case the bar stays indeterminate (rows still show).
    const totalBytes = (await headObject(fileKey, 'workspace'))?.size ?? 0

    // Stream the file rather than buffering it — a ~1M-row import must never be held in memory.
    source = await downloadFileStream({ key: fileKey, context: 'workspace' })

    // 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 / threaded
    // order keys from it.
    const basePosition = mode === 'append' ? await nextImportStartPosition(tableId) : 0
    let lastOrderKey = mode === 'append' ? await nextImportStartOrderKey(tableId) : null

    // Append keeps the existing rows; create/replace start from empty (replace deletes
    // existing rows in resolveSetup). Per-batch capacity is checked against this base + the
    // running total, so a stream that crosses the plan limit fails within one batch.
    const existingRowCount = mode === 'append' ? table.rowCount : 0

    // Count bytes as they flow so the row total can be extrapolated from byte progress.
    let bytesRead = 0
    const byteCounter = new Transform({
      transform(chunk: Buffer, _enc, cb) {
        bytesRead += chunk.length
        cb(null, chunk)
      },
    })

    const parser = createCsvParser(delimiter)
    // `.pipe` doesn't forward source errors; forward so the iterator throws.
    source.on('error', (err) => parser.destroy(err))
    byteCounter.on('error', (err) => parser.destroy(err))
    source.pipe(byteCounter).pipe(parser)

    let schema: TableSchema | null = null
    let headerToColumn: Map<string, string> | null = null
    let inserted = 0
    let lastReported = 0
    const sample: Record<string, unknown>[] = []
    let batch: Record<string, unknown>[] = []

    /**
     * Resolve the schema + header→column mapping from the buffered sample (runs once).
     * `create` infers a fresh schema and overwrites the placeholder; `append`/`replace`
     * map onto the existing schema, optionally auto-creating `createColumns` first.
     */
    const resolveSetup = async () => {