diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/__tests__/workspace-event-emitter.service.spec.ts b/packages/twenty-server/src/engine/workspace-event-emitter/__tests__/workspace-event-emitter.service.spec.ts new file mode 100644 index 0000000000..4fa3c95426 --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-event-emitter/__tests__/workspace-event-emitter.service.spec.ts @@ -0,0 +1,980 @@ +import { Test, type TestingModule } from '@nestjs/testing'; + +import { + type ObjectsPermissionsByRoleId, + type RecordGqlOperationFilter, +} from 'twenty-shared/types'; + +import { createEmptyFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/constant/create-empty-flat-entity-maps.constant'; +import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type'; +import { addFlatEntityToFlatEntityMapsOrThrow } from 'src/engine/metadata-modules/flat-entity/utils/add-flat-entity-to-flat-entity-maps-or-throw.util'; +import { COMPANY_FLAT_FIELDS_MOCK } from 'src/engine/metadata-modules/flat-field-metadata/__mocks__/company-flat-fields.mock'; +import { getFlatFieldMetadataMock } from 'src/engine/metadata-modules/flat-field-metadata/__mocks__/get-flat-field-metadata.mock'; +import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type'; +import { COMPANY_FLAT_OBJECT_MOCK } from 'src/engine/metadata-modules/flat-object-metadata/__mocks__/company-flat-object.mock'; +import { type FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type'; +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 { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; +import { type WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type'; +import { WorkspaceEventEmitterService } from 'src/engine/workspace-event-emitter/workspace-event-emitter.service'; + +jest.mock( + 'src/engine/twenty-orm/utils/build-row-level-permission-record-filter.util', + () => ({ + buildRowLevelPermissionRecordFilter: jest.fn(), + }), +); + +jest.mock( + 'src/engine/twenty-orm/utils/is-record-matching-rls-row-level-permission-predicate.util', + () => ({ + isRecordMatchingRLSRowLevelPermissionPredicate: jest.fn(), + }), +); + +const { + buildRowLevelPermissionRecordFilter, +} = require('src/engine/twenty-orm/utils/build-row-level-permission-record-filter.util'); +const { + isRecordMatchingRLSRowLevelPermissionPredicate, +} = require('src/engine/twenty-orm/utils/is-record-matching-rls-row-level-permission-predicate.util'); + +type MockObjectRecordEvent = { + recordId: string; + userId?: string; + workspaceMemberId?: string; + properties: { + before?: object; + after?: object; + updatedFields?: string[]; + diff?: object; + }; +}; + +const buildFlatFieldMetadataMaps = ( + fields: FlatFieldMetadata[], +): FlatEntityMaps => + fields.reduce( + (maps, field) => + addFlatEntityToFlatEntityMapsOrThrow({ + flatEntity: field, + flatEntityMaps: maps, + }), + createEmptyFlatEntityMaps() as FlatEntityMaps, + ); + +describe('WorkspaceEventEmitterService', () => { + let service: WorkspaceEventEmitterService; + let mockSubscriptionService: jest.Mocked< + Pick + >; + let mockEventStreamService: jest.Mocked< + Pick< + EventStreamService, + 'getActiveStreamIds' | 'getStreamsData' | 'removeFromActiveStreams' + > + >; + let mockWorkspaceCacheService: { + getOrRecompute: jest.Mock; + }; + + const workspaceId = COMPANY_FLAT_OBJECT_MOCK.workspaceId; + const streamChannelId = 'test-stream-channel-id'; + const userWorkspaceId = 'test-user-workspace-id'; + const roleId = 'test-role-id'; + + const companyObjectMetadata: FlatObjectMetadata = COMPANY_FLAT_OBJECT_MOCK; + + const companyNameField = COMPANY_FLAT_FIELDS_MOCK.name; + + const mockFlatFieldMetadataMaps = buildFlatFieldMetadataMaps([ + companyNameField, + ]); + + const mockUserWorkspaceRoleMap: Record = { + [userWorkspaceId]: roleId, + }; + + const mockRolesPermissions: ObjectsPermissionsByRoleId = { + [roleId]: { + [companyObjectMetadata.id]: { + canReadObjectRecords: true, + canUpdateObjectRecords: true, + canSoftDeleteObjectRecords: true, + canDestroyObjectRecords: true, + restrictedFields: {}, + rowLevelPermissionPredicates: [], + rowLevelPermissionPredicateGroups: [], + }, + }, + }; + + const mockStreamData: EventStreamData = { + authContext: { + userWorkspaceId, + userId: 'test-user-id', + workspaceMemberId: 'test-workspace-member-id', + }, + workspaceId, + queries: { + 'query-1': { + objectNameSingular: 'company', + variables: {}, + }, + }, + createdAt: Date.now(), + }; + + const createMockEvent = ( + overrides: Partial = {}, + ): MockObjectRecordEvent => ({ + recordId: 'record-1', + userId: 'test-user-id', + workspaceMemberId: 'test-workspace-member-id', + properties: { + after: { id: 'record-1', name: 'Test Company' }, + }, + ...overrides, + }); + + const createPermissionsContext = ( + overrides: { + flatFieldMetadataMaps?: FlatEntityMaps; + userWorkspaceRoleMap?: Record; + rolesPermissions?: ObjectsPermissionsByRoleId; + } = {}, + ) => ({ + flatRowLevelPermissionPredicateMaps: { + byId: {}, + idByUniversalIdentifier: {}, + universalIdentifiersByApplicationId: {}, + }, + flatRowLevelPermissionPredicateGroupMaps: { + byId: {}, + idByUniversalIdentifier: {}, + universalIdentifiersByApplicationId: {}, + }, + flatFieldMetadataMaps: + overrides.flatFieldMetadataMaps ?? mockFlatFieldMetadataMaps, + userWorkspaceRoleMap: + overrides.userWorkspaceRoleMap ?? mockUserWorkspaceRoleMap, + rolesPermissions: overrides.rolesPermissions ?? mockRolesPermissions, + }); + + beforeEach(async () => { + jest.clearAllMocks(); + + mockSubscriptionService = { + publish: jest.fn().mockResolvedValue(undefined), + publishToEventStream: jest.fn().mockResolvedValue(undefined), + }; + + mockEventStreamService = { + getActiveStreamIds: jest.fn().mockResolvedValue([streamChannelId]), + getStreamsData: jest + .fn() + .mockResolvedValue( + new Map([[streamChannelId, mockStreamData]]) as Map< + string, + EventStreamData | undefined + >, + ), + removeFromActiveStreams: jest.fn().mockResolvedValue(undefined), + }; + + mockWorkspaceCacheService = { + getOrRecompute: jest.fn().mockResolvedValue(createPermissionsContext()), + }; + + (buildRowLevelPermissionRecordFilter as jest.Mock).mockReturnValue({}); + ( + isRecordMatchingRLSRowLevelPermissionPredicate as jest.Mock + ).mockReturnValue(true); + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + WorkspaceEventEmitterService, + { + provide: SubscriptionService, + useValue: mockSubscriptionService, + }, + { + provide: EventStreamService, + useValue: mockEventStreamService, + }, + { + provide: WorkspaceCacheService, + useValue: mockWorkspaceCacheService, + }, + ], + }).compile(); + + service = module.get( + WorkspaceEventEmitterService, + ); + }); + + it('should be defined', () => { + expect(service).toBeDefined(); + }); + + describe('publish', () => { + it('should skip publishing to event streams when no active streams exist', async () => { + mockEventStreamService.getActiveStreamIds.mockResolvedValue([]); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect(mockEventStreamService.getActiveStreamIds).toHaveBeenCalledWith( + workspaceId, + ); + expect(mockEventStreamService.getStreamsData).not.toHaveBeenCalled(); + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + + it('should publish events when record matches query and permissions', async () => { + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect(mockSubscriptionService.publishToEventStream).toHaveBeenCalled(); + const publishCall = ( + mockSubscriptionService.publishToEventStream as jest.Mock + ).mock.calls[0][0]; + + expect(publishCall.workspaceId).toBe(workspaceId); + expect(publishCall.eventStreamChannelId).toBe(streamChannelId); + expect(publishCall.payload).toHaveLength(1); + expect(publishCall.payload[0].queryIds).toContain('query-1'); + }); + + it('should not publish events when object-level read permission is denied', async () => { + const permissionsWithoutRead: ObjectsPermissionsByRoleId = { + [roleId]: { + [companyObjectMetadata.id]: { + canReadObjectRecords: false, + canUpdateObjectRecords: true, + canSoftDeleteObjectRecords: true, + canDestroyObjectRecords: true, + restrictedFields: {}, + rowLevelPermissionPredicates: [], + rowLevelPermissionPredicateGroups: [], + }, + }, + }; + + mockWorkspaceCacheService.getOrRecompute.mockResolvedValue( + createPermissionsContext({ rolesPermissions: permissionsWithoutRead }), + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + + it('should not publish events when query object name does not match event', async () => { + const streamDataWithDifferentObject: EventStreamData = { + ...mockStreamData, + queries: { + 'query-1': { + objectNameSingular: 'person', + variables: {}, + }, + }, + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, streamDataWithDifferentObject]]) as Map< + string, + EventStreamData | undefined + >, + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + + it('should not publish events when record does not match RLS filter', async () => { + ( + isRecordMatchingRLSRowLevelPermissionPredicate as jest.Mock + ).mockReturnValue(false); + + const streamDataWithFilter: EventStreamData = { + ...mockStreamData, + queries: { + 'query-1': { + objectNameSingular: 'company', + variables: { + filter: { name: { eq: 'Other Company' } }, + }, + }, + }, + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, streamDataWithFilter]]) as Map< + string, + EventStreamData | undefined + >, + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + + it('should filter restricted fields from events', async () => { + const restrictedField = getFlatFieldMetadataMock({ + objectMetadataId: companyObjectMetadata.id, + type: COMPANY_FLAT_FIELDS_MOCK.name.type, + name: 'secretField', + universalIdentifier: 'restricted-field-universal-id', + workspaceId, + }); + + const permissionsWithRestrictedFields: ObjectsPermissionsByRoleId = { + [roleId]: { + [companyObjectMetadata.id]: { + canReadObjectRecords: true, + canUpdateObjectRecords: true, + canSoftDeleteObjectRecords: true, + canDestroyObjectRecords: true, + restrictedFields: { + [restrictedField.id]: { canRead: false, canUpdate: false }, + }, + rowLevelPermissionPredicates: [], + rowLevelPermissionPredicateGroups: [], + }, + }, + }; + + const fieldMetadataMapsWithRestricted = buildFlatFieldMetadataMaps([ + restrictedField, + ]); + + mockWorkspaceCacheService.getOrRecompute.mockResolvedValue( + createPermissionsContext({ + flatFieldMetadataMaps: fieldMetadataMapsWithRestricted, + rolesPermissions: permissionsWithRestrictedFields, + }), + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [ + createMockEvent({ + properties: { + after: { + id: 'record-1', + name: 'Test Company', + secretField: 'secret-value', + }, + }, + }), + ], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect(mockSubscriptionService.publishToEventStream).toHaveBeenCalled(); + const publishCall = ( + 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', + ); + }); + + it('should skip update events when all updated fields are restricted', async () => { + const restrictedField = getFlatFieldMetadataMock({ + objectMetadataId: companyObjectMetadata.id, + type: COMPANY_FLAT_FIELDS_MOCK.name.type, + name: 'secretField', + universalIdentifier: 'restricted-field-universal-id', + workspaceId, + }); + + const permissionsWithRestrictedFields: ObjectsPermissionsByRoleId = { + [roleId]: { + [companyObjectMetadata.id]: { + canReadObjectRecords: true, + canUpdateObjectRecords: true, + canSoftDeleteObjectRecords: true, + canDestroyObjectRecords: true, + restrictedFields: { + [restrictedField.id]: { canRead: false, canUpdate: false }, + }, + rowLevelPermissionPredicates: [], + rowLevelPermissionPredicateGroups: [], + }, + }, + }; + + const fieldMetadataMapsWithRestricted = buildFlatFieldMetadataMaps([ + restrictedField, + ]); + + mockWorkspaceCacheService.getOrRecompute.mockResolvedValue( + createPermissionsContext({ + flatFieldMetadataMaps: fieldMetadataMapsWithRestricted, + rolesPermissions: permissionsWithRestrictedFields, + }), + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.updated', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [ + createMockEvent({ + properties: { + before: { id: 'record-1', secretField: 'old-secret' }, + after: { id: 'record-1', secretField: 'new-secret' }, + updatedFields: ['secretField'], + diff: { + secretField: { before: 'old-secret', after: 'new-secret' }, + }, + }, + }), + ], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + + it('should filter diff when restricted fields are updated', async () => { + const restrictedField = getFlatFieldMetadataMock({ + objectMetadataId: companyObjectMetadata.id, + type: COMPANY_FLAT_FIELDS_MOCK.name.type, + name: 'secretField', + universalIdentifier: 'restricted-field-universal-id', + workspaceId, + }); + + const permissionsWithRestrictedFields: ObjectsPermissionsByRoleId = { + [roleId]: { + [companyObjectMetadata.id]: { + canReadObjectRecords: true, + canUpdateObjectRecords: true, + canSoftDeleteObjectRecords: true, + canDestroyObjectRecords: true, + restrictedFields: { + [restrictedField.id]: { canRead: false, canUpdate: false }, + }, + rowLevelPermissionPredicates: [], + rowLevelPermissionPredicateGroups: [], + }, + }, + }; + + const fieldMetadataMapsWithRestricted = buildFlatFieldMetadataMaps([ + restrictedField, + ]); + + mockWorkspaceCacheService.getOrRecompute.mockResolvedValue( + createPermissionsContext({ + flatFieldMetadataMaps: fieldMetadataMapsWithRestricted, + rolesPermissions: permissionsWithRestrictedFields, + }), + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.updated', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [ + createMockEvent({ + properties: { + before: { + id: 'record-1', + name: 'Old Name', + secretField: 'old-secret', + }, + after: { + id: 'record-1', + name: 'New Name', + secretField: 'new-secret', + }, + updatedFields: ['name', 'secretField'], + diff: { + name: { before: 'Old Name', after: 'New Name' }, + secretField: { before: 'old-secret', after: 'new-secret' }, + }, + }, + }), + ], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect(mockSubscriptionService.publishToEventStream).toHaveBeenCalled(); + const publishCall = ( + mockSubscriptionService.publishToEventStream as jest.Mock + ).mock.calls[0][0]; + + const eventPayload = publishCall.payload[0].event; + + expect(eventPayload.properties.updatedFields).toEqual(['name']); + expect(eventPayload.properties.diff).not.toHaveProperty('secretField'); + expect(eventPayload.properties.diff).toHaveProperty('name'); + expect(eventPayload.properties.before).not.toHaveProperty('secretField'); + expect(eventPayload.properties.after).not.toHaveProperty('secretField'); + }); + + it('should remove stale streams from active streams', async () => { + const staleStreamId = 'stale-stream-id'; + + mockEventStreamService.getActiveStreamIds.mockResolvedValue([ + streamChannelId, + staleStreamId, + ]); + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([ + [streamChannelId, mockStreamData], + [staleStreamId, undefined], + ]) as Map, + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockEventStreamService.removeFromActiveStreams, + ).toHaveBeenCalledWith(workspaceId, [staleStreamId]); + }); + + it('should skip streams with no registered queries', async () => { + const streamDataWithNoQueries: EventStreamData = { + ...mockStreamData, + queries: {}, + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, streamDataWithNoQueries]]) as Map< + string, + EventStreamData | undefined + >, + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + + it('should not publish when user has no role assigned', async () => { + mockWorkspaceCacheService.getOrRecompute.mockResolvedValue( + createPermissionsContext({ userWorkspaceRoleMap: {} }), + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + + it('should combine query filter with RLS filter', async () => { + const rlsFilter: RecordGqlOperationFilter = { status: { eq: 'active' } }; + + (buildRowLevelPermissionRecordFilter as jest.Mock).mockReturnValue( + rlsFilter, + ); + + const streamDataWithFilter: EventStreamData = { + ...mockStreamData, + queries: { + 'query-1': { + objectNameSingular: 'company', + variables: { + filter: { name: { eq: 'Test Company' } }, + }, + }, + }, + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, streamDataWithFilter]]) as Map< + string, + EventStreamData | undefined + >, + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [ + createMockEvent({ + properties: { + after: { + id: 'record-1', + name: 'Test Company', + status: 'active', + }, + }, + }), + ], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + isRecordMatchingRLSRowLevelPermissionPredicate, + ).toHaveBeenCalledWith( + expect.objectContaining({ + record: expect.objectContaining({ + name: 'Test Company', + status: 'active', + }), + filter: expect.objectContaining({ + and: expect.arrayContaining([ + { name: { eq: 'Test Company' } }, + { status: { eq: 'active' } }, + ]), + }), + }), + ); + }); + + it('should handle multiple events in a batch', async () => { + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [ + createMockEvent({ + recordId: 'record-1', + properties: { after: { id: 'record-1', name: 'Company 1' } }, + }), + createMockEvent({ + recordId: 'record-2', + properties: { after: { id: 'record-2', name: 'Company 2' } }, + }), + ], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect(mockSubscriptionService.publishToEventStream).toHaveBeenCalled(); + const publishCall = ( + mockSubscriptionService.publishToEventStream as jest.Mock + ).mock.calls[0][0]; + + expect(publishCall.payload).toHaveLength(2); + }); + + it('should handle multiple matching queries', async () => { + const streamDataWithMultipleQueries: EventStreamData = { + ...mockStreamData, + queries: { + 'query-1': { + objectNameSingular: 'company', + variables: {}, + }, + 'query-2': { + objectNameSingular: 'company', + variables: {}, + }, + }, + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, streamDataWithMultipleQueries]]) as Map< + string, + EventStreamData | undefined + >, + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect(mockSubscriptionService.publishToEventStream).toHaveBeenCalled(); + const publishCall = ( + 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'); + }); + + it('should use before record for delete events', async () => { + const streamDataWithFilter: EventStreamData = { + ...mockStreamData, + queries: { + 'query-1': { + objectNameSingular: 'company', + variables: { + filter: { name: { eq: 'Deleted Company' } }, + }, + }, + }, + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, streamDataWithFilter]]) as Map< + string, + EventStreamData | undefined + >, + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.deleted', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [ + createMockEvent({ + properties: { + before: { id: 'record-1', name: 'Deleted Company' }, + }, + }), + ], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + isRecordMatchingRLSRowLevelPermissionPredicate, + ).toHaveBeenCalledWith( + expect.objectContaining({ + record: expect.objectContaining({ + id: 'record-1', + name: 'Deleted Company', + }), + }), + ); + }); + + describe('subscribers without valid authentication', () => { + it('should not publish events and log warning when subscriber has no userWorkspaceId', async () => { + const anonymousStreamData: EventStreamData = { + authContext: {}, + workspaceId, + queries: { + 'query-1': { + objectNameSingular: 'company', + variables: {}, + }, + }, + createdAt: Date.now(), + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, anonymousStreamData]]) as Map< + string, + EventStreamData | undefined + >, + ); + + mockWorkspaceCacheService.getOrRecompute.mockResolvedValue( + createPermissionsContext(), + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + + it('should not publish events and log warning when user role cannot be found', async () => { + const unknownUserStreamData: EventStreamData = { + authContext: { + userWorkspaceId: 'unknown-user-workspace-id', + userId: 'unknown-user-id', + }, + workspaceId, + queries: { + 'query-1': { + objectNameSingular: 'company', + variables: {}, + }, + }, + createdAt: Date.now(), + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, unknownUserStreamData]]) as Map< + string, + EventStreamData | undefined + >, + ); + + mockWorkspaceCacheService.getOrRecompute.mockResolvedValue( + createPermissionsContext({ userWorkspaceRoleMap: {} }), + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [createMockEvent()], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect( + mockSubscriptionService.publishToEventStream, + ).not.toHaveBeenCalled(); + }); + }); + + describe('dynamic RLS predicates', () => { + it('should pass workspaceMemberId to buildRowLevelPermissionRecordFilter for dynamic predicates', async () => { + const workspaceMemberId = 'test-workspace-member-id'; + const rlsFilter: RecordGqlOperationFilter = { + assigneeId: { eq: workspaceMemberId }, + }; + + (buildRowLevelPermissionRecordFilter as jest.Mock).mockReturnValue( + rlsFilter, + ); + + const streamDataWithWorkspaceMember: EventStreamData = { + authContext: { + userWorkspaceId, + userId: 'test-user-id', + workspaceMemberId, + }, + workspaceId, + queries: { + 'query-1': { + objectNameSingular: 'company', + variables: { + filter: { name: { eq: 'Test Company' } }, + }, + }, + }, + createdAt: Date.now(), + }; + + mockEventStreamService.getStreamsData.mockResolvedValue( + new Map([[streamChannelId, streamDataWithWorkspaceMember]]) as Map< + string, + EventStreamData | undefined + >, + ); + + mockWorkspaceCacheService.getOrRecompute.mockResolvedValue( + createPermissionsContext(), + ); + + const eventBatch: WorkspaceEventBatch = { + name: 'company.created', + workspaceId, + objectMetadata: companyObjectMetadata, + events: [ + createMockEvent({ + properties: { + after: { + id: 'record-1', + name: 'Test Company', + assigneeId: workspaceMemberId, + }, + }, + }), + ], + }; + + await service.publish(eventBatch as WorkspaceEventBatch); + + expect(buildRowLevelPermissionRecordFilter).toHaveBeenCalledWith( + expect.objectContaining({ + authContext: expect.objectContaining({ + userWorkspaceId, + workspaceMemberId, + }), + }), + ); + }); + }); + }); +}); diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts index 14dfaaf57c..cc5b08bf4c 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.module.ts @@ -1,13 +1,14 @@ import { Global, Module } from '@nestjs/common'; import { SubscriptionsModule } from 'src/engine/subscriptions/subscriptions.module'; +import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache.module'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkspaceEventEmitterResolver } from 'src/engine/workspace-event-emitter/workspace-event-emitter.resolver'; import { WorkspaceEventEmitterService } from 'src/engine/workspace-event-emitter/workspace-event-emitter.service'; @Global() @Module({ - imports: [SubscriptionsModule], + imports: [SubscriptionsModule, WorkspaceCacheModule], providers: [ WorkspaceEventEmitter, WorkspaceEventEmitterService, diff --git a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts index 879b9da49d..20cb98fe0b 100644 --- a/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts +++ b/packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts @@ -1,14 +1,28 @@ import { Injectable } from '@nestjs/common'; import { type ObjectRecordEvent } from 'twenty-shared/database-events'; -import { isDefined } from 'twenty-shared/utils'; +import { + type ObjectsPermissionsByRoleId, + type RecordGqlOperationFilter, + type RestrictedFieldsPermissions, +} from 'twenty-shared/types'; +import { combineFilters, isDefined } from 'twenty-shared/utils'; +import { type SerializableAuthContext } from 'src/engine/core-modules/auth/types/auth-context.type'; import { transformEventToWebhookEvent } from 'src/engine/core-modules/webhook/utils/transform-event-to-webhook-event'; +import { type FlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/flat-entity-maps.type'; +import { type FlatFieldMetadata } from 'src/engine/metadata-modules/flat-field-metadata/types/flat-field-metadata.type'; +import { type FlatObjectMetadata } from 'src/engine/metadata-modules/flat-object-metadata/types/flat-object-metadata.type'; +import { type FlatRowLevelPermissionPredicateGroupMaps } from 'src/engine/metadata-modules/row-level-permission-predicate/types/flat-row-level-permission-predicate-group-maps.type'; +import { type FlatRowLevelPermissionPredicateMaps } from 'src/engine/metadata-modules/row-level-permission-predicate/types/flat-row-level-permission-predicate-maps.type'; 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 { ObjectRecordSubscriptionEvent } from 'src/engine/subscriptions/types/object-record-subscription-event.type'; +import { buildRowLevelPermissionRecordFilter } from 'src/engine/twenty-orm/utils/build-row-level-permission-record-filter.util'; +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 { parseEventNameOrThrow } from 'src/engine/workspace-event-emitter/utils/parse-event-name'; @@ -17,6 +31,7 @@ export class WorkspaceEventEmitterService { constructor( private readonly subscriptionService: SubscriptionService, private readonly eventStreamService: EventStreamService, + private readonly workspaceCacheService: WorkspaceCacheService, ) {} async publish( @@ -66,6 +81,8 @@ export class WorkspaceEventEmitterService { activeStreamIds, ); + const permissionsContext = await this.fetchPermissionsContext(workspaceId); + const streamIdsToRemove: string[] = []; for (const [streamChannelId, streamData] of streamsData) { @@ -74,10 +91,15 @@ export class WorkspaceEventEmitterService { continue; } + if (Object.keys(streamData.queries).length === 0) { + continue; + } + await this.processStreamEvents( streamChannelId, streamData, workspaceEventBatch, + permissionsContext, ); } @@ -91,7 +113,35 @@ export class WorkspaceEventEmitterService { streamChannelId: string, streamData: EventStreamData, workspaceEventBatch: WorkspaceEventBatch, + permissionsContext: { + flatRowLevelPermissionPredicateMaps: FlatRowLevelPermissionPredicateMaps; + flatRowLevelPermissionPredicateGroupMaps: FlatRowLevelPermissionPredicateGroupMaps; + flatFieldMetadataMaps: FlatEntityMaps; + userWorkspaceRoleMap: Record; + rolesPermissions: ObjectsPermissionsByRoleId; + }, ): Promise { + const { userWorkspaceId } = streamData.authContext; + + if (!isDefined(userWorkspaceId)) { + return; + } + + const roleId = permissionsContext.userWorkspaceRoleMap[userWorkspaceId]; + + if (!isDefined(roleId)) { + return; + } + + const objectPermissions = + permissionsContext.rolesPermissions[roleId]?.[ + workspaceEventBatch.objectMetadata.id + ]; + + if (!objectPermissions?.canReadObjectRecords) { + return; + } + const matchedEvents: { queryIds: string[]; event: ObjectRecordEvent & { objectNameSingular: string }; @@ -99,6 +149,15 @@ export class WorkspaceEventEmitterService { const objectNameSingular = workspaceEventBatch.objectMetadata.nameSingular; + const subscriberRLSFilter = this.buildSubscriberRLSFilter( + streamData.authContext, + roleId, + workspaceEventBatch.objectMetadata, + permissionsContext, + ); + + const restrictedFields = objectPermissions.restrictedFields; + for (const event of workspaceEventBatch.events) { const { action } = parseEventNameOrThrow(workspaceEventBatch.name); @@ -108,9 +167,29 @@ export class WorkspaceEventEmitterService { ...event, }; - const matchedQueryIds = this.eventStreamService.matchQueriesWithEvent( - streamData.queries, + const filteredEvent = this.filterRestrictedFieldsFromEvent( eventWithObjectName, + restrictedFields, + permissionsContext.flatFieldMetadataMaps, + ); + + const filteredProperties = filteredEvent.properties as { + updatedFields?: string[]; + }; + + if ( + isDefined(filteredProperties.updatedFields) && + filteredProperties.updatedFields.length === 0 + ) { + continue; + } + + const matchedQueryIds = this.getMatchingQueryIds( + streamData.queries, + filteredEvent, + subscriberRLSFilter, + workspaceEventBatch.objectMetadata, + permissionsContext.flatFieldMetadataMaps, ); if (matchedQueryIds.length === 0) { @@ -119,7 +198,7 @@ export class WorkspaceEventEmitterService { matchedEvents.push({ queryIds: matchedQueryIds, - event: eventWithObjectName, + event: filteredEvent, }); } @@ -131,4 +210,201 @@ export class WorkspaceEventEmitterService { }); } } + + private buildSubscriberRLSFilter( + subscriberAuthContext: SerializableAuthContext, + roleId: string, + objectMetadata: FlatObjectMetadata, + permissionsContext: { + flatRowLevelPermissionPredicateMaps: FlatRowLevelPermissionPredicateMaps; + flatRowLevelPermissionPredicateGroupMaps: FlatRowLevelPermissionPredicateGroupMaps; + flatFieldMetadataMaps: FlatEntityMaps; + }, + ): RecordGqlOperationFilter | null { + return buildRowLevelPermissionRecordFilter({ + flatRowLevelPermissionPredicateMaps: + permissionsContext.flatRowLevelPermissionPredicateMaps, + flatRowLevelPermissionPredicateGroupMaps: + permissionsContext.flatRowLevelPermissionPredicateGroupMaps, + flatFieldMetadataMaps: permissionsContext.flatFieldMetadataMaps, + objectMetadata, + roleId, + // TODO(t.trompette): For dynamic predicates, we would need to load workspaceMember data + authContext: { + userWorkspaceId: subscriberAuthContext.userWorkspaceId, + workspaceMemberId: subscriberAuthContext.workspaceMemberId, + }, + }); + } + + private filterRestrictedFieldsFromEvent( + event: ObjectRecordEvent & { objectNameSingular: string }, + restrictedFields: RestrictedFieldsPermissions | undefined, + flatFieldMetadataMaps: FlatEntityMaps, + ): ObjectRecordEvent & { objectNameSingular: string } { + if (!restrictedFields || Object.keys(restrictedFields).length === 0) { + return event; + } + + const restrictedFieldNames = new Set( + Object.entries(restrictedFields) + .filter(([, permissions]) => permissions.canRead === false) + .map(([fieldMetadataId]) => { + const fieldMetadata = flatFieldMetadataMaps.byId[fieldMetadataId]; + + return fieldMetadata?.name; + }) + .filter(isDefined), + ); + + if (restrictedFieldNames.size === 0) { + return event; + } + + const filterRecord = (record: object | undefined): object | undefined => { + if (!record) { + return record; + } + + return Object.fromEntries( + Object.entries(record).filter( + ([key]) => !restrictedFieldNames.has(key), + ), + ); + }; + + const properties = event.properties as { + before?: object; + after?: object; + updatedFields?: string[]; + diff?: object; + }; + + const filteredBefore = filterRecord(properties.before); + const filteredAfter = filterRecord(properties.after); + const filteredDiff = filterRecord(properties.diff); + + const filteredProperties = { + ...properties, + ...(filteredBefore !== undefined && { before: filteredBefore }), + ...(filteredAfter !== undefined && { after: filteredAfter }), + ...(filteredDiff !== undefined && { diff: filteredDiff }), + updatedFields: properties.updatedFields?.filter( + (field) => !restrictedFieldNames.has(field), + ), + }; + + return { + ...event, + properties: filteredProperties, + } as ObjectRecordEvent & { objectNameSingular: string }; + } + + private getMatchingQueryIds( + queries: Record< + string, + { + objectNameSingular: string; + variables?: { filter?: RecordGqlOperationFilter }; + } + >, + event: ObjectRecordEvent & { objectNameSingular: string }, + subscriberRLSFilter: RecordGqlOperationFilter | null, + objectMetadata: FlatObjectMetadata, + flatFieldMetadataMaps: FlatEntityMaps, + ): string[] { + const matchedQueryIds: string[] = []; + + for (const [queryId, operationSignature] of Object.entries(queries)) { + if ( + this.isQueryMatchingEvent( + operationSignature, + event, + subscriberRLSFilter, + objectMetadata, + flatFieldMetadataMaps, + ) + ) { + matchedQueryIds.push(queryId); + } + } + + return matchedQueryIds; + } + + private isQueryMatchingEvent( + operationSignature: { + objectNameSingular: string; + variables?: { filter?: RecordGqlOperationFilter }; + }, + event: ObjectRecordEvent & { objectNameSingular: string }, + subscriberRLSFilter: RecordGqlOperationFilter | null, + objectMetadata: FlatObjectMetadata, + flatFieldMetadataMaps: FlatEntityMaps, + ): boolean { + if (operationSignature.objectNameSingular !== event.objectNameSingular) { + return false; + } + + const properties = event.properties as { + after?: object; + before?: object; + }; + const record = properties?.after ?? properties?.before; + + if (!isDefined(record)) { + 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; + } + + return isRecordMatchingRLSRowLevelPermissionPredicate({ + record, + filter: combinedFilter, + flatObjectMetadata: objectMetadata, + flatFieldMetadataMaps, + }); + } + + private async fetchPermissionsContext(workspaceId: string): Promise<{ + flatRowLevelPermissionPredicateMaps: FlatRowLevelPermissionPredicateMaps; + flatRowLevelPermissionPredicateGroupMaps: FlatRowLevelPermissionPredicateGroupMaps; + flatFieldMetadataMaps: FlatEntityMaps; + userWorkspaceRoleMap: Record; + rolesPermissions: ObjectsPermissionsByRoleId; + }> { + const { + flatRowLevelPermissionPredicateMaps, + flatRowLevelPermissionPredicateGroupMaps, + flatFieldMetadataMaps, + userWorkspaceRoleMap, + rolesPermissions, + } = await this.workspaceCacheService.getOrRecompute(workspaceId, [ + 'flatRowLevelPermissionPredicateMaps', + 'flatRowLevelPermissionPredicateGroupMaps', + 'flatFieldMetadataMaps', + 'userWorkspaceRoleMap', + 'rolesPermissions', + ]); + + return { + flatRowLevelPermissionPredicateMaps, + flatRowLevelPermissionPredicateGroupMaps, + flatFieldMetadataMaps, + userWorkspaceRoleMap, + rolesPermissions, + }; + } }