构建异步数据导出服务

项目概述

本项目实现异步数据导出服务：请求创建导出任务并立即返回任务 id，工作进程以游标流式写出 CSV 到对象存储而不整表载入内存，完成后签发短时私有下载链接。两个核心约束是内存与隐私：流式让百万行导出不至于变成百万行堆占用，签名链接让私有数据集不至于成为公开下载。

架构与关键决策

任务行是导出的唯一事实来源，保存状态、进度、产物键与过期时间，使客户端、运营者与清理任务读到同一份状态。工作进程使用服务端游标，峰值内存与行数无关；进度按周期更新而非逐行写入，避免把进度写变成瓶颈。完成时把对象置于私有键下并签发短时效签名 URL，签名即能力，过期即真正的访问控制。

实施与测试

导出超过单批次规模的数据集断言行数与源一致且峰值内存有界；断言过期任务拒绝下载、被篡改签名返回 403；断言导出失败会把错误记在任务上而非永久挂起；并以含逗号、引号与换行的夹具验证 CSV 转义。

限制与后续

• 导出为单遍顺序执行，无并行分片读取。
• 产物仅 CSV，列集固定不可自定义。
• 无完成回调，客户端轮询任务状态。

代码示例

pnpm init
pnpm add pg @aws-sdk/client-s3 express
pnpm add -D typescript tsx vitest

data-export/
├── src/enqueue.ts
├── src/worker.ts
├── src/csv-stream.ts
├── src/download.ts
├── tests/export.test.ts
└── package.json

{ id, filter, requestedBy, status: 'queued', progress: 0 }

const q = new QueryStream('SELECT * FROM orders WHERE ...');
await pipeline(q, new CsvTransform(columns), upload);

if (written % 1000 === 0) await db.setProgress(jobId, written / total);

const url = await s3.getSignedUrl('getObject', { Bucket, Key, Expires: 3600 });

await s3.deleteObject({ Bucket, Key }); await db.markExpired(jobId);

pnpm vitest run