feat: export all project contacts as CSV

CONTRIBUTING.md

@@ -142,3 +142,29 @@ between groups.
 ## License
 
 By contributing to Plunk, you agree that your contributions will be licensed under the AGPL-3.0 License.
+
+## Contact CSV exports
+
+Contact exports use the API and the contact-export BullMQ worker. The worker reads the
+whole authenticated project in batches of 100, snapshots rows to a temporary file,
+and uploads CSV using bounded multipart buffers. Search, table pagination and selection
+do not affect the export. Contacts are evaluated while the worker reads them; concurrent
+contact edits are not a point-in-time database backup.
+
+Standard columns are id, email, subscribed, snoozedUntil, createdAt and updatedAt.
+Custom keys are sorted and prefixed with `data.` to avoid collisions. Null/missing
+values become empty cells, booleans/numbers use their JSON spelling, dates use ISO 8601,
+and nested custom values use JSON. Formula-like strings gain an apostrophe. Exports
+support up to 1000 distinct custom keys and fail rather than silently omit extra fields.
+
+The existing S3 credentials must be allowed to create/manage a separate private bucket
+named `<S3_BUCKET>-contact-exports`, read its policy, configure lifecycle rules, and
+upload/download/delete objects and multipart uploads there. Do not grant anonymous
+access or configure a bucket policy on this bucket: exports fail closed if a policy is
+present. Keep S3 account/bucket ACLs private. Export status and downloads expire
+24 hours after the request. The worker requests one-day object expiry and incomplete-multipart
+abort rules; verify lifecycle support on your storage backend. Handled failures abort
+uploads, but crash-abandoned upload cleanup depends on the backend's support. Downloads
+go through the authenticated API and membership checks; no public URL is returned.
+Workers need temporary disk space for the project snapshot; temporary files are removed
+on success, cancellation or failure.

apps/api/src/controllers/Contacts.ts

@@ -1,3 +1,5 @@
+import {pipeline} from 'node:stream/promises';
+
 import {Controller, Delete, Get, Middleware, Patch, Post} from '@overnightjs/core';
 import type {NextFunction, Request, Response} from 'express';
 import multer from 'multer';
@@ -7,6 +9,8 @@ import signale from 'signale';
 import {requireAuth, requireEmailVerified} from '../middleware/auth.js';
 import {contactWriteRateLimit} from '../middleware/rateLimit.js';
 import {HttpException} from '../exceptions/index.js';
+import {ContactExportService} from '../services/ContactExportService.js';
+import {ContactExportStorage} from '../services/ContactExportStorage.js';
 import {ContactService} from '../services/ContactService.js';
 import {QueueService} from '../services/QueueService.js';
 import {CatchAsync} from '../utils/asyncHandler.js';
@@ -142,6 +146,51 @@ export class Contacts {
     }
   }
 
+  @Post('exports')
+  @Middleware([requireAuth, requireEmailVerified])
+  @CatchAsync
+  public async exportCsv(_req: Request, res: Response) {
+    return res.status(202).json(await ContactExportService.start(res.locals.auth.projectId!));
+  }
+
+  @Get('exports/:jobId')
+  @Middleware([requireAuth, requireEmailVerified])
+  @CatchAsync
+  public async exportStatus(req: Request, res: Response) {
+    res.setHeader('Cache-Control', 'no-store');
+    return res.json(await ContactExportService.status(req.params.jobId!, res.locals.auth.projectId!));
+  }
+
+  @Delete('exports/:jobId')
+  @Middleware([requireAuth, requireEmailVerified])
+  @CatchAsync
+  public async cancelExport(req: Request, res: Response) {
+    await ContactExportService.cancel(req.params.jobId!, res.locals.auth.projectId!);
+    return res.status(204).send();
+  }
+
+  @Get('exports/:jobId/download')
+  @Middleware([exportDownloadAuth, requireEmailVerified])
+  @CatchAsync
+  public async downloadExport(req: Request, res: Response) {
+    const job = await ContactExportService.getJob(req.params.jobId!, res.locals.auth.projectId!);
+    if (job.data.cancelled || (await job.getState()) !== 'completed') {
+      throw new HttpException(409, 'Contact export is not ready');
+    }
+    const file = job.returnvalue;
+    const stream = await ContactExportStorage.download(file.key);
+    res.setHeader('Cache-Control', 'private, no-store');
+    res.setHeader('Content-Type', 'text/csv; charset=utf-8');
+    res.attachment(file.filename);
+    // pipeline propagates backpressure and destroys the S3 stream on disconnect.
+    try {
+      await pipeline(stream, res);
+    } catch (error) {
+      // A disconnected client has no response left for the JSON error middleware.
+      if (!res.destroyed) throw error;
+    }
+  }
+
   /**
    * GET /contacts/:id
    * Get a specific contact by ID
@@ -606,3 +655,17 @@ async function queueBulkAction(
     });
   }
 }
+
+/** Native browser downloads cannot set X-Project-Id. Resolve the locator from the
+ * job, then use the normal authentication AND membership checks for that project.
+ * A job id alone never grants access to a CSV.
+ */
+async function exportDownloadAuth(req: Request, res: Response, next: NextFunction) {
+  try {
+    const job = await ContactExportService.getJob(req.params.jobId!);
+    req.headers['x-project-id'] = job.data.projectId;
+    await requireAuth(req, res, next);
+  } catch (error) {
+    next(error);
+  }
+}

apps/api/src/controllers/__tests__/Contacts.export.test.ts

@@ -0,0 +1,212 @@
+import {once} from 'node:events';
+import {get} from 'node:http';
+import {Readable} from 'node:stream';
+
+import {Server} from '@overnightjs/core';
+import type {ContactExportResult} from '@plunk/types';
+import {QueueEvents} from 'bullmq';
+import cookies from 'cookie-parser';
+import type {NextFunction, Request, Response} from 'express';
+import {json} from 'express';
+import request from 'supertest';
+import {afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi} from 'vitest';
+
+import {factories, getPrismaClient} from '../../../../../test/helpers';
+import {HttpException} from '../../exceptions/index.js';
+import {createContactExportWorker} from '../../jobs/contact-export-processor.js';
+import {jwt} from '../../middleware/auth.js';
+import {ContactExportStorage} from '../../services/ContactExportStorage.js';
+import {ContactExportService} from '../../services/ContactExportService.js';
+import {contactExportQueue} from '../../services/QueueService.js';
+import {Contacts} from '../Contacts.js';
+
+vi.mock('../../app/constants.js', async importOriginal => ({
+  ...(await importOriginal<typeof import('../../app/constants.js')>()),
+  PLUNK_ENABLED: true,
+}));
+
+class TestServer extends Server {
+  constructor() {
+    super();
+    this.app.use(json(), cookies());
+    this.addControllers(new Contacts());
+    this.app.use((error: HttpException, _req: Request, res: Response, _next: NextFunction) => {
+      if (res.headersSent) return res.destroy();
+      res.status(error.code || 500).json({message: error.message});
+    });
+  }
+  get handler() {
+    return this.app;
+  }
+}
+
+const app = new TestServer().handler;
+let worker: ReturnType<typeof createContactExportWorker>;
+let events: QueueEvents;
+let project: Awaited<ReturnType<typeof factories.createProject>>;
+let userId: string;
+const keys: string[] = [];
+
+async function completedExport() {
+  const response = await request(app).post('/contacts/exports').auth(project.secret, {type: 'bearer'}).expect(202);
+  const job = await ContactExportService.getJob(response.body.jobId, project.id);
+  const result = (await job.waitUntilFinished(events, 15000)) as ContactExportResult;
+  keys.push(result.key);
+  return {job, result};
+}
+
+describe('Authenticated contact CSV routes and worker', () => {
+  beforeAll(async () => {
+    worker = createContactExportWorker();
+    events = new QueueEvents(contactExportQueue.name, {connection: contactExportQueue.opts.connection});
+    await events.waitUntilReady();
+  });
+  beforeEach(async () => {
+    const fixture = await factories.createUserWithProject();
+    project = fixture.project;
+    userId = fixture.user.id;
+    await getPrismaClient().user.update({where: {id: userId}, data: {emailVerified: true}});
+  });
+  afterEach(async () => {
+    vi.restoreAllMocks();
+    await Promise.all(keys.splice(0).map(key => ContactExportStorage.remove(key)));
+  });
+  afterAll(async () => {
+    await worker.close();
+    await events.close();
+  });
+
+  it('runs the real queue and downloads the whole project using a browser JWT without a project header', async () => {
+    await factories.createContact({projectId: project.id, email: 'export@example.test', data: {name: '世界'}});
+    const {job, result} = await completedExport();
+    await request(app)
+      .get(`/contacts/exports/${job.id}`)
+      .auth(project.secret, {type: 'bearer'})
+      .expect(200)
+      .expect(({body}) => {
+        expect(body).toMatchObject({status: 'completed', rows: 1});
+        expect(body).not.toHaveProperty('key');
+      });
+    const response = await request(app)
+      .get(`/contacts/exports/${job.id}/download`)
+      .set('Cookie', `next_token=${jwt.sign(userId)}`)
+      .expect(200);
+    expect(response.headers['content-type']).toContain('text/csv');
+    expect(response.headers['content-disposition']).toContain(result.filename);
+    expect(response.headers['cache-control']).toContain('no-store');
+    expect(response.text).toContain('export@example.test');
+    expect(response.text).toContain('世界');
+  });
+
+  it('rejects unauthenticated, invalid-key and public-key creation', async () => {
+    await request(app).post('/contacts/exports').expect(401);
+    await request(app).post('/contacts/exports').auth('sk_invalid', {type: 'bearer'}).expect(401);
+    await request(app).post('/contacts/exports').auth(project.public, {type: 'bearer'}).expect(401);
+  });
+
+  it('requires authentication and project membership at the download boundary', async () => {
+    const {job} = await completedExport();
+    const other = await factories.createUserWithProject();
+    await request(app).get(`/contacts/exports/${job.id}/download`).expect(401);
+    await request(app)
+      .get(`/contacts/exports/${job.id}/download`)
+      .set('Cookie', `next_token=${jwt.sign(other.user.id)}`)
+      .expect(403);
+    await request(app)
+      .get(`/contacts/exports/${job.id}/download`)
+      .auth(other.project.secret, {type: 'bearer'})
+      .expect(404);
+    await request(app).get(`/contacts/exports/${job.id}`).auth(other.project.secret, {type: 'bearer'}).expect(404);
+    await request(app).delete(`/contacts/exports/${job.id}`).auth(other.project.secret, {type: 'bearer'}).expect(404);
+  });
+
+  it('ignores client project ids and filters when queuing an export', async () => {
+    const other = await factories.createProject();
+    await factories.createContact({projectId: project.id, email: 'own@example.test'});
+    await factories.createContact({projectId: other.id, email: 'other@example.test'});
+    const response = await request(app)
+      .post('/contacts/exports?search=missing&status=unsubscribed')
+      .auth(project.secret, {type: 'bearer'})
+      .send({projectId: other.id})
+      .expect(202);
+    const job = await ContactExportService.getJob(response.body.jobId, project.id);
+    const result = (await job.waitUntilFinished(events, 15000)) as ContactExportResult;
+    keys.push(result.key);
+    expect(result.rows).toBe(1);
+    expect(job.data.projectId).toBe(project.id);
+  });
+
+  it('cancels completed exports and blocks future downloads', async () => {
+    const {job, result} = await completedExport();
+    await request(app).delete(`/contacts/exports/${job.id}`).auth(project.secret, {type: 'bearer'}).expect(204);
+    await request(app).get(`/contacts/exports/${job.id}/download`).auth(project.secret, {type: 'bearer'}).expect(409);
+    await expect(ContactExportStorage.download(result.key)).rejects.toThrow();
+  });
+
+  it('rejects an unverified dashboard user at creation and download', async () => {
+    const {job} = await completedExport();
+    await getPrismaClient().user.update({where: {id: userId}, data: {emailVerified: false}});
+    const cookie = `next_token=${jwt.sign(userId)}`;
+    await request(app).post('/contacts/exports').set('Cookie', cookie).set('X-Project-Id', project.id).expect(403);
+    await request(app).get(`/contacts/exports/${job.id}/download`).set('Cookie', cookie).expect(403);
+  });
+
+  it('returns a useful failure without exposing worker details or contact data', async () => {
+    vi.spyOn(ContactExportService, 'process').mockRejectedValueOnce(new Error('private@example.test storage failure'));
+    const response = await request(app).post('/contacts/exports').auth(project.secret, {type: 'bearer'}).expect(202);
+    const job = await ContactExportService.getJob(response.body.jobId, project.id);
+    await expect(job.waitUntilFinished(events, 15000)).rejects.toThrow('storage failure');
+    const status = await request(app)
+      .get(`/contacts/exports/${job.id}`)
+      .auth(project.secret, {type: 'bearer'})
+      .expect(200);
+    expect(status.body).toEqual({status: 'failed', rows: 0, error: 'Could not export contacts. Please try again.'});
+  });
+
+  it('destroys the source stream when the browser interrupts a download', async () => {
+    const {job} = await completedExport();
+    const source = Readable.from(
+      (async function* () {
+        while (true) {
+          yield Buffer.alloc(65536, 'a');
+          await new Promise(resolve => setTimeout(resolve, 5));
+        }
+      })(),
+    );
+    const closed = new Promise<void>(resolve => source.once('close', resolve));
+    vi.spyOn(ContactExportStorage, 'download').mockResolvedValueOnce(source);
+    const listener = app.listen(0, '127.0.0.1');
+    await once(listener, 'listening');
+    const address = listener.address() as {port: number};
+    try {
+      await new Promise<void>((resolve, reject) => {
+        const req = get(
+          `http://127.0.0.1:${address.port}/contacts/exports/${job.id}/download`,
+          {
+            headers: {Authorization: `Bearer ${project.secret}`},
+          },
+          response => {
+            response.once('data', () => {
+              response.destroy();
+              resolve();
+            });
+          },
+        );
+        req.on('error', reject);
+      });
+      await closed;
+      expect(source.destroyed).toBe(true);
+    } finally {
+      listener.close();
+      source.destroy();
+    }
+  });
+
+  it('handles unknown jobs through the static route rather than contact lookup', async () => {
+    const response = await request(app)
+      .get('/contacts/exports/does-not-exist')
+      .auth(project.secret, {type: 'bearer'})
+      .expect(404);
+    expect(response.body.message).toContain('export not found');
+  });
+});

apps/api/src/jobs/contact-export-processor.ts

@@ -0,0 +1,20 @@
+import type {ContactExportJobData, ContactExportResult} from '@plunk/types';
+import {Worker} from 'bullmq';
+import signale from 'signale';
+
+import {ContactExportService} from '../services/ContactExportService.js';
+import {contactExportQueue} from '../services/QueueService.js';
+
+export function createContactExportWorker() {
+  const worker = new Worker<ContactExportJobData, ContactExportResult>(
+    contactExportQueue.name,
+    job => ContactExportService.process(job),
+    {
+      connection: contactExportQueue.opts.connection,
+      concurrency: 1,
+    },
+  );
+  worker.on('error', error => signale.error('[CONTACT-EXPORT] Worker error:', error));
+  worker.on('failed', (_job, error) => signale.error('[CONTACT-EXPORT] Export failed:', error.message));
+  return worker;
+}

apps/api/src/jobs/worker.ts

@@ -19,6 +19,7 @@ import {createCardVerificationWorker} from './card-verification-processor.js';
 import {createCampaignStatsSweepWorker} from './campaign-stats-sweep-processor.js';
 import {createSnoozeSweepWorker} from './snooze-sweep-processor.js';
 import {createCardVerificationSweepWorker} from './card-verification-sweep-processor.js';
+import {createContactExportWorker} from './contact-export-processor.js';
 import {createDomainVerificationWorker} from './domain-verification-processor.js';
 import {createEmailBodyCleanupWorker} from './email-body-cleanup-processor.js';
 import {createEmailWorker} from './email-processor.js';
@@ -69,6 +70,8 @@ async function startWorkers() {
     workers.push({name: 'workflow', worker: workflowWorker});
     signale.success('[WORKER] Workflow worker started');
 
+    workers.push({name: 'contact-export', worker: createContactExportWorker()});
+
     // Start import worker
     const importWorker = createImportWorker();
     workers.push({name: 'import', worker: importWorker});