Publish opted-in form submissions to Kafka through Bus API

.env.example

@@ -19,3 +19,14 @@ TRUST_PROXY_CIDRS=
 # Local Docker Compose host ports.
 FORMS_HTTP_PORT=3006
 FORMS_DATABASE_PORT=5546
+
+# Optional outbound Kafka publication through Bus API (required for kafka=true submissions).
+# The wrapper adds /bus/events; use the API base ending in /v6.
+# BUSAPI_URL=https://api.topcoder-dev.com/v6
+# AUTH0_URL=https://topcoder-dev.auth0.com/oauth/token
+# AUTH0_AUDIENCE=https://www.topcoder-dev.com
+# AUTH0_CLIENT_ID=
+# AUTH0_CLIENT_SECRET=
+# TOKEN_CACHE_TIME=86400000
+# AUTH0_PROXY_SERVER_URL=
+# KAFKA_ERROR_TOPIC=common.error.reporting

README.md

@@ -66,7 +66,8 @@ Stop the Compose API first if it already occupies port 3006. `pnpm start:dev` us
 3. Publish it. The API creates `forms.event_interest_v1` transactionally.
 4. Add a Payload form block referencing `event_interest` to a content page.
 5. The website fetches the current API schema and submits the exact displayed revision with a UUID `Idempotency-Key`.
-6. Read private paginated JSON/CSV reports or query the SQL view with a reporting database role.
+6. Optionally include `"kafka": true` alongside the submitted answers to publish a `form.submitted` event through Bus API; see [the event contract and retry behavior](docs/api.md#optional-kafka-publication).
+7. Read private paginated JSON/CSV reports or query the SQL view with a reporting database role.
 
 Published definitions are immutable. Changes use a new sequential revision; publication retires the previous one. An already-open older page receives a 409 and must reload. Historical reports remain available. Exact retries of accepted submissions return the original receipt, including after retirement.
 

docs/api.md

@@ -69,9 +69,42 @@ Idempotency-Key: 6b3d585c-f134-4e06-a6d5-6ea42fc1c7ce
 
 Receipts contain no answers or member details. All answers must belong to the pinned stored revision; client-supplied field definitions and extra answer keys are rejected. Server validation does not coerce number or boolean strings. Decimal values are exact strings; multi-select answers are arrays of option keys. Every answer and its selections commit atomically with the envelope.
 
+## Optional Kafka publication
+
+Include `"kafka": true` in the submission body alongside `version`, `answers`, and `sourcePage` to publish the accepted submission to **`form.submitted`** through the authenticated Topcoder Bus API. The exact string `"true"` is also accepted; omitted, `false`, or `"false"` does not publish. Other non-null values are rejected. This is envelope metadata, not an answer field or query parameter. Answer values retain their strict type validation.
+
+```json
+{
+  "version": 1,
+  "answers": { "email": "member@example.com" },
+  "sourcePage": "/events",
+  "kafka": true
+}
+```
+
+The Bus API envelope uses `topic: "form.submitted"`, `originator: "forms-api-v6"`, `mime-type: "application/json"`, the submission timestamp, and the submission UUID as `key`. Its `payload` is:
+
+```json
+{
+  "submissionId": "c1ab5d27-722a-431f-a811-46737109de1f",
+  "formKey": "event_interest",
+  "version": 1,
+  "submittedAt": "2026-09-28T01:00:00.000Z",
+  "memberId": null,
+  "sourcePage": "/events",
+  "answers": { "email": "member@example.com" }
+}
+```
+
+Answers use their validated, canonical values (including exact decimal strings, booleans, numbers, date strings, and option keys); omitted optional answers stay omitted. Member identity comes from verified claims. The honeypot and Kafka flag are excluded from the event. The HTTP receipt remains unchanged.
+
+Publication happens only after the submission and all answers commit. A successful delivery is recorded separately and identical retries, including concurrent retries, do not publish again. Changing the Kafka opt-in for an existing retry key returns 409; omission and false are equivalent and preserve compatibility with older submissions.
+
+If Bus API is unavailable or unconfigured, the submission remains saved and the request returns **503**. Retry the identical body and `Idempotency-Key` to resume delivery, even after retirement. There is no background retry worker. A crash or lost acknowledgement after Bus API accepts an event but before the delivery receipt commits can cause redelivery; consumers must deduplicate by `submissionId`. Invalid or rejected submissions never publish.
+
 ## Retry and error handling
 
-Generate one cryptographically random UUID for a logical submission attempt. Reuse it with the same revision, answers, member identity, and source page after a timeout or lost response. The service canonicalizes decimals and multi-select order before hashing. An identical retry returns the original receipt; changed content with that key returns 409. This key remains reserved while the submission is retained.
+Generate one cryptographically random UUID for a logical submission attempt. Reuse it with the same revision, answers, member identity, source page, and Kafka opt-in after a timeout or lost response. The service canonicalizes decimals and multi-select order before hashing. An identical retry returns the original receipt; changed content with that key returns 409. This key remains reserved while the submission is retained.
 
 A previously accepted attempt can be retried after a version is retired. A new attempt targeting a draft/retired version returns 409. A member-form retry still requires the member token. The API does not replay a submission under a different member identity.
 

docs/data-model.md

@@ -17,6 +17,7 @@ erDiagram
     FormVersion ||--o{ FormField : defines
     FormField ||--o{ FieldOption : choices
     FormVersion ||--o{ Submission : receives
+    Submission ||--o| SubmissionEvent : delivery
     Submission ||--o{ Answer : contains
     FormField ||--o{ Answer : constrains
     Answer ||--o{ AnswerSelection : selects
@@ -32,6 +33,7 @@ erDiagram
 | `FormField`       | Named key unique within a version; explicit type, position, required flag, text length and numeric bounds.                                                          |
 | `FieldOption`     | Named option key and label, owned by a field/version.                                                                                                               |
 | `Submission`      | Exact version, server timestamp, UUID idempotency key, canonical request hash, verified member ID, and optional source page path.                                   |
+| `SubmissionEvent` | Optional one-to-one delivery receipt with submission UUID and nullable publication timestamp; cascades on submission deletion. |
 | `Answer`          | One scalar value in a type-specific column, or a multi-select answer parent; unique per submission/field.                                                           |
 | `AnswerSelection` | One row per selected option, with composite ownership foreign keys and duplicate prevention.                                                                        |
 
@@ -48,6 +50,8 @@ The API additionally validates email syntax, exact decimal input precision, real
 
 API form registration and revisions are serialized on the stable form row. Publication, retirement, and submission acceptance use the same lock. This favors simple consistency for occasional website forms; it serializes submissions to the same form. Revisit this locking strategy if measured submission volume requires greater throughput.
 
+Kafka opt-in creates a `SubmissionEvent` row atomically with the submission. Its publication timestamp is updated after Bus API acceptance in a separate transaction; submission envelopes and answers remain immutable. Delivery retries lock this row to prevent concurrent duplicate sends. A failed or unacknowledged delivery remains pending until the caller retries. No JSON payload is stored; the validated request and immutable revision reconstruct the event.
+
 ## Field semantics
 
 - `INTEGER` is PostgreSQL `integer` (signed 32-bit), sent as a JSON number.

docs/operations.md

@@ -20,6 +20,28 @@ All Forms tables, enums, functions, migration history, and reporting views live
 
 The Prisma client uses the PostgreSQL driver adapter and generated TypeScript code. Connection URLs are configured in `prisma.config.ts`, consistent with the [Prisma 7 migration guide](https://www.prisma.io/docs/orm/more/upgrade-guides/upgrading-versions/upgrading-to-prisma-7). Client generation and compilation require no live database. Migrations and runtime require `DATABASE_URL`.
 
+## Outbound Bus API
+
+Ordinary submissions need no Bus API configuration. To accept `kafka=true` submissions successfully, configure:
+
+| Variable | Meaning |
+| --- | --- |
+| `BUSAPI_URL` | HTTP(S) API base ending in `/v6`, e.g. `https://api.topcoder-dev.com/v6`. The shared wrapper appends `/bus/events`. |
+| `AUTH0_URL` | Auth0 token endpoint used by the shared Topcoder M2M client. |
+| `AUTH0_AUDIENCE` | Outbound M2M audience; separate from inbound `AUTH_AUDIENCE`. |
+| `AUTH0_CLIENT_ID`, `AUTH0_CLIENT_SECRET` | Service credentials authorized to publish Bus API events. Required when `BUSAPI_URL` is set. |
+| `TOKEN_CACHE_TIME` | Optional M2M token cache duration in milliseconds, 0–86400000; otherwise uses the wrapper default. |
+| `AUTH0_PROXY_SERVER_URL` | Optional Auth0 proxy supported by the shared wrapper. |
+| `KAFKA_ERROR_TOPIC` | Wrapper error-topic setting, default `common.error.reporting`; submission topic is always `form.submitted`. |
+
+For ECS, expose the five required Bus API/Auth0 settings to the runtime container through its task-definition secrets (the existing template only maps inbound authentication and database settings). Store them under the service parameter prefix so its existing SSM read permissions apply; never put credentials in the template.
+
+Apply migration `20260928010000_submission_events` before deploying this version. It adds the `forms.SubmissionEvent` delivery table; it does not change existing submissions or reporting views. The service role needs SELECT/INSERT/UPDATE on this table. Ensure `form.submitted` is available through Bus API and downstream consumers deduplicate on the payload's `submissionId`.
+
+Delivery failures return 503 after saving the submission; callers must retry the same body and key. Pending attempts have `publishedAt IS NULL` in `forms.SubmissionEvent`. No background delivery job is included. Provider error bodies, credentials, and answers are not logged by the publisher. Delivery uses a separate transaction with a row lock and a 15-second transaction timeout; failure to record an accepted event may result in redelivery.
+
+The shared wrapper and its Topcoder core dependency are pinned to Git commits. `pnpm-workspace.yaml` allows Git subdependencies for this integration and explicitly skips optional native DTrace builds.
+
 ## Database access
 
 The checked-in migrations create tables, enums, foreign keys, CHECKs, indexes, and lifecycle/answer triggers inside `forms`. Use `pnpm migrate:deploy`; **do not use `prisma db push`**, which does not reproduce the custom integrity triggers and checks.

package.json

@@ -40,7 +40,8 @@
     "jose": "6.2.10",
     "pg": "8.23.0",
     "reflect-metadata": "0.2.2",
-    "rxjs": "7.8.2"
+    "rxjs": "7.8.2",
+    "tc-bus-api-wrapper": "github:topcoder-platform/tc-bus-api-wrapper#297a9c0adcdb97661257e7825bee9c3f5578b833"
   },
   "devDependencies": {
     "@eslint/js": "9.39.2",