Add SSE for metadata and plug front components (#17998)

Create the necessary tooling to listen to metadata events and plug it to the front components. Now we have a hot reload like experience when we edit a component in an app. ## Backend - Split `EventWithQueryIds` into `ObjectRecordEventWithQueryIds` and `MetadataEventWithQueryIds` - Publish metadata event batches to active SSE streams in `MetadataEventsToDbListener` ## Frontend - Create a metadata event dispatching pipeline: SSE metadata events are grouped by metadata name, transformed into `MetadataOperationBrowserEventDetail` objects, and dispatched as browser `CustomEvents` - Add `useListenToMetadataOperationBrowserEvent` hook for consuming metadata operation events filtered by metadata name and operation type - Rename `useListenToObjectRecordEventsForQuery` to `useListenToEventsForQuery`, now accepting both `RecordGqlOperationSignature` and `MetadataGqlOperationSignature` - Implement `useOnFrontComponentUpdated` which subscribes to front component metadata events and updates the Apollo cache when the component is modified - Add `builtComponentChecksum` to the front component query and appends it to the component URL for browser cache invalidation
2026-02-18 12:26:20 +01:00
parent e3753bf822
commit 2455c859b4
76 changed files with 1636 additions and 173 deletions
@@ -352,8 +352,12 @@ describe('WorkspaceEventEmitterService', () => {

      expect(publishCall.workspaceId).toBe(workspaceId);
      expect(publishCall.eventStreamChannelId).toBe(streamChannelId);
-      expect(publishCall.payload).toHaveLength(1);
-      expect(publishCall.payload[0].queryIds).toContain('query-1');
+      expect(publishCall.payload.objectRecordEventsWithQueryIds).toHaveLength(
+        1,
+      );
+      expect(
+        publishCall.payload.objectRecordEventsWithQueryIds[0].queryIds,
+      ).toContain('query-1');
    });

    it('should not publish events when object-level read permission is denied', async () => {
@@ -519,12 +523,14 @@ describe('WorkspaceEventEmitterService', () => {
        mockSubscriptionService.publishToEventStream as jest.Mock
      ).mock.calls[0][0];

-      expect(publishCall.payload[0].event.properties.after).not.toHaveProperty(
-        'secretField',
-      );
-      expect(publishCall.payload[0].event.properties.after).toHaveProperty(
-        'name',
-      );
+      expect(
+        publishCall.payload.objectRecordEventsWithQueryIds[0].objectRecordEvent
+          .properties.after,
+      ).not.toHaveProperty('secretField');
+      expect(
+        publishCall.payload.objectRecordEventsWithQueryIds[0].objectRecordEvent
+          .properties.after,
+      ).toHaveProperty('name');
    });

    it('should skip update events when all updated fields are restricted', async () => {
@@ -658,7 +664,8 @@ describe('WorkspaceEventEmitterService', () => {
        mockSubscriptionService.publishToEventStream as jest.Mock
      ).mock.calls[0][0];

-      const eventPayload = publishCall.payload[0].event;
+      const eventPayload =
+        publishCall.payload.objectRecordEventsWithQueryIds[0].objectRecordEvent;

      expect(eventPayload.properties.updatedFields).toEqual(['name']);
      expect(eventPayload.properties.diff).not.toHaveProperty('secretField');
@@ -829,7 +836,9 @@ describe('WorkspaceEventEmitterService', () => {
        mockSubscriptionService.publishToEventStream as jest.Mock
      ).mock.calls[0][0];

-      expect(publishCall.payload).toHaveLength(2);
+      expect(publishCall.payload.objectRecordEventsWithQueryIds).toHaveLength(
+        2,
+      );
    });

    it('should handle multiple matching queries', async () => {
@@ -868,8 +877,12 @@ describe('WorkspaceEventEmitterService', () => {
        mockSubscriptionService.publishToEventStream as jest.Mock
      ).mock.calls[0][0];

-      expect(publishCall.payload[0].queryIds).toContain('query-1');
-      expect(publishCall.payload[0].queryIds).toContain('query-2');
+      expect(
+        publishCall.payload.objectRecordEventsWithQueryIds[0].queryIds,
+      ).toContain('query-1');
+      expect(
+        publishCall.payload.objectRecordEventsWithQueryIds[0].queryIds,
+      ).toContain('query-2');
    });

    it('should use before record for delete events', async () => {
@@ -0,0 +1,512 @@
+import { isMetadataRecordMatchingFilter } from 'src/engine/workspace-event-emitter/utils/is-metadata-record-matching-filter.util';
+
+const record = {
+  id: '1',
+  name: 'Test Object',
+  label: 'testObject',
+  isActive: true,
+  count: 42,
+};
+
+describe('isMetadataRecordMatchingFilter', () => {
+  describe('empty filters', () => {
+    it('should match any record when filter is empty', () => {
+      expect(isMetadataRecordMatchingFilter({ record, filter: {} })).toBe(true);
+    });
+  });
+
+  describe('and filter', () => {
+    it('should match when all sub-filters match', () => {
+      const filter = {
+        and: [{ name: { eq: 'Test Object' } }, { isActive: { eq: true } }],
+      };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(true);
+    });
+
+    it('should not match when any sub-filter fails', () => {
+      const filter = {
+        and: [{ name: { eq: 'Test Object' } }, { isActive: { eq: false } }],
+      };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(false);
+    });
+
+    it('should match when and array is empty', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { and: [] },
+        }),
+      ).toBe(true);
+    });
+
+    it('should throw when and value is not an array', () => {
+      expect(() =>
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { and: 'invalid' } as any,
+        }),
+      ).toThrow('Unexpected value for "and" filter');
+    });
+  });
+
+  describe('or filter', () => {
+    it('should match when any sub-filter matches', () => {
+      const filter = {
+        or: [{ name: { eq: 'Wrong Name' } }, { name: { eq: 'Test Object' } }],
+      };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(true);
+    });
+
+    it('should not match when no sub-filter matches', () => {
+      const filter = {
+        or: [{ name: { eq: 'Wrong' } }, { name: { eq: 'Also Wrong' } }],
+      };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(false);
+    });
+
+    it('should match when or array is empty', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { or: [] },
+        }),
+      ).toBe(true);
+    });
+
+    it('should treat or with an object as an and', () => {
+      const filter = {
+        or: { name: { eq: 'Test Object' } },
+      };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(true);
+    });
+
+    it('should throw when or value is neither array nor object', () => {
+      expect(() =>
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { or: 'invalid' } as any,
+        }),
+      ).toThrow('Unexpected value for "or" filter');
+    });
+  });
+
+  describe('not filter', () => {
+    it('should negate a matching filter', () => {
+      const filter = { not: { name: { eq: 'Test Object' } } };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(false);
+    });
+
+    it('should negate a non-matching filter', () => {
+      const filter = { not: { name: { eq: 'Wrong' } } };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(true);
+    });
+
+    it('should match when not contains an empty object', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { not: {} },
+        }),
+      ).toBe(true);
+    });
+  });
+
+  describe('implicit and (multi-key filter)', () => {
+    it('should treat multiple keys as an implicit AND', () => {
+      const filter = {
+        name: { eq: 'Test Object' },
+        isActive: { eq: true },
+      };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(true);
+    });
+
+    it('should fail if any key in implicit AND does not match', () => {
+      const filter = {
+        name: { eq: 'Test Object' },
+        isActive: { eq: false },
+      };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(false);
+    });
+  });
+
+  describe('eq operator', () => {
+    it('should match equal string values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { eq: 'Test Object' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match different string values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { eq: 'Other' } },
+        }),
+      ).toBe(false);
+    });
+
+    it('should match equal boolean values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { isActive: { eq: true } },
+        }),
+      ).toBe(true);
+    });
+  });
+
+  describe('neq operator', () => {
+    it('should match when values are different', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { neq: 'Other' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match when values are equal', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { neq: 'Test Object' } },
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('in operator', () => {
+    it('should match when value is in the array', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { in: ['Test Object', 'Other'] } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match when value is not in the array', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { in: ['A', 'B'] } },
+        }),
+      ).toBe(false);
+    });
+
+    it('should return false when in value is not an array', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { in: 'not-array' } } as any,
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('is operator', () => {
+    it('should match NULL for undefined values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record: { ...record, optional: undefined },
+          filter: { optional: { is: 'NULL' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match NULL for defined values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { is: 'NULL' } },
+        }),
+      ).toBe(false);
+    });
+
+    it('should match NOT_NULL for defined values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { is: 'NOT_NULL' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match NOT_NULL for undefined values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record: { ...record, optional: undefined },
+          filter: { optional: { is: 'NOT_NULL' } },
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('like operator', () => {
+    it('should match with wildcard prefix', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { like: '%Object' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should match with wildcard suffix', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { like: 'Test%' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should match with wildcards on both sides', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { like: '%est Obj%' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match when pattern does not match', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { like: 'wrong%' } },
+        }),
+      ).toBe(false);
+    });
+
+    it('should escape regex special characters in pattern', () => {
+      const specialRecord = { ...record, name: 'foo.bar' };
+
+      expect(
+        isMetadataRecordMatchingFilter({
+          record: specialRecord,
+          filter: { name: { like: 'foo.bar' } },
+        }),
+      ).toBe(true);
+
+      // A dot in the pattern should NOT match any character
+      expect(
+        isMetadataRecordMatchingFilter({
+          record: { ...record, name: 'fooXbar' },
+          filter: { name: { like: 'foo.bar' } },
+        }),
+      ).toBe(false);
+    });
+
+    it('should return false for non-string values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { like: '42' } },
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('ilike operator', () => {
+    it('should match case-insensitively', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { ilike: '%test object%' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should escape regex special characters', () => {
+      const specialRecord = { ...record, name: 'foo(bar)' };
+
+      expect(
+        isMetadataRecordMatchingFilter({
+          record: specialRecord,
+          filter: { name: { ilike: 'FOO(BAR)' } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should return false for non-string values', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { ilike: '42' } },
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('gt operator', () => {
+    it('should match when value is greater', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { gt: 40 } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match when value is equal', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { gt: 42 } },
+        }),
+      ).toBe(false);
+    });
+
+    it('should not match when value is less', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { gt: 50 } },
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('gte operator', () => {
+    it('should match when value is greater or equal', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { gte: 42 } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match when value is less', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { gte: 43 } },
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('lt operator', () => {
+    it('should match when value is less', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { lt: 50 } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match when value is equal', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { lt: 42 } },
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('lte operator', () => {
+    it('should match when value is less or equal', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { lte: 42 } },
+        }),
+      ).toBe(true);
+    });
+
+    it('should not match when value is greater', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { count: { lte: 41 } },
+        }),
+      ).toBe(false);
+    });
+  });
+
+  describe('unknown operator', () => {
+    it('should throw for unsupported operators', () => {
+      expect(() =>
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: { unknownOp: 'value' } } as any,
+        }),
+      ).toThrow('Unsupported filter operator');
+    });
+  });
+
+  describe('nested logical operators', () => {
+    it('should handle deeply nested and/or/not', () => {
+      const filter = {
+        and: [
+          {
+            or: [
+              { name: { eq: 'Wrong' } },
+              { not: { isActive: { eq: false } } },
+            ],
+          },
+          { count: { gte: 40 } },
+        ],
+      };
+
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(true);
+    });
+
+    it('should handle not with nested and', () => {
+      const filter = {
+        not: {
+          and: [{ name: { eq: 'Test Object' } }, { count: { gt: 100 } }],
+        },
+      };
+
+      // name matches but count > 100 fails, so AND = false, NOT = true
+      expect(isMetadataRecordMatchingFilter({ record, filter })).toBe(true);
+    });
+  });
+
+  describe('edge cases', () => {
+    it('should skip non-object field filters', () => {
+      // When fieldFilter is a primitive, it is skipped (returns true)
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: 'Test Object' },
+        }),
+      ).toBe(true);
+    });
+
+    it('should skip undefined field filters', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record,
+          filter: { name: undefined },
+        }),
+      ).toBe(true);
+    });
+
+    it('should handle records with missing fields', () => {
+      expect(
+        isMetadataRecordMatchingFilter({
+          record: { id: '1' },
+          filter: { name: { eq: 'Test' } },
+        }),
+      ).toBe(false);
+    });
+  });
+});
@@ -0,0 +1,198 @@
+import { isObject } from '@sniptt/guards';
+import { isDefined } from 'twenty-shared/utils';
+
+type MetadataFilter = Record<string, unknown>;
+
+type MetadataScalarFilter = {
+  is?: 'NULL' | 'NOT_NULL';
+  eq?: unknown;
+  neq?: unknown;
+  in?: unknown[];
+  like?: string;
+  ilike?: string;
+  gt?: number;
+  gte?: number;
+  lt?: number;
+  lte?: number;
+};
+
+const isEmptyFilter = (filter: MetadataFilter): boolean =>
+  Object.keys(filter).length === 0;
+
+const isAndFilter = (filter: MetadataFilter): boolean =>
+  'and' in filter && isDefined(filter.and);
+
+const isOrFilter = (filter: MetadataFilter): boolean =>
+  'or' in filter && isDefined(filter.or);
+
+const isNotFilter = (filter: MetadataFilter): boolean =>
+  'not' in filter && isDefined(filter.not);
+
+const isImplicitAndFilter = (filter: MetadataFilter): boolean =>
+  Object.keys(filter).length > 1;
+
+const escapeRegExp = (str: string): string =>
+  str.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
+
+export const isMetadataRecordMatchingFilter = ({
+  record,
+  filter,
+}: {
+  record: Record<string, unknown>;
+  filter: MetadataFilter;
+}): boolean => {
+  if (isEmptyFilter(filter)) {
+    return true;
+  }
+
+  if (isImplicitAndFilter(filter)) {
+    return Object.entries(filter).every(([key, value]) =>
+      isMetadataRecordMatchingFilter({
+        record,
+        filter: { [key]: value },
+      }),
+    );
+  }
+
+  if (isAndFilter(filter)) {
+    const andValue = filter.and;
+
+    if (!Array.isArray(andValue)) {
+      throw new Error(
+        'Unexpected value for "and" filter: ' + JSON.stringify(andValue),
+      );
+    }
+
+    return (
+      andValue.length === 0 ||
+      andValue.every((subFilter: MetadataFilter) =>
+        isMetadataRecordMatchingFilter({ record, filter: subFilter }),
+      )
+    );
+  }
+
+  if (isOrFilter(filter)) {
+    const orValue = filter.or;
+
+    if (Array.isArray(orValue)) {
+      return (
+        orValue.length === 0 ||
+        orValue.some((subFilter: MetadataFilter) =>
+          isMetadataRecordMatchingFilter({ record, filter: subFilter }),
+        )
+      );
+    }
+
+    if (isObject(orValue)) {
+      return isMetadataRecordMatchingFilter({
+        record,
+        filter: orValue as MetadataFilter,
+      });
+    }
+
+    throw new Error(
+      'Unexpected value for "or" filter: ' + JSON.stringify(orValue),
+    );
+  }
+
+  if (isNotFilter(filter)) {
+    const notValue = filter.not;
+
+    if (isObject(notValue) && isEmptyFilter(notValue as MetadataFilter)) {
+      return true;
+    }
+
+    return !isMetadataRecordMatchingFilter({
+      record,
+      filter: notValue as MetadataFilter,
+    });
+  }
+
+  return Object.entries(filter).every(([fieldName, fieldFilter]) => {
+    if (!isDefined(fieldFilter) || !isObject(fieldFilter)) {
+      return true;
+    }
+
+    const recordValue = record[fieldName];
+
+    return isScalarValueMatchingFilter(
+      recordValue,
+      fieldFilter as MetadataScalarFilter,
+    );
+  });
+};
+
+const isScalarValueMatchingFilter = (
+  value: unknown,
+  fieldFilter: MetadataScalarFilter,
+): boolean => {
+  if ('is' in fieldFilter) {
+    if (fieldFilter.is === 'NULL') {
+      return !isDefined(value);
+    }
+
+    return isDefined(value);
+  }
+
+  if ('eq' in fieldFilter) {
+    return value === fieldFilter.eq;
+  }
+
+  if ('neq' in fieldFilter) {
+    return value !== fieldFilter.neq;
+  }
+
+  if ('in' in fieldFilter) {
+    if (!Array.isArray(fieldFilter.in)) {
+      return false;
+    }
+
+    return fieldFilter.in.includes(value);
+  }
+
+  if ('like' in fieldFilter) {
+    if (typeof value !== 'string') {
+      return false;
+    }
+
+    const pattern = String(fieldFilter.like)
+      .split('%')
+      .map(escapeRegExp)
+      .join('.*');
+
+    return new RegExp(`^${pattern}$`).test(value);
+  }
+
+  if ('ilike' in fieldFilter) {
+    if (typeof value !== 'string') {
+      return false;
+    }
+
+    const pattern = String(fieldFilter.ilike)
+      .split('%')
+      .map(escapeRegExp)
+      .join('.*');
+
+    return new RegExp(`^${pattern}$`, 'i').test(value);
+  }
+
+  if ('gt' in fieldFilter) {
+    return isDefined(value) && (value as number) > (fieldFilter.gt as number);
+  }
+
+  if ('gte' in fieldFilter) {
+    return isDefined(value) && (value as number) >= (fieldFilter.gte as number);
+  }
+
+  if ('lt' in fieldFilter) {
+    return isDefined(value) && (value as number) < (fieldFilter.lt as number);
+  }
+
+  if ('lte' in fieldFilter) {
+    return isDefined(value) && (value as number) <= (fieldFilter.lte as number);
+  }
+
+  throw new Error(
+    'Unsupported filter operator: ' + JSON.stringify(fieldFilter),
+  );
+};
@@ -18,10 +18,7 @@ import { UserAuthGuard } from 'src/engine/guards/user-auth.guard';
 import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
 import { EVENT_STREAM_TTL_MS } from 'src/engine/subscriptions/constants/event-stream-ttl.constant';
 import { AddQuerySubscriptionInput } from 'src/engine/subscriptions/dtos/add-query-subscription.input';
-import {
-  EventSubscriptionDTO,
-  EventWithQueryIdsDTO,
-} from 'src/engine/subscriptions/dtos/event-subscription.dto';
+import { EventSubscriptionDTO } from 'src/engine/subscriptions/dtos/event-subscription.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';
@@ -32,6 +29,7 @@ import {
 } from 'src/engine/subscriptions/event-stream.exception';
 import { EventStreamService } from 'src/engine/subscriptions/event-stream.service';
 import { SubscriptionService } from 'src/engine/subscriptions/subscription.service';
+import { type EventStreamPayload } from 'src/engine/subscriptions/types/event-stream-payload.type';
 import { wrapAsyncIteratorWithLifecycle } from 'src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-lifecycle';
 import { WorkspaceEventEmitterExceptionFilter } from 'src/engine/workspace-event-emitter/workspace-event-emitter-exception.filter';

@@ -86,12 +84,13 @@ export class WorkspaceEventEmitterResolver {
  @Subscription(() => EventSubscriptionDTO, {
    nullable: true,
    resolve: (
-      payload: EventWithQueryIdsDTO[],
+      payload: EventStreamPayload,
      variables: { eventStreamId: string },
    ) => {
      return {
        eventStreamId: variables.eventStreamId,
-        eventWithQueryIdsList: payload,
+        objectRecordEventsWithQueryIds: payload.objectRecordEventsWithQueryIds,
+        metadataEventsWithQueryIds: payload.metadataEventsWithQueryIds,
      };
    },
  })
@@ -126,7 +125,7 @@ export class WorkspaceEventEmitterResolver {
      },
    });

-    let iterator: AsyncIterableIterator<EventWithQueryIdsDTO[]>;
+    let iterator: AsyncIterableIterator<EventStreamPayload>;

    try {
      iterator = await this.subscriptionService.subscribeToEventStream({
@@ -142,7 +141,10 @@ export class WorkspaceEventEmitterResolver {
    }

    return wrapAsyncIteratorWithLifecycle(iterator, {
-      initialValue: [],
+      initialValue: {
+        objectRecordEventsWithQueryIds: [],
+        metadataEventsWithQueryIds: [],
+      },
      onHeartbeat: () =>
        this.eventStreamService.refreshEventStreamTTL({
          workspaceId: workspace.id,
@@ -7,9 +7,16 @@ import {
  ObjectRecord,
  type ObjectsPermissionsByRoleId,
  type RecordGqlOperationFilter,
+  type RecordGqlOperationSignature,
  type RestrictedFieldsPermissions,
 } from 'twenty-shared/types';
-import { combineFilters, isDefined } from 'twenty-shared/utils';
+import {
+  combineFilters,
+  isDefined,
+  isMetadataGqlOperationSignature,
+  isNonEmptyArray,
+  isRecordGqlOperationSignature,
+} from 'twenty-shared/utils';
 import { FindOptionsRelations, ObjectLiteral } from 'typeorm';

 import { ProcessNestedRelationsHelper } from 'src/engine/api/common/common-nested-relations-processor/process-nested-relations.helper';
@@ -18,6 +25,7 @@ import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner
 import { GraphqlQueryParser } from 'src/engine/api/graphql/graphql-query-runner/graphql-query-parsers/graphql-query.parser';
 import { type SerializableAuthContext } from 'src/engine/core-modules/auth/types/auth-context.type';
 import { type FlatWorkspaceMemberMaps } from 'src/engine/core-modules/user/types/flat-workspace-member-maps.type';
+import { type MetadataEventBatch } from 'src/engine/metadata-event-emitter/types/metadata-event-batch.type';
 import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
 import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type';
 import { findFlatEntityByIdInFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-id-in-flat-entity-maps.util';
@@ -30,7 +38,11 @@ import { transformEventToWebhookEvent } from 'src/engine/metadata-modules/webhoo
 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 { type EventStreamData } from 'src/engine/subscriptions/types/event-stream-data.type';
+import {
+  type EventStreamData,
+  type RecordOrMetadataGqlOperationSignature,
+} from 'src/engine/subscriptions/types/event-stream-data.type';
+import { type EventStreamPayload } from 'src/engine/subscriptions/types/event-stream-payload.type';
 import { ObjectRecordSubscriptionEvent } from 'src/engine/subscriptions/types/object-record-subscription-event.type';
 import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
 import { RolePermissionConfig } from 'src/engine/twenty-orm/types/role-permission-config';
@@ -38,7 +50,9 @@ import { buildRowLevelPermissionRecordFilter } from 'src/engine/twenty-orm/utils
 import { isRecordMatchingRLSRowLevelPermissionPredicate } from 'src/engine/twenty-orm/utils/is-record-matching-rls-row-level-permission-predicate.util';
 import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
 import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
+import { isMetadataRecordMatchingFilter } from 'src/engine/workspace-event-emitter/utils/is-metadata-record-matching-filter.util';
 import { parseEventNameOrThrow } from 'src/engine/workspace-event-emitter/utils/parse-event-name';
+import { type MetadataEvent } from 'src/engine/workspace-manager/workspace-migration/workspace-migration-runner/types/metadata-event';

@Injectable()
 export class WorkspaceEventEmitterService {
@@ -53,13 +67,29 @@ export class WorkspaceEventEmitterService {
  ) {}

  async publish(
-    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
+    eventBatch: WorkspaceEventBatch<ObjectRecordEvent> | MetadataEventBatch,
  ): Promise<void> {
-    const [nameSingular, operation] = workspaceEventBatch.name.split('.');
+    if (!this.isMetadataEventBatch(eventBatch)) {
+      await this.publishToLegacyChannel(eventBatch);
+    }

-    for (const eventData of workspaceEventBatch.events) {
+    await this.publishToEventStreams(eventBatch);
+  }
+
+  private isMetadataEventBatch(
+    eventBatch: WorkspaceEventBatch<ObjectRecordEvent> | MetadataEventBatch,
+  ): eventBatch is MetadataEventBatch {
+    return 'metadataName' in eventBatch;
+  }
+
+  private async publishToLegacyChannel(
+    eventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
+  ): Promise<void> {
+    const [nameSingular, operation] = eventBatch.name.split('.');
+
+    for (const eventData of eventBatch.events) {
      const { record, updatedFields } = transformEventToWebhookEvent({
-        eventName: workspaceEventBatch.name,
+        eventName: eventBatch.name,
        event: eventData,
      });

@@ -71,21 +101,19 @@ export class WorkspaceEventEmitterService {
        ...(updatedFields && { updatedFields }),
      };

-      // Publish individual events to legacy channel (onDbEvent)
      await this.subscriptionService.publish({
        channel: SubscriptionChannel.DATABASE_EVENT_CHANNEL,
-        workspaceId: workspaceEventBatch.workspaceId,
+        workspaceId: eventBatch.workspaceId,
        payload: { onDbEvent: event },
      });
    }
-
-    await this.publishToEventStreams(workspaceEventBatch);
  }

  private async publishToEventStreams(
-    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
+    eventBatch: WorkspaceEventBatch<ObjectRecordEvent> | MetadataEventBatch,
  ): Promise<void> {
-    const workspaceId = workspaceEventBatch.workspaceId;
+    const workspaceId = eventBatch.workspaceId;
+    const isMetadata = this.isMetadataEventBatch(eventBatch);

    const activeStreamIds =
      await this.eventStreamService.getActiveStreamIds(workspaceId);
@@ -99,15 +127,12 @@ export class WorkspaceEventEmitterService {
      activeStreamIds,
    );

-    const permissionsContext = await this.fetchPermissionsContext(workspaceId);
-
-    const { flatWorkspaceMemberMaps } =
-      await this.workspaceCacheService.getOrRecompute(workspaceId, [
-        'flatWorkspaceMemberMaps',
-      ]);
-
    const streamIdsToRemove: string[] = [];

+    const objectRecordStreamContext = !isMetadata
+      ? await this.fetchObjectRecordStreamContext(workspaceId)
+      : undefined;
+
    for (const [streamChannelId, streamData] of streamsData) {
      if (!isDefined(streamData)) {
        streamIdsToRemove.push(streamChannelId);
@@ -118,13 +143,25 @@ export class WorkspaceEventEmitterService {
        continue;
      }

-      await this.processStreamEvents(
-        streamChannelId,
-        streamData,
-        workspaceEventBatch,
-        permissionsContext,
-        flatWorkspaceMemberMaps,
-      );
+      if (isMetadata) {
+        await this.processMetadataStreamEvents(
+          streamChannelId,
+          streamData,
+          eventBatch as MetadataEventBatch,
+        );
+      } else {
+        if (!isDefined(objectRecordStreamContext)) {
+          continue;
+        }
+
+        await this.processObjectRecordStreamEvents(
+          streamChannelId,
+          streamData,
+          eventBatch as WorkspaceEventBatch<ObjectRecordEvent>,
+          objectRecordStreamContext.permissionsContext,
+          objectRecordStreamContext.flatWorkspaceMemberMaps,
+        );
+      }
    }

    await this.eventStreamService.removeFromActiveStreams(
@@ -133,7 +170,102 @@ export class WorkspaceEventEmitterService {
    );
  }

-  private async processStreamEvents(
+  private async fetchObjectRecordStreamContext(workspaceId: string) {
+    const permissionsContext = await this.fetchPermissionsContext(workspaceId);
+    const { flatWorkspaceMemberMaps } =
+      await this.workspaceCacheService.getOrRecompute(workspaceId, [
+        'flatWorkspaceMemberMaps',
+      ]);
+
+    return { permissionsContext, flatWorkspaceMemberMaps };
+  }
+
+  private async processMetadataStreamEvents(
+    streamChannelId: string,
+    streamData: EventStreamData,
+    metadataEventBatch: MetadataEventBatch,
+  ): Promise<void> {
+    const metadataEventsWithQueryIds: {
+      queryIds: string[];
+      metadataEvent: MetadataEvent;
+    }[] = [];
+
+    for (const metadataEvent of metadataEventBatch.events) {
+      const matchedQueryIds = this.getMatchingMetadataQueryIds(
+        streamData.queries,
+        metadataEvent,
+      );
+
+      if (!isNonEmptyArray(matchedQueryIds)) {
+        continue;
+      }
+
+      metadataEventsWithQueryIds.push({
+        queryIds: matchedQueryIds,
+        metadataEvent,
+      });
+    }
+
+    if (!isNonEmptyArray(metadataEventsWithQueryIds)) {
+      return;
+    }
+
+    const payload: EventStreamPayload = {
+      objectRecordEventsWithQueryIds: [],
+      metadataEventsWithQueryIds,
+    };
+
+    await this.subscriptionService.publishToEventStream({
+      workspaceId: metadataEventBatch.workspaceId,
+      eventStreamChannelId: streamChannelId,
+      payload,
+    });
+  }
+
+  private getMatchingMetadataQueryIds(
+    queries: Record<string, RecordOrMetadataGqlOperationSignature>,
+    metadataEvent: MetadataEvent,
+  ): string[] {
+    const properties = metadataEvent.properties as {
+      after?: Record<string, unknown>;
+      before?: Record<string, unknown>;
+    };
+
+    const record = properties?.after ?? properties?.before;
+
+    return Object.entries(queries)
+      .filter(([, operationSignature]) => {
+        if (!isMetadataGqlOperationSignature(operationSignature)) {
+          return false;
+        }
+
+        if (operationSignature.metadataName !== metadataEvent.metadataName) {
+          return false;
+        }
+
+        const queryFilter = (
+          operationSignature.variables as {
+            filter?: Record<string, unknown>;
+          }
+        )?.filter;
+
+        if (!isDefined(queryFilter) || Object.keys(queryFilter).length === 0) {
+          return true;
+        }
+
+        if (!isDefined(record)) {
+          return false;
+        }
+
+        return isMetadataRecordMatchingFilter({
+          record,
+          filter: queryFilter,
+        });
+      })
+      .map(([queryId]) => queryId);
+  }
+
+  private async processObjectRecordStreamEvents(
    streamChannelId: string,
    streamData: EventStreamData,
    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
@@ -169,7 +301,7 @@ export class WorkspaceEventEmitterService {

    const matchedEvents: {
      queryIds: string[];
-      event: ObjectRecordEvent & { objectNameSingular: string };
+      objectRecordEvent: ObjectRecordSubscriptionEvent;
    }[] = [];

    const objectNameSingular = workspaceEventBatch.objectMetadata.nameSingular;
@@ -210,7 +342,7 @@ export class WorkspaceEventEmitterService {
        continue;
      }

-      const matchedQueryIds = this.getMatchingQueryIds(
+      const matchedQueryIds = this.getMatchingObjectRecordQueryIds(
        streamData.queries,
        filteredEvent,
        subscriberRLSFilter,
@@ -224,24 +356,29 @@ export class WorkspaceEventEmitterService {

      matchedEvents.push({
        queryIds: matchedQueryIds,
-        event: filteredEvent,
+        objectRecordEvent: filteredEvent,
      });
    }

    if (matchedEvents.length > 0) {
      await this.enrichEventBatchWithNestedRelations({
        objectMetadata: workspaceEventBatch.objectMetadata,
-        events: matchedEvents.map((e) => e.event),
+        events: matchedEvents.map((e) => e.objectRecordEvent),
        streamData,
        permissionsContext,
        workspaceId: workspaceEventBatch.workspaceId,
        roleId,
      });

+      const payload: EventStreamPayload = {
+        objectRecordEventsWithQueryIds: matchedEvents,
+        metadataEventsWithQueryIds: [],
+      };
+
      await this.subscriptionService.publishToEventStream({
        workspaceId: workspaceEventBatch.workspaceId,
        eventStreamChannelId: streamChannelId,
-        payload: matchedEvents,
+        payload,
      });
    }
  }
@@ -435,14 +572,8 @@ export class WorkspaceEventEmitterService {
    } as ObjectRecordSubscriptionEvent;
  }

-  private getMatchingQueryIds(
-    queries: Record<
-      string,
-      {
-        objectNameSingular: string;
-        variables?: { filter?: RecordGqlOperationFilter };
-      }
-    >,
+  private getMatchingObjectRecordQueryIds(
+    queries: Record<string, RecordOrMetadataGqlOperationSignature>,
    event: ObjectRecordSubscriptionEvent,
    subscriberRLSFilter: RecordGqlOperationFilter | null,
    objectMetadata: FlatObjectMetadata,
@@ -451,8 +582,12 @@ export class WorkspaceEventEmitterService {
    const matchedQueryIds: string[] = [];

    for (const [queryId, operationSignature] of Object.entries(queries)) {
+      if (!isRecordGqlOperationSignature(operationSignature)) {
+        continue;
+      }
+
      if (
-        this.isQueryMatchingEvent(
+        this.isQueryMatchingObjectRecordEvent(
          operationSignature,
          event,
          subscriberRLSFilter,
@@ -467,11 +602,8 @@ export class WorkspaceEventEmitterService {
    return matchedQueryIds;
  }

-  private isQueryMatchingEvent(
-    operationSignature: {
-      objectNameSingular: string;
-      variables?: { filter?: RecordGqlOperationFilter };
-    },
+  private isQueryMatchingObjectRecordEvent(
+    operationSignature: RecordGqlOperationSignature,
    event: ObjectRecordSubscriptionEvent,
    subscriberRLSFilter: RecordGqlOperationFilter | null,
    objectMetadata: FlatObjectMetadata,
@@ -485,6 +617,7 @@ export class WorkspaceEventEmitterService {
      after?: object;
      before?: object;
    };
+
    const record = properties?.after ?? properties?.before;

    if (!isDefined(record)) {