Add events to workflows

2025-12-01 18:37:53 +01:00
parent 5158756650
commit b1f0043940
11 changed files with 927 additions and 51 deletions
@@ -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);
@@ -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<string[]> {
+    // 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<string>();
+
+    for (const event of events) {
+      if (event.data && typeof event.data === 'object' && !Array.isArray(event.data)) {
+        const data = event.data as Record<string, unknown>;
+        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
@@ -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<string, unknown>)
        : {};
+    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
@@ -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: '[email protected]',
+        openedAt: new Date().toISOString(),
+      });
+
+      await EventService.trackEvent(projectId, 'email.opened', contact.id, undefined, {
+        subject: 'Newsletter',
+        from: '[email protected]',
+        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: '[email protected]',
+        },
+        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: '[email protected]',
+        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});
@@ -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: '[email protected]',
+            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: '[email protected]',
+          } 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: '[email protected]',
+      });
+      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: '[email protected]',
+            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: '[email protected]',
+        messageId: 'msg-123',
+        isFirstOpen: true,
+        openedAt: '2024-01-15T10:00:00Z',
+      });
+
+      expect(body.contact.email).toBe('[email protected]');
+    });
+  });
 });