houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove subscriptions job (#14250)

Browse Source 
Remove subscriptions publish job

---------

Co-authored-by: Charles Bochet <charlesBochet@users.noreply.github.com>
This commit is contained in:
martmull
2025-09-02 14:40:30 +02:00
committed by GitHub
parent c613bdc238
commit 1801b5086a
10 changed files with 67 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-front/src/generated-metadata/graphql.ts
+1
View File
@@ -970,6 +970,7 @@ export enum FeatureFlagKey {
IS_API_KEY_ROLES_ENABLED = 'IS_API_KEY_ROLES_ENABLED',
IS_CORE_VIEW_ENABLED = 'IS_CORE_VIEW_ENABLED',
IS_CORE_VIEW_SYNCING_ENABLED = 'IS_CORE_VIEW_SYNCING_ENABLED',
IS_DATABASE_EVENT_TRIGGER_ENABLED = 'IS_DATABASE_EVENT_TRIGGER_ENABLED',
IS_IMAP_SMTP_CALDAV_ENABLED = 'IS_IMAP_SMTP_CALDAV_ENABLED',
IS_JSON_FILTER_ENABLED = 'IS_JSON_FILTER_ENABLED',
IS_MESSAGE_FOLDER_CONTROL_ENABLED = 'IS_MESSAGE_FOLDER_CONTROL_ENABLED',
packages/twenty-front/src/generated/graphql.ts
+1
View File
@@ -934,6 +934,7 @@ export enum FeatureFlagKey {
IS_API_KEY_ROLES_ENABLED = 'IS_API_KEY_ROLES_ENABLED',
IS_CORE_VIEW_ENABLED = 'IS_CORE_VIEW_ENABLED',
IS_CORE_VIEW_SYNCING_ENABLED = 'IS_CORE_VIEW_SYNCING_ENABLED',
IS_DATABASE_EVENT_TRIGGER_ENABLED = 'IS_DATABASE_EVENT_TRIGGER_ENABLED',
IS_IMAP_SMTP_CALDAV_ENABLED = 'IS_IMAP_SMTP_CALDAV_ENABLED',
IS_JSON_FILTER_ENABLED = 'IS_JSON_FILTER_ENABLED',
IS_MESSAGE_FOLDER_CONTROL_ENABLED = 'IS_MESSAGE_FOLDER_CONTROL_ENABLED',
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts
+49 -36
View File
@@ -13,12 +13,14 @@ import { type ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emit
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { SubscriptionsJob } from 'src/engine/subscriptions/subscriptions.job';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { UpsertTimelineActivityFromInternalEvent } from 'src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job';
import { CallWebhookJobsJob } from 'src/engine/core-modules/webhook/jobs/call-webhook-jobs.job';
import { type ObjectRecordEventForWebhook } from 'src/engine/core-modules/webhook/types/object-record-event-for-webhook.type';
import { CallDatabaseEventTriggerJobsJob } from 'src/engine/metadata-modules/trigger/jobs/call-database-event-trigger-jobs.job';
import { SubscriptionsService } from 'src/engine/subscriptions/subscriptions.service';
import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum';
@Injectable()
export class EntityEventsToDbListener {
@@ -27,10 +29,10 @@ export class EntityEventsToDbListener {
private readonly entityEventsToDbQueueService: MessageQueueService,
@InjectMessageQueue(MessageQueue.webhookQueue)
private readonly webhookQueueService: MessageQueueService,
@InjectMessageQueue(MessageQueue.subscriptionsQueue)
private readonly subscriptionsQueueService: MessageQueueService,
@InjectMessageQueue(MessageQueue.triggerQueue)
private readonly triggerQueueService: MessageQueueService,
private readonly subscriptionsService: SubscriptionsService,
private readonly featureFlagService: FeatureFlagService,
) {}
@OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED)
@@ -79,17 +81,14 @@ export class EntityEventsToDbListener {
},
}));
await Promise.all([
this.subscriptionsQueueService.add<WorkspaceEventBatch<T>>(
SubscriptionsJob.name,
batchEvent,
{ retryLimit: 3 },
),
this.triggerQueueService.add<WorkspaceEventBatch<T>>(
CallDatabaseEventTriggerJobsJob.name,
batchEvent,
{ retryLimit: 3 },
),
const isDatabaseEventTriggerEnabled =
await this.featureFlagService.isFeatureEnabled(
FeatureFlagKey.IS_DATABASE_EVENT_TRIGGER_ENABLED,
batchEvent.workspaceId,
);
const promises = [
this.subscriptionsService.publish(batchEvent),
this.webhookQueueService.add<
WorkspaceEventBatch<ObjectRecordEventForWebhook>
>(
@@ -99,27 +98,41 @@ export class EntityEventsToDbListener {
retryLimit: 3,
},
),
...(auditLogsEvents.length > 0
? [
this.entityEventsToDbQueueService.add<WorkspaceEventBatch<T>>(
CreateAuditLogFromInternalEvent.name,
{
...batchEvent,
events: auditLogsEvents,
},
),
]
: []),
...(action !== DatabaseEventAction.DESTROYED && auditLogsEvents.length > 0
? [
this.entityEventsToDbQueueService.add<
WorkspaceEventBatch<ObjectRecordNonDestructiveEvent>
>(UpsertTimelineActivityFromInternalEvent.name, {
...batchEvent,
events: auditLogsEvents,
}),
]
: []),
]);
];
if (isDatabaseEventTriggerEnabled) {
promises.push(
this.triggerQueueService.add<WorkspaceEventBatch<T>>(
CallDatabaseEventTriggerJobsJob.name,
batchEvent,
{ retryLimit: 3 },
),
);
}
if (auditLogsEvents.length > 0) {
promises.push(
this.entityEventsToDbQueueService.add<WorkspaceEventBatch<T>>(
CreateAuditLogFromInternalEvent.name,
{
...batchEvent,
events: auditLogsEvents,
},
),
);
if (action !== DatabaseEventAction.DESTROYED) {
promises.push(
this.entityEventsToDbQueueService.add<
WorkspaceEventBatch<ObjectRecordNonDestructiveEvent>
>(UpsertTimelineActivityFromInternalEvent.name, {
...batchEvent,
events: auditLogsEvents,
}),
);
}
}
await Promise.all(promises);
}
}
packages/twenty-server/src/engine/api/graphql/workspace-query-runner/workspace-query-runner.module.ts
+2
View File
@@ -14,6 +14,7 @@ import { RecordPositionModule } from 'src/engine/core-modules/record-position/re
import { RecordTransformerModule } from 'src/engine/core-modules/record-transformer/record-transformer.module';
import { TelemetryModule } from 'src/engine/core-modules/telemetry/telemetry.module';
import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
import { SubscriptionsModule } from 'src/engine/subscriptions/subscriptions.module';
import { EntityEventsToDbListener } from './listeners/entity-events-to-db.listener';
@@ -30,6 +31,7 @@ import { EntityEventsToDbListener } from './listeners/entity-events-to-db.listen
FeatureFlagModule,
RecordTransformerModule,
RecordPositionModule,
SubscriptionsModule,
],
providers: [
...workspaceQueryRunnerFactories,
packages/twenty-server/src/engine/core-modules/feature-flag/enums/feature-flag-key.enum.ts
+1
View File
@@ -16,4 +16,5 @@ export enum FeatureFlagKey {
IS_PAGE_LAYOUT_ENABLED = 'IS_PAGE_LAYOUT_ENABLED',
IS_MESSAGE_FOLDER_CONTROL_ENABLED = 'IS_MESSAGE_FOLDER_CONTROL_ENABLED',
IS_WORKFLOW_ITERATOR_ENABLED = 'IS_WORKFLOW_ITERATOR_ENABLED',
IS_DATABASE_EVENT_TRIGGER_ENABLED = 'IS_DATABASE_EVENT_TRIGGER_ENABLED',
}
packages/twenty-server/src/engine/core-modules/message-queue/message-queue.constants.ts
-1
View File
@@ -15,7 +15,6 @@ export enum MessageQueue {
entityEventsToDbQueue = 'entity-events-to-db-queue',
workflowQueue = 'workflow-queue',
deleteCascadeQueue = 'delete-cascade-queue',
subscriptionsQueue = 'subscriptions-queue',
serverlessFunctionQueue = 'serverless-function-queue',
triggerQueue = 'trigger-queue',
}
packages/twenty-server/src/engine/subscriptions/subscriptions.module.ts
+3 -3
View File
@@ -4,10 +4,9 @@ import { RedisPubSub } from 'graphql-redis-subscriptions';
import { RedisClientService } from 'src/engine/core-modules/redis-client/redis-client.service';
import { SubscriptionsResolver } from 'src/engine/subscriptions/subscriptions.resolver';
import { SubscriptionsJob } from 'src/engine/subscriptions/subscriptions.job';
import { SubscriptionsService } from 'src/engine/subscriptions/subscriptions.service';
@Module({
exports: ['PUB_SUB'],
providers: [
{
provide: 'PUB_SUB',
@@ -20,8 +19,9 @@ import { SubscriptionsJob } from 'src/engine/subscriptions/subscriptions.job';
}),
},
SubscriptionsResolver,
SubscriptionsJob,
SubscriptionsService,
],
exports: ['PUB_SUB', SubscriptionsService],
})
export class SubscriptionsModule implements OnModuleDestroy {
constructor(@Inject('PUB_SUB') private readonly pubSub: RedisPubSub) {}
packages/twenty-server/src/engine/subscriptions/subscriptions.job.ts → packages/twenty-server/src/engine/subscriptions/subscriptions.service.ts
+4 -8
View File
@@ -1,21 +1,17 @@
import { Inject } from '@nestjs/common';
import { Inject, Injectable } from '@nestjs/common';
import { RedisPubSub } from 'graphql-redis-subscriptions';
import { isDefined } from 'twenty-shared/utils';
import { type ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
import { removeSecretFromWebhookRecord } from 'src/utils/remove-secret-from-webhook-record';
@Processor(MessageQueue.subscriptionsQueue)
export class SubscriptionsJob {
@Injectable()
export class SubscriptionsService {
constructor(@Inject('PUB_SUB') private readonly pubSub: RedisPubSub) {}
@Process(SubscriptionsJob.name)
async handle(
async publish(
workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>,
): Promise<void> {
for (const eventData of workspaceEventBatch.events) {
packages/twenty-server/src/engine/twenty-orm/entity-manager/workspace-entity-manager.spec.ts
+1
View File
@@ -139,6 +139,7 @@ describe('WorkspaceEntityManager', () => {
IS_PAGE_LAYOUT_ENABLED: false,
IS_MESSAGE_FOLDER_CONTROL_ENABLED: false,
IS_WORKFLOW_ITERATOR_ENABLED: false,
IS_DATABASE_EVENT_TRIGGER_ENABLED: false,
},
eventEmitterService: {
emitMutationEvent: jest.fn(),
packages/twenty-server/src/engine/workspace-manager/dev-seeder/core/utils/seed-feature-flags.util.ts
+5
View File
@@ -85,6 +85,11 @@ export const seedFeatureFlags = async (
workspaceId: workspaceId,
value: false,
},
{
key: FeatureFlagKey.IS_DATABASE_EVENT_TRIGGER_ENABLED,
workspaceId: workspaceId,
value: false,
},
])
.execute();
};