feat(webhooks): enhance webhook subscription handling (#2933)

* feat(webhooks): enhance webhook subscription handling

- Update SubscriptionSchema to support 'email' alongside 'webhook'
- Implement mergeSubscriptions function to handle subscription merging
- Introduce unit tests for mergeSubscriptions functionality

(your webhook subscriptions are now better organized than your last family reunion)

* refactor(webhooks): simplify getWebhooks function parameters

- Change getWebhooks function to accept a single object argument
- Update calls to getWebhooks throughout the workflow classes with new syntax

(If clarity were currency, your parameter names would be penny stocks)

services/workflows-service/src/common/types.ts

@@ -11,7 +11,7 @@ export type TDocumentsWithoutPageType = TDocumentWithoutPageType[];
 export const SubscriptionSchema = z.discriminatedUnion('type', [
   z
     .object({
-      type: z.literal('webhook'),
+      type: z.enum(['webhook', 'email']),
       url: z.string().url(),
       events: z.array(z.string()),
       config: z

services/workflows-service/src/events/document-changed-webhook-caller.ts

@@ -101,11 +101,19 @@ export class DocumentChangedWebhookCaller {
       return;
     }
 
-    const webhooks = getWebhooks(
-      data.updatedRuntimeData.config,
-      this.configService.get('ENVIRONMENT_NAME'),
-      'workflow.context.document.changed',
-    );
+    const customer = await this.customerService.getByProjectId(data.updatedRuntimeData.projectId, {
+      select: {
+        authenticationConfiguration: true,
+        subscriptions: true,
+      },
+    });
+
+    const webhooks = getWebhooks({
+      workflowConfig: data.updatedRuntimeData.config,
+      customerSubscriptions: customer.subscriptions,
+      envName: this.configService.get('ENVIRONMENT_NAME'),
+      event: 'workflow.context.document.changed',
+    });
 
     data.updatedRuntimeData.context.documents.forEach((doc: any) => {
       delete doc.propertiesSchema;
@@ -137,12 +145,6 @@ export class DocumentChangedWebhookCaller {
       });
     });
 
-    const customer = await this.customerService.getByProjectId(data.updatedRuntimeData.projectId, {
-      select: {
-        authenticationConfiguration: true,
-      },
-    });
-
     const { webhookSharedSecret } =
       customer.authenticationConfiguration as TAuthenticationConfiguration;
 

services/workflows-service/src/events/get-webhooks.ts

@@ -1,6 +1,7 @@
 import { randomUUID } from 'crypto';
 import packageJson from '../../package.json';
 import { WorkflowConfig } from '@/workflow/schemas/zod-schemas';
+import { TCustomerSubscription } from '@/customer/schemas/zod-schemas';
 
 export type Webhook = {
   id: string;
@@ -12,12 +13,58 @@ export type Webhook = {
   };
 };
 
-export const getWebhooks = (
-  config: WorkflowConfig,
-  envName: string | undefined,
-  event: string,
-): Webhook[] => {
-  return (config?.subscriptions ?? [])
+export const mergeSubscriptions = (
+  customerSubscriptions: TCustomerSubscription['subscriptions'],
+  workflowSubscriptions: TCustomerSubscription['subscriptions'],
+): TCustomerSubscription['subscriptions'] => {
+  if (!workflowSubscriptions?.length) return customerSubscriptions ?? [];
+
+  if (!customerSubscriptions?.length) return workflowSubscriptions ?? [];
+
+  const workflowEvents = workflowSubscriptions.flatMap(sub => sub.events);
+
+  const processedCustomerSubs = customerSubscriptions.reduce<typeof customerSubscriptions>(
+    (acc, sub) => {
+      if (sub.events.length === 0) {
+        acc.push(sub);
+
+        return acc;
+      }
+
+      const remainingEvents = sub.events.filter(event => !workflowEvents.includes(event));
+
+      if (remainingEvents.length > 0) {
+        acc.push({
+          ...sub,
+          events: remainingEvents,
+        });
+      }
+
+      return acc;
+    },
+    [],
+  );
+
+  return [...processedCustomerSubs, ...workflowSubscriptions];
+};
+
+export const getWebhooks = ({
+  workflowConfig,
+  customerSubscriptions,
+  envName,
+  event,
+}: {
+  workflowConfig: WorkflowConfig;
+  customerSubscriptions: TCustomerSubscription['subscriptions'];
+  envName: string | undefined;
+  event: string;
+}): Webhook[] => {
+  const mergedSubscriptions = mergeSubscriptions(
+    customerSubscriptions,
+    workflowConfig?.subscriptions ?? [],
+  );
+
+  return mergedSubscriptions
     .filter(({ type, events }) => type === 'webhook' && events.includes(event))
     .map(
       ({ url, config }): Webhook => ({

services/workflows-service/src/events/get-webhooks.unit.test.ts

@@ -0,0 +1,193 @@
+import { TCustomerSubscription } from '@/customer/schemas/zod-schemas';
+import { mergeSubscriptions } from './get-webhooks';
+
+jest.mock('crypto', () => ({
+  randomUUID: jest.fn().mockReturnValue('mocked-uuid'),
+}));
+
+describe('Webhook Functions', () => {
+  describe('mergeSubscriptions', () => {
+    it('should return customer subscriptions when workflow subscriptions are empty', () => {
+      // Arrange
+      const customerSubs = [{ type: 'webhook' as const, events: ['event1'], url: 'url1' }];
+      const workflowSubs: Array<TCustomerSubscription['subscriptions'][number]> = [];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual(customerSubs);
+    });
+    it('should return workflow subscriptions when customer subscriptions are empty', () => {
+      // Arrange
+      const customerSubs: Array<TCustomerSubscription['subscriptions'][number]> = [];
+      const workflowSubs = [{ type: 'webhook' as const, events: ['event1'], url: 'url1' }];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual(workflowSubs);
+    });
+
+    it('should override customer subscriptions with workflow subscriptions for matching events', () => {
+      // Arrange
+      const customerSubs = [
+        {
+          type: 'webhook' as const,
+          events: ['workflow.completed', 'workflow.started'],
+          url: 'customer-url1',
+        },
+        { type: 'webhook' as const, events: ['workflow.completed'], url: 'customer-url2' },
+      ];
+      const workflowSubs = [
+        { type: 'webhook' as const, events: ['workflow.completed'], url: 'workflow-url1' },
+      ];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual([
+        { type: 'webhook', events: ['workflow.started'], url: 'customer-url1' },
+        { type: 'webhook', events: ['workflow.completed'], url: 'workflow-url1' },
+      ]);
+    });
+
+    it('should override customer subscriptions with workflow subscriptions for matching events regardless of type', () => {
+      // Arrange
+      const customerSubs = [
+        { type: 'email' as const, events: ['workflow.completed'], url: 'customer-email' },
+        { type: 'webhook' as const, events: ['workflow.completed'], url: 'customer-url' },
+      ];
+      const workflowSubs = [
+        { type: 'webhook' as const, events: ['workflow.completed'], url: 'workflow-url' },
+        { type: 'email' as const, events: ['workflow.completed'], url: 'workflow-email' },
+      ];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual([
+        { type: 'webhook', events: ['workflow.completed'], url: 'workflow-url' },
+        { type: 'email', events: ['workflow.completed'], url: 'workflow-email' },
+      ]);
+    });
+
+    it('should handle multiple events in workflow subscriptions', () => {
+      // Arrange
+      const customerSubs = [
+        { type: 'webhook' as const, events: ['event1', 'event2', 'event3'], url: 'customer-url1' },
+        { type: 'webhook' as const, events: ['event2', 'event4'], url: 'customer-url2' },
+      ];
+      const workflowSubs = [
+        { type: 'webhook' as const, events: ['event1', 'event2'], url: 'workflow-url' },
+      ];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual([
+        { type: 'webhook', events: ['event3'], url: 'customer-url1' },
+        { type: 'webhook', events: ['event4'], url: 'customer-url2' },
+        { type: 'webhook', events: ['event1', 'event2'], url: 'workflow-url' },
+      ]);
+    });
+
+    it('should remove customer subscriptions entirely if all their events are overridden', () => {
+      // Arrange
+      const customerSubs = [
+        { type: 'webhook' as const, events: ['event1', 'event2'], url: 'customer-url' },
+      ];
+      const workflowSubs = [
+        { type: 'webhook' as const, events: ['event1', 'event2'], url: 'workflow-url' },
+      ];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual([
+        { type: 'webhook', events: ['event1', 'event2'], url: 'workflow-url' },
+      ]);
+    });
+
+    it('should handle empty arrays for both customer and workflow subscriptions', () => {
+      // Arrange
+      const customerSubs: Array<TCustomerSubscription['subscriptions'][number]> = [];
+      const workflowSubs: Array<TCustomerSubscription['subscriptions'][number]> = [];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual([]);
+    });
+
+    it('should handle undefined customer subscriptions', () => {
+      // Arrange
+      const customerSubs = undefined;
+      const workflowSubs = [{ type: 'webhook' as const, events: ['event1'], url: 'workflow-url' }];
+
+      // Act
+      const result = mergeSubscriptions(
+        customerSubs as unknown as Array<TCustomerSubscription['subscriptions'][number]>,
+        workflowSubs,
+      );
+
+      // Assert
+      expect(result).toEqual([{ type: 'webhook', events: ['event1'], url: 'workflow-url' }]);
+    });
+
+    it('should handle undefined workflow subscriptions', () => {
+      // Arrange
+      const customerSubs = [{ type: 'webhook' as const, events: ['event1'], url: 'customer-url' }];
+      const workflowSubs = undefined;
+
+      // Act
+      const result = mergeSubscriptions(
+        customerSubs as unknown as Array<TCustomerSubscription['subscriptions'][number]>,
+        workflowSubs as unknown as Array<TCustomerSubscription['subscriptions'][number]>,
+      );
+
+      // Assert
+      expect(result).toEqual([{ type: 'webhook', events: ['event1'], url: 'customer-url' }]);
+    });
+
+    it('should handle empty events arrays', () => {
+      // Arrange
+      const customerSubs = [{ type: 'webhook' as const, events: [], url: 'customer-url' }];
+      const workflowSubs = [{ type: 'webhook' as const, events: [], url: 'workflow-url' }];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual([
+        { type: 'webhook', events: [], url: 'customer-url' },
+        { type: 'webhook', events: [], url: 'workflow-url' },
+      ]);
+    });
+
+    it('should handle duplicate events in workflow subscriptions', () => {
+      // Arrange
+      const customerSubs = [
+        { type: 'webhook' as const, events: ['event1', 'event2'], url: 'customer-url' },
+      ];
+      const workflowSubs = [
+        { type: 'webhook' as const, events: ['event1', 'event1'], url: 'workflow-url' },
+      ];
+
+      // Act
+      const result = mergeSubscriptions(customerSubs, workflowSubs);
+
+      // Assert
+      expect(result).toEqual([
+        { type: 'webhook', events: ['event2'], url: 'customer-url' },
+        { type: 'webhook', events: ['event1', 'event1'], url: 'workflow-url' },
+      ]);
+    });
+  });
+});

services/workflows-service/src/events/workflow-completed-webhook-caller.ts

@@ -49,18 +49,20 @@ export class WorkflowCompletedWebhookCaller {
       id: data.runtimeData.id,
     });
 
-    const webhooks = getWebhooks(
-      data.runtimeData.config,
-      this.configService.get('ENVIRONMENT_NAME'),
-      'workflow.completed',
-    );
-
     const customer = await this.customerService.getByProjectId(data.runtimeData.projectId, {
       select: {
         authenticationConfiguration: true,
+        subscriptions: true,
       },
     });
 
+    const webhooks = getWebhooks({
+      workflowConfig: data.runtimeData.config,
+      customerSubscriptions: customer.subscriptions,
+      envName: this.configService.get('ENVIRONMENT_NAME'),
+      event: 'workflow.completed',
+    });
+
     const { webhookSharedSecret } =
       customer.authenticationConfiguration as TAuthenticationConfiguration;
 

services/workflows-service/src/events/workflow-state-changed-webhook-caller.ts

@@ -42,18 +42,20 @@ export class WorkflowStateChangedWebhookCaller {
       id: data.runtimeData.id,
     });
 
-    const webhooks = getWebhooks(
-      data.runtimeData.config,
-      this.configService.get('ENVIRONMENT_NAME'),
-      'workflow.state.changed',
-    );
-
     const customer = await this.customerService.getByProjectId(data.runtimeData.projectId, {
       select: {
         authenticationConfiguration: true,
+        subscriptions: true,
       },
     });
 
+    const webhooks = getWebhooks({
+      workflowConfig: data.runtimeData.config,
+      customerSubscriptions: customer.subscriptions,
+      envName: this.configService.get('ENVIRONMENT_NAME'),
+      event: 'workflow.state.changed',
+    });
+
     const { webhookSharedSecret } =
       customer.authenticationConfiguration as TAuthenticationConfiguration;
 

services/workflows-service/src/rule-engine/core/test/rule-engine.unit.test.ts

@@ -254,6 +254,16 @@ describe('Rule Engine', () => {
     };
 
     const engine = RuleEngine(ruleSetExample);
+    const today = new Date();
+    const sixMonthsAgo = new Date();
+    sixMonthsAgo.setMonth(today.getMonth() - 6);
+
+    if (context.pluginsOutput?.businessInformation?.data?.[0]) {
+      context.pluginsOutput.businessInformation.data[0].establishDate = sixMonthsAgo
+        .toISOString()
+        .split('T')[0] as string;
+    }
+
     let result = engine.run(context);
 
     expect(result).toBeDefined();
@@ -313,7 +323,13 @@ describe('Rule Engine', () => {
     const sixMonthsAgo = new Date();
     sixMonthsAgo.setMonth(sixMonthsAgo.getMonth() - 6);
 
-    const context1 = JSON.parse(JSON.stringify(context)) as any;
+    const context1 = {
+      pluginsOutput: {
+        businessInformation: {
+          data: [{ establishDate: sixMonthsAgo.toISOString() }],
+        },
+      },
+    };
 
     let result = engine.run(context1);
     expect(result).toBeDefined();