From 655e808e03e71d6989b4a1310ddcfa016480f462 Mon Sep 17 00:00:00 2001 From: Pratham2511 <145864145+Pratham2511@users.noreply.github.com> Date: Tue, 20 Oct 2026 08:33:42 +0530 Subject: [PATCH] feat: export all project contacts as CSV --- CONTRIBUTING.md | 26 ++ apps/api/src/controllers/Contacts.ts | 63 +++ .../__tests__/Contacts.export.test.ts | 212 ++++++++++ apps/api/src/jobs/contact-export-processor.ts | 20 + apps/api/src/jobs/worker.ts | 3 + apps/api/src/services/ContactExportService.ts | 164 ++++++++ apps/api/src/services/ContactExportStorage.ts | 140 ++++++ apps/api/src/services/QueueService.ts | 13 + .../__tests__/ContactExportService.test.ts | 398 ++++++++++++++++++ .../src/components/ContactExportDialog.tsx | 134 ++++++ apps/web/src/pages/contacts/index.tsx | 11 + packages/types/src/jobs/contact-export.ts | 17 + packages/types/src/jobs/index.ts | 2 + 13 files changed, 1203 insertions(+) create mode 100644 apps/api/src/controllers/__tests__/Contacts.export.test.ts create mode 100644 apps/api/src/jobs/contact-export-processor.ts create mode 100644 apps/api/src/services/ContactExportService.ts create mode 100644 apps/api/src/services/ContactExportStorage.ts create mode 100644 apps/api/src/services/__tests__/ContactExportService.test.ts create mode 100644 apps/web/src/components/ContactExportDialog.tsx create mode 100644 packages/types/src/jobs/contact-export.ts diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index f2d860ee2..822ac4011 100644 --- a/CONTRIBUTING.md +++ b/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 `-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. diff --git a/apps/api/src/controllers/Contacts.ts b/apps/api/src/controllers/Contacts.ts index 311b599cf..7f3fed4e1 100644 --- a/apps/api/src/controllers/Contacts.ts +++ b/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); + } +} diff --git a/apps/api/src/controllers/__tests__/Contacts.export.test.ts b/apps/api/src/controllers/__tests__/Contacts.export.test.ts new file mode 100644 index 000000000..24e9958ea --- /dev/null +++ b/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()), + 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; +let events: QueueEvents; +let project: Awaited>; +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(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((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'); + }); +}); diff --git a/apps/api/src/jobs/contact-export-processor.ts b/apps/api/src/jobs/contact-export-processor.ts new file mode 100644 index 000000000..54f42fe92 --- /dev/null +++ b/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( + 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; +} diff --git a/apps/api/src/jobs/worker.ts b/apps/api/src/jobs/worker.ts index a090c01ee..480268d7e 100644 --- a/apps/api/src/jobs/worker.ts +++ b/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}); diff --git a/apps/api/src/services/ContactExportService.ts b/apps/api/src/services/ContactExportService.ts new file mode 100644 index 000000000..daef97754 --- /dev/null +++ b/apps/api/src/services/ContactExportService.ts @@ -0,0 +1,164 @@ +import {createReadStream} from 'node:fs'; +import {mkdtemp, open, rm} from 'node:fs/promises'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; +import {createInterface} from 'node:readline'; + +import type {Contact, Prisma} from '@plunk/db'; +import type {ContactExportJobData, ContactExportResult, ContactExportStatus} from '@plunk/types'; +import type {Job} from 'bullmq'; + +import {S3_ENABLED} from '../app/constants.js'; +import {prisma} from '../database/prisma.js'; +import {HttpException} from '../exceptions/index.js'; +import {ContactExportStorage} from './ContactExportStorage.js'; +import {contactExportQueue} from './QueueService.js'; + +const BATCH_SIZE = 100; +const MAX_CUSTOM_FIELDS = 1000; +const LIFETIME = 24 * 60 * 60 * 1000; +const COLUMNS = ['id', 'email', 'subscribed', 'snoozedUntil', 'createdAt', 'updatedAt'] as const; + +function customData(data: Prisma.JsonValue | null): Prisma.JsonObject { + return data && typeof data === 'object' && !Array.isArray(data) ? data : {}; +} + +/** RFC 4180 quoting alone does not prevent spreadsheet formula execution. */ +export function csvCell(value: unknown): string { + let text = value == null ? '' : typeof value === 'object' ? JSON.stringify(value) : String(value); + if (typeof value === 'string' && (/^[\t\r\n]/.test(text) || /^\s*[=+\-@]/u.test(text))) text = `'${text}`; + return `"${text.replace(/"/g, '""')}"`; +} + +export class ContactExportService { + static async start(projectId: string) { + if (!S3_ENABLED) throw new HttpException(503, 'Contact exports require S3-compatible storage to be configured.'); + for (let attempt = 0; attempt < 3; attempt++) { + const added = await contactExportQueue.add('export', {projectId}, {deduplication: {id: projectId}}); + // add() can return a deduplicated id with the NEW request's data/timestamp. + // Read the stored job to check the lifecycle of the actual queued work. + const job = await contactExportQueue.getJob(added.id!); + if (!job) continue; // Another requester just removed the same obsolete job. + if (job.data.projectId !== projectId) throw new HttpException(404, 'Contact export not found'); + if (!job.data.cancelled && Date.now() - job.timestamp < LIFETIME) return {jobId: job.id!}; + + // Keep the deduplication boundary until an active worker has finished cleanup. + // If a waiting job becomes active after this check, BullMQ's atomic remove + // refuses its lock. Never delete a deduplication key separately from its job: + // concurrent retries must not release a newer job's boundary. + if ((await job.getState()) === 'active' || (await contactExportQueue.remove(job.id!)) === 0) { + throw new HttpException(409, 'The previous contact export is still stopping. Please try again shortly.'); + } + } + throw new HttpException(409, 'Contact export changed while starting. Please try again.'); + } + + static async getJob(jobId: string, projectId?: string) { + const job = await contactExportQueue.getJob(jobId); + if (!job || (projectId && job.data.projectId !== projectId) || Date.now() - job.timestamp >= LIFETIME) { + throw new HttpException(404, 'Contact export not found or expired'); + } + return job; + } + + static async status(jobId: string, projectId: string): Promise { + const job = await this.getJob(jobId, projectId); + const state = await job.getState(); + if (job.data.cancelled) return {status: 'cancelled', rows: 0}; + if (state === 'completed') { + const result = job.returnvalue as ContactExportResult; + return {status: 'completed', rows: result.rows, filename: result.filename}; + } + if (state === 'failed') { + const error = + job.failedReason === 'Contact export exceeds 1000 custom fields' + ? 'This project exceeds the export limit of 1000 custom fields.' + : 'Could not export contacts. Please try again.'; + return {status: 'failed', rows: 0, error}; + } + return {status: state === 'active' ? 'active' : 'waiting', rows: Number(job.progress) || 0}; + } + + static async cancel(jobId: string, projectId: string) { + const job = await this.getJob(jobId, projectId); + await job.updateData({...job.data, cancelled: true}); + if (job.returnvalue?.key) await ContactExportStorage.remove(job.returnvalue.key); + } + + /** Snapshot to disk once, then derive the header from every row before streaming CSV. + * Memory is limited to one page, one row and at most 1000 custom keys. Contacts + * are evaluated as the worker reads them; this is not a database point-in-time backup. + */ + static async process(job: Job): Promise { + const checkCancelled = async () => { + const current = await this.getJob(job.id!, job.data.projectId); + if (current.data.cancelled) throw new Error('Contact export cancelled'); + }; + const directory = await mkdtemp(join(tmpdir(), 'plunk-contact-export-')); + const path = join(directory, 'contacts.ndjson'); + let key: string | undefined; + try { + const fields = new Set(); + const snapshot = await open(path, 'wx', 0o600); + let cursor: string | undefined; + let rows = 0; + try { + while (true) { + await checkCancelled(); + // The existing unique (projectId, email) index serves this ordered walk. + const contacts: Contact[] = await prisma.contact.findMany({ + where: {projectId: job.data.projectId, ...(cursor !== undefined ? {email: {gt: cursor}} : {})}, + orderBy: {email: 'asc'}, + take: BATCH_SIZE, + }); + if (contacts.length === 0) break; + for (const contact of contacts) { + for (const field of Object.keys(customData(contact.data))) { + fields.add(field); + if (fields.size > MAX_CUSTOM_FIELDS) throw new Error('Contact export exceeds 1000 custom fields'); + } + await snapshot.writeFile(`${JSON.stringify(contact)}\n`); + } + rows += contacts.length; + cursor = contacts[contacts.length - 1]!.email; + await job.updateProgress(rows); + } + } finally { + await snapshot.close(); + } + const customFields = [...fields].sort(); + async function* csv() { + // UTF-8 BOM makes Unicode contacts open correctly in spreadsheet applications. + yield '\uFEFF' + [...COLUMNS, ...customFields.map(field => `data.${field}`)].map(csvCell).join(',') + '\r\n'; + const stream = createReadStream(path); + const lines = createInterface({input: stream, crlfDelay: Infinity}); + try { + let read = 0; + for await (const line of lines) { + if (read++ % BATCH_SIZE === 0) await checkCancelled(); + const contact = JSON.parse(line) as Contact; + const data = customData(contact.data); + yield [ + ...COLUMNS.map(column => contact[column]), + ...customFields.map(field => (Object.hasOwn(data, field) ? data[field] : null)), + ] + .map(csvCell) + .join(',') + '\r\n'; + } + } finally { + lines.close(); + stream.destroy(); + } + } + key = await ContactExportStorage.upload(csv(), checkCancelled); + // Cancellation may have arrived while the final upload request was in flight. + await checkCancelled(); + return {key, rows, filename: `contacts-${new Date(job.timestamp).toISOString().slice(0, 10)}.csv`}; + } catch (error) { + if (key) await ContactExportStorage.remove(key); + throw error; + } finally { + await rm(directory, {recursive: true, force: true}); + } + } +} diff --git a/apps/api/src/services/ContactExportStorage.ts b/apps/api/src/services/ContactExportStorage.ts new file mode 100644 index 000000000..b6d0f27c3 --- /dev/null +++ b/apps/api/src/services/ContactExportStorage.ts @@ -0,0 +1,140 @@ +import {randomUUID} from 'node:crypto'; +import type {Readable} from 'node:stream'; + +import { + AbortMultipartUploadCommand, + CompleteMultipartUploadCommand, + CreateBucketCommand, + CreateMultipartUploadCommand, + DeleteObjectCommand, + GetBucketPolicyCommand, + GetObjectCommand, + HeadBucketCommand, + PutBucketLifecycleConfigurationCommand, + S3Client, + UploadPartCommand, +} from '@aws-sdk/client-s3'; + +import { + S3_ACCESS_KEY_ID, + S3_ACCESS_KEY_SECRET, + S3_BUCKET, + S3_ENABLED, + S3_ENDPOINT, + S3_FORCE_PATH_STYLE, +} from '../app/constants.js'; + +// Never use S3Service's public upload bucket for contact data. +const bucket = `${S3_BUCKET}-contact-exports`; +const client = new S3Client({ + region: 'us-east-1', + endpoint: S3_ENDPOINT, + forcePathStyle: S3_FORCE_PATH_STYLE, + credentials: {accessKeyId: S3_ACCESS_KEY_ID, secretAccessKey: S3_ACCESS_KEY_SECRET}, +}); +const PART_SIZE = 5 * 1024 * 1024; + +export class ContactExportStorage { + static async initialize() { + if (!S3_ENABLED) throw new Error('Contact exports require S3-compatible storage'); + try { + await client.send(new HeadBucketCommand({Bucket: bucket})); + } catch (error) { + if ((error as {$metadata?: {httpStatusCode?: number}}).$metadata?.httpStatusCode !== 404) throw error; + try { + await client.send(new CreateBucketCommand({Bucket: bucket})); + } catch (createError) { + if ((createError as {name?: string}).name !== 'BucketAlreadyOwnedByYou') throw createError; + } + } + // Fail closed if somebody configured a policy on the private bucket. + try { + const policy = await client.send(new GetBucketPolicyCommand({Bucket: bucket})); + if (policy.Policy) throw new Error('The contact export bucket must have no bucket policy'); + } catch (error) { + if ((error as {name?: string}).name !== 'NoSuchBucketPolicy') throw error; + } + await client.send( + new PutBucketLifecycleConfigurationCommand({ + Bucket: bucket, + LifecycleConfiguration: { + Rules: [ + { + ID: 'expire-contact-exports', + Status: 'Enabled', + Filter: {Prefix: ''}, + Expiration: {Days: 1}, + AbortIncompleteMultipartUpload: {DaysAfterInitiation: 1}, + }, + ], + }, + }), + ); + } + + static async upload(source: AsyncIterable, checkCancelled: () => Promise): Promise { + await this.initialize(); + const key = randomUUID(); + const created = await client.send( + new CreateMultipartUploadCommand({ + Bucket: bucket, + Key: key, + ContentType: 'text/csv; charset=utf-8', + }), + ); + const uploadId = created.UploadId!; + const parts: {ETag: string; PartNumber: number}[] = []; + const buffer = Buffer.alloc(PART_SIZE); + let used = 0; + const uploadPart = async () => { + await checkCancelled(); + if (parts.length >= 10000) throw new Error('Contact export exceeds the storage size limit'); + const result = await client.send( + new UploadPartCommand({ + Bucket: bucket, + Key: key, + UploadId: uploadId, + PartNumber: parts.length + 1, + Body: buffer.subarray(0, used), + }), + ); + parts.push({ETag: result.ETag!, PartNumber: parts.length + 1}); + used = 0; + }; + try { + for await (const text of source) { + const chunk = Buffer.from(text, 'utf8'); + let offset = 0; + while (offset < chunk.length) { + const copied = chunk.copy(buffer, used, offset, offset + PART_SIZE - used); + used += copied; + offset += copied; + if (used === PART_SIZE) await uploadPart(); + } + } + if (used > 0) await uploadPart(); + await checkCancelled(); + await client.send( + new CompleteMultipartUploadCommand({ + Bucket: bucket, + Key: key, + UploadId: uploadId, + MultipartUpload: {Parts: parts}, + }), + ); + return key; + } catch (error) { + await client.send(new AbortMultipartUploadCommand({Bucket: bucket, Key: key, UploadId: uploadId})); + throw error; + } + } + + static async download(key: string): Promise { + const result = await client.send(new GetObjectCommand({Bucket: bucket, Key: key})); + return result.Body as Readable; + } + + static async remove(key: string) { + await client.send(new DeleteObjectCommand({Bucket: bucket, Key: key})); + } +} diff --git a/apps/api/src/services/QueueService.ts b/apps/api/src/services/QueueService.ts index f273b1fb2..9e4ec8eda 100644 --- a/apps/api/src/services/QueueService.ts +++ b/apps/api/src/services/QueueService.ts @@ -11,6 +11,8 @@ import type { CardVerificationJobData, CardVerificationSweepJobData, ContactImportJobData, + ContactExportJobData, + ContactExportResult, DomainVerificationJobData, CampaignCancelCleanupJobData, EmailBodyCleanupJobData, @@ -107,6 +109,16 @@ export const importQueue = new Queue('import', { }, }); +// Only one unfinished export per project. Results expire independently in private S3 storage. +export const contactExportQueue = new Queue('contact-export', { + connection: redisConnection, + defaultJobOptions: { + attempts: 1, + removeOnComplete: {age: 86400, count: 1000}, + removeOnFail: {age: 86400, count: 1000}, + }, +}); + export const segmentCountQueue = new Queue('segment-count', { connection: redisConnection, defaultJobOptions: { @@ -788,6 +800,7 @@ export class QueueService { workflowQueue.close(), scheduledQueue.close(), importQueue.close(), + contactExportQueue.close(), segmentCountQueue.close(), domainVerificationQueue.close(), apiRequestCleanupQueue.close(), diff --git a/apps/api/src/services/__tests__/ContactExportService.test.ts b/apps/api/src/services/__tests__/ContactExportService.test.ts new file mode 100644 index 000000000..513b93807 --- /dev/null +++ b/apps/api/src/services/__tests__/ContactExportService.test.ts @@ -0,0 +1,398 @@ +import {access, readFile, readdir} from 'node:fs/promises'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; + +import {QueueEvents} from 'bullmq'; +import {parse} from 'csv-parse/sync'; +import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest'; + +import {factories, getPrismaClient} from '../../../../../test/helpers'; +import {prisma} from '../../database/prisma.js'; +import {createContactExportWorker} from '../../jobs/contact-export-processor.js'; +import {ContactExportService, csvCell} from '../ContactExportService.js'; +import {ContactExportStorage} from '../ContactExportStorage.js'; +import {contactExportQueue} from '../QueueService.js'; + +const database = getPrismaClient(); +const keys: string[] = []; +let projectId: string; + +async function exportCsv() { + const {jobId} = await ContactExportService.start(projectId); + const job = await ContactExportService.getJob(jobId, projectId); + const result = await ContactExportService.process(job); + keys.push(result.key); + const stream = await ContactExportStorage.download(result.key); + const chunks: Buffer[] = []; + for await (const chunk of stream) chunks.push(Buffer.from(chunk)); + const csv = Buffer.concat(chunks).toString('utf8'); + return {result, csv, records: parse(csv, {bom: true, columns: true}) as Record[]}; +} + +describe('Contact CSV export', () => { + beforeEach(async () => { + await contactExportQueue.drain(); + const {project} = await factories.createUserWithProject(); + projectId = project.id; + }); + + afterEach(async () => { + vi.restoreAllMocks(); + await Promise.all(keys.splice(0).map(key => ContactExportStorage.remove(key))); + await contactExportQueue.drain(); + }); + + it('exports more than two pages, including later custom fields, in deterministic columns', async () => { + await database.contact.createMany({ + data: Array.from({length: 205}, (_, i) => ({ + projectId, + email: `contact-${String(i).padStart(3, '0')}@example.test`, + data: i === 204 ? {'z': 'late', 'email': 'custom email', 'data.z': 'literal dot'} : {a: 'early'}, + })), + }); + const query = vi.spyOn(prisma.contact, 'findMany'); + const {result, csv, records} = await exportCsv(); + expect(result.rows).toBe(205); + expect(records).toHaveLength(205); + expect(Object.keys(records[0]!)).toEqual([ + 'id', + 'email', + 'subscribed', + 'snoozedUntil', + 'createdAt', + 'updatedAt', + 'data.a', + 'data.data.z', + 'data.email', + 'data.z', + ]); + expect(records[0]!['data.z']).toBe(''); + expect(records[204]).toMatchObject({'data.z': 'late', 'data.email': 'custom email', 'data.data.z': 'literal dot'}); + expect(csv.startsWith('\uFEFF')).toBe(true); + expect(query).toHaveBeenCalledTimes(4); + for (const [args] of query.mock.calls) { + expect(args).toMatchObject({where: {projectId}, take: 100, orderBy: {email: 'asc'}}); + expect(args).not.toHaveProperty('skip'); + } + expect(query.mock.calls[1]?.[0]).toMatchObject({where: {email: {gt: records[99]!.email}}}); + }); + + it('exports only headers for an empty project', async () => { + const {result, records, csv} = await exportCsv(); + expect(result.rows).toBe(0); + expect(records).toEqual([]); + expect(parse(csv, {bom: true})).toHaveLength(1); + }); + + it('preserves quoting, Unicode, nulls, booleans, numbers, dates and nested JSON', async () => { + await factories.createContact({ + projectId, + data: { + text: 'Hello, "世界"\r\nnext line', + empty: null, + flag: false, + count: -12.5, + object: {nested: ['α', 2, true]}, + formula: ' =HYPERLINK("https://example.test")', + __proto__: 'ignored by object literal', + }, + }); + const {records} = await exportCsv(); + expect(records[0]).toMatchObject({ + 'data.text': 'Hello, "世界"\r\nnext line', + 'data.empty': '', + 'data.flag': 'false', + 'data.count': '-12.5', + 'data.object': '{"nested":["α",2,true]}', + 'data.formula': '\' =HYPERLINK("https://example.test")', + }); + expect(records[0]!.createdAt).toMatch(/^\d{4}-\d{2}-\d{2}T.*Z$/); + }); + + it.each(['=1+1', '+SUM(A1)', '-2+1', '@SUM(A1)', '\tvalue', '\rvalue', '\nvalue', ' =1', '\uFEFF=1'])( + 'neutralizes spreadsheet formulas/control prefixes: %j', + value => { + expect(parse(csvCell(value))[0][0]).toBe(`'${value}`); + }, + ); + + it('does not include another project or accept its export id', async () => { + const {project: other} = await factories.createUserWithProject(); + await factories.createContact({projectId: other.id, email: 'private@example.test', data: {secret: 'never export'}}); + const {records, csv} = await exportCsv(); + expect(records).toEqual([]); + expect(csv).not.toContain('secret'); + const {jobId} = await ContactExportService.start(projectId); + await expect(ContactExportService.getJob(jobId, other.id)).rejects.toThrow('not found'); + await expect(ContactExportService.cancel(jobId, other.id)).rejects.toThrow('not found'); + }); + + it('deduplicates unfinished jobs for the project', async () => { + const first = await ContactExportService.start(projectId); + const second = await ContactExportService.start(projectId); + expect(second.jobId).toBe(first.jobId); + }); + + it('fails explicitly for too many custom columns and removes its snapshot', async () => { + await factories.createContact({ + projectId, + data: Object.fromEntries(Array.from({length: 1001}, (_, i) => [`k${i}`, i])), + }); + const before = await readdir(tmpdir()); + const {jobId} = await ContactExportService.start(projectId); + const job = await ContactExportService.getJob(jobId, projectId); + await expect(ContactExportService.process(job)).rejects.toThrow('1000 custom fields'); + expect( + (await readdir(tmpdir())).filter(name => name.startsWith('plunk-contact-export-') && !before.includes(name)), + ).toEqual([]); + }); + + it('expires access even when Redis still retains the job', async () => { + const {jobId} = await ContactExportService.start(projectId); + const now = Date.now(); + vi.spyOn(Date, 'now').mockReturnValue(now + 86400001); + await expect(ContactExportService.getJob(jobId, projectId)).rejects.toThrow('expired'); + }); + + it('streams multipart CSV and aborts an interrupted upload', async () => { + const source = async function* () { + yield 'a'.repeat(5 * 1024 * 1024); + throw new Error('interrupted source'); + }; + await expect(ContactExportStorage.upload(source(), async () => {})).rejects.toThrow('interrupted source'); + const content = '世界,'.repeat(2 * 1024 * 1024); + const key = await ContactExportStorage.upload( + (async function* () { + yield content; + })(), + async () => {}, + ); + keys.push(key); + const stream = await ContactExportStorage.download(key); + let size = 0; + for await (const chunk of stream) size += Buffer.byteLength(chunk); + expect(size).toBe(Buffer.byteLength(content)); + const response = await fetch(`${process.env.S3_ENDPOINT}/${process.env.S3_BUCKET}-contact-exports/${key}`); + expect(response.status).toBe(403); + }); +}); + +function deferred() { + let resolve!: () => void; + const promise = new Promise(done => { + resolve = done; + }); + return {promise, resolve}; +} + +describe('Contact export queue lifecycle', () => { + let projectId: string; + let worker: ReturnType | undefined; + let events: QueueEvents; + let releaseWorker: (() => void) | undefined; + + beforeEach(async () => { + await contactExportQueue.pause(); + await contactExportQueue.drain(); + projectId = (await factories.createUserWithProject()).project.id; + events = new QueueEvents(contactExportQueue.name, {connection: contactExportQueue.opts.connection}); + await events.waitUntilReady(); + }); + + afterEach(async () => { + releaseWorker?.(); + releaseWorker = undefined; + await worker?.close(); + worker = undefined; + await events.close(); + vi.restoreAllMocks(); + await contactExportQueue.drain(); + await contactExportQueue.resume(); + }); + + it('restarts immediately after queued cancellation, without a worker consuming the old job', async () => { + const first = await ContactExportService.start(projectId); + await ContactExportService.cancel(first.jobId, projectId); + expect(await ContactExportService.status(first.jobId, projectId)).toEqual({status: 'cancelled', rows: 0}); + const retry = await ContactExportService.start(projectId); + expect(retry.jobId).not.toBe(first.jobId); + expect(await ContactExportService.status(retry.jobId, projectId)).toMatchObject({status: 'waiting'}); + expect(await contactExportQueue.getJob(first.jobId)).toBeUndefined(); + expect(await contactExportQueue.getDeduplicationJobId(projectId)).toBe(retry.jobId); + }); + + it('replaces an expired waiting job using its stored timestamp', async () => { + const first = await ContactExportService.start(projectId); + const client = await contactExportQueue.client; + await client.hset(contactExportQueue.toKey(first.jobId), 'timestamp', String(Date.now() - 86400001)); + await expect(ContactExportService.getJob(first.jobId, projectId)).rejects.toThrow('expired'); + const retry = await ContactExportService.start(projectId); + expect(retry.jobId).not.toBe(first.jobId); + expect(await ContactExportService.status(retry.jobId, projectId)).toMatchObject({status: 'waiting'}); + expect(await contactExportQueue.getJob(first.jobId)).toBeUndefined(); + }); + + it.each(['new', 'cancelled', 'expired'] as const)('deduplicates concurrent starts for a %s export', async state => { + if (state !== 'new') { + const first = await ContactExportService.start(projectId); + if (state === 'cancelled') await ContactExportService.cancel(first.jobId, projectId); + else { + const client = await contactExportQueue.client; + await client.hset(contactExportQueue.toKey(first.jobId), 'timestamp', String(Date.now() - 86400001)); + } + } + const starts = await Promise.all(Array.from({length: 20}, () => ContactExportService.start(projectId))); + expect(new Set(starts.map(start => start.jobId)).size).toBe(1); + const id = starts[0]!.jobId; + expect(await contactExportQueue.getDeduplicationJobId(projectId)).toBe(id); + expect(await ContactExportService.status(id, projectId)).toMatchObject({status: 'waiting'}); + const queued = await contactExportQueue.getJobs(['waiting', 'paused']); + expect(queued.filter(job => job.data.projectId === projectId).map(job => job.id)).toEqual([id]); + }); + + it.each(['cancelled', 'expired'] as const)( + 'keeps an %s active export locked until real worker cleanup completes', + async state => { + await database.contact.createMany({ + data: Array.from({length: 150}, (_, i) => ({projectId, email: `lifecycle-${i}@example.test`})), + }); + const before = await readdir(tmpdir()); + const entered = deferred(); + const release = deferred(); + releaseWorker = release.resolve; + const upload = ContactExportStorage.upload.bind(ContactExportStorage); + vi.spyOn(ContactExportStorage, 'upload').mockImplementationOnce(async (source, checkCancelled) => { + entered.resolve(); + await release.promise; + return upload(source, checkCancelled); + }); + const query = vi.spyOn(prisma.contact, 'findMany'); + const first = await ContactExportService.start(projectId); + const job = await ContactExportService.getJob(first.jobId, projectId); + // Attach failure handling before the worker can reject the cancelled/expired job. + const finished = job.waitUntilFinished(events, 15000).then( + () => undefined, + error => error as Error, + ); + worker = createContactExportWorker(); + await contactExportQueue.resume(); + await entered.promise; + expect(await job.getState()).toBe('active'); + expect(query).toHaveBeenCalledTimes(3); + const snapshots = await projectSnapshots(projectId, before); + expect(snapshots).toHaveLength(1); + const concurrent = await Promise.all(Array.from({length: 20}, () => ContactExportService.start(projectId))); + expect(new Set(concurrent.map(result => result.jobId))).toEqual(new Set([first.jobId])); + if (state === 'cancelled') await ContactExportService.cancel(first.jobId, projectId); + else { + const client = await contactExportQueue.client; + await client.hset(contactExportQueue.toKey(first.jobId), 'timestamp', String(Date.now() - 86400001)); + } + await expect(ContactExportService.start(projectId)).rejects.toMatchObject({ + code: 409, + message: expect.stringContaining('still stopping'), + }); + expect(await contactExportQueue.getDeduplicationJobId(projectId)).toBe(first.jobId); + expect( + (await contactExportQueue.getJobs(['active'])).filter(job => job.data.projectId === projectId), + ).toHaveLength(1); + await contactExportQueue.pause(); + release.resolve(); + expect(await finished).toMatchObject({ + message: expect.stringContaining(state === 'cancelled' ? 'cancelled' : 'expired'), + }); + for (const snapshot of snapshots) await expect(access(snapshot)).rejects.toMatchObject({code: 'ENOENT'}); + const retry = await ContactExportService.start(projectId); + expect(retry.jobId).not.toBe(first.jobId); + expect(await ContactExportService.status(retry.jobId, projectId)).toMatchObject({status: 'waiting'}); + }, + ); + + it('stops after the current page on active cancellation and removes its snapshot before unlocking', async () => { + await database.contact.createMany({ + data: Array.from({length: 150}, (_, i) => ({projectId, email: `page-${i}@example.test`})), + }); + const before = await readdir(tmpdir()); + const entered = deferred(); + const release = deferred(); + releaseWorker = release.resolve; + const process = ContactExportService.process.bind(ContactExportService); + vi.spyOn(ContactExportService, 'process').mockImplementationOnce(async job => { + const updateProgress = job.updateProgress.bind(job); + vi.spyOn(job, 'updateProgress').mockImplementationOnce(async progress => { + await updateProgress(progress); + entered.resolve(); + await release.promise; + }); + return process(job); + }); + const query = vi.spyOn(prisma.contact, 'findMany'); + const first = await ContactExportService.start(projectId); + const job = await ContactExportService.getJob(first.jobId, projectId); + const finished = job.waitUntilFinished(events, 15000).then( + () => undefined, + error => error as Error, + ); + worker = createContactExportWorker(); + await contactExportQueue.resume(); + await entered.promise; + const snapshots = await projectSnapshots(projectId, before); + expect(snapshots).toHaveLength(1); + await ContactExportService.cancel(first.jobId, projectId); + await expect(ContactExportService.start(projectId)).rejects.toMatchObject({code: 409}); + await contactExportQueue.pause(); + release.resolve(); + expect(await finished).toMatchObject({message: expect.stringContaining('cancelled')}); + expect(query).toHaveBeenCalledTimes(1); + for (const snapshot of snapshots) await expect(access(snapshot)).rejects.toMatchObject({code: 'ENOENT'}); + expect(await ContactExportService.status(first.jobId, projectId)).toEqual({status: 'cancelled', rows: 0}); + expect((await ContactExportService.start(projectId)).jobId).not.toBe(first.jobId); + }); + + it('does not release deduplication when a queued job becomes active just before removal', async () => { + const entered = deferred(); + const release = deferred(); + releaseWorker = release.resolve; + const process = ContactExportService.process.bind(ContactExportService); + vi.spyOn(ContactExportService, 'process').mockImplementationOnce(async job => { + entered.resolve(); + await release.promise; + return process(job); + }); + const first = await ContactExportService.start(projectId); + await ContactExportService.cancel(first.jobId, projectId); + const job = await ContactExportService.getJob(first.jobId, projectId); + const finished = job.waitUntilFinished(events, 15000).then( + () => undefined, + error => error as Error, + ); + worker = createContactExportWorker(); + const remove = contactExportQueue.remove.bind(contactExportQueue); + vi.spyOn(contactExportQueue, 'remove').mockImplementationOnce(async (jobId, options) => { + // Promote in the gap AFTER getState reported waiting, BEFORE atomic remove. + await contactExportQueue.resume(); + await entered.promise; + expect(await job.getState()).toBe('active'); + return remove(jobId, options); + }); + await expect(ContactExportService.start(projectId)).rejects.toMatchObject({code: 409}); + expect(await contactExportQueue.getDeduplicationJobId(projectId)).toBe(first.jobId); + expect(await contactExportQueue.getJob(first.jobId)).toBeDefined(); + await contactExportQueue.pause(); + release.resolve(); + expect(await finished).toMatchObject({message: expect.stringContaining('cancelled')}); + expect((await ContactExportService.start(projectId)).jobId).not.toBe(first.jobId); + }); +}); + +// Other test files export in parallel; identify only this test project's files. +async function projectSnapshots(projectId: string, before: string[]) { + const matches: string[] = []; + for (const name of await readdir(tmpdir())) { + if (!name.startsWith('plunk-contact-export-') || before.includes(name)) continue; + const path = join(tmpdir(), name, 'contacts.ndjson'); + const contents = await readFile(path, 'utf8').catch(() => ''); + if (contents.includes(`"projectId":"${projectId}"`)) matches.push(join(tmpdir(), name)); + } + return matches; +} diff --git a/apps/web/src/components/ContactExportDialog.tsx b/apps/web/src/components/ContactExportDialog.tsx new file mode 100644 index 000000000..d9fc7029e --- /dev/null +++ b/apps/web/src/components/ContactExportDialog.tsx @@ -0,0 +1,134 @@ +import type {ContactExportStatus} from '@plunk/types'; +import {Button, Dialog, DialogContent, DialogDescription, DialogFooter, DialogHeader, DialogTitle} from '@plunk/ui'; +import {Download, Loader2} from 'lucide-react'; +import {useEffect, useRef, useState} from 'react'; + +import {API_URI} from '../lib/constants'; +import {network} from '../lib/network'; + +export function ContactExportDialog({open, onOpenChange}: {open: boolean; onOpenChange: (open: boolean) => void}) { + const [jobId, setJobId] = useState(); + const [status, setStatus] = useState(); + const [pollAttempt, setPollAttempt] = useState(0); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(); + const requestInFlight = useRef(false); + + useEffect(() => { + if (!open || !jobId) return; + let disposed = false; + let timer: ReturnType; + const poll = async () => { + try { + const result = await network.fetch('GET', `/contacts/exports/${jobId}`); + if (disposed) return; + setStatus(result); + if (result.status === 'failed') setError(result.error); + if (result.status === 'waiting' || result.status === 'active') timer = setTimeout(() => void poll(), 1500); + } catch (err) { + if (!disposed) setError(err instanceof Error ? err.message : 'Could not check the export. Please try again.'); + } + }; + void poll(); + return () => { + disposed = true; + clearTimeout(timer); + }; + }, [open, jobId, pollAttempt]); + + const start = async () => { + if (requestInFlight.current) return; + requestInFlight.current = true; + setBusy(true); + setError(undefined); + try { + const result = await network.fetch<{jobId: string}>('POST', '/contacts/exports'); + setJobId(result.jobId); + setPollAttempt(attempt => attempt + 1); + setStatus({status: 'waiting', rows: 0}); + } catch (err) { + setError(err instanceof Error ? err.message : 'Could not start the export. Please try again.'); + } finally { + requestInFlight.current = false; + setBusy(false); + } + }; + + const close = async () => { + if (requestInFlight.current) return; + requestInFlight.current = true; + setBusy(true); + try { + if (jobId && (status?.status === 'waiting' || status?.status === 'active')) { + await network.fetch('DELETE', `/contacts/exports/${jobId}`); + } + setJobId(undefined); + setStatus(undefined); + setError(undefined); + onOpenChange(false); + } catch (err) { + setError(err instanceof Error ? err.message : 'Could not cancel the export. Please try again.'); + } finally { + requestInFlight.current = false; + setBusy(false); + } + }; + + const running = status?.status === 'waiting' || status?.status === 'active'; + return ( + { + if (!value) void close(); + }} + > + + + Export all contacts + + + Download every contact in this project as CSV, including custom fields. Search, filters and selected rows do + not limit this export. + +

+ Contacts are read when the export runs. Exports expire 24 hours after they are requested. Formula-like text is + prefixed with an apostrophe for spreadsheet safety. +

+
+ {running && ( +

+

+ )} + {status?.status === 'completed' &&

Ready: {status.rows.toLocaleString()} contacts.

} + {status?.status === 'cancelled' &&

Export cancelled.

} +
+ {error && ( +

+ {error} +

+ )} + + + {status?.status === 'completed' && jobId ? ( + + ) : ( + + )} + + + + ); +} diff --git a/apps/web/src/pages/contacts/index.tsx b/apps/web/src/pages/contacts/index.tsx index da75e7822..114ea8fbd 100644 --- a/apps/web/src/pages/contacts/index.tsx +++ b/apps/web/src/pages/contacts/index.tsx @@ -26,6 +26,7 @@ import { type SortingState, type VisibilityState, } from '@tanstack/react-table'; +import {ContactExportDialog} from '../../components/ContactExportDialog'; import {DashboardLayout} from '../../components/DashboardLayout'; import { BulkActionBar, @@ -45,6 +46,7 @@ import {KeyValueEditor} from '../../components/KeyValueEditor'; import {network} from '../../lib/network'; import {ContactStatusBadge, ContactStatusIcon} from '../../lib/contactStatus'; import {formatRelativeTime} from '../../lib/dateUtils'; +import {useActiveProject} from '../../lib/contexts/ActiveProjectProvider'; import {useColumnVisibility} from '../../lib/hooks/useColumnVisibility'; import {usePersistentState} from '../../lib/hooks/usePersistentState'; import { @@ -54,6 +56,7 @@ import { ChevronLeft, ChevronRight, Edit, + Download, EyeOff, FileUp, Loader2, @@ -99,6 +102,7 @@ const DEFAULT_COLUMN_VISIBILITY: VisibilityState = { }; export default function ContactsPage() { + const {activeProject} = useActiveProject(); const [cursor, setCursor] = useState(undefined); const [cursorHistory, setCursorHistory] = useState<(string | undefined)[]>([undefined]); const [currentPage, setCurrentPage] = useState(0); @@ -111,6 +115,7 @@ export default function ContactsPage() { const [columnVisibility, setColumnVisibility] = useColumnVisibility(COLUMNS_STORAGE_KEY, DEFAULT_COLUMN_VISIBILITY); const [showCreateDialog, setShowCreateDialog] = useState(false); const [showImportDialog, setShowImportDialog] = useState(false); + const [showExportDialog, setShowExportDialog] = useState(false); const [showDeleteDialog, setShowDeleteDialog] = useState(false); const [contactToDelete, setContactToDelete] = useState(null); const [totalCount, setTotalCount] = useState(0); @@ -449,6 +454,7 @@ export default function ContactsPage() { return ( <> +
@@ -462,6 +468,11 @@ export default function ContactsPage() {

+