diff --git a/apps/api/src/controllers/Webhooks.ts b/apps/api/src/controllers/Webhooks.ts index 87f61d4..628362f 100644 --- a/apps/api/src/controllers/Webhooks.ts +++ b/apps/api/src/controllers/Webhooks.ts @@ -8,6 +8,7 @@ import type Stripe from 'stripe'; import {STRIPE_ENABLED, STRIPE_WEBHOOK_SECRET} from '../app/constants.js'; import {stripe} from '../app/stripe.js'; import {prisma} from '../database/prisma.js'; +import {EventService} from '../services/EventService.js'; import {SecurityService} from '../services/SecurityService.js'; import {CatchAsync} from '../utils/asyncHandler.js'; @@ -23,7 +24,7 @@ export class Webhooks { */ @Post('sns') @CatchAsync - public async receiveSNSWebhook(req: Request, res: Response, next: NextFunction) { + public async receiveSNSWebhook(req: Request, res: Response) { try { // Handle SNS subscription confirmation FIRST (before parsing Message field) if (req.body.Type === 'SubscriptionConfirmation') { @@ -90,7 +91,18 @@ export class Webhooks { const now = new Date(); const updateData: Prisma.EmailUpdateInput = {}; const eventName = `email.${eventType.toLowerCase()}`; - let eventData: Record = {}; + + // Base event data with email metadata + const baseEventData = { + subject: email.subject, + from: email.from, + fromName: email.fromName, + messageId: email.messageId, + templateId: email.templateId, + campaignId: email.campaignId, + sourceType: email.sourceType, + }; + let eventData: Record = baseEventData; // Process event based on type switch (eventType) { @@ -98,6 +110,10 @@ export class Webhooks { signale.success(`[WEBHOOK] Delivery confirmed for ${email.contact.email} from ${email.project.name}`); updateData.status = EmailStatus.DELIVERED; updateData.deliveredAt = now; + eventData = { + ...baseEventData, + deliveredAt: now.toISOString(), + }; break; case 'Open': @@ -108,6 +124,12 @@ export class Webhooks { } updateData.opens = (email.opens || 0) + 1; updateData.status = EmailStatus.OPENED; + eventData = { + ...baseEventData, + openedAt: email.openedAt?.toISOString() || now.toISOString(), + opens: (email.opens || 0) + 1, + isFirstOpen: !email.openedAt, + }; break; case 'Click': { @@ -119,7 +141,13 @@ export class Webhooks { } updateData.clicks = (email.clicks || 0) + 1; updateData.status = EmailStatus.CLICKED; - eventData = {link: clickedLink}; + eventData = { + ...baseEventData, + link: clickedLink, + clickedAt: email.clickedAt?.toISOString() || now.toISOString(), + clicks: (email.clicks || 0) + 1, + isFirstClick: !email.clickedAt, + }; break; } @@ -132,7 +160,11 @@ export class Webhooks { where: {id: email.contactId}, data: {subscribed: false}, }); - eventData = body.bounce ? {bounceType: body.bounce.bounceType} : {}; + eventData = { + ...baseEventData, + bounceType: body.bounce?.bounceType, + bouncedAt: now.toISOString(), + }; break; case 'Complaint': @@ -144,6 +176,10 @@ export class Webhooks { where: {id: email.contactId}, data: {subscribed: false}, }); + eventData = { + ...baseEventData, + complainedAt: now.toISOString(), + }; break; default: @@ -157,16 +193,8 @@ export class Webhooks { data: updateData, }); - // Track event - await prisma.event.create({ - data: { - projectId: email.projectId, - contactId: email.contactId, - emailId: email.id, - name: eventName, - data: eventData as Prisma.InputJsonValue, - }, - }); + // Track event (this will trigger workflows) + await EventService.trackEvent(email.projectId, eventName, email.contactId, email.id, eventData); // Check security limits for bounce and complaint events if (eventType === 'Bounce' || eventType === 'Complaint') { @@ -188,7 +216,7 @@ export class Webhooks { */ @Post('incoming/stripe') @CatchAsync - public async receiveStripeWebhook(req: Request, res: Response, next: NextFunction) { + public async receiveStripeWebhook(req: Request, res: Response) { // Return 404 if billing is disabled if (!STRIPE_ENABLED || !stripe) { signale.warn('[WEBHOOK] Stripe webhook received but billing is disabled'); diff --git a/apps/api/src/controllers/Workflows.ts b/apps/api/src/controllers/Workflows.ts index 810f2ea..74a1de1 100644 --- a/apps/api/src/controllers/Workflows.ts +++ b/apps/api/src/controllers/Workflows.ts @@ -4,6 +4,8 @@ import type {NextFunction, Request, Response} from 'express'; import type {AuthResponse} from '../middleware/auth.js'; import {requireAuth} from '../middleware/auth.js'; +import {ContactService} from '../services/ContactService.js'; +import {EventService} from '../services/EventService.js'; import {WorkflowService} from '../services/WorkflowService.js'; import {CatchAsync} from '../utils/asyncHandler.js'; @@ -27,6 +29,45 @@ export class Workflows { return res.status(200).json(result); } + /** + * GET /workflows/fields + * Get all available fields for workflow conditions (contact fields + event fields) + * Query param: eventName - Optional event name to filter event fields + * NOTE: This must be defined BEFORE the :id route to avoid conflicts + */ + @Get('fields') + @Middleware([requireAuth]) + @CatchAsync + public async getAvailableFields(req: Request, res: Response, next: NextFunction) { + const auth = res.locals.auth as AuthResponse; + const eventName = req.query.eventName as string | undefined; + + try { + // Get contact fields (standard + custom data fields) + const contactFields = await ContactService.getAvailableFields(auth.projectId!); + + // Add standard contact fields + const standardContactFields = ['contact.email', 'contact.subscribed']; + + // Get event fields by analyzing actual event data + // This will only show fields that have been seen in actual events + const eventFields = await EventService.getAvailableEventFields(auth.projectId!, eventName); + + // Combine all fields + const allFields = [...standardContactFields, ...contactFields, ...eventFields].sort(); + + return res.status(200).json({ + fields: allFields, + count: allFields.length, + }); + } catch (error) { + console.error('[WORKFLOWS] Failed to get available fields:', error); + return res.status(500).json({ + error: error instanceof Error ? error.message : 'Failed to get available fields', + }); + } + } + /** * GET /workflows/:id * Get a specific workflow with all steps and transitions diff --git a/apps/api/src/controllers/__tests__/Workflows.test.ts b/apps/api/src/controllers/__tests__/Workflows.test.ts new file mode 100644 index 0000000..e236d31 --- /dev/null +++ b/apps/api/src/controllers/__tests__/Workflows.test.ts @@ -0,0 +1,269 @@ +import {describe, it, expect, beforeEach} from 'vitest'; +import {WorkflowTriggerType} from '@plunk/db'; +import request from 'supertest'; +import jwt from 'jsonwebtoken'; +import {app} from '../../app'; +import {JWT_SECRET} from '../../app/constants'; +import {factories, getPrismaClient} from '../../../../../test/helpers'; +import {EventService} from '../../services/EventService'; + +describe('Workflows Controller', () => { + let projectId: string; + let userId: string; + let authToken: string; + const prisma = getPrismaClient(); + + beforeEach(async () => { + const {project, user} = await factories.createUserWithProject(); + projectId = project.id; + userId = user.id; + authToken = jwt.sign({userId: user.id, projectId: project.id}, JWT_SECRET); + }); + + // ======================================== + // GET /workflows/fields + // ======================================== + describe('GET /workflows/fields', () => { + it('should return contact and event fields', async () => { + const contact = await factories.createContact({ + projectId, + data: { + firstName: 'John', + lastName: 'Doe', + plan: 'premium', + }, + }); + + // Create email events with data + await EventService.trackEvent(projectId, 'email.opened', contact.id, undefined, { + subject: 'Welcome', + from: 'hello@example.com', + openedAt: new Date().toISOString(), + isFirstOpen: true, + }); + + const response = await request(app) + .get('/workflows/fields') + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + expect(response.body.fields).toBeInstanceOf(Array); + expect(response.body.count).toBeGreaterThan(0); + + // Should include standard contact fields + expect(response.body.fields).toContain('contact.email'); + expect(response.body.fields).toContain('contact.subscribed'); + + // Should include custom contact data fields + expect(response.body.fields).toContain('data.firstName'); + expect(response.body.fields).toContain('data.lastName'); + expect(response.body.fields).toContain('data.plan'); + + // Should include event fields + expect(response.body.fields).toContain('event.subject'); + expect(response.body.fields).toContain('event.from'); + expect(response.body.fields).toContain('event.openedAt'); + expect(response.body.fields).toContain('event.isFirstOpen'); + }); + + it('should filter event fields by eventName query param', async () => { + const contact = await factories.createContact({projectId}); + + // Create email.opened events + await EventService.trackEvent(projectId, 'email.opened', contact.id, undefined, { + subject: 'Test', + openedAt: new Date().toISOString(), + isFirstOpen: true, + }); + + // Create email.clicked events + await EventService.trackEvent(projectId, 'email.clicked', contact.id, undefined, { + subject: 'Test', + link: 'https://example.com', + clickedAt: new Date().toISOString(), + }); + + // Request fields for email.opened only + const response = await request(app) + .get('/workflows/fields') + .query({eventName: 'email.opened'}) + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + // Should include email.opened fields + expect(response.body.fields).toContain('event.subject'); + expect(response.body.fields).toContain('event.openedAt'); + expect(response.body.fields).toContain('event.isFirstOpen'); + + // Should NOT include email.clicked fields + expect(response.body.fields).not.toContain('event.link'); + expect(response.body.fields).not.toContain('event.clickedAt'); + }); + + it('should return fields in sorted order', async () => { + const contact = await factories.createContact({ + projectId, + data: { + zebra: 'last', + apple: 'first', + }, + }); + + await EventService.trackEvent(projectId, 'test.event', contact.id, undefined, { + zoo: 'value', + aardvark: 'value', + }); + + const response = await request(app) + .get('/workflows/fields') + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + const fields = response.body.fields as string[]; + + // Verify fields are sorted + const sortedFields = [...fields].sort(); + expect(fields).toEqual(sortedFields); + }); + + it('should return empty event fields when no events exist', async () => { + const response = await request(app) + .get('/workflows/fields') + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + // Should still have contact fields + expect(response.body.fields).toContain('contact.email'); + expect(response.body.fields).toContain('contact.subscribed'); + + // But no event fields + const eventFields = response.body.fields.filter((f: string) => f.startsWith('event.')); + expect(eventFields).toHaveLength(0); + }); + + it('should require authentication', async () => { + await request(app).get('/workflows/fields').expect(401); + }); + + it('should only return fields for authenticated project', async () => { + // Create another project with events + const {project: otherProject} = await factories.createUserWithProject(); + const otherContact = await factories.createContact({projectId: otherProject.id}); + + await EventService.trackEvent(otherProject.id, 'other.event', otherContact.id, undefined, { + secretField: 'secret value', + }); + + // Request fields with our auth token (different project) + const response = await request(app) + .get('/workflows/fields') + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + // Should not include fields from other project + expect(response.body.fields).not.toContain('event.secretField'); + }); + + it('should handle URL encoded event names', async () => { + const contact = await factories.createContact({projectId}); + + await EventService.trackEvent(projectId, 'email.opened', contact.id, undefined, { + field1: 'value', + }); + + const response = await request(app) + .get('/workflows/fields') + .query({eventName: 'email.opened'}) + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + expect(response.body.fields).toContain('event.field1'); + }); + }); + + // ======================================== + // Integration Tests + // ======================================== + describe('Workflow field discovery integration', () => { + it('should discover correct fields for email.sent events', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id); + + // Create email.sent event with actual structure + await EventService.trackEvent(projectId, 'email.sent', contact.id, email.id, { + subject: 'Welcome Email', + from: 'hello@example.com', + fromName: 'Example Team', + messageId: 'msg-123', + templateId: 'tpl-456', + campaignId: null, + sourceType: 'TRANSACTIONAL', + sentAt: new Date().toISOString(), + }); + + const response = await request(app) + .get('/workflows/fields') + .query({eventName: 'email.sent'}) + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + expect(response.body.fields).toContain('event.subject'); + expect(response.body.fields).toContain('event.from'); + expect(response.body.fields).toContain('event.fromName'); + expect(response.body.fields).toContain('event.messageId'); + expect(response.body.fields).toContain('event.templateId'); + expect(response.body.fields).toContain('event.sourceType'); + expect(response.body.fields).toContain('event.sentAt'); + }); + + it('should discover correct fields for email.opened events', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id); + + await EventService.trackEvent(projectId, 'email.opened', contact.id, email.id, { + subject: 'Newsletter', + from: 'news@example.com', + fromName: 'Newsletter Team', + messageId: 'msg-456', + openedAt: new Date().toISOString(), + opens: 1, + isFirstOpen: true, + }); + + const response = await request(app) + .get('/workflows/fields') + .query({eventName: 'email.opened'}) + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + expect(response.body.fields).toContain('event.openedAt'); + expect(response.body.fields).toContain('event.opens'); + expect(response.body.fields).toContain('event.isFirstOpen'); + }); + + it('should discover correct fields for email.clicked events', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id); + + await EventService.trackEvent(projectId, 'email.clicked', contact.id, email.id, { + subject: 'Sale Announcement', + from: 'sales@example.com', + link: 'https://example.com/sale', + clickedAt: new Date().toISOString(), + clicks: 1, + isFirstClick: true, + }); + + const response = await request(app) + .get('/workflows/fields') + .query({eventName: 'email.clicked'}) + .set('Authorization', `Bearer ${authToken}`) + .expect(200); + + expect(response.body.fields).toContain('event.link'); + expect(response.body.fields).toContain('event.clickedAt'); + expect(response.body.fields).toContain('event.clicks'); + expect(response.body.fields).toContain('event.isFirstClick'); + }); + }); +}); diff --git a/apps/api/src/jobs/email-processor.ts b/apps/api/src/jobs/email-processor.ts index 6dfccea..19f84c5 100644 --- a/apps/api/src/jobs/email-processor.ts +++ b/apps/api/src/jobs/email-processor.ts @@ -9,6 +9,7 @@ import {type Job, Worker} from 'bullmq'; import {prisma} from '../database/prisma.js'; import {EmailService} from '../services/EmailService.js'; +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'; @@ -119,19 +120,16 @@ export function createEmailWorker() { await MeterService.recordEmailSent(email.project.customer, emailCount, `email_${emailId}`); } - // Track event - await prisma.event.create({ - data: { - projectId: email.projectId, - contactId: email.contactId, - emailId: email.id, - name: 'email.sent', - data: { - subject: formattedEmail.subject, - from: email.from, - messageId: result.messageId, - } as Prisma.InputJsonValue, - }, + // Track event (this will trigger workflows) + await EventService.trackEvent(email.projectId, 'email.sent', email.contactId, email.id, { + subject: formattedEmail.subject, + from: email.from, + fromName: email.fromName, + messageId: result.messageId, + templateId: email.templateId, + campaignId: email.campaignId, + sourceType: email.sourceType, + sentAt: new Date().toISOString(), }); } catch (error) { console.error(`[EMAIL-PROCESSOR] Failed to send email ${emailId}:`, error); diff --git a/apps/api/src/services/EmailService.ts b/apps/api/src/services/EmailService.ts index 3253503..572bd96 100644 --- a/apps/api/src/services/EmailService.ts +++ b/apps/api/src/services/EmailService.ts @@ -8,6 +8,7 @@ import {HttpException} from '../exceptions/index.js'; import {BillingLimitService} from './BillingLimitService.js'; import {DomainService} from './DomainService.js'; +import {EventService} from './EventService.js'; import {QueueService} from './QueueService.js'; import {sendRawEmail} from './SESService.js'; @@ -386,19 +387,16 @@ export class EmailService { }, }); - // Track event - await prisma.event.create({ - data: { - projectId: email.projectId, - contactId: email.contactId, - emailId: email.id, - name: 'email.sent', - data: { - subject: formattedEmail.subject, - from: email.from, - messageId: result.messageId, - }, - }, + // Track event (this will trigger workflows) + await EventService.trackEvent(email.projectId, 'email.sent', email.contactId, email.id, { + subject: formattedEmail.subject, + from: email.from, + fromName: email.fromName, + messageId: result.messageId, + templateId: email.templateId, + campaignId: email.campaignId, + sourceType: email.sourceType, + sentAt: new Date().toISOString(), }); } catch (error) { console.error(`[EMAIL] Failed to send email ${emailId}:`, error); diff --git a/apps/api/src/services/EventService.ts b/apps/api/src/services/EventService.ts index 3b77e6a..d1c2203 100644 --- a/apps/api/src/services/EventService.ts +++ b/apps/api/src/services/EventService.ts @@ -1,4 +1,5 @@ -import type {Event, Prisma} from '@plunk/db'; +import type {Event} from '@plunk/db'; +import {Prisma} from '@plunk/db'; import {prisma} from '../database/prisma.js'; import {redis} from '../database/redis.js'; @@ -139,6 +140,43 @@ export class EventService { return events.map(e => e.name); } + /** + * Get available event data fields for a specific event name + * Analyzes actual event data to discover which fields are present + * This is optimized for large datasets - only samples recent events + */ + public static async getAvailableEventFields(projectId: string, eventName?: string): Promise { + // Query recent events to discover data fields (limit to 100 for performance) + const events = await prisma.event.findMany({ + where: { + projectId, + ...(eventName ? {name: eventName} : {}), + data: { + not: Prisma.DbNull, // Only events with data (not null) + }, + }, + select: { + data: true, + }, + orderBy: {createdAt: 'desc'}, + take: 100, // Sample recent events for performance + }); + + // Extract all unique keys from event data + const fieldSet = new Set(); + + for (const event of events) { + if (event.data && typeof event.data === 'object' && !Array.isArray(event.data)) { + const data = event.data as Record; + for (const key of Object.keys(data)) { + fieldSet.add(`event.${key}`); + } + } + } + + return Array.from(fieldSet).sort(); + } + /** * Trigger workflows based on an event * Uses Redis caching for enabled workflows to improve performance diff --git a/apps/api/src/services/WorkflowExecutionService.ts b/apps/api/src/services/WorkflowExecutionService.ts index e46d189..a28c041 100644 --- a/apps/api/src/services/WorkflowExecutionService.ts +++ b/apps/api/src/services/WorkflowExecutionService.ts @@ -591,7 +591,8 @@ export class WorkflowExecutionService { // Structure allows access to: // - contact.email, contact.subscribed // - data.firstName, data.lastName, etc. - // - workflow.* (execution context) + // - workflow.* (execution context - alias for event data) + // - event.* (event data that triggered the workflow) const actualValue = this.resolveField(field, { contact: { email: contact.email, @@ -599,6 +600,7 @@ export class WorkflowExecutionService { }, data: contactData, workflow: context, + event: context, // Alias for easier access to event data }); // Evaluate the condition @@ -665,6 +667,7 @@ export class WorkflowExecutionService { contact.data && typeof contact.data === 'object' && !Array.isArray(contact.data) ? (contact.data as Record) : {}; + const context = execution.context || {}; const payload = body || { contact: { @@ -680,6 +683,7 @@ export class WorkflowExecutionService { id: execution.id, startedAt: execution.startedAt, }, + event: context, // Include event data that triggered the workflow }; // Make HTTP request diff --git a/apps/api/src/services/__tests__/EventService.test.ts b/apps/api/src/services/__tests__/EventService.test.ts index b90d835..af1b60b 100644 --- a/apps/api/src/services/__tests__/EventService.test.ts +++ b/apps/api/src/services/__tests__/EventService.test.ts @@ -599,6 +599,197 @@ describe('EventService', () => { }); }); + // ======================================== + // FIELD DISCOVERY + // ======================================== + describe('getAvailableEventFields', () => { + it('should discover fields from event data', async () => { + const contact = await factories.createContact({projectId}); + + // Create events with various fields + await EventService.trackEvent(projectId, 'email.opened', contact.id, undefined, { + subject: 'Welcome', + from: 'hello@example.com', + openedAt: new Date().toISOString(), + }); + + await EventService.trackEvent(projectId, 'email.opened', contact.id, undefined, { + subject: 'Newsletter', + from: 'news@example.com', + isFirstOpen: true, + opens: 1, + }); + + const fields = await EventService.getAvailableEventFields(projectId, 'email.opened'); + + expect(fields).toContain('event.subject'); + expect(fields).toContain('event.from'); + expect(fields).toContain('event.openedAt'); + expect(fields).toContain('event.isFirstOpen'); + expect(fields).toContain('event.opens'); + expect(fields).toHaveLength(5); + }); + + it('should filter fields by event name', async () => { + const contact = await factories.createContact({projectId}); + + // Create email.opened events + await EventService.trackEvent(projectId, 'email.opened', contact.id, undefined, { + subject: 'Test', + openedAt: new Date().toISOString(), + }); + + // Create email.clicked events + await EventService.trackEvent(projectId, 'email.clicked', contact.id, undefined, { + subject: 'Test', + link: 'https://example.com', + clickedAt: new Date().toISOString(), + }); + + // Get fields for email.opened only + const openedFields = await EventService.getAvailableEventFields(projectId, 'email.opened'); + + expect(openedFields).toContain('event.subject'); + expect(openedFields).toContain('event.openedAt'); + expect(openedFields).not.toContain('event.link'); + expect(openedFields).not.toContain('event.clickedAt'); + }); + + it('should return all event fields when no event name specified', async () => { + const contact = await factories.createContact({projectId}); + + await EventService.trackEvent(projectId, 'email.opened', contact.id, undefined, { + openedAt: new Date().toISOString(), + }); + + await EventService.trackEvent(projectId, 'email.clicked', contact.id, undefined, { + link: 'https://example.com', + }); + + const allFields = await EventService.getAvailableEventFields(projectId); + + expect(allFields).toContain('event.openedAt'); + expect(allFields).toContain('event.link'); + }); + + it('should return empty array when no events exist', async () => { + const fields = await EventService.getAvailableEventFields(projectId, 'nonexistent.event'); + + expect(fields).toHaveLength(0); + }); + + it('should ignore events with null data', async () => { + const contact = await factories.createContact({projectId}); + + // Event without data + await EventService.trackEvent(projectId, 'simple.event', contact.id); + + // Event with data + await EventService.trackEvent(projectId, 'simple.event', contact.id, undefined, { + field1: 'value1', + }); + + const fields = await EventService.getAvailableEventFields(projectId, 'simple.event'); + + expect(fields).toEqual(['event.field1']); + }); + + it('should handle nested field structures', async () => { + const contact = await factories.createContact({projectId}); + + await EventService.trackEvent(projectId, 'complex.event', contact.id, undefined, { + user: { + name: 'John', + email: 'john@example.com', + }, + metadata: { + source: 'web', + }, + }); + + const fields = await EventService.getAvailableEventFields(projectId, 'complex.event'); + + // Should return top-level keys + expect(fields).toContain('event.user'); + expect(fields).toContain('event.metadata'); + }); + + it('should deduplicate fields across multiple events', async () => { + const contact = await factories.createContact({projectId}); + + // Create 5 events with same fields + for (let i = 0; i < 5; i++) { + await EventService.trackEvent(projectId, 'test.event', contact.id, undefined, { + field1: `value${i}`, + field2: `value${i}`, + }); + } + + const fields = await EventService.getAvailableEventFields(projectId, 'test.event'); + + expect(fields).toEqual(['event.field1', 'event.field2']); + }); + + it('should respect 100 event sample limit for performance', async () => { + const contact = await factories.createContact({projectId}); + + // Create 150 events with different fields + for (let i = 0; i < 150; i++) { + await EventService.trackEvent(projectId, 'test.event', contact.id, undefined, { + [`field${i}`]: `value${i}`, + }); + } + + const fields = await EventService.getAvailableEventFields(projectId, 'test.event'); + + // Should sample only last 100 events (most recent) + // So we should see field50-field149 (100 fields) + expect(fields.length).toBeLessThanOrEqual(100); + }); + + it('should return fields in alphabetical order', async () => { + const contact = await factories.createContact({projectId}); + + await EventService.trackEvent(projectId, 'test.event', contact.id, undefined, { + zebra: 'last', + apple: 'first', + middle: 'middle', + }); + + const fields = await EventService.getAvailableEventFields(projectId, 'test.event'); + + expect(fields).toEqual(['event.apple', 'event.middle', 'event.zebra']); + }); + + it('should handle email event data structure', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail(projectId, contact.id); + + // Simulate actual email.sent event data + await EventService.trackEvent(projectId, 'email.sent', contact.id, email.id, { + subject: 'Welcome Email', + from: 'hello@example.com', + fromName: 'Example Team', + messageId: 'msg-123', + templateId: 'tpl-456', + campaignId: null, + sourceType: 'TRANSACTIONAL', + sentAt: new Date().toISOString(), + }); + + const fields = await EventService.getAvailableEventFields(projectId, 'email.sent'); + + expect(fields).toContain('event.subject'); + expect(fields).toContain('event.from'); + expect(fields).toContain('event.fromName'); + expect(fields).toContain('event.messageId'); + expect(fields).toContain('event.templateId'); + expect(fields).toContain('event.campaignId'); + expect(fields).toContain('event.sourceType'); + expect(fields).toContain('event.sentAt'); + }); + }); + describe('Event Data - Persistent vs Non-Persistent', () => { it('should store all event data (persistent + non-persistent) in event record', async () => { const contact = await factories.createContact({projectId}); diff --git a/apps/api/src/services/__tests__/WorkflowExecutionService.integration.test.ts b/apps/api/src/services/__tests__/WorkflowExecutionService.integration.test.ts index dcb3b16..7d10564 100644 --- a/apps/api/src/services/__tests__/WorkflowExecutionService.integration.test.ts +++ b/apps/api/src/services/__tests__/WorkflowExecutionService.integration.test.ts @@ -954,4 +954,299 @@ describe('WorkflowExecutionService - Integration Tests', () => { expect(savedContact?.data).not.toHaveProperty('trackingUrl'); }); }); + + // ======================================== + // EVENT DATA IN WORKFLOW CONDITIONS + // ======================================== + describe('Event Data in Workflow Conditions', () => { + it('should evaluate conditions using event.* fields', async () => { + const contact = await factories.createContact({projectId}); + const workflow = await factories.createWorkflow({projectId}); + + const triggerStep = await prisma.workflowStep.findFirst({ + where: {workflowId: workflow.id, type: 'TRIGGER'}, + }); + + // Create condition step that checks event data + const conditionStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.CONDITION, + name: 'Check if first open', + position: {x: 100, y: 0}, + config: { + field: 'event.isFirstOpen', + operator: 'equals', + value: true, // Use boolean, not string + } as Prisma.InputJsonValue, + }, + }); + + const yesStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'First Open', + position: {x: 200, y: 0}, + config: {reason: 'first_open'} as Prisma.InputJsonValue, + }, + }); + + const noStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'Not First Open', + position: {x: 200, y: 100}, + config: {reason: 'not_first_open'} as Prisma.InputJsonValue, + }, + }); + + // Connect steps + await prisma.workflowTransition.create({ + data: {fromStepId: triggerStep!.id, toStepId: conditionStep.id}, + }); + await prisma.workflowTransition.create({ + data: {fromStepId: conditionStep.id, toStepId: yesStep.id, condition: {branch: 'yes'} as Prisma.InputJsonValue}, + }); + await prisma.workflowTransition.create({ + data: {fromStepId: conditionStep.id, toStepId: noStep.id, condition: {branch: 'no'} as Prisma.InputJsonValue}, + }); + + // Create execution with event data + const execution = await prisma.workflowExecution.create({ + data: { + workflowId: workflow.id, + contactId: contact.id, + status: WorkflowExecutionStatus.RUNNING, + currentStepId: triggerStep!.id, + context: { + subject: 'Welcome Email', + from: 'hello@example.com', + isFirstOpen: true, + openedAt: new Date().toISOString(), + } as Prisma.InputJsonValue, + }, + }); + + // Process the workflow + await WorkflowExecutionService.processStepExecution(execution.id, triggerStep!.id); + + // Verify it followed YES branch (first open) + const completedExecution = await prisma.workflowExecution.findUnique({ + where: {id: execution.id}, + include: {stepExecutions: true}, + }); + + expect(completedExecution?.status).toBe(WorkflowExecutionStatus.COMPLETED); + expect(completedExecution?.exitReason).toBe('first_open'); + }); + + it('should evaluate conditions using event.subject field', async () => { + const contact = await factories.createContact({projectId}); + const workflow = await factories.createWorkflow({projectId}); + + const triggerStep = await prisma.workflowStep.findFirst({ + where: {workflowId: workflow.id, type: 'TRIGGER'}, + }); + + const conditionStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.CONDITION, + name: 'Check subject', + position: {x: 100, y: 0}, + config: { + field: 'event.subject', + operator: 'contains', + value: 'Welcome', + } as Prisma.InputJsonValue, + }, + }); + + const exitStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'Done', + position: {x: 200, y: 0}, + config: {reason: 'matched'} as Prisma.InputJsonValue, + }, + }); + + await prisma.workflowTransition.create({ + data: {fromStepId: triggerStep!.id, toStepId: conditionStep.id}, + }); + await prisma.workflowTransition.create({ + data: {fromStepId: conditionStep.id, toStepId: exitStep.id, condition: {branch: 'yes'} as Prisma.InputJsonValue}, + }); + + const execution = await prisma.workflowExecution.create({ + data: { + workflowId: workflow.id, + contactId: contact.id, + status: WorkflowExecutionStatus.RUNNING, + currentStepId: triggerStep!.id, + context: { + subject: 'Welcome to Plunk!', + from: 'team@plunk.com', + } as Prisma.InputJsonValue, + }, + }); + + await WorkflowExecutionService.processStepExecution(execution.id, triggerStep!.id); + + const completedExecution = await prisma.workflowExecution.findUnique({ + where: {id: execution.id}, + }); + + expect(completedExecution?.status).toBe(WorkflowExecutionStatus.COMPLETED); + expect(completedExecution?.exitReason).toBe('matched'); + }); + + it('should handle numeric event fields in conditions', async () => { + const contact = await factories.createContact({projectId}); + const workflow = await factories.createWorkflow({projectId}); + + const triggerStep = await prisma.workflowStep.findFirst({ + where: {workflowId: workflow.id, type: 'TRIGGER'}, + }); + + const conditionStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.CONDITION, + name: 'Check opens count', + position: {x: 100, y: 0}, + config: { + field: 'event.opens', + operator: 'greaterThan', + value: '3', + } as Prisma.InputJsonValue, + }, + }); + + const exitStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'Done', + position: {x: 200, y: 0}, + config: {reason: 'engaged'} as Prisma.InputJsonValue, + }, + }); + + await prisma.workflowTransition.create({ + data: {fromStepId: triggerStep!.id, toStepId: conditionStep.id}, + }); + await prisma.workflowTransition.create({ + data: {fromStepId: conditionStep.id, toStepId: exitStep.id, condition: {branch: 'yes'} as Prisma.InputJsonValue}, + }); + + const execution = await prisma.workflowExecution.create({ + data: { + workflowId: workflow.id, + contactId: contact.id, + status: WorkflowExecutionStatus.RUNNING, + currentStepId: triggerStep!.id, + context: { + subject: 'Newsletter', + opens: 5, + } as Prisma.InputJsonValue, + }, + }); + + await WorkflowExecutionService.processStepExecution(execution.id, triggerStep!.id); + + const completedExecution = await prisma.workflowExecution.findUnique({ + where: {id: execution.id}, + }); + + expect(completedExecution?.status).toBe(WorkflowExecutionStatus.COMPLETED); + expect(completedExecution?.exitReason).toBe('engaged'); + }); + }); + + // ======================================== + // EVENT DATA IN WEBHOOK CALLS + // ======================================== + describe('Event Data in Webhook Calls', () => { + it('should include event data in webhook payload', async () => { + // Mock fetch to capture webhook calls + const mockFetch = vi.fn(async () => ({ + ok: true, + status: 200, + text: async () => JSON.stringify({success: true}), + })); + global.fetch = mockFetch as any; + + const contact = await factories.createContact({ + projectId, + email: 'test@example.com', + }); + const workflow = await factories.createWorkflow({projectId}); + + const triggerStep = await prisma.workflowStep.findFirst({ + where: {workflowId: workflow.id, type: 'TRIGGER'}, + }); + + const webhookStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.WEBHOOK, + name: 'Send Webhook', + position: {x: 100, y: 0}, + config: { + url: 'https://webhook.example.com/test', + method: 'POST', + } as Prisma.InputJsonValue, + }, + }); + + await prisma.workflowTransition.create({ + data: {fromStepId: triggerStep!.id, toStepId: webhookStep.id}, + }); + + const execution = await prisma.workflowExecution.create({ + data: { + workflowId: workflow.id, + contactId: contact.id, + status: WorkflowExecutionStatus.RUNNING, + currentStepId: triggerStep!.id, + context: { + subject: 'Welcome Email', + from: 'hello@example.com', + messageId: 'msg-123', + isFirstOpen: true, + openedAt: '2024-01-15T10:00:00Z', + } as Prisma.InputJsonValue, + }, + }); + + await WorkflowExecutionService.processStepExecution(execution.id, triggerStep!.id); + + // Verify webhook was called with event data + expect(mockFetch).toHaveBeenCalledWith( + 'https://webhook.example.com/test', + expect.objectContaining({ + method: 'POST', + body: expect.stringContaining('"event"'), + }), + ); + + // Parse the body to verify event data structure + const callArgs = mockFetch.mock.calls[0]; + const body = JSON.parse(callArgs[1].body); + + expect(body.event).toMatchObject({ + subject: 'Welcome Email', + from: 'hello@example.com', + messageId: 'msg-123', + isFirstOpen: true, + openedAt: '2024-01-15T10:00:00Z', + }); + + expect(body.contact.email).toBe('test@example.com'); + }); + }); }); diff --git a/apps/web/next-env.d.ts b/apps/web/next-env.d.ts index 7996d35..1970904 100644 --- a/apps/web/next-env.d.ts +++ b/apps/web/next-env.d.ts @@ -1,6 +1,6 @@ /// /// -import "./.next/dev/types/routes.d.ts"; +import "./.next/types/routes.d.ts"; // NOTE: This file should not be edited // see https://nextjs.org/docs/pages/api-reference/config/typescript for more information. diff --git a/apps/web/src/pages/workflows/[id].tsx b/apps/web/src/pages/workflows/[id].tsx index 6764533..755945b 100644 --- a/apps/web/src/pages/workflows/[id].tsx +++ b/apps/web/src/pages/workflows/[id].tsx @@ -654,14 +654,21 @@ function AddStepDialog({open, onOpenChange, workflowId, onSuccess}: AddStepDialo const [isSubmitting, setIsSubmitting] = useState(false); const {data: templatesData} = useSWR<{templates: Template[]}>('/templates?pageSize=100'); + const {data: workflow} = useSWR(workflowId ? `/workflows/${workflowId}` : null); - // Fetch available contact fields when dialog opens and type is CONDITION + // Fetch available fields when dialog opens and type is CONDITION useEffect(() => { const fetchAvailableFields = async () => { - if (type === 'CONDITION' && open) { + if (type === 'CONDITION' && open && workflow) { setLoadingFields(true); try { - const response = await network.fetch<{fields: string[]}>('GET', '/contacts/fields'); + // Get event name from workflow trigger config + const triggerConfig = workflow.triggerConfig as {eventName?: string} | null; + const eventName = triggerConfig?.eventName; + + // Pass eventName as query param to filter event fields + const url = eventName ? `/workflows/fields?eventName=${encodeURIComponent(eventName)}` : '/workflows/fields'; + const response = await network.fetch<{fields: string[]}>('GET', url); setAvailableFields(response.fields); // Set default field if available @@ -679,7 +686,7 @@ function AddStepDialog({open, onOpenChange, workflowId, onSuccess}: AddStepDialo void fetchAvailableFields(); // eslint-disable-next-line react-hooks/exhaustive-deps - }, [type, open]); + }, [type, open, workflow]); const handleSubmit = async (e: React.FormEvent) => { e.preventDefault(); @@ -1162,14 +1169,21 @@ function EditStepDialog({step, workflowId, open, onOpenChange, onSuccess}: EditS const [exitReason, setExitReason] = useState(String(config?.reason || 'completed')); const {data: templatesData} = useSWR<{templates: Template[]}>('/templates?pageSize=100'); + const {data: workflow} = useSWR(workflowId ? `/workflows/${workflowId}` : null); // Fetch available contact fields when dialog opens and type is CONDITION useEffect(() => { const fetchAvailableFields = async () => { - if (step.type === 'CONDITION' && open) { + if (step.type === 'CONDITION' && open && workflow) { setLoadingFields(true); try { - const response = await network.fetch<{fields: string[]}>('GET', '/contacts/fields'); + // Get event name from workflow trigger config + const triggerConfig = workflow.triggerConfig as {eventName?: string} | null; + const eventName = triggerConfig?.eventName; + + // Pass eventName as query param to filter event fields + const url = eventName ? `/workflows/fields?eventName=${encodeURIComponent(eventName)}` : '/workflows/fields'; + const response = await network.fetch<{fields: string[]}>('GET', url); setAvailableFields(response.fields); } catch (error) { console.error('Failed to fetch available fields:', error); @@ -1181,7 +1195,7 @@ function EditStepDialog({step, workflowId, open, onOpenChange, onSuccess}: EditS }; void fetchAvailableFields(); - }, [step.type, open]); + }, [step.type, open, workflow, workflowId]); const handleSubmit = async (e: React.FormEvent) => { e.preventDefault();