Add permissions to SSE (#17201)
Flow: - fetch user role from context stored in cache - do not support api context yet - check object permissions - filter restricted fields on event - fetch role rls predicate and combine it with the query filter Do not support dynamic predicates yet.
This commit is contained in:
Thomas Trompette
+980
@@ -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<FlatFieldMetadata> =>
|
||||
fields.reduce(
|
||||
(maps, field) =>
|
||||
addFlatEntityToFlatEntityMapsOrThrow({
|
||||
flatEntity: field,
|
||||
flatEntityMaps: maps,
|
||||
}),
|
||||
createEmptyFlatEntityMaps() as FlatEntityMaps<FlatFieldMetadata>,
|
||||
);
|
||||
|
||||
describe('WorkspaceEventEmitterService', () => {
|
||||
let service: WorkspaceEventEmitterService;
|
||||
let mockSubscriptionService: jest.Mocked<
|
||||
Pick<SubscriptionService, 'publish' | 'publishToEventStream'>
|
||||
>;
|
||||
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<string, string> = {
|
||||
[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> = {},
|
||||
): 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<FlatFieldMetadata>;
|
||||
userWorkspaceRoleMap?: Record<string, string>;
|
||||
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>(
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
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,
|
||||
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<MockObjectRecordEvent> = {
|
||||
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<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
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<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
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<never>);
|
||||
|
||||
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<string, EventStreamData | undefined>,
|
||||
);
|
||||
|
||||
const eventBatch: WorkspaceEventBatch<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
expect(
|
||||
mockSubscriptionService.publishToEventStream,
|
||||
).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should not publish when user has no role assigned', async () => {
|
||||
mockWorkspaceCacheService.getOrRecompute.mockResolvedValue(
|
||||
createPermissionsContext({ userWorkspaceRoleMap: {} }),
|
||||
);
|
||||
|
||||
const eventBatch: WorkspaceEventBatch<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [
|
||||
createMockEvent({
|
||||
properties: {
|
||||
after: {
|
||||
id: 'record-1',
|
||||
name: 'Test Company',
|
||||
status: 'active',
|
||||
},
|
||||
},
|
||||
}),
|
||||
],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
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<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.deleted',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [
|
||||
createMockEvent({
|
||||
properties: {
|
||||
before: { id: 'record-1', name: 'Deleted Company' },
|
||||
},
|
||||
}),
|
||||
],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [createMockEvent()],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
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<MockObjectRecordEvent> = {
|
||||
name: 'company.created',
|
||||
workspaceId,
|
||||
objectMetadata: companyObjectMetadata,
|
||||
events: [
|
||||
createMockEvent({
|
||||
properties: {
|
||||
after: {
|
||||
id: 'record-1',
|
||||
name: 'Test Company',
|
||||
assigneeId: workspaceMemberId,
|
||||
},
|
||||
},
|
||||
}),
|
||||
],
|
||||
};
|
||||
|
||||
await service.publish(eventBatch as WorkspaceEventBatch<never>);
|
||||
|
||||
expect(buildRowLevelPermissionRecordFilter).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
authContext: expect.objectContaining({
|
||||
userWorkspaceId,
|
||||
workspaceMemberId,
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
+2
-1
@@ -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,
|
||||
|
||||
+280
-4
@@ -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<ObjectRecordEvent>,
|
||||
permissionsContext: {
|
||||
flatRowLevelPermissionPredicateMaps: FlatRowLevelPermissionPredicateMaps;
|
||||
flatRowLevelPermissionPredicateGroupMaps: FlatRowLevelPermissionPredicateGroupMaps;
|
||||
flatFieldMetadataMaps: FlatEntityMaps<FlatFieldMetadata>;
|
||||
userWorkspaceRoleMap: Record<string, string>;
|
||||
rolesPermissions: ObjectsPermissionsByRoleId;
|
||||
},
|
||||
): Promise<void> {
|
||||
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<FlatFieldMetadata>;
|
||||
},
|
||||
): 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<FlatFieldMetadata>,
|
||||
): 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<FlatFieldMetadata>,
|
||||
): 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<FlatFieldMetadata>,
|
||||
): 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<FlatFieldMetadata>;
|
||||
userWorkspaceRoleMap: Record<string, string>;
|
||||
rolesPermissions: ObjectsPermissionsByRoleId;
|
||||
}> {
|
||||
const {
|
||||
flatRowLevelPermissionPredicateMaps,
|
||||
flatRowLevelPermissionPredicateGroupMaps,
|
||||
flatFieldMetadataMaps,
|
||||
userWorkspaceRoleMap,
|
||||
rolesPermissions,
|
||||
} = await this.workspaceCacheService.getOrRecompute(workspaceId, [
|
||||
'flatRowLevelPermissionPredicateMaps',
|
||||
'flatRowLevelPermissionPredicateGroupMaps',
|
||||
'flatFieldMetadataMaps',
|
||||
'userWorkspaceRoleMap',
|
||||
'rolesPermissions',
|
||||
]);
|
||||
|
||||
return {
|
||||
flatRowLevelPermissionPredicateMaps,
|
||||
flatRowLevelPermissionPredicateGroupMaps,
|
||||
flatFieldMetadataMaps,
|
||||
userWorkspaceRoleMap,
|
||||
rolesPermissions,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user