From fd495ee61bf40969ee40af7f64b13806f4261c60 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?F=C3=A9lix=20Malfait?= Date: Sat, 18 Apr 2026 12:26:21 +0200 Subject: [PATCH] fix(server): deliver user-scoped metadata events only to the owning user (#19832) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Fixes cross-user cache contamination for AI chat threads in multi-user workspaces. `agentChatThread` is the only user-scoped (`userWorkspaceId`-filtered) entry in `METADATA_NAME_TO_ENTITY_KEY`, but its create/update events were going through `WorkspaceEventBroadcaster`, which fans out to every active SSE stream in the workspace. Every client therefore received other users' threads into their local `agentChatThreads` metadata store (which is persisted to localStorage). On a subsequent session, `AgentChatThreadInitializationEffect` would pick the most-recently-updated thread — potentially another user's — and fire `GetChatMessages` against it; the server's `userWorkspaceId` filter didn't match, producing "Thread not found" errors (now 404 thanks to twentyhq/twenty#19831). ### The fix mirrors the RLS pattern already used for object records `ObjectRecordEventPublisher` already reads `streamData.authContext.userWorkspaceId` to filter per subscriber. Metadata events had no equivalent. This PR closes that gap with a minimal, opt-in change: - Add optional `recipientUserWorkspaceIds?: string[]` to `WorkspaceBroadcastEvent`. Omit → workspace-wide (unchanged for views, objects, fields, etc.). Set → delivered only to streams whose `authContext.userWorkspaceId` is in the list. - `WorkspaceEventBroadcaster.broadcast` builds the payload per stream and skips events whose recipient doesn't match. - Both `agentChatThread` broadcast call sites in `AgentChatService` now pass `recipientUserWorkspaceIds: [userWorkspaceId]`. ### Scope 3 files, +40/-11 LOC: - `subscriptions/workspace-event-broadcaster/types/workspace-broadcast-event.type.ts` - `subscriptions/workspace-event-broadcaster/workspace-event-broadcaster.service.ts` - `metadata-modules/ai/ai-chat/services/agent-chat.service.ts` ### Stale cache note Existing clients still carry poisoned localStorage from before this lands. They self-heal on sign-out/in because `clearAllSessionLocalStorageKeys` already drops `agentChatThreads`. If we want to actively flush on deploy, a frontend cache-version bump can follow in a separate PR. ### Related - twentyhq/twenty#19831 turns the resulting "Thread not found" from 500 to 404. This PR addresses the root cause; #19831 stops the Sentry noise. ## Test plan - [x] `npx nx typecheck twenty-server` - [x] `npx oxlint --type-aware` on all 3 files (0 warnings/errors) - [x] `npx prettier --check` on all 3 files - [ ] Manual: two users in the same workspace, user A creates/sends a chat thread — confirm user B's session does not receive the event and their `chatThreads` sidebar is unaffected - [ ] CI --- .../ai/ai-chat/services/agent-chat.service.ts | 2 + .../types/workspace-broadcast-event.type.ts | 5 +++ .../workspace-event-broadcaster.service.ts | 44 ++++++++++++++----- 3 files changed, 40 insertions(+), 11 deletions(-) diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service.ts index 678efa303ca..71303870400 100644 --- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service.ts +++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/services/agent-chat.service.ts @@ -76,6 +76,7 @@ export class AgentChatService { type: 'created', entityName: 'agentChatThread', recordId: savedThread.id, + recipientUserWorkspaceIds: [userWorkspaceId], properties: { after: serializeThreadForBroadcast(savedThread), }, @@ -339,6 +340,7 @@ export class AgentChatService { type: 'updated', entityName: 'agentChatThread', recordId: threadId, + recipientUserWorkspaceIds: [thread.userWorkspaceId], properties: { updatedFields: ['title'], after: serializeThreadForBroadcast({ ...thread, title }), diff --git a/packages/twenty-server/src/engine/subscriptions/workspace-event-broadcaster/types/workspace-broadcast-event.type.ts b/packages/twenty-server/src/engine/subscriptions/workspace-event-broadcaster/types/workspace-broadcast-event.type.ts index c03a4ba76a2..bc73d038ceb 100644 --- a/packages/twenty-server/src/engine/subscriptions/workspace-event-broadcaster/types/workspace-broadcast-event.type.ts +++ b/packages/twenty-server/src/engine/subscriptions/workspace-event-broadcaster/types/workspace-broadcast-event.type.ts @@ -8,4 +8,9 @@ export type WorkspaceBroadcastEvent = { after?: Record; diff?: Record; }; + // Restricts delivery to streams whose authContext.userWorkspaceId is in this + // list. Omit for workspace-wide events (shared metadata like views, objects, + // fields). Set for user-scoped entities (e.g. agentChatThread) so other users + // in the same workspace don't receive them. + recipientUserWorkspaceIds?: string[]; }; diff --git a/packages/twenty-server/src/engine/subscriptions/workspace-event-broadcaster/workspace-event-broadcaster.service.ts b/packages/twenty-server/src/engine/subscriptions/workspace-event-broadcaster/workspace-event-broadcaster.service.ts index 119a512fed6..36f6ca5dae7 100644 --- a/packages/twenty-server/src/engine/subscriptions/workspace-event-broadcaster/workspace-event-broadcaster.service.ts +++ b/packages/twenty-server/src/engine/subscriptions/workspace-event-broadcaster/workspace-event-broadcaster.service.ts @@ -41,23 +41,45 @@ export class WorkspaceEventBroadcaster { const streamIdsToRemove: string[] = []; - const payload: EventStreamPayload = { - objectRecordEventsWithQueryIds: [], - metadataEvents: events.map((event) => ({ - metadataName: event.entityName, - type: event.type, - recordId: event.recordId, - properties: event.properties, - updatedCollectionHash, - })), - }; - for (const [streamChannelId, streamData] of streamsData) { if (!isDefined(streamData)) { streamIdsToRemove.push(streamChannelId); continue; } + const streamUserWorkspaceId = streamData.authContext.userWorkspaceId; + + const metadataEventsForStream = events + .filter((event) => { + // Events without recipientUserWorkspaceIds are workspace-wide; delivered + // to every stream. Events with the field are user-scoped; only delivered + // to streams whose authContext.userWorkspaceId is in the list. + if (!isDefined(event.recipientUserWorkspaceIds)) { + return true; + } + + return ( + isDefined(streamUserWorkspaceId) && + event.recipientUserWorkspaceIds.includes(streamUserWorkspaceId) + ); + }) + .map((event) => ({ + metadataName: event.entityName, + type: event.type, + recordId: event.recordId, + properties: event.properties, + updatedCollectionHash, + })); + + if (metadataEventsForStream.length === 0) { + continue; + } + + const payload: EventStreamPayload = { + objectRecordEventsWithQueryIds: [], + metadataEvents: metadataEventsForStream, + }; + await this.subscriptionService.publishToEventStream({ workspaceId, eventStreamChannelId: streamChannelId,