[pull] next from novuhq:next

.github/workflows/release-packages.yml

@@ -56,6 +56,21 @@ jobs:
           fetch-tags: true
           ref: ${{ github.ref_name }}
 
+      - name: Normalize and validate version
+        env:
+          RAW_VERSION: ${{ github.event.inputs.version }}
+        run: |
+          SEMVER=$(echo "$RAW_VERSION" | sed 's/^[vV]\.*//')
+          # Numeric identifiers must not have leading zeros; prerelease ids must be
+          # non-empty and dot-separated (rejects 3.1.0-rc., 3.1.0-rc..1, 3.1.0-01).
+          semver_re='^[0-9]+\.[0-9]+\.[0-9]+(-((0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*)(\.(0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*))*))?$'
+          if ! [[ "$SEMVER" =~ $semver_re ]]; then
+            echo "❌ ERROR: Invalid version '$RAW_VERSION'. Must be semver (e.g. v3.1.0 or 3.1.0), got cleaned: '$SEMVER'"
+            exit 1
+          fi
+          echo "RELEASE_SEMVER=$SEMVER" >> "$GITHUB_ENV"
+          echo "✅ Normalized version: v$SEMVER (from input: $RAW_VERSION)"
+
       - name: Install pnpm
         uses: pnpm/action-setup@0e279bb959325dab635dd2c09392533439d90093 # v6.0.8
         with:
@@ -84,16 +99,15 @@ jobs:
         if: github.event.inputs.release_type != 'stable'
         env:
           INPUT_RELEASE_TYPE: ${{ github.event.inputs.release_type }}
-          INPUT_VERSION: ${{ github.event.inputs.version }}
         run: |
           COMMIT_SHA=$(git rev-parse --short HEAD)
           if [ "$INPUT_RELEASE_TYPE" = "nightly" ]; then
             DATE=$(date +'%Y%m%d')
-            echo "RELEASE_VERSION=${INPUT_VERSION}-nightly.${DATE}.${COMMIT_SHA}" >> $GITHUB_ENV
-            echo "Using nightly version: $RELEASE_VERSION"
+            echo "RELEASE_VERSION=v${RELEASE_SEMVER}-nightly.${DATE}.${COMMIT_SHA}" >> $GITHUB_ENV
+            echo "Using nightly version: v${RELEASE_SEMVER}-nightly.${DATE}.${COMMIT_SHA}"
           elif [ "$INPUT_RELEASE_TYPE" = "rc" ]; then
-            echo "RELEASE_VERSION=${INPUT_VERSION}-rc.${COMMIT_SHA}" >> $GITHUB_ENV
-            echo "Using rc version: $RELEASE_VERSION"
+            echo "RELEASE_VERSION=v${RELEASE_SEMVER}-rc.${COMMIT_SHA}" >> $GITHUB_ENV
+            echo "Using rc version: v${RELEASE_SEMVER}-rc.${COMMIT_SHA}"
           fi
 
       - name: Configure Git
@@ -104,7 +118,6 @@ jobs:
       - name: Release version (without commit)
         env:
           INPUT_RELEASE_TYPE: ${{ github.event.inputs.release_type }}
-          INPUT_VERSION: ${{ github.event.inputs.version }}
           INPUT_PACKAGES: ${{ github.event.inputs.packages }}
         run: |
           if [ "$INPUT_RELEASE_TYPE" = "nightly" ]; then
@@ -114,19 +127,18 @@ jobs:
             echo "Running rc release with version: $RELEASE_VERSION"
             pnpm nx release version "$RELEASE_VERSION" --projects="$INPUT_PACKAGES" --preid rc --git-commit=false --verbose
           else
-            echo "Running stable release with version: $INPUT_VERSION"
-            pnpm nx release version "$INPUT_VERSION" --projects="$INPUT_PACKAGES" --git-commit=false --verbose
+            echo "Running stable release with version: v$RELEASE_SEMVER"
+            pnpm nx release version "v$RELEASE_SEMVER" --projects="$INPUT_PACKAGES" --git-commit=false --verbose
           fi
 
       - name: Generate changelog (without commit)
         env:
           INPUT_RELEASE_TYPE: ${{ github.event.inputs.release_type }}
-          INPUT_VERSION: ${{ github.event.inputs.version }}
           INPUT_PACKAGES: ${{ github.event.inputs.packages }}
           INPUT_PREVIOUS_TAG: ${{ github.event.inputs.previous_tag }}
         run: |
           if [ "$INPUT_RELEASE_TYPE" = "stable" ]; then
-            pnpm nx release changelog "$INPUT_VERSION" --projects="$INPUT_PACKAGES" --from="$INPUT_PREVIOUS_TAG" --git-commit=false
+            pnpm nx release changelog "v$RELEASE_SEMVER" --projects="$INPUT_PACKAGES" --from="$INPUT_PREVIOUS_TAG" --git-commit=false
           else
             pnpm nx release changelog "$RELEASE_VERSION" --projects="$INPUT_PACKAGES" --from="$INPUT_PREVIOUS_TAG" --git-commit=false
           fi
@@ -161,20 +173,20 @@ jobs:
         uses: peter-evans/create-pull-request@4e1beaa7521e8b457b572c090b25bd3db56bf1c5 # v5
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
-          commit-message: "chore: release ${{ github.event.inputs.version }} (${{ github.event.inputs.packages }})"
-          title: "🚀 Release ${{ github.event.inputs.version }} - ${{ github.event.inputs.packages }}"
+          commit-message: "chore: release v${{ env.RELEASE_SEMVER }} (${{ github.event.inputs.packages }})"
+          title: "🚀 Release v${{ env.RELEASE_SEMVER }} - ${{ github.event.inputs.packages }}"
           body: |
-            ## 🚀 Release ${{ github.event.inputs.version }}
+            ## 🚀 Release v${{ env.RELEASE_SEMVER }}
 
-            This PR contains the release changes for version **${{ github.event.inputs.version }}**
+            This PR contains the release changes for version **v${{ env.RELEASE_SEMVER }}**
 
             ### 📦 Packages Released:
             ```
             ${{ github.event.inputs.packages }}
             ```
 
             ### 🔄 Changes:
-            - Updated package versions to ${{ github.event.inputs.version }}
+            - Updated package versions to v${{ env.RELEASE_SEMVER }}
             - Generated changelogs from ${{ github.event.inputs.previous_tag }}
 
             **After merging this PR:**
@@ -183,7 +195,7 @@ jobs:
             - GitHub releases will be created
 
             **Please review the changes and merge when ready.**
-          branch: release-${{ github.event.inputs.version }}
+          branch: release-v${{ env.RELEASE_SEMVER }}
           base: ${{ github.ref_name }}
           delete-branch: false
           labels: automated-npm-release
@@ -211,10 +223,10 @@ jobs:
           echo ""
           echo "=== Extraction ==="
 
-          VERSION=$(echo "$PR_TITLE" | sed -n 's/.*Release \([v0-9\.]*\).*/\1/p')
+          RAW_VERSION=$(echo "$PR_TITLE" | sed -n 's/.*Release \([v0-9\.]*\).*/\1/p')
           PACKAGES=$(echo "$PR_TITLE" | sed -n 's/.*- \(@novu.*\)/\1/p')
 
-          if [ -z "$VERSION" ]; then
+          if [ -z "$RAW_VERSION" ]; then
             echo "❌ ERROR: Failed to extract version from PR title"
             echo "Expected format: '🚀 Release vX.Y.Z - @novu/package1,@novu/package2'"
             exit 1
@@ -226,6 +238,11 @@ jobs:
             exit 1
           fi
 
+          VERSION=$(echo "$RAW_VERSION" | sed 's/^[vV]\.*/v/')
+          if [ "$VERSION" = "v" ]; then
+            VERSION="$RAW_VERSION"
+          fi
+
           echo "version=$VERSION" >> $GITHUB_OUTPUT
           echo "packages=$PACKAGES" >> $GITHUB_OUTPUT
           echo "✅ Extracted version: $VERSION"

apps/api/src/app/agents/conversation-runtime/ingress/inbound-connection-context.resolver.spec.ts

@@ -43,9 +43,11 @@ function makeContextRepository(contexts: FakeContext[]) {
   const byKey = new Map(contexts.map((context) => [context.key, context]));
 
   return {
-    findByKeys: sinon.stub().callsFake(async (_env: string, _org: string, keys: string[]) =>
-      keys.map((key) => byKey.get(key)).filter((context): context is FakeContext => !!context)
-    ),
+    findByKeys: sinon
+      .stub()
+      .callsFake(async (_env: string, _org: string, keys: string[]) =>
+        keys.map((key) => byKey.get(key)).filter((context): context is FakeContext => !!context)
+      ),
   };
 }
 

apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.ts

@@ -17,9 +17,9 @@ import { AgentConversationService } from '../conversation/agent-conversation.ser
 import {
   buildWorkflowOriginSummary,
   extractAgentEmailOriginToken,
+  extractTeamsQuotedActivityId,
   extractTelegramChatIdFromThreadId,
   extractTelegramQuotedMessageId,
-  extractTeamsQuotedActivityId,
   extractWhatsAppQuotedWamid,
   isSendblueDirectThreadId,
   RECHECK_WORKFLOW_ORIGIN_PLATFORMS,

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
 

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,
     });
   }
 

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<CreateExecutionDetails>;
   let markNotificationAsMock: sinon.SinonStubbedInstance<MarkNotificationAs>;
   let getSubscriberMock: sinon.SinonStubbedInstance<GetSubscriber>;
+  let schedulerServiceMock: sinon.SinonStubbedInstance<EventBridgeSchedulerService>;
 
   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);

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');
+      });
+  }
 }