Merge pull request #310 from topcoder-platform/PM-5716

Submission confirmation email fixes

Author: jmgasper

docs/KAFKA_SETUP.md

@@ -109,8 +109,12 @@ docker exec -it kafka kafka-console-consumer --topic avscan.action.scan --from-b
    }
    ```
 
-2. Register the handler in the `src/shared/modules/kafka/handlers/registered-handlers.config.ts` config handlers array.
-3. The handler will automatically be registered and start consuming messages.
+2. Add the handler to
+   `src/shared/modules/kafka/handlers/registered-handlers.config.ts` so the
+   handler catalog remains complete.
+3. Add the handler class to `GlobalProvidersModule.providers`. Nest must
+   instantiate the provider for its `onModuleInit` registration to run.
+4. The registered topic is included when the Kafka consumer starts.
 
 ### Dead Letter Queue (DLQ) Support
 
@@ -127,9 +131,13 @@ The application includes a robust Dead Letter Queue implementation for handling
 
 2. **Retry Mechanism**:
 
-   - Failed messages are automatically retried up to the configured maximum number of retries
+   - Failed messages are automatically retried up to the configured maximum
+     number of retries, whether or not DLQ publication is enabled
    - Retry count is tracked per message using a unique key based on topic, partition, and offset
    - Exponential backoff is applied between retries
+   - `KAFKA_DLQ_ENABLED` controls whether an event is copied to a DLQ after the
+     retry budget is exhausted; exhausted messages are logged and committed
+     when DLQ publication is disabled
 
 3. **DLQ Processing**:
 
@@ -156,6 +164,12 @@ The application includes a robust Dead Letter Queue implementation for handling
 
 - The service uses `@platformatic/kafka` 2.8.0 for broker failover and consumer group recovery fixes.
 - Platformatic Kafka 2.x raises the aggregate consumer Fetch limit to 50 MiB. The service deliberately retains the previous 10 MiB `maxBytes` limit to avoid increasing its per-consumer memory envelope.
+- Streams use committed-offset mode so restarts resume the consumer group's
+  last successful offset. A topic without a committed offset starts at latest,
+  which avoids replaying historical events when a handler is first deployed.
+- Submission confirmation additionally persists a request in the same database
+  operation as each normal member submission. Its scheduled recovery pass does
+  not depend on Kafka redelivery, so a missed source event remains recoverable.
 - Terminal consumer or producer client errors and offset commit timeouts mark Kafka health as `reconnecting` and start the shared reconnect lifecycle. A successful reconnect returns health to `ready`; exhausted attempts mark it as `failed` with the last failure reason.
 
 ### Environment Variables

docs/MANUAL_UPLOAD_FLOW.md

@@ -40,14 +40,21 @@ sequenceDiagram
     participant S3DMZ as S3 DMZ bucket
     participant AV as AV scan pipeline
     participant Kafka as Kafka
+    participant Email as email-service-v6
+    participant Member
     participant AI as AI workflow queue
     participant ReviewDB as review DB
 
     Client->>ReviewAPI: POST /submissions/manual-upload
     ReviewAPI->>ReviewAPI: Validate auth, file, submitter, challenge, phase window
     ReviewAPI->>S3DMZ: Upload file to DMZ
-    ReviewAPI->>ReviewDB: Create submission row
+    ReviewAPI->>ReviewDB: Create submission + confirmation request atomically
     ReviewAPI->>Kafka: Publish avscan.action.scan
+    ReviewAPI->>Kafka: Publish submission.notification.create
+    Kafka->>ReviewAPI: submission.notification.create
+    ReviewAPI->>Kafka: Publish submission.notification.send
+    Kafka->>Email: submission.notification.send
+    Email-->>Member: Submission confirmation
     AV->>Kafka: Publish submission.scan.complete
     Kafka->>ReviewAPI: submission.scan.complete
     ReviewAPI->>ReviewDB: Update submission URL + virusScan=true
@@ -127,6 +134,11 @@ The created submission row is initially stored with:
 - `eventRaised = false`
 - `url = <DMZ URL>`
 
+The same database operation also creates one
+`submissionConfirmationEmail` request with no `publishedAt` timestamp. This
+request is the durable recovery source for the member receipt; historical and
+validation-only submissions are not automatically added to it.
+
 At this point the file is not yet considered clean.
 
 ### 6. Events emitted immediately after creation
@@ -147,6 +159,55 @@ The AV scan event includes:
 - quarantine destination bucket
 - callback topic `submission.scan.complete`
 
+### 7. Submission receipt confirmation
+
+The Kafka handler for `submission.notification.create` asks the durable
+confirmation dispatcher to claim the related request. The dispatcher re-reads
+the persisted submission instead of trusting member or challenge fields in the
+event. It then:
+
+- resolves the submitter email and handle from member storage
+- resolves the challenge title from challenge storage
+- publishes `submission.notification.send` with Kafka key
+  `submission-confirmation:{submissionId}`
+- records the request's `publishedAt` only after Bus API accepts that event
+
+A once-per-minute recovery pass selects unpublished requests. A five-minute
+database lease with an opaque per-claim token prevents the Kafka handler,
+multiple ECS tasks, and the recovery pass from concurrently publishing the same
+request. A stale worker cannot release a newer worker's lease. If review-api
+stops after the submission is stored but before
+`submission.notification.create` reaches Kafka, the recovery pass still
+publishes the confirmation from the durable request.
+
+The email-ready payload uses the `v3` dynamic-template shape:
+
+```json
+{
+  "recipients": ["member@example.com"],
+  "version": "v3",
+  "data": {
+    "submitter": { "handle": "memberHandle" },
+    "challenge": { "challengeTitle": "Challenge title" },
+    "submission": {
+      "id": "submission-id",
+      "challengeId": "challenge-id"
+    }
+  }
+}
+```
+
+The email service selects its configured template from the
+`submission.notification.send` topic, so review-api does not hardcode a
+SendGrid template ID. A populated request `publishedAt` suppresses ordinary
+replay after successful publication. It records Bus acceptance, not a SendGrid
+delivery receipt; a process failure after Bus acceptance but before the request
+is marked can still produce at-least-once delivery. Transient enrichment or
+publication failures release the lease and become eligible for scheduled
+recovery after five minutes, preventing permanently bad records from starving
+new requests. Keep `KAFKA_DLQ_ENABLED=true` to retain exhausted source events
+for operational inspection and replay.
+
 ## What Happens After AV Scan
 
 When `submission.scan.complete` is received:

prisma/migrations/20260730120000_add_submission_confirmation_email_publication/migration.sql

@@ -0,0 +1,24 @@
+-- Persist one recoverable confirmation request for each new member submission.
+-- The table is intentionally empty after migration; historical submissions are
+-- not backfilled or emailed without an explicit operational replay.
+CREATE TABLE "submissionConfirmationEmail" (
+    "submissionId" VARCHAR(14) NOT NULL,
+    "processingStartedAt" TIMESTAMP(3),
+    "processingToken" VARCHAR(36),
+    "publishedAt" TIMESTAMP(3),
+    "nextAttemptAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "attemptCount" INTEGER NOT NULL DEFAULT 0,
+    "lastError" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "submissionConfirmationEmail_pkey" PRIMARY KEY ("submissionId")
+);
+
+CREATE INDEX "submissionConfirmationEmail_pending_idx"
+ON "submissionConfirmationEmail"("publishedAt", "nextAttemptAt", "processingStartedAt", "createdAt");
+
+ALTER TABLE "submissionConfirmationEmail"
+ADD CONSTRAINT "submissionConfirmationEmail_submissionId_fkey"
+FOREIGN KEY ("submissionId") REFERENCES "submission"("id")
+ON DELETE CASCADE ON UPDATE CASCADE;

prisma/schema.prisma

@@ -447,6 +447,7 @@ model submission {
   aiReviewDecisions   aiReviewDecision[]
   auditLogs           reviewAudit[]
   accessAudits        submissionAccessAudit[]
+  confirmationEmail   submissionConfirmationEmail?
 
   @@index([memberId])
   @@index([challengeId])
@@ -455,6 +456,23 @@ model submission {
   @@index([submittedDate])
 }
 
+/// Durable submission-confirmation request created atomically with a member submission.
+model submissionConfirmationEmail {
+  submissionId        String    @id @db.VarChar(14)
+  processingStartedAt DateTime?
+  processingToken     String?   @db.VarChar(36)
+  publishedAt         DateTime?
+  nextAttemptAt       DateTime  @default(now())
+  attemptCount        Int       @default(0)
+  lastError           String?   @db.Text
+  createdAt           DateTime  @default(now())
+  updatedAt           DateTime  @updatedAt
+
+  submission submission @relation(fields: [submissionId], references: [id], onDelete: Cascade)
+
+  @@index([publishedAt, nextAttemptAt, processingStartedAt, createdAt], map: "submissionConfirmationEmail_pending_idx")
+}
+
 enum ReviewOpportunityStatus {
   OPEN
   CLOSED

src/api/submission/submission.service.spec.ts

@@ -377,6 +377,9 @@ describe('SubmissionService', () => {
           virusScan: true,
         }),
       });
+      expect(
+        prisma.submission.create.mock.calls[0][0].data,
+      ).not.toHaveProperty('confirmationEmail');
     });
 
     it('rejects validation upload requests without file contents', async () => {
@@ -1896,7 +1899,13 @@ describe('SubmissionService', () => {
         challengeApiServiceMock.validateFinalFixSubmissionCreation,
       ).toHaveBeenCalledWith('challenge-final-fix');
       expect(challengePrismaMock.$queryRaw).toHaveBeenCalled();
-      expect(prismaMock.submission.create).toHaveBeenCalled();
+      expect(prismaMock.submission.create).toHaveBeenCalledWith({
+        data: expect.objectContaining({
+          confirmationEmail: {
+            create: {},
+          },
+        }),
+      });
       expect(result.type).toBe(SubmissionType.STUDIO_FINAL_FIX_SUBMISSION);
     });
 

src/api/submission/submission.service.ts

@@ -2842,6 +2842,15 @@ export class SubmissionService {
     }
   }
 
+  /**
+   * Publishes the canonical submission-created event after persistence.
+   *
+   * @param submission Persisted submission fields used to build the public event payload.
+   * @returns A promise that resolves after Bus API accepts the event.
+   * @throws InternalServerErrorException when Bus API publication fails.
+   * Used by createSubmission to trigger downstream scanning, automation, and
+   * submission-confirmation consumers with stable per-submission ordering.
+   */
   private async publishSubmissionCreateEvent(
     submission: SubmissionBusPayloadSource,
   ): Promise<void> {
@@ -2886,6 +2895,7 @@ export class SubmissionService {
     await this.eventBusService.publish(
       'submission.notification.create',
       payload,
+      `submission:${submission.id}`,
     );
     this.logger.log(
       `Published submission.notification.create event for submission ${submission.id}`,
@@ -3446,6 +3456,9 @@ export class SubmissionService {
           type: body.type as SubmissionType,
           virusScan: false,
           eventRaised: false,
+          confirmationEmail: {
+            create: {},
+          },
         },
       });
       this.logger.log(`Submission created with ID: ${data.id}`);

src/shared/modules/global/eventBus.service.spec.ts

@@ -0,0 +1,46 @@
+import { HttpStatus } from '@nestjs/common';
+import { of } from 'rxjs';
+import { CommonConfig } from 'src/shared/config/common.config';
+import { EventBusService } from './eventBus.service';
+
+describe('EventBusService', () => {
+  it('forwards a stable Kafka key in the Bus API event envelope', async () => {
+    const m2mService = {
+      getM2MToken: jest.fn().mockResolvedValue('m2m-token'),
+    };
+    const httpService = {
+      post: jest.fn().mockReturnValue(
+        of({
+          status: HttpStatus.ACCEPTED,
+        }),
+      ),
+    };
+    const service = new EventBusService(
+      m2mService as any,
+      httpService as any,
+    );
+
+    await service.publish(
+      'submission.notification.send',
+      { recipients: ['member@example.com'] },
+      'submission-confirmation:submission-1',
+    );
+
+    expect(httpService.post).toHaveBeenCalledWith(
+      CommonConfig.apis.busApiUrl,
+      expect.objectContaining({
+        topic: 'submission.notification.send',
+        originator: 'review-api-v6',
+        'mime-type': 'application/json',
+        timestamp: expect.any(String),
+        payload: { recipients: ['member@example.com'] },
+        key: 'submission-confirmation:submission-1',
+      }),
+      {
+        headers: {
+          Authorization: 'Bearer m2m-token',
+        },
+      },
+    );
+  });
+});

src/shared/modules/global/eventBus.service.ts

@@ -18,9 +18,12 @@ class EventBusMessage<T> {
   'mime-type': string = 'application/json';
   timestamp: string = new Date().toISOString();
   payload: T;
+  key?: string;
 }
 
-// event bus send email payload
+/**
+ * Payload accepted by the legacy external.action.email topic.
+ */
 export class EventBusSendEmailPayload {
   // Template-specific variables payload. Structure depends on the sendgrid template.
   data: Record<string, any>;
@@ -40,10 +43,22 @@ export class EventBusService {
     private readonly httpService: HttpService,
   ) {}
 
+  /**
+   * Posts a message to Bus API using the review-api M2M identity.
+   *
+   * @param topic Event topic to publish.
+   * @param payload Topic-specific JSON payload.
+   * @param originator Service name recorded on the event.
+   * @param key Optional stable Kafka partitioning and correlation key.
+   * @returns A promise that resolves after Bus API accepts the event.
+   * @throws InternalServerErrorException when token acquisition or Bus API publication fails.
+   * Used by sendEmail and publish for all review-api event delivery.
+   */
   private async postMessage<T>(
     topic: string,
     payload: T,
     originator = 'review-api-v6',
+    key?: string,
   ): Promise<void> {
     // Get M2M token
     const token = await this.m2mService.getM2MToken();
@@ -52,6 +67,9 @@ export class EventBusService {
     msg.topic = topic;
     msg.originator = originator;
     msg.payload = payload;
+    if (key !== undefined) {
+      msg.key = key;
+    }
     // send message to event bus
     const url = CommonConfig.apis.busApiUrl;
     try {
@@ -80,14 +98,28 @@ export class EventBusService {
 
   /**
    * Send email message to Event bus.
-   * @param payload send email payload
+   *
+   * @param payload Legacy external.action.email payload, including its template ID.
+   * @returns A promise that resolves after Bus API accepts the event.
+   * @throws InternalServerErrorException when Bus API publication fails.
+   * Used by review-api features that select their SendGrid template directly.
    */
   async sendEmail(payload: EventBusSendEmailPayload): Promise<void> {
     console.log(`${JSON.stringify(payload, null, 2)}`);
     await this.postMessage('external.action.email', payload);
   }
 
-  async publish<T>(topic: string, payload: T): Promise<void> {
-    await this.postMessage(topic, payload);
+  /**
+   * Publishes a topic-specific event through Bus API.
+   *
+   * @param topic Event topic to publish.
+   * @param payload Topic-specific JSON payload.
+   * @param key Optional stable Kafka partitioning and correlation key.
+   * @returns A promise that resolves after Bus API accepts the event.
+   * @throws InternalServerErrorException when Bus API publication fails.
+   * Used by submission, scan, workflow, and notification event producers.
+   */
+  async publish<T>(topic: string, payload: T, key?: string): Promise<void> {
+    await this.postMessage(topic, payload, 'review-api-v6', key);
   }
 }