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

Add contact data

Browse Source
This commit is contained in:
Dries Augustyns
2025-12-03 13:28:41 +01:00
parent d96bce75d7
commit 8defb9f8ab
9 changed files with 79 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/jobs/email-processor.ts
+5 -1
View File
@@ -3,7 +3,6 @@
* Processes individual emails from the queue (for all sources: transactional, campaign, workflow)
*/
import type {Prisma} from '@plunk/db';
import {EmailSourceType, EmailStatus} from '@plunk/db';
import {type Job, Worker} from 'bullmq';
@@ -13,6 +12,7 @@ import {EventService} from '../services/EventService.js';
import {MeterService} from '../services/MeterService.js';
import {emailQueue, type SendEmailJobData} from '../services/QueueService.js';
import {sendRawEmail} from '../services/SESService.js';
import {DASHBOARD_URI} from '../app/constants.js';
export function createEmailWorker() {
const worker = new Worker<SendEmailJobData>(
@@ -64,6 +64,10 @@ export function createEmailWorker() {
data: {
email: email.contact.email,
...contactData,
data: contactData,
unsubscribeUrl: `${DASHBOARD_URI}/unsubscribe/${email.contact.id}`,
subscribeUrl: `${DASHBOARD_URI}/subscribe/${email.contact.id}`,
manageUrl: `${DASHBOARD_URI}/manage/${email.contact.id}`,
},
});
apps/api/src/services/CampaignService.ts
+60 -56
View File
@@ -10,6 +10,8 @@ import {DomainService} from './DomainService.js';
import {EmailService} from './EmailService.js';
import {QueueService} from './QueueService.js';
import {SegmentService} from './SegmentService.js';
import {DASHBOARD_URI} from '../app/constants.js';
import {sendRawEmail} from './SESService.js';
const BATCH_SIZE = 500; // Number of emails to process per batch (increased for better performance)
@@ -369,6 +371,10 @@ export class CampaignService {
const variables = {
email: contact.email,
...contactData,
data: contactData,
unsubscribeUrl: `${DASHBOARD_URI}/unsubscribe/${contact.id}`,
subscribeUrl: `${DASHBOARD_URI}/subscribe/${contact.id}`,
manageUrl: `${DASHBOARD_URI}/manage/${contact.id}`,
};
const renderedSubject = EmailService.format({
@@ -505,6 +511,60 @@ export class CampaignService {
};
}
/**
* Send a test email for a campaign
*/
public static async sendTest(projectId: string, campaignId: string, testEmail: string): Promise<void> {
const campaign = await this.get(projectId, campaignId);
// Validate that the test email belongs to a project member
const membership = await prisma.membership.findFirst({
where: {
projectId,
user: {
email: testEmail,
},
},
include: {
user: true,
},
});
if (!membership) {
throw new HttpException(403, 'Test emails can only be sent to project members');
}
// Verify domain is registered and verified before sending
await DomainService.verifyEmailDomain(campaign.from, projectId);
// Get project to validate from address
const project = await prisma.project.findUnique({
where: {id: projectId},
});
if (!project) {
throw new HttpException(404, 'Project not found');
}
// Prepare the email content (no variable replacement for test emails)
await sendRawEmail({
from: {
name: campaign.fromName || project.name || 'Plunk',
email: campaign.from,
},
to: [testEmail],
content: {
subject: `[TEST] ${campaign.subject}`,
html: campaign.body,
},
reply: campaign.replyTo || undefined,
headers: {
'X-Plunk-Test': 'true',
},
tracking: false, // Disable tracking for test emails
});
}
/**
* Get recipient count for a campaign
*/
@@ -632,60 +692,4 @@ export class CampaignService {
...segmentWhere,
};
}
/**
* Send a test email for a campaign
*/
public static async sendTest(projectId: string, campaignId: string, testEmail: string): Promise<void> {
const campaign = await this.get(projectId, campaignId);
// Validate that the test email belongs to a project member
const membership = await prisma.membership.findFirst({
where: {
projectId,
user: {
email: testEmail,
},
},
include: {
user: true,
},
});
if (!membership) {
throw new HttpException(403, 'Test emails can only be sent to project members');
}
// Verify domain is registered and verified before sending
await DomainService.verifyEmailDomain(campaign.from, projectId);
// Get project to validate from address
const project = await prisma.project.findUnique({
where: {id: projectId},
});
if (!project) {
throw new HttpException(404, 'Project not found');
}
// Prepare the email content (no variable replacement for test emails)
const {sendRawEmail} = await import('./SESService.js');
await sendRawEmail({
from: {
name: campaign.fromName || project.name || 'Plunk',
email: campaign.from,
},
to: [testEmail],
content: {
subject: `[TEST] ${campaign.subject}`,
html: campaign.body,
},
reply: campaign.replyTo || undefined,
headers: {
'X-Plunk-Test': 'true',
},
tracking: false, // Disable tracking for test emails
});
}
}
apps/api/src/services/ContactService.ts
+1 -4
View File
@@ -2,6 +2,7 @@ import {type Contact, Prisma} from '@plunk/db';
import {prisma} from '../database/prisma.js';
import {HttpException} from '../exceptions/index.js';
import {EventService} from './EventService.js';
export interface PaginatedContacts {
contacts: Contact[];
@@ -155,7 +156,6 @@ export class ContactService {
// Track subscription event if status changed
if (isSubscriptionChanging) {
const {EventService} = await import('./EventService.js');
if (data.subscribed && !wasSubscribed) {
await EventService.trackEvent(projectId, 'contact.subscribed', contactId);
} else if (!data.subscribed && wasSubscribed) {
@@ -263,7 +263,6 @@ export class ContactService {
// Track subscription event if status changed
if (isSubscriptionChanging) {
const {EventService} = await import('./EventService.js');
if (subscribed && !wasSubscribed) {
await EventService.trackEvent(projectId, 'contact.subscribed', updated.id);
} else if (!subscribed && wasSubscribed) {
@@ -351,7 +350,6 @@ export class ContactService {
});
// Track subscription event
const {EventService} = await import('./EventService.js');
await EventService.trackEvent(contact.projectId, 'contact.subscribed', contactId);
return contact;
@@ -367,7 +365,6 @@ export class ContactService {
});
// Track unsubscription event
const {EventService} = await import('./EventService.js');
await EventService.trackEvent(contact.projectId, 'contact.unsubscribed', contactId);
return contact;
apps/api/src/services/EmailService.ts
+4 -2
View File
@@ -333,6 +333,10 @@ export class EmailService {
data: {
email: email.contact.email,
...contactData,
data: contactData,
unsubscribeUrl: `${DASHBOARD_URI}/unsubscribe/${email.contact.id}`,
subscribeUrl: `${DASHBOARD_URI}/subscribe/${email.contact.id}`,
manageUrl: `${DASHBOARD_URI}/manage/${email.contact.id}`,
},
});
@@ -466,7 +470,6 @@ export class EmailService {
data: {subscribed: false},
});
// Track unsubscription event
const {EventService} = await import('./EventService.js');
await EventService.trackEvent(email.projectId, 'contact.unsubscribed', email.contactId, email.id, {
reason: 'bounce',
});
@@ -483,7 +486,6 @@ export class EmailService {
data: {subscribed: false},
});
// Track unsubscription event
const {EventService} = await import('./EventService.js');
await EventService.trackEvent(email.projectId, 'contact.unsubscribed', email.contactId, email.id, {
reason: 'complaint',
});
apps/api/src/services/QueueService.ts
+1 -4
View File
@@ -2,6 +2,7 @@ import {type Job, Queue} from 'bullmq';
import type {RedisOptions} from 'ioredis';
import {REDIS_URL} from '../app/constants.js';
import {prisma} from '../database/prisma.js';
/**
* Queue Job Data Types
@@ -441,7 +442,6 @@ export class QueueService {
for (const job of scheduledCampaigns) {
// We need to check if the campaign belongs to this project
// by looking up the campaign in the database
const {prisma} = await import('../database/prisma.js');
const campaign = await prisma.campaign.findUnique({
where: {id: job.data.campaignId},
select: {projectId: true},
@@ -456,7 +456,6 @@ export class QueueService {
// Cancel all pending emails for this project
const pendingEmails = await emailQueue.getJobs(['waiting', 'delayed']);
for (const job of pendingEmails) {
const {prisma} = await import('../database/prisma.js');
const email = await prisma.email.findUnique({
where: {id: job.data.emailId},
select: {projectId: true},
@@ -471,7 +470,6 @@ export class QueueService {
// Cancel all pending campaign batches for this project
const campaignBatches = await campaignQueue.getJobs(['waiting', 'delayed']);
for (const job of campaignBatches) {
const {prisma} = await import('../database/prisma.js');
const campaign = await prisma.campaign.findUnique({
where: {id: job.data.campaignId},
select: {projectId: true},
@@ -486,7 +484,6 @@ export class QueueService {
// Cancel all pending workflow steps for this project
const workflowSteps = await workflowQueue.getJobs(['waiting', 'delayed']);
for (const job of workflowSteps) {
const {prisma} = await import('../database/prisma.js');
const execution = await prisma.workflowExecution.findUnique({
where: {id: job.data.executionId},
select: {workflow: {select: {projectId: true}}},
apps/api/src/services/SecurityService.ts
+1 -1
View File
@@ -2,6 +2,7 @@ import signale from 'signale';
import {prisma} from '../database/prisma.js';
import {redis} from '../database/redis.js';
import {QueueService} from './QueueService.js';
/**
* Security thresholds for bounce and complaint rates
@@ -324,7 +325,6 @@ export class SecurityService {
// Cancel all pending jobs for this project
try {
const {QueueService} = await import('./QueueService.js');
await QueueService.cancelAllProjectJobs(projectId);
signale.info(`[SECURITY] Cancelled all pending jobs for project ${projectId}`);
} catch (error) {
apps/api/src/services/WorkflowExecutionService.ts
+5
View File
@@ -12,6 +12,7 @@ import {WorkflowStepConfigSchemas} from '@plunk/shared';
import {prisma} from '../database/prisma.js';
import {HttpException} from '../exceptions/index.js';
import {DASHBOARD_URI} from '../app/constants.js';
import {EmailService} from './EmailService.js';
import {QueueService} from './QueueService.js';
@@ -446,6 +447,10 @@ export class WorkflowExecutionService {
email: contact.email,
...contactData,
...executionContext,
data: contactData,
unsubscribeUrl: `${DASHBOARD_URI}/unsubscribe/${contact.id}`,
subscribeUrl: `${DASHBOARD_URI}/subscribe/${contact.id}`,
manageUrl: `${DASHBOARD_URI}/manage/${contact.id}`,
};
const renderedSubject = this.renderTemplate(step.template.subject, variables);
apps/api/src/services/WorkflowService.ts
+1 -1
View File
@@ -6,6 +6,7 @@ import {HttpException} from '../exceptions/index.js';
import {EventService} from './EventService.js';
import {ContactService} from './ContactService.js';
import {WorkflowExecutionService} from './WorkflowExecutionService.js';
export interface PaginatedWorkflows {
workflows: Workflow[];
@@ -749,7 +750,6 @@ export class WorkflowService {
// Start executing the workflow asynchronously
// Don't await - let it run in background
const {WorkflowExecutionService} = await import('./WorkflowExecutionService.js');
WorkflowExecutionService.processStepExecution(execution.id, triggerStep.id).catch(error => {
console.error('Error executing workflow:', error);
});