fix(twenty-server): notify SSE subscribers when an update leaves their filtered view (#23858)

## Context Record tables subscribe to SSE with their query signature (object + filter) and the server pushes only matching DB events. `isQueryMatchingObjectRecordEvent` evaluated `after ?? before` — for UPDATED events that is always `after`, so an update moving a record **out** of a filtered view never matched: no event, and the open table keeps the stale row until a manual reload. Symptom on twenty-internal: a Sales Action Item marked Done by an AI-chat tool (or any API/workflow write) stays visible in the `status = OPEN` view. Records *entering* a view appear live; records *leaving* never disappear. UI edits mask the bug via the local Apollo cache. ## Fix For UPDATED events, view membership now matches on either snapshot (a before-only match = the record just left the view). Row-level authorization is unchanged: still evaluated against the delivered snapshot, so a before-state match cannot authorize a payload the subscriber lost RLS access to. A pre-existing, unrelated payload leak spotted during review (before values delivered when a record enters RLS scope) is fixed separately in the stacked #23870. ## Test plan - Unit: leave-view update publishes (fails on the old matcher — verified), neither-state-matches does not, RLS-failing delivered state does not even when before matched. 36/36 on the spec, full server suite green. - End-to-end on a local stack: companies table filtered `Name contains 'Open'`, `updateCompany` renamed a row out of the filter via the API → row disappeared from the open table within seconds, no reload.
2026-08-06 17:02:26 +02:00
parent f5a42cdbae
commit a47f566eb1
2 changed files with 199 additions and 30 deletions
@@ -475,6 +475,157 @@ describe('ObjectRecordEventPublisher', () => {
      ).not.toHaveBeenCalled();
    });

+    it('should publish update events when only the BEFORE state matches the filter (record leaving the view)', async () => {
+      (
+        isRecordMatchingRLSRowLevelPermissionPredicate as jest.Mock
+      ).mockImplementation(
+        ({ record }: { record: { name?: string } }) =>
+          record.name === 'Open Company',
+      );
+
+      const streamDataWithFilter: EventStreamData = {
+        ...mockStreamData,
+        queries: {
+          'query-1': {
+            objectNameSingular: 'company',
+            variables: {
+              filter: { name: { eq: 'Open Company' } },
+            },
+          },
+        },
+      };
+
+      mockEventStreamService.getStreamsData.mockResolvedValue(
+        new Map([[streamChannelId, streamDataWithFilter]]) as Map<
+          string,
+          EventStreamData | undefined
+        >,
+      );
+
+      const eventBatch: WorkspaceEventBatch<MockObjectRecordEvent> = {
+        name: 'company.updated',
+        workspaceId,
+        objectMetadata: companyObjectMetadata,
+        events: [
+          createMockEvent({
+            properties: {
+              before: { id: 'record-1', name: 'Open Company' },
+              after: { id: 'record-1', name: 'Done Company' },
+            } as MockObjectRecordEvent['properties'],
+          }),
+        ],
+      };
+
+      await service.publish(eventBatch as WorkspaceEventBatch<never>);
+
+      expect(
+        mockSubscriptionService.publishToEventStream,
+      ).toHaveBeenCalledTimes(1);
+      const publishCall = (
+        mockSubscriptionService.publishToEventStream as jest.Mock
+      ).mock.calls[0][0];
+
+      expect(publishCall.payload.objectRecordEventsWithQueryIds).toHaveLength(
+        1,
+      );
+      expect(
+        publishCall.payload.objectRecordEventsWithQueryIds[0].queryIds,
+      ).toEqual(['query-1']);
+    });
+
+    it('should not publish update events when the delivered state fails the RLS filter, even if the before state matched', async () => {
+      const rlsFilter: RecordGqlOperationFilter = { status: { eq: 'active' } };
+
+      (buildRowLevelPermissionRecordFilter as jest.Mock).mockReturnValue(
+        rlsFilter,
+      );
+
+      (
+        isRecordMatchingRLSRowLevelPermissionPredicate as jest.Mock
+      ).mockImplementation(
+        ({
+          record,
+          filter,
+        }: {
+          record: { status?: string };
+          filter: RecordGqlOperationFilter;
+        }) => ('status' in filter ? record.status === 'active' : true),
+      );
+
+      const eventBatch: WorkspaceEventBatch<MockObjectRecordEvent> = {
+        name: 'company.updated',
+        workspaceId,
+        objectMetadata: companyObjectMetadata,
+        events: [
+          createMockEvent({
+            properties: {
+              before: {
+                id: 'record-1',
+                name: 'Test Company',
+                status: 'active',
+              },
+              after: {
+                id: 'record-1',
+                name: 'Test Company',
+                status: 'archived',
+              },
+            } as MockObjectRecordEvent['properties'],
+          }),
+        ],
+      };
+
+      await service.publish(eventBatch as WorkspaceEventBatch<never>);
+
+      expect(
+        mockSubscriptionService.publishToEventStream,
+      ).not.toHaveBeenCalled();
+    });
+
+    it('should not publish update events when neither state matches the filter', async () => {
+      (
+        isRecordMatchingRLSRowLevelPermissionPredicate as jest.Mock
+      ).mockReturnValue(false);
+
+      const streamDataWithFilter: EventStreamData = {
+        ...mockStreamData,
+        queries: {
+          'query-1': {
+            objectNameSingular: 'company',
+            variables: {
+              filter: { name: { eq: 'Open Company' } },
+            },
+          },
+        },
+      };
+
+      mockEventStreamService.getStreamsData.mockResolvedValue(
+        new Map([[streamChannelId, streamDataWithFilter]]) as Map<
+          string,
+          EventStreamData | undefined
+        >,
+      );
+
+      const eventBatch: WorkspaceEventBatch<MockObjectRecordEvent> = {
+        name: 'company.updated',
+        workspaceId,
+        objectMetadata: companyObjectMetadata,
+        events: [
+          createMockEvent({
+            properties: {
+              before: { id: 'record-1', name: 'Unrelated A' },
+              after: { id: 'record-1', name: 'Unrelated B' },
+            } as MockObjectRecordEvent['properties'],
+          }),
+        ],
+      };
+
+      await service.publish(eventBatch as WorkspaceEventBatch<never>);
+
+      expect(
+        mockSubscriptionService.publishToEventStream,
+      ).not.toHaveBeenCalled();
+    });
+
    it('should filter restricted fields from events', async () => {
      const restrictedField = getFlatFieldMetadataMock({
        objectMetadataId: companyObjectMetadata.id,
@@ -915,7 +1066,7 @@ describe('ObjectRecordEventPublisher', () => {
      });
    });

-    it('should combine query filter with RLS filter', async () => {
+    it('should check the RLS filter and the query filter separately', async () => {
      const rlsFilter: RecordGqlOperationFilter = { status: { eq: 'active' } };

      (buildRowLevelPermissionRecordFilter as jest.Mock).mockReturnValue(
@@ -968,12 +1119,18 @@ describe('ObjectRecordEventPublisher', () => {
            name: 'Test Company',
            status: 'active',
          }),
-          filter: expect.objectContaining({
-            and: expect.arrayContaining([
-              { name: { eq: 'Test Company' } },
-              { status: { eq: 'active' } },
-            ]),
+          filter: { status: { eq: 'active' } },
+        }),
+      );
+      expect(
+        isRecordMatchingRLSRowLevelPermissionPredicate,
+      ).toHaveBeenCalledWith(
+        expect.objectContaining({
+          record: expect.objectContaining({
+            name: 'Test Company',
+            status: 'active',
          }),
+          filter: { name: { eq: 'Test Company' } },
        }),
      );
    });
@@ -12,7 +12,6 @@ import {
  type RestrictedFieldsPermissions,
 } from 'twenty-shared/types';
 import {
-  combineFilters,
  isDefined,
  isNonEmptyArray,
  isRecordGqlOperationSignature,
@@ -547,37 +546,50 @@ export class ObjectRecordEventPublisher {
      before?: object;
    };

-    const record = properties?.after ?? properties?.before;
+    const deliveredRecord = properties?.after ?? properties?.before;

-    if (!isDefined(record)) {
+    if (!isDefined(deliveredRecord)) {
      return false;
    }

-    const queryFilter = operationSignature.variables?.filter ?? {};
-
-    const filtersToApply: RecordGqlOperationFilter[] = [queryFilter];
-
-    if (subscriberRLSFilter && Object.keys(subscriberRLSFilter).length > 0) {
-      filtersToApply.push(subscriberRLSFilter);
-    }
-
-    const combinedFilter = combineFilters(filtersToApply);
-
-    if (Object.keys(combinedFilter).length === 0) {
-      return true;
-    }
-
    const shouldIgnoreSoftDeleteDefaultFilter =
      event.action === DatabaseEventAction.DELETED ||
      event.action === DatabaseEventAction.RESTORED;

-    return isRecordMatchingRLSRowLevelPermissionPredicate({
-      record,
-      filter: combinedFilter,
-      flatObjectMetadata: objectMetadata,
-      flatFieldMetadataMaps,
-      shouldIgnoreSoftDeleteDefaultFilter,
-    });
+    if (
+      isDefined(subscriberRLSFilter) &&
+      Object.keys(subscriberRLSFilter).length > 0 &&
+      !isRecordMatchingRLSRowLevelPermissionPredicate({
+        record: deliveredRecord,
+        filter: subscriberRLSFilter,
+        flatObjectMetadata: objectMetadata,
+        flatFieldMetadataMaps,
+        shouldIgnoreSoftDeleteDefaultFilter,
+      })
+    ) {
+      return false;
+    }
+
+    const queryFilter = operationSignature.variables?.filter ?? {};
+
+    if (Object.keys(queryFilter).length === 0) {
+      return true;
+    }
+
+    const candidateRecords =
+      event.action === DatabaseEventAction.UPDATED
+        ? [properties?.after, properties?.before].filter(isDefined)
+        : [deliveredRecord];
+
+    return candidateRecords.some((record) =>
+      isRecordMatchingRLSRowLevelPermissionPredicate({
+        record,
+        filter: queryFilter,
+        flatObjectMetadata: objectMetadata,
+        flatFieldMetadataMaps,
+        shouldIgnoreSoftDeleteDefaultFilter,
+      }),
+    );
  }

  private async fetchPermissionsContext(