diff --git a/apps/api/src/jobs/import-processor.ts b/apps/api/src/jobs/import-processor.ts index 9160768..65a499f 100644 --- a/apps/api/src/jobs/import-processor.ts +++ b/apps/api/src/jobs/import-processor.ts @@ -6,7 +6,9 @@ import {type Job, Worker} from 'bullmq'; import {parse} from 'csv-parse/sync'; +import {prisma} from '../database/prisma.js'; import {ContactService} from '../services/ContactService.js'; +import {NtfyService} from '../services/NtfyService.js'; import {type ContactImportJobData, importQueue} from '../services/QueueService.js'; const BATCH_SIZE = 100; // Process contacts in batches of 100 @@ -28,6 +30,14 @@ export function createImportWorker() { console.log(`[IMPORT-PROCESSOR] Processing import for project ${projectId} (${filename})`); + // Fetch project information for notifications + const project = await prisma.project.findUnique({ + where: {id: projectId}, + select: {name: true}, + }); + + const projectName = project?.name || projectId; + const result: ImportResult = { totalRows: 0, successCount: 0, @@ -58,6 +68,9 @@ export function createImportWorker() { console.log(`[IMPORT-PROCESSOR] Parsed ${records.length} rows from CSV`); + // Notify that import has started + await NtfyService.notifyContactImportStarted(projectName, projectId, filename, result.totalRows); + // Validate that 'email' column exists const firstRecord = records[0]; if (firstRecord && typeof firstRecord === 'object' && !('email' in firstRecord)) { @@ -142,15 +155,30 @@ export function createImportWorker() { `[IMPORT-PROCESSOR] Import completed: ${result.createdCount} created, ${result.updatedCount} updated, ${result.failureCount} failed`, ); + // Notify that import has completed + await NtfyService.notifyContactImportCompleted( + projectName, + projectId, + filename, + result.successCount, + result.createdCount, + result.updatedCount, + result.failureCount, + ); + return result; } catch (error) { console.error(`[IMPORT-PROCESSOR] Failed to process import:`, error); + // Notify that import has failed + const errorMessage = error instanceof Error ? error.message : 'Unknown error'; + await NtfyService.notifyContactImportFailed(projectName, projectId, filename, errorMessage); + // Return partial results with error result.errors.push({ row: 0, email: '', - error: error instanceof Error ? error.message : 'Unknown error', + error: errorMessage, }); throw error; // Re-throw to mark job as failed diff --git a/apps/api/src/jobs/segment-count-processor.ts b/apps/api/src/jobs/segment-count-processor.ts index e80e2bb..26ccd70 100644 --- a/apps/api/src/jobs/segment-count-processor.ts +++ b/apps/api/src/jobs/segment-count-processor.ts @@ -7,6 +7,7 @@ import {type Job, Worker} from 'bullmq'; import signale from 'signale'; import {prisma} from '../database/prisma.js'; +import {NtfyService} from '../services/NtfyService.js'; import {type SegmentCountJobData, segmentCountQueue} from '../services/QueueService.js'; import {SegmentService} from '../services/SegmentService.js'; @@ -42,6 +43,11 @@ async function processProjectSegments(projectId: string, projectName?: string): signale.success( `[SEGMENT-COUNT-WORKER] Segment "${segment.name}": +${result.added} entries, -${result.removed} exits, ${result.total} total members`, ); + + // Notify about segment membership update (only for tracked segments) + if (projectName) { + await NtfyService.notifySegmentMembershipComputed(segment.name, projectName, projectId, result.total); + } } catch (error) { signale.error(`[SEGMENT-COUNT-WORKER] Failed to compute membership for segment ${segment.id}:`, error); // Continue with other segments diff --git a/apps/api/src/services/SegmentService.ts b/apps/api/src/services/SegmentService.ts index d393020..e5d59d3 100644 --- a/apps/api/src/services/SegmentService.ts +++ b/apps/api/src/services/SegmentService.ts @@ -5,6 +5,7 @@ import {prisma} from '../database/prisma.js'; import {HttpException} from '../exceptions/index.js'; import {EventService} from './EventService.js'; +import {NtfyService} from './NtfyService.js'; // Re-export types for use in other services export type {FilterCondition, FilterGroup, SegmentFilter} from '@plunk/types'; @@ -115,7 +116,7 @@ export class SegmentService { const where = this.buildWhereClause(projectId, data.condition); const memberCount = await prisma.contact.count({where}); - return prisma.segment.create({ + const segment = await prisma.segment.create({ data: { projectId, name: data.name, @@ -124,7 +125,17 @@ export class SegmentService { trackMembership: data.trackMembership ?? false, memberCount, }, + include: { + project: { + select: {name: true}, + }, + }, }); + + // Notify about segment creation + await NtfyService.notifySegmentCreated(segment.name, segment.project.name, projectId); + + return segment; } /** @@ -178,7 +189,21 @@ export class SegmentService { */ public static async delete(projectId: string, segmentId: string): Promise { // First verify segment exists and belongs to project - await this.get(projectId, segmentId); + const segment = await prisma.segment.findFirst({ + where: { + id: segmentId, + projectId, + }, + include: { + project: { + select: {name: true}, + }, + }, + }); + + if (!segment) { + throw new HttpException(404, 'Segment not found'); + } // Check if segment is used in any active campaigns const campaignsUsingSegment = await prisma.campaign.count({ @@ -200,6 +225,9 @@ export class SegmentService { await prisma.segment.delete({ where: {id: segmentId}, }); + + // Notify about segment deletion + await NtfyService.notifySegmentDeleted(segment.name, segment.project.name, projectId); } /**