From 775a07fb6fe63e7676b99d0e3047a342d1c3e272 Mon Sep 17 00:00:00 2001 From: Himanshu Garg Date: Thu, 20 Aug 2026 19:25:34 +0530 Subject: [PATCH 1/7] feat(api-service, worker): aws eventbridge scheduler for delay more than 900 sec (#12362) --- .../snooze-notification.spec.ts | 55 +++ .../snooze-notification.usecase.ts | 13 +- .../unsnooze-notification.spec.ts | 45 ++- .../unsnooze-notification.usecase.ts | 27 +- apps/api/src/config/env.validators.ts | 7 + .../workflow/services/standard.worker.spec.ts | 10 +- .../usecases/add-job/add-job.usecase.ts | 25 ++ apps/worker/src/config/env.validators.ts | 7 + libs/application-generic/jest.config.js | 4 +- libs/application-generic/package.json | 2 + .../src/modules/queues.module.ts | 15 +- .../application-generic/src/services/index.ts | 1 + .../queues/queue-base.service.spec.ts | 244 ++++++++++++++ .../src/services/queues/queue-base.service.ts | 248 ++++++++++---- .../queues/standard-queue.service.spec.ts | 13 +- .../services/queues/standard-queue.service.ts | 8 +- .../readiness/readiness.service.spec.ts | 10 +- .../event-bridge-scheduler.service.spec.ts | 264 +++++++++++++++ .../event-bridge-scheduler.service.ts | 270 +++++++++++++++ .../src/services/scheduler/index.ts | 2 + .../src/services/scheduler/types.ts | 53 +++ packages/shared/src/types/feature-flags.ts | 8 + pnpm-lock.yaml | 317 +++++++++++++++++- 23 files changed, 1564 insertions(+), 84 deletions(-) create mode 100644 libs/application-generic/src/services/queues/queue-base.service.spec.ts create mode 100644 libs/application-generic/src/services/scheduler/event-bridge-scheduler.service.spec.ts create mode 100644 libs/application-generic/src/services/scheduler/event-bridge-scheduler.service.ts create mode 100644 libs/application-generic/src/services/scheduler/index.ts create mode 100644 libs/application-generic/src/services/scheduler/types.ts diff --git a/apps/api/src/app/inbox/usecases/snooze-notification/snooze-notification.spec.ts b/apps/api/src/app/inbox/usecases/snooze-notification/snooze-notification.spec.ts index 2d7d9a5d3ba..eb1bb54cced 100644 --- a/apps/api/src/app/inbox/usecases/snooze-notification/snooze-notification.spec.ts +++ b/apps/api/src/app/inbox/usecases/snooze-notification/snooze-notification.spec.ts @@ -223,6 +223,61 @@ describe('SnoozeNotification', () => { expect(createExecutionDetailsMock.execute.called).to.be.true; }); + it('should enqueue the unsnooze job only after the transaction has closed', async () => { + const command = createCommand(SNOOZE_DURATION.ONE_DAY); + const sequence: string[] = []; + let transactionDepth = 0; + let transactionDepthAtEnqueue = -1; + + // Mirrors session.withTransaction: the session stays open for the whole callback. + // @ts-expect-error Mocking the withTransaction method + messageRepositoryMock.withTransaction = sinon.stub().callsFake(async (callback) => { + transactionDepth += 1; + sequence.push('transaction:begin'); + try { + return await callback(); + } finally { + sequence.push('transaction:end'); + transactionDepth -= 1; + } + }); + + jobRepositoryMock.create.callsFake(async () => { + sequence.push('job:create'); + + return mockJob; + }); + + markNotificationAsMock.execute.callsFake(async () => { + sequence.push('notification:snoozed'); + + return mockNotification; + }); + + standardQueueServiceMock.add.callsFake(async () => { + transactionDepthAtEnqueue = transactionDepth; + sequence.push('queue:add'); + }); + + await snoozeNotification.execute(command); + + /* + * Enqueueing is an external call - SQS, or a CreateSchedule round trip to + * EventBridge Scheduler for any snooze past the 900s delay cap. Doing it + * inside the transaction pins a Mongo connection and its locks for the + * length of that call, and an abort afterwards strands the schedule. + */ + expect(standardQueueServiceMock.add.calledOnce).to.be.true; + expect(transactionDepthAtEnqueue).to.equal(0); + expect(sequence).to.deep.equal([ + 'transaction:begin', + 'job:create', + 'notification:snoozed', + 'transaction:end', + 'queue:add', + ]); + }); + it('should enqueue job with correct parameters', async () => { const delay = 3600000; // 1 hour in milliseconds diff --git a/apps/api/src/app/inbox/usecases/snooze-notification/snooze-notification.usecase.ts b/apps/api/src/app/inbox/usecases/snooze-notification/snooze-notification.usecase.ts index b8dff03f7d2..ae28d99d851 100644 --- a/apps/api/src/app/inbox/usecases/snooze-notification/snooze-notification.usecase.ts +++ b/apps/api/src/app/inbox/usecases/snooze-notification/snooze-notification.usecase.ts @@ -10,6 +10,7 @@ import { AnalyticsService, CreateExecutionDetails, CreateExecutionDetailsCommand, + DeferReasonEnum, DetailEnum, getEffectiveJobPayload, PinoLogger, @@ -72,9 +73,18 @@ export class SnoozeNotification { await this.messageRepository.withTransaction(async () => { scheduledJob = await this.createScheduledUnsnoozeJob(notification, snoozeDurationMs); snoozedNotification = await this.markNotificationAsSnoozed(command); - await this.enqueueJob(scheduledJob, snoozeDurationMs); }); + /* + * Enqueueing has to stay outside the transaction: it is an external call, + * and once the snooze outlives the 900s SQS delay cap - which any snooze + * measured in hours does - it becomes a CreateSchedule round trip to + * EventBridge. Inside the transaction that held the Mongo session, and its + * locks, open for the length of an AWS call, and an abort after the call + * had succeeded would leave a schedule behind with no job left to wake. + */ + await this.enqueueJob(scheduledJob, snoozeDurationMs); + // fire and forget this.createExecutionDetails .execute( @@ -116,6 +126,7 @@ export class SnoozeNotification { }, groupId: job._organizationId, options: { delay, attempts: this.RETRY_ATTEMPTS, backoff: { type: 'exponential', delay: 5000 } }, + deferReason: DeferReasonEnum.SNOOZE, }); } diff --git a/apps/api/src/app/inbox/usecases/unsnooze-notification/unsnooze-notification.spec.ts b/apps/api/src/app/inbox/usecases/unsnooze-notification/unsnooze-notification.spec.ts index 12def471ed5..b2d8f89f546 100644 --- a/apps/api/src/app/inbox/usecases/unsnooze-notification/unsnooze-notification.spec.ts +++ b/apps/api/src/app/inbox/usecases/unsnooze-notification/unsnooze-notification.spec.ts @@ -1,5 +1,11 @@ import { NotFoundException } from '@nestjs/common'; -import { CreateExecutionDetails, CreateExecutionDetailsCommand, PinoLogger } from '@novu/application-generic'; +import { + CreateExecutionDetails, + CreateExecutionDetailsCommand, + DeferReasonEnum, + EventBridgeSchedulerService, + PinoLogger, +} from '@novu/application-generic'; import { JobEntity, JobRepository, MessageEntity, MessageRepository } from '@novu/dal'; import { ChannelTypeEnum, JobStatusEnum, SeverityLevelEnum } from '@novu/shared'; import { expect } from 'chai'; @@ -26,6 +32,7 @@ describe('UnsnoozeNotification', () => { let createExecutionDetailsMock: sinon.SinonStubbedInstance; let markNotificationAsMock: sinon.SinonStubbedInstance; let getSubscriberMock: sinon.SinonStubbedInstance; + let schedulerServiceMock: sinon.SinonStubbedInstance; const snoozedUntil = new Date(); snoozedUntil.setHours(snoozedUntil.getHours() + 1); @@ -80,6 +87,8 @@ describe('UnsnoozeNotification', () => { createExecutionDetailsMock = sinon.createStubInstance(CreateExecutionDetails); markNotificationAsMock = sinon.createStubInstance(MarkNotificationAs); getSubscriberMock = sinon.createStubInstance(GetSubscriber); + schedulerServiceMock = sinon.createStubInstance(EventBridgeSchedulerService); + schedulerServiceMock.deleteSchedule.resolves(); sinon.stub(MarkNotificationAsCommand, 'create').returns({ environmentId: validEnvId, @@ -101,7 +110,8 @@ describe('UnsnoozeNotification', () => { jobRepositoryMock as any, markNotificationAsMock as any, createExecutionDetailsMock as any, - getSubscriberMock as any + getSubscriberMock as any, + schedulerServiceMock as any ); jobRepositoryMock.findOneAndDelete.resolves(mockJob); @@ -150,6 +160,37 @@ describe('UnsnoozeNotification', () => { expect(createExecutionDetailsMock.execute.calledOnce).to.be.true; }); + it('should delete the snooze schedule so a stale fire cannot churn on SQS', async () => { + const command = createCommand(); + + await unsnoozeNotification.execute(command); + + expect(schedulerServiceMock.deleteSchedule.calledOnce).to.be.true; + expect(schedulerServiceMock.deleteSchedule.firstCall.args[0]).to.deep.equal({ + deferReason: DeferReasonEnum.SNOOZE, + organizationId: validOrgId, + scheduleId: validJobId, + }); + }); + + it('should still unsnooze when deleting the schedule fails', async () => { + const command = createCommand(); + schedulerServiceMock.deleteSchedule.rejects(new Error('AccessDeniedException')); + + const result = await unsnoozeNotification.execute(command); + + expect(result).to.deep.equal(mockNotification); + }); + + it('should not attempt a schedule delete when there was no scheduled job', async () => { + const command = createCommand(); + jobRepositoryMock.findOneAndDelete.resolves(null); + + await unsnoozeNotification.execute(command); + + expect(schedulerServiceMock.deleteSchedule.called).to.be.false; + }); + it('should handle missing scheduled job gracefully', async () => { const command = createCommand(); jobRepositoryMock.findOneAndDelete.resolves(null); diff --git a/apps/api/src/app/inbox/usecases/unsnooze-notification/unsnooze-notification.usecase.ts b/apps/api/src/app/inbox/usecases/unsnooze-notification/unsnooze-notification.usecase.ts index 37ce9088d9a..dd58a1d2cc3 100644 --- a/apps/api/src/app/inbox/usecases/unsnooze-notification/unsnooze-notification.usecase.ts +++ b/apps/api/src/app/inbox/usecases/unsnooze-notification/unsnooze-notification.usecase.ts @@ -2,7 +2,9 @@ import { BadRequestException, Injectable, InternalServerErrorException, NotFound import { CreateExecutionDetails, CreateExecutionDetailsCommand, + DeferReasonEnum, DetailEnum, + EventBridgeSchedulerService, PinoLogger, } from '@novu/application-generic'; import { ChannelTypeEnum, JobEntity, JobRepository, JobStatusEnum, MessageRepository } from '@novu/dal'; @@ -21,7 +23,8 @@ export class UnsnoozeNotification { private jobRepository: JobRepository, private markNotificationAs: MarkNotificationAs, private createExecutionDetails: CreateExecutionDetails, - private getSubscriber: GetSubscriber + private getSubscriber: GetSubscriber, + private schedulerService: EventBridgeSchedulerService ) { this.logger.setContext(this.constructor.name); } @@ -90,6 +93,8 @@ export class UnsnoozeNotification { }); if (scheduledJob) { + this.deleteSnoozeSchedule(scheduledJob); + // fire and forget this.createExecutionDetails .execute( @@ -114,4 +119,24 @@ export class UnsnoozeNotification { return unsnoozedNotification; } + + /** + * Snooze is the one defer reason whose schedule is worth removing: the job + * document has just been deleted, so a later fire would find nothing and + * churn through SQS redeliveries until the redrive policy gives up. Every + * other reason relies on the fire happening and `RunJob` deciding it is a + * no-op. Best effort by design - the unsnooze has already been committed and + * a leftover schedule is only noise, never a correctness problem. + */ + private deleteSnoozeSchedule(job: JobEntity): void { + this.schedulerService + .deleteSchedule({ + deferReason: DeferReasonEnum.SNOOZE, + organizationId: job._organizationId, + scheduleId: job._id, + }) + .catch((error) => { + this.logger.warn({ err: error, jobId: job._id }, 'Failed to delete the snooze schedule'); + }); + } } diff --git a/apps/api/src/config/env.validators.ts b/apps/api/src/config/env.validators.ts index 948d398850f..15fe6bb1cee 100644 --- a/apps/api/src/config/env.validators.ts +++ b/apps/api/src/config/env.validators.ts @@ -80,6 +80,13 @@ export const envValidators = { SQS_ENDPOINT: str({ default: undefined }), SQS_PAYLOAD_OFFLOAD_BUCKET: str({ default: undefined }), SQS_PAYLOAD_SIZE_THRESHOLD: num({ default: undefined }), + // EventBridge Scheduler for delays beyond the SQS 900s cap (optional - when + // unset, long delays keep going to BullMQ) + EVENTBRIDGE_SCHEDULER_GROUP_PREFIX: str({ default: undefined }), + EVENTBRIDGE_SCHEDULER_ROLE_ARN: str({ default: undefined }), + EVENTBRIDGE_SCHEDULER_DLQ_ARN: str({ default: undefined }), + EVENTBRIDGE_SCHEDULER_MAX_RETRY_ATTEMPTS: num({ default: undefined }), + EVENTBRIDGE_SCHEDULER_MAX_EVENT_AGE_SECONDS: num({ default: undefined }), ENABLE_OTEL: bool({ default: false }), ENABLE_OTEL_LOGS: bool({ default: false }), OTEL_PROMETHEUS_PORT: num({ default: 9464 }), diff --git a/apps/worker/src/app/workflow/services/standard.worker.spec.ts b/apps/worker/src/app/workflow/services/standard.worker.spec.ts index eafbe28310b..ca8a68c8d78 100644 --- a/apps/worker/src/app/workflow/services/standard.worker.spec.ts +++ b/apps/worker/src/app/workflow/services/standard.worker.spec.ts @@ -1,6 +1,7 @@ import { faker } from '@faker-js/faker'; import { Test } from '@nestjs/testing'; import { + EventBridgeSchedulerService, FeatureFlagsService, JobsOptions, PinoLogger, @@ -70,6 +71,12 @@ const mockLogger = { error: () => {}, } as unknown as PinoLogger; +const mockSchedulerService = { + isConfigured: () => false, + createDelayedFire: async () => {}, + deleteSchedule: async () => {}, +} as unknown as EventBridgeSchedulerService; + describe('Standard Worker', () => { let jobRepository: JobRepository; let notificationRepository: NotificationRepository; @@ -131,7 +138,8 @@ describe('Standard Worker', () => { mockSqsService, mockFeatureFlagsService, mockOrganizationRepository, - mockLogger + mockLogger, + mockSchedulerService ); await standardQueueService.queue.obliterate(); }); diff --git a/apps/worker/src/app/workflow/usecases/add-job/add-job.usecase.ts b/apps/worker/src/app/workflow/usecases/add-job/add-job.usecase.ts index a91b9b8539d..26cf822ab13 100644 --- a/apps/worker/src/app/workflow/usecases/add-job/add-job.usecase.ts +++ b/apps/worker/src/app/workflow/usecases/add-job/add-job.usecase.ts @@ -5,6 +5,7 @@ import { ConditionsFilterCommand, CreateExecutionDetails, CreateExecutionDetailsCommand, + DeferReasonEnum, DetailEnum, DurationUtils, getDigestType, @@ -70,6 +71,17 @@ export enum BackoffStrategiesEnum { WEBHOOK_FILTER_BACKOFF = 'webhookFilterBackoff', } +/** + * Groups the job's EventBridge schedule when the delay outlives the SQS cap. + * Only the deferring step types appear here; anything else that reaches + * queueJob is either immediate or a schedule extension, both handled below. + */ +const DEFER_REASON_BY_STEP_TYPE: Partial> = { + [StepTypeEnum.DELAY]: DeferReasonEnum.DELAY, + [StepTypeEnum.DIGEST]: DeferReasonEnum.DIGEST, + [StepTypeEnum.THROTTLE]: DeferReasonEnum.THROTTLE, +}; + /* * @description: This is the result of the add job usecase * @@ -1114,6 +1126,7 @@ export class AddJob { }, groupId: job._organizationId, options, + deferReason: this.resolveDeferReason(job), }); if (delay) { @@ -1121,6 +1134,18 @@ export class AddJob { } } + /** + * A quiet-hours extension re-queues a channel-typed job, so the step type + * alone cannot tell the two apart - the extension counter can. + */ + private resolveDeferReason(job: JobEntity): DeferReasonEnum { + if (job.scheduleExtensionsCount) { + return DeferReasonEnum.SCHEDULE_EXTENSION; + } + + return (job.type && DEFER_REASON_BY_STEP_TYPE[job.type]) || DeferReasonEnum.DELAY; + } + private async createDelayExecutionDetails(job: JobEntity, delay: number, untilDate: Date | null, timezone?: string) { const logMessage = job.type === StepTypeEnum.DELAY diff --git a/apps/worker/src/config/env.validators.ts b/apps/worker/src/config/env.validators.ts index 838f9557dd5..55a0765b533 100644 --- a/apps/worker/src/config/env.validators.ts +++ b/apps/worker/src/config/env.validators.ts @@ -97,6 +97,13 @@ export const envValidators = { SQS_ENDPOINT: str({ default: undefined }), SQS_PAYLOAD_OFFLOAD_BUCKET: str({ default: undefined }), SQS_PAYLOAD_SIZE_THRESHOLD: num({ default: undefined }), + // EventBridge Scheduler for delays beyond the SQS 900s cap (optional - when + // unset, long delays keep going to BullMQ) + EVENTBRIDGE_SCHEDULER_GROUP_PREFIX: str({ default: undefined }), + EVENTBRIDGE_SCHEDULER_ROLE_ARN: str({ default: undefined }), + EVENTBRIDGE_SCHEDULER_DLQ_ARN: str({ default: undefined }), + EVENTBRIDGE_SCHEDULER_MAX_RETRY_ATTEMPTS: num({ default: undefined }), + EVENTBRIDGE_SCHEDULER_MAX_EVENT_AGE_SECONDS: num({ default: undefined }), SOCKET_WORKER_URL: str({ default: undefined }), INTERNAL_SERVICES_API_KEY: str({ default: undefined }), STEP_RESOLVER_DISPATCH_URL: str({ default: undefined }), diff --git a/libs/application-generic/jest.config.js b/libs/application-generic/jest.config.js index 138be3f37e8..406a400d7d8 100644 --- a/libs/application-generic/jest.config.js +++ b/libs/application-generic/jest.config.js @@ -1,7 +1,9 @@ /** @type {import('ts-jest/dist/types').InitialOptionsTsJest} */ module.exports = { preset: 'ts-jest', - testEnvironment: 'node', + // Pinned explicitly: the bare 'node' alias resolves to the hoisted + // jest-environment-node 30, which jest 27's runner cannot drive. + testEnvironment: require.resolve('jest-environment-node'), moduleNameMapper: { axios: 'axios/dist/node/axios.cjs', }, diff --git a/libs/application-generic/package.json b/libs/application-generic/package.json index e7595f941f6..811e3523379 100644 --- a/libs/application-generic/package.json +++ b/libs/application-generic/package.json @@ -40,6 +40,7 @@ "@anthropic-ai/aws-sdk": "0.3.0", "@anthropic-ai/sdk": "0.95.1", "@aws-sdk/client-s3": "^3.996.0", + "@aws-sdk/client-scheduler": "3.996.0", "@aws-sdk/client-secrets-manager": "3.996.0", "@aws-sdk/client-sqs": "^3.996.0", "@aws-sdk/s3-request-presigner": "^3.996.0", @@ -138,6 +139,7 @@ "cpx": "^1.5.0", "dotenv": "^16.6.1", "jest": "^27.1.0", + "jest-environment-node": "^27.5.1", "npm-run-all": "^4.1.5", "nyc": "^15.1.0", "rimraf": "^3.0.2", diff --git a/libs/application-generic/src/modules/queues.module.ts b/libs/application-generic/src/modules/queues.module.ts index 2b3753c3deb..ea6e4b173dc 100644 --- a/libs/application-generic/src/modules/queues.module.ts +++ b/libs/application-generic/src/modules/queues.module.ts @@ -10,7 +10,13 @@ import { WebSocketsQueueServiceHealthIndicator, WorkflowQueueServiceHealthIndicator, } from '../health'; -import { ReadinessService, SocketWorkerService, SqsService, WorkflowInMemoryProviderService } from '../services'; +import { + EventBridgeSchedulerService, + ReadinessService, + SocketWorkerService, + SqsService, + WorkflowInMemoryProviderService, +} from '../services'; import { ActiveJobsMetricQueueService, InboundParseQueueService, @@ -33,7 +39,12 @@ const memoryQueueService = { }; const INTERNAL_MODULE_PROVIDERS = [memoryQueueService, featureFlagsService]; -const BASE_PROVIDERS: Provider[] = [ReadinessService, CommunityOrganizationRepository, SqsService]; +const BASE_PROVIDERS: Provider[] = [ + ReadinessService, + CommunityOrganizationRepository, + SqsService, + EventBridgeSchedulerService, +]; @Module({ providers: [], diff --git a/libs/application-generic/src/services/index.ts b/libs/application-generic/src/services/index.ts index 4337b2b6567..9eb1dddfa56 100644 --- a/libs/application-generic/src/services/index.ts +++ b/libs/application-generic/src/services/index.ts @@ -47,6 +47,7 @@ export { export * from './safe-outbound-http'; export * from './sanitize/sanitizer.service'; export * from './sanitize/sanitizer-v0.service'; +export * from './scheduler'; export * from './socket-worker'; export * from './sqs'; export { diff --git a/libs/application-generic/src/services/queues/queue-base.service.spec.ts b/libs/application-generic/src/services/queues/queue-base.service.spec.ts new file mode 100644 index 00000000000..d2c564f9f62 --- /dev/null +++ b/libs/application-generic/src/services/queues/queue-base.service.spec.ts @@ -0,0 +1,244 @@ +import { CommunityOrganizationRepository } from '@novu/dal'; +import { ApiServiceLevelEnum, FeatureFlagsKeysEnum, JobTopicNameEnum, QueueBackendMode } from '@novu/shared'; + +import { PinoLogger } from '../../logging'; +import { BullMqService } from '../bull-mq'; +import { FeatureFlagsService } from '../feature-flags'; +import { DeferReasonEnum, EventBridgeSchedulerService } from '../scheduler'; +import { SqsService } from '../sqs'; +import { SQS_MAX_DELAY_SECONDS } from '../sqs/types'; +import { QueueBaseService } from './queue-base.service'; + +const ORGANIZATION_ID = '65f1a2b3c4d5e6f708192a3b'; +const JOB_ID = '65f1a2b3c4d5e6f708192a3c'; +const LONG_DELAY_MS = (SQS_MAX_DELAY_SECONDS + 1) * 1000; +const SHORT_DELAY_MS = 60_000; + +type Harness = { + service: QueueBaseService; + bullMq: { add: jest.Mock; addBulk: jest.Mock }; + sqs: { isConfigured: jest.Mock; send: jest.Mock; sendBulk: jest.Mock }; + scheduler: { isConfigured: jest.Mock; createDelayedFire: jest.Mock; deleteSchedule: jest.Mock }; + getFlag: jest.Mock; +}; + +function buildHarness( + options: { + sqsConfigured?: boolean; + schedulerConfigured?: boolean; + schedulerEnabled?: boolean; + backendMode?: string; + organization?: unknown; + } = {} +): Harness { + const { + sqsConfigured = true, + schedulerConfigured = true, + schedulerEnabled = true, + backendMode = QueueBackendMode.COMPLETE, + organization = { _id: ORGANIZATION_ID, apiServiceLevel: ApiServiceLevelEnum.BUSINESS }, + } = options; + + const bullMq = { add: jest.fn(), addBulk: jest.fn() }; + const sqs = { isConfigured: jest.fn(() => sqsConfigured), send: jest.fn(), sendBulk: jest.fn() }; + const scheduler = { + isConfigured: jest.fn(() => schedulerConfigured), + createDelayedFire: jest.fn(), + deleteSchedule: jest.fn(), + }; + + const getFlag = jest.fn(({ key }: { key: FeatureFlagsKeysEnum }) => { + if (key === FeatureFlagsKeysEnum.IS_EVENTBRIDGE_SCHEDULER_ENABLED) { + return Promise.resolve(schedulerEnabled); + } + + return Promise.resolve(backendMode); + }); + + const service = new QueueBaseService( + JobTopicNameEnum.STANDARD, + bullMq as unknown as BullMqService, + sqs as unknown as SqsService, + { getFlag } as unknown as FeatureFlagsService, + { findOne: jest.fn().mockResolvedValue(organization) } as unknown as CommunityOrganizationRepository, + undefined as unknown as PinoLogger, + scheduler as unknown as EventBridgeSchedulerService + ); + + return { service, bullMq, sqs, scheduler, getFlag }; +} + +function longDelayJob(overrides: Record = {}) { + return { + name: JOB_ID, + data: { _id: JOB_ID, _organizationId: ORGANIZATION_ID }, + groupId: ORGANIZATION_ID, + options: { delay: LONG_DELAY_MS, jobId: JOB_ID }, + deferReason: DeferReasonEnum.DIGEST, + ...overrides, + }; +} + +describe('QueueBaseService long-delay routing', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('should keep long delays on BullMQ when SQS is not configured for the topic', async () => { + const { service, bullMq, sqs, scheduler } = buildHarness({ sqsConfigured: false }); + + await service.add(longDelayJob()); + + expect(bullMq.add).toHaveBeenCalledTimes(1); + expect(scheduler.createDelayedFire).not.toHaveBeenCalled(); + expect(sqs.send).not.toHaveBeenCalled(); + }); + + it('should keep long delays on BullMQ while the scheduler flag is off', async () => { + const { service, bullMq, sqs, scheduler } = buildHarness({ schedulerEnabled: false }); + + await service.add(longDelayJob()); + + expect(bullMq.add).toHaveBeenCalledTimes(1); + expect(scheduler.createDelayedFire).not.toHaveBeenCalled(); + expect(sqs.send).not.toHaveBeenCalled(); + }); + + it('should keep long delays on BullMQ when the scheduler is unconfigured, without evaluating the flag', async () => { + const { service, bullMq, scheduler, getFlag } = buildHarness({ schedulerConfigured: false }); + + await service.add(longDelayJob()); + + expect(bullMq.add).toHaveBeenCalledTimes(1); + expect(scheduler.createDelayedFire).not.toHaveBeenCalled(); + expect( + getFlag.mock.calls.some(([args]) => args.key === FeatureFlagsKeysEnum.IS_EVENTBRIDGE_SCHEDULER_ENABLED) + ).toBe(false); + }); + + it('should create a schedule instead of an SQS message once enabled', async () => { + const { service, bullMq, sqs, scheduler } = buildHarness(); + + await service.add(longDelayJob()); + + expect(scheduler.createDelayedFire).toHaveBeenCalledTimes(1); + expect(sqs.send).not.toHaveBeenCalled(); + expect(bullMq.add).not.toHaveBeenCalled(); + }); + + it('should pass the defer reason, schedule id and fire time to the scheduler', async () => { + const { service, scheduler } = buildHarness(); + const before = Date.now(); + + await service.add(longDelayJob()); + + const [topic, params] = scheduler.createDelayedFire.mock.calls[0]; + expect(topic).toBe(JobTopicNameEnum.STANDARD); + expect(params).toMatchObject({ + deferReason: DeferReasonEnum.DIGEST, + organizationId: ORGANIZATION_ID, + scheduleId: JOB_ID, + messageBody: JSON.stringify({ _id: JOB_ID, _organizationId: ORGANIZATION_ID }), + }); + expect(params.fireAt.getTime()).toBeGreaterThanOrEqual(before + LONG_DELAY_MS); + expect(params.fireAt.getTime()).toBeLessThanOrEqual(Date.now() + LONG_DELAY_MS); + }); + + it('should use the extension-suffixed job id so each extension gets its own schedule', async () => { + const { service, scheduler } = buildHarness(); + + await service.add( + longDelayJob({ + options: { delay: LONG_DELAY_MS, jobId: `${JOB_ID}-ext2` }, + deferReason: DeferReasonEnum.SCHEDULE_EXTENSION, + }) + ); + + expect(scheduler.createDelayedFire.mock.calls[0][1]).toMatchObject({ + scheduleId: `${JOB_ID}-ext2`, + deferReason: DeferReasonEnum.SCHEDULE_EXTENSION, + }); + }); + + it('should default the defer reason when a producer does not set one', async () => { + const { service, scheduler } = buildHarness(); + + await service.add(longDelayJob({ deferReason: undefined })); + + expect(scheduler.createDelayedFire.mock.calls[0][1].deferReason).toBe(DeferReasonEnum.DELAY); + }); + + it('should fall back to BullMQ when the schedule cannot be created', async () => { + const { service, bullMq, scheduler } = buildHarness(); + scheduler.createDelayedFire.mockRejectedValueOnce(new Error('ThrottlingException')); + + await service.add(longDelayJob()); + + expect(bullMq.add).toHaveBeenCalledTimes(1); + }); + + it('should leave short delays on the direct SQS path', async () => { + const { service, sqs, scheduler } = buildHarness(); + + await service.add(longDelayJob({ options: { delay: SHORT_DELAY_MS, jobId: JOB_ID } })); + + expect(sqs.send).toHaveBeenCalledTimes(1); + expect(sqs.send.mock.calls[0][1]).toMatchObject({ delaySeconds: SHORT_DELAY_MS / 1000 }); + expect(scheduler.createDelayedFire).not.toHaveBeenCalled(); + }); + + it('should honour a BULLMQ backend mode even when the scheduler is enabled', async () => { + const { service, bullMq, scheduler } = buildHarness({ backendMode: QueueBackendMode.BULLMQ }); + + await service.add(longDelayJob()); + + expect(bullMq.add).toHaveBeenCalledTimes(1); + expect(scheduler.createDelayedFire).not.toHaveBeenCalled(); + }); + + it('should skip the job entirely when the organization cannot be read', async () => { + const { service, bullMq, sqs, scheduler } = buildHarness({ organization: null }); + + await service.add(longDelayJob()); + + expect(bullMq.add).not.toHaveBeenCalled(); + expect(sqs.send).not.toHaveBeenCalled(); + expect(scheduler.createDelayedFire).not.toHaveBeenCalled(); + }); + + it('should route a job without an organization id to BullMQ', async () => { + const { service, bullMq, scheduler } = buildHarness(); + + await service.add(longDelayJob({ groupId: undefined })); + + expect(bullMq.add).toHaveBeenCalledTimes(1); + expect(scheduler.createDelayedFire).not.toHaveBeenCalled(); + }); + + describe('addBulk', () => { + it('should split a mixed batch between the scheduler and SQS', async () => { + const { service, sqs, scheduler } = buildHarness(); + + await service.addBulk([ + longDelayJob(), + longDelayJob({ options: { delay: SHORT_DELAY_MS, jobId: 'short-1' }, name: 'short-1' }), + ] as never); + + expect(scheduler.createDelayedFire).toHaveBeenCalledTimes(1); + expect(sqs.send).toHaveBeenCalledTimes(1); + }); + + it('should send only the short-delay jobs to SQS when the scheduler is off', async () => { + const { service, bullMq, sqs, scheduler } = buildHarness({ schedulerEnabled: false }); + + await service.addBulk([ + longDelayJob(), + longDelayJob({ options: { delay: SHORT_DELAY_MS, jobId: 'short-1' }, name: 'short-1' }), + ] as never); + + expect(scheduler.createDelayedFire).not.toHaveBeenCalled(); + expect(bullMq.add).toHaveBeenCalledTimes(1); + expect(sqs.send).toHaveBeenCalledTimes(1); + }); + }); +}); diff --git a/libs/application-generic/src/services/queues/queue-base.service.ts b/libs/application-generic/src/services/queues/queue-base.service.ts index 46d9f15fbbf..f560bd7f5e8 100644 --- a/libs/application-generic/src/services/queues/queue-base.service.ts +++ b/libs/application-generic/src/services/queues/queue-base.service.ts @@ -5,11 +5,34 @@ import { PinoLogger } from '../../logging'; import { BulkJobOptions, BullMqService, JobsOptions, Queue, QueueOptions } from '../bull-mq'; import { FeatureFlagsService } from '../feature-flags'; +import { DeferReasonEnum, EventBridgeSchedulerService } from '../scheduler'; import { SqsService } from '../sqs'; import { SQS_MAX_DELAY_SECONDS } from '../sqs/types'; const LOG_CONTEXT = 'QueueService'; +type OrganizationRouting = { _id: string; apiServiceLevel?: ApiServiceLevelEnum }; + +/** The per-organization decisions `add` and `addBulk` share, resolved once per call. */ +interface IQueueRouting { + organization: OrganizationRouting; + backendMode: string; +} + +function exceedsSqsDelayCap(delayMs: number | undefined): boolean { + return (delayMs || 0) > SQS_MAX_DELAY_SECONDS * 1000; +} + +/** + * Mirrors the id BullMQ dedups on, which the producers already make unique per + * fire (a schedule extension re-queues the same job under `-ext{N}`). Reusing + * it as the schedule name gives EventBridge the same dedup semantics: a + * repeated enqueue collides by name instead of creating a second fire. + */ +function resolveScheduleId(job: IJobParams | IBulkJobParams): string { + return job.options?.jobId || job.data?._id || job.name; +} + export class QueueBaseService implements OnModuleDestroy { private bullMqService: BullMqService; @@ -22,7 +45,8 @@ export class QueueBaseService implements OnModuleDestroy { protected sqsService?: SqsService, protected featureFlagsService?: FeatureFlagsService, protected organizationRepository?: CommunityOrganizationRepository, - protected logger?: PinoLogger + protected logger?: PinoLogger, + protected schedulerService?: EventBridgeSchedulerService ) { this.bullMqService = bullMqService; if (logger) { @@ -107,18 +131,6 @@ export class QueueBaseService implements OnModuleDestroy { } public async add(params: IJobParams) { - const delayMs = params.options?.delay || 0; - - if (delayMs > SQS_MAX_DELAY_SECONDS * 1000) { - Logger.log( - { topic: this.topic, delay: delayMs }, - 'Job delay exceeds SQS max (15min), routing to BullMQ', - LOG_CONTEXT - ); - - return await this.addToBullMQ(params); - } - /* * When no SQS queue URL is configured for this topic (community edition, * self-hosted, or a partially rolled-out deployment), skip the whole @@ -143,18 +155,69 @@ export class QueueBaseService implements OnModuleDestroy { return await this.addToBullMQ(params); } - const queueBackendMode = await this.getQueueBackendMode(organizationId); - if (queueBackendMode === null) { + const routing = await this.resolveRouting(organizationId); + if (!routing) { return; } - Logger.debug({ topic: this.topic, queueBackendMode, organizationId }, 'Queue backend mode evaluation', LOG_CONTEXT); + return await this.dispatch([params], routing); + } + + /** + * Resolves every per-organization routing decision in one place so `add` and + * `addBulk` cannot drift apart. Returns undefined when the organization + * cannot be read, which callers treat as "skip the job". + */ + private async resolveRouting(organizationId: string): Promise { + const organization = await this.findOrganization(organizationId); + if (!organization) { + return undefined; + } + + const backendMode = await this.getQueueBackendMode(organization); + + Logger.debug({ topic: this.topic, backendMode, organizationId }, 'Queue backend mode evaluation', LOG_CONTEXT); + + return { organization, backendMode }; + } + + /** + * The single gate for long delays. SQS caps a per-message delay at 900s, so + * anything longer can only reach the queue through EventBridge Scheduler; + * until that is enabled for the organization such jobs keep going to BullMQ + * as they always have. Everything past this point may assume a long-delayed + * job is allowed to be scheduled. + */ + private async dispatch(jobs: (IJobParams | IBulkJobParams)[], routing: IQueueRouting): Promise { + const { longDelayed, sqsEligible } = this.separateByDelay(jobs); + + // Evaluated only when it can change the outcome, keeping the common + // short-delay path down to a single flag lookup. + if (longDelayed.length === 0 || (await this.isSchedulerEnabled(routing.organization))) { + return await this.routeByMode(jobs, routing); + } + + Logger.debug( + { topic: this.topic, count: longDelayed.length }, + 'Job delay exceeds SQS max (15min) and EventBridge Scheduler is off, routing to BullMQ', + LOG_CONTEXT + ); + await this.addJobsToBullMQ(longDelayed); + + if (sqsEligible.length === 0) { + return; + } - return await this.routeByMode([params], queueBackendMode, organizationId); + return await this.routeByMode(sqsEligible, routing); } - private async getQueueBackendMode(organizationId: string): Promise { - let organization: { _id: string; apiServiceLevel?: ApiServiceLevelEnum } | undefined; + /** + * Returns undefined when the organization cannot be read, which callers treat + * as "skip the job": if Mongo is unavailable the job cannot be executed + * downstream either, so enqueueing it anywhere would only defer the failure. + */ + private async findOrganization(organizationId: string): Promise { + let organization: OrganizationRouting | undefined; try { organization = await this.organizationRepository?.findOne({ _id: organizationId }, 'apiServiceLevel', { readPreference: 'secondaryPreferred', @@ -167,21 +230,32 @@ export class QueueBaseService implements OnModuleDestroy { ); } - /* - * If the organization is not found, we return null to indicate that the job should be skipped. - * There is no point in trying to route the job to SQS or BullMQ if the organization is not found. - */ - if (!organization) { Logger.warn({ organizationId, topic: this.topic }, 'Organization not found, skipping job', LOG_CONTEXT); - return null; + return undefined; } + return organization; + } + + private async getQueueBackendMode(organization: OrganizationRouting): Promise { return await this.featureFlagsService.getFlag({ key: FeatureFlagsKeysEnum.QUEUE_BACKEND_MODE, defaultValue: QueueBackendMode.BULLMQ, - organization: { _id: organizationId, apiServiceLevel: organization.apiServiceLevel }, + organization: { _id: organization._id, apiServiceLevel: organization.apiServiceLevel }, + }); + } + + private async isSchedulerEnabled(organization: OrganizationRouting): Promise { + if (!this.schedulerService?.isConfigured(this.topic)) { + return false; + } + + return await this.featureFlagsService.getFlag({ + key: FeatureFlagsKeysEnum.IS_EVENTBRIDGE_SCHEDULER_ENABLED, + defaultValue: false, + organization: { _id: organization._id, apiServiceLevel: organization.apiServiceLevel }, }); } @@ -189,11 +263,10 @@ export class QueueBaseService implements OnModuleDestroy { return jobs.map((job) => ({ ...job, data: { ...job.data, skipProcessing: true } })); } - private async routeByMode( - jobs: (IJobParams | IBulkJobParams)[], - queueBackendMode: string, - organizationId: string - ): Promise { + private async routeByMode(jobs: (IJobParams | IBulkJobParams)[], routing: IQueueRouting): Promise { + const { backendMode: queueBackendMode } = routing; + const organizationId = routing.organization._id; + switch (queueBackendMode) { case QueueBackendMode.BULLMQ: return await this.addJobsToBullMQ(jobs); @@ -287,7 +360,24 @@ export class QueueBaseService implements OnModuleDestroy { } private async addJobsToSQS(jobs: (IJobParams | IBulkJobParams)[], organizationId: string): Promise { - const messages = jobs.map((job, index) => ({ + /* + * Transport detail, not policy: `dispatch` has already decided these jobs + * may use the scheduler, and all this picks is which AWS call delivers + * them. Splitting here rather than above keeps long delays inside the + * mode's existing BullMQ fallback, since a schedule is just a deferred + * send to this same queue. + */ + const { longDelayed, sqsEligible } = this.separateByDelay(jobs); + + if (longDelayed.length > 0) { + await this.addJobsToScheduler(longDelayed, organizationId); + } + + if (sqsEligible.length === 0) { + return; + } + + const messages = sqsEligible.map((job, index) => ({ id: `${job.groupId || job.name}-${index}`, body: JSON.stringify(job.data || {}), groupId: organizationId, @@ -297,7 +387,11 @@ export class QueueBaseService implements OnModuleDestroy { if (messages.length === 1) { await this.sqsService.send(this.topic, messages[0]); Logger.debug( - { topic: this.topic, jobName: jobs[0].name, payloadSizeBytes: this.calculatePayloadSize(jobs[0].data) }, + { + topic: this.topic, + jobName: sqsEligible[0].name, + payloadSizeBytes: this.calculatePayloadSize(sqsEligible[0].data), + }, 'Added job to SQS', LOG_CONTEXT ); @@ -307,6 +401,37 @@ export class QueueBaseService implements OnModuleDestroy { } } + /** + * Throws when the scheduler is unavailable so the mode's existing catch + * blocks fall back to BullMQ, which is exactly what should happen to a job + * that has no other way to be delivered. + */ + private async addJobsToScheduler(jobs: (IJobParams | IBulkJobParams)[], organizationId: string): Promise { + if (!this.schedulerService) { + throw new Error(`EventBridge Scheduler is unavailable for long-delayed jobs on topic: ${this.topic}`); + } + + const now = Date.now(); + + await Promise.all( + jobs.map((job) => + this.schedulerService.createDelayedFire(this.topic, { + deferReason: job.deferReason || DeferReasonEnum.DELAY, + fireAt: new Date(now + (job.options?.delay || 0)), + organizationId, + scheduleId: resolveScheduleId(job), + messageBody: JSON.stringify(job.data || {}), + }) + ) + ); + + Logger.log( + { topic: this.topic, count: jobs.length, organizationId }, + 'Scheduled long-delayed jobs through EventBridge Scheduler', + LOG_CONTEXT + ); + } + protected async addToBullMQ(params: IJobParams) { const jobOptions = { removeOnComplete: true, @@ -332,50 +457,37 @@ export class QueueBaseService implements OnModuleDestroy { return await this.bullMqService.addBulk(data); } - const { longDelayed, sqsEligible } = this.separateByDelay(data); + const organizationId = data.find((job) => job.groupId)?.groupId; - if (longDelayed.length > 0) { + if (!organizationId) { Logger.debug( - { topic: this.topic, count: longDelayed.length }, - 'Routing long-delayed jobs (>15min) to BullMQ', + { topic: this.topic, count: data.length }, + 'Jobs without organization ID, routing to BullMQ fallback', LOG_CONTEXT ); - await this.bullMqService.addBulk(longDelayed); - } - if (sqsEligible.length > 0) { - const organizationId = sqsEligible[0]?.groupId; - - if (!organizationId) { - Logger.debug( - { topic: this.topic, count: sqsEligible.length }, - 'Jobs without organization ID, routing to BullMQ fallback', - LOG_CONTEXT - ); - await this.addJobsToBullMQ(sqsEligible); - - return; - } - - const queueBackendMode = await this.getQueueBackendMode(organizationId); - if (queueBackendMode === null) { - return; - } + return await this.addJobsToBullMQ(data); + } - await this.routeByMode(sqsEligible, queueBackendMode, organizationId); + const routing = await this.resolveRouting(organizationId); + if (!routing) { + return; } + + return await this.dispatch(data, routing); } - private separateByDelay(jobs: IBulkJobParams[]): { - longDelayed: IBulkJobParams[]; - sqsEligible: IBulkJobParams[]; + private separateByDelay( + jobs: T[] + ): { + longDelayed: T[]; + sqsEligible: T[]; } { - const longDelayed: IBulkJobParams[] = []; - const sqsEligible: IBulkJobParams[] = []; + const longDelayed: T[] = []; + const sqsEligible: T[] = []; for (const job of jobs) { - const delayMs = job.options?.delay || 0; - if (delayMs > SQS_MAX_DELAY_SECONDS * 1000) { + if (exceedsSqsDelayCap(job.options?.delay)) { longDelayed.push(job); } else { sqsEligible.push(job); @@ -435,6 +547,11 @@ export interface IJobParams { data?: any; groupId?: string; options?: JobsOptions; + /** + * Selects the EventBridge schedule group when the delay exceeds the SQS cap. + * Only meaningful for producers that can defer beyond 900s; ignored otherwise. + */ + deferReason?: DeferReasonEnum; } export interface IBulkJobParams { @@ -442,4 +559,5 @@ export interface IBulkJobParams { data: any; groupId?: string; options?: BulkJobOptions; + deferReason?: DeferReasonEnum; } diff --git a/libs/application-generic/src/services/queues/standard-queue.service.spec.ts b/libs/application-generic/src/services/queues/standard-queue.service.spec.ts index b0d54d40c5e..22bd5d49e0c 100644 --- a/libs/application-generic/src/services/queues/standard-queue.service.spec.ts +++ b/libs/application-generic/src/services/queues/standard-queue.service.spec.ts @@ -3,6 +3,7 @@ import { ApiServiceLevelEnum, QueueBackendMode } from '@novu/shared'; import { PinoLogger } from '../../logging'; import { FeatureFlagsService } from '../feature-flags'; import { WorkflowInMemoryProviderService } from '../in-memory-provider'; +import { EventBridgeSchedulerService } from '../scheduler'; import { SqsService } from '../sqs'; import { StandardQueueService } from './standard-queue.service'; @@ -32,6 +33,12 @@ const mockLogger = { error: jest.fn(), } as unknown as PinoLogger; +const mockSchedulerService = { + isConfigured: jest.fn(() => false), + createDelayedFire: jest.fn(), + deleteSchedule: jest.fn(), +} as unknown as EventBridgeSchedulerService; + describe('Standard Queue service', () => { describe('General', () => { beforeAll(async () => { @@ -40,7 +47,8 @@ describe('Standard Queue service', () => { mockSqsService, mockFeatureFlagsService, mockOrganizationRepository, - mockLogger + mockLogger, + mockSchedulerService ); await standardQueueService.queue.obliterate(); }); @@ -208,7 +216,8 @@ describe('Standard Queue service', () => { mockSqsService, mockFeatureFlagsService, mockOrganizationRepository, - mockLogger + mockLogger, + mockSchedulerService ); await standardQueueService.queue.obliterate(); }); diff --git a/libs/application-generic/src/services/queues/standard-queue.service.ts b/libs/application-generic/src/services/queues/standard-queue.service.ts index 88093d27175..c883b5c64ee 100644 --- a/libs/application-generic/src/services/queues/standard-queue.service.ts +++ b/libs/application-generic/src/services/queues/standard-queue.service.ts @@ -6,6 +6,7 @@ import { PinoLogger } from '../../logging'; import { BullMqService } from '../bull-mq'; import { FeatureFlagsService } from '../feature-flags'; import { WorkflowInMemoryProviderService } from '../in-memory-provider'; +import { EventBridgeSchedulerService } from '../scheduler'; import { SqsService } from '../sqs'; import { QueueBaseService } from './queue-base.service'; @@ -18,7 +19,8 @@ export class StandardQueueService extends QueueBaseService { sqsService: SqsService, featureFlagsService: FeatureFlagsService, organizationRepository: CommunityOrganizationRepository, - logger: PinoLogger + logger: PinoLogger, + schedulerService: EventBridgeSchedulerService ) { super( JobTopicNameEnum.STANDARD, @@ -26,7 +28,9 @@ export class StandardQueueService extends QueueBaseService { sqsService, featureFlagsService, organizationRepository, - logger + logger, + // Standard is the only topic that ever carries a delay. + schedulerService ); Logger.log({ topic: this.topic }, 'Creating queue', LOG_CONTEXT); diff --git a/libs/application-generic/src/services/readiness/readiness.service.spec.ts b/libs/application-generic/src/services/readiness/readiness.service.spec.ts index ef021b96041..74a42f0362c 100644 --- a/libs/application-generic/src/services/readiness/readiness.service.spec.ts +++ b/libs/application-generic/src/services/readiness/readiness.service.spec.ts @@ -9,6 +9,7 @@ import { BullMqService } from '../bull-mq'; import { FeatureFlagsService } from '../feature-flags'; import { WorkflowInMemoryProviderService } from '../in-memory-provider'; import { StandardQueueService, SubscriberProcessQueueService, WorkflowQueueService } from '../queues'; +import { EventBridgeSchedulerService } from '../scheduler'; import { SqsService } from '../sqs'; import { StandardWorkerService, WorkerBaseService } from '../workers'; import { ReadinessService } from './readiness.service'; @@ -41,6 +42,12 @@ const mockLogger = { error: jest.fn(), } as unknown as PinoLogger; +const mockSchedulerService = { + isConfigured: jest.fn(() => false), + createDelayedFire: jest.fn(), + deleteSchedule: jest.fn(), +} as unknown as EventBridgeSchedulerService; + describe('Readiness Service', () => { beforeAll(async () => { process.env.IN_MEMORY_CLUSTER_MODE_ENABLED = 'false'; @@ -51,7 +58,8 @@ describe('Readiness Service', () => { mockSqsService, mockFeatureFlagsService, mockOrganizationRepository, - mockLogger + mockLogger, + mockSchedulerService ); workflowQueueService = new WorkflowQueueService( new WorkflowInMemoryProviderService(), diff --git a/libs/application-generic/src/services/scheduler/event-bridge-scheduler.service.spec.ts b/libs/application-generic/src/services/scheduler/event-bridge-scheduler.service.spec.ts new file mode 100644 index 00000000000..a5ab4fba636 --- /dev/null +++ b/libs/application-generic/src/services/scheduler/event-bridge-scheduler.service.spec.ts @@ -0,0 +1,264 @@ +import { ConflictException, CreateScheduleCommand, DeleteScheduleCommand } from '@aws-sdk/client-scheduler'; +import { JobTopicNameEnum } from '@novu/shared'; + +import { SqsService } from '../sqs'; +import { deriveQueueArnFromUrl, EventBridgeSchedulerService } from './event-bridge-scheduler.service'; +import { DeferReasonEnum, SCHEDULER_MAX_INPUT_BYTES } from './types'; + +const mockSend = jest.fn(); + +jest.mock('@aws-sdk/client-scheduler', () => { + const actual = jest.requireActual('@aws-sdk/client-scheduler'); + + return { + ...actual, + SchedulerClient: jest.fn(() => ({ send: mockSend })), + }; +}); + +const QUEUE_URL = 'https://sqs.eu-west-2.amazonaws.com/354725113120/novu-standard-queue'; +const QUEUE_ARN = 'arn:aws:sqs:eu-west-2:354725113120:novu-standard-queue'; +const ROLE_ARN = 'arn:aws:iam::354725113120:role/novu-scheduler-role'; +const DLQ_ARN = 'arn:aws:sqs:eu-west-2:354725113120:novu-scheduler-dlq'; +const ORGANIZATION_ID = '65f1a2b3c4d5e6f708192a3b'; +const JOB_ID = '65f1a2b3c4d5e6f708192a3c'; + +/** Pass `null` to model a topic that has no SQS queue URL configured. */ +function buildSqsService(queueUrl: string | null = QUEUE_URL): SqsService { + return { getQueueUrl: jest.fn(() => queueUrl ?? undefined) } as unknown as SqsService; +} + +function configureEnv(): void { + process.env.EVENTBRIDGE_SCHEDULER_GROUP_PREFIX = 'novu-test'; + process.env.EVENTBRIDGE_SCHEDULER_ROLE_ARN = ROLE_ARN; + process.env.EVENTBRIDGE_SCHEDULER_DLQ_ARN = DLQ_ARN; +} + +function clearEnv(): void { + delete process.env.EVENTBRIDGE_SCHEDULER_GROUP_PREFIX; + delete process.env.EVENTBRIDGE_SCHEDULER_ROLE_ARN; + delete process.env.EVENTBRIDGE_SCHEDULER_DLQ_ARN; + delete process.env.EVENTBRIDGE_SCHEDULER_MAX_RETRY_ATTEMPTS; + delete process.env.EVENTBRIDGE_SCHEDULER_MAX_EVENT_AGE_SECONDS; +} + +function createParams(overrides: Record = {}) { + return { + deferReason: DeferReasonEnum.DELAY, + fireAt: new Date('2026-09-01T10:30:45.123Z'), + organizationId: ORGANIZATION_ID, + scheduleId: JOB_ID, + messageBody: JSON.stringify({ _id: JOB_ID }), + ...overrides, + } as Parameters[1]; +} + +describe('EventBridgeSchedulerService', () => { + beforeEach(() => { + jest.clearAllMocks(); + clearEnv(); + configureEnv(); + }); + + afterEach(() => { + clearEnv(); + }); + + describe('isConfigured', () => { + it('should be configured when the prefix, role, DLQ and queue URL are all present', () => { + const service = new EventBridgeSchedulerService(buildSqsService()); + + expect(service.isConfigured(JobTopicNameEnum.STANDARD)).toBe(true); + }); + + it.each([ + ['EVENTBRIDGE_SCHEDULER_GROUP_PREFIX'], + ['EVENTBRIDGE_SCHEDULER_ROLE_ARN'], + ['EVENTBRIDGE_SCHEDULER_DLQ_ARN'], + ])('should not be configured when %s is missing', (envKey) => { + delete process.env[envKey]; + const service = new EventBridgeSchedulerService(buildSqsService()); + + expect(service.isConfigured(JobTopicNameEnum.STANDARD)).toBe(false); + }); + + it('should not be configured when the topic has no SQS queue URL', () => { + const service = new EventBridgeSchedulerService(buildSqsService(null)); + + expect(service.isConfigured(JobTopicNameEnum.STANDARD)).toBe(false); + }); + + it('should not be configured when the queue URL is not an AWS host', () => { + const service = new EventBridgeSchedulerService(buildSqsService('http://localhost:4566/000000000000/standard')); + + expect(service.isConfigured(JobTopicNameEnum.STANDARD)).toBe(false); + }); + }); + + describe('createDelayedFire', () => { + it('should create a self-deleting schedule targeting the SQS queue', async () => { + const service = new EventBridgeSchedulerService(buildSqsService()); + + await service.createDelayedFire(JobTopicNameEnum.STANDARD, createParams()); + + expect(mockSend).toHaveBeenCalledTimes(1); + const command = mockSend.mock.calls[0][0] as CreateScheduleCommand; + expect(command).toBeInstanceOf(CreateScheduleCommand); + expect(command.input).toMatchObject({ + Name: `${ORGANIZATION_ID}-${JOB_ID}`, + GroupName: 'novu-test-delay', + ScheduleExpression: 'at(2026-09-01T10:30:45)', + FlexibleTimeWindow: { Mode: 'OFF' }, + ActionAfterCompletion: 'DELETE', + Target: { + Arn: QUEUE_ARN, + RoleArn: ROLE_ARN, + Input: JSON.stringify({ _id: JOB_ID }), + DeadLetterConfig: { Arn: DLQ_ARN }, + }, + }); + }); + + it('should not set a MessageGroupId, which EventBridge rejects for non-FIFO queues', async () => { + const service = new EventBridgeSchedulerService(buildSqsService()); + + await service.createDelayedFire(JobTopicNameEnum.STANDARD, createParams()); + + const command = mockSend.mock.calls[0][0] as CreateScheduleCommand; + expect(command.input.Target?.SqsParameters).toBeUndefined(); + }); + + it.each([ + [DeferReasonEnum.DELAY, 'novu-test-delay'], + [DeferReasonEnum.DIGEST, 'novu-test-digest'], + [DeferReasonEnum.THROTTLE, 'novu-test-throttle'], + [DeferReasonEnum.SNOOZE, 'novu-test-snooze'], + [DeferReasonEnum.SCHEDULE_EXTENSION, 'novu-test-schedule-extension'], + ])('should place a %s job in the %s group', async (deferReason, expectedGroup) => { + const service = new EventBridgeSchedulerService(buildSqsService()); + + await service.createDelayedFire(JobTopicNameEnum.STANDARD, createParams({ deferReason })); + + const command = mockSend.mock.calls[0][0] as CreateScheduleCommand; + expect(command.input.GroupName).toBe(expectedGroup); + }); + + it('should keep the schedule-extension suffix in the schedule name', async () => { + const service = new EventBridgeSchedulerService(buildSqsService()); + + await service.createDelayedFire( + JobTopicNameEnum.STANDARD, + createParams({ scheduleId: `${JOB_ID}-ext2`, deferReason: DeferReasonEnum.SCHEDULE_EXTENSION }) + ); + + const command = mockSend.mock.calls[0][0] as CreateScheduleCommand; + expect(command.input.Name).toBe(`${ORGANIZATION_ID}-${JOB_ID}-ext2`); + expect((command.input.Name as string).length).toBeLessThanOrEqual(64); + }); + + it('should apply the configured retry policy', async () => { + process.env.EVENTBRIDGE_SCHEDULER_MAX_RETRY_ATTEMPTS = '4'; + process.env.EVENTBRIDGE_SCHEDULER_MAX_EVENT_AGE_SECONDS = '600'; + const service = new EventBridgeSchedulerService(buildSqsService()); + + await service.createDelayedFire(JobTopicNameEnum.STANDARD, createParams()); + + const command = mockSend.mock.calls[0][0] as CreateScheduleCommand; + expect(command.input.Target?.RetryPolicy).toEqual({ + MaximumRetryAttempts: 4, + MaximumEventAgeInSeconds: 600, + }); + }); + + it('should treat a ConflictException as success so producer replays stay idempotent', async () => { + mockSend.mockRejectedValueOnce( + new ConflictException({ message: 'already exists', Message: 'already exists', $metadata: {} }) + ); + const service = new EventBridgeSchedulerService(buildSqsService()); + + await expect(service.createDelayedFire(JobTopicNameEnum.STANDARD, createParams())).resolves.toBeUndefined(); + }); + + it('should propagate other AWS errors so the caller can fall back to BullMQ', async () => { + mockSend.mockRejectedValueOnce(new Error('ThrottlingException')); + const service = new EventBridgeSchedulerService(buildSqsService()); + + await expect(service.createDelayedFire(JobTopicNameEnum.STANDARD, createParams())).rejects.toThrow( + 'ThrottlingException' + ); + }); + + it('should reject a body larger than the EventBridge input limit', async () => { + const service = new EventBridgeSchedulerService(buildSqsService()); + const oversized = 'x'.repeat(SCHEDULER_MAX_INPUT_BYTES + 1); + + await expect( + service.createDelayedFire(JobTopicNameEnum.STANDARD, createParams({ messageBody: oversized })) + ).rejects.toThrow(/exceeds the EventBridge limit/); + expect(mockSend).not.toHaveBeenCalled(); + }); + + it('should throw when the topic is not configured', async () => { + const service = new EventBridgeSchedulerService(buildSqsService(null)); + + await expect(service.createDelayedFire(JobTopicNameEnum.STANDARD, createParams())).rejects.toThrow( + /not configured/ + ); + expect(mockSend).not.toHaveBeenCalled(); + }); + }); + + describe('deleteSchedule', () => { + it('should delete the schedule from the reason group', async () => { + const service = new EventBridgeSchedulerService(buildSqsService()); + + await service.deleteSchedule({ + deferReason: DeferReasonEnum.SNOOZE, + organizationId: ORGANIZATION_ID, + scheduleId: JOB_ID, + }); + + const command = mockSend.mock.calls[0][0] as DeleteScheduleCommand; + expect(command).toBeInstanceOf(DeleteScheduleCommand); + expect(command.input).toEqual({ + Name: `${ORGANIZATION_ID}-${JOB_ID}`, + GroupName: 'novu-test-snooze', + }); + }); + + it('should be a no-op when the scheduler is not configured', async () => { + clearEnv(); + const service = new EventBridgeSchedulerService(buildSqsService()); + + await service.deleteSchedule({ + deferReason: DeferReasonEnum.SNOOZE, + organizationId: ORGANIZATION_ID, + scheduleId: JOB_ID, + }); + + expect(mockSend).not.toHaveBeenCalled(); + }); + }); + + describe('deriveQueueArnFromUrl', () => { + it.each([ + ['https://sqs.eu-west-2.amazonaws.com/354725113120/novu-standard-queue', QUEUE_ARN], + [ + 'https://eu-west-2.queue.amazonaws.com/354725113120/novu-standard-queue', + 'arn:aws:sqs:eu-west-2:354725113120:novu-standard-queue', + ], + ['https://sqs.us-gov-west-1.amazonaws.com/354725113120/q', 'arn:aws-us-gov:sqs:us-gov-west-1:354725113120:q'], + ['https://sqs.cn-north-1.amazonaws.com.cn/354725113120/q', 'arn:aws-cn:sqs:cn-north-1:354725113120:q'], + ])('should derive an ARN from %s', (url, expected) => { + expect(deriveQueueArnFromUrl(url)).toBe(expected); + }); + + it.each([ + ['http://localhost:4566/000000000000/standard'], + ['not-a-url'], + ['https://sqs.eu-west-2.amazonaws.com/onlyonesegment'], + ])('should return undefined for the unsupported URL %s', (url) => { + expect(deriveQueueArnFromUrl(url)).toBeUndefined(); + }); + }); +}); diff --git a/libs/application-generic/src/services/scheduler/event-bridge-scheduler.service.ts b/libs/application-generic/src/services/scheduler/event-bridge-scheduler.service.ts new file mode 100644 index 00000000000..6262e58d877 --- /dev/null +++ b/libs/application-generic/src/services/scheduler/event-bridge-scheduler.service.ts @@ -0,0 +1,270 @@ +import { + ActionAfterCompletion, + ConflictException, + CreateScheduleCommand, + DeleteScheduleCommand, + FlexibleTimeWindowMode, + ResourceNotFoundException, + SchedulerClient, +} from '@aws-sdk/client-scheduler'; +import { Injectable, Logger } from '@nestjs/common'; +import { JobTopicNameEnum } from '@novu/shared'; + +import { SqsService } from '../sqs'; +import { + DeferReasonEnum, + ICreateDelayedFireParams, + IDeleteScheduleParams, + SCHEDULER_DEFAULT_MAX_EVENT_AGE_SECONDS, + SCHEDULER_DEFAULT_MAX_RETRY_ATTEMPTS, + SCHEDULER_MAX_INPUT_BYTES, + SCHEDULER_MAX_NAME_LENGTH, + SCHEDULER_NAME_PATTERN, +} from './types'; + +const LOG_CONTEXT = 'EventBridgeSchedulerService'; + +/** + * Creates one-shot EventBridge schedules for job delays that exceed the SQS + * per-message cap of 900s. Each schedule targets the standard SQS queue + * directly and self-deletes once it fires. + * + * Fires deliberately carry no `MessageGroupId`: EventBridge Scheduler's + * templated SQS target rejects it for non-FIFO queues, so long-delay fires + * arrive without the per-org fair-queue attribution that the direct SQS path + * sets. Acceptable because they are a small fraction of queue volume. + */ +@Injectable() +export class EventBridgeSchedulerService { + private client?: SchedulerClient; + private readonly groupPrefix?: string; + private readonly roleArn?: string; + private readonly dlqArn?: string; + private readonly maxRetryAttempts: number; + private readonly maxEventAgeSeconds: number; + + constructor(private readonly sqsService: SqsService) { + this.groupPrefix = normalizeEnv(process.env.EVENTBRIDGE_SCHEDULER_GROUP_PREFIX); + this.roleArn = normalizeEnv(process.env.EVENTBRIDGE_SCHEDULER_ROLE_ARN); + this.dlqArn = normalizeEnv(process.env.EVENTBRIDGE_SCHEDULER_DLQ_ARN); + this.maxRetryAttempts = toPositiveInt( + process.env.EVENTBRIDGE_SCHEDULER_MAX_RETRY_ATTEMPTS, + SCHEDULER_DEFAULT_MAX_RETRY_ATTEMPTS + ); + this.maxEventAgeSeconds = toPositiveInt( + process.env.EVENTBRIDGE_SCHEDULER_MAX_EVENT_AGE_SECONDS, + SCHEDULER_DEFAULT_MAX_EVENT_AGE_SECONDS + ); + + if (this.hasCredentials()) { + const region = process.env.AWS_REGION || process.env.NOVU_REGION || 'us-east-1'; + this.client = new SchedulerClient({ region }); + Logger.log({ groupPrefix: this.groupPrefix, region }, 'EventBridge Scheduler service initialized', LOG_CONTEXT); + } else { + Logger.log('EventBridge Scheduler service initialized with no configuration', LOG_CONTEXT); + } + } + + /** + * A schedule is only usable when the group prefix, the execution role and + * the dead-letter queue are all present: without the DLQ a failed fire is + * lost with no way to recover the job, which is worse than never leaving + * BullMQ. + */ + public isConfigured(topic: JobTopicNameEnum): boolean { + return this.hasCredentials() && !!this.client && !!this.resolveQueueArn(topic); + } + + public async createDelayedFire(topic: JobTopicNameEnum, params: ICreateDelayedFireParams): Promise { + const { deferReason, fireAt, organizationId, scheduleId, messageBody } = params; + + const queueArn = this.resolveQueueArn(topic); + if (!this.client || !queueArn) { + throw new Error(`EventBridge Scheduler is not configured for topic: ${topic}`); + } + + const name = buildScheduleName(organizationId, scheduleId); + const inputBytes = Buffer.byteLength(messageBody, 'utf8'); + if (inputBytes > SCHEDULER_MAX_INPUT_BYTES) { + throw new Error( + `Schedule input of ${inputBytes} bytes exceeds the EventBridge limit of ${SCHEDULER_MAX_INPUT_BYTES}` + ); + } + + try { + await this.client.send( + new CreateScheduleCommand({ + Name: name, + GroupName: this.buildGroupName(deferReason), + // The delay is already an absolute instant, so the expression is + // formatted in UTC and ScheduleExpressionTimezone is left unset + // (EventBridge defaults to UTC). + ScheduleExpression: toScheduleExpression(fireAt), + FlexibleTimeWindow: { Mode: FlexibleTimeWindowMode.OFF }, + ActionAfterCompletion: ActionAfterCompletion.DELETE, + Target: { + Arn: queueArn, + RoleArn: this.roleArn, + Input: messageBody, + RetryPolicy: { + MaximumRetryAttempts: this.maxRetryAttempts, + MaximumEventAgeInSeconds: this.maxEventAgeSeconds, + }, + DeadLetterConfig: { Arn: this.dlqArn }, + }, + }) + ); + + Logger.debug( + { topic, name, deferReason, fireAt: fireAt.toISOString(), inputBytes }, + 'Created EventBridge schedule for long-delayed job', + LOG_CONTEXT + ); + } catch (error) { + /* + * A same-name schedule already exists, which means an earlier attempt + * succeeded and the producer is being replayed (an SQS redelivery of the + * parent message re-runs AddJob). The job is already scheduled, so this + * is a success, not a failure to fall back on. + */ + if (error instanceof ConflictException) { + Logger.debug({ topic, name, deferReason }, 'EventBridge schedule already exists, skipping', LOG_CONTEXT); + + return; + } + + throw error; + } + } + + /** + * Deletes a schedule. Only used for snooze, where the job document is + * removed on unsnooze and a later fire would otherwise fail to find it and + * churn through SQS redeliveries. A schedule that is already gone (fired, or + * never created because the job fell back to BullMQ) is not an error. + */ + public async deleteSchedule(params: IDeleteScheduleParams): Promise { + const { deferReason, organizationId, scheduleId } = params; + + if (!this.client) { + return; + } + + const name = buildScheduleName(organizationId, scheduleId); + + try { + await this.client.send(new DeleteScheduleCommand({ Name: name, GroupName: this.buildGroupName(deferReason) })); + + Logger.debug({ name, deferReason }, 'Deleted EventBridge schedule', LOG_CONTEXT); + } catch (error) { + if (error instanceof ResourceNotFoundException) { + return; + } + + throw error; + } + } + + public buildGroupName(deferReason: DeferReasonEnum): string { + return `${this.groupPrefix}-${deferReason}`; + } + + private hasCredentials(): boolean { + return !!this.groupPrefix && !!this.roleArn && !!this.dlqArn; + } + + private resolveQueueArn(topic: JobTopicNameEnum): string | undefined { + const queueUrl = this.sqsService?.getQueueUrl(topic); + + return queueUrl ? deriveQueueArnFromUrl(queueUrl) : undefined; + } +} + +function normalizeEnv(value: string | undefined): string | undefined { + return value && value.trim() !== '' ? value.trim() : undefined; +} + +function toPositiveInt(value: string | undefined, fallback: number): number { + const parsed = Number(value); + + return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : fallback; +} + +/** `at()` takes a local-to-the-schedule-timezone stamp with no offset suffix. */ +function toScheduleExpression(fireAt: Date): string { + return `at(${fireAt.toISOString().slice(0, 19)})`; +} + +/** + * The org prefix makes schedules enumerable per tenant via + * `ListSchedules --name-prefix`. Two ObjectIds plus a separator is 49 + * characters, and the `-ext{N}` suffix a schedule extension carries adds a + * handful more, so the AWS 64-character ceiling is never a practical concern. + */ +export function buildScheduleName(organizationId: string, scheduleId: string): string { + const name = `${organizationId}-${scheduleId}`; + + if (name.length > SCHEDULER_MAX_NAME_LENGTH) { + throw new Error(`Schedule name '${name}' exceeds the ${SCHEDULER_MAX_NAME_LENGTH} character limit`); + } + + if (!SCHEDULER_NAME_PATTERN.test(name)) { + throw new Error(`Schedule name '${name}' contains characters EventBridge does not accept`); + } + + return name; +} + +/** + * SQS exposes queues by URL but EventBridge targets them by ARN, and the + * deployment already configures the URL. Supports both the modern + * (`sqs.{region}.amazonaws.com`) and legacy (`{region}.queue.amazonaws.com`) + * host forms; anything else (LocalStack) yields no ARN, which disables the + * scheduler path rather than emitting a malformed target. + */ +export function deriveQueueArnFromUrl(queueUrl: string): string | undefined { + let url: URL; + try { + url = new URL(queueUrl); + } catch { + return undefined; + } + + const [accountId, queueName] = url.pathname.split('/').filter(Boolean); + if (!accountId || !queueName) { + return undefined; + } + + const region = extractRegion(url.hostname); + if (!region) { + return undefined; + } + + return `arn:${partitionForRegion(region)}:sqs:${region}:${accountId}:${queueName}`; +} + +function extractRegion(hostname: string): string | undefined { + const modern = /^sqs\.([a-z0-9-]+)\.amazonaws\.com(\.cn)?$/.exec(hostname); + if (modern) { + return modern[1]; + } + + const legacy = /^([a-z0-9-]+)\.queue\.amazonaws\.com(\.cn)?$/.exec(hostname); + if (legacy) { + return legacy[1]; + } + + return undefined; +} + +function partitionForRegion(region: string): string { + if (region.startsWith('us-gov-')) { + return 'aws-us-gov'; + } + + if (region.startsWith('cn-')) { + return 'aws-cn'; + } + + return 'aws'; +} diff --git a/libs/application-generic/src/services/scheduler/index.ts b/libs/application-generic/src/services/scheduler/index.ts new file mode 100644 index 00000000000..514cafb3963 --- /dev/null +++ b/libs/application-generic/src/services/scheduler/index.ts @@ -0,0 +1,2 @@ +export * from './event-bridge-scheduler.service'; +export * from './types'; diff --git a/libs/application-generic/src/services/scheduler/types.ts b/libs/application-generic/src/services/scheduler/types.ts new file mode 100644 index 00000000000..59df5acef5b --- /dev/null +++ b/libs/application-generic/src/services/scheduler/types.ts @@ -0,0 +1,53 @@ +/** + * Why a job was deferred. Each reason maps to a pre-created EventBridge + * Scheduler group (`${prefix}-${reason}`) so schedules can be listed, metered + * and cleaned up per reason. Derived from the job rather than read off + * `job.type` directly: snooze and schedule extensions are channel-typed jobs. + */ +export enum DeferReasonEnum { + DELAY = 'delay', + DIGEST = 'digest', + THROTTLE = 'throttle', + SNOOZE = 'snooze', + SCHEDULE_EXTENSION = 'schedule-extension', +} + +/** AWS limit on `CreateSchedule.Name`. */ +export const SCHEDULER_MAX_NAME_LENGTH = 64; + +/** Characters AWS accepts in a schedule name. */ +export const SCHEDULER_NAME_PATTERN = /^[0-9a-zA-Z-_.]+$/; + +/** + * AWS limit on `Target.Input`. Standard-topic bodies are ~160 bytes (four + * ObjectIds), so this only ever guards against an unexpected payload shape. + */ +export const SCHEDULER_MAX_INPUT_BYTES = 256 * 1024; + +/** + * Scheduler retries delivery of the fire to SQS, not execution of the job. + * Once exhausted the fire goes to the schedule's dead-letter queue. Kept far + * below the AWS default (185 attempts / 24h) so a job cannot silently surface + * a day late; SQS SendMessage failures are rare and short-lived. + */ +export const SCHEDULER_DEFAULT_MAX_RETRY_ATTEMPTS = 10; +export const SCHEDULER_DEFAULT_MAX_EVENT_AGE_SECONDS = 3600; + +export interface ICreateDelayedFireParams { + /** Selects the schedule group. */ + deferReason: DeferReasonEnum; + /** Absolute UTC instant at which the message should land on the queue. */ + fireAt: Date; + /** Tenant, used as the schedule name prefix so schedules are enumerable per org. */ + organizationId: string; + /** Job id, already carrying the `-ext{N}` suffix for schedule extensions. */ + scheduleId: string; + /** Serialized queue message, delivered verbatim as the SQS message body. */ + messageBody: string; +} + +export interface IDeleteScheduleParams { + deferReason: DeferReasonEnum; + organizationId: string; + scheduleId: string; +} diff --git a/packages/shared/src/types/feature-flags.ts b/packages/shared/src/types/feature-flags.ts index d1464034019..51303b1a06b 100644 --- a/packages/shared/src/types/feature-flags.ts +++ b/packages/shared/src/types/feature-flags.ts @@ -212,6 +212,14 @@ export enum FeatureFlagsKeysEnum { */ IS_SUBSCRIBER_CHAT_OAUTH_HMAC_REQUIRED_ENABLED = 'IS_SUBSCRIBER_CHAT_OAUTH_HMAC_REQUIRED_ENABLED', + /** + * Route job delays longer than the SQS 900s per-message cap through + * EventBridge Scheduler instead of BullMQ. Only consulted once + * `QUEUE_BACKEND_MODE` has SQS as the primary backend; when false (default) + * long delays keep going to BullMQ. + */ + IS_EVENTBRIDGE_SCHEDULER_ENABLED = 'IS_EVENTBRIDGE_SCHEDULER_ENABLED', + // String flags QUEUE_BACKEND_MODE = 'QUEUE_BACKEND_MODE', // Values: "bullmq" | "shadow" | "live" | "complete" USAGE_REPORT_TRIGGER_SECRET = 'USAGE_REPORT_TRIGGER_SECRET', diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b2266c516c3..3feb556ecd6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -461,7 +461,7 @@ importers: version: link:../../libs/testing '@novu/thalamus': specifier: 0.1.0-alpha.18 - version: 0.1.0-alpha.18(@anthropic-ai/sdk@0.95.1(zod@3.25.20))(@aws-crypto/sha256-js@5.2.0)(@smithy/signature-v4@5.4.3)(openai@6.17.0(ws@8.21.0)(zod@3.25.20)) + version: 0.1.0-alpha.18(@anthropic-ai/sdk@0.95.1(zod@3.25.20))(@aws-crypto/sha256-js@5.2.0)(@smithy/signature-v4@5.6.12)(openai@6.17.0(ws@8.21.0)(zod@3.25.20)) '@sendgrid/mail': specifier: ^8.1.6 version: 8.1.6 @@ -2480,7 +2480,7 @@ importers: dependencies: '@novu/thalamus': specifier: 0.1.0-alpha.18 - version: 0.1.0-alpha.18(@anthropic-ai/sdk@0.95.1(zod@4.3.6))(@aws-crypto/sha256-js@5.2.0)(@smithy/signature-v4@5.4.3)(openai@6.17.0(ws@8.21.0)(zod@4.3.6)) + version: 0.1.0-alpha.18(@anthropic-ai/sdk@0.95.1(zod@4.3.6))(@aws-crypto/sha256-js@5.2.0)(@smithy/signature-v4@5.6.12)(openai@6.17.0(ws@8.21.0)(zod@4.3.6)) agents: specifier: ^0.21.0 version: 0.21.0(@ai-sdk/react@3.0.51(react@19.2.3)(zod@4.3.6))(@babel/core@7.29.7)(@babel/runtime@7.28.3)(@cloudflare/workers-types@4.20260702.1)(@modelcontextprotocol/client@2.0.0)(@modelcontextprotocol/sdk@1.30.0(@cfworker/json-schema@4.1.1)(zod@4.3.6))(@modelcontextprotocol/server@2.0.0)(ai@7.0.31(zod@4.3.6))(chat@4.33.0(ai@7.0.31(zod@4.3.6))(zod@4.3.6))(react@19.2.3)(rolldown@1.2.4)(zod@4.3.6) @@ -2540,6 +2540,9 @@ importers: '@aws-sdk/client-s3': specifier: ^3.996.0 version: 3.996.0 + '@aws-sdk/client-scheduler': + specifier: 3.996.0 + version: 3.996.0 '@aws-sdk/client-secrets-manager': specifier: 3.996.0 version: 3.996.0 @@ -2844,6 +2847,9 @@ importers: jest: specifier: ^27.1.0 version: 27.5.1(ts-node@10.9.2(@swc/core@1.7.26(@swc/helpers@0.5.15))(@types/node@22.15.13)(typescript@5.6.2)) + jest-environment-node: + specifier: ^27.5.1 + version: 27.5.1 npm-run-all: specifier: ^4.1.5 version: 4.1.5 @@ -5075,6 +5081,10 @@ packages: resolution: {integrity: sha512-BZsCeq8Sgqbm6xs8VfjyVVwhQZvxDR45P22dcbNNDFaGkkQ/TbJ5KxER19APR9aK+IC7l4KuLxInqeVab2DFfg==} engines: {node: '>=20.0.0'} + '@aws-sdk/client-scheduler@3.996.0': + resolution: {integrity: sha512-5TAGuf7F0hYzVxcy0nNVd46vt53K2E4QM6IXwbTQIdvqcnOi65uR0g7PLXDKRWkzASU53viAGAJoz3Rd87RQcQ==} + engines: {node: '>=20.0.0'} + '@aws-sdk/client-secrets-manager@3.996.0': resolution: {integrity: sha512-90EfmhOcj7/m4NyKG5B2LM0/elN1zniqHwUGzJMRADXAXl973jnATT6NbKIDyzHtHYZxo8RughyBIl+uKbt2JA==} engines: {node: '>=20.0.0'} @@ -5127,6 +5137,10 @@ packages: resolution: {integrity: sha512-njR2qoG6ZuB0kvAS2FyICsFZJ6gmCcf2X/7JcD14sUvGDm26wiZ5BrA6LOiUxKFEF+IVe7kdroxyE00YlkiYsw==} engines: {node: '>=20.0.0'} + '@aws-sdk/core@3.977.6': + resolution: {integrity: sha512-QiaJV4/zDrB4ZY2mfeSXSzSTc36W16sZXcGz+SPFk0CJ26gziO0cS+4LjJUMAbdeeBOvS0k0Aq1cZpfGdUXxSw==} + engines: {node: '>=20.0.0'} + '@aws-sdk/crc64-nvme@3.972.0': resolution: {integrity: sha512-ThlLhTqX68jvoIVv+pryOdb5coP1cX1/MaTbB9xkGDCbWbsqQcLqzPxuSoW1DCnAAIacmXCWpzUNOB9pv+xXQw==} engines: {node: '>=20.0.0'} @@ -5155,6 +5169,10 @@ packages: resolution: {integrity: sha512-29wX9zpAvEt1vcj0psha+y6ygBHy2V/S72mp6e7q0KARLWXq+pwE/lR6qGkwknQvruh52lXvlqZIga8Hdxkucw==} engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-env@3.972.67': + resolution: {integrity: sha512-rcIpk5kxUqDaaNa6Xk23pQ6ViY7jlqzmfFWCahQcBT97ddXaXYYwzCen9Tz1Jvo6aJft6wDl5bN44/Jw5B4oLA==} + engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-http@3.575.0': resolution: {integrity: sha512-xQfVmYI+9KqRvhWY8fyElnpcVUBBUgi/Hoji3oU6WLrUjrX98k93He7gKDQSyHf7ykMLUAJYWwsV4AjQ2j6njA==} engines: {node: '>=16.0.0'} @@ -5175,6 +5193,10 @@ packages: resolution: {integrity: sha512-IA3CQTjtJkb6u1H4mE4936c8OPBMa9Jggtwe8U2Mqw/vvb/tZ5Ebd0mcZcX0uKWQhOyYo/+qNIwkV5Xh+FeJJA==} engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-http@3.972.69': + resolution: {integrity: sha512-nggwJtZ4eeNsUw5IeWBMXsi1ryct5idi0K+/SCRF3kybLubOMaNTb3XCihXpWMiVpyzyPeIrl0zTkzhBH9porA==} + engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-ini@3.575.0': resolution: {integrity: sha512-BdM6a/5VUuNge3c6yRuxvO+4srLoSfqHfkQGfUDfhTdTJpljlpfnc9h3z2Ni1+aueOHPZMNFWIktHDcX5wUGBg==} engines: {node: '>=16.0.0'} @@ -5193,6 +5215,10 @@ packages: resolution: {integrity: sha512-4mzII+3mZEVXXE1xzrLQrCJL7/r62A63bA6SVzZoNL5rqCJghpf+xgGltVrIBBs0n+mOZBKrQl2tRREtvZ5l6A==} engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-ini@3.973.12': + resolution: {integrity: sha512-pNEf/OeyN5X3VmLKlgSO6TqaWmW10CvI3TfwL1XhsuhYjSLT2VDaxFnCPHnOeQXSaFisMX4jNhpETriqN8DOmg==} + engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-login@3.972.10': resolution: {integrity: sha512-7Me+/EkY3kQC1nehBjb9ryc558N+a8R4Dg3rSV3zpiB7iQtvXh4gU3rV14h/dIbn2/VkK9sh55YdXamSjfdb/Q==} engines: {node: '>=20.0.0'} @@ -5205,6 +5231,10 @@ packages: resolution: {integrity: sha512-HG7kQCwXtbv3oBV61Ins0oNX8KKyvrMqqRkb6ZiAfQHbMuHaiNaEb2KnpKLPkNpqImSBK82UkVE/kaY6IfWikA==} engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-login@3.972.74': + resolution: {integrity: sha512-0AQfDcf99TNmqVKv0owHrw/TQs6i4ZE5t9qmz6NvO53bE/sA/tpXhXL9AAcEP1qHc6Zzjd1UMb69+/9zdhvY3g==} + engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-node@3.575.0': resolution: {integrity: sha512-rEdNpqW2jEc5kwbf/s9XQywMLQlIkMjuCK6mw9sF2OVRGHGVnh+6eh/1JFx8Kj+eU51ctifQ7KaHe8dGco8HYQ==} engines: {node: '>=16.0.0'} @@ -5221,6 +5251,10 @@ packages: resolution: {integrity: sha512-sDaBIT0yrNNIPfvlsiTCmANm07zKju+ipWODjEXgZlsjMeIJR3LVp7RDyAOzUoAsTbDfYKDWp+i5WrFiQP6rmQ==} engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-node@3.972.78': + resolution: {integrity: sha512-OgPAnfvbGAMWac6yvxJ1ihslrvDpPVwR68D2csospdNCCyPvHk9JLzYKwz48SNiS1T2znDwHauywRKRFfpyYng==} + engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-process@3.575.0': resolution: {integrity: sha512-2/5NJV7MZysKglqJSQ/O8OELNcwLcH3xknabL9NagtzB7RNB2p1AUXR0UlTey9sSDLL4oCmNa/+unYuglW/Ahg==} engines: {node: '>=16.0.0'} @@ -5241,6 +5275,10 @@ packages: resolution: {integrity: sha512-2k/amBifLd75eXNwgvPw/2lKYSQ3NhvHQgkVKVjfUq13/eJ3JRtHmznuFenn74OK3sSfp4SMy1YB2w+UVXoKqA==} engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-process@3.972.67': + resolution: {integrity: sha512-IlUEejorGTWKb4/Dm7K5Yw4QxUmXLThLhrvBmzVBqZFTbW72cv9LTcITmo1dsnYriALE4h68mOq4LB99x6sQ7Q==} + engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-sso@3.575.0': resolution: {integrity: sha512-NtXA9OPIKsqavs2F7hhLT/t2ZDjwJsvQevj31ov1NpmTNYMc7OWFWDptOG7rppsWMsk5KKmfiL2qViQJnezXNA==} engines: {node: '>=16.0.0'} @@ -5261,6 +5299,10 @@ packages: resolution: {integrity: sha512-LPc3+Y4vhH1T4x6CMqwCM6hk5+SRf/Lwmgm8INm95wxTtIRHcMwQUVkDzWu4Iw/RSncxYM2BC01OrYbxOPZvyg==} engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-sso@3.973.11': + resolution: {integrity: sha512-gAQBkBZxUB84d71+pPcI9L+jh2ujhuAVxc/4FgGiWFDjkPBlMKxzd5XDtkSXTFX8Ro7ansnT88+XadasxMeCRw==} + engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-web-identity@3.575.0': resolution: {integrity: sha512-QcvVH7wpvpFRXGAGgCBfQeiF/ptD0NJ+Hrc8dDYfPGhFeZ0EoVQBYNphLi25xe7JZ+XbaqCKrURHZtr4fAEOJw==} engines: {node: '>=16.0.0'} @@ -5283,6 +5325,10 @@ packages: resolution: {integrity: sha512-wQtL34lUD/09VXjwAUo2T+I3aEXRDxMB3DKmTJL/Zj0Gi6sLDTrVhae1XVt01yzkquOWajI/sZW72JGDZ1ciTw==} engines: {node: '>=20.0.0'} + '@aws-sdk/credential-provider-web-identity@3.972.73': + resolution: {integrity: sha512-SnlEmQa6SjOgs6iOPLUQl1Eyq4AKiAdPQlkOhFhqNfDtDCwibMGvL6QlkSmf3o6vAUSImzdPCxowT5dfQUZP1A==} + engines: {node: '>=20.0.0'} + '@aws-sdk/credential-providers@3.1052.0': resolution: {integrity: sha512-PpF9zZ8Mkzb2M9u/O4blveBRF4vo9FdMhsnUcfZBjam06mfrUnhnKZw9yTfdGxvO9noxyPIrxQZ4tyj2NxEDTg==} engines: {node: '>=20.0.0'} @@ -5427,6 +5473,10 @@ packages: resolution: {integrity: sha512-nWXXJ1r/r8N2Gw1pWolRgED38/A9A8DHR2ETWIv220zh4PZHcybbR4hUVWWktmNXTRHzDJwRluapHn0rZxuoqA==} engines: {node: '>=20.0.0'} + '@aws-sdk/nested-clients@3.997.41': + resolution: {integrity: sha512-RDHqPGQWlF6tatA/Tp3rg6oIwtgN9IVderxE+9av2Y93Dfyu+mO1hZ5Bu2jpfZg2rwdNbsssnwM+sLafIczMlQ==} + engines: {node: '>=20.0.0'} + '@aws-sdk/nested-clients@3.997.6': resolution: {integrity: sha512-WBDnqatJl+kGObpfmfSxqnXeYTu3Me8wx8WCtvoxX3pfWrrTv8I4WTMSSs7PZqcRcVh8WeUKMgGFjMG+52SR1w==} engines: {node: '>=20.0.0'} @@ -5475,6 +5525,10 @@ packages: resolution: {integrity: sha512-qs9z5LqXO/CZC2Lg9SGKpoLU8Rhi+m2pFKZqfO9pytX1clc0katqtsDNupJxFy0xT9wsZSPzM2v1y+/H/zfp5Q==} engines: {node: '>=20.0.0'} + '@aws-sdk/signature-v4-multi-region@3.996.43': + resolution: {integrity: sha512-lKekx8bLBXSv4O+cslk9Zfnw2XKSkWBs3uWL5QGhH2ZAQfNS7FE0vcSSN2vD/AhxX54ZTywWxR4STThoeOXlBA==} + engines: {node: '>=20.0.0'} + '@aws-sdk/token-providers@3.1021.0': resolution: {integrity: sha512-TKY6h9spUk3OLs5v1oAgW9mAeBE3LAGNBwJokLy96wwmd4W2v/tYlXseProyed9ValDj2u1jK/4Rg1T+1NXyJA==} engines: {node: '>=20.0.0'} @@ -5487,6 +5541,10 @@ packages: resolution: {integrity: sha512-QqZNB3so7UIDxZtroc85TQaLVxdZRFm0eWM1CSR2N+b06as9TOrilvrlTZuj3guYlxMs6yLOgGxnklJ5qMYtTw==} engines: {node: '>=20.0.0'} + '@aws-sdk/token-providers@3.1103.0': + resolution: {integrity: sha512-N4wy26MNn31ItGVHYHPrEuCIFY4MBBjC+C5v1lJKqIUSA7OZBdhleCY53zCCrXn27hsk7YNOaTuhQu807S4AfQ==} + engines: {node: '>=20.0.0'} + '@aws-sdk/token-providers@3.575.0': resolution: {integrity: sha512-EPNDPQoQkjKqn4D2t70qVzbfdtlaAy9KBdG58qD1yNWVxq8Rh/lXdwmB+aE2PSahtyfVikZdCRoZiFzxDh5IUA==} engines: {node: '>=16.0.0'} @@ -5517,6 +5575,10 @@ packages: resolution: {integrity: sha512-kuBfgQVdcz5Bmapc4A13YbpVw/pXkesfhetcFYwbntqas8sF41OHyd4o28+/TG2ZQdHBsv90Lsu5y6oitvYCdg==} engines: {node: '>=20.0.0'} + '@aws-sdk/types@3.974.2': + resolution: {integrity: sha512-3W6IUtSxFbH6X7Wb7DzGCV5QiFQsd0g8bOfntpmDxQlzBoKWUMBu/JPQR0DwkE+Hpnxd6db1tXbOwdeHddG6cA==} + engines: {node: '>=20.0.0'} + '@aws-sdk/util-arn-parser@3.568.0': resolution: {integrity: sha512-XUKJWWo+KOB7fbnPP0+g/o5Ulku/X53t7i/h+sPHr5xxYTJJ9CYnbToo95mzxe7xWvkLrsNtJ8L+MnNn9INs2w==} engines: {node: '>=16.0.0'} @@ -5624,6 +5686,10 @@ packages: resolution: {integrity: sha512-GH+Kjz4nPKWKHnsiQpnhP1MJdTGIcK4rAka6tzakgjjUkVgNsmPeEbbRAf09SzS1hjGu6duGHCBsxYke0BhHjQ==} engines: {node: '>=20.0.0'} + '@aws-sdk/xml-builder@3.972.37': + resolution: {integrity: sha512-zKq4HQum8JwDyEuyfuI4bbiAcU0KxP6qy+9PR/IsR92IyE/DaBAikzAS50tjxip4bqIIANpCcG+Yyj6CVhXupg==} + engines: {node: '>=20.0.0'} + '@aws-sdk/xml-builder@3.972.5': resolution: {integrity: sha512-mCae5Ys6Qm1LDu0qdGwx2UQ63ONUe+FHw908fJzLDqFKTDBK4LDZUqKWm4OkTCNFq19bftjsBSESIGLD/s3/rA==} engines: {node: '>=20.0.0'} @@ -5632,6 +5698,10 @@ packages: resolution: {integrity: sha512-oLvsaPMTBejkkmHhjf09xTgk71mOqyr/409NKhRIL08If7AhVfUsJhVsx386uJaqNd42v9kWamQ9lFbkoC2dYw==} engines: {node: '>=18.0.0'} + '@aws/lambda-invoke-store@0.3.0': + resolution: {integrity: sha512-sl4Bm6yiMNYrZKkqqDFWN0UfnWhlS8ivKxrYl+6t0gCLrqr8y3B2IqZZbFRkfaVVp7C/baApyh71P+LeE1A2sQ==} + engines: {node: '>=18.0.0'} + '@azure/abort-controller@1.1.0': resolution: {integrity: sha512-TrRLIoSQVzfAJX9H1JeFjzAoDGcoK1IYX1UImfceTZpsyYfWr09Ss1aHW1y5TrrR3iq6RZLBwJ3E24uwPhwahw==} engines: {node: '>=12.0.0'} @@ -12970,6 +13040,10 @@ packages: resolution: {integrity: sha512-Ep/7tPamGY8mgESE3LyLKtxJyy6U52WWAqr/3wial47Sj4u3PiIF73AOGI27UyLy9duTkhZbgzodOfLV4TduZg==} engines: {node: '>=18.0.0'} + '@smithy/core@3.31.1': + resolution: {integrity: sha512-CyogUINxvi7C7LDsh8Syo6hVJOT9ckz4rG8dRZfTJ8r91HkMY59PnNooaj7WcHyxEkxPfBAmbgztZU+xTo76lg==} + engines: {node: '>=18.0.0'} + '@smithy/credential-provider-imds@3.2.0': resolution: {integrity: sha512-0SCIzgd8LYZ9EJxUjLXBmEKSZR/P/w6l7Rz/pab9culE/RWuqelAKGJvn5qUOl8BgX8Yj5HWM50A5hiB/RzsgA==} engines: {node: '>=16.0.0'} @@ -12990,6 +13064,10 @@ packages: resolution: {integrity: sha512-I2Bti0DKFo2IJyN28ijCsx51BAumEYR4/1yZ1FXyBygy9MqbnMqCev4JPth/MbpRfBSRAX35hITSnAdJRo1u5w==} engines: {node: '>=18.0.0'} + '@smithy/credential-provider-imds@4.4.16': + resolution: {integrity: sha512-QfuLWAkLzptffFW980AFeHZFdqds2B64rpEd3uJ6lgs3xVn9QegGMUgUcj+4d7dRrAsya3r58ZKpku97WcFb4w==} + engines: {node: '>=18.0.0'} + '@smithy/eventstream-codec@3.0.0': resolution: {integrity: sha512-PUtyEA0Oik50SaEFCZ0WPVtF9tz/teze2fDptW6WRXl+RrEenH8UbEjudOz8iakiMl3lE3lCVqYf2Y+znL8QFQ==} @@ -13048,6 +13126,10 @@ packages: resolution: {integrity: sha512-F+DRf8IJazRJgYog2A/yJK7eYVc0rqTlRzO+5ZxjJd4WkZoKz0IJRncf7G6t1pdVT3kryJcwuTFhN1c5m6N47A==} engines: {node: '>=18.0.0'} + '@smithy/fetch-http-handler@5.6.13': + resolution: {integrity: sha512-4fW86pEUOMbrD5nkbyl/tTvPHHWJFbuB2odl6ps9lWfHoXf9HWh3Q/Smh59qH1g7+c/BSZghX6bbUk4gsiMs8A==} + engines: {node: '>=18.0.0'} + '@smithy/hash-blob-browser@3.0.0': resolution: {integrity: sha512-/Wbpdg+bwJvW7lxR/zpWAc1/x/YkcqguuF2bAzkJrvXriZu1vm8r+PUdE4syiVwQg7PPR2dXpi3CLBb9qRDaVQ==} @@ -13233,6 +13315,10 @@ packages: resolution: {integrity: sha512-/jPhevcTFPMVl6KNjbaI47iOg1zxC7IsnX4PQDGVZKMFceOXtB8IEYaB7a9VvkP/3oC60WzTeKocvSI7vLT0vA==} engines: {node: '>=18.0.0'} + '@smithy/node-http-handler@4.9.13': + resolution: {integrity: sha512-Nmd/Nl35zfYrd+a6OO2cDJb3GPh9bgTjIUhcM+JFfjpp8/osCgboDV5nCT1I01Pv6R13eSKDKLSoVa5ZB6Zsfw==} + engines: {node: '>=18.0.0'} + '@smithy/property-provider@3.1.3': resolution: {integrity: sha512-zahyOVR9Q4PEoguJ/NrFP4O7SMAfYO1HLhB18M+q+Z4KFd4V2obiMnlVoUFzFLSPeVt1POyNWneHHrZaTMoc/g==} engines: {node: '>=16.0.0'} @@ -13357,6 +13443,10 @@ packages: resolution: {integrity: sha512-53+75QuPl6DL+ct6vVEB51FDO5oulXr20TPV46VvJZg76lIlXNWfxi8j+G2V/t0I2qxCBOa3vX/8bmjrpFVo9g==} engines: {node: '>=18.0.0'} + '@smithy/signature-v4@5.6.12': + resolution: {integrity: sha512-I6KLtq3H0qqSuV9vLglfi8puHqzygzWHOnI4z/Rdoo+q50vvo18vBRdPAvvEtcaKROz7Zn6qnPa14kRfPH6PcQ==} + engines: {node: '>=18.0.0'} + '@smithy/smithy-client@3.2.0': resolution: {integrity: sha512-pDbtxs8WOhJLJSeaF/eAbPgXg4VVYFlRcL/zoNYA5WbG3wBL06CHtBSg53ppkttDpAJ/hdiede+xApip1CwSLw==} engines: {node: '>=16.0.0'} @@ -13397,6 +13487,10 @@ packages: resolution: {integrity: sha512-P+otAxbV4CqBybp7EkcJCrig63yE2E7PuNVOmilVMRcx/O+QDzGULTrKsq4DV13gSfak9ObPrWaHl/9bL5YcWw==} engines: {node: '>=18.0.0'} + '@smithy/types@4.16.1': + resolution: {integrity: sha512-0JFs3V2y2M9tKW5na/qxe69Zv+uxLMO7QBbhxF/FHu/Gp2NFZAAL9tWl9PU02xxo07pb3G9FTyjNc6D5uZrJIg==} + engines: {node: '>=18.0.0'} + '@smithy/url-parser@3.0.3': resolution: {integrity: sha512-pw3VtZtX2rg+s6HMs6/+u9+hu6oY6U7IohGhVNnjbgKy86wcIsSZwgHrFR+t67Uyxvp4Xz3p3kGXXIpTNisq8A==} @@ -29200,6 +29294,50 @@ snapshots: transitivePeerDependencies: - aws-crt + '@aws-sdk/client-scheduler@3.996.0': + dependencies: + '@aws-crypto/sha256-browser': 5.2.0 + '@aws-crypto/sha256-js': 5.2.0 + '@aws-sdk/core': 3.977.6 + '@aws-sdk/credential-provider-node': 3.972.78 + '@aws-sdk/middleware-host-header': 3.972.10 + '@aws-sdk/middleware-logger': 3.972.10 + '@aws-sdk/middleware-recursion-detection': 3.972.11 + '@aws-sdk/middleware-user-agent': 3.972.38 + '@aws-sdk/region-config-resolver': 3.972.13 + '@aws-sdk/types': 3.974.2 + '@aws-sdk/util-endpoints': 3.996.0 + '@aws-sdk/util-user-agent-browser': 3.972.10 + '@aws-sdk/util-user-agent-node': 3.973.24 + '@smithy/config-resolver': 4.4.17 + '@smithy/core': 3.31.1 + '@smithy/fetch-http-handler': 5.6.13 + '@smithy/hash-node': 4.2.14 + '@smithy/invalid-dependency': 4.2.14 + '@smithy/middleware-content-length': 4.2.14 + '@smithy/middleware-endpoint': 4.4.32 + '@smithy/middleware-retry': 4.5.7 + '@smithy/middleware-serde': 4.2.20 + '@smithy/middleware-stack': 4.2.14 + '@smithy/node-config-provider': 4.3.14 + '@smithy/node-http-handler': 4.9.13 + '@smithy/protocol-http': 5.3.14 + '@smithy/smithy-client': 4.12.13 + '@smithy/types': 4.16.1 + '@smithy/url-parser': 4.2.14 + '@smithy/util-base64': 4.3.2 + '@smithy/util-body-length-browser': 4.2.2 + '@smithy/util-body-length-node': 4.2.3 + '@smithy/util-defaults-mode-browser': 4.3.49 + '@smithy/util-defaults-mode-node': 4.2.54 + '@smithy/util-endpoints': 3.4.2 + '@smithy/util-middleware': 4.2.14 + '@smithy/util-retry': 4.3.8 + '@smithy/util-utf8': 4.2.2 + tslib: 2.8.1 + transitivePeerDependencies: + - aws-crt + '@aws-sdk/client-secrets-manager@3.996.0': dependencies: '@aws-crypto/sha256-browser': 5.2.0 @@ -29626,6 +29764,17 @@ snapshots: '@smithy/util-utf8': 4.2.2 tslib: 2.8.1 + '@aws-sdk/core@3.977.6': + dependencies: + '@aws-sdk/types': 3.974.2 + '@aws-sdk/xml-builder': 3.972.37 + '@aws/lambda-invoke-store': 0.3.0 + '@smithy/core': 3.31.1 + '@smithy/signature-v4': 5.6.12 + '@smithy/types': 4.16.1 + bowser: 2.11.0 + tslib: 2.8.1 + '@aws-sdk/crc64-nvme@3.972.0': dependencies: '@smithy/types': 4.14.2 @@ -29678,6 +29827,14 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/credential-provider-env@3.972.67': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/credential-provider-http@3.575.0': dependencies: '@aws-sdk/types': 3.575.0 @@ -29739,6 +29896,16 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/credential-provider-http@3.972.69': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/fetch-http-handler': 5.6.13 + '@smithy/node-http-handler': 4.9.13 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/credential-provider-ini@3.575.0(@aws-sdk/client-sso-oidc@3.575.0)(@aws-sdk/client-sts@3.575.0)': dependencies: '@aws-sdk/client-sts': 3.575.0(@aws-sdk/client-sso-oidc@3.575.0) @@ -29810,6 +29977,22 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/credential-provider-ini@3.973.12': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/credential-provider-env': 3.972.67 + '@aws-sdk/credential-provider-http': 3.972.69 + '@aws-sdk/credential-provider-login': 3.972.74 + '@aws-sdk/credential-provider-process': 3.972.67 + '@aws-sdk/credential-provider-sso': 3.973.11 + '@aws-sdk/credential-provider-web-identity': 3.972.73 + '@aws-sdk/nested-clients': 3.997.41 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/credential-provider-imds': 4.4.16 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/credential-provider-login@3.972.10': dependencies: '@aws-sdk/core': 3.974.13 @@ -29845,6 +30028,15 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/credential-provider-login@3.972.74': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/nested-clients': 3.997.41 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/credential-provider-node@3.575.0(@aws-sdk/client-sso-oidc@3.575.0)(@aws-sdk/client-sts@3.575.0)': dependencies: '@aws-sdk/credential-provider-env': 3.575.0 @@ -29912,6 +30104,20 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/credential-provider-node@3.972.78': + dependencies: + '@aws-sdk/credential-provider-env': 3.972.67 + '@aws-sdk/credential-provider-http': 3.972.69 + '@aws-sdk/credential-provider-ini': 3.973.12 + '@aws-sdk/credential-provider-process': 3.972.67 + '@aws-sdk/credential-provider-sso': 3.973.11 + '@aws-sdk/credential-provider-web-identity': 3.972.73 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/credential-provider-imds': 4.4.16 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/credential-provider-process@3.575.0': dependencies: '@aws-sdk/types': 3.575.0 @@ -29955,6 +30161,14 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/credential-provider-process@3.972.67': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/credential-provider-sso@3.575.0(@aws-sdk/client-sso-oidc@3.575.0)': dependencies: '@aws-sdk/client-sso': 3.575.0 @@ -30017,6 +30231,16 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/credential-provider-sso@3.973.11': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/nested-clients': 3.997.41 + '@aws-sdk/token-providers': 3.1103.0 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/credential-provider-web-identity@3.575.0(@aws-sdk/client-sts@3.575.0)': dependencies: '@aws-sdk/client-sts': 3.575.0(@aws-sdk/client-sso-oidc@3.575.0) @@ -30070,6 +30294,15 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/credential-provider-web-identity@3.972.73': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/nested-clients': 3.997.41 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/credential-providers@3.1052.0': dependencies: '@aws-sdk/client-cognito-identity': 3.1052.0 @@ -30480,6 +30713,17 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/nested-clients@3.997.41': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/signature-v4-multi-region': 3.996.43 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/fetch-http-handler': 5.6.13 + '@smithy/node-http-handler': 4.9.13 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/nested-clients@3.997.6': dependencies: '@aws-crypto/sha256-browser': 5.2.0 @@ -30623,6 +30867,13 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/signature-v4-multi-region@3.996.43': + dependencies: + '@aws-sdk/types': 3.974.2 + '@smithy/signature-v4': 5.6.12 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/token-providers@3.1021.0': dependencies: '@aws-sdk/core': 3.974.13 @@ -30656,6 +30907,15 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/token-providers@3.1103.0': + dependencies: + '@aws-sdk/core': 3.977.6 + '@aws-sdk/nested-clients': 3.997.41 + '@aws-sdk/types': 3.974.2 + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/token-providers@3.575.0(@aws-sdk/client-sso-oidc@3.575.0)': dependencies: '@aws-sdk/client-sso-oidc': 3.575.0 @@ -30702,6 +30962,11 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@aws-sdk/types@3.974.2': + dependencies: + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/util-arn-parser@3.568.0': dependencies: tslib: 2.8.1 @@ -30853,6 +31118,11 @@ snapshots: fast-xml-parser: 5.7.3 tslib: 2.8.1 + '@aws-sdk/xml-builder@3.972.37': + dependencies: + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@aws-sdk/xml-builder@3.972.5': dependencies: '@smithy/types': 4.14.2 @@ -30861,6 +31131,8 @@ snapshots: '@aws/lambda-invoke-store@0.2.3': {} + '@aws/lambda-invoke-store@0.3.0': {} + '@azure/abort-controller@1.1.0': dependencies: tslib: 2.8.1 @@ -36122,18 +36394,18 @@ snapshots: - bufferutil - utf-8-validate - '@novu/thalamus@0.1.0-alpha.18(@anthropic-ai/sdk@0.95.1(zod@3.25.20))(@aws-crypto/sha256-js@5.2.0)(@smithy/signature-v4@5.4.3)(openai@6.17.0(ws@8.21.0)(zod@3.25.20))': + '@novu/thalamus@0.1.0-alpha.18(@anthropic-ai/sdk@0.95.1(zod@3.25.20))(@aws-crypto/sha256-js@5.2.0)(@smithy/signature-v4@5.6.12)(openai@6.17.0(ws@8.21.0)(zod@3.25.20))': optionalDependencies: '@anthropic-ai/sdk': 0.95.1(zod@3.25.20) '@aws-crypto/sha256-js': 5.2.0 - '@smithy/signature-v4': 5.4.3 + '@smithy/signature-v4': 5.6.12 openai: 6.17.0(ws@8.21.0)(zod@3.25.20) - '@novu/thalamus@0.1.0-alpha.18(@anthropic-ai/sdk@0.95.1(zod@4.3.6))(@aws-crypto/sha256-js@5.2.0)(@smithy/signature-v4@5.4.3)(openai@6.17.0(ws@8.21.0)(zod@4.3.6))': + '@novu/thalamus@0.1.0-alpha.18(@anthropic-ai/sdk@0.95.1(zod@4.3.6))(@aws-crypto/sha256-js@5.2.0)(@smithy/signature-v4@5.6.12)(openai@6.17.0(ws@8.21.0)(zod@4.3.6))': optionalDependencies: '@anthropic-ai/sdk': 0.95.1(zod@4.3.6) '@aws-crypto/sha256-js': 5.2.0 - '@smithy/signature-v4': 5.4.3 + '@smithy/signature-v4': 5.6.12 openai: 6.17.0(ws@8.21.0)(zod@4.3.6) '@nrwl/nx-cloud@19.1.0': @@ -41156,6 +41428,11 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@smithy/core@3.31.1': + dependencies: + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@smithy/credential-provider-imds@3.2.0': dependencies: '@smithy/node-config-provider': 3.1.4 @@ -41194,6 +41471,12 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@smithy/credential-provider-imds@4.4.16': + dependencies: + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@smithy/eventstream-codec@3.0.0': dependencies: '@aws-crypto/crc32': 3.0.0 @@ -41292,6 +41575,12 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@smithy/fetch-http-handler@5.6.13': + dependencies: + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@smithy/hash-blob-browser@3.0.0': dependencies: '@smithy/chunked-blob-reader': 3.0.0 @@ -41619,6 +41908,12 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@smithy/node-http-handler@4.9.13': + dependencies: + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@smithy/property-provider@3.1.3': dependencies: '@smithy/types': 3.3.0 @@ -41803,6 +42098,12 @@ snapshots: '@smithy/types': 4.14.2 tslib: 2.8.1 + '@smithy/signature-v4@5.6.12': + dependencies: + '@smithy/core': 3.31.1 + '@smithy/types': 4.16.1 + tslib: 2.8.1 + '@smithy/smithy-client@3.2.0': dependencies: '@smithy/middleware-endpoint': 3.1.0 @@ -41866,6 +42167,10 @@ snapshots: dependencies: tslib: 2.8.1 + '@smithy/types@4.16.1': + dependencies: + tslib: 2.8.1 + '@smithy/url-parser@3.0.3': dependencies: '@smithy/querystring-parser': 3.0.3 From 7a9669ed5cb2a5f3bc5933868900ee664faa4349 Mon Sep 17 00:00:00 2001 From: Pawan Jain Date: Thu, 20 Aug 2026 20:10:20 +0530 Subject: [PATCH 2/7] feat(docs): enhance chat step documentation with block editor details and rich card examples fixes DOC-429 (#12313) --- docs/docs.json | 11 +- docs/framework/chat-channel.mdx | 33 ++++ docs/framework/typescript/steps/chat.mdx | 151 ++++++++++++++++-- docs/platform/integrations/chat.mdx | 6 +- .../integrations/chat/adding-chat.mdx | 3 +- docs/platform/integrations/chat/line.mdx | 4 +- docs/platform/integrations/chat/ms-teams.mdx | 4 + docs/platform/integrations/chat/slack.mdx | 13 +- docs/platform/integrations/chat/whats-app.mdx | 2 +- .../chat/writing-chat-template.mdx | 124 ++++++++++++++ .../workflow/add-and-configure-steps.mdx | 2 +- .../add-and-configure-steps/code-steps.mdx | 2 +- .../channels-template-editors.mdx | 11 +- 13 files changed, 333 insertions(+), 33 deletions(-) create mode 100644 docs/platform/integrations/chat/writing-chat-template.mdx diff --git a/docs/docs.json b/docs/docs.json index 08dad79a68b..086cfe9372f 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -194,6 +194,7 @@ "group": "Chat", "root": "platform/integrations/chat", "pages": [ + "platform/integrations/chat/writing-chat-template", "platform/integrations/chat/discord", "platform/integrations/chat/line", "platform/integrations/chat/mattermost", @@ -1034,16 +1035,16 @@ { "group": "Steps", "pages": [ + "framework/typescript/steps", "framework/typescript/steps/chat", - "framework/typescript/steps/custom", - "framework/typescript/steps/delay", - "framework/typescript/steps/digest", "framework/typescript/steps/email", "framework/typescript/steps/inApp", - "framework/typescript/steps", "framework/typescript/steps/push", "framework/typescript/steps/sms", - "framework/typescript/steps/tool" + "framework/typescript/steps/tool", + "framework/typescript/steps/custom", + "framework/typescript/steps/delay", + "framework/typescript/steps/digest" ] }, "framework/schema/zod", diff --git a/docs/framework/chat-channel.mdx b/docs/framework/chat-channel.mdx index fed62a6fd37..bfbbcfffb8b 100644 --- a/docs/framework/chat-channel.mdx +++ b/docs/framework/chat-channel.mdx @@ -7,6 +7,8 @@ Novu brings chat notifications into your development workflow, giving you a unif Learn more about the [Chat Channel](/platform/integrations/chat). +## Plain text + ```tsx await step.chat('chat', async () => { return { @@ -14,3 +16,34 @@ await step.chat('chat', async () => { }; }); ``` + +## Rich card + +Return a `card` for structured messages (text, images, link buttons, and more). Existing `body`-only steps stay valid. + +```tsx +import { Actions, Card, CardText, Divider } from '@novu/framework'; + +await step.chat('chat', async () => { + return { + card: Card({ + title: 'Deploy finished', + children: [ + CardText('All checks passed.'), + Divider(), + Actions([ + { + type: 'link-button', + id: 'view', + label: 'View deploy', + url: 'https://example.com/deploys/123', + style: 'primary', + }, + ]), + ], + }), + }; +}); +``` + +See the [Chat step reference](/framework/typescript/steps/chat) for the full output schema, and [Writing chat templates](/platform/integrations/chat/writing-chat-template) for the Dashboard editors. diff --git a/docs/framework/typescript/steps/chat.mdx b/docs/framework/typescript/steps/chat.mdx index aa5bf0c7874..539a00245b3 100644 --- a/docs/framework/typescript/steps/chat.mdx +++ b/docs/framework/typescript/steps/chat.mdx @@ -1,11 +1,15 @@ --- title: 'Chat' -description: "Use the chat step in Novu Framework to send messages to Slack, Discord, Microsoft Teams, and other chat providers as part of a workflow." +description: "Use the chat step in Novu Framework to send plain-text or rich card messages to Slack, Discord, Microsoft Teams, and other chat providers." --- -The `chat` step allows you to send messages to various chat platforms like Slack, Discord, and Microsoft Teams. +The `chat` step sends a message to chat platforms such as Slack, Discord, and Microsoft Teams. -## Example Usage +Return a plain-text `body` or a rich `card`. If you return both, Novu uses `card`. Existing workflows that only return `body` keep working without changes. + +## Example usage + +### Plain text body ```tsx await step.chat('chat', async () => { @@ -15,12 +19,141 @@ await step.chat('chat', async () => { }); ``` -## Chat Step Output +### Rich card + +Build cards with the Card DSL helpers from `@novu/framework`. The Card DSL extends the raw Chat SDK card model with provider-agnostic Markdown formatting. Novu converts that formatting to the syntax each provider supports. + +```tsx +import { + Actions, + Card, + CardLink, + CardText, + Divider, + Image, +} from '@novu/framework'; + +await step.chat('chat', async () => { + return { + card: Card({ + title: 'Deploy finished', + children: [ + CardText('All checks **passed** for `production`.'), + Divider(), + Image({ + url: 'https://example.com/status.png', + alt: 'Status', + }), + Actions([ + { + type: 'link-button', + id: 'view', + label: 'View deploy', + url: 'https://example.com/deploys/123', + style: 'primary', + }, + ]), + CardLink({ + url: 'https://example.com/deploys/123', + label: 'Open in dashboard', + }), + ], + }), + }; +}); +``` + +You can also return a raw card object that matches the schema below. Prefer the helpers so TypeScript catches shape mistakes early. + +Card text supports `**bold**`, `_italic_`, `~~strikethrough~~`, inline code, and `[links](https://example.com)`. For example, Slack receives equivalent `mrkdwn`, while Novu converts the same source text for other providers. + + + Link button URLs must be absolute (include `https://`). Incomplete URLs can fail provider validation at send time. + + +## Chat step output + +Provide at least one of `body` or `card`. + +### body + +- **Type**: `string` +- **Required**: No (required if `card` is omitted) +- **Description**: Plain-text message body. If the output also includes `card`, Novu uses the card and ignores `body`. + +### card + +- **Type**: `object` (`CardElement`) +- **Required**: No (required if `body` is omitted) +- **Description**: Structured chat card. Novu provides provider-specific rendering for Slack, Microsoft Teams, WhatsApp, and Telegram. Other providers receive a Markdown fallback generated from the card. + +#### Card properties + +| Property | Type | Required | Description | +| --- | --- | --- | --- | +| `type` | `'card'` | Yes | Discriminator. Set automatically when you use `Card()`. | +| `title` | `string` | No | Card title. | +| `subtitle` | `string` | No | Secondary title line. | +| `imageUrl` | `string` | No | Optional header image URL. | +| `children` | `array` | Yes | Ordered list of card child elements. | + +#### Card child types + +| Type | Helpers | Description | +| --- | --- | --- | +| `text` | `CardText` | Text block. Optional `style`: `plain`, `bold`, `muted`. | +| `image` | `Image` | Image with `url` and optional `alt`. | +| `divider` | `Divider` | Visual separator. | +| `link` | `CardLink` | Inline link with `label` and `url`. | +| `actions` | `Actions` | Wrapper for link buttons. | +| `link-button` | (inside `Actions`) | Button that opens a URL. Requires `label` and `url`. Optional `style`: `primary`, `danger`, `default`. | +| `fields` | | Label/value pairs (`field` children). | +| `table` | | Table with `headers` and `rows`. | +| `section` | | Nested group of child elements. | + +The Dashboard [block editor](/platform/integrations/chat/writing-chat-template) authors a common subset of this model (text, image, divider, link buttons, lists, and related layout blocks). Framework also supports the layout elements listed above, including sections, fields, and tables. + +## Provider overrides + +When you need a provider-native payload that the shared card cannot express, use the step `providers` object. For example, Slack Block Kit: + +```tsx +await step.chat( + 'slack-rich', + async () => ({ + body: 'Fallback text for notifications and plain-text clients', + }), + { + providers: { + slack: async ({ controls }) => ({ + blocks: [ + { + type: 'section', + text: { + type: 'mrkdwn', + text: '*Deploy finished*', + }, + }, + ], + }), + }, + } +); +``` + +See [Providers overrides](/framework/typescript/steps#providers-overrides-object) and the [Chat channel](/platform/integrations/chat#provider-content-overrides) docs for Dashboard overrides. + +## Chat step result -| Property | Type | Required | Description | -| -------- | ------ | -------- | ------------------------------------------ | -| body | string | Yes | The message to be sent to the chat channel | +The `chat` step does not return a result object. -## Chat Step Result +## Related -The `chat` step does not return any result object. + + + Dashboard block and text editors, preview, and backward compatibility. + + + Full card kit usage for agent replies and `onAction`. + + diff --git a/docs/platform/integrations/chat.mdx b/docs/platform/integrations/chat.mdx index 7fb94df2374..73abca324a2 100644 --- a/docs/platform/integrations/chat.mdx +++ b/docs/platform/integrations/chat.mdx @@ -42,8 +42,8 @@ This step defines when and how a chat message should be sent as part of your not ### Define the chat content -Within the Chat step editor, write the message body. -The editor supports dynamic data for personalized and contextual messages. +Within the Chat step editor, compose the message with the **block editor** (structured cards with text, images, and link buttons) or the **text editor** (plain text and Liquid). +Both modes support dynamic data for personalized messages. See [Writing chat templates](/platform/integrations/chat/writing-chat-template). @@ -243,7 +243,7 @@ curl -L -X PUT 'https://api.novu.co/v1/subscribers//credentials' ## Provider content overrides -The Chat step body is plain text. When you need a platform's richer message format, such as Slack Block Kit, configure a **provider content override** on the Chat step in the workflow editor. +For many workflows, the [Chat block editor](/platform/integrations/chat/writing-chat-template) is enough: you author one shared rich message and Novu maps it to each provider. Use a **provider content override** when you need a platform's native message format instead, such as Slack Block Kit fields that the shared card model does not expose. An override is a JSON object that Novu merges into the request it sends to that provider. It is saved on the step, so it is versioned and promoted between environments with the rest of the workflow, and it applies to every trigger without any change to your trigger call. diff --git a/docs/platform/integrations/chat/adding-chat.mdx b/docs/platform/integrations/chat/adding-chat.mdx index a61eb3877e7..13bc3815219 100644 --- a/docs/platform/integrations/chat/adding-chat.mdx +++ b/docs/platform/integrations/chat/adding-chat.mdx @@ -38,8 +38,9 @@ Chat channels allow you to deliver instant, contextual messages to your subscrib - Click the **"Add a Workflow"** button - Add a step and select **"Chat"** as the channel - Configure the Chat content: - - Message body (e.g., `{{subscriber.firstName}}, your order {{orderId}} has shipped.`) + - Use the **block editor** for structured messages (text, images, link buttons), or the **text editor** for a plain-text / Liquid body (for example `{{subscriber.firstName}}, your order {{orderId}} has shipped.`) - Dynamic placeholders for personalized content + - See [Writing chat templates](/platform/integrations/chat/writing-chat-template) for editor details and backward compatibility - Optionally, set fallback channels to ensure reliable delivery if Chat fails diff --git a/docs/platform/integrations/chat/line.mdx b/docs/platform/integrations/chat/line.mdx index 1bc5b92574c..e5b9d513148 100644 --- a/docs/platform/integrations/chat/line.mdx +++ b/docs/platform/integrations/chat/line.mdx @@ -127,11 +127,11 @@ await novu.trigger({ }); ``` -The chat step content is sent as a plain text LINE message unless you override the message type at trigger time. +The chat step content is sent as a LINE text message by default. Block or card content is converted to a text/markdown-style fallback. Use trigger overrides for `flex`, `image`, or `sticker` when you need LINE-native message types. See [Writing chat templates](/platform/integrations/chat/writing-chat-template). ## Rich message types -Pass `flex`, `image`, or `sticker` under `overrides.chat` when triggering a workflow. When present, Novu sends that message type instead of plain text. +Pass `flex`, `image`, or `sticker` under `overrides.chat` when triggering a workflow. When present, Novu sends that message type instead of the rendered Chat step text. diff --git a/docs/platform/integrations/chat/ms-teams.mdx b/docs/platform/integrations/chat/ms-teams.mdx index 7ff095cdb58..cb37bf9af05 100644 --- a/docs/platform/integrations/chat/ms-teams.mdx +++ b/docs/platform/integrations/chat/ms-teams.mdx @@ -14,6 +14,10 @@ Once a customer connects their workspace, Novu establishes a secure channel conn Check out the [agents](/agents) documentation for more information on how to build agents using Microsoft Teams. + + Compose rich Chat step content with the [block editor](/platform/integrations/chat/writing-chat-template), or keep a plain-text / Liquid body. Novu renders shared cards as Microsoft Teams Adaptive Cards. Use provider overrides when you need Teams-native payloads beyond the shared card model. + + ## Prerequisites - Access to [Azure Portal](https://portal.azure.com/) with App Registration permissions. diff --git a/docs/platform/integrations/chat/slack.mdx b/docs/platform/integrations/chat/slack.mdx index 74e442d658d..8d9d2a3857b 100644 --- a/docs/platform/integrations/chat/slack.mdx +++ b/docs/platform/integrations/chat/slack.mdx @@ -20,7 +20,7 @@ This guide walks you through setting up Slack chat, connect workspaces, and deli - The Chat step body is plain text. To send [Slack Block Kit](https://api.slack.com/block-kit) messages, either save a Slack content override on the Chat step in the workflow editor (rolling out gradually — see [Configure Slack overrides in the dashboard](#configure-slack-overrides-in-the-dashboard)), or pass `blocks` in [trigger overrides](/platform/integrations/trigger-overrides) when you call the API. + Compose rich Chat step content with the [block editor](/platform/integrations/chat/writing-chat-template), or keep a plain-text / Liquid body. To send [Slack Block Kit](https://api.slack.com/block-kit) payloads that go beyond the shared card model, save a Slack content override on the Chat step (rolling out gradually; see [Configure Slack overrides in the dashboard](#configure-slack-overrides-in-the-dashboard)), or pass `blocks` in [trigger overrides](/platform/integrations/trigger-overrides) when you call the API. @@ -864,23 +864,24 @@ When the workflow is triggered, Novu will: - Novu sends the notification to each configured Slack destination. By default, the Chat step body is sent as plain text. To send Block Kit formatting, see [Format messages with Block Kit](#format-messages-with-block-kit). + Novu sends the notification to each configured Slack destination. Block-editor cards are rendered to Slack-native formatting when possible. Text-editor bodies are sent as plain text. To send custom Block Kit, see [Format messages with Block Kit](#format-messages-with-block-kit). ## Format messages with Block Kit -The Chat step body itself is plain text. To send richer Slack messages, layer a **Slack content override** on top of it. An override is a JSON object whose keys map to [Slack `chat.postMessage`](https://api.slack.com/methods/chat.postMessage) arguments, and Novu merges it into the request it sends to Slack. +Use a **Slack content override** when you need Slack-native Block Kit that the shared [Chat block editor](/platform/integrations/chat/writing-chat-template) does not cover. An override is a JSON object whose keys map to [Slack `chat.postMessage`](https://api.slack.com/methods/chat.postMessage) arguments, and Novu merges it into the request it sends to Slack. You can supply that object from three places: | Approach | Best for | | --- | --- | +| **Chat block editor (default content)** | Shared rich messages (text, images, link buttons) that Novu maps to Slack | | **Slack overrides on the step (dashboard)** | Block Kit that is part of the workflow itself, saved and versioned with the step | | **Trigger overrides (API)** | Block Kit that only your calling code can build, decided per trigger | | **Framework provider overrides** | Full control in code-first workflows, including dynamic digest formatting | -If you send no override at all, Slack receives the rendered Chat step body as plain text, which is enough for summaries built from [digest variables](/platform/workflow/add-notification-content/personalize-content#digest-variables). +If you use the text editor and send no override, Slack receives the rendered Chat step body as plain text, which is enough for summaries built from [digest variables](/platform/workflow/add-notification-content/personalize-content#digest-variables). Step overrides and trigger overrides are combined rather than chosen between. See [How Slack overrides are combined](#how-slack-overrides-are-combined). @@ -902,10 +903,6 @@ This renders as plain text in Slack. It does not produce Block Kit formatting su Save a Slack override on the Chat step and every trigger of that workflow sends it, with no change to your trigger call. The override is stored on the step, so it is versioned and promoted between environments along with the rest of the workflow. - - Provider content overrides in the workflow editor are rolling out gradually and may not be available on your Chat step yet. Overrides passed at trigger time work regardless. - - In the [Novu Dashboard](https://dashboard.novu.co), open your workflow and select the Chat step that uses the Slack integration. diff --git a/docs/platform/integrations/chat/whats-app.mdx b/docs/platform/integrations/chat/whats-app.mdx index 791df375756..86491f606a8 100644 --- a/docs/platform/integrations/chat/whats-app.mdx +++ b/docs/platform/integrations/chat/whats-app.mdx @@ -284,7 +284,7 @@ curl --location 'https://api.novu.co/v1/events/trigger' \ ## Configure WhatsApp overrides -Save a WhatsApp Business override on the Chat step when you need Cloud API message shapes beyond the plain-text step body — for example templates, interactive buttons, or media. The override is stored on the step, so it is versioned and promoted with the workflow, and every trigger sends it without changing your trigger call. +Save a WhatsApp Business override on the Chat step when you need Cloud API message shapes beyond the shared Chat step content (text editor body or block/card fallback text), for example templates, interactive buttons, or media. The override is stored on the step, so it is versioned and promoted with the workflow, and every trigger sends it without changing your trigger call. Provider content overrides in the workflow editor are rolling out gradually and may not be available on your Chat step yet. Overrides passed at trigger time work regardless. See [provider content overrides](/platform/integrations/chat#provider-content-overrides). diff --git a/docs/platform/integrations/chat/writing-chat-template.mdx b/docs/platform/integrations/chat/writing-chat-template.mdx new file mode 100644 index 00000000000..b467be05430 --- /dev/null +++ b/docs/platform/integrations/chat/writing-chat-template.mdx @@ -0,0 +1,124 @@ +--- +title: 'How to Write Chat Templates in Novu' +sidebarTitle: 'Writing Chat Template' +description: 'Compose chat notifications in the Novu Dashboard with the block editor or text editor. Use text, images, buttons, and provider previews while keeping existing plain-text steps working.' +--- + +You can compose Chat step content in the Novu Dashboard with either the **block editor** or the **text editor**. + +- **Block editor**: Visual blocks for structured messages (text, images, buttons, lists). Best when you want a shared rich message that Novu maps to each chat provider. +- **Text editor**: Plain text and [Liquid](/platform/workflow/add-notification-content/personalize-content#apply-logic-with-liquidjs) for variables, conditions, and loops. Best when you already rely on Liquid-heavy templates, or when you need full control over a single string body. + +Both editors support variables and notification preview. For code-first workflows, see the [Framework chat step](/framework/typescript/steps/chat). + +## Choose an editor + +Open a Chat step in the workflow editor. Use the **Block editor** / **Text editor** control to switch modes. + +| Editor | When to use it | What you author | +| --- | --- | --- | +| Block editor | New rich messages, cards with images and link buttons, provider-aware previews | Structured blocks, saved as block JSON | +| Text editor | Existing Liquid templates, digests built as a single string, provider-agnostic plain text | A string body with Liquid | + +### Backward compatibility + +Existing Chat steps keep working without changes. + +- Steps that already use a plain-text or Liquid body open in the **Text editor**. +- Delivery for those steps stays the same. +- New empty Chat steps default to the **Block editor**. +- Novu does not auto-convert Liquid-heavy text templates into blocks. Keep those steps on **Text editor**, or rebuild the content in **Block editor** if you want the structured format. + + + Switching from **Block editor** to **Text editor** (or the other way) replaces the step content for that editor mode. Copy anything you need before switching if you are experimenting. + + +## Block editor + +The block editor is a visual composer for chat cards. Add a block with the plus (`+`) control or by typing `/` in the editor. Choose a block from the menu, then edit its fields inline. + +### Supported blocks + +| Block | Description | +| --- | --- | +| **Text** | Message copy. Supports variables. | +| **Image** | Image from an absolute URL. Optional alt text. | +| **Button** | Link button with a label and redirect URL. Variables are supported in the label and URL. | +| **Divider** | Horizontal separator between sections. | +| **Bullet list** / **Numbered list** | Structured lists. | +| **Blockquote** | Quoted text. | +| **Hard break** | Line break. | +| **Repeat** | Iterate over an array from the payload or digest, same idea as the [email Repeat block](/platform/integrations/email/writing-email-template#repeat-block). | +| **Digest** | Insert digest summary content when a digest step runs before the Chat step. | + +Buttons in the Dashboard block editor are **link buttons** only. They open a URL. Interactive actions that call back into your app or an agent are not part of this editor. + + + Button URLs must be valid absolute URLs (for example `https://novu.co`). Values without a protocol can fail delivery on providers such as Slack. + + +### Variables and personalization + +Use the variable picker or Liquid-style `{{ ... }}` placeholders inside text, button labels, and button URLs. The same subscriber, payload, and digest variables work as in other channel editors. See [Personalize notification content](/platform/workflow/add-notification-content/personalize-content). + +### Preview + +The Chat step preview gives you an approximate view of the message for providers configured in your environment, such as Slack or Microsoft Teams. The delivered message can look different because each provider has its own layout, spacing, and supported features. + +Use the provider switcher in the preview panel to inspect each rendering. Some providers support native rich layouts (Slack Block Kit, Microsoft Teams Adaptive Cards). Others receive a text or Markdown-style fallback derived from the same card. + +If a block cannot be represented fully on a provider, the preview may surface a warning. Treat that as a signal to simplify the card or add a [provider content override](/platform/integrations/chat#provider-content-overrides) for that provider. + +## Text editor + +The text editor is a single message body field. Write plain text or Liquid: + +```liquid +{{ subscriber.firstName }}, your order {{ payload.orderId }} has shipped. +``` + +Use this mode when: + +- The step already uses Liquid conditionals or loops that you do not want to rebuild as blocks. +- You only need a short plain-text notification. +- You plan to supply the rich format through a [provider content override](/platform/integrations/chat#provider-content-overrides) (for example Slack Block Kit) and keep the default body as a fallback string. + +## Rich chat content vs provider overrides + +These layers solve different problems: + +| Approach | What it is | Use when | +| --- | --- | --- | +| **Block editor (default content)** | Novu card model, authored once, rendered per provider | You want one shared rich message across chat providers | +| **Provider content override** | Provider-native JSON saved on the step (Slack Block Kit, WhatsApp fields, and so on) | You need a format or field that only that provider supports | +| **Trigger overrides** | Provider payload passed at trigger time | The shape depends on runtime data outside the workflow definition | + +The block editor does not replace provider overrides. Overrides still apply for provider-specific APIs when you set them. If both exist, follow the precedence rules in [Provider content overrides](/platform/integrations/chat#provider-content-overrides) and [Trigger overrides](/platform/integrations/trigger-overrides). + +Use provider content overrides when you need more control over one provider's payload than the shared blocks offer. The block editor supports variables, but it does not support Liquid conditions or loops. Use the text editor when you need Liquid logic. + +## Code-first and Framework + +If you define the Chat step in code with `@novu/framework`, return a `body` string or a `card` object. If you return both, Novu uses `card`. The Dashboard block editor covers the common card subset. Framework also supports sections, fields, tables, and other layout elements. + +See: + +- [Framework chat step](/framework/typescript/steps/chat) +- [Code steps](/platform/workflow/add-and-configure-steps/code-steps) for publishing a handler from the Dashboard **Custom Code** toggle + +## Related + + + + Delivery flow, credentials, and provider overrides. + + + How template editors work across channels. + + + `body` and `card` output for code-first workflows. + + + Slack setup and Block Kit overrides when you need Slack-native payloads. + + diff --git a/docs/platform/workflow/add-and-configure-steps.mdx b/docs/platform/workflow/add-and-configure-steps.mdx index c6337f771b1..b6e6407f48e 100644 --- a/docs/platform/workflow/add-and-configure-steps.mdx +++ b/docs/platform/workflow/add-and-configure-steps.mdx @@ -51,7 +51,7 @@ Supported channel types include: title="Chat" icon="messages-square" href="/platform/integrations/chat" - description="Send notifications to chat platforms such as Slack or Microsoft Teams." + description="Send chat notifications with plain text or rich cards (Slack, Microsoft Teams, and more)." /> diff --git a/docs/platform/workflow/add-and-configure-steps/code-steps.mdx b/docs/platform/workflow/add-and-configure-steps/code-steps.mdx index e046a92ee47..107a74d9029 100644 --- a/docs/platform/workflow/add-and-configure-steps/code-steps.mdx +++ b/docs/platform/workflow/add-and-configure-steps/code-steps.mdx @@ -14,7 +14,7 @@ You can create both UI-managed steps and code-managed steps within the same work | Email | `subject`, `body` (HTML string) | | SMS | `body` | | Push | `subject`, `body` | -| Chat | `body` | +| Chat | `body` or `card` (see [Chat step](/framework/typescript/steps/chat)) | | In-App | `subject`, `body` (plus optional `avatar`, `primaryAction`, `secondaryAction`, `data`, `redirect`) | | Tool | `body` | diff --git a/docs/platform/workflow/add-notification-content/channels-template-editors.mdx b/docs/platform/workflow/add-notification-content/channels-template-editors.mdx index cc6a158e94b..213ca42c564 100644 --- a/docs/platform/workflow/add-notification-content/channels-template-editors.mdx +++ b/docs/platform/workflow/add-notification-content/channels-template-editors.mdx @@ -1,6 +1,6 @@ --- title: 'Channels Template Editors' -description: "Design and configure notification content for email, in-app, push, and SMS channels using the Novu template editors in the dashboard." +description: "Design and configure notification content for email, in-app, push, SMS, and chat channels using the Novu template editors in the dashboard." --- @@ -8,7 +8,8 @@ The channel template editors are where you define the content of the notificatio Each channel exposes an editor aligned with how messages are delivered on that channel: -- **SMS and chat**: Body field only +- **SMS**: Body field only +- **Chat**: Block editor for rich cards, or a text editor for plain text / Liquid - **Push, email, and in-app**: Subject and body fields All editors (including the Subject field editors) support variables and dynamic data that can be used to [personalize notification content](/platform/workflow/add-notification-content/personalize-content). Notification preview is also supported on all editors. @@ -99,6 +100,12 @@ The data object is a customizable key-value store available in the in-app step. To learn more about how to use the data object to customize how messages appear in the [``](/platform/inbox), refer to the [Data Object](/platform/inbox/configuration/data-object) documentation. +## Chat template editor + +The Chat template editor supports a **block editor** (structured cards) and a **text editor** (plain text / Liquid). Existing text steps stay on the text editor, and new empty steps default to the block editor. + +For supported blocks, preview, backward compatibility, and how this relates to provider overrides, see [Writing chat templates](/platform/integrations/chat/writing-chat-template). + ## Email template editor The email template editor defines how email notifications are structured, rendered, and delivered. Novu dashboard provides multiple authoring modes: From 5311d4a988f90ca8167b39ad5f553a5fff5611b2 Mon Sep 17 00:00:00 2001 From: Adam Chmara Date: Thu, 20 Aug 2026 17:05:53 +0200 Subject: [PATCH 3/7] feat(dashboard): redesign agent chat onboarding and preview drawer fixes NV-8612 (#12390) --- .../light/square/novu-agent-chat.svg | 10 +- .../agent-chat-panel/agent-chat-drawer.tsx | 92 ++++ .../agent-chat-panel/agent-chat-panel.tsx | 220 ++++----- .../agent-chat-panel/agent-chat-parts.tsx | 444 ++++++++++-------- .../agents/agent-chat-panel/index.ts | 1 + .../agents/agent-chat-setup-content.tsx | 16 +- .../agents/agent-chat-setup-guide.tsx | 40 +- .../agents/agent-chat-setup-steps.tsx | 64 ++- .../agent-chat-agent-integration-guide.tsx | 23 +- .../components/agents/agent-setup-steps.tsx | 8 +- .../agents/is-agent-integration-connected.ts | 9 +- .../agents/provider-card-display-state.ts | 2 +- .../src/components/agents/provider-cards.tsx | 47 +- .../agents/setup-guide-primitives.tsx | 3 + .../connect-agent/prebuilt-prompt-banner.tsx | 124 +++-- .../src/hooks/use-agent-chat-preview.ts | 68 +++ .../src/hooks/use-agent-chat-prompt.ts | 19 +- apps/dashboard/src/pages/agent-details.tsx | 61 ++- apps/dashboard/src/utils/routes.ts | 5 +- .../src/utils/agent-chat-connect-prompt.ts | 37 +- 20 files changed, 787 insertions(+), 506 deletions(-) create mode 100644 apps/dashboard/src/components/agents/agent-chat-panel/agent-chat-drawer.tsx create mode 100644 apps/dashboard/src/hooks/use-agent-chat-preview.ts diff --git a/apps/dashboard/public/images/providers/light/square/novu-agent-chat.svg b/apps/dashboard/public/images/providers/light/square/novu-agent-chat.svg index 9de3bff3a40..19b4e50afcd 100644 --- a/apps/dashboard/public/images/providers/light/square/novu-agent-chat.svg +++ b/apps/dashboard/public/images/providers/light/square/novu-agent-chat.svg @@ -1,9 +1,9 @@ - + + - - - + + + - diff --git a/apps/dashboard/src/components/agents/agent-chat-panel/agent-chat-drawer.tsx b/apps/dashboard/src/components/agents/agent-chat-panel/agent-chat-drawer.tsx new file mode 100644 index 00000000000..346499ce6e2 --- /dev/null +++ b/apps/dashboard/src/components/agents/agent-chat-panel/agent-chat-drawer.tsx @@ -0,0 +1,92 @@ +import { RiArrowRightUpLine, RiCloseLine } from 'react-icons/ri'; +import type { AgentResponse } from '@/api/agents'; +import { AgentChatPanel } from '@/components/agents/agent-chat-panel/agent-chat-panel'; +import { AGENT_CHAT_DOCS_URL } from '@/components/agents/agent-chat-setup-content'; +import { CursorPromptActions } from '@/components/onboarding/connect-agent/prebuilt-prompt-banner'; +import { CompactButton } from '@/components/primitives/button-compact'; +import { Sheet, SheetClose, SheetContent, SheetDescription, SheetTitle } from '@/components/primitives/sheet'; +import { VisuallyHidden } from '@/components/primitives/visually-hidden'; +import { useAgentChatPrompt } from '@/hooks/use-agent-chat-prompt'; + +type AgentChatDrawerProps = { + open: boolean; + onOpenChange: (open: boolean) => void; + agent: AgentResponse; + /** Hide once the customer's app has sent a first inbound message (`connectedAt`). */ + showAddToAppCallouts?: boolean; + /** Channels tab for this agent's web-chat integration. */ + addToAppHref?: string; +}; + +export function AgentChatDrawer({ + open, + onOpenChange, + agent, + showAddToAppCallouts = false, + addToAppHref, +}: AgentChatDrawerProps) { + const prompt = useAgentChatPrompt(agent); + + return ( + + event.preventDefault()} + > +
+
+
+ + Web chat preview +
+

+ Send a message to see how it replies.{' '} + + Read docs + + +

+ + Preview this agent in web chat before adding it to your app. + +
+ window.open(AGENT_CHAT_DOCS_URL, '_blank', 'noopener,noreferrer')} + /> + + + +
+ +
+ {showAddToAppCallouts ? ( +
+ +

+ Add Web Chat to your app +

+ +
+ ) : null} + + +
+ + + ); +} diff --git a/apps/dashboard/src/components/agents/agent-chat-panel/agent-chat-panel.tsx b/apps/dashboard/src/components/agents/agent-chat-panel/agent-chat-panel.tsx index 5756d108f7a..9967a4f51b4 100644 --- a/apps/dashboard/src/components/agents/agent-chat-panel/agent-chat-panel.tsx +++ b/apps/dashboard/src/components/agents/agent-chat-panel/agent-chat-panel.tsx @@ -1,36 +1,43 @@ import { NovuProvider, useAgentChat } from '@novu/react'; import { buildDashboardAgentChatSubscriberId } from '@novu/shared'; import { useLayoutEffect, useMemo, useRef, useState } from 'react'; -import { RiArrowUpLine, RiCodeSSlashLine, RiErrorWarningLine, RiLoader4Line } from 'react-icons/ri'; -import { useLocation, useNavigate } from 'react-router-dom'; +import { RiArrowUpLine, RiCloseFill, RiErrorWarningLine, RiLoader4Line } from 'react-icons/ri'; +import { Link } from 'react-router-dom'; import type { AgentResponse } from '@/api/agents'; -import { - ChatEmptyState, - ChatMessageRow, - ChatPendingActionCard, - ChatTypingRow, -} from '@/components/agents/agent-chat-panel/agent-chat-parts'; +import { ChatEmptyState, ChatMessageRow, ChatTypingRow } from '@/components/agents/agent-chat-panel/agent-chat-parts'; import { Button } from '@/components/primitives/button'; -import { Kbd } from '@/components/primitives/kbd'; import { Skeleton } from '@/components/primitives/skeleton'; +import { Tooltip, TooltipContent, TooltipTrigger } from '@/components/primitives/tooltip'; import { useAuth } from '@/context/auth/hooks'; import { useEnvironment } from '@/context/environment/hooks'; -import { useAgentRoutes } from '@/hooks/use-agent-routes'; import { apiHostnameManager } from '@/utils/api-hostname-manager'; -import { buildRoute } from '@/utils/routes'; import { cn } from '@/utils/ui'; const COMPOSER_MAX_HEIGHT_PX = 128; +const CHANNEL_PROMO_ICONS = [ + { src: '/images/providers/light/square/slack.svg', label: 'Slack' }, + { src: '/images/providers/light/square/msteams.svg', label: 'Microsoft Teams' }, + { src: '/images/providers/light/square/whatsapp-business.svg', label: 'WhatsApp' }, + { src: '/images/providers/light/square/imessages.svg', label: 'Messages' }, +] as const; + +const PREVIEW_STRIPE_STYLE = { + backgroundImage: [ + 'linear-gradient(to top, rgba(255,255,255,0) 20%, #fff 85%)', + 'repeating-linear-gradient(-38deg, rgba(255,132,71,0.11) 0px, rgba(255,132,71,0.11) 1.5px, rgba(255,132,71,0.07) 1.5px, rgba(255,132,71,0.07) 5px)', + ].join(', '), +} as const; + type AgentChatPanelProps = { agent: AgentResponse; - agentChatIntegrationIdentifier?: string; + showAddToAppCallouts?: boolean; + addToAppHref?: string; }; -export function AgentChatPanel({ agent, agentChatIntegrationIdentifier }: AgentChatPanelProps) { +export function AgentChatPanel({ agent, showAddToAppCallouts = false, addToAppHref }: AgentChatPanelProps) { const { currentUser, isUserLoaded } = useAuth(); const { currentEnvironment } = useEnvironment(); - const testerName = currentUser?.firstName?.trim() || 'yourself'; const testerSubscriberId = currentUser?._id ? buildDashboardAgentChatSubscriberId(currentUser._id) : ''; const isReady = isUserLoaded && Boolean(testerSubscriberId) && Boolean(currentEnvironment?.identifier); const subscriber = useMemo( @@ -46,9 +53,9 @@ export function AgentChatPanel({ agent, agentChatIntegrationIdentifier }: AgentC if (!isReady || !currentEnvironment) { return ( -
+
- +
); } @@ -61,12 +68,12 @@ export function AgentChatPanel({ agent, agentChatIntegrationIdentifier }: AgentC apiUrl={apiHostnameManager.getHostname()} socketUrl={apiHostnameManager.getWebSocketHostname()} > -
+
@@ -76,15 +83,16 @@ export function AgentChatPanel({ agent, agentChatIntegrationIdentifier }: AgentC function AgentChatSurface({ agentId, agentName, - testerName, - agentChatIntegrationIdentifier, + showAddToAppCallouts, + addToAppHref, }: { agentId: string; agentName: string; - testerName: string; - agentChatIntegrationIdentifier?: string; + showAddToAppCallouts: boolean; + addToAppHref?: string; }) { const [draft, setDraft] = useState(''); + const [showChannelPromo, setShowChannelPromo] = useState(true); const textareaRef = useRef(null); const scrollRef = useRef(null); const { messages, pendingActions, sendMessage, sendAction, respondToAction, error, isRunning, isLoading, typing } = @@ -96,7 +104,7 @@ function AgentChatSurface({ const canSend = !composerDisabled && Boolean(draft.trim()); const isEmpty = messages.length === 0 && !isRunning && !isLoading; const lastMessage = messages[messages.length - 1]; - const showTypingRow = Boolean(typing) || isRunning; + const showTypingRow = (Boolean(typing) || isRunning) && pendingActions.length === 0; const lastMessageSignature = lastMessage ? `${lastMessage.id}:${lastMessage.parts.map((part) => (part.type === 'text' ? part.text : part.type)).join('\0')}` : ''; @@ -110,6 +118,7 @@ function AgentChatSurface({ el.style.height = `${Math.min(el.scrollHeight, COMPOSER_MAX_HEIGHT_PX)}px`; }, [draft]); + // biome-ignore lint/correctness/useExhaustiveDependencies: scroll when the transcript changes useLayoutEffect(() => { const el = scrollRef.current; if (!el || isEmpty) return; @@ -127,28 +136,22 @@ function AgentChatSurface({ }; return ( -
-
-

- Chatting as {testerName} -

- -
- +
{isEmpty ? ( -
+
) : ( -
- {messages.map((message, index) => ( +
+ {messages.map((message) => ( void sendAction(action)} + onRespondToAction={(action) => void respondToAction(action)} /> ))} {showTypingRow ? : null} @@ -156,17 +159,8 @@ function AgentChatSurface({ )}
-
-
- {pendingActions.map((action) => ( - void respondToAction({ actionId: action.id, decision })} - /> - ))} - +
+
{error ? (
+ {showChannelPromo ? ( +
+ + +
+

+ Talk to your agent from wherever you work +

+
+ {CHANNEL_PROMO_ICONS.map((icon) => ( + + ))} +
+
+ + + Your users can natively connect to Slack, Teams, WhatsApp, and more to talk to this agent. + + + +
+ ) : null} +
+
+