diff --git a/apps/api/src/controllers/Contacts.ts b/apps/api/src/controllers/Contacts.ts index d05302a..7e78121 100644 --- a/apps/api/src/controllers/Contacts.ts +++ b/apps/api/src/controllers/Contacts.ts @@ -379,4 +379,137 @@ export class Contacts { }); } } + + /** + * POST /contacts/bulk-subscribe + * Queue bulk subscribe operation + */ + @Post('bulk-subscribe') + @Middleware([requireAuth]) + @CatchAsync + public async bulkSubscribe(req: Request, res: Response, _next: NextFunction) { + const auth = res.locals.auth as AuthResponse; + const {contactIds} = req.body; + + if (!Array.isArray(contactIds) || contactIds.length === 0) { + return res.status(400).json({error: 'contactIds array is required'}); + } + + // Validate limit + if (contactIds.length > 1000) { + return res.status(400).json({error: 'Maximum 1000 contacts can be processed at once'}); + } + + try { + const job = await QueueService.queueBulkContactAction(auth.projectId!, contactIds, 'subscribe'); + + return res.status(202).json({ + message: 'Bulk subscribe queued successfully', + jobId: job.id, + }); + } catch (error) { + signale.error('[CONTACTS] Failed to queue bulk subscribe:', error); + return res.status(500).json({ + error: error instanceof Error ? error.message : 'Failed to queue bulk subscribe', + }); + } + } + + /** + * POST /contacts/bulk-unsubscribe + * Queue bulk unsubscribe operation + */ + @Post('bulk-unsubscribe') + @Middleware([requireAuth]) + @CatchAsync + public async bulkUnsubscribe(req: Request, res: Response, _next: NextFunction) { + const auth = res.locals.auth as AuthResponse; + const {contactIds} = req.body; + + if (!Array.isArray(contactIds) || contactIds.length === 0) { + return res.status(400).json({error: 'contactIds array is required'}); + } + + if (contactIds.length > 1000) { + return res.status(400).json({error: 'Maximum 1000 contacts can be processed at once'}); + } + + try { + const job = await QueueService.queueBulkContactAction(auth.projectId!, contactIds, 'unsubscribe'); + + return res.status(202).json({ + message: 'Bulk unsubscribe queued successfully', + jobId: job.id, + }); + } catch (error) { + signale.error('[CONTACTS] Failed to queue bulk unsubscribe:', error); + return res.status(500).json({ + error: error instanceof Error ? error.message : 'Failed to queue bulk unsubscribe', + }); + } + } + + /** + * POST /contacts/bulk-delete + * Queue bulk delete operation + */ + @Post('bulk-delete') + @Middleware([requireAuth]) + @CatchAsync + public async bulkDelete(req: Request, res: Response, _next: NextFunction) { + const auth = res.locals.auth as AuthResponse; + const {contactIds} = req.body; + + if (!Array.isArray(contactIds) || contactIds.length === 0) { + return res.status(400).json({error: 'contactIds array is required'}); + } + + if (contactIds.length > 1000) { + return res.status(400).json({error: 'Maximum 1000 contacts can be processed at once'}); + } + + try { + const job = await QueueService.queueBulkContactAction(auth.projectId!, contactIds, 'delete'); + + return res.status(202).json({ + message: 'Bulk delete queued successfully', + jobId: job.id, + }); + } catch (error) { + signale.error('[CONTACTS] Failed to queue bulk delete:', error); + return res.status(500).json({ + error: error instanceof Error ? error.message : 'Failed to queue bulk delete', + }); + } + } + + /** + * GET /contacts/bulk/:jobId + * Get bulk action job status + */ + @Get('bulk/:jobId') + @Middleware([requireAuth]) + @CatchAsync + public async getBulkActionStatus(req: Request, res: Response, _next: NextFunction) { + const jobId = req.params.jobId; + + if (!jobId) { + return res.status(400).json({error: 'Job ID is required'}); + } + + try { + const status = await QueueService.getBulkActionJobStatus(jobId); + + if (!status) { + return res.status(404).json({error: 'Bulk action job not found'}); + } + + return res.status(200).json(status); + } catch (error) { + signale.error('[CONTACTS] Failed to get bulk action status:', error); + return res.status(500).json({ + error: error instanceof Error ? error.message : 'Failed to get bulk action status', + }); + } + } } diff --git a/apps/api/src/jobs/bulk-contact-processor.ts b/apps/api/src/jobs/bulk-contact-processor.ts new file mode 100644 index 0000000..aec45e0 --- /dev/null +++ b/apps/api/src/jobs/bulk-contact-processor.ts @@ -0,0 +1,114 @@ +/** + * Background Job: Bulk Contact Action Processor + * Processes bulk subscribe, unsubscribe, and delete operations + */ + +import {type Job, Worker} from 'bullmq'; +import signale from 'signale'; + +import {ContactService} from '../services/ContactService.js'; +import {type BulkContactActionJobData, bulkContactQueue} from '../services/QueueService.js'; + +const BATCH_SIZE = 100; // Process contacts in batches of 100 + +interface BulkActionResult { + operation: 'subscribe' | 'unsubscribe' | 'delete'; + totalRequested: number; + successCount: number; + failureCount: number; + errors: {contactId: string; email: string; error: string}[]; +} + +export function createBulkContactWorker() { + const worker = new Worker( + bulkContactQueue.name, + async (job: Job) => { + const {projectId, contactIds, operation} = job.data; + + signale.info( + `[BULK-CONTACT-PROCESSOR] Processing ${operation} for ${contactIds.length} contacts in project ${projectId}`, + ); + + const result: BulkActionResult = { + operation, + totalRequested: contactIds.length, + successCount: 0, + failureCount: 0, + errors: [], + }; + + try { + // Process contacts in batches + for (let i = 0; i < contactIds.length; i += BATCH_SIZE) { + const batchIds = contactIds.slice(i, Math.min(i + BATCH_SIZE, contactIds.length)); + + try { + let batchResult: {updated?: number; deleted?: number}; + + switch (operation) { + case 'subscribe': + batchResult = await ContactService.bulkSubscribe(projectId, batchIds); + result.successCount += batchResult.updated || 0; + break; + case 'unsubscribe': + batchResult = await ContactService.bulkUnsubscribe(projectId, batchIds); + result.successCount += batchResult.updated || 0; + break; + case 'delete': + batchResult = await ContactService.bulkDelete(projectId, batchIds); + result.successCount += batchResult.deleted || 0; + break; + } + + // If some contacts in batch weren't processed, track them as failures + const processedCount = batchResult.updated || batchResult.deleted || 0; + const failedCount = batchIds.length - processedCount; + if (failedCount > 0) { + result.failureCount += failedCount; + // Note: We don't have individual contact details for batch failures + } + } catch (error) { + signale.error(`[BULK-CONTACT-PROCESSOR] Batch failed:`, error); + result.failureCount += batchIds.length; + result.errors.push({ + contactId: 'batch', + email: '', + error: error instanceof Error ? error.message : 'Batch processing failed', + }); + } + + // Update progress + const progress = Math.round(((i + batchIds.length) / contactIds.length) * 100); + await job.updateProgress(progress); + } + + signale.info( + `[BULK-CONTACT-PROCESSOR] ${operation} completed: ${result.successCount} succeeded, ${result.failureCount} failed`, + ); + + return result; + } catch (error) { + signale.error(`[BULK-CONTACT-PROCESSOR] Failed to process ${operation}:`, error); + throw error; + } + }, + { + connection: bulkContactQueue.opts.connection, + concurrency: 3, // Process max 3 bulk operations concurrently + }, + ); + + worker.on('completed', job => { + signale.info(`[BULK-CONTACT-PROCESSOR] Job ${job.id} completed`); + }); + + worker.on('failed', (job, err) => { + signale.error(`[BULK-CONTACT-PROCESSOR] Job ${job?.id} failed:`, err.message); + }); + + worker.on('error', err => { + signale.error('[BULK-CONTACT-PROCESSOR] Worker error:', err); + }); + + return worker; +} diff --git a/apps/api/src/jobs/worker.ts b/apps/api/src/jobs/worker.ts index 23bcd03..5735c0e 100644 --- a/apps/api/src/jobs/worker.ts +++ b/apps/api/src/jobs/worker.ts @@ -10,6 +10,7 @@ import {Worker} from 'bullmq'; import signale from 'signale'; import {createApiRequestCleanupWorker} from './api-request-cleanup-processor.js'; +import {createBulkContactWorker} from './bulk-contact-processor.js'; import {createCampaignWorker} from './campaign-processor.js'; import {createDomainVerificationWorker} from './domain-verification-processor.js'; import {createEmailWorker} from './email-processor.js'; @@ -49,6 +50,11 @@ async function startWorkers() { workers.push({name: 'import', worker: importWorker}); signale.success('[WORKER] Import worker started'); + // Start bulk contact action worker + const bulkContactWorker = createBulkContactWorker(); + workers.push({name: 'bulk-contact-actions', worker: bulkContactWorker}); + signale.success('[WORKER] Bulk contact action worker started'); + // Start segment count worker const segmentCountWorker = createSegmentCountWorker(); workers.push({name: 'segment-count', worker: segmentCountWorker}); diff --git a/apps/api/src/services/ContactService.ts b/apps/api/src/services/ContactService.ts index 7b396ad..3e91ea8 100644 --- a/apps/api/src/services/ContactService.ts +++ b/apps/api/src/services/ContactService.ts @@ -679,4 +679,148 @@ export class ContactService { return false; } + + /** + * Bulk subscribe contacts + * Updates multiple contacts to subscribed=true in batches + */ + public static async bulkSubscribe( + projectId: string, + contactIds: string[], + ): Promise<{updated: number}> { + // Verify all contacts belong to this project + const contacts = await prisma.contact.findMany({ + where: { + id: {in: contactIds}, + projectId, + }, + select: {id: true, subscribed: true}, + }); + + const validIds = contacts.map(c => c.id); + + if (validIds.length === 0) { + return {updated: 0}; + } + + // Only update contacts that are currently unsubscribed + const unsubscribedIds = contacts.filter(c => !c.subscribed).map(c => c.id); + + if (unsubscribedIds.length === 0) { + return {updated: 0}; + } + + // Update in a single query for performance + const result = await prisma.contact.updateMany({ + where: { + id: {in: unsubscribedIds}, + projectId, + }, + data: { + subscribed: true, + }, + }); + + // Track events for changed contacts sequentially to avoid database deadlocks + // Process in background to avoid blocking the API response + this.trackEventsSequentially(projectId, 'contact.subscribed', unsubscribedIds).catch(error => { + // Silently ignore errors in tests due to cleanup race conditions + if (process.env.NODE_ENV !== 'test') { + console.error('[ContactService] Failed to track bulk subscribe events:', error); + } + }); + + return {updated: result.count}; + } + + /** + * Bulk unsubscribe contacts + */ + public static async bulkUnsubscribe( + projectId: string, + contactIds: string[], + ): Promise<{updated: number}> { + const contacts = await prisma.contact.findMany({ + where: { + id: {in: contactIds}, + projectId, + }, + select: {id: true, subscribed: true}, + }); + + const validIds = contacts.map(c => c.id); + + if (validIds.length === 0) { + return {updated: 0}; + } + + // Only update contacts that are currently subscribed + const subscribedIds = contacts.filter(c => c.subscribed).map(c => c.id); + + if (subscribedIds.length === 0) { + return {updated: 0}; + } + + const result = await prisma.contact.updateMany({ + where: { + id: {in: subscribedIds}, + projectId, + }, + data: { + subscribed: false, + }, + }); + + // Track events for changed contacts sequentially to avoid database deadlocks + // Process in background to avoid blocking the API response + this.trackEventsSequentially(projectId, 'contact.unsubscribed', subscribedIds).catch(error => { + // Silently ignore errors in tests due to cleanup race conditions + if (process.env.NODE_ENV !== 'test') { + console.error('[ContactService] Failed to track bulk unsubscribe events:', error); + } + }); + + return {updated: result.count}; + } + + /** + * Bulk delete contacts + */ + public static async bulkDelete( + projectId: string, + contactIds: string[], + ): Promise<{deleted: number}> { + const result = await prisma.contact.deleteMany({ + where: { + id: {in: contactIds}, + projectId, + }, + }); + + return {deleted: result.count}; + } + + /** + * Track events sequentially to avoid database deadlocks + * Processes events one at a time with error handling + * + * @private + */ + private static async trackEventsSequentially( + projectId: string, + eventName: string, + contactIds: string[], + ): Promise { + for (const contactId of contactIds) { + try { + await EventService.trackEvent(projectId, eventName, contactId); + } catch (error) { + // Log error but continue processing remaining events + // Suppress logging in test environments to reduce noise from cleanup race conditions + if (process.env.NODE_ENV !== 'test') { + console.error(`[ContactService] Failed to track event ${eventName} for contact ${contactId}:`, error); + } + } + } + } } diff --git a/apps/api/src/services/QueueService.ts b/apps/api/src/services/QueueService.ts index 9d42e00..d1560dc 100644 --- a/apps/api/src/services/QueueService.ts +++ b/apps/api/src/services/QueueService.ts @@ -52,6 +52,12 @@ export interface ApiRequestCleanupJobData { // Empty - cleans up old API request logs } +export interface BulkContactActionJobData { + projectId: string; + contactIds: string[]; + operation: 'subscribe' | 'unsubscribe' | 'delete'; +} + /** * Queue Configuration */ @@ -181,6 +187,19 @@ export const apiRequestCleanupQueue = new Queue('api-r }, }); +export const bulkContactQueue = new Queue('bulk-contact-actions', { + connection: redisConnection, + defaultJobOptions: { + attempts: 2, // Limited retries for bulk operations + backoff: { + type: 'exponential', + delay: 5000, + }, + removeOnComplete: 50, // Keep last 50 completed bulk operations + removeOnFail: 100, // Keep last 100 failed bulk operations + }, +}); + /** * Queue Service - Centralized queue management */ @@ -328,6 +347,48 @@ export class QueueService { }; } + /** + * Queue bulk contact action job + */ + public static async queueBulkContactAction( + projectId: string, + contactIds: string[], + operation: 'subscribe' | 'unsubscribe' | 'delete', + ): Promise> { + return bulkContactQueue.add( + 'bulk-contact-action', + {projectId, contactIds, operation}, + { + jobId: `bulk-${operation}-${projectId}-${Date.now()}`, + }, + ); + } + + /** + * Get bulk action job status and progress + */ + public static async getBulkActionJobStatus(jobId: string) { + const job = await bulkContactQueue.getJob(jobId); + + if (!job) { + return null; + } + + const state = await job.getState(); + const progress = job.progress; + const returnValue = job.returnvalue; + const failedReason = job.failedReason; + + return { + id: job.id, + state, + progress, + result: returnValue, + data: job.data, + failedReason, + }; + } + /** * Queue segment count update job */ @@ -354,6 +415,7 @@ export class QueueService { segmentCountCounts, domainVerificationCounts, apiRequestCleanupCounts, + bulkContactCounts, ] = await Promise.all([ emailQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), campaignQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), @@ -363,6 +425,7 @@ export class QueueService { segmentCountQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), domainVerificationQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), apiRequestCleanupQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), + bulkContactQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), ]); return { @@ -374,6 +437,7 @@ export class QueueService { segmentCount: segmentCountCounts, domainVerification: domainVerificationCounts, apiRequestCleanup: apiRequestCleanupCounts, + bulkContact: bulkContactCounts, }; } @@ -390,6 +454,7 @@ export class QueueService { segmentCountQueue.pause(), domainVerificationQueue.pause(), apiRequestCleanupQueue.pause(), + bulkContactQueue.pause(), ]); } @@ -406,6 +471,7 @@ export class QueueService { segmentCountQueue.resume(), domainVerificationQueue.resume(), apiRequestCleanupQueue.resume(), + bulkContactQueue.resume(), ]); } @@ -430,6 +496,8 @@ export class QueueService { segmentCountQueue.clean(gracePeriod * 7, 50, 'failed'), domainVerificationQueue.clean(gracePeriod, 10, 'completed'), domainVerificationQueue.clean(gracePeriod * 7, 50, 'failed'), + bulkContactQueue.clean(gracePeriod, 50, 'completed'), + bulkContactQueue.clean(gracePeriod * 7, 100, 'failed'), ]); } @@ -514,6 +582,7 @@ export class QueueService { segmentCountQueue.close(), domainVerificationQueue.close(), apiRequestCleanupQueue.close(), + bulkContactQueue.close(), ]); } } diff --git a/apps/api/src/services/__tests__/ContactService.test.ts b/apps/api/src/services/__tests__/ContactService.test.ts index b587792..5f0621e 100644 --- a/apps/api/src/services/__tests__/ContactService.test.ts +++ b/apps/api/src/services/__tests__/ContactService.test.ts @@ -439,4 +439,271 @@ describe('ContactService - Duplicate Prevention & Data Merging', () => { expect(unsubscribed?.subscribed).toBe(false); }); }); + + describe('Bulk Contact Operations', () => { + describe('bulkSubscribe', () => { + it('should subscribe multiple unsubscribed contacts', async () => { + const contact1 = await factories.createContact({projectId, subscribed: false}); + const contact2 = await factories.createContact({projectId, subscribed: false}); + const contact3 = await factories.createContact({projectId, subscribed: false}); + + const result = await ContactService.bulkSubscribe(projectId, [contact1.id, contact2.id, contact3.id]); + + expect(result.updated).toBe(3); + + const contacts = await prisma.contact.findMany({ + where: {id: {in: [contact1.id, contact2.id, contact3.id]}}, + }); + + expect(contacts.every(c => c.subscribed)).toBe(true); + }); + + it('should only update unsubscribed contacts, not already subscribed ones', async () => { + const unsubscribed1 = await factories.createContact({projectId, subscribed: false}); + const unsubscribed2 = await factories.createContact({projectId, subscribed: false}); + const alreadySubscribed = await factories.createContact({projectId, subscribed: true}); + + const result = await ContactService.bulkSubscribe(projectId, [ + unsubscribed1.id, + unsubscribed2.id, + alreadySubscribed.id, + ]); + + expect(result.updated).toBe(2); + }); + + it('should return 0 if no contacts need updating', async () => { + const contact1 = await factories.createContact({projectId, subscribed: true}); + const contact2 = await factories.createContact({projectId, subscribed: true}); + + const result = await ContactService.bulkSubscribe(projectId, [contact1.id, contact2.id]); + + expect(result.updated).toBe(0); + }); + + it('should only update contacts belonging to the specified project', async () => { + const {project: otherProject} = await factories.createUserWithProject(); + const ownContact = await factories.createContact({projectId, subscribed: false}); + const otherContact = await factories.createContact({projectId: otherProject.id, subscribed: false}); + + const result = await ContactService.bulkSubscribe(projectId, [ownContact.id, otherContact.id]); + + expect(result.updated).toBe(1); + + const ownContactAfter = await prisma.contact.findUnique({where: {id: ownContact.id}}); + const otherContactAfter = await prisma.contact.findUnique({where: {id: otherContact.id}}); + + expect(ownContactAfter?.subscribed).toBe(true); + expect(otherContactAfter?.subscribed).toBe(false); + }); + + it('should handle empty contact IDs array', async () => { + const result = await ContactService.bulkSubscribe(projectId, []); + + expect(result.updated).toBe(0); + }); + + it('should handle non-existent contact IDs gracefully', async () => { + const result = await ContactService.bulkSubscribe(projectId, ['non-existent-1', 'non-existent-2']); + + expect(result.updated).toBe(0); + }); + + it('should handle large batches efficiently', async () => { + const contacts = await Promise.all( + Array.from({length: 150}, () => factories.createContact({projectId, subscribed: false})), + ); + const contactIds = contacts.map(c => c.id); + + const result = await ContactService.bulkSubscribe(projectId, contactIds); + + expect(result.updated).toBe(150); + + const updatedContacts = await prisma.contact.findMany({ + where: {id: {in: contactIds}}, + }); + + expect(updatedContacts.every(c => c.subscribed)).toBe(true); + }); + }); + + describe('bulkUnsubscribe', () => { + it('should unsubscribe multiple subscribed contacts', async () => { + const contact1 = await factories.createContact({projectId, subscribed: true}); + const contact2 = await factories.createContact({projectId, subscribed: true}); + const contact3 = await factories.createContact({projectId, subscribed: true}); + + const result = await ContactService.bulkUnsubscribe(projectId, [contact1.id, contact2.id, contact3.id]); + + expect(result.updated).toBe(3); + + const contacts = await prisma.contact.findMany({ + where: {id: {in: [contact1.id, contact2.id, contact3.id]}}, + }); + + expect(contacts.every(c => !c.subscribed)).toBe(true); + }); + + it('should only update subscribed contacts, not already unsubscribed ones', async () => { + const subscribed1 = await factories.createContact({projectId, subscribed: true}); + const subscribed2 = await factories.createContact({projectId, subscribed: true}); + const alreadyUnsubscribed = await factories.createContact({projectId, subscribed: false}); + + const result = await ContactService.bulkUnsubscribe(projectId, [ + subscribed1.id, + subscribed2.id, + alreadyUnsubscribed.id, + ]); + + expect(result.updated).toBe(2); + }); + + it('should return 0 if no contacts need updating', async () => { + const contact1 = await factories.createContact({projectId, subscribed: false}); + const contact2 = await factories.createContact({projectId, subscribed: false}); + + const result = await ContactService.bulkUnsubscribe(projectId, [contact1.id, contact2.id]); + + expect(result.updated).toBe(0); + }); + + it('should only update contacts belonging to the specified project', async () => { + const {project: otherProject} = await factories.createUserWithProject(); + const ownContact = await factories.createContact({projectId, subscribed: true}); + const otherContact = await factories.createContact({projectId: otherProject.id, subscribed: true}); + + const result = await ContactService.bulkUnsubscribe(projectId, [ownContact.id, otherContact.id]); + + expect(result.updated).toBe(1); + + const ownContactAfter = await prisma.contact.findUnique({where: {id: ownContact.id}}); + const otherContactAfter = await prisma.contact.findUnique({where: {id: otherContact.id}}); + + expect(ownContactAfter?.subscribed).toBe(false); + expect(otherContactAfter?.subscribed).toBe(true); + }); + + it('should handle empty contact IDs array', async () => { + const result = await ContactService.bulkUnsubscribe(projectId, []); + + expect(result.updated).toBe(0); + }); + + it('should handle non-existent contact IDs gracefully', async () => { + const result = await ContactService.bulkUnsubscribe(projectId, ['non-existent-1', 'non-existent-2']); + + expect(result.updated).toBe(0); + }); + }); + + describe('bulkDelete', () => { + it('should delete multiple contacts', async () => { + const contact1 = await factories.createContact({projectId}); + const contact2 = await factories.createContact({projectId}); + const contact3 = await factories.createContact({projectId}); + + const result = await ContactService.bulkDelete(projectId, [contact1.id, contact2.id, contact3.id]); + + expect(result.deleted).toBe(3); + + const contacts = await prisma.contact.findMany({ + where: {id: {in: [contact1.id, contact2.id, contact3.id]}}, + }); + + expect(contacts).toHaveLength(0); + }); + + it('should only delete contacts belonging to the specified project', async () => { + const {project: otherProject} = await factories.createUserWithProject(); + const ownContact = await factories.createContact({projectId}); + const otherContact = await factories.createContact({projectId: otherProject.id}); + + const result = await ContactService.bulkDelete(projectId, [ownContact.id, otherContact.id]); + + expect(result.deleted).toBe(1); + + const ownContactAfter = await prisma.contact.findUnique({where: {id: ownContact.id}}); + const otherContactAfter = await prisma.contact.findUnique({where: {id: otherContact.id}}); + + expect(ownContactAfter).toBeNull(); + expect(otherContactAfter).not.toBeNull(); + }); + + it('should handle empty contact IDs array', async () => { + const result = await ContactService.bulkDelete(projectId, []); + + expect(result.deleted).toBe(0); + }); + + it('should handle non-existent contact IDs gracefully', async () => { + const result = await ContactService.bulkDelete(projectId, ['non-existent-1', 'non-existent-2']); + + expect(result.deleted).toBe(0); + }); + + it('should handle large batches efficiently', async () => { + const contacts = await Promise.all(Array.from({length: 200}, () => factories.createContact({projectId}))); + const contactIds = contacts.map(c => c.id); + + const result = await ContactService.bulkDelete(projectId, contactIds); + + expect(result.deleted).toBe(200); + + const remainingContacts = await prisma.contact.findMany({ + where: {id: {in: contactIds}}, + }); + + expect(remainingContacts).toHaveLength(0); + }); + + it('should delete both subscribed and unsubscribed contacts', async () => { + const subscribed = await factories.createContact({projectId, subscribed: true}); + const unsubscribed = await factories.createContact({projectId, subscribed: false}); + + const result = await ContactService.bulkDelete(projectId, [subscribed.id, unsubscribed.id]); + + expect(result.deleted).toBe(2); + }); + + it('should handle partial matches (some exist, some do not)', async () => { + const existingContact = await factories.createContact({projectId}); + + const result = await ContactService.bulkDelete(projectId, [existingContact.id, 'non-existent-id']); + + expect(result.deleted).toBe(1); + + const contact = await prisma.contact.findUnique({where: {id: existingContact.id}}); + expect(contact).toBeNull(); + }); + }); + + describe('Bulk Operations - Project Isolation', () => { + it('should never leak contacts between projects in bulk operations', async () => { + const {project: project1} = await factories.createUserWithProject(); + const {project: project2} = await factories.createUserWithProject(); + + const p1Contact1 = await factories.createContact({projectId: project1.id, subscribed: false}); + const p1Contact2 = await factories.createContact({projectId: project1.id, subscribed: false}); + const p2Contact1 = await factories.createContact({projectId: project2.id, subscribed: false}); + const p2Contact2 = await factories.createContact({projectId: project2.id, subscribed: false}); + + await ContactService.bulkSubscribe(project1.id, [ + p1Contact1.id, + p1Contact2.id, + p2Contact1.id, + p2Contact2.id, + ]); + + const p1ContactsAfter = await prisma.contact.findMany({ + where: {projectId: project1.id}, + }); + const p2ContactsAfter = await prisma.contact.findMany({ + where: {projectId: project2.id}, + }); + + expect(p1ContactsAfter.every(c => c.subscribed)).toBe(true); + expect(p2ContactsAfter.every(c => !c.subscribed)).toBe(true); + }); + }); + }); }); diff --git a/apps/web/src/pages/contacts/index.tsx b/apps/web/src/pages/contacts/index.tsx index 3e28bd3..5479063 100644 --- a/apps/web/src/pages/contacts/index.tsx +++ b/apps/web/src/pages/contacts/index.tsx @@ -5,6 +5,7 @@ import { CardDescription, CardHeader, CardTitle, + Checkbox, ConfirmDialog, Dialog, DialogContent, @@ -62,6 +63,9 @@ export default function ContactsPage() { const [showDeleteDialog, setShowDeleteDialog] = useState(false); const [contactToDelete, setContactToDelete] = useState(null); const [totalCount, setTotalCount] = useState(0); + const [selectedContacts, setSelectedContacts] = useState>(new Set()); + const [showBulkActionsDialog, setShowBulkActionsDialog] = useState(false); + const [bulkOperation, setBulkOperation] = useState<'subscribe' | 'unsubscribe' | 'delete' | null>(null); const pageSize = 50; const {data, mutate, isLoading} = useSWR( @@ -93,6 +97,7 @@ export default function ContactsPage() { const newPage = currentPage + 1; setCursor(data.cursor); setCurrentPage(newPage); + setSelectedContacts(new Set()); // Clear selection on page change // Store cursor in history if not already there if (cursorHistory.length <= newPage) { @@ -107,9 +112,37 @@ export default function ContactsPage() { const previousCursor = cursorHistory[newPage]; setCursor(previousCursor); setCurrentPage(newPage); + setSelectedContacts(new Set()); // Clear selection on page change } }; + const handleSelectAll = () => { + if (selectedContacts.size === contacts.length && contacts.length > 0) { + setSelectedContacts(new Set()); + } else { + setSelectedContacts(new Set(contacts.map(c => c.id))); + } + }; + + const handleSelectContact = (contactId: string) => { + const newSelected = new Set(selectedContacts); + if (newSelected.has(contactId)) { + newSelected.delete(contactId); + } else { + newSelected.add(contactId); + } + setSelectedContacts(newSelected); + }; + + const handleBulkAction = (operation: 'subscribe' | 'unsubscribe' | 'delete') => { + setBulkOperation(operation); + setShowBulkActionsDialog(true); + }; + + const clearSelection = () => { + setSelectedContacts(new Set()); + }; + const promptDelete = (contactId: string) => { setContactToDelete(contactId); setShowDeleteDialog(true); @@ -194,6 +227,50 @@ export default function ContactsPage() { + {/* Bulk Actions Toolbar */} + {selectedContacts.size > 0 && ( + + +
+
+ + {selectedContacts.size} contact{selectedContacts.size !== 1 ? 's' : ''} selected + +
+ + + +
+
+ +
+ + + )} + {/* Contacts Table */} @@ -244,6 +321,12 @@ export default function ContactsPage() { + @@ -261,6 +344,12 @@ export default function ContactsPage() { {contacts.map(contact => ( +
+ 0} + onCheckedChange={handleSelectAll} + /> + Email
+ handleSelectContact(contact.id)} + /> +
{contact.subscribed ? ( @@ -399,6 +488,18 @@ export default function ContactsPage() { {/* Import Contacts Dialog */} mutate()} /> + {/* Bulk Actions Dialog */} + { + mutate(); + clearSelection(); + }} + /> + {/* Delete Confirmation Dialog */} ); } + +interface BulkActionsDialogProps { + open: boolean; + onOpenChange: (open: boolean) => void; + operation: 'subscribe' | 'unsubscribe' | 'delete' | null; + contactIds: string[]; + onSuccess: () => void; +} + +interface BulkActionResult { + operation: string; + totalRequested: number; + successCount: number; + failureCount: number; + errors: Array<{contactId: string; email: string; error: string}>; +} + +function BulkActionsDialog({open, onOpenChange, operation, contactIds, onSuccess}: BulkActionsDialogProps) { + const [jobId, setJobId] = useState(null); + const [isProcessing, setIsProcessing] = useState(false); + const [progress, setProgress] = useState(0); + const [status, setStatus] = useState<'idle' | 'processing' | 'completed' | 'failed'>('idle'); + const [result, setResult] = useState(null); + const [errorMessage, setErrorMessage] = useState(null); + const pollIntervalRef = useRef(null); + const [showCloseConfirmDialog, setShowCloseConfirmDialog] = useState(false); + + // Clean up polling on unmount or dialog close + useEffect(() => { + if (!open) { + if (pollIntervalRef.current) { + clearInterval(pollIntervalRef.current); + pollIntervalRef.current = null; + } + setTimeout(() => { + setJobId(null); + setProgress(0); + setStatus('idle'); + setResult(null); + setErrorMessage(null); + }, 300); + } + }, [open]); + + const pollJobStatus = async (jobId: string) => { + try { + const response = await network.fetch<{ + id: string; + state: string; + progress: number; + result: BulkActionResult | null; + failedReason?: string; + }>('GET', `/contacts/bulk/${jobId}`); + + setProgress(response.progress || 0); + + if (response.state === 'completed') { + setStatus('completed'); + setResult(response.result); + if (pollIntervalRef.current) { + clearInterval(pollIntervalRef.current); + pollIntervalRef.current = null; + } + + if (response.result) { + const {successCount, failureCount} = response.result; + toast.success(`Completed: ${successCount} succeeded${failureCount > 0 ? `, ${failureCount} failed` : ''}`); + } + + onSuccess(); + } else if (response.state === 'failed') { + setStatus('failed'); + if (pollIntervalRef.current) { + clearInterval(pollIntervalRef.current); + pollIntervalRef.current = null; + } + const errorMsg = response.failedReason || 'Operation failed'; + setErrorMessage(errorMsg); + toast.error(errorMsg); + } else if (response.state === 'active') { + setStatus('processing'); + } + } catch (error) { + console.error('Failed to poll job status:', error); + if (pollIntervalRef.current) { + clearInterval(pollIntervalRef.current); + pollIntervalRef.current = null; + } + setStatus('failed'); + toast.error('Failed to check operation status'); + } + }; + + const handleConfirm = async () => { + if (!operation) return; + + setIsProcessing(true); + setStatus('processing'); + + try { + const endpoint = `/contacts/bulk-${operation}`; + const data = await network.fetch<{jobId: string; message: string}, typeof ContactSchemas.bulkAction>( + 'POST', + endpoint, + {contactIds}, + ); + + setJobId(data.jobId); + + // Start polling for job status + pollIntervalRef.current = setInterval(() => { + void pollJobStatus(data.jobId); + }, 1000); + } catch (error) { + const errorMsg = error instanceof Error ? error.message : 'Failed to start operation'; + setErrorMessage(errorMsg); + toast.error(errorMsg); + setStatus('failed'); + } finally { + setIsProcessing(false); + } + }; + + const handleClose = () => { + if (status === 'processing') { + setShowCloseConfirmDialog(true); + return; + } + onOpenChange(false); + }; + + const confirmClose = () => { + onOpenChange(false); + }; + + const getOperationLabel = () => { + switch (operation) { + case 'subscribe': return 'Subscribe'; + case 'unsubscribe': return 'Unsubscribe'; + case 'delete': return 'Delete'; + default: return 'Process'; + } + }; + + const getOperationColor = () => { + switch (operation) { + case 'subscribe': return 'green'; + case 'unsubscribe': return 'yellow'; + case 'delete': return 'red'; + default: return 'blue'; + } + }; + + return ( + <> + + + + {getOperationLabel()} Contacts + + +
+ {status === 'idle' && ( +
+

+ Are you sure you want to {operation} {contactIds.length} contact + {contactIds.length !== 1 ? 's' : ''}? +

+ {operation === 'delete' && ( +

+ This action cannot be undone. +

+ )} +
+ )} + + {(status === 'processing') && ( +
+
+ Processing contacts... + {progress}% +
+
+
+
+
+ )} + + {status === 'completed' && result && ( +
+
+
+
+ +
{result.successCount}
+
+
Succeeded
+
+ {result.failureCount > 0 && ( +
+
+ +
{result.failureCount}
+
+
Failed
+
+ )} +
+ + {result.errors && result.errors.length > 0 && ( +
+

Errors

+
+ {result.errors.slice(0, 10).map((error, idx) => ( +
{error.error}
+ ))} + {result.errors.length > 10 && ( +
+ ...and {result.errors.length - 10} more errors +
+ )} +
+
+ )} +
+ )} + + {status === 'failed' && ( +
+
+ + Operation failed +
+

+ {errorMessage || 'Please try again.'} +

+
+ )} +
+ + + {status === 'idle' ? ( + <> + + + + ) : status === 'completed' ? ( + + ) : ( + + )} + + +
+ + + + ); +} diff --git a/packages/shared/src/schemas/index.ts b/packages/shared/src/schemas/index.ts index 3650b9c..58f049c 100644 --- a/packages/shared/src/schemas/index.ts +++ b/packages/shared/src/schemas/index.ts @@ -69,7 +69,10 @@ export const ContactSchemas = { subscribed: z.boolean().default(true), data: jsonSchema.optional(), }), -}; + bulkAction: z.object({ + contactIds: z.array(uuid).min(1).max(1000), + }), +} as const; const segmentFilterSchema = z.object({ field: z.string().min(1), diff --git a/test/helpers/database.ts b/test/helpers/database.ts index 96dd4c2..7f933cd 100644 --- a/test/helpers/database.ts +++ b/test/helpers/database.ts @@ -63,54 +63,79 @@ class TestDatabase { * Clean up database after each test * Deletes all records in reverse order of dependencies * Uses batched deletes to prevent memory issues with large datasets + * Retries on deadlock to handle race conditions with background event tracking */ async cleanup() { if (!this.prisma) return; - try { - // Use a transaction to ensure all deletes happen atomically - // This prevents foreign key constraint violations and race conditions - await this.prisma.$transaction([ - // Level 1: Delete deepest dependencies first - this.prisma.event.deleteMany(), - this.prisma.workflowStepExecution.deleteMany(), + const maxRetries = 3; + let lastError: Error | null = null; - // Level 2: Delete entities that depend on Level 1 - this.prisma.email.deleteMany(), - this.prisma.workflowExecution.deleteMany(), + for (let attempt = 1; attempt <= maxRetries; attempt++) { + try { + // Use a transaction to ensure all deletes happen atomically + // This prevents foreign key constraint violations and race conditions + await this.prisma.$transaction([ + // Level 1: Delete deepest dependencies first + this.prisma.event.deleteMany(), + this.prisma.workflowStepExecution.deleteMany(), - // Level 3: Delete workflow structure - this.prisma.workflowTransition.deleteMany(), - this.prisma.workflowStep.deleteMany(), - this.prisma.workflow.deleteMany(), + // Level 2: Delete entities that depend on Level 1 + this.prisma.email.deleteMany(), + this.prisma.workflowExecution.deleteMany(), - // Level 4: Delete campaigns and templates - this.prisma.campaign.deleteMany(), - this.prisma.template.deleteMany(), + // Level 3: Delete workflow structure + this.prisma.workflowTransition.deleteMany(), + this.prisma.workflowStep.deleteMany(), + this.prisma.workflow.deleteMany(), - // Level 5: Delete segment relationships - this.prisma.segmentMembership.deleteMany(), - this.prisma.segment.deleteMany(), + // Level 4: Delete campaigns and templates + this.prisma.campaign.deleteMany(), + this.prisma.template.deleteMany(), - // Level 6: Delete contacts - this.prisma.contact.deleteMany(), + // Level 5: Delete segment relationships + this.prisma.segmentMembership.deleteMany(), + this.prisma.segment.deleteMany(), - // Level 7: Delete domains - this.prisma.domain.deleteMany(), + // Level 6: Delete contacts + this.prisma.contact.deleteMany(), - // Level 8: Delete memberships (has FK to both user and project) - this.prisma.membership.deleteMany(), + // Level 7: Delete domains + this.prisma.domain.deleteMany(), - // Level 9: Delete projects - this.prisma.project.deleteMany(), + // Level 8: Delete memberships (has FK to both user and project) + this.prisma.membership.deleteMany(), - // Level 10: Delete users last - this.prisma.user.deleteMany(), - ]); - } catch (error) { - console.error('Error cleaning up database:', error); - throw error; + // Level 9: Delete projects + this.prisma.project.deleteMany(), + + // Level 10: Delete users last + this.prisma.user.deleteMany(), + ]); + + // Success - exit retry loop + return; + } catch (error) { + lastError = error as Error; + + // Check if this is a deadlock error (PostgreSQL error code 40P01) + const isDeadlock = error instanceof Error && error.message?.includes('deadlock detected'); + + if (isDeadlock && attempt < maxRetries) { + // Wait before retrying (exponential backoff) + const delay = Math.pow(2, attempt) * 50; // 100ms, 200ms, 400ms + await new Promise(resolve => setTimeout(resolve, delay)); + continue; + } + + // Not a deadlock or out of retries + break; + } } + + // If we get here, all retries failed + console.error(`Error cleaning up database after ${maxRetries} attempts:`, lastError); + throw lastError; } /**