Bulk Data Processing in Go SaaS Backends: Imports, Exports, and Progress Tracking Without Timeouts

Every SaaS product eventually gets a user who tries to import 100,000 rows from a spreadsheet. Or export three years of transaction history. Or run a recalculation across an entire tenant's dataset. These operations break HTTP handlers, time out load balancers, and exhaust database connections if you do not have a deliberate architecture for them.

This is a class of problem we handle repeatedly when building backends for restaurant systems, inventory platforms, and business operations tools across Lebanon and the MENA region. The solution is not complicated, but it requires moving bulk operations out of the request-response cycle entirely.

Why HTTP handlers cannot run bulk operations

HTTP handlers have at least three limits that make them unsuitable for bulk operations:

Load balancer timeouts. AWS ALB has a default 60-second idle timeout. Most ECS-hosted APIs are behind a load balancer. A 100,000-row import that takes 3 minutes will get a TCP timeout at the load balancer before it finishes, even if the backend is still running.

Request context cancellation. If the client disconnects or the browser tab closes, r.Context() is cancelled. Any database queries tied to that context fail. A long-running import that is halfway through inserting rows will abort.

Memory. Reading a 200MB CSV file into memory in a request handler will spike RSS on every concurrent import. On a t3.medium with 4GB, two simultaneous imports can cause OOM.

The correct architecture: accept the file, enqueue a job, return a job ID to the client, and let the client poll for progress or receive a notification when done.

Job model in PostgreSQL

A minimal but complete bulk job table:

CREATE TYPE bulk_job_status AS ENUM (
    'pending', 'running', 'completed', 'failed', 'cancelled'
);

CREATE TABLE bulk_jobs (
    id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    workspace_id  UUID NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
    type          TEXT NOT NULL,  -- 'import_contacts', 'export_invoices', etc.
    status        bulk_job_status NOT NULL DEFAULT 'pending',
    total_rows    INTEGER,
    processed_rows INTEGER NOT NULL DEFAULT 0,
    failed_rows   INTEGER NOT NULL DEFAULT 0,
    error_summary JSONB,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    started_at    TIMESTAMPTZ,
    completed_at  TIMESTAMPTZ,
    created_by    UUID NOT NULL REFERENCES accounts(id)
);

CREATE INDEX ON bulk_jobs (workspace_id, created_at DESC);
CREATE INDEX ON bulk_jobs (status) WHERE status IN ('pending', 'running');

error_summary stores a JSONB array of row-level errors so users can see which rows failed and why, without reading a separate error log file.

The HTTP handler: accept, store, enqueue

The handler is intentionally thin:

func (h *BulkHandler) ImportContacts(w http.ResponseWriter, r *http.Request) {
    workspaceID := getWorkspaceID(r)
    accountID := getAccountID(r)

    // Parse multipart form with memory limit
    if err := r.ParseMultipartForm(10 << 20); err != nil { // 10MB in memory
        http.Error(w, "file too large for direct parse", http.StatusBadRequest)
        return
    }

    file, header, err := r.FormFile("file")
    if err != nil {
        http.Error(w, "missing file", http.StatusBadRequest)
        return
    }
    defer file.Close()

    // Stream upload to object storage (S3/R2)
    fileKey := fmt.Sprintf("imports/%s/%s.csv", workspaceID, uuid.New())
    if err := h.storage.Upload(r.Context(), fileKey, file); err != nil {
        http.Error(w, "upload failed", http.StatusInternalServerError)
        return
    }

    // Create job record
    jobID, err := h.createJob(r.Context(), workspaceID, accountID, "import_contacts", fileKey)
    if err != nil {
        http.Error(w, "could not create job", http.StatusInternalServerError)
        return
    }

    // Enqueue for background processing
    h.queue.Enqueue(r.Context(), "bulk_import", map[string]string{
        "job_id":  jobID,
        "file_key": fileKey,
    })

    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]string{"job_id": jobID})
}

The file never touches the request body again after it is streamed to object storage. The handler returns in under a second. The client receives a job ID and starts polling.

The worker: chunked processing

The worker downloads the file from storage and processes it in chunks:

const chunkSize = 500

func (w *ImportWorker) ProcessImport(ctx context.Context, jobID, fileKey string) error {
    // Mark job as running
    if err := w.jobs.Start(ctx, jobID); err != nil {
        return err
    }

    // Download file from storage to temp file
    tmpFile, err := downloadToTemp(ctx, w.storage, fileKey)
    if err != nil {
        return w.jobs.Fail(ctx, jobID, "download failed: "+err.Error())
    }
    defer os.Remove(tmpFile.Name())

    // Count total rows first (for progress percentage)
    totalRows, err := countCSVRows(tmpFile)
    if err != nil {
        return w.jobs.Fail(ctx, jobID, "could not count rows")
    }
    w.jobs.SetTotal(ctx, jobID, totalRows)

    // Reset file to beginning
    tmpFile.Seek(0, io.SeekStart)

    reader := csv.NewReader(tmpFile)
    reader.Read() // skip header

    var (
        processed int
        errors    []RowError
        batch     []ContactRow
    )

    for {
        record, err := reader.Read()
        if err == io.EOF {
            break
        }
        if err != nil {
            errors = append(errors, RowError{Row: processed + 1, Message: err.Error()})
            processed++
            continue
        }

        row, parseErr := parseContactRow(record)
        if parseErr != nil {
            errors = append(errors, RowError{Row: processed + 1, Message: parseErr.Error()})
            processed++
            continue
        }

        batch = append(batch, row)
        processed++

        if len(batch) >= chunkSize {
            if insertErr := w.insertBatch(ctx, batch); insertErr != nil {
                // Log batch-level error
                errors = append(errors, RowError{Row: processed, Message: "batch insert failed"})
            }
            batch = batch[:0]

            // Update progress every chunk
            w.jobs.SetProgress(ctx, jobID, processed, len(errors))

            // Check for cancellation
            select {
            case <-ctx.Done():
                return w.jobs.Fail(ctx, jobID, "cancelled")
            default:
            }
        }
    }

    // Final batch
    if len(batch) > 0 {
        w.insertBatch(ctx, batch)
    }

    return w.jobs.Complete(ctx, jobID, processed, errors)
}

A few important choices here:

Process in chunks of 500, not row by row. Row-by-row inserts are dramatically slower due to per-statement overhead in PostgreSQL. Chunks of 500 use a single INSERT ... VALUES with 500 value groups.

Update progress every chunk. The client polling /jobs/{id} sees a meaningful update every 500 rows, not just at the start and end.

Check for cancellation in the loop. If the user cancels the job through the UI, the worker checks the context and exits cleanly. The job record is marked as cancelled, not completed.

Batch insert with PostgreSQL unnest

For inserting a batch of rows efficiently, use the unnest approach rather than constructing a giant VALUES clause:

func (w *ImportWorker) insertBatch(ctx context.Context, rows []ContactRow) error {
    emails := make([]string, len(rows))
    names  := make([]string, len(rows))
    phones := make([]string, len(rows))
    for i, r := range rows {
        emails[i] = r.Email
        names[i]  = r.Name
        phones[i] = r.Phone
    }

    _, err := w.db.ExecContext(ctx, `
        INSERT INTO contacts (workspace_id, email, name, phone)
        SELECT $1, u.email, u.name, u.phone
        FROM unnest($2::text[], $3::text[], $4::text[])
             AS u(email, name, phone)
        ON CONFLICT (workspace_id, email) DO UPDATE
            SET name = EXCLUDED.name, phone = EXCLUDED.phone`,
        w.workspaceID, pq.Array(emails), pq.Array(names), pq.Array(phones),
    )
    return err
}

This sends one query per batch instead of one per row. For 500 rows, the difference is roughly 2ms versus 1000ms.

Exports: streaming large datasets

Exports have a different memory problem. You cannot load a 200,000-row query result into memory, convert it to CSV, and write it to a response. Use a database cursor or LIMIT/OFFSET with streaming:

func (w *ExportWorker) ExportInvoices(ctx context.Context, jobID string, opts ExportOpts) error {
    tmpFile, _ := os.CreateTemp("", "export-*.csv")
    defer os.Remove(tmpFile.Name())

    writer := csv.NewWriter(tmpFile)
    writer.Write([]string{"id", "date", "amount", "customer"})

    const pageSize = 1000
    offset := 0

    for {
        rows, err := w.db.QueryContext(ctx,
            `SELECT id, created_at, amount, customer_name
             FROM invoices
             WHERE workspace_id = $1
             ORDER BY created_at DESC
             LIMIT $2 OFFSET $3`,
            opts.WorkspaceID, pageSize, offset,
        )
        if err != nil {
            return err
        }

        count := 0
        for rows.Next() {
            var inv Invoice
            rows.Scan(&inv.ID, &inv.CreatedAt, &inv.Amount, &inv.Customer)
            writer.Write([]string{inv.ID, inv.CreatedAt.Format(time.RFC3339), fmt.Sprintf("%.2f", inv.Amount), inv.Customer})
            count++
        }
        rows.Close()

        offset += pageSize
        if count < pageSize {
            break
        }

        w.jobs.SetProgress(ctx, jobID, offset, 0)
    }

    writer.Flush()

    // Upload completed file to storage
    downloadKey := fmt.Sprintf("exports/%s/%s.csv", opts.WorkspaceID, jobID)
    tmpFile.Seek(0, io.SeekStart)
    w.storage.Upload(ctx, downloadKey, tmpFile)

    // Generate a presigned URL valid for 1 hour
    url := w.storage.PresignedURL(ctx, downloadKey, time.Hour)
    return w.jobs.CompleteWithURL(ctx, jobID, url)
}

The final job record contains the presigned download URL. The client receives it when polling and shows a download button.

Progress polling and notification

Clients poll GET /bulk-jobs/{id} which returns:

{
  "id": "...",
  "status": "running",
  "total_rows": 12400,
  "processed_rows": 6500,
  "failed_rows": 12,
  "percent": 52
}

Poll every 2 to 3 seconds while status is pending or running. Stop polling when status is completed, failed, or cancelled.

For products where users care about being notified without keeping the tab open, push a notification via WebSocket or send an email when the job completes. The job's completed_at timestamp triggers the notification logic in the worker before it exits.

Key lessons from production

Bulk operations must never run inside HTTP handlers. Stream uploads to object storage immediately. Process in chunks of 500 using unnest inserts. Update progress every chunk. Check for cancellation signals in the processing loop. Stream exports page-by-page to control memory. Return presigned download URLs rather than serving files directly.