Designing a Tenant Data Import System: File Parsing, Schema Mapping, and Error Recovery for SaaS Onboarding at Scale

Every mature SaaS product eventually ships a data import feature. A customer decides to migrate from a competitor, or they have years of records in a spreadsheet, or their previous system let them export a CSV. Without import, you are asking them to re-enter everything by hand, and most will not bother.

The deceptive part is that import looks like a solved problem. You read a file, validate the rows, write to the database. The reality is far more interesting. Files arrive in inconsistent formats. Column names do not match your schema. Rows fail validation partway through a 200,000-row upload. The import crashes and the customer wants to retry without duplicating what already processed. Five customers start large imports simultaneously and your database write throughput collapses.

This article covers the complete architecture for a production-grade import system, from file ingestion through background processing to error recovery and multi-tenant concurrency.

The Problem with Synchronous Import

The first instinct is a synchronous endpoint: receive the file, parse it, validate it, write it, return a result. This works for small files in demos. It fails predictably in production.

HTTP timeouts are typically 30-60 seconds at the load balancer level. A 50,000-row CSV takes several seconds to parse and validate even before any database writes. The connection dies before you finish, the client has no idea what happened, and partial writes may already be in place.

Memory is the second issue. Loading a 100MB file into a Node.js process buffer to parse it will spike RSS by several hundred megabytes. Under concurrent load, this pattern will OOM your API servers.

The correct model separates concerns cleanly: accept the file, acknowledge immediately with a job ID, parse and process in the background, and push progress updates over WebSocket. The user-facing latency is the upload time only. Everything else is async.

File Ingestion and Storage

Before any parsing happens, the file needs to land somewhere durable. Accepting it directly into the API server process and then passing it to a queue is fragile. The right pattern is to have the client upload directly to object storage (S3, GCS, or R2) using a presigned URL, then notify your API with the storage key.

interface ImportUploadRequest {
  tenantId: string;
  filename: string;
  contentType: "text/csv" | "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
  sizeBytes: number;
}

interface ImportUploadResponse {
  importId: string;
  uploadUrl: string;       // presigned PUT URL, expires in 15 minutes
  uploadFields?: Record<string, string>;  // for multipart POST
}

async function createImportUpload(req: ImportUploadRequest): Promise<ImportUploadResponse> {
  const importId = crypto.randomUUID();

  // Write the import record first, before issuing the presigned URL.
  // This ensures the record exists when the upload completes.
  await db.insert(imports).values({
    id: importId,
    tenantId: req.tenantId,
    filename: req.filename,
    status: "awaiting_upload",
    createdAt: new Date(),
  });

  const storageKey = `imports/${req.tenantId}/${importId}/${req.filename}`;
  const uploadUrl = await storage.getSignedPutUrl(storageKey, {
    contentType: req.contentType,
    expiresIn: 900,
    maxSizeBytes: 250 * 1024 * 1024,  // 250MB hard cap
  });

  return { importId, uploadUrl };
}

After the upload completes, the client calls a confirmation endpoint. This is where you enqueue the background job:

async function confirmImportUpload(importId: string, tenantId: string): Promise<void> {
  const imp = await db.query.imports.findFirst({
    where: and(eq(imports.id, importId), eq(imports.tenantId, tenantId)),
  });

  if (!imp || imp.status !== "awaiting_upload") {
    throw new Error("Import not found or already confirmed");
  }

  await db.update(imports).set({ status: "queued" }).where(eq(imports.id, importId));

  await importQueue.add("parse-import", { importId, tenantId }, {
    jobId: importId,        // idempotent: re-queuing the same importId is a no-op
    attempts: 3,
    backoff: { type: "exponential", delay: 5000 },
  });
}

Streaming File Parsing

A 200MB CSV file should never be loaded into memory in full. The correct approach is a streaming parser that emits rows as it reads. Node.js streams make this straightforward:

import { createReadStream } from "fs";
import { parse } from "csv-parse";
import { pipeline } from "stream/promises";

interface ParsedRow {
  rowIndex: number;
  rawValues: Record<string, string>;
}

async function* streamCsvRows(storageKey: string): AsyncGenerator<ParsedRow> {
  const fileStream = await storage.createReadStream(storageKey);
  const parser = parse({
    columns: true,            // use first row as header
    skip_empty_lines: true,
    trim: true,
    relax_column_count: true, // tolerate rows with extra/missing columns
  });

  let rowIndex = 0;
  fileStream.pipe(parser);

  for await (const record of parser) {
    yield { rowIndex: rowIndex++, rawValues: record };
  }
}

For Excel files (XLSX), the streaming story is more complicated because the format is a ZIP archive containing XML. The xlsx-stream-reader or exceljs libraries support streaming reads, but row-level memory usage is still higher than CSV. Set a practical limit on Excel file sizes (10-25MB is reasonable) and convert them to CSV server-side before the main processing pipeline.

Column Mapping and Schema Inference

Customer files rarely have column names that match your schema. A CRM export might have “First Name” where you expect first_name. An accounting export might have “Account Number (Legacy)” where you need account_id. Manual column mapping via a UI is the gold standard for data quality, but you need sensible inference to pre-populate those mappings.

interface FieldDefinition {
  name: string;           // canonical field name in your schema
  displayName: string;
  type: "string" | "email" | "phone" | "date" | "number" | "boolean";
  required: boolean;
  aliases: string[];      // known alternate column names from common sources
}

interface ColumnMapping {
  sourceColumn: string;   // header from uploaded file
  targetField: string | null;  // null = ignore this column
  confidence: number;     // 0-1, for UI display
}

function inferColumnMappings(
  sourceHeaders: string[],
  targetFields: FieldDefinition[]
): ColumnMapping[] {
  return sourceHeaders.map((header) => {
    const normalized = header.toLowerCase().replace(/[^a-z0-9]/g, "_");

    // Exact match first
    const exactMatch = targetFields.find(
      (f) => f.name === normalized || f.aliases.includes(normalized)
    );
    if (exactMatch) {
      return { sourceColumn: header, targetField: exactMatch.name, confidence: 1.0 };
    }

    // Fuzzy match: strip common noise words and compare tokens
    const sourceTokens = new Set(normalized.split("_").filter((t) => t.length > 2));
    let bestMatch: { field: FieldDefinition; score: number } | null = null;

    for (const field of targetFields) {
      const fieldTokens = new Set(
        [field.name, ...field.aliases].join("_").split("_").filter((t) => t.length > 2)
      );
      const intersection = [...sourceTokens].filter((t) => fieldTokens.has(t)).length;
      const union = new Set([...sourceTokens, ...fieldTokens]).size;
      const score = union > 0 ? intersection / union : 0;

      if (score > 0.4 && (!bestMatch || score > bestMatch.score)) {
        bestMatch = { field, score };
      }
    }

    if (bestMatch) {
      return {
        sourceColumn: header,
        targetField: bestMatch.field.name,
        confidence: bestMatch.score,
      };
    }

    return { sourceColumn: header, targetField: null, confidence: 0 };
  });
}

Store the confirmed mapping (after user review in the UI) alongside the import record. This becomes the execution plan for the processing phase.

Validation Pipeline

Validation is where most import complexity lives. You need to distinguish between row-level errors (a specific cell has a bad value), structural errors (the file is missing a required column entirely), and business logic errors (a foreign key reference does not exist for this tenant).

interface ValidationRule {
  field: string;
  check: (value: string, context: ValidationContext) => ValidationError | null;
}

interface ValidationError {
  rowIndex: number;
  field: string;
  value: string;
  code: string;
  message: string;
}

interface ValidationContext {
  tenantId: string;
  existingIds?: Set<string>;  // preloaded for FK checks
}

const builtInRules: ValidationRule[] = [
  {
    field: "email",
    check: (value) =>
      /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(value)
        ? null
        : { rowIndex: -1, field: "email", value, code: "INVALID_EMAIL", message: `"${value}" is not a valid email address` },
  },
  {
    field: "phone",
    check: (value) =>
      /^\+?[\d\s\-().]{7,20}$/.test(value)
        ? null
        : { rowIndex: -1, field: "phone", value, code: "INVALID_PHONE", message: `"${value}" does not look like a phone number` },
  },
];

async function* validateRows(
  rows: AsyncGenerator<ParsedRow>,
  mapping: ColumnMapping[],
  rules: ValidationRule[],
  context: ValidationContext
): AsyncGenerator<{ row: ParsedRow; errors: ValidationError[] }> {
  for await (const row of rows) {
    const errors: ValidationError[] = [];

    for (const col of mapping) {
      if (!col.targetField) continue;
      const rawValue = row.rawValues[col.sourceColumn] ?? "";
      const applicableRules = rules.filter((r) => r.targetField === col.targetField);

      for (const rule of applicableRules) {
        const error = rule.check(rawValue, context);
        if (error) {
          errors.push({ ...error, rowIndex: row.rowIndex });
        }
      }
    }

    yield { row, errors };
  }
}

Batch FK validation separately. Checking whether account_id exists for every row with an individual query is a guaranteed way to serialize your import into single-digit rows per second. Load the set of valid IDs for the tenant into memory before starting (or into a temporary table for very large sets) and do set-membership checks in process.

Background Processing with Progress Tracking

The processing job reads from the validation stream and writes to the database in batches. It reports progress by updating a processed_rows counter in Redis every N rows. WebSocket clients poll or subscribe to these updates:

interface ImportProgress {
  importId: string;
  status: "running" | "completed" | "failed" | "partial";
  totalRows: number;
  processedRows: number;
  errorRows: number;
  skippedRows: number;
  currentPhase: "parsing" | "validating" | "writing" | "finalizing";
  startedAt: Date;
  estimatedCompletionAt?: Date;
}

async function processImportJob(importId: string, tenantId: string): Promise<void> {
  const imp = await db.query.imports.findFirst({ where: eq(imports.id, importId) });
  const mapping: ColumnMapping[] = JSON.parse(imp.mappingJson);
  const config: ImportConfig = JSON.parse(imp.configJson);

  await setImportStatus(importId, "running");

  const rows = streamCsvRows(imp.storageKey);
  const validated = validateRows(rows, mapping, getRulesForImport(config), {
    tenantId,
    existingIds: await loadExistingIds(tenantId, config.entityType),
  });

  const writeBatch: object[] = [];
  const errors: ValidationError[] = [];
  let processedRows = 0;

  for await (const { row, errors: rowErrors } of validated) {
    if (rowErrors.length > 0) {
      errors.push(...rowErrors);
      if (config.errorStrategy === "fail_fast") {
        await finalizeImport(importId, "failed", processedRows, errors);
        return;
      }
      // skip_on_error: record the error but continue
      continue;
    }

    writeBatch.push(mapRowToEntity(row, mapping, tenantId));
    processedRows++;

    if (writeBatch.length >= 500) {
      await flushBatch(writeBatch, config.entityType, tenantId);
      writeBatch.length = 0;
      await updateImportProgress(importId, processedRows, errors.length);
    }
  }

  if (writeBatch.length > 0) {
    await flushBatch(writeBatch, config.entityType, tenantId);
  }

  await persistImportErrors(importId, errors);
  const finalStatus = errors.length > 0 ? "partial" : "completed";
  await finalizeImport(importId, finalStatus, processedRows, errors);
}

Batch size of 500 rows is a reasonable default for Postgres INSERT ... VALUES payloads. Beyond 1000 rows per batch, you see diminishing returns on throughput and increasing lock contention on adjacent rows.

Idempotency and Resume

Imports fail. The job runner reboots, the database connection drops, the worker process is evicted. When the job retries, you need to avoid writing duplicate rows.

The cleanest approach is to assign each source row a deterministic ID before writing, derived from the import ID and the row index:

function deriveRowId(importId: string, rowIndex: number): string {
  return `${importId}:${rowIndex}`;
}

async function flushBatch(
  batch: EntityWithSourceId[],
  entityType: string,
  tenantId: string
): Promise<void> {
  // ON CONFLICT DO NOTHING means retrying the same batch is safe.
  // The source_import_row_id column has a UNIQUE INDEX.
  await db
    .insert(getEntityTable(entityType))
    .values(batch)
    .onConflictDoNothing({ target: source_import_row_id });
}

This works because the row index is stable across retries. Row 4,721 in the CSV is always row 4,721. If the job crashes after writing rows 0-4,999 and retries from the beginning, rows 0-4,999 will hit ON CONFLICT DO NOTHING and the batch continues without duplicates.

For the progress counter, store a last_committed_row_index on the import record after each batch flush. On retry, skip rows up to that index before resuming writes:

async function* skipAlreadyProcessed(
  rows: AsyncGenerator<ParsedRow>,
  resumeFromIndex: number
): AsyncGenerator<ParsedRow> {
  for await (const row of rows) {
    if (row.rowIndex <= resumeFromIndex) continue;
    yield row;
  }
}

Error Strategy Tradeoffs

The choice between fail-fast and skip-on-error is not purely technical. It reflects what your customers expect.

Strategy	Behavior	When to Use	Tradeoff
Fail-fast	Abort on first validation error	Strict data integrity requirements	All-or-nothing; customer must fix file before any data lands
Skip-on-error	Write valid rows, collect errors	Migration-heavy use cases	Partial data in system; error report must be clear and actionable
Threshold-based	Abort if error rate exceeds N%	Balance of both	Requires configuring threshold; can surprise customers near boundary
Two-pass	Validate all rows before writing any	Maximum predictability	Requires full file read twice; doubles processing time

Two-pass is worth the cost for entity types where a partial import leaves your data model in an inconsistent state. If contacts reference companies that appear later in the file, a single-pass approach will fail FK checks for all contacts before the company rows have been written. Two-pass validation with a deferred-write phase is the correct model for those cases.

Multi-Tenant Concurrency

With multiple tenants running large imports simultaneously, you face two resource contention problems: CPU (parsing and validation) and database write throughput.

The parsing and validation stages are stateless and can scale horizontally by adding more worker processes. Use a per-tenant concurrency limit in the job queue to prevent one tenant from monopolizing all workers:

const importQueue = new Queue("imports", {
  connection: redis,
  defaultJobOptions: {
    concurrency: 2,       // max 2 jobs per tenant at once
    removeOnComplete: 100,
    removeOnFail: 500,
  },
});

// When adding jobs, use a group key so the queue respects per-tenant limits
await importQueue.add(
  "parse-import",
  { importId, tenantId },
  {
    jobId: importId,
    group: { id: tenantId, concurrency: 2 },
  }
);

Database writes are harder to isolate. Batch inserts from multiple imports hitting the same table simultaneously cause lock contention on shared index pages. Two mitigations help here:

First, time your batch flushes with a small random jitter (50-150ms) so simultaneous jobs do not issue writes in lock-step.

Second, consider a write queue in front of the database for very high import volume: workers publish batches to a Redis stream, and a dedicated database writer process consumes from the stream at a controlled rate. This decouples import worker scaling from database write throughput, at the cost of a more complex pipeline.

Production Considerations

File format sniffing. Customers upload files with the wrong extension or content type header. Sniff the actual format by reading the first few bytes: CSV files start with printable ASCII, XLSX files start with the PK ZIP magic bytes (50 4B 03 04). Reject mismatches with a clear error rather than letting the parser fail cryptically mid-file.

Encoding detection. Not all CSV files are UTF-8. Files exported from Excel on Windows are often Windows-1252. Use a library like chardet to detect encoding before parsing, and transcode to UTF-8 before handing to the parser.

Memory bounds for validation preloads. Loading existing IDs into memory for FK checks is fine at 100,000 IDs (a Set<string> of 100K UUIDs is roughly 10-20MB). At 10 million IDs, load into a temporary Postgres table and do the check as a bulk NOT IN (SELECT id FROM temp_valid_ids) query per batch instead.

Error report delivery. When an import completes with errors, the error report can itself be large. Write it to object storage as a CSV file and return a presigned download URL, not a JSON array in the API response. This keeps the API response size bounded regardless of how many rows failed.

Tenant storage quotas. Track per-tenant storage consumption for uploaded import files. Files should be deleted from object storage after processing completes (or after a configurable retention window). Orphaned files from failed uploads accumulate quickly without a cleanup job.

Observability. Instrument each phase separately: upload completion time, queue wait time, parse throughput (rows per second), validation error rate, write throughput, and total job duration. A slow parse phase points to parser inefficiency or network-bottlenecked object storage reads. A slow write phase points to database contention. Without phase-level metrics, these problems look identical from the outside.

Closing

The surface area of a data import system is larger than it appears at first. File ingestion, streaming parsing, column mapping, multi-phase validation, background processing, idempotent resume, error recovery, and multi-tenant fairness each have failure modes that only appear under real load with real customer data. The architecture described here handles each of those failure modes explicitly rather than hoping they do not occur. The investment pays back during onboarding: customers who can import their existing data quickly become customers who stay.