Publish batch events for sse subscriptions (#16943)

- publish batch events to avoid multiple redis / graphql sse
publications
- add event to the endpoint output

packages/twenty-server/src/engine/subscriptions/dtos/subscription-matches.dto.ts

@@ -1,13 +1,18 @@
 import { Field, ObjectType } from '@nestjs/graphql';
 
+import { OnDbEventDTO } from './on-db-event.dto';
+
 @ObjectType('SubscriptionMatch')
 export class SubscriptionMatchDTO {
-  @Field(() => String)
-  id: string;
+  @Field(() => [String])
+  subscriptionIds: string[];
+
+  @Field(() => OnDbEventDTO)
+  event: OnDbEventDTO;
 }
 
 @ObjectType('SubscriptionMatches')
 export class SubscriptionMatchesDTO {
   @Field(() => [SubscriptionMatchDTO])
-  subscriptions: SubscriptionMatchDTO[];
+  matches: SubscriptionMatchDTO[];
 }

packages/twenty-server/src/engine/subscriptions/enums/subscription-channel.enum.ts

@@ -1,4 +1,5 @@
 export enum SubscriptionChannel {
   DATABASE_EVENT_CHANNEL = 'DATABASE_EVENT_CHANNEL',
+  DATABASE_BATCH_EVENTS_CHANNEL = 'DATABASE_BATCH_EVENTS_CHANNEL',
   SERVERLESS_FUNCTION_LOGS_CHANNEL = 'SERVERLESS_FUNCTION_LOGS_CHANNEL',
 }

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts

@@ -61,36 +61,41 @@ export class WorkspaceEventEmitterResolver {
     nullable: true,
     resolve: async function (
       this: WorkspaceEventEmitterResolver,
-      payload: { onDbEvent: OnDbEventDTO },
+      payload: { onDbEvents: OnDbEventDTO[] },
       args: { subscriptions: SubscriptionInput[] },
       context: { req: { workspace: { id: string } } },
     ): Promise<SubscriptionMatchesDTO> {
       const workspaceId = context.req.workspace.id;
 
-      const matchedSubscriptionIds = await Promise.all(
-        args.subscriptions.map(async (subscription) => {
-          const matches =
-            await this.subscriptionService.isSubscriptionMatchingEvent(
-              subscription,
-              payload.onDbEvent,
-              workspaceId,
-            );
+      const matches: { subscriptionIds: string[]; event: OnDbEventDTO }[] = [];
 
-          return matches ? subscription.id : null;
-        }),
-      );
+      for (const event of payload.onDbEvents) {
+        const matchedSubscriptionIds = await Promise.all(
+          args.subscriptions.map(async (subscription) => {
+            const isMatch =
+              await this.subscriptionService.isSubscriptionMatchingEvent(
+                subscription,
+                event,
+                workspaceId,
+              );
 
-      const filteredIds = matchedSubscriptionIds.filter(
-        (id): id is string => id !== null,
-      );
+            return isMatch ? subscription.id : null;
+          }),
+        );
 
-      if (filteredIds.length === 0) {
-        return { subscriptions: [] };
+        const filteredIds = matchedSubscriptionIds.filter(
+          (id): id is string => id !== null,
+        );
+
+        if (filteredIds.length > 0) {
+          matches.push({
+            subscriptionIds: filteredIds,
+            event,
+          });
+        }
       }
 
-      return {
-        subscriptions: filteredIds.map((id) => ({ id })),
-      };
+      return { matches };
     },
   })
   onSubscriptionMatch(
@@ -99,7 +104,7 @@ export class WorkspaceEventEmitterResolver {
     @AuthWorkspace() workspace: WorkspaceEntity,
   ) {
     return this.subscriptionService.subscribe({
-      channel: SubscriptionChannel.DATABASE_EVENT_CHANNEL,
+      channel: SubscriptionChannel.DATABASE_BATCH_EVENTS_CHANNEL,
       workspaceId: workspace.id,
     });
   }

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts

@@ -2,10 +2,10 @@ import { Injectable } from '@nestjs/common';
 
 import { type ObjectRecordEvent } from 'twenty-shared/database-events';
 
-import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
 import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event';
-import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
 import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum';
+import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
+import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
 
 @Injectable()
 export class WorkspaceEventEmitterService {
@@ -16,25 +16,36 @@ export class WorkspaceEventEmitterService {
   ): Promise<void> {
     const [nameSingular, operation] = workspaceEventBatch.name.split('.');
 
+    const batchEvents = [];
+
     for (const eventData of workspaceEventBatch.events) {
       const { record, updatedFields } = transformEventToWebhookEvent({
         eventName: workspaceEventBatch.name,
         event: eventData,
       });
 
+      const event = {
+        action: operation,
+        objectNameSingular: nameSingular,
+        eventDate: new Date(),
+        record,
+        ...(updatedFields && { updatedFields }),
+      };
+
+      batchEvents.push(event);
+
+      // Publish individual events to legacy channel (onDbEvent)
       await this.subscriptionService.publish({
         channel: SubscriptionChannel.DATABASE_EVENT_CHANNEL,
         workspaceId: workspaceEventBatch.workspaceId,
-        payload: {
-          onDbEvent: {
-            action: operation,
-            objectNameSingular: nameSingular,
-            eventDate: new Date(),
-            record,
-            ...(updatedFields && { updatedFields }),
-          },
-        },
+        payload: { onDbEvent: event },
       });
     }
+
+    await this.subscriptionService.publish({
+      channel: SubscriptionChannel.DATABASE_BATCH_EVENTS_CHANNEL,
+      workspaceId: workspaceEventBatch.workspaceId,
+      payload: { onDbEvents: batchEvents },
+    });
   }
 }