vynte-infra/plunk
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement bulk contact action selector for improved flexibility in bulk operations

Browse Source
This commit is contained in:
Dries Augustyns
2026-05-10 10:40:40 +02:00
parent 7658a59b5d
commit de6335e999
8 changed files with 713 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/controllers/Contacts.ts
+38 -73
View File
@@ -1,6 +1,8 @@
import {Controller, Delete, Get, Middleware, Patch, Post} from '@overnightjs/core';
import type {NextFunction, Request, Response} from 'express';
import multer from 'multer';
import {ContactSchemas} from '@plunk/shared';
import type {BulkContactActionSelector} from '@plunk/types';
import signale from 'signale';
import {requireAuth, requireEmailVerified} from '../middleware/auth.js';
import {ContactService} from '../services/ContactService.js';
@@ -424,31 +426,7 @@ export class Contacts {
@Middleware([requireAuth, requireEmailVerified])
@CatchAsync
public async bulkSubscribe(req: Request, res: Response, _next: NextFunction) {
const auth = res.locals.auth;
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',
});
}
return queueBulkAction(req, res, 'subscribe');
}
/**
@@ -459,30 +437,7 @@ export class Contacts {
@Middleware([requireAuth, requireEmailVerified])
@CatchAsync
public async bulkUnsubscribe(req: Request, res: Response, _next: NextFunction) {
const auth = res.locals.auth;
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',
});
}
return queueBulkAction(req, res, 'unsubscribe');
}
/**
@@ -493,30 +448,7 @@ export class Contacts {
@Middleware([requireAuth, requireEmailVerified])
@CatchAsync
public async bulkDelete(req: Request, res: Response, _next: NextFunction) {
const auth = res.locals.auth;
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',
});
}
return queueBulkAction(req, res, 'delete');
}
/**
@@ -550,3 +482,36 @@ export class Contacts {
}
}
}
async function queueBulkAction(
req: Request,
res: Response,
operation: 'subscribe' | 'unsubscribe' | 'delete',
) {
const auth = res.locals.auth;
const parsed = ContactSchemas.bulkAction.safeParse(req.body);
if (!parsed.success) {
return res.status(400).json({
error: parsed.error.errors[0]?.message ?? 'Invalid bulk action payload',
});
}
const selector: BulkContactActionSelector =
parsed.data.mode === 'ids'
? {mode: 'ids', contactIds: parsed.data.contactIds}
: {mode: 'query', filter: parsed.data.filter, excludeIds: parsed.data.excludeIds};
try {
const job = await QueueService.queueBulkContactAction(auth.projectId!, selector, operation);
return res.status(202).json({
message: `Bulk ${operation} queued successfully`,
jobId: job.id,
});
} catch (error) {
signale.error(`[CONTACTS] Failed to queue bulk ${operation}:`, error);
return res.status(500).json({
error: error instanceof Error ? error.message : `Failed to queue bulk ${operation}`,
});
}
}
apps/api/src/jobs/bulk-contact-processor.ts
+122 -49
View File
@@ -3,73 +3,149 @@
* Processes bulk subscribe, unsubscribe, and delete operations
*/
import type {BulkContactActionJobData} from '@plunk/types';
import {Prisma} from '@plunk/db';
import type {BulkContactActionJobData, BulkContactActionSelector} from '@plunk/types';
import {type Job, Worker} from 'bullmq';
import signale from 'signale';
import {prisma} from '../database/prisma.js';
import {ContactService} from '../services/ContactService.js';
import {bulkContactQueue} from '../services/QueueService.js';
const BATCH_SIZE = 100; // Process contacts in batches of 100
const BATCH_SIZE = 100;
interface BulkActionResult {
operation: 'subscribe' | 'unsubscribe' | 'delete';
totalRequested: number;
/** Contacts whose state was actually changed by this run. */
successCount: number;
/** Subscribe/unsubscribe only: contacts already in the target state. */
unchangedCount: number;
/** Contacts that errored or weren't found (e.g. wrong project). */
failureCount: number;
errors: {contactId: string; email: string; error: string}[];
}
function buildQueryWhere(projectId: string, selector: Extract<BulkContactActionSelector, {mode: 'query'}>): Prisma.ContactWhereInput {
const search = selector.filter?.search;
const excludeIds = selector.excludeIds ?? [];
return {
projectId,
...(search ? {email: {contains: search, mode: 'insensitive' as const}} : {}),
...(excludeIds.length > 0 ? {id: {notIn: excludeIds}} : {}),
};
}
async function applyBatch(
projectId: string,
operation: BulkActionResult['operation'],
ids: string[],
): Promise<{changed: number; unchanged: number}> {
switch (operation) {
case 'subscribe': {
const r = await ContactService.bulkSubscribe(projectId, ids);
return {changed: r.updated, unchanged: r.unchanged};
}
case 'unsubscribe': {
const r = await ContactService.bulkUnsubscribe(projectId, ids);
return {changed: r.updated, unchanged: r.unchanged};
}
case 'delete': {
const r = await ContactService.bulkDelete(projectId, ids);
return {changed: r.deleted, unchanged: 0};
}
}
}
export function createBulkContactWorker() {
const worker = new Worker<BulkContactActionJobData>(
bulkContactQueue.name,
async (job: Job<BulkContactActionJobData>) => {
const {projectId, contactIds, operation} = job.data;
signale.info(
`[BULK-CONTACT-PROCESSOR] Processing ${operation} for ${contactIds.length} contacts in project ${projectId}`,
);
const {projectId, operation, selector} = job.data;
const result: BulkActionResult = {
operation,
totalRequested: contactIds.length,
totalRequested: 0,
successCount: 0,
unchangedCount: 0,
failureCount: 0,
errors: [],
};
try {
// Process contacts in batches
if (selector.mode === 'ids') {
const {contactIds} = selector;
result.totalRequested = contactIds.length;
signale.info(
`[BULK-CONTACT-PROCESSOR] Processing ${operation} for ${contactIds.length} contacts (ids mode) in project ${projectId}`,
);
for (let i = 0; i < contactIds.length; i += BATCH_SIZE) {
const batchIds = contactIds.slice(i, Math.min(i + BATCH_SIZE, contactIds.length));
const batchIds = contactIds.slice(i, i + BATCH_SIZE);
try {
const {changed, unchanged} = await applyBatch(projectId, operation, batchIds);
result.successCount += changed;
result.unchangedCount += unchanged;
const failed = batchIds.length - changed - unchanged;
if (failed > 0) result.failureCount += failed;
} 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',
});
}
await job.updateProgress(Math.round(((i + batchIds.length) / contactIds.length) * 100));
}
} else {
const where = buildQueryWhere(projectId, selector);
const total = await prisma.contact.count({where});
result.totalRequested = total;
signale.info(
`[BULK-CONTACT-PROCESSOR] Processing ${operation} for ${total} contacts (query mode) in project ${projectId}`,
);
if (total === 0) {
await job.updateProgress(100);
return result;
}
// Cursor-based iteration over matching contacts. We re-evaluate the where clause
// each batch (with id < cursor) instead of Prisma's `cursor:` because for `delete`
// the rows we just processed disappear — a stable cursor would either skip survivors
// or revisit deletions. Sorting by id desc + `id < lastId` is idempotent under either.
let lastId: string | undefined;
let processedRows = 0;
// Cap the loop so a runaway query (e.g. growing table) can't spin forever.
const maxIterations = Math.ceil(total / BATCH_SIZE) + 50;
for (let iter = 0; iter < maxIterations; iter += 1) {
const batch = await prisma.contact.findMany({
where: {
...where,
...(lastId ? {id: {...(where.id as object | undefined), lt: lastId}} : {}),
},
select: {id: true},
orderBy: {id: 'desc'},
take: BATCH_SIZE,
});
if (batch.length === 0) break;
const batchIds = batch.map(c => c.id);
lastId = batchIds[batchIds.length - 1];
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
}
const {changed, unchanged} = await applyBatch(projectId, operation, batchIds);
result.successCount += changed;
result.unchangedCount += unchanged;
const failed = batchIds.length - changed - unchanged;
if (failed > 0) result.failureCount += failed;
} catch (error) {
signale.error(`[BULK-CONTACT-PROCESSOR] Batch failed:`, error);
signale.error('[BULK-CONTACT-PROCESSOR] Batch failed:', error);
result.failureCount += batchIds.length;
result.errors.push({
contactId: 'batch',
@@ -78,24 +154,21 @@ export function createBulkContactWorker() {
});
}
// Update progress
const progress = Math.round(((i + batchIds.length) / contactIds.length) * 100);
await job.updateProgress(progress);
processedRows += batchIds.length;
await job.updateProgress(Math.min(100, Math.round((processedRows / total) * 100)));
if (batch.length < BATCH_SIZE) break;
}
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;
}
signale.info(
`[BULK-CONTACT-PROCESSOR] ${operation} completed: ${result.successCount} succeeded, ${result.failureCount} failed`,
);
return result;
},
{
connection: bulkContactQueue.opts.connection,
concurrency: 3, // Process max 3 bulk operations concurrently
concurrency: 3,
},
);
apps/api/src/services/ContactService.ts
+30 -48
View File
@@ -721,99 +721,81 @@ export class ContactService {
/**
* Bulk subscribe contacts
* Updates multiple contacts to subscribed=true in batches
* Updates multiple contacts to subscribed=true in batches.
* `updated` = contacts flipped from unsubscribed to subscribed.
* `unchanged` = contacts that were already subscribed (no-op, not a failure).
*/
public static async bulkSubscribe(projectId: string, contactIds: string[]): Promise<{updated: number}> {
// Verify all contacts belong to this project
public static async bulkSubscribe(
projectId: string,
contactIds: string[],
): Promise<{updated: number; unchanged: number}> {
const contacts = await prisma.contact.findMany({
where: {
id: {in: contactIds},
projectId,
},
where: {id: {in: contactIds}, projectId},
select: {id: true, subscribed: true},
});
const validIds = contacts.map(c => c.id);
if (validIds.length === 0) {
return {updated: 0};
if (contacts.length === 0) {
return {updated: 0, unchanged: 0};
}
// Only update contacts that are currently unsubscribed
const unsubscribedIds = contacts.filter(c => !c.subscribed).map(c => c.id);
const unchanged = contacts.length - unsubscribedIds.length;
if (unsubscribedIds.length === 0) {
return {updated: 0};
return {updated: 0, unchanged};
}
// Update in a single query for performance
const result = await prisma.contact.updateMany({
where: {
id: {in: unsubscribedIds},
projectId,
},
data: {
subscribed: true,
},
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};
return {updated: result.count, unchanged};
}
/**
* Bulk unsubscribe contacts
* Bulk unsubscribe contacts.
* `updated` = contacts flipped from subscribed to unsubscribed.
* `unchanged` = contacts that were already unsubscribed (no-op, not a failure).
*/
public static async bulkUnsubscribe(projectId: string, contactIds: string[]): Promise<{updated: number}> {
public static async bulkUnsubscribe(
projectId: string,
contactIds: string[],
): Promise<{updated: number; unchanged: number}> {
const contacts = await prisma.contact.findMany({
where: {
id: {in: contactIds},
projectId,
},
where: {id: {in: contactIds}, projectId},
select: {id: true, subscribed: true},
});
const validIds = contacts.map(c => c.id);
if (validIds.length === 0) {
return {updated: 0};
if (contacts.length === 0) {
return {updated: 0, unchanged: 0};
}
// Only update contacts that are currently subscribed
const subscribedIds = contacts.filter(c => c.subscribed).map(c => c.id);
const unchanged = contacts.length - subscribedIds.length;
if (subscribedIds.length === 0) {
return {updated: 0};
return {updated: 0, unchanged};
}
const result = await prisma.contact.updateMany({
where: {
id: {in: subscribedIds},
projectId,
},
data: {
subscribed: false,
},
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};
return {updated: result.count, unchanged};
}
/**
apps/api/src/services/QueueService.ts
+3 -2
View File
@@ -5,6 +5,7 @@ import signale from 'signale';
import type {
ApiRequestCleanupJobData,
BulkContactActionJobData,
BulkContactActionSelector,
CampaignBatchJobData,
ContactImportJobData,
DomainVerificationJobData,
@@ -350,12 +351,12 @@ export class QueueService {
*/
public static async queueBulkContactAction(
projectId: string,
contactIds: string[],
selector: BulkContactActionSelector,
operation: 'subscribe' | 'unsubscribe' | 'delete',
): Promise<Job<BulkContactActionJobData>> {
return bulkContactQueue.add(
'bulk-contact-action',
{projectId, contactIds, operation},
{projectId, operation, selector},
{
jobId: `bulk-${operation}-${projectId}-${Date.now()}`,
},