Building a SaaS Data Export Pipeline: Async Jobs, Streaming CSV/JSON, and S3 Signed URLs

Every SaaS product eventually needs a data export feature. Users want their contacts, orders, analytics — whatever data they've accumulated — in a spreadsheet they can open in Excel. Simple enough in theory. In practice, exporting 500,000 rows synchronously in an HTTP request will time out, eat memory, and bring your server to its knees.

The right approach is an async export pipeline: queue the job, stream the data to S3 as it's generated, then send the user a time-limited download link. This guide builds that system end to end.

Architecture Overview

User clicks "Export" →
  POST /api/exports (creates job, returns jobId) →
    BullMQ worker picks up job →
      Streams query results → CSV/JSON →
        Uploads to S3 multipart →
          Updates job status in DB →
            Emails signed URL to user

The user sees immediate feedback ("Export queued — you'll receive an email when ready") and the server never blocks waiting for a large query to complete.

Database Schema

-- export_jobs table
CREATE TYPE export_status AS ENUM (
  'pending', 'processing', 'completed', 'failed', 'expired'
);

CREATE TYPE export_format AS ENUM ('csv', 'json', 'xlsx');

CREATE TABLE export_jobs (
  id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id    UUID NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  user_id         UUID NOT NULL REFERENCES users(id),
  entity_type     TEXT NOT NULL,          -- 'contacts', 'orders', 'transactions'
  format          export_format NOT NULL DEFAULT 'csv',
  filters         JSONB NOT NULL DEFAULT '{}',
  status          export_status NOT NULL DEFAULT 'pending',
  total_rows      INTEGER,
  processed_rows  INTEGER NOT NULL DEFAULT 0,
  s3_key          TEXT,
  s3_bucket       TEXT,
  download_url    TEXT,
  expires_at      TIMESTAMPTZ,
  error_message   TEXT,
  started_at      TIMESTAMPTZ,
  completed_at    TIMESTAMPTZ,
  created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_export_jobs_workspace ON export_jobs(workspace_id, created_at DESC);
CREATE INDEX idx_export_jobs_user ON export_jobs(user_id, created_at DESC);
CREATE INDEX idx_export_jobs_status ON export_jobs(status) WHERE status IN ('pending', 'processing');

Prisma Schema

model ExportJob {
  id             String        @id @default(uuid())
  workspaceId    String
  userId         String
  entityType     String
  format         ExportFormat  @default(CSV)
  filters        Json          @default("{}")
  status         ExportStatus  @default(PENDING)
  totalRows      Int?
  processedRows  Int           @default(0)
  s3Key          String?
  s3Bucket       String?
  downloadUrl    String?
  expiresAt      DateTime?
  errorMessage   String?
  startedAt      DateTime?
  completedAt    DateTime?
  createdAt      DateTime      @default(now())
  updatedAt      DateTime      @updatedAt

  workspace Workspace @relation(fields: [workspaceId], references: [id], onDelete: Cascade)
  user      User      @relation(fields: [userId], references: [id])

  @@index([workspaceId, createdAt(sort: Desc)])
  @@index([userId, createdAt(sort: Desc)])
}

enum ExportStatus {
  PENDING
  PROCESSING
  COMPLETED
  FAILED
  EXPIRED
}

enum ExportFormat {
  CSV
  JSON
  XLSX
}

API Route: Queue the Export Job

// app/api/exports/route.ts
import { NextRequest, NextResponse } from "next/server";
import { z } from "zod";
import { auth } from "@/auth";
import { prisma } from "@/lib/prisma";
import { exportQueue } from "@/lib/queues/export-queue";

const ExportRequestSchema = z.object({
  entityType: z.enum(["contacts", "orders", "transactions", "users"]),
  format: z.enum(["csv", "json"]).default("csv"),
  filters: z
    .object({
      dateFrom: z.string().datetime().optional(),
      dateTo: z.string().datetime().optional(),
      status: z.string().optional(),
      tags: z.array(z.string()).optional(),
    })
    .default({}),
});

export async function POST(req: NextRequest) {
  const session = await auth();
  if (!session?.user) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  const body = await req.json();
  const parsed = ExportRequestSchema.safeParse(body);
  if (!parsed.success) {
    return NextResponse.json(
      { error: parsed.error.issues[0].message },
      { status: 400 }
    );
  }

  // Rate limit: max 3 active exports per workspace
  const activeExports = await prisma.exportJob.count({
    where: {
      workspaceId: session.user.organizationId,
      status: { in: ["PENDING", "PROCESSING"] },
    },
  });

  if (activeExports >= 3) {
    return NextResponse.json(
      { error: "Too many active exports. Wait for existing exports to complete." },
      { status: 429 }
    );
  }

  const job = await prisma.exportJob.create({
    data: {
      workspaceId: session.user.organizationId,
      userId: session.user.id,
      entityType: parsed.data.entityType,
      format: parsed.data.format.toUpperCase() as "CSV" | "JSON",
      filters: parsed.data.filters,
      status: "PENDING",
    },
  });

  // Enqueue the export worker
  await exportQueue.add(
    "export",
    { jobId: job.id },
    {
      jobId: job.id,
      attempts: 3,
      backoff: { type: "exponential", delay: 5000 },
    }
  );

  return NextResponse.json({ jobId: job.id, status: "pending" }, { status: 202 });
}

export async function GET(req: NextRequest) {
  const session = await auth();
  if (!session?.user) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  const jobs = await prisma.exportJob.findMany({
    where: { workspaceId: session.user.organizationId },
    orderBy: { createdAt: "desc" },
    take: 20,
    select: {
      id: true,
      entityType: true,
      format: true,
      status: true,
      totalRows: true,
      processedRows: true,
      downloadUrl: true,
      expiresAt: true,
      errorMessage: true,
      createdAt: true,
      completedAt: true,
    },
  });

  return NextResponse.json({ jobs });
}

BullMQ Queue Setup

// lib/queues/export-queue.ts
import { Queue } from "bullmq";
import { redis } from "@/lib/redis";

export const exportQueue = new Queue("exports", {
  connection: redis,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: "exponential", delay: 5000 },
    removeOnComplete: { age: 86400 }, // keep 24h for debugging
    removeOnFail: { age: 604800 },    // keep 7d for inspection
  },
});

Export Worker: Streaming Data to S3

// workers/export-worker.ts
import { Worker, Job } from "bullmq";
import { Readable, Transform } from "stream";
import { pipeline } from "stream/promises";
import {
  S3Client,
  CreateMultipartUploadCommand,
  UploadPartCommand,
  CompleteMultipartUploadCommand,
  GetObjectCommand,
} from "@aws-sdk/client-s3";
import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
import { stringify } from "csv-stringify";
import { prisma } from "@/lib/prisma";
import { redis } from "@/lib/redis";
import { sendExportReadyEmail } from "@/lib/email";

const s3 = new S3Client({ region: process.env.AWS_REGION });
const BUCKET = process.env.EXPORTS_BUCKET!;
const SIGNED_URL_TTL = 24 * 60 * 60; // 24 hours

interface ExportJobData {
  jobId: string;
}

export const exportWorker = new Worker<ExportJobData>(
  "exports",
  async (job: Job<ExportJobData>) => {
    const { jobId } = job.data;

    const exportJob = await prisma.exportJob.update({
      where: { id: jobId },
      data: { status: "PROCESSING", startedAt: new Date() },
      include: { user: { select: { email: true, name: true } }, workspace: true },
    });

    try {
      const s3Key = `exports/${exportJob.workspaceId}/${jobId}.${exportJob.format.toLowerCase()}`;

      // Count total rows first (for progress tracking)
      const totalRows = await countRows(exportJob);
      await prisma.exportJob.update({
        where: { id: jobId },
        data: { totalRows },
      });

      // Stream data → S3 multipart upload
      await streamToS3(exportJob, s3Key, jobId, totalRows);

      // Generate signed download URL
      const downloadUrl = await getSignedUrl(
        s3,
        new GetObjectCommand({
          Bucket: BUCKET,
          Key: s3Key,
          ResponseContentDisposition: `attachment; filename="${exportJob.entityType}-export-${jobId}.${exportJob.format.toLowerCase()}"`,
        }),
        { expiresIn: SIGNED_URL_TTL }
      );

      const expiresAt = new Date(Date.now() + SIGNED_URL_TTL * 1000);

      await prisma.exportJob.update({
        where: { id: jobId },
        data: {
          status: "COMPLETED",
          s3Key,
          s3Bucket: BUCKET,
          downloadUrl,
          expiresAt,
          completedAt: new Date(),
          processedRows: totalRows,
        },
      });

      // Notify user
      await sendExportReadyEmail({
        to: exportJob.user.email,
        name: exportJob.user.name ?? "there",
        entityType: exportJob.entityType,
        rowCount: totalRows,
        downloadUrl,
        expiresAt,
      });

      // Publish progress event for SSE listeners
      await redis.publish(
        `export:${jobId}`,
        JSON.stringify({ status: "completed", processedRows: totalRows, totalRows, downloadUrl })
      );
    } catch (error) {
      const message = error instanceof Error ? error.message : "Unknown error";
      await prisma.exportJob.update({
        where: { id: jobId },
        data: { status: "FAILED", errorMessage: message },
      });

      await redis.publish(
        `export:${jobId}`,
        JSON.stringify({ status: "failed", error: message })
      );

      throw error; // Let BullMQ handle retry
    }
  },
  { connection: redis, concurrency: 3 }
);

async function countRows(exportJob: any): Promise<number> {
  const filters = exportJob.filters as Record<string, any>;
  const where = buildWhereClause(exportJob.workspaceId, exportJob.entityType, filters);

  switch (exportJob.entityType) {
    case "contacts":
      return prisma.contact.count({ where });
    case "orders":
      return prisma.order.count({ where });
    default:
      return 0;
  }
}

async function streamToS3(
  exportJob: any,
  s3Key: string,
  jobId: string,
  totalRows: number
): Promise<void> {
  // Initiate multipart upload
  const { UploadId } = await s3.send(
    new CreateMultipartUploadCommand({
      Bucket: BUCKET,
      Key: s3Key,
      ContentType:
        exportJob.format === "CSV" ? "text/csv" : "application/json",
      ServerSideEncryption: "AES256",
      Metadata: {
        workspaceId: exportJob.workspaceId,
        jobId,
        entityType: exportJob.entityType,
      },
    })
  );

  const parts: { ETag: string; PartNumber: number }[] = [];
  let partNumber = 1;
  let buffer = Buffer.alloc(0);
  const MIN_PART_SIZE = 5 * 1024 * 1024; // 5MB minimum for multipart
  let processedRows = 0;

  const flushPart = async (data: Buffer, isFinal = false) => {
    if (data.length === 0) return;
    const { ETag } = await s3.send(
      new UploadPartCommand({
        Bucket: BUCKET,
        Key: s3Key,
        UploadId,
        PartNumber: partNumber,
        Body: data,
      })
    );
    parts.push({ ETag: ETag!, PartNumber: partNumber });
    partNumber++;
  };

  // Stream rows from database using cursor-based pagination
  const BATCH_SIZE = 500;
  let cursor: string | undefined;

  // Write CSV header if needed
  if (exportJob.format === "CSV") {
    const header = getCSVHeader(exportJob.entityType);
    buffer = Buffer.concat([buffer, Buffer.from(header + "\n")]);
  }

  let isFirstJsonRow = true;
  if (exportJob.format === "JSON") {
    buffer = Buffer.concat([buffer, Buffer.from("[\n")]);
  }

  while (true) {
    const rows = await fetchBatch(
      exportJob.workspaceId,
      exportJob.entityType,
      exportJob.filters,
      BATCH_SIZE,
      cursor
    );

    if (rows.length === 0) break;

    for (const row of rows) {
      let rowStr: string;
      if (exportJob.format === "CSV") {
        rowStr = formatCSVRow(exportJob.entityType, row) + "\n";
      } else {
        rowStr =
          (isFirstJsonRow ? "  " : ",\n  ") + JSON.stringify(row);
        isFirstJsonRow = false;
      }
      buffer = Buffer.concat([buffer, Buffer.from(rowStr)]);
      processedRows++;
    }

    // Flush part if buffer is large enough
    if (buffer.length >= MIN_PART_SIZE) {
      await flushPart(buffer);
      buffer = Buffer.alloc(0);
    }

    // Update progress every batch
    await prisma.exportJob.update({
      where: { id: jobId },
      data: { processedRows },
    });

    await redis.publish(
      `export:${jobId}`,
      JSON.stringify({
        status: "processing",
        processedRows,
        totalRows,
        pct: Math.round((processedRows / totalRows) * 100),
      })
    );

    cursor = rows[rows.length - 1].id;
    if (rows.length < BATCH_SIZE) break;
  }

  // Close JSON array
  if (exportJob.format === "JSON") {
    buffer = Buffer.concat([buffer, Buffer.from("\n]")]);
  }

  // Flush remaining buffer as final part
  await flushPart(buffer, true);

  // Complete multipart upload
  await s3.send(
    new CompleteMultipartUploadCommand({
      Bucket: BUCKET,
      Key: s3Key,
      UploadId,
      MultipartUpload: { Parts: parts },
    })
  );
}

async function fetchBatch(
  workspaceId: string,
  entityType: string,
  filters: Record<string, any>,
  limit: number,
  cursor?: string
): Promise<any[]> {
  const where = buildWhereClause(workspaceId, entityType, filters);
  const cursorClause = cursor ? { id: { gt: cursor } } : {};

  switch (entityType) {
    case "contacts":
      return prisma.contact.findMany({
        where: { ...where, ...cursorClause },
        orderBy: { id: "asc" },
        take: limit,
        select: {
          id: true,
          firstName: true,
          lastName: true,
          email: true,
          phone: true,
          company: true,
          status: true,
          tags: true,
          createdAt: true,
        },
      });
    case "orders":
      return prisma.order.findMany({
        where: { ...where, ...cursorClause },
        orderBy: { id: "asc" },
        take: limit,
        select: {
          id: true,
          orderNumber: true,
          status: true,
          totalAmount: true,
          currency: true,
          createdAt: true,
        },
      });
    default:
      return [];
  }
}

function buildWhereClause(
  workspaceId: string,
  entityType: string,
  filters: Record<string, any>
) {
  const base = { workspaceId };
  const dateFilter =
    filters.dateFrom || filters.dateTo
      ? {
          createdAt: {
            ...(filters.dateFrom && { gte: new Date(filters.dateFrom) }),
            ...(filters.dateTo && { lte: new Date(filters.dateTo) }),
          },
        }
      : {};
  const statusFilter = filters.status ? { status: filters.status } : {};

  return { ...base, ...dateFilter, ...statusFilter };
}

function getCSVHeader(entityType: string): string {
  const headers: Record<string, string> = {
    contacts: "id,first_name,last_name,email,phone,company,status,tags,created_at",
    orders: "id,order_number,status,total_amount,currency,created_at",
  };
  return headers[entityType] ?? "id,created_at";
}

function formatCSVRow(entityType: string, row: any): string {
  const escape = (v: any) => {
    if (v === null || v === undefined) return "";
    const str = Array.isArray(v) ? v.join(";") : String(v);
    return str.includes(",") || str.includes('"') || str.includes("\n")
      ? `"${str.replace(/"/g, '""')}"`
      : str;
  };

  switch (entityType) {
    case "contacts":
      return [
        row.id,
        escape(row.firstName),
        escape(row.lastName),
        escape(row.email),
        escape(row.phone),
        escape(row.company),
        row.status,
        escape(row.tags),
        row.createdAt.toISOString(),
      ].join(",");
    case "orders":
      return [
        row.id,
        escape(row.orderNumber),
        row.status,
        row.totalAmount,
        row.currency,
        row.createdAt.toISOString(),
      ].join(",");
    default:
      return row.id;
  }
}

Real-Time Progress with Server-Sent Events

// app/api/exports/[jobId]/progress/route.ts
import { NextRequest, NextResponse } from "next/server";
import { auth } from "@/auth";
import { prisma } from "@/lib/prisma";
import { redis } from "@/lib/redis";

export async function GET(
  req: NextRequest,
  { params }: { params: { jobId: string } }
) {
  const session = await auth();
  if (!session?.user) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  const job = await prisma.exportJob.findFirst({
    where: {
      id: params.jobId,
      workspaceId: session.user.organizationId,
    },
  });

  if (!job) {
    return NextResponse.json({ error: "Not found" }, { status: 404 });
  }

  // If already terminal, return immediately
  if (job.status === "COMPLETED" || job.status === "FAILED") {
    return NextResponse.json({
      status: job.status.toLowerCase(),
      processedRows: job.processedRows,
      totalRows: job.totalRows,
      downloadUrl: job.downloadUrl,
      errorMessage: job.errorMessage,
    });
  }

  const encoder = new TextEncoder();
  const stream = new ReadableStream({
    async start(controller) {
      const send = (data: object) => {
        controller.enqueue(
          encoder.encode(`data: ${JSON.stringify(data)}\n\n`)
        );
      };

      // Subscribe to Redis pub/sub for live updates
      const subscriber = redis.duplicate();
      await subscriber.subscribe(`export:${params.jobId}`);

      subscriber.on("message", (_channel, message) => {
        const parsed = JSON.parse(message);
        send(parsed);

        if (parsed.status === "completed" || parsed.status === "failed") {
          subscriber.unsubscribe();
          subscriber.quit();
          controller.close();
        }
      });

      // Poll DB as fallback every 3s (in case Redis message was missed)
      const pollInterval = setInterval(async () => {
        const current = await prisma.exportJob.findUnique({
          where: { id: params.jobId },
          select: {
            status: true,
            processedRows: true,
            totalRows: true,
            downloadUrl: true,
            errorMessage: true,
          },
        });

        if (!current) return;

        send({
          status: current.status.toLowerCase(),
          processedRows: current.processedRows,
          totalRows: current.totalRows ?? 0,
          pct: current.totalRows
            ? Math.round((current.processedRows / current.totalRows) * 100)
            : 0,
          downloadUrl: current.downloadUrl,
        });

        if (
          current.status === "COMPLETED" ||
          current.status === "FAILED"
        ) {
          clearInterval(pollInterval);
          subscriber.unsubscribe().catch(() => {});
          subscriber.quit().catch(() => {});
          controller.close();
        }
      }, 3000);

      // Cleanup on client disconnect
      req.signal.addEventListener("abort", () => {
        clearInterval(pollInterval);
        subscriber.unsubscribe().catch(() => {});
        subscriber.quit().catch(() => {});
        controller.close();
      });
    },
  });

  return new NextResponse(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    },
  });
}

Frontend: Export UI Component

// components/exports/export-dialog.tsx
"use client";

import { useState, useEffect, useRef } from "react";
import { Download, Loader2, CheckCircle, XCircle } from "lucide-react";

interface ExportProgress {
  status: "pending" | "processing" | "completed" | "failed";
  processedRows: number;
  totalRows: number;
  pct: number;
  downloadUrl?: string;
  errorMessage?: string;
}

interface ExportDialogProps {
  entityType: "contacts" | "orders" | "transactions";
  onClose: () => void;
}

export function ExportDialog({ entityType, onClose }: ExportDialogProps) {
  const [format, setFormat] = useState<"csv" | "json">("csv");
  const [jobId, setJobId] = useState<string | null>(null);
  const [progress, setProgress] = useState<ExportProgress | null>(null);
  const [isStarting, setIsStarting] = useState(false);
  const eventSourceRef = useRef<EventSource | null>(null);

  const startExport = async () => {
    setIsStarting(true);
    try {
      const res = await fetch("/api/exports", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ entityType, format }),
      });

      if (!res.ok) {
        const err = await res.json();
        throw new Error(err.error);
      }

      const { jobId } = await res.json();
      setJobId(jobId);
      setProgress({ status: "pending", processedRows: 0, totalRows: 0, pct: 0 });
    } catch (err) {
      alert(err instanceof Error ? err.message : "Failed to start export");
    } finally {
      setIsStarting(false);
    }
  };

  useEffect(() => {
    if (!jobId) return;

    const es = new EventSource(`/api/exports/${jobId}/progress`);
    eventSourceRef.current = es;

    es.onmessage = (event) => {
      const data = JSON.parse(event.data) as ExportProgress;
      setProgress(data);

      if (data.status === "completed" || data.status === "failed") {
        es.close();
      }
    };

    es.onerror = () => {
      es.close();
    };

    return () => {
      es.close();
    };
  }, [jobId]);

  return (
    <div className="fixed inset-0 bg-black/50 flex items-center justify-center z-50">
      <div className="bg-white rounded-xl p-6 w-full max-w-md shadow-xl">
        <h2 className="text-xl font-semibold mb-4">
          Export {entityType.charAt(0).toUpperCase() + entityType.slice(1)}
        </h2>

        {!jobId ? (
          <>
            <div className="mb-4">
              <label className="block text-sm font-medium text-gray-700 mb-2">
                Format
              </label>
              <div className="flex gap-3">
                {(["csv", "json"] as const).map((f) => (
                  <button
                    key={f}
                    onClick={() => setFormat(f)}
                    className={`flex-1 py-2 border rounded-lg text-sm font-medium transition ${
                      format === f
                        ? "border-blue-600 bg-blue-50 text-blue-700"
                        : "border-gray-200 hover:border-gray-300"
                    }`}
                  >
                    {f.toUpperCase()}
                  </button>
                ))}
              </div>
            </div>

            <div className="flex gap-3 mt-6">
              <button
                onClick={onClose}
                className="flex-1 py-2 border border-gray-200 rounded-lg text-sm hover:bg-gray-50"
              >
                Cancel
              </button>
              <button
                onClick={startExport}
                disabled={isStarting}
                className="flex-1 py-2 bg-blue-600 text-white rounded-lg text-sm font-medium hover:bg-blue-700 disabled:opacity-50 flex items-center justify-center gap-2"
              >
                {isStarting ? (
                  <Loader2 className="w-4 h-4 animate-spin" />
                ) : (
                  <Download className="w-4 h-4" />
                )}
                Start Export
              </button>
            </div>
          </>
        ) : (
          <div className="space-y-4">
            {progress?.status === "completed" ? (
              <div className="text-center py-4">
                <CheckCircle className="w-12 h-12 text-green-500 mx-auto mb-3" />
                <p className="font-medium text-gray-900">Export ready!</p>
                <p className="text-sm text-gray-500 mt-1">
                  {progress.totalRows.toLocaleString()} rows exported
                </p>
                <a
                  href={progress.downloadUrl}
                  download
                  className="mt-4 inline-flex items-center gap-2 px-4 py-2 bg-green-600 text-white rounded-lg text-sm font-medium hover:bg-green-700"
                >
                  <Download className="w-4 h-4" />
                  Download {format.toUpperCase()}
                </a>
                <p className="text-xs text-gray-400 mt-2">
                  Link expires in 24 hours
                </p>
              </div>
            ) : progress?.status === "failed" ? (
              <div className="text-center py-4">
                <XCircle className="w-12 h-12 text-red-500 mx-auto mb-3" />
                <p className="font-medium text-gray-900">Export failed</p>
                <p className="text-sm text-red-600 mt-1">
                  {progress.errorMessage ?? "An unknown error occurred"}
                </p>
              </div>
            ) : (
              <div>
                <div className="flex justify-between text-sm text-gray-600 mb-2">
                  <span>
                    {progress?.status === "pending"
                      ? "Queued..."
                      : "Exporting..."}
                  </span>
                  <span>{progress?.pct ?? 0}%</span>
                </div>
                <div className="w-full bg-gray-100 rounded-full h-2">
                  <div
                    className="bg-blue-600 h-2 rounded-full transition-all duration-500"
                    style={{ width: `${progress?.pct ?? 0}%` }}
                  />
                </div>
                <p className="text-xs text-gray-500 mt-2 text-center">
                  {(progress?.processedRows ?? 0).toLocaleString()} /{" "}
                  {(progress?.totalRows ?? 0).toLocaleString()} rows
                </p>
                <p className="text-xs text-gray-400 text-center mt-1">
                  You'll also receive an email when the export is ready.
                </p>
              </div>
            )}

            <button
              onClick={onClose}
              className="w-full py-2 border border-gray-200 rounded-lg text-sm hover:bg-gray-50"
            >
              Close
            </button>
          </div>
        )}
      </div>
    </div>
  );
}

Expiry Cleanup Cron Job

// workers/export-cleanup-worker.ts
import { CronJob } from "cron";
import { S3Client, DeleteObjectCommand } from "@aws-sdk/client-s3";
import { prisma } from "@/lib/prisma";

const s3 = new S3Client({ region: process.env.AWS_REGION });

export const exportCleanupJob = new CronJob(
  "0 2 * * *", // 2 AM daily
  async () => {
    console.log("Running export cleanup...");

    // Mark expired exports
    const expired = await prisma.exportJob.findMany({
      where: {
        status: "COMPLETED",
        expiresAt: { lt: new Date() },
      },
      select: { id: true, s3Key: true, s3Bucket: true },
    });

    for (const job of expired) {
      try {
        if (job.s3Key && job.s3Bucket) {
          await s3.send(
            new DeleteObjectCommand({
              Bucket: job.s3Bucket,
              Key: job.s3Key,
            })
          );
        }

        await prisma.exportJob.update({
          where: { id: job.id },
          data: {
            status: "EXPIRED",
            downloadUrl: null,
            s3Key: null,
          },
        });
      } catch (err) {
        console.error(`Failed to cleanup export ${job.id}:`, err);
      }
    }

    console.log(`Cleaned up ${expired.length} expired exports`);
  },
  null,
  true,
  "UTC"
);

Cost and Timeline Estimates

AWS S3 running costs (for reference):

• Storage: $0.023/GB/month
• Requests: $0.005 per 1,000 PUT, $0.0004 per 1,000 GET
• Data transfer out: $0.09/GB (first 10TB/month)

A SaaS with 1,000 exports/month averaging 5MB each costs ~$5–10/month in S3 costs.

See Also

• SaaS CSV Import Pipeline with Validation and Progress Tracking
• AWS SQS and SNS Event-Driven Patterns
• SaaS Audit Trail Implementation
• PostgreSQL Row-Level Security for Multi-Tenant Apps
• React Query Infinite Scroll with Cursor Pagination

---

Working With Viprasol

Data export is one of those features that looks simple but has real complexity hiding underneath — streaming large datasets without memory spikes, multipart S3 uploads, progress tracking, signed URL expiry, and cleanup crons. Our team has built export pipelines for SaaS platforms handling millions of records.

What we deliver:

• Async export queue with BullMQ or SQS
• Streaming CSV/JSON/XLSX generation with multipart S3 upload
• Real-time progress via SSE or WebSockets
• Signed URL delivery with configurable expiry
• Automated S3 cleanup and audit logging

Talk to our team about your data export requirements →

Or explore our SaaS development services to see how we approach full-stack product work.