Build a Data Export Pipeline with Streaming

// src/exports/export-engine.ts — Stream large datasets without memory pressure
import { pipeline, Transform } from "node:stream";
import { promisify } from "node:util";
import { createWriteStream, createReadStream } from "node:fs";
import { createGzip } from "node:zlib";
import { pool } from "../db";
import { Redis } from "ioredis";
import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3";
import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
import QueryStream from "pg-query-stream";

const pipelineAsync = promisify(pipeline);
const redis = new Redis(process.env.REDIS_URL!);
const s3 = new S3Client({ region: process.env.AWS_REGION || "us-east-1" });

interface ExportJob {
  id: string;
  customerId: string;
  query: string;
  params: any[];
  format: "csv" | "json" | "jsonl";
  compress: boolean;
  status: "pending" | "running" | "completed" | "failed";
  totalRows: number;
  processedRows: number;
  fileSizeBytes: number;
  downloadUrl: string | null;
}

// Start an export job
export async function startExport(job: ExportJob): Promise<void> {
  const client = await pool.connect();

  try {
    await updateExportStatus(job.id, "running", 0);

    // Count total rows for progress tracking
    const countQuery = `SELECT COUNT(*) as total FROM (${job.query}) sub`;
    const { rows: [{ total }] } = await client.query(countQuery, job.params);
    const totalRows = parseInt(total);
    await redis.hset(`export:${job.id}`, "totalRows", totalRows);

    // Stream query results using pg-query-stream (cursor-based, low memory)
    const queryStream = new QueryStream(job.query, job.params, {
      batchSize: 1000,  // fetch 1000 rows at a time from Postgres
    });

    const dbStream = client.query(queryStream);

    // Transform rows to output format
    let processedRows = 0;
    const formatter = new Transform({
      objectMode: true,
      transform(row, _, callback) {
        processedRows++;

        // Update progress every 5000 rows
        if (processedRows % 5000 === 0) {
          redis.hset(`export:${job.id}`, "processedRows", processedRows).catch(() => {});
        }

        let output: string;
        if (job.format === "csv") {
          if (processedRows === 1) {
            // Header row
            output = Object.keys(row).join(",") + "\n";
            output += formatCsvRow(row) + "\n";
          } else {
            output = formatCsvRow(row) + "\n";
          }
        } else if (job.format === "jsonl") {
          output = JSON.stringify(row) + "\n";
        } else {
          // JSON array format
          const prefix = processedRows === 1 ? "[\n" : "";
          const suffix = processedRows === totalRows ? "\n]" : ",\n";
          output = prefix + JSON.stringify(row) + suffix;
        }

        callback(null, output);
      },
    });

    // Output: local file → compress → upload to S3
    const localPath = `/tmp/export-${job.id}.${job.format}${job.compress ? ".gz" : ""}`;
    const fileStream = createWriteStream(localPath);

    const streams: any[] = [dbStream, formatter];
    if (job.compress) streams.push(createGzip());
    streams.push(fileStream);

    await pipelineAsync(...streams);

    // Upload to S3
    const s3Key = `exports/${job.customerId}/${job.id}.${job.format}${job.compress ? ".gz" : ""}`;

    await s3.send(new PutObjectCommand({
      Bucket: process.env.EXPORT_BUCKET!,
      Key: s3Key,
      Body: createReadStream(localPath),
      ContentType: job.format === "csv" ? "text/csv" : "application/json",
      ContentEncoding: job.compress ? "gzip" : undefined,
    }));

    // Generate signed download URL (expires in 7 days)
    const downloadUrl = await getSignedUrl(s3, new PutObjectCommand({
      Bucket: process.env.EXPORT_BUCKET!,
      Key: s3Key,
    }), { expiresIn: 604800 });

    // Update job status
    const { size } = await import("node:fs").then((fs) => fs.promises.stat(localPath));
    await pool.query(
      `UPDATE export_jobs SET status = 'completed', processed_rows = $2, total_rows = $3,
       file_size_bytes = $4, download_url = $5, completed_at = NOW()
       WHERE id = $1`,
      [job.id, processedRows, totalRows, size, downloadUrl]
    );

    await redis.hset(`export:${job.id}`, "status", "completed", "downloadUrl", downloadUrl);

    // Clean up local file
    await import("node:fs").then((fs) => fs.promises.unlink(localPath));
  } catch (err: any) {
    await pool.query(
      "UPDATE export_jobs SET status = 'failed', error = $2 WHERE id = $1",
      [job.id, err.message]
    );
    await redis.hset(`export:${job.id}`, "status", "failed", "error", err.message);
    throw err;
  } finally {
    client.release();
  }
}

function formatCsvRow(row: Record<string, any>): string {
  return Object.values(row)
    .map((v) => {
      if (v === null || v === undefined) return "";
      const str = String(v);
      if (str.includes(",") || str.includes('"') || str.includes("\n")) {
        return `"${str.replace(/"/g, '""')}"`;
      }
      return str;
    })
    .join(",");
}

async function updateExportStatus(jobId: string, status: string, processedRows: number): Promise<void> {
  await pool.query("UPDATE export_jobs SET status = $2, processed_rows = $3 WHERE id = $1", [jobId, status, processedRows]);
  await redis.hset(`export:${jobId}`, "status", status, "processedRows", processedRows);
}

// Check export progress
export async function getExportProgress(jobId: string): Promise<{
  status: string;
  processedRows: number;
  totalRows: number;
  percentComplete: number;
  downloadUrl: string | null;
}> {
  const data = await redis.hgetall(`export:${jobId}`);
  const processed = parseInt(data.processedRows || "0");
  const total = parseInt(data.totalRows || "0");

  return {
    status: data.status || "pending",
    processedRows: processed,
    totalRows: total,
    percentComplete: total > 0 ? Math.round((processed / total) * 100) : 0,
    downloadUrl: data.downloadUrl || null,
  };
}