Why not just POST the CSV file to a Next.js API route?

Because you hit two ceilings at once. Serverless request bodies are capped (4.5 MB on Vercel by default), and a route handler that parses 40,000 rows will run past the function timeout long before it finishes. Hand the browser a signed upload URL instead so the bytes go straight to Supabase Storage, then process the file in a background job that has its own timeout budget and its own retries.

How do I stop one bad row from killing a 50,000 row import?

Never throw inside the row loop. Validate each row with a Zod schema and safeParse, push the failures into an import_row_errors table with the row number and the field that failed, and keep going. Then upsert the valid rows in chunks, with each chunk wrapped in its own Inngest step.run so a failed chunk retries on its own instead of replaying the whole file.

How do I export a large table as CSV without running out of memory?

Return a ReadableStream from the route handler and page through the table with a keyset cursor (order by id, then gt on the last id you saw). Each page of 1,000 rows is encoded and enqueued, so peak memory is one page and the browser starts downloading before the query finishes. Set Content-Type to text/csv, add a Content-Disposition attachment header, and send Cache-Control no-store.

CSV Import Pipeline

Bulk import is where a clean SaaS gets ugly. A customer drops in a 60,000 row export from their old CRM, and your API route times out, one malformed email kills the whole run, and a retry duplicates every contact that already made it in. This guide wires the whole path end to end: signed upload, streamed parse, Zod row validation, a chunked Inngest upsert that survives partial failure, a per-row error report the customer can actually fix, and a streaming export on the way back out.

What You're Building

Six moving parts, each with one job. The browser never sends the file through your API. Your server hands out a signed upload token, the bytes go straight to Supabase Storage, and a row in import_jobs tracks the run. An Inngest function then streams the file, validates every row, stages the good ones and records the bad ones, and upserts the staged rows in chunks. Chunks are the unit of retry, so a network blip on chunk 37 does not replay chunks 0 through 36.

At the end the customer gets two things: a count of what landed, and a downloadable CSV listing every row that did not, with the row number, the offending column, and the original value. That error report is the difference between a feature people use and a feature people email support about.

Give Claude Code the rules before it writes anything. Drop this in CLAUDE.md and it stops reaching for the naive multipart POST:

## CSV import

- Files NEVER pass through a route handler. Signed upload URL to the "imports" bucket.
- Parsing is streamed with csv-parse. Never JSON.parse or readFile the whole CSV.
- Every row goes through a Zod schema with safeParse. Bad rows are collected, never thrown.
- Writes happen in Inngest step.run chunks of 500. Chunk index is the step id.
- Contacts are upserted on the (user_id, email) unique constraint. Never plain insert.
- Every upsert batch is deduped on its conflict key before it is written.
- Exports stream via ReadableStream with a keyset cursor. Never select the whole table.

The Schema

Four tables. contacts is the destination, and the unique constraint on (user_id, email) is what makes the whole pipeline safe to retry. import_jobs is the run record. import_rows is a staging table that holds validated rows between the parse step and the upsert steps. import_row_errors is what the error report is built from.

Ask Claude Code to write the migration, then check two things by hand: the unique constraints, and that RLS is on everywhere.

-- supabase/migrations/20260722000000_csv_import.sql
create table contacts (
  id uuid primary key default gen_random_uuid(),
  user_id uuid not null references auth.users(id) on delete cascade,
  email text not null,
  full_name text not null,
  company text,
  plan text not null default 'free',
  mrr numeric not null default 0,
  signed_up_at timestamptz,
  updated_at timestamptz not null default now(),
  unique (user_id, email)
);

create table import_jobs (
  id uuid primary key,
  user_id uuid not null references auth.users(id) on delete cascade,
  storage_path text not null,
  status text not null default 'pending',
  total_rows int not null default 0,
  valid_rows int not null default 0,
  failed_rows int not null default 0,
  error_report_path text,
  created_at timestamptz not null default now()
);

create table import_rows (
  job_id uuid not null references import_jobs(id) on delete cascade,
  row_number int not null,
  chunk int not null,
  payload jsonb not null,
  primary key (job_id, row_number)
);
create index import_rows_chunk_idx on import_rows (job_id, chunk);

create table import_row_errors (
  id bigserial primary key,
  job_id uuid not null references import_jobs(id) on delete cascade,
  row_number int not null,
  field text not null,
  message text not null,
  raw jsonb,
  unique (job_id, row_number, field)
);

alter table contacts enable row level security;
alter table import_jobs enable row level security;
alter table import_rows enable row level security;
alter table import_row_errors enable row level security;

create policy "own contacts" on contacts for all to authenticated
  using ((select auth.uid()) = user_id)
  with check ((select auth.uid()) = user_id);

create policy "own jobs" on import_jobs for all to authenticated
  using ((select auth.uid()) = user_id)
  with check ((select auth.uid()) = user_id);

-- No policies on import_rows or import_row_errors on purpose. RLS is enabled
-- with zero policies, which denies every client. Only the background worker
-- (service role) touches them, and customers read failures via the CSV report.

The primary key on import_rows is (job_id, row_number), not a serial id. That is deliberate. If the parse step dies at row 40,000 and Inngest retries it from the top, the staging writes are upserts against that key, so the rerun overwrites instead of duplicating.

Signed Upload, Not a Request Body

The upload procedure creates the job row and mints a one-time upload token in the same call. It never touches the file bytes. Note that the job id is generated server side and used as both the primary key and the storage filename, so the path is guessable only by someone who already owns the job.

The size cap here is a courtesy check on a number the client reports. Storage enforces the real one at the bucket level, which is where a cap belongs.

// server/imports.ts
import { z } from "zod";
import { randomUUID } from "node:crypto";
import { authed } from "./base";
import { inngest } from "@/lib/inngest/client";

export const createImport = authed
  .input(
    z.object({
      filename: z.string().min(1).max(200),
      sizeBytes: z
        .number()
        .int()
        .positive()
        .max(50 * 1024 * 1024),
    })
  )
  .handler(async ({ input, context }) => {
    const { supabase, user } = context;

    if (!input.filename.toLowerCase().endsWith(".csv")) {
      throw new Error("Only .csv files are accepted");
    }

    const jobId = randomUUID();
    const path = `${user.id}/${jobId}.csv`;

    const { data: signed, error } = await supabase.storage
      .from("imports")
      .createSignedUploadUrl(path);

    if (error) throw new Error(error.message);

    const { error: insertError } = await supabase.from("import_jobs").insert({
      id: jobId,
      user_id: user.id,
      storage_path: path,
      status: "pending",
    });

    if (insertError) throw new Error(insertError.message);

    return { jobId, path: signed.path, token: signed.token };
  });

export const startImport = authed
  .input(z.object({ jobId: z.string().uuid() }))
  .handler(async ({ input, context }) => {
    const { supabase, user } = context;

    const { error } = await supabase
      .from("import_jobs")
      .update({ status: "queued" })
      .eq("id", input.jobId)
      .eq("user_id", user.id);

    if (error) throw new Error(error.message);

    await inngest.send({
      name: "import/csv.uploaded",
      data: { jobId: input.jobId, userId: user.id },
    });

    return { queued: true };
  });

Two procedures instead of one, because the upload happens between them. The browser calls createImport, pushes the bytes with the token it got back, then calls startImport. If the upload fails, the job row sits at pending and no event is ever sent.

// components/import-form.tsx (inside the submit handler)
const supabase = createClient();

const { jobId, path, token } = await client.imports.createImport({
  filename: file.name,
  sizeBytes: file.size,
});

const { error } = await supabase.storage
  .from("imports")
  .uploadToSignedUrl(path, token, file, { contentType: "text/csv" });

if (error) throw new Error(error.message);

await client.imports.startImport({ jobId });

Zod Row Validation

Real customer CSVs have "Email Address" in one file and "e-mail" in the next, $1,200 in a revenue column, and blank cells where you expect a date. Handle all of it in one place: a header normalizer plus a row schema.

The alias map is the part worth curating. Every alias you add is a support ticket you never receive. Anything unrecognized falls through to a snake_case version of the header, which will simply fail validation with a clear message.

// lib/imports/schema.ts
import { z } from "zod";

const ALIASES: Record<string, string> = {
  email: "email",
  "email address": "email",
  "e mail": "email",
  "work email": "email",
  name: "full_name",
  "full name": "full_name",
  "contact name": "full_name",
  company: "company",
  organization: "company",
  organisation: "company",
  account: "company",
  plan: "plan",
  tier: "plan",
  mrr: "mrr",
  "monthly revenue": "mrr",
  "signed up": "signed_up_at",
  "signup date": "signed_up_at",
  "created at": "signed_up_at",
};

export function normalizeHeader(raw: string): string {
  const key = raw.trim().toLowerCase().replace(/[_-]+/g, " ").replace(/\s+/g, " ");
  return ALIASES[key] ?? key.replace(/\s+/g, "_");
}

const text = (max: number) =>
  z.preprocess((v) => String(v ?? "").trim(), z.string().max(max));

export const ContactRow = z.object({
  email: z.preprocess(
    (v) => String(v ?? "").trim().toLowerCase(),
    z.string().email("Not a valid email address")
  ),
  full_name: z.preprocess(
    (v) => String(v ?? "").trim(),
    z.string().min(1, "Name is required").max(120)
  ),
  company: text(120).transform((v) => (v === "" ? null : v)),
  plan: z.preprocess(
    (v) => String(v ?? "").trim().toLowerCase() || "free",
    z.enum(["free", "pro", "enterprise"])
  ),
  mrr: z.preprocess((v) => {
    const s = String(v ?? "").replace(/[$,\s]/g, "");
    return s === "" ? 0 : Number(s);
  }, z.number().min(0, "Revenue cannot be negative")),
  signed_up_at: z
    .preprocess((v) => {
      const s = String(v ?? "").trim();
      return s === "" ? null : new Date(s);
    }, z.date().nullable())
    .transform((v) => (v ? v.toISOString() : null)),
});

export type Contact = z.infer<typeof ContactRow>;

z.preprocess does the coercion, the inner schema does the judging, and the error message you write is the one the customer reads in the report. new Date("not a date") produces an Invalid Date, which z.date() rejects, so garbage in the date column fails loudly instead of writing a null.

Everything transforms to something JSON serializable. The date becomes an ISO string, because these rows get staged in a jsonb column and passed through Inngest step output.

Streaming the Parse

csv-parse implements the Node stream API and exposes an async iterator, which is exactly what you want: one row in memory at a time regardless of file size. The bridge from the web ReadableStream you get from fetch to a Node stream is Readable.fromWeb.

The columns option accepts a function, so header normalization happens during parsing rather than in a second pass. bom: true strips the byte order mark Excel writes on Windows, and relax_column_count keeps a row with a stray trailing comma from aborting the parse.

// lib/imports/parse.ts
import { Readable } from "node:stream";
import { parse } from "csv-parse";
import { ContactRow, normalizeHeader, type Contact } from "./schema";

export const CHUNK_SIZE = 500;

export type ParsedRow =
  | { ok: true; rowNumber: number; value: Contact }
  | {
      ok: false;
      rowNumber: number;
      raw: Record<string, string>;
      errors: { field: string; message: string }[];
    };

export async function* parseContacts(
  body: ReadableStream<Uint8Array>
): AsyncGenerator<ParsedRow> {
  const parser = Readable.fromWeb(body as never).pipe(
    parse({
      bom: true,
      trim: true,
      skip_empty_lines: true,
      relax_column_count: true,
      columns: (header: string[]) => header.map(normalizeHeader),
    })
  );

  // Row 1 is the header, so data rows start at 2 and match what the user sees.
  let rowNumber = 1;

  for await (const record of parser as AsyncIterable<Record<string, string>>) {
    rowNumber += 1;
    const result = ContactRow.safeParse(record);

    if (result.success) {
      yield { ok: true, rowNumber, value: result.data };
    } else {
      yield {
        ok: false,
        rowNumber,
        raw: record,
        errors: result.error.issues.map((issue) => ({
          field: String(issue.path[0] ?? "row"),
          message: issue.message,
        })),
      };
    }
  }
}

Counting from 2 is a small thing that matters a lot. The customer opens the file in Excel, and row 2 in your error report is row 2 on their screen.

The Inngest Batch Upsert

The job runs in three phases. Parse and stage happens once, inside a single step, because streaming a file is not something you want re-running on every step boundary. Then one step per chunk of validated rows. Then a finalize step.

Note the download: it uses a signed URL and fetch rather than the Storage client's download(). download() resolves to a Blob, which means the whole file lands in memory first. fetch gives you res.body, a real stream, which is the entire point here.

// lib/inngest/functions/import-contacts.ts
import { inngest } from "@/lib/inngest/client";
import { admin } from "@/lib/supabase/admin";
import { parseContacts, CHUNK_SIZE } from "@/lib/imports/parse";
import { buildErrorReport } from "@/lib/imports/error-report";

const FLUSH_AT = 1000;

export const importContacts = inngest.createFunction(
  {
    id: "import-contacts",
    retries: 3,
    concurrency: { limit: 3, key: "event.data.userId" },
    onFailure: async ({ event }) => {
      const db = admin();
      await db
        .from("import_jobs")
        .update({ status: "failed" })
        .eq("id", event.data.event.data.jobId);
    },
  },
  { event: "import/csv.uploaded" },
  async ({ event, step }) => {
    const { jobId, userId } = event.data;
    const db = admin();

    const job = await step.run("load-job", async () => {
      const { data, error } = await db
        .from("import_jobs")
        .select("id, storage_path")
        .eq("id", jobId)
        .single();
      if (error) throw new Error(error.message);
      return data;
    });

    const counts = await step.run("parse-and-stage", async () => {
      const { data: signed, error: signError } = await db.storage
        .from("imports")
        .createSignedUrl(job.storage_path, 900);
      if (signError) throw new Error(signError.message);

      const res = await fetch(signed.signedUrl);
      if (!res.ok || !res.body) {
        throw new Error(`Could not read ${job.storage_path}`);
      }

      let total = 0;
      let valid = 0;
      let failed = 0;
      let staged: Record<string, unknown>[] = [];
      let errors: Record<string, unknown>[] = [];

      const flush = async (
        table: string,
        rows: Record<string, unknown>[],
        onConflict: string
      ) => {
        if (rows.length === 0) return;
        const { error } = await db.from(table).upsert(rows, { onConflict });
        if (error) throw new Error(error.message);
      };

      const flushStaged = async () => {
        await flush("import_rows", staged, "job_id,row_number");
        staged = [];
      };

      const flushErrors = async () => {
        await flush("import_row_errors", errors, "job_id,row_number,field");
        errors = [];
      };

      for await (const row of parseContacts(res.body)) {
        total += 1;

        if (row.ok) {
          valid += 1;
          staged.push({
            job_id: jobId,
            row_number: row.rowNumber,
            chunk: Math.floor((valid - 1) / CHUNK_SIZE),
            payload: { ...row.value, user_id: userId },
          });
          if (staged.length >= FLUSH_AT) await flushStaged();
        } else {
          failed += 1;
          for (const issue of row.errors) {
            errors.push({
              job_id: jobId,
              row_number: row.rowNumber,
              field: issue.field,
              message: issue.message,
              raw: row.raw,
            });
          }
          if (errors.length >= FLUSH_AT) await flushErrors();
        }
      }

      await flushStaged();
      await flushErrors();

      await db
        .from("import_jobs")
        .update({
          status: "upserting",
          total_rows: total,
          valid_rows: valid,
          failed_rows: failed,
        })
        .eq("id", jobId);

      return { total, valid, failed, chunks: Math.ceil(valid / CHUNK_SIZE) };
    });

    for (let chunk = 0; chunk < counts.chunks; chunk += 1) {
      await step.run(`upsert-chunk-${chunk}`, async () => {
        const { data: rows, error } = await db
          .from("import_rows")
          .select("row_number, payload")
          .eq("job_id", jobId)
          .eq("chunk", chunk)
          .order("row_number", { ascending: true });
        if (error) throw new Error(error.message);

        const payloads = (rows ?? []).map(
          (r) => r.payload as Record<string, unknown>
        );
        if (payloads.length === 0) return { chunk, written: 0 };

        // One INSERT cannot update the same row twice, so a chunk carrying the
        // same email on two rows would throw. Later row in the file wins.
        const deduped = [
          ...new Map(payloads.map((p) => [p.email as string, p] as const)).values(),
        ];

        const { error: upsertError } = await db
          .from("contacts")
          .upsert(deduped, { onConflict: "user_id,email" });
        if (upsertError) throw new Error(upsertError.message);

        return { chunk, written: deduped.length };
      });
    }

    const reportPath = await step.run("write-error-report", async () =>
      counts.failed === 0 ? null : buildErrorReport(db, jobId, userId)
    );

    await step.run("finalize", async () => {
      await db
        .from("import_jobs")
        .update({ status: "completed", error_report_path: reportPath })
        .eq("id", jobId);
      await db.from("import_rows").delete().eq("job_id", jobId);
    });

    return { ...counts, reportPath };
  }
);

The step id upsert-chunk-${chunk} is what makes partial failure survivable. Inngest memoizes each completed step by id, so if chunk 37 throws on a connection reset, the retry replays chunks 0 through 36 from memoized results in milliseconds and picks up at 37. Combined with the (user_id, email) upsert, replaying a chunk that already landed is a no-op rather than a duplicate.

The in-chunk dedupe is not optional either. Postgres refuses to let one INSERT ... ON CONFLICT DO UPDATE touch the same row twice, so a chunk carrying the same email on two rows fails with ON CONFLICT DO UPDATE command cannot affect row a second time and takes the other 499 rows down with it. Customer exports have duplicate emails constantly. Collapsing on email before the write, later row wins, is the difference between a shrug and a support ticket. A chunk of 500 also keeps every chunk select under the 1,000 row ceiling PostgREST applies by default.

This function uses the service role client, which bypasses RLS by design (a background worker has no user session). Keep lib/supabase/admin.ts server only and never import it from a component.

One caveat worth knowing before you scale it: this loop creates one step per chunk, so a million-row file would create thousands of steps. Past a few hundred chunks, switch the loop to a fan-out with step.sendEvent("dispatch-chunks", ...) and process chunks in a second function with its own concurrency limit.

The parse step has a ceiling too, and it is easy to miss. Every step.run executes inside one request to your Inngest route, so the streaming parse is still bounded by that route's function timeout. Set maxDuration explicitly in app/api/inngest/route.ts instead of inheriting the platform default.

The Per-Row Error Report

A count of failures is useless. A file the customer can open, fix, and re-upload is the actual feature. Build it from import_row_errors, including the original value so they can see what you choked on.

The sync stringifier is fine here because errors are a small fraction of a healthy import, and the total is capped. The paging is not optional though. PostgREST returns at most the project's Max rows setting, which is 1,000 by default, so a single limit(10000) would hand you 1,000 rows and a report that quietly under-reports the damage. If a file is more broken than the cap, the customer has a bigger problem than pagination.

// lib/imports/error-report.ts
import { stringify } from "csv-stringify/sync";
import type { SupabaseClient } from "@supabase/supabase-js";

type ErrorRow = {
  row_number: number;
  field: string;
  message: string;
  raw: Record<string, string> | null;
};

const PAGE = 1000;
const MAX_ERROR_ROWS = 10000;

export async function buildErrorReport(
  db: SupabaseClient,
  jobId: string,
  userId: string
): Promise<string> {
  const rows: ErrorRow[] = [];

  // PostgREST caps a response at the project's Max rows setting, so page
  // instead of trusting one big limit(). Order by both key parts: a single
  // row can fail on more than one field.
  for (let offset = 0; offset < MAX_ERROR_ROWS; offset += PAGE) {
    const { data, error } = await db
      .from("import_row_errors")
      .select("row_number, field, message, raw")
      .eq("job_id", jobId)
      .order("row_number", { ascending: true })
      .order("field", { ascending: true })
      .range(offset, offset + PAGE - 1);

    if (error) throw new Error(error.message);
    if (!data || data.length === 0) break;

    rows.push(...(data as ErrorRow[]));
    if (data.length < PAGE) break;
  }

  const csv = stringify(
    rows.map((e) => ({
      row: e.row_number,
      column: e.field,
      problem: e.message,
      original_value: e.raw?.[e.field] ?? "",
    })),
    { header: true, columns: ["row", "column", "problem", "original_value"] }
  );

  const path = `${userId}/${jobId}-errors.csv`;

  const { error: uploadError } = await db.storage
    .from("imports")
    .upload(path, csv, { contentType: "text/csv", upsert: true });

  if (uploadError) throw new Error(uploadError.message);

  return path;
}

Serving it is a route handler with an awaited params, which is the Next.js 16 shape. The user session client is doing real work here: the select on import_jobs runs under RLS, so a request for someone else's job id returns nothing and falls through to a 404.

// app/api/imports/[id]/errors/route.ts
import { NextResponse, type NextRequest } from "next/server";
import { createClient } from "@/lib/supabase/server";

export async function GET(
  _request: NextRequest,
  { params }: { params: Promise<{ id: string }> }
) {
  const { id } = await params;
  const supabase = await createClient();

  const {
    data: { user },
  } = await supabase.auth.getUser();
  if (!user) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });

  const { data: job } = await supabase
    .from("import_jobs")
    .select("error_report_path")
    .eq("id", id)
    .single();

  if (!job?.error_report_path) {
    return NextResponse.json({ error: "No error report" }, { status: 404 });
  }

  const { data, error } = await supabase.storage
    .from("imports")
    .createSignedUrl(job.error_report_path, 300, {
      download: `import-${id}-errors.csv`,
    });

  if (error) return NextResponse.json({ error: error.message }, { status: 500 });

  return NextResponse.redirect(data.signedUrl);
}

The Streaming Export

Data has to come back out too, and the naive version (select everything, join it into a string, return it) dies on the first customer with real volume. Stream it instead: page through the table with a keyset cursor and enqueue each page as it arrives. Peak memory is one page, and the browser's download starts before the last query runs.

Two details in escapeCell are not optional. Quoting cells that contain commas, quotes, or newlines is correctness. Prefixing cells that start with =, +, -, or @ with an apostrophe is security: without it, a contact named =cmd|... becomes a live formula when someone opens your export in Excel.

// app/api/contacts/export/route.ts
import { type NextRequest } from "next/server";
import { createClient } from "@/lib/supabase/server";

export const runtime = "nodejs";
export const dynamic = "force-dynamic";

const COLUMNS = ["email", "full_name", "company", "plan", "mrr", "signed_up_at"] as const;
const PAGE = 1000;

function escapeCell(value: unknown): string {
  if (value === null || value === undefined) return "";
  let cell = String(value);
  if (/^[=+\-@\t\r]/.test(cell)) cell = `'${cell}`;
  if (/["\n\r,]/.test(cell)) cell = `"${cell.replace(/"/g, '""')}"`;
  return cell;
}

export async function GET(_request: NextRequest) {
  const supabase = await createClient();

  const {
    data: { user },
  } = await supabase.auth.getUser();
  if (!user) return new Response("Unauthorized", { status: 401 });

  const encoder = new TextEncoder();

  const stream = new ReadableStream<Uint8Array>({
    async start(controller) {
      // BOM first so Excel reads UTF-8 accents correctly.
      controller.enqueue(encoder.encode(`\uFEFF${COLUMNS.join(",")}\n`));

      let cursor: string | null = null;

      try {
        for (;;) {
          let filter = supabase
            .from("contacts")
            .select(`id, ${COLUMNS.join(", ")}`)
            .eq("user_id", user.id);

          if (cursor) filter = filter.gt("id", cursor);

          const { data, error } = await filter
            .order("id", { ascending: true })
            .limit(PAGE);

          if (error) throw new Error(error.message);
          if (!data || data.length === 0) break;

          const lines = data
            .map((row) =>
              COLUMNS.map((c) => escapeCell((row as Record<string, unknown>)[c])).join(",")
            )
            .join("\n");

          controller.enqueue(encoder.encode(`${lines}\n`));

          cursor = (data[data.length - 1] as { id: string }).id;
          if (data.length < PAGE) break;
        }

        controller.close();
      } catch (err) {
        controller.error(err);
      }
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/csv; charset=utf-8",
      "Content-Disposition": `attachment; filename="contacts-${new Date()
        .toISOString()
        .slice(0, 10)}.csv"`,
      "Cache-Control": "no-store",
    },
  });
}

Keyset pagination beats offset here for a reason worth internalizing: offset 50000 makes Postgres walk and discard 50,000 rows on every page, so the export gets slower the further it goes. gt("id", cursor) is an index seek every time.

This is also a good place to resist the "use cache" directive. Cache Components are excellent for a dashboard that reads the same aggregate for everyone, and wrong for a per-user export of live data. dynamic = "force-dynamic" plus Cache-Control: no-store is the correct pairing.

What Actually Breaks

These are the failures that show up once real customers touch the feature.

Symptom	Cause	Fix
Upload fails around 4.5 MB	File POSTed through a route handler	Signed upload URL straight to Storage
Job times out on big files	One long transaction	Chunked `step.run` writes
Duplicate contacts after a retry	Plain `insert`	Unique `(user_id, email)` plus `onConflict`
A whole chunk fails on a valid file	Same email twice in one upsert	Dedupe the chunk on the conflict key first
Error report stops at 1,000 rows	PostgREST Max rows cap	Page the query with `range`
Whole import dies on row 12	Throwing inside the row loop	`safeParse` plus an errors table
Accents look like mojibake in Excel	Missing byte order mark	`bom: true` on parse, a leading `\uFEFF` on export
A cell executes as a formula	CSV injection	Prefix leading `=` `+` `-` `@` with an apostrophe
Memory spikes on export	Selecting the whole table	Keyset cursor plus `ReadableStream`

Driving It With Claude Code

Build this in plan mode first. The pipeline has an order to it, and letting Claude sketch the file layout before writing anything catches the wrong-shaped decisions while they are still cheap to change.

claude --plan "add bulk CSV contact import: signed upload to the imports bucket,
an Inngest function that streams the file with csv-parse, validates rows with Zod,
stages valid rows, and upserts them in chunks of 500 with one step.run per chunk"

Then check four things in the output before you accept it. That parsing never buffers the whole file (no readFile, no await res.text()). That the row loop uses safeParse and cannot throw. That every write to contacts is an upsert on the unique constraint. That each batch is deduped on that same constraint before it is written.

Test against a deliberately hostile fixture, not a clean one. Ask for it explicitly:

claude "write a fixture CSV with 20 rows: 15 that must pass and 5 that must fail.
Among the valid rows put a name with a comma inside quotes, an mrr written as $1,200,
an accented name, a cell starting with =, and one row that repeats an email used
earlier in the file. Among the failures put a blank email, a malformed email, an
invalid date, a negative mrr, and an unknown plan. Then write a test asserting the
import produces 5 error rows and 14 contacts, since the repeated email upserts onto
the earlier one"

That file is worth keeping in the repo forever. It is the one that catches the regression when someone later swaps the parser or refactors the schema.

Where the Harness Fits

Claude Code writes each of these files well on its own. What it does not do on its own is hold a six-part pipeline to a standard across a whole session, and a broken import pipeline is expensive precisely because the damage is silent. That is what the $29 Code Kit adds on top of Claude Code: a harness that plans the feature before any code exists, builds it, evaluates the result, runs tests against it, and finishes with quality gates (zero type errors, zero lint errors, a clean build) before anything is considered done. It is $29 one time with no subscription. Claude Code itself is separate and needs a paid Anthropic plan.

Posted by @speedy_devv

CSV Import Pipeline

On this page