Move query matching before publication by thomtrp · Pull Request #17121 · twentyhq/twenty

thomtrp · 2026-01-13T13:38:22Z

new channel EVENT_STREAM_CHANNEL based on event stream id
on event, perform the matching and publish only to the right streams
store a list of active streams per workspace
store the user id along with the queries for each stream

Bonus:

remove onSubscriptionMatch

greptile-apps

Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.

packages/twenty-server/src/engine/subscriptions/event-stream.service.ts

cubic-dev-ai

3 issues found across 21 files

Prompt for AI agents (all issues)


Check if these issues are valid — if so, understand the root cause of each and fix them.


<file name="packages/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup.ts">

<violation number="1" location="packages/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup.ts:7">
P2: If `onClose()` throws an error, `iterator.return()` will never be called, potentially causing a resource leak. Consider wrapping `onClose()` in a try-finally block to ensure the underlying iterator is always closed.</violation>
</file>

<file name="packages/twenty-server/src/engine/subscriptions/event-stream.service.ts">

<violation number="1" location="packages/twenty-server/src/engine/subscriptions/event-stream.service.ts:74">
P1: This method will always return an empty array in Redis environments. The data is stored using `setAdd()` (which uses Redis SADD command for SET data type), but retrieved using `get()` (which uses GET for STRING data type). These are incompatible Redis data types. Consider adding a `setMembers` method to `CacheStorageService` that uses `SMEMBERS` command, or use a different storage approach.</violation>
</file>

<file name="packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts">

<violation number="1" location="packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts:101">
P2: Resource leak: if `subscribeToEventStream` throws after `createEventStream` succeeds, the event stream will never be cleaned up. Wrap the subscription in a try-catch to ensure cleanup on failure.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}

packages/twenty-server/src/engine/subscriptions/event-stream.service.ts

...s/twenty-server/src/engine/workspace-event-emitter/utils/wrap-async-iterator-with-cleanup.ts

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts

github-actions · 2026-01-13T13:49:42Z

🚀 Preview Environment Ready!

Your preview environment is available at: http://bore.pub:23159

This environment will automatically shut down when the PR is closed or after 5 hours.

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts

+    for (const event of workspaceEventBatch.events) {
+      const eventWithObjectName = {
+        objectNameSingular,
+        ...event,
+      };


github-actions · 2026-01-13T13:53:35Z

📊 API Changes Report

GraphQL Schema Changes

[log]
Detected the following changes (4) between schemas:

[log] ✖ Field onSubscriptionMatch was removed from object type Subscription
[log] ✖ Type SubscriptionInput was removed
[log] ✖ Type SubscriptionMatch was removed
[log] ✖ Type SubscriptionMatches was removed
[error] Detected 4 breaking changes
⚠️ Breaking changes or errors detected in GraphQL schema

[log] 
Detected the following changes (4) between schemas:

[log] ✖  Field onSubscriptionMatch was removed from object type Subscription
[log] ✖  Type SubscriptionInput was removed
[log] ✖  Type SubscriptionMatch was removed
[log] ✖  Type SubscriptionMatches was removed
[error] Detected 4 breaking changes
Error generating diff

GraphQL Metadata Schema Changes

[log]
Detected the following changes (4) between schemas:

[log] ✖ Field onSubscriptionMatch was removed from object type Subscription
[log] ✖ Type SubscriptionInput was removed
[log] ✖ Type SubscriptionMatch was removed
[log] ✖ Type SubscriptionMatches was removed
[error] Detected 4 breaking changes
⚠️ Breaking changes or errors detected in GraphQL metadata schema

[log] 
Detected the following changes (4) between schemas:

[log] ✖  Field onSubscriptionMatch was removed from object type Subscription
[log] ✖  Type SubscriptionInput was removed
[log] ✖  Type SubscriptionMatch was removed
[log] ✖  Type SubscriptionMatches was removed
[error] Detected 4 breaking changes
Error generating diff

⚠️ Please review these API changes carefully before merging.

⚠️ Breaking Change Protocol

Breaking changes detected but PR title does not contain "breaking" - CI will pass but action needed.

🔄 Options:

If this IS a breaking change: Add "breaking" to your PR title and add BREAKING CHANGE: to your commit message
If this is NOT a breaking change: The API diff tool may have false positives - please review carefully

For breaking changes, add to commit message:

feat: add new API endpoint

BREAKING CHANGE: removed deprecated field from User schema

packages/twenty-server/src/engine/subscriptions/event-stream.service.ts

+    await this.cacheLockService.withLock(async () => {
+      await this.cacheStorageService.setAdd(activeStreamsKey, [
+        eventStreamChannelId,


charlesBochet · 2026-01-13T17:47:53Z

Leaving it to you @thomtrp as tests are red :)

Weiko

This is much cleaner imho!

Note: activeStreams does not have a TTL, if destroy is never called (connection drop, tab crash, etc) this would stay in the cache forever, see my comments about this issue.

Otherwise LGTM! 👏 @thomtrp

Weiko · 2026-01-14T09:43:45Z

packages/twenty-server/src/engine/subscriptions/event-stream.service.ts

+  }): Promise<void> {
+    const key = this.getEventStreamKey(workspaceId, eventStreamChannelId);
+    const streamData: EventStreamData = {
+      userId,


What about api keys / agents mode (that are not running behind a user). Should we use an authContext here instead?

Weiko · 2026-01-14T09:58:29Z

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts

+  ): Promise<void> {
+    const workspaceId = workspaceEventBatch.workspaceId;
+
+    const activeStreamIds =


This might not get cleared and have stale data due to missing TTL

Weiko · 2026-01-14T10:00:47Z

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts

+      return;
+    }
+
+    const streamsData = await this.eventStreamService.getStreamsData(


This however would be cleared, meaning you might iterate below hover dead steam data with your

if (!isDefined(streamData)) { continue; }

for no reason and this would add up.

Weiko · 2026-01-14T10:01:18Z

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts

+
+    for (const [streamChannelId, streamData] of streamsData) {
+      if (!isDefined(streamData)) {
+        continue;


instead of continue, you could also add some "pruning" logic here for the issue I've mentioned above

sentry · 2026-01-14T10:37:59Z

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts

+      iterator = await this.subscriptionService.subscribeToEventStream({
+        workspaceId: workspace.id,
+        eventStreamChannelId,
+      });


Bug: A client disconnect before a GraphQL subscription starts iterating can cause the cleanup logic in wrapAsyncIteratorWithCleanup to be skipped, orphaning the event stream in Redis.
_{Severity: HIGH}

Suggested Fix

The cleanup mechanism relies on the GraphQL async iterator's return() method, which is not guaranteed to be called if the client disconnects before iteration begins. Implement a more robust cleanup mechanism, for example by using a try...finally block around the subscription processing logic or handling the disconnect event on the server to trigger the destroyEventStream call explicitly, ensuring resources are always released.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's not valid. Location: packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.resolver.ts#L110-L113 Potential issue: The cleanup logic for the `onEventSubscription` resolver, which calls `destroyEventStream`, is wrapped in `wrapAsyncIteratorWithCleanup`. This cleanup relies on the async iterator's `return()` method. If a client disconnects after the resolver returns the iterator but before the GraphQL engine begins iterating over it, the `return()` method is never called. This orphans the event stream data and its ID in Redis until the 30-minute TTL expires. Consequently, this leads to a memory leak in Redis and causes the server to waste CPU cycles attempting to publish events to disconnected clients.

_{Did we get this right? 👍 / 👎 to inform future reviews.}

twenty-eng-sync · 2026-01-14T12:58:37Z

Hey @thomtrp! After you've done the QA of your Pull Request, you can mark it as done here. Thank you!

twenty-eng-sync · 2026-01-14T12:58:39Z

Hey @thomtrp! After you've done the QA of your Pull Request, you can mark it as done here. Thank you!

- new channel EVENT_STREAM_CHANNEL based on event stream id - on event, perform the matching and publish only to the right streams - store a list of active streams per workspace - store the user id along with the queries for each stream Bonus: - remove onSubscriptionMatch

greptile-apps bot reviewed Jan 13, 2026

View reviewed changes

sentry bot reviewed Jan 13, 2026

View reviewed changes

packages/twenty-server/src/engine/subscriptions/event-stream.service.ts Show resolved Hide resolved

cubic-dev-ai bot reviewed Jan 13, 2026

View reviewed changes

thomtrp force-pushed the tt-add-permission-checks-on-streams branch from ff93b25 to ce4c6db Compare January 13, 2026 13:45

sentry bot reviewed Jan 13, 2026

View reviewed changes

packages/twenty-server/src/engine/workspace-event-emitter/workspace-event-emitter.service.ts

Comment on lines +92 to +96

for (const event of workspaceEventBatch.events) {

const eventWithObjectName = {

objectNameSingular,

...event,

};

This comment was marked as outdated.

Sign in to view

thomtrp force-pushed the tt-add-permission-checks-on-streams branch from ce4c6db to 4dd30ee Compare January 13, 2026 13:55

sentry bot reviewed Jan 13, 2026

View reviewed changes

packages/twenty-server/src/engine/subscriptions/event-stream.service.ts Outdated

Comment on lines +47 to +49

await this.cacheLockService.withLock(async () => {

await this.cacheStorageService.setAdd(activeStreamsKey, [

eventStreamChannelId,

This comment was marked as outdated.

Sign in to view

charlesBochet requested a review from Weiko January 13, 2026 17:47

charlesBochet assigned thomtrp Jan 13, 2026

charlesBochet added -PR: awaiting review -PR: awaiting author and removed -PR: awaiting review labels Jan 13, 2026

thomtrp added -PR: awaiting review and removed -PR: awaiting author labels Jan 14, 2026

Weiko approved these changes Jan 14, 2026

View reviewed changes

thomtrp added 3 commits January 14, 2026 11:07

Move query matching before publication

5e18083

Regen schema

1222202

Store authcontext + clean active streams

6b7444c

thomtrp force-pushed the tt-add-permission-checks-on-streams branch from 0e63cd9 to 6b7444c Compare January 14, 2026 10:32

sentry bot reviewed Jan 14, 2026

View reviewed changes

thomtrp added this pull request to the merge queue Jan 14, 2026

Merged via the queue into main with commit ec87b29 Jan 14, 2026
114 of 122 checks passed

thomtrp deleted the tt-add-permission-checks-on-streams branch January 14, 2026 12:58

Conversation

thomtrp commented Jan 13, 2026

Uh oh!

greptile-apps bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

cubic-dev-ai bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Jan 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

This comment was marked as outdated.

Uh oh!

github-actions bot commented Jan 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

📊 API Changes Report

GraphQL Schema Changes

GraphQL Schema Changes

GraphQL Metadata Schema Changes

GraphQL Metadata Schema Changes

⚠️ Breaking Change Protocol

Uh oh!

This comment was marked as outdated.

Uh oh!

charlesBochet commented Jan 13, 2026

Uh oh!

Weiko left a comment

Choose a reason for hiding this comment

Uh oh!

Weiko Jan 14, 2026

Choose a reason for hiding this comment

Uh oh!

Weiko Jan 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Weiko Jan 14, 2026

Choose a reason for hiding this comment

Uh oh!

Weiko Jan 14, 2026

Choose a reason for hiding this comment

Uh oh!

sentry bot Jan 14, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

twenty-eng-sync bot commented Jan 14, 2026

Uh oh!

twenty-eng-sync bot commented Jan 14, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

github-actions bot commented Jan 13, 2026 •

edited

Loading

github-actions bot commented Jan 13, 2026 •

edited

Loading

Weiko Jan 14, 2026 •

edited

Loading