Merge remote-tracking branch 'origin/develop' into PM-5747

# Conflicts:
#	src/api/submission/submission.controller.ts
#	src/api/submission/submission.service.ts

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.controller.ts

@@ -405,7 +405,7 @@ export class SubmissionController {
   @ApiOperation({
     summary: 'Download the submission',
     description:
-      'Roles: Copilot, Admin, User, Reviewer. After challenge completion, the exact metadata value allowAllRegistrantsToDownloadWinningSubmissions=true lets every registered Submitter download only an exact final winning submission and denies non-winners without legacy fallback. Other values require passing-submission eligibility, except non-Design First2Finish challenges retain legacy submitter eligibility. Design challenges also require submissionsViewable. | Scopes: read:submission',
+      'Roles: Copilot, Admin, User, Reviewer. After challenge completion, the exact metadata value allowAllRegistrantsToDownloadWinningSubmissions=true lets every registered Submitter download only an exact final winning submission and denies non-winners without legacy fallback. Other values require passing-submission eligibility, except non-Design First2Finish challenges retain legacy submitter eligibility. | Scopes: read:submission',
   })
   @ApiParam({
     name: 'submissionId',

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 () => {
@@ -1269,7 +1272,7 @@ describe('SubmissionService', () => {
       expect(s3Send).not.toHaveBeenCalled();
     });
 
-    it('lets the Design visibility gate block registrants even when the new flag is enabled', async () => {
+    it('allows all Design registrants to download winner submissions when legacy visibility is disabled', async () => {
       resourceApiService.getMemberResourcesRoles.mockResolvedValue([
         { roleName: 'Submitter' },
       ]);
@@ -1282,66 +1285,6 @@ describe('SubmissionService', () => {
         },
         winners: [{ userId: 'owner-user', placement: 1 }],
       });
-      prismaMock.submission.findFirst.mockResolvedValue({ id: 'passing-sub' });
-
-      await expect(
-        service.getSubmissionFileStream(
-          {
-            userId: 'registered-user',
-            isMachine: false,
-            roles: [],
-          } as any,
-          'sub-123',
-        ),
-      ).rejects.toBeInstanceOf(ForbiddenException);
-
-      expect(prismaMock.submission.findFirst).not.toHaveBeenCalled();
-      expect(s3Send).not.toHaveBeenCalled();
-    });
-
-    it('applies the Design visibility gate to legacy Design tracks', async () => {
-      resourceApiService.getMemberResourcesRoles.mockResolvedValue([
-        { roleName: 'Submitter' },
-      ]);
-      challengeApiServiceMock.getChallengeDetail.mockResolvedValue({
-        status: ChallengeStatus.COMPLETED,
-        track: 'Legacy',
-        legacy: { track: 'DESIGN' },
-        metadata: {
-          submissionsViewable: 'false',
-          allowAllRegistrantsToDownloadWinningSubmissions: 'true',
-        },
-        winners: [{ userId: 'owner-user', placement: 1 }],
-      });
-
-      await expect(
-        service.getSubmissionFileStream(
-          {
-            userId: 'registered-user',
-            isMachine: false,
-            roles: [],
-          } as any,
-          'sub-123',
-        ),
-      ).rejects.toBeInstanceOf(ForbiddenException);
-
-      expect(prismaMock.submission.findFirst).not.toHaveBeenCalled();
-      expect(s3Send).not.toHaveBeenCalled();
-    });
-
-    it('allows all Design registrants to download winner submissions when both gates are enabled', async () => {
-      resourceApiService.getMemberResourcesRoles.mockResolvedValue([
-        { roleName: 'Submitter' },
-      ]);
-      challengeApiServiceMock.getChallengeDetail.mockResolvedValue({
-        status: ChallengeStatus.COMPLETED,
-        track: 'Design',
-        metadata: {
-          submissionsViewable: 'true',
-          allowAllRegistrantsToDownloadWinningSubmissions: 'true',
-        },
-        winners: [{ userId: 'owner-user', placement: 1 }],
-      });
       prismaMock.challengeResult.findUnique.mockResolvedValue({
         submissionId: 'sub-123',
         userId: 'owner-user',
@@ -1362,15 +1305,15 @@ describe('SubmissionService', () => {
       expect(s3Send).toHaveBeenCalledTimes(2);
     });
 
-    it('uses passing-submitter eligibility for viewable Design challenges when the new flag is disabled', async () => {
+    it('uses passing-submitter eligibility for Design challenges when legacy visibility is disabled', async () => {
       resourceApiService.getMemberResourcesRoles.mockResolvedValue([
         { roleName: 'Submitter' },
       ]);
       challengeApiServiceMock.getChallengeDetail.mockResolvedValue({
         status: ChallengeStatus.COMPLETED,
         track: 'Design',
         metadata: {
-          submissionsViewable: 'true',
+          submissionsViewable: 'false',
           allowAllRegistrantsToDownloadWinningSubmissions: 'false',
         },
         winners: [{ userId: 'owner-user', placement: 1 }],
@@ -1391,15 +1334,15 @@ describe('SubmissionService', () => {
       expect(s3Send).toHaveBeenCalledTimes(2);
     });
 
-    it('denies a non-passing submitter for a viewable Design challenge when the new flag is disabled', async () => {
+    it('denies a non-passing Design submitter when the new flag is disabled regardless of legacy visibility', async () => {
       resourceApiService.getMemberResourcesRoles.mockResolvedValue([
         { roleName: 'Submitter' },
       ]);
       challengeApiServiceMock.getChallengeDetail.mockResolvedValue({
         status: ChallengeStatus.COMPLETED,
         track: 'Design',
         metadata: {
-          submissionsViewable: 'true',
+          submissionsViewable: 'false',
           allowAllRegistrantsToDownloadWinningSubmissions: 'false',
         },
         winners: [{ userId: 'owner-user', placement: 1 }],
@@ -1942,7 +1885,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);
     });