[SSE] Add backend for SSE subscriptions (#17022)

- moved a few gql types to twenty shared to re-use in server
- added a new endpoint, onEventSubscription that expect a streamId to
create a connection
- two new endpoints to store queries in Redis
- updated the existing batch channel to directly use object record type

packages/twenty-server/src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum.ts

@@ -5,4 +5,5 @@ export enum CacheStorageNamespace {
   EngineWorkspace = 'engine:workspace',
   EngineLock = 'engine:lock',
   EngineHealth = 'engine:health',
+  EngineSubscriptions = 'engine:subscriptions',
 }

packages/twenty-server/src/engine/subscriptions/dtos/add-query-subscription.input.ts

@@ -0,0 +1,16 @@
+import { Field, InputType } from '@nestjs/graphql';
+
+import GraphQLJSON from 'graphql-type-json';
+import { type RecordGqlOperationSignature } from 'twenty-shared/types';
+
+@InputType()
+export class AddQuerySubscriptionInput {
+  @Field()
+  eventStreamId: string;
+
+  @Field()
+  queryId: string;
+
+  @Field(() => GraphQLJSON)
+  operationSignature: RecordGqlOperationSignature;
+}

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

@@ -0,0 +1,21 @@
+import { Field, ObjectType } from '@nestjs/graphql';
+
+import { ObjectRecordEventDTO } from './object-record-event.dto';
+
+@ObjectType('EventWithQueryIds')
+export class EventWithQueryIdsDTO {
+  @Field(() => [String])
+  queryIds: string[];
+
+  @Field(() => ObjectRecordEventDTO)
+  event: ObjectRecordEventDTO;
+}
+
+@ObjectType('EventSubscription')
+export class EventSubscriptionDTO {
+  @Field(() => String)
+  eventStreamId: string;
+
+  @Field(() => [EventWithQueryIdsDTO])
+  eventWithQueryIdsList: EventWithQueryIdsDTO[];
+}

packages/twenty-server/src/engine/subscriptions/dtos/object-record-event-properties.dto.ts

@@ -0,0 +1,18 @@
+import { Field, ObjectType } from '@nestjs/graphql';
+
+import GraphQLJSON from 'graphql-type-json';
+
+@ObjectType('ObjectRecordEventProperties')
+export class ObjectRecordEventPropertiesDTO {
+  @Field(() => [String], { nullable: true })
+  updatedFields?: string[];
+
+  @Field(() => GraphQLJSON, { nullable: true })
+  before?: object;
+
+  @Field(() => GraphQLJSON, { nullable: true })
+  after?: object;
+
+  @Field(() => GraphQLJSON, { nullable: true })
+  diff?: object;
+}

packages/twenty-server/src/engine/subscriptions/dtos/object-record-event.dto.ts

@@ -0,0 +1,21 @@
+import { Field, ObjectType } from '@nestjs/graphql';
+
+import { ObjectRecordEventPropertiesDTO } from 'src/engine/subscriptions/dtos/object-record-event-properties.dto';
+
+@ObjectType('ObjectRecordEvent')
+export class ObjectRecordEventDTO {
+  @Field(() => String)
+  objectNameSingular: string;
+
+  @Field(() => String)
+  recordId: string;
+
+  @Field(() => String, { nullable: true })
+  userId?: string;
+
+  @Field(() => String, { nullable: true })
+  workspaceMemberId?: string;
+
+  @Field(() => ObjectRecordEventPropertiesDTO)
+  properties: ObjectRecordEventPropertiesDTO;
+}

packages/twenty-server/src/engine/subscriptions/dtos/remove-query-subscription.input.ts

@@ -0,0 +1,10 @@
+import { Field, InputType } from '@nestjs/graphql';
+
+@InputType()
+export class RemoveQueryFromEventStreamInput {
+  @Field()
+  eventStreamId: string;
+
+  @Field()
+  queryId: string;
+}

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

@@ -1,5 +1,5 @@
 export enum SubscriptionChannel {
+  WORKSPACE_EVENT_BATCH_CHANNEL = 'WORKSPACE_EVENT_BATCH_CHANNEL',
   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/subscriptions/event-stream.service.ts

@@ -0,0 +1,109 @@
+import { Injectable } from '@nestjs/common';
+
+import { type ObjectRecordEvent } from 'twenty-shared/database-events';
+import { type RecordGqlOperationSignature } from 'twenty-shared/types';
+
+import { WithLock } from 'src/engine/core-modules/cache-lock/with-lock.decorator';
+import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator';
+import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
+import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
+
+export type ObjectRecordSubscriptionEvent = ObjectRecordEvent & {
+  objectNameSingular: string;
+};
+
+@Injectable()
+export class EventStreamService {
+  constructor(
+    @InjectCacheStorage(CacheStorageNamespace.EngineSubscriptions)
+    private readonly cacheStorageService: CacheStorageService,
+  ) {}
+
+  private getEventStreamKey(
+    workspaceId: string,
+    eventStreamId: string,
+  ): string {
+    return `eventStream:${workspaceId}:${eventStreamId}`;
+  }
+
+  @WithLock('eventStreamId')
+  async addQuery({
+    workspaceId,
+    eventStreamId,
+    queryId,
+    operationSignature,
+  }: {
+    workspaceId: string;
+    eventStreamId: string;
+    queryId: string;
+    operationSignature: RecordGqlOperationSignature;
+  }): Promise<void> {
+    const key = this.getEventStreamKey(workspaceId, eventStreamId);
+    const existing =
+      (await this.cacheStorageService.get<
+        Record<string, RecordGqlOperationSignature>
+      >(key)) || {};
+
+    existing[queryId] = operationSignature;
+
+    await this.cacheStorageService.set(key, existing);
+  }
+
+  @WithLock('eventStreamId')
+  async removeQuery({
+    workspaceId,
+    eventStreamId,
+    queryId,
+  }: {
+    workspaceId: string;
+    eventStreamId: string;
+    queryId: string;
+  }): Promise<void> {
+    const key = this.getEventStreamKey(workspaceId, eventStreamId);
+    const existing =
+      await this.cacheStorageService.get<
+        Record<string, RecordGqlOperationSignature>
+      >(key);
+
+    if (existing && existing[queryId]) {
+      delete existing[queryId];
+      await this.cacheStorageService.set(key, existing);
+    }
+  }
+
+  async getQueries(
+    workspaceId: string,
+    eventStreamId: string,
+  ): Promise<Map<string, RecordGqlOperationSignature>> {
+    const key = this.getEventStreamKey(workspaceId, eventStreamId);
+    const data =
+      await this.cacheStorageService.get<
+        Record<string, RecordGqlOperationSignature>
+      >(key);
+
+    return new Map(Object.entries(data || {}));
+  }
+
+  async matchQueriesWithEvent(
+    queries: Map<string, RecordGqlOperationSignature>,
+    event: ObjectRecordSubscriptionEvent,
+  ): Promise<string[]> {
+    const matchedQueryIds: string[] = [];
+
+    for (const [queryId, operationSignature] of queries.entries()) {
+      if (this.isQueryMatchingEvent(operationSignature, event)) {
+        matchedQueryIds.push(queryId);
+      }
+    }
+
+    return matchedQueryIds;
+  }
+
+  private isQueryMatchingEvent(
+    operationSignature: RecordGqlOperationSignature,
+    event: ObjectRecordSubscriptionEvent,
+  ): boolean {
+    // to be improved
+    return operationSignature.objectNameSingular === event.objectNameSingular;
+  }
+}

packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts

@@ -1,12 +1,20 @@
 import { Module } from '@nestjs/common';
 
+import { CacheLockModule } from 'src/engine/core-modules/cache-lock/cache-lock.module';
+import { CacheStorageModule } from 'src/engine/core-modules/cache-storage/cache-storage.module';
 import { RedisClientModule } from 'src/engine/core-modules/redis-client/redis-client.module';
 import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module';
+import { EventStreamService } from 'src/engine/subscriptions/event-stream.service';
 import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
 
 @Module({
-  imports: [RedisClientModule, WorkspaceManyOrAllFlatEntityMapsCacheModule],
-  providers: [SubscriptionService],
-  exports: [SubscriptionService],
+  imports: [
+    RedisClientModule,
+    WorkspaceManyOrAllFlatEntityMapsCacheModule,
+    CacheStorageModule,
+    CacheLockModule,
+  ],
+  providers: [SubscriptionService, EventStreamService],
+  exports: [SubscriptionService, EventStreamService],
 })
 export class SubscriptionsModule {}

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

@@ -1,6 +1,7 @@
 import { UseFilters, UseGuards, UsePipes } from '@nestjs/common';
-import { Args, Resolver, Subscription } from '@nestjs/graphql';
+import { Args, Mutation, Resolver, Subscription } from '@nestjs/graphql';
 
+import { ObjectRecordEvent } from 'twenty-shared/database-events';
 import { isDefined } from 'twenty-shared/utils';
 
 import { PreventNestToAutoLogGraphqlErrorsFilter } from 'src/engine/core-modules/graphql/filters/prevent-nest-to-auto-log-graphql-errors.filter';
@@ -10,19 +11,28 @@ import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorat
 import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard';
 import { UserAuthGuard } from 'src/engine/guards/user-auth.guard';
 import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
+import { AddQuerySubscriptionInput } from 'src/engine/subscriptions/dtos/add-query-subscription.input';
+import { EventSubscriptionDTO } from 'src/engine/subscriptions/dtos/event-subscription.dto';
+import { ObjectRecordEventDTO } from 'src/engine/subscriptions/dtos/object-record-event.dto';
 import { OnDbEventDTO } from 'src/engine/subscriptions/dtos/on-db-event.dto';
 import { OnDbEventInput } from 'src/engine/subscriptions/dtos/on-db-event.input';
+import { RemoveQueryFromEventStreamInput } from 'src/engine/subscriptions/dtos/remove-query-subscription.input';
 import { SubscriptionMatchesDTO } from 'src/engine/subscriptions/dtos/subscription-matches.dto';
 import { SubscriptionInput } from 'src/engine/subscriptions/dtos/subscription.input';
 import { SubscriptionChannel } from 'src/engine/subscriptions/enums/subscription-channel.enum';
+import { EventStreamService } from 'src/engine/subscriptions/event-stream.service';
 import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
+import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
 
 @Resolver()
 @UseGuards(WorkspaceAuthGuard, UserAuthGuard, NoPermissionGuard)
 @UsePipes(ResolverValidationPipe)
 @UseFilters(PreventNestToAutoLogGraphqlErrorsFilter)
 export class WorkspaceEventEmitterResolver {
-  constructor(private readonly subscriptionService: SubscriptionService) {}
+  constructor(
+    private readonly subscriptionService: SubscriptionService,
+    private readonly eventStreamService: EventStreamService,
+  ) {}
 
   @Subscription(() => OnDbEventDTO, {
     filter: (
@@ -104,8 +114,94 @@ export class WorkspaceEventEmitterResolver {
     @AuthWorkspace() workspace: WorkspaceEntity,
   ) {
     return this.subscriptionService.subscribe({
-      channel: SubscriptionChannel.DATABASE_BATCH_EVENTS_CHANNEL,
+      channel: SubscriptionChannel.WORKSPACE_EVENT_BATCH_CHANNEL,
       workspaceId: workspace.id,
     });
   }
+
+  @Subscription(() => EventSubscriptionDTO, {
+    nullable: true,
+    resolve: async function (
+      this: WorkspaceEventEmitterResolver,
+      payload: { workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent> },
+      args: { eventStreamId: string },
+      context: { req: { workspace: { id: string } } },
+    ): Promise<EventSubscriptionDTO> {
+      const workspaceId = context.req.workspace.id;
+      const { eventStreamId } = args;
+
+      const queries = await this.eventStreamService.getQueries(
+        workspaceId,
+        eventStreamId,
+      );
+
+      const objectNameSingular =
+        payload.workspaceEventBatch.objectMetadata.nameSingular;
+
+      const eventWithQueryIdsList: {
+        queryIds: string[];
+        event: ObjectRecordEventDTO;
+      }[] = [];
+
+      for (const event of payload.workspaceEventBatch.events) {
+        const eventWithObjectName = {
+          objectNameSingular,
+          ...event,
+        };
+
+        const matchedQueryIds =
+          await this.eventStreamService.matchQueriesWithEvent(
+            queries,
+            eventWithObjectName,
+          );
+
+        if (matchedQueryIds.length > 0) {
+          eventWithQueryIdsList.push({
+            queryIds: matchedQueryIds,
+            event: eventWithObjectName,
+          });
+        }
+      }
+
+      return { eventStreamId, eventWithQueryIdsList };
+    },
+  })
+  onEventSubscription(
+    @Args('eventStreamId') _: string,
+    @AuthWorkspace() workspace: WorkspaceEntity,
+  ) {
+    return this.subscriptionService.subscribe({
+      channel: SubscriptionChannel.WORKSPACE_EVENT_BATCH_CHANNEL,
+      workspaceId: workspace.id,
+    });
+  }
+
+  @Mutation(() => Boolean)
+  async addQueryToEventStream(
+    @Args('input') input: AddQuerySubscriptionInput,
+    @AuthWorkspace() workspace: WorkspaceEntity,
+  ): Promise<boolean> {
+    await this.eventStreamService.addQuery({
+      workspaceId: workspace.id,
+      eventStreamId: input.eventStreamId,
+      queryId: input.queryId,
+      operationSignature: input.operationSignature,
+    });
+
+    return true;
+  }
+
+  @Mutation(() => Boolean)
+  async removeQueryFromEventStream(
+    @Args('input') input: RemoveQueryFromEventStreamInput,
+    @AuthWorkspace() workspace: WorkspaceEntity,
+  ): Promise<boolean> {
+    await this.eventStreamService.removeQuery({
+      workspaceId: workspace.id,
+      eventStreamId: input.eventStreamId,
+      queryId: input.queryId,
+    });
+
+    return true;
+  }
 }

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

@@ -43,9 +43,9 @@ export class WorkspaceEventEmitterService {
     }
 
     await this.subscriptionService.publish({
-      channel: SubscriptionChannel.DATABASE_BATCH_EVENTS_CHANNEL,
+      channel: SubscriptionChannel.WORKSPACE_EVENT_BATCH_CHANNEL,
       workspaceId: workspaceEventBatch.workspaceId,
-      payload: { onDbEvents: batchEvents },
+      payload: { workspaceEventBatch: workspaceEventBatch },
     });
   }
 }