Fix auth issue

docs/api.md

@@ -23,9 +23,9 @@ Keys are lowercase snake_case, begin with a letter, and contain at most 40 chara
 
 ## Authentication
 
-Authorization is deny-by-default. Public routes may omit a token; supplying an invalid token still returns 401. Configure either HTTPS JWKS/RS256 or the legacy HS256 shared-secret mode. The two verification paths have separate algorithm allowlists. Signature, issuer, audience, expiry, issued-at presence, and subject are required.
+Authorization is deny-by-default. Public routes may omit a token; supplying an invalid token still returns 401. Configure either HTTPS JWKS/RS256 or the legacy HS256 shared-secret mode. The two verification paths have separate algorithm allowlists. Signature, trusted issuer, expiry, and issued-at presence are always required. Standard tokens also require the configured audience and a nonempty subject. In HS256 mode, legacy human tokens issued by `https://api.topcoder-dev.com` or `https://api.topcoder.com` may omit both `aud` and `sub` when they contain a positive numeric `userId`; the issuer must still appear in `VALID_ISSUERS`. Their verified `userId` supplies the audit subject. This exception does not apply to machine tokens, JWKS/RS256 tokens, or tokens containing only one of `aud` and `sub`.
 
-Roles are matched case-insensitively. Roles and userId are read from direct claims or the exact `AUTH_CLAIM_NAMESPACE` prefix, default `https://topcoder.com/`. Subject is retained for editor audit; member forms require an actual `userId` claim. A missing userId is not silently replaced with an Auth0 subject.
+Roles are matched case-insensitively. Roles and userId are read from direct claims or the exact `AUTH_CLAIM_NAMESPACE` prefix, default `https://topcoder.com/`. Subject (or the verified legacy `userId`) is retained for editor audit; member forms require an actual `userId` claim. A missing userId is not silently replaced with an Auth0 subject.
 
 | Caller    | Manage access                                 | Report access                            | Submit access                                           |
 | --------- | --------------------------------------------- | ---------------------------------------- | ------------------------------------------------------- |

docs/data-model.md

@@ -107,3 +107,31 @@ SQL views are computed rather than materialized. They are indexed through the su
 The service supports `DRAFT -> PUBLISHED -> RETIRED`. Published versions cannot return to draft; retired versions cannot be reopened. Create the next sequential revision to reopen or change a form. Draft content becomes immutable through the API when first saved; local Payload drafts can be edited freely until synchronized.
 
 Versions and historical reporting views are retained. The service does not impose an arbitrary retention period or add deletion endpoints. An approved retention policy can delete whole submission envelopes in batches; cascading foreign keys remove child data. Definition records remain for interpretation of retained historical data. Reports and database grants expose personal submission information only to their authorized readers.
+
+## Submission processing
+
+`ProcessorFlow`, `ProcessorEvent`, and `ProcessorDelivery` support forms-processor-v6.
+The API owns their migrations; the processor uses these schema-qualified tables
+without running DDL at startup. A flow has a stable ID, form key, enabled flag,
+AND-combined answer predicates (`rules.all`), action name, and JSON action settings.
+The seeded `lets-talk-sales-email` flow matches all `lets-talk` submissions and is
+disabled with TBD recipients, sender, and template. Manage it externally in SQL;
+set `updatedAt` when changing settings. Disabling delivery preserves queued work.
+
+The inbox stores the submitted event before Kafka commit. Routing records a unique
+receipt per submission/flow, including disabled matching flows. Retries read current
+action settings. No matching flow means the event is retained but has no actions.
+Flow additions/rule edits apply to events not yet routed; replay requires explicitly
+clearing `routedAt` and never deletes existing delivery receipts. Bus API email-event acceptance
+and the receipt cannot be atomic; a crash in between can cause a duplicate email.
+
+Processor events contain answers and must be included in personal-data retention
+and erasure procedures separately from `Submission`. Deleting a ProcessorEvent
+cascades its delivery records and removes deduplication protection, so retain it
+through the Kafka retention/replay window. No automatic deletion policy is enabled.
+See forms-processor-v6/README.md for rules, settings, replay, and operational SQL.
+
+The `sendgrid-email` action name is retained for compatibility, but forms-processor-v6
+now publishes the v3 email contract to `external.action.email` through Bus API.
+`deliveredAt` records Bus API acceptance; email-service-v6 handles provider delivery
+and retries. Optional `fromEmail` overrides that service's default sender.

docs/operations.md

@@ -10,7 +10,7 @@
 | `JWKS_URL`             | Required HTTPS URL in JWKS mode; only RS256 is accepted.                                                                    |
 | `AUTH_SECRET`          | At least 32 characters in HS256 mode; only HS256 is accepted.                                                               |
 | `VALID_ISSUERS`        | Required comma-separated exact JWT issuers. Unlike some older APIs, this setting is not a JSON array.                       |
-| `AUTH_AUDIENCE`        | Required expected JWT audience.                                                                                             |
+| `AUTH_AUDIENCE`        | Required expected JWT audience for standard tokens; legacy Topcoder human HS256 tokens omit it (see [authentication](api.md#authentication)).                                                                                             |
 | `AUTH_CLAIM_NAMESPACE` | Exact roles/userId custom-claim prefix; default `https://topcoder.com/`.                                                    |
 | `CORS_ORIGINS`         | Comma-separated exact HTTP(S) origins, no trailing slash/wildcard. Empty means no browser origins are allowed.              |
 | `THROTTLE_LIMIT`       | Requests per minute per client IP and handler, per replica; default 30. Health probes are exempt.                           |

prisma/migrations/20260928020000_processor_flows/migration.sql

@@ -0,0 +1,45 @@
+-- Workflow definitions and durable processing state belong to the forms API migration history.
+CREATE TABLE forms."ProcessorFlow" (
+  id VARCHAR(100) PRIMARY KEY,
+  "formKey" VARCHAR(40) NOT NULL,
+  enabled BOOLEAN NOT NULL DEFAULT false,
+  action VARCHAR(100) NOT NULL,
+  rules JSONB NOT NULL DEFAULT '{"all":[]}',
+  settings JSONB NOT NULL DEFAULT '{}',
+  "createdAt" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "updatedAt" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  CONSTRAINT "ProcessorFlow_rules_object" CHECK (jsonb_typeof(rules) = 'object'),
+  CONSTRAINT "ProcessorFlow_settings_object" CHECK (jsonb_typeof(settings) = 'object')
+);
+CREATE INDEX "ProcessorFlow_formKey_idx" ON forms."ProcessorFlow" ("formKey");
+
+CREATE TABLE forms."ProcessorEvent" (
+  "submissionId" UUID PRIMARY KEY,
+  "formKey" VARCHAR(40) NOT NULL,
+  payload JSONB NOT NULL,
+  "receivedAt" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "routedAt" TIMESTAMPTZ(3),
+  attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+  "nextAttemptAt" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "lastError" VARCHAR(100),
+  CONSTRAINT "ProcessorEvent_payload_object" CHECK (jsonb_typeof(payload) = 'object')
+);
+CREATE INDEX "ProcessorEvent_routedAt_nextAttemptAt_idx" ON forms."ProcessorEvent" ("routedAt", "nextAttemptAt");
+
+CREATE TABLE forms."ProcessorDelivery" (
+  "submissionId" UUID NOT NULL REFERENCES forms."ProcessorEvent"("submissionId") ON DELETE CASCADE,
+  "flowId" VARCHAR(100) NOT NULL REFERENCES forms."ProcessorFlow"(id) ON DELETE RESTRICT,
+  attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+  "nextAttemptAt" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "lastError" VARCHAR(100),
+  "deliveredAt" TIMESTAMPTZ(3),
+  "createdAt" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY ("submissionId", "flowId")
+);
+CREATE INDEX "ProcessorDelivery_deliveredAt_nextAttemptAt_idx" ON forms."ProcessorDelivery" ("deliveredAt", "nextAttemptAt");
+CREATE INDEX "ProcessorDelivery_flowId_idx" ON forms."ProcessorDelivery" ("flowId");
+
+-- TBD values remain null/empty and the flow disabled until configured externally.
+INSERT INTO forms."ProcessorFlow" (id, "formKey", action, settings)
+VALUES ('lets-talk-sales-email', 'lets-talk', 'sendgrid-email',
+        '{"recipients":[],"templateId":null,"fromEmail":null,"fromName":"Topcoder"}');

prisma/schema.prisma

@@ -187,3 +187,53 @@ model SubmissionEvent {
 
   @@schema("forms")
 }
+
+/// Externally managed content rules and action settings. Disabled flows retain pending deliveries.
+model ProcessorFlow {
+  id         String              @id @db.VarChar(100)
+  formKey    String              @db.VarChar(40)
+  enabled    Boolean             @default(false)
+  action     String              @db.VarChar(100)
+  rules      Json                @default("{\"all\":[]}")
+  settings   Json                @default("{}")
+  createdAt  DateTime            @default(now()) @db.Timestamptz(3)
+  updatedAt  DateTime            @default(now()) @db.Timestamptz(3)
+  deliveries ProcessorDelivery[]
+
+  @@index([formKey])
+  @@schema("forms")
+}
+
+/// Durable Kafka inbox; independent of submission retention and committed before acknowledging Kafka.
+model ProcessorEvent {
+  submissionId  String              @id @db.Uuid
+  formKey       String              @db.VarChar(40)
+  payload       Json
+  receivedAt    DateTime            @default(now()) @db.Timestamptz(3)
+  routedAt      DateTime?           @db.Timestamptz(3)
+  attempts      Int                 @default(0)
+  nextAttemptAt DateTime            @default(now()) @db.Timestamptz(3)
+  lastError     String?             @db.VarChar(100)
+  deliveries    ProcessorDelivery[]
+
+  @@index([routedAt, nextAttemptAt])
+  @@schema("forms")
+}
+
+/// Per-flow acceptance receipt and retry schedule; Bus API acceptance is not recipient delivery confirmation.
+model ProcessorDelivery {
+  submissionId  String         @db.Uuid
+  flowId        String         @db.VarChar(100)
+  attempts      Int            @default(0)
+  nextAttemptAt DateTime       @default(now()) @db.Timestamptz(3)
+  lastError     String?        @db.VarChar(100)
+  deliveredAt   DateTime?      @db.Timestamptz(3)
+  createdAt     DateTime       @default(now()) @db.Timestamptz(3)
+  event         ProcessorEvent @relation(fields: [submissionId], references: [submissionId], onDelete: Cascade, onUpdate: NoAction)
+  flow          ProcessorFlow  @relation(fields: [flowId], references: [id], onDelete: Restrict, onUpdate: NoAction)
+
+  @@id([submissionId, flowId])
+  @@index([deliveredAt, nextAttemptAt])
+  @@index([flowId])
+  @@schema("forms")
+}

src/auth.ts

@@ -48,7 +48,8 @@ export class AuthGuard implements CanActivate {
   ) {}
 
   /**
-   * Authenticates an optional public-route token or a required administration token.
+   * Authenticates standard JWTs or legacy Topcoder HS256 member tokens, then checks route access.
+   * Legacy tokens without sub/aud use their verified userId as the audit subject.
    * @param context Current HTTP route and request.
    * @returns True after assigning a verified actor, or for an anonymous public request.
    * @throws UnauthorizedException for invalid/missing JWTs; ForbiddenException for insufficient roles/scopes.
@@ -75,14 +76,38 @@ export class AuthGuard implements CanActivate {
       const verificationOptions = {
         algorithms: [this.config.authMode === 'hs256' ? 'HS256' : 'RS256'],
         issuer: this.config.issuers,
-        audience: this.config.audience,
-        requiredClaims: ['exp', 'sub', 'iat'],
+        requiredClaims: ['exp', 'iat'],
       };
       const { payload } =
         this.key instanceof Uint8Array
           ? await jwtVerify(header.slice(7), this.key, verificationOptions)
           : await jwtVerify(header.slice(7), this.key, verificationOptions);
-      request.actor = normalizeActor(payload, this.config.claimNamespace);
+      // Identity API's legacy member JWTs use userId and omit both sub and aud.
+      // Select this compatibility profile only after signature/issuer verification.
+      const legacyMember =
+        this.config.authMode === 'hs256' &&
+        payload.sub === undefined &&
+        payload.aud === undefined &&
+        /^https:\/\/api\.topcoder(?:-dev)?\.com$/.test(payload.iss ?? '') &&
+        ((typeof payload.userId === 'string' &&
+          /^[1-9]\d{0,19}$/.test(payload.userId)) ||
+          (typeof payload.userId === 'number' &&
+            Number.isSafeInteger(payload.userId) &&
+            payload.userId > 0));
+      if (!legacyMember) {
+        const audiences = Array.isArray(payload.aud)
+          ? payload.aud
+          : [payload.aud];
+        if (!audiences.includes(this.config.audience))
+          throw new Error('Invalid token audience.');
+      }
+      const actor = normalizeActor(
+        legacyMember ? { ...payload, sub: String(payload.userId) } : payload,
+        this.config.claimNamespace,
+      );
+      if (legacyMember && actor.machine)
+        throw new Error('Machine tokens require subject and audience.');
+      request.actor = actor;
     } catch {
       throw new UnauthorizedException('Invalid or expired bearer token.');
     }
@@ -110,7 +135,7 @@ export class AuthGuard implements CanActivate {
  */
 export function normalizeActor(claims: JWTPayload, namespace: string): Actor {
   const subject = claims.sub;
-  if (!subject || subject.length > 200)
+  if (typeof subject !== 'string' || !subject.trim() || subject.length > 200)
     throw new Error('Invalid token subject.');
   const roles = stringList(
     claims.roles ?? claims[`${namespace}roles`],

test/auth.spec.ts

@@ -0,0 +1,151 @@
+import 'reflect-metadata';
+import type { ExecutionContext } from '@nestjs/common';
+import { Reflector } from '@nestjs/core';
+import { describe, expect, it } from 'vitest';
+import { AuthGuard, type ActorRequest, type Permission } from '../src/auth';
+import { readConfig } from '../src/config';
+import { testSecret } from './fixtures';
+
+const config = readConfig({
+  DATABASE_URL: 'postgresql://localhost/forms',
+  AUTH_MODE: 'hs256',
+  AUTH_SECRET: testSecret,
+  VALID_ISSUERS:
+    'https://api.topcoder-dev.com,https://api.topcoder.com,https://forms.test',
+  AUTH_AUDIENCE: 'forms-api',
+});
+
+/**
+ * Signs Identity API-shaped claims and runs the actual guard without a database.
+ * @param overrides Claim overrides, including undefined to omit a claim.
+ * @param permission Route access level. @param secret Signing key for negative tests.
+ * @param algorithm Signing algorithm for allowlist tests.
+ * @returns Verified actor. @throws Authentication/authorization errors from the guard.
+ */
+async function authenticate(
+  overrides: Record<string, unknown> = {},
+  permission: Permission = 'report',
+  secret = testSecret,
+  algorithm = 'HS256',
+) {
+  const { SignJWT } = await import('jose');
+  const now = Math.floor(Date.now() / 1000);
+  const jwt = await new SignJWT({
+    userId: '12345',
+    roles: ['Administrator'],
+    iss: 'https://api.topcoder-dev.com',
+    iat: now,
+    exp: now + 300,
+    ...overrides,
+  })
+    .setProtectedHeader({ alg: algorithm })
+    .sign(new TextEncoder().encode(secret));
+  const request = {
+    headers: { authorization: `Bearer ${jwt}` },
+  } as ActorRequest;
+  const handler = () => undefined;
+  Reflect.defineMetadata('forms.permission', permission, handler);
+  const context = {
+    getHandler: () => handler,
+    getClass: () => AuthGuard,
+    switchToHttp: () => ({ getRequest: () => request }),
+  } as unknown as ExecutionContext;
+  await new AuthGuard(config, new Reflector()).canActivate(context);
+  return request.actor;
+}
+
+describe('Topcoder bearer authentication', () => {
+  it.each(['https://api.topcoder-dev.com', 'https://api.topcoder.com'])(
+    'accepts legacy administrator tokens issued by %s',
+    async (iss) => {
+      expect(await authenticate({ iss })).toEqual({
+        subject: '12345',
+        memberId: '12345',
+        machine: false,
+        roles: ['administrator'],
+        scopes: [],
+      });
+    },
+  );
+
+  it('uses a numeric userId for legacy editor audit and member identity', async () => {
+    expect(await authenticate({ userId: 12345 }, 'manage')).toMatchObject({
+      subject: '12345',
+      memberId: '12345',
+    });
+    expect(await authenticate({ roles: ['Member'] }, 'public')).toMatchObject({
+      memberId: '12345',
+      machine: false,
+    });
+  });
+
+  it('still enforces human report permissions', async () => {
+    await expect(authenticate({ roles: ['Member'] })).rejects.toMatchObject({
+      status: 403,
+    });
+    await expect(
+      authenticate({ roles: ['Forms Administrator'] }),
+    ).rejects.toMatchObject({ status: 403 });
+    expect(await authenticate({ roles: ['Forms Reporter'] })).toBeDefined();
+  });
+
+  it.each([
+    { exp: 1 },
+    { exp: undefined },
+    { iat: undefined },
+    { iss: 'https://untrusted.example' },
+    { iss: 'https://forms.test' },
+    { userId: undefined },
+    { userId: '' },
+    { userId: {} },
+    { userId: 1.5 },
+    { userId: 0 },
+    { userId: '1'.repeat(21) },
+    { sub: '', aud: 'forms-api' },
+    { sub: 42, aud: 'forms-api' },
+    { sub: 'auth0|person' },
+    { aud: 'forms-api' },
+    { sub: 'auth0|person', aud: 'wrong-api' },
+    { gty: 'client-credentials' },
+    { isMachine: true },
+    { scope: 'read:forms-submissions', roles: [] },
+  ])('rejects invalid or nonlegacy claims %#', async (claims) => {
+    await expect(authenticate(claims)).rejects.toMatchObject({ status: 401 });
+  });
+
+  it('rejects invalid signatures even for the legacy profile', async () => {
+    await expect(
+      authenticate({}, 'report', `${testSecret}-wrong`),
+    ).rejects.toMatchObject({ status: 401 });
+  });
+
+  it('rejects other signing algorithms even for the legacy profile', async () => {
+    await expect(
+      authenticate({}, 'report', testSecret, 'HS384'),
+    ).rejects.toMatchObject({ status: 401 });
+  });
+
+  it.each(['forms-api', ['another-api', 'forms-api']])(
+    'accepts standard audience %j',
+    async (aud) => {
+      expect(await authenticate({ sub: 'auth0|person', aud })).toMatchObject({
+        subject: 'auth0|person',
+      });
+    },
+  );
+
+  it('keeps machine scope authorization separate from administrator roles', async () => {
+    const machine = {
+      sub: 'client@clients',
+      aud: 'forms-api',
+      gty: 'client-credentials',
+    };
+    await expect(authenticate(machine)).rejects.toMatchObject({ status: 403 });
+    expect(
+      await authenticate({ ...machine, scope: 'read:forms-submissions' }),
+    ).toMatchObject({
+      machine: true,
+      memberId: undefined,
+    });
+  });
+});