Function flush

apps/sim/lib/table/import-runner.ts:195–247

(rows: Record<string, unknown>[])

    }

    const flush = async (rows: Record<string, unknown>[]) => {
      if (rows.length === 0 || !schema || !headerToColumn) return
      // Ownership gate before every insert: once this run loses the table (cancel/supersede),
      // updateJobProgress returns false and we stop before writing into a table a newer import
      // may own. Runs per batch (not just at the emit cadence) so we stop within one batch.
      const owns = await updateJobProgress(tableId, inserted, importId)
      if (!owns) throw new ImportSupersededError()
      const coerced = coerceRowsForTable(rows, schema, headerToColumn)
      const rowLimit = await assertRowCapacity({
        workspaceId,
        currentRowCount: existingRowCount + inserted,
        addedRows: coerced.length,
      })
      const result = await bulkInsertImportBatch(
        {
          tableId,
          workspaceId,
          userId,
          rows: coerced,
          startPosition: basePosition + inserted,
          afterOrderKey: lastOrderKey,
        },
        { ...table, schema },
        requestId
      )
      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.
      if (
        inserted - lastReported >= PROGRESS_INTERVAL_ROWS ||
        (lastReported === 0 && inserted > 0)
      ) {
        lastReported = inserted
        // Exact, monotonic completion from bytes consumed — no wobbly row estimate.
        const percent =
          totalBytes > 0 ? Math.min(99, Math.round((bytesRead / totalBytes) * 100)) : undefined
        void appendTableEvent({
          kind: 'job',
          type: 'import',
          tableId,
          jobId: importId,
          status: 'running',
          progress: inserted,
          percent,
        })
      }
    }

    let ready = false
    for await (const record of parser as AsyncIterable<Record<string, unknown>>) {