From 55642ba444844edae536e3c6b674eeb32b7a0b10 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:11:13 +0530 Subject: [PATCH 01/11] docs(spec): API repository-layer refactor design Repository ports + Drizzle adapters for apps/api (no domain layer yet). Executor pattern preserves the one-transaction outbox write and RLS scoping. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../2026-06-14-api-repository-layer-design.md | 129 ++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 docs/superpowers/specs/2026-06-14-api-repository-layer-design.md diff --git a/docs/superpowers/specs/2026-06-14-api-repository-layer-design.md b/docs/superpowers/specs/2026-06-14-api-repository-layer-design.md new file mode 100644 index 0000000..cd05688 --- /dev/null +++ b/docs/superpowers/specs/2026-06-14-api-repository-layer-design.md @@ -0,0 +1,129 @@ +# API Repository Layer — Design + +**Date:** 2026-06-14 +**Status:** Approved (design); implementation pending +**Scope:** `apps/api` only. Worker and a separate domain layer are explicitly out of scope (a later, separate effort). + +## Context + +Today the API's NestJS services (`forms`, `endpoints`, `submissions`, `deliveries`, `public`, `auth/tenants`) reach the database directly: they import `drizzle-orm` operators, the `@eventform/db` schema, and raw `pg` `Pool`/`PoolClient`. Business logic, persistence, and transaction handling are mixed in one layer. + +This refactor extracts **persistence behind repository ports (interfaces) + Drizzle adapters**, so services depend on an abstraction instead of the ORM. It is a structural change only — no behavior changes. + +## Goal & non-goals + +**Goal:** Each API module exposes a repository **port** (interface) implemented by a Drizzle **adapter**. Services stop importing `drizzle-orm`/`pg` and instead depend on the port via DI (mirroring the existing `TOKEN_VERIFIER` / `SECRET_CIPHER` seams). + +**Non-goals (deferred):** +- No framework-free domain entities / value objects. Services remain the application layer and keep their current logic. +- No mock-based unit tests. The existing live-Postgres integration suite is the safety net. +- No new top-level `infrastructure/` directory (fits better when the domain layer lands). +- No worker changes. + +## Guardrails (must not regress) + +1. **Atomicity:** the submit path writes `submission` + N `deliveries` + N `outbox` rows in **one** `withTenant` transaction. The **manual-retry path** is a second atomic write — it updates the delivery and inserts a new `outbox` row in one transaction. Splitting persistence across repositories must not split either transaction (both use the shared tx executor inside one `withTenant`). +2. **RLS tenant scoping:** tenant-scoped operations run inside `withTenant` (which issues `SET LOCAL app.tenant_id`); the anonymous public-form read runs on the pool with no tenant set (relying on the RLS public-read policy). Both behaviors preserved exactly. +3. **Tests green at every step:** the existing integration tests (`apps/api`, `packages/db`) must pass after each module is migrated. + +## Architecture — the executor pattern + +A repository method must be able to participate in a caller-controlled transaction, so **every repository method takes an `Executor` as its first argument**: + +```ts +import type { Db } from "@eventform/db"; +export type Executor = Db; // a Drizzle handle: tx-bound (inside withTenant) OR pool-bound + +export interface SubmissionRepository { + insert(x: Executor, row: NewSubmission): Promise; +} +``` + +The **service** decides which executor to pass and owns the transaction policy: + +- **Atomic writes** — the service opens the unit of work and passes the *tx-bound* handle to each repo call, so all writes share one transaction and one `SET LOCAL app.tenant_id`: + ```ts + await withTenant(pool, tenantId, async (db) => { + const sub = await this.submissions.insert(db, ...); + await this.deliveries.insert(db, ...); // same tx + await this.outbox.insert(db, ...); // same tx + }); + ``` +- **Anonymous / non-tenant reads** — the service passes a **pool-bound** executor (no `SET LOCAL`), so RLS public-read policies apply as today. + +Repositories therefore hold **zero** knowledge of tenancy or transaction policy — they only run queries against whatever executor they're handed. That policy lives in the service, where it already is. + +**Supporting infra:** add a pool-bound executor helper alongside `withTenant` in `@eventform/db`, e.g. `poolExecutor(pool): Db` (a memoized `drizzle(pool, { schema })`). `withTenant`'s signature is unchanged. + +**Rejected alternative:** tx-bound repository *instances* via a UnitOfWork factory (`uow.withTenant(tenantId, repos => ...)`). Cleaner call sites but more infrastructure; not worth it for a repository-only first pass. Revisit if/when the domain layer lands. + +## Components — the repositories + +One port + one Drizzle adapter per module, derived from the current service's queries: + +| Repository | Responsibilities (method surface, indicative) | +|---|---| +| `FormRepository` | CRUD forms + form_fields; `findPublishedBySlug` (anonymous, pool executor); publish/unpublish | +| `EndpointRepository` | CRUD endpoints (stores/returns ciphertext only); list active by tenant | +| `SubmissionRepository` | `insert` | +| `DeliveryRepository` | list/get deliveries; read `delivery_attempts` for the UI; manual-retry reads/writes | +| `OutboxRepository` | `insert` (used by submit + manual retry) | +| `TenantRepository` | find/provision by `cognito_sub`; update display name | + +Each port + adapter is named/derived directly from the queries the corresponding service runs today — no new query behavior. + +## File layout & DI wiring + +Per-module, mirroring existing patterns: + +``` +apps/api/src//.repository.ts # port interface + Executor re-export + a DI token const +apps/api/src//.repository.drizzle.ts # Drizzle adapter (the only file importing drizzle-orm/schema) +``` + +- DI token per repo (e.g. `export const FORM_REPOSITORY = "FORM_REPOSITORY"`), provided in the module with `useClass` the Drizzle adapter, exactly like `TOKEN_VERIFIER` (`auth.module.ts`) and `SECRET_CIPHER` (`db.module.ts`). +- The pool-bound executor + `API_POOL` are provided by the existing `DbModule` (global) and injected into adapters that need anonymous reads. +- After the refactor, `drizzle-orm` and `pg` imports exist **only** in `*.repository.drizzle.ts` files and `@eventform/db` — not in any service. + +## Data flow & responsibilities + +``` +Controller (driving adapter) + → Service (application/use-case: input already validated by zod pipe; + owns withTenant boundary; maps to HTTP exceptions; builds outbox payload; encrypts via SECRET_CIPHER) + → Repository port + → Drizzle adapter + → Postgres +``` + +- Repos return plain row data or `null`; they do **not** throw HTTP exceptions. +- Services keep translating absence/conflict into `NotFoundException` / `ConflictException`, keep building the `SubmissionReceivedEvent` payload, and keep calling `SECRET_CIPHER` (repos persist/return ciphertext only — encryption never moves into persistence). + +## Error handling + +- `drizzle-exception.filter.ts` stays as-is (still catches DB constraint errors raised through the adapters). +- Service-level `NotFound`/`Conflict` mapping is unchanged. +- The `withTenant` rollback-on-error + poisoned-connection handling is unchanged (still the transaction owner's concern). + +## Testing strategy + +- No new unit tests, no mocks. The existing live-Postgres integration suites are the contract. +- Migrate one module at a time; run `pnpm --filter @eventform/api test` and `pnpm --filter @eventform/db test` after each. The whole suite green is the per-step success criterion. +- The Playwright smoke (sign-in → build → publish → submit → delivered) is the end-to-end backstop. + +## Migration plan (incremental, tests green each step) + +1. Add `poolExecutor(pool)` + `Executor` type to `@eventform/db`. +2. `endpoints` — simplest CRUD; establishes the port/adapter/DI shape. +3. `submissions` — single `insert`. +4. `deliveries` — lists + attempts + manual-retry. +5. `forms` — CRUD + fields + the anonymous `findPublishedBySlug` (exercises the pool executor). +6. `public` — the **atomic outbox write**; done last and most carefully (submission + deliveries + outbox via the tx executor inside `withTenant`). +7. `auth/tenants` (+ `me`) — provision/update. +8. Final sweep: confirm no service imports `drizzle-orm`/`pg`; full suite + smoke green. + +## Out of scope / future + +- **Domain layer** (entities, value objects, pure domain logic, application/use-case separation) — deferred; the user will decide based on the results of this pass. +- Worker repository/domain refactor. +- Any move to a top-level `infrastructure/` layout or a UnitOfWork factory — revisit alongside the domain layer. From 7b65edfd640e26069d9b84379bc3d7ea5e0f7246 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:17:13 +0530 Subject: [PATCH 02/11] docs(plan): API repository-layer implementation plan 8 tasks: Executor foundation + one repository (port + Drizzle adapter) per module, services keep withTenant/logic, boundary-verification sweep. Refactor under the existing green integration suite (no new tests). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../plans/2026-06-14-api-repository-layer.md | 1375 +++++++++++++++++ 1 file changed, 1375 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-14-api-repository-layer.md diff --git a/docs/superpowers/plans/2026-06-14-api-repository-layer.md b/docs/superpowers/plans/2026-06-14-api-repository-layer.md new file mode 100644 index 0000000..4da4b30 --- /dev/null +++ b/docs/superpowers/plans/2026-06-14-api-repository-layer.md @@ -0,0 +1,1375 @@ +# API Repository Layer Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Extract `apps/api` persistence behind repository ports (interfaces) + Drizzle adapters, so services depend on an abstraction instead of importing `drizzle-orm`/`pg` directly — with no behavior change. + +**Architecture:** Per-module repository **port** (interface) + Drizzle **adapter**, wired by a DI token (mirroring `TOKEN_VERIFIER`/`SECRET_CIPHER`). Every repo method takes an **`Executor`** (a Drizzle handle) as its first arg; the **service** keeps owning the `withTenant` transaction and passes the tx-bound handle (atomic writes) or a pool-bound handle (anonymous reads). Services retain all business logic, HTTP exceptions, encryption, and payload building. + +**Tech Stack:** NestJS, Drizzle ORM, `pg`, Postgres (RLS), Vitest (existing integration suite against live Postgres). + +**Verification model:** This is a refactor under existing coverage. There are no new tests. After each task, the existing suites must stay green: +- `pnpm --filter @eventform/api test` +- `pnpm --filter @eventform/db test` + +If a task makes a suite fail, the refactor for that module is wrong — fix before committing. + +--- + +## File Structure + +For each module under `apps/api/src//`: +- `*.repository.ts` — **port**: the interface, a DI token const, and the row type(s) re-exported from `@eventform/db`. No `drizzle-orm` import. +- `*.repository.drizzle.ts` — **adapter**: the only file importing `drizzle-orm` + schema tables for that module. `@Injectable()`. +- `*.service.ts` — **modified**: stops importing `drizzle-orm`/schema-tables; injects the port; keeps `withTenant`, exceptions, cipher, payloads. +- `*.module.ts` — **modified**: provides the port via `{ provide: TOKEN, useClass: Adapter }`. + +Foundation (shared): +- `packages/db/src/client.ts` — add `Executor` type alias + `poolExecutor(pool)`. +- `packages/db/src/index.ts` — export them. + +After completion, `drizzle-orm` and `pg` are imported **only** in `*.repository.drizzle.ts` files and `@eventform/db` — not in any `*.service.ts`. + +--- + +## Task 1: Executor foundation in @eventform/db + +**Files:** +- Modify: `packages/db/src/client.ts` +- Modify: `packages/db/src/index.ts` + +- [ ] **Step 1: Add `Executor` alias and `poolExecutor` to client.ts** + +Open `packages/db/src/client.ts`. It already has `export type Db = NodePgDatabase;` and `withTenant`. Add, right after the `Db` type definition: + +```ts +/** + * A query executor handed to repositories. It is just a Drizzle handle: + * - tx-bound (from withTenant) for atomic, tenant-scoped work, or + * - pool-bound (from poolExecutor) for anonymous reads under RLS public policies. + * Repositories never open their own connection; the caller chooses the executor. + */ +export type Executor = Db; + +let pooled: Db | undefined; + +/** + * A pool-bound Drizzle handle for non-transactional / anonymous reads (no + * `SET LOCAL app.tenant_id`). Memoized per process. Each query borrows a pool + * connection. Do NOT use for multi-statement atomic work — use withTenant. + */ +export function poolExecutor(pool: Pool): Db { + if (!pooled) { + pooled = drizzle(pool, { schema }); + } + return pooled; +} +``` + +`drizzle`, `schema`, and `Pool` are already imported in this file (used by `withTenant`). Confirm those imports exist; if `Pool` is only imported as a type, ensure `import { Pool } from "pg"` / the existing import covers the value use (it is already used as a parameter type in `withTenant`, so the import is present). + +- [ ] **Step 2: Export from index.ts** + +Open `packages/db/src/index.ts`. It re-exports from `./client`. Ensure `Executor` and `poolExecutor` are exported. If the file does `export * from "./client";`, no change is needed — verify with: + +Run: `grep -nE "client|Executor|poolExecutor" packages/db/src/index.ts` +Expected: a re-export of `./client` is present (so the new symbols are exported). If it uses named exports instead of `export *`, add `Executor` (type) and `poolExecutor` to the list. + +- [ ] **Step 3: Build the db package to typecheck** + +Run: `pnpm --filter @eventform/db build` +Expected: PASS (no type errors). + +- [ ] **Step 4: Run db tests (unchanged, must stay green)** + +Run: `pnpm --filter @eventform/db test` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add packages/db/src/client.ts packages/db/src/index.ts +git commit -m "feat(db): add Executor type + poolExecutor for the repository layer" +``` + +--- + +## Task 2: EndpointRepository + +**Files:** +- Create: `apps/api/src/endpoints/endpoints.repository.ts` +- Create: `apps/api/src/endpoints/endpoints.repository.drizzle.ts` +- Modify: `apps/api/src/endpoints/endpoints.service.ts` +- Modify: `apps/api/src/endpoints/endpoints.module.ts` + +- [ ] **Step 1: Create the port** (`endpoints.repository.ts`) + +```ts +import type { Executor } from "@eventform/db"; +import { endpoints } from "@eventform/db"; + +export const ENDPOINT_REPOSITORY = "ENDPOINT_REPOSITORY"; + +export type EndpointRow = typeof endpoints.$inferSelect; + +export interface EndpointRepository { + countByTenant(x: Executor, tenantId: string): Promise; + insert( + x: Executor, + values: { tenantId: string; name: string; url: string; secretCiphertext: string }, + ): Promise; + listByTenant(x: Executor): Promise; + findById(x: Executor, id: string, tenantId: string): Promise; + update( + x: Executor, + id: string, + tenantId: string, + patch: Partial<{ name: string; url: string; active: boolean }>, + ): Promise; + updateSecret( + x: Executor, + id: string, + tenantId: string, + secretCiphertext: string, + ): Promise; + remove(x: Executor, id: string, tenantId: string): Promise; +} +``` + +Note: `endpoints` is imported only to derive `EndpointRow` via `$inferSelect`; no `drizzle-orm` operators here. + +- [ ] **Step 2: Create the adapter** (`endpoints.repository.drizzle.ts`) + +```ts +import { Injectable } from "@nestjs/common"; +import { and, asc, count, eq } from "drizzle-orm"; +import { endpoints, type Executor } from "@eventform/db"; +import { EndpointRepository, EndpointRow } from "./endpoints.repository"; + +@Injectable() +export class DrizzleEndpointRepository implements EndpointRepository { + async countByTenant(x: Executor, tenantId: string): Promise { + const [{ value }] = await x + .select({ value: count() }) + .from(endpoints) + .where(eq(endpoints.tenantId, tenantId)); + return value; + } + + async insert( + x: Executor, + values: { tenantId: string; name: string; url: string; secretCiphertext: string }, + ): Promise { + const [created] = await x.insert(endpoints).values(values).returning(); + return created; + } + + listByTenant(x: Executor): Promise { + return x.select().from(endpoints).orderBy(asc(endpoints.createdAt)); + } + + async findById(x: Executor, id: string, tenantId: string): Promise { + const [found] = await x + .select() + .from(endpoints) + .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))); + return found; + } + + async update( + x: Executor, + id: string, + tenantId: string, + patch: Partial<{ name: string; url: string; active: boolean }>, + ): Promise { + const [updated] = await x + .update(endpoints) + .set(patch) + .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))) + .returning(); + return updated; + } + + async updateSecret( + x: Executor, + id: string, + tenantId: string, + secretCiphertext: string, + ): Promise { + const [updated] = await x + .update(endpoints) + .set({ secretCiphertext }) + .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))) + .returning(); + return updated; + } + + async remove(x: Executor, id: string, tenantId: string): Promise { + const [removed] = await x + .delete(endpoints) + .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))) + .returning(); + return removed; + } +} +``` + +- [ ] **Step 3: Rewrite the service** (`endpoints.service.ts`) — full replacement + +```ts +import { ConflictException, Inject, Injectable, NotFoundException } from "@nestjs/common"; +import { Pool } from "pg"; +import { withTenant } from "@eventform/db"; +import { generateEndpointSecret, SecretCipher } from "@eventform/shared"; +import { API_POOL, SECRET_CIPHER } from "../db/db.module"; +import { CreateEndpointDto, UpdateEndpointDto } from "./endpoints.schemas"; +import { ENDPOINT_REPOSITORY, EndpointRepository, EndpointRow } from "./endpoints.repository"; + +function publicView(row: EndpointRow) { + const { secretCiphertext: _omitted, ...rest } = row; + return rest; +} + +@Injectable() +export class EndpointsService { + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(SECRET_CIPHER) private readonly cipher: SecretCipher, + @Inject(ENDPOINT_REPOSITORY) private readonly repo: EndpointRepository, + ) {} + + async create(tenantId: string, dto: CreateEndpointDto) { + const secret = generateEndpointSecret(); + const secretCiphertext = await this.cipher.encrypt(secret, tenantId); + const row = await withTenant(this.pool, tenantId, async (db) => { + // Cap endpoints at 20 per tenant. + const existing = await this.repo.countByTenant(db, tenantId); + if (existing >= 20) { + throw new ConflictException("endpoint limit reached (20)"); + } + return this.repo.insert(db, { tenantId, name: dto.name, url: dto.url, secretCiphertext }); + }); + return { ...publicView(row), secret }; + } + + list(tenantId: string) { + return withTenant(this.pool, tenantId, async (db) => { + const rows = await this.repo.listByTenant(db); + return rows.map(publicView); + }); + } + + async update(tenantId: string, id: string, dto: UpdateEndpointDto) { + const row = await withTenant(this.pool, tenantId, (db) => + this.repo.update(db, id, tenantId, dto), + ); + if (!row) { + throw new NotFoundException("endpoint not found"); + } + return publicView(row); + } + + async remove(tenantId: string, id: string) { + const removed = await withTenant(this.pool, tenantId, (db) => + this.repo.remove(db, id, tenantId), + ); + if (!removed) { + throw new NotFoundException("endpoint not found"); + } + } + + async revealSecret(tenantId: string, id: string) { + const row = await withTenant(this.pool, tenantId, (db) => + this.repo.findById(db, id, tenantId), + ); + if (!row) { + throw new NotFoundException("endpoint not found"); + } + const secret = await this.cipher.decrypt(row.secretCiphertext, tenantId); + return { secret }; + } + + async rotateSecret(tenantId: string, id: string) { + const secret = generateEndpointSecret(); + const secretCiphertext = await this.cipher.encrypt(secret, tenantId); + const row = await withTenant(this.pool, tenantId, (db) => + this.repo.updateSecret(db, id, tenantId, secretCiphertext), + ); + if (!row) { + throw new NotFoundException("endpoint not found"); + } + return { ...publicView(row), secret }; + } +} +``` + +Note: `update` passes `dto` (an `UpdateEndpointDto`) directly as the patch — same as the original `.set(dto)`. + +- [ ] **Step 4: Provide the repo in the module** (`endpoints.module.ts`) + +Add to the module's `providers` array (alongside the existing service): + +```ts +import { ENDPOINT_REPOSITORY } from "./endpoints.repository"; +import { DrizzleEndpointRepository } from "./endpoints.repository.drizzle"; +// ... +providers: [ + EndpointsService, + { provide: ENDPOINT_REPOSITORY, useClass: DrizzleEndpointRepository }, +], +``` + +(Keep existing controller/exports as they are.) + +- [ ] **Step 5: Run the API suite (must stay green)** + +Run: `pnpm --filter @eventform/api test` +Expected: PASS — endpoints e2e tests behave identically. + +- [ ] **Step 6: Commit** + +```bash +git add apps/api/src/endpoints/ +git commit -m "refactor(api): extract EndpointRepository port + Drizzle adapter" +``` + +--- + +## Task 3: SubmissionRepository + +**Files:** +- Create: `apps/api/src/submissions/submissions.repository.ts` +- Create: `apps/api/src/submissions/submissions.repository.drizzle.ts` +- Modify: `apps/api/src/submissions/submissions.service.ts` +- Modify: `apps/api/src/submissions/submissions.module.ts` + +- [ ] **Step 1: Create the port** (`submissions.repository.ts`) + +```ts +import type { Executor } from "@eventform/db"; +import { submissions } from "@eventform/db"; + +export const SUBMISSION_REPOSITORY = "SUBMISSION_REPOSITORY"; + +export type SubmissionRow = typeof submissions.$inferSelect; + +export interface SubmissionListItem { + id: string; + formId: string; + formTitle: string; + answers: SubmissionRow["answers"]; + submittedAt: SubmissionRow["submittedAt"]; + sourceIp: SubmissionRow["sourceIp"]; +} + +export interface SubmissionRepository { + /** All responses across the tenant's forms, newest first (latest 200). */ + listAll(x: Executor): Promise; + /** Returns undefined if the form doesn't exist (under the current tenant/RLS). */ + formExists(x: Executor, formId: string): Promise; + listForForm(x: Executor, formId: string): Promise; +} +``` + +- [ ] **Step 2: Create the adapter** (`submissions.repository.drizzle.ts`) + +```ts +import { Injectable } from "@nestjs/common"; +import { desc, eq } from "drizzle-orm"; +import { forms, submissions, type Executor } from "@eventform/db"; +import { + SubmissionListItem, + SubmissionRepository, + SubmissionRow, +} from "./submissions.repository"; + +@Injectable() +export class DrizzleSubmissionRepository implements SubmissionRepository { + listAll(x: Executor): Promise { + return x + .select({ + id: submissions.id, + formId: submissions.formId, + formTitle: forms.title, + answers: submissions.answers, + submittedAt: submissions.submittedAt, + sourceIp: submissions.sourceIp, + }) + .from(submissions) + .innerJoin(forms, eq(forms.id, submissions.formId)) + .orderBy(desc(submissions.submittedAt)) + .limit(200); + } + + async formExists(x: Executor, formId: string): Promise { + const [form] = await x.select({ id: forms.id }).from(forms).where(eq(forms.id, formId)); + return Boolean(form); + } + + listForForm(x: Executor, formId: string): Promise { + return x + .select() + .from(submissions) + .where(eq(submissions.formId, formId)) + .orderBy(desc(submissions.submittedAt)); + } +} +``` + +- [ ] **Step 3: Rewrite the service** (`submissions.service.ts`) — full replacement + +```ts +import { Inject, Injectable, NotFoundException } from "@nestjs/common"; +import { Pool } from "pg"; +import { withTenant } from "@eventform/db"; +import { API_POOL } from "../db/db.module"; +import { SUBMISSION_REPOSITORY, SubmissionRepository } from "./submissions.repository"; + +@Injectable() +export class SubmissionsService { + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(SUBMISSION_REPOSITORY) private readonly repo: SubmissionRepository, + ) {} + + /** All responses across the tenant's forms, newest first (latest 200). */ + listAll(tenantId: string) { + return withTenant(this.pool, tenantId, (db) => this.repo.listAll(db)); + } + + listForForm(tenantId: string, formId: string) { + return withTenant(this.pool, tenantId, async (db) => { + if (!(await this.repo.formExists(db, formId))) { + throw new NotFoundException("form not found"); + } + return this.repo.listForForm(db, formId); + }); + } +} +``` + +- [ ] **Step 4: Provide the repo in the module** (`submissions.module.ts`) + +```ts +import { SUBMISSION_REPOSITORY } from "./submissions.repository"; +import { DrizzleSubmissionRepository } from "./submissions.repository.drizzle"; +// ... +providers: [ + SubmissionsService, + { provide: SUBMISSION_REPOSITORY, useClass: DrizzleSubmissionRepository }, +], +``` + +- [ ] **Step 5: Run the API suite** + +Run: `pnpm --filter @eventform/api test` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add apps/api/src/submissions/ +git commit -m "refactor(api): extract SubmissionRepository port + Drizzle adapter" +``` + +--- + +## Task 4: DeliveryRepository + +**Files:** +- Create: `apps/api/src/deliveries/deliveries.repository.ts` +- Create: `apps/api/src/deliveries/deliveries.repository.drizzle.ts` +- Modify: `apps/api/src/deliveries/deliveries.service.ts` +- Modify: `apps/api/src/deliveries/deliveries.module.ts` + +Note: the manual-retry path writes an `outbox` row **and** updates the delivery in one transaction. The OutboxRepository (Task 6) doesn't exist yet at this point, so this task keeps the outbox insert inside the DeliveryRepository for now via a dedicated `reEmitRetry` method that does both writes on the same executor. Task 6 does NOT need to revisit this (the outbox write stays co-located with the delivery update because they are one atomic unit owned by the deliveries use-case). + +- [ ] **Step 1: Create the port** (`deliveries.repository.ts`) + +```ts +import type { Executor } from "@eventform/db"; +import { deliveries, deliveryAttempts } from "@eventform/db"; +import type { ListDeliveriesQuery } from "./deliveries.schemas"; + +export const DELIVERY_REPOSITORY = "DELIVERY_REPOSITORY"; + +export type DeliveryRow = typeof deliveries.$inferSelect; +export type DeliveryAttemptRow = typeof deliveryAttempts.$inferSelect; + +export interface DeliveryListItem { + id: string; + endpointId: string; + endpointName: string; + payload: DeliveryRow["payload"]; + status: DeliveryRow["status"]; + attemptCount: DeliveryRow["attemptCount"]; + nextRetryAt: DeliveryRow["nextRetryAt"]; + lastError: DeliveryRow["lastError"]; + responseCode: DeliveryRow["responseCode"]; + deliveredAt: DeliveryRow["deliveredAt"]; + createdAt: DeliveryRow["createdAt"]; +} + +export interface DeliveryRepository { + list(x: Executor, tenantId: string, query: ListDeliveriesQuery): Promise; + findById(x: Executor, id: string): Promise; + listAttempts(x: Executor, deliveryId: string): Promise; + /** SELECT ... FOR UPDATE — used to lock the row before a manual retry. */ + findByIdForUpdate(x: Executor, id: string): Promise; + /** + * Manual retry, one atomic unit: insert a fresh outbox row from the stored + * payload and reset the delivery to pending. Both writes run on `x`. + */ + reEmitRetry( + x: Executor, + args: { delivery: DeliveryRow; tenantId: string; eventId: string; payload: DeliveryRow["payload"] }, + ): Promise; +} +``` + +- [ ] **Step 2: Create the adapter** (`deliveries.repository.drizzle.ts`) + +```ts +import { Injectable } from "@nestjs/common"; +import { and, asc, desc, eq, SQL } from "drizzle-orm"; +import { deliveries, deliveryAttempts, endpoints, outbox, type Executor } from "@eventform/db"; +import type { ListDeliveriesQuery } from "./deliveries.schemas"; +import { + DeliveryAttemptRow, + DeliveryListItem, + DeliveryRepository, + DeliveryRow, +} from "./deliveries.repository"; + +@Injectable() +export class DrizzleDeliveryRepository implements DeliveryRepository { + list(x: Executor, tenantId: string, query: ListDeliveriesQuery): Promise { + const conditions: SQL[] = [eq(deliveries.tenantId, tenantId)]; + if (query.status) { + conditions.push(eq(deliveries.status, query.status)); + } + if (query.endpointId) { + conditions.push(eq(deliveries.endpointId, query.endpointId)); + } + return x + .select({ + id: deliveries.id, + endpointId: deliveries.endpointId, + endpointName: endpoints.name, + payload: deliveries.payload, + status: deliveries.status, + attemptCount: deliveries.attemptCount, + nextRetryAt: deliveries.nextRetryAt, + lastError: deliveries.lastError, + responseCode: deliveries.responseCode, + deliveredAt: deliveries.deliveredAt, + createdAt: deliveries.createdAt, + }) + .from(deliveries) + .innerJoin(endpoints, eq(endpoints.id, deliveries.endpointId)) + .where(and(...conditions)) + .orderBy(desc(deliveries.createdAt)) + .limit(200); + } + + async findById(x: Executor, id: string): Promise { + const [delivery] = await x.select().from(deliveries).where(eq(deliveries.id, id)); + return delivery; + } + + listAttempts(x: Executor, deliveryId: string): Promise { + return x + .select() + .from(deliveryAttempts) + .where(eq(deliveryAttempts.deliveryId, deliveryId)) + .orderBy(asc(deliveryAttempts.attemptNo)); + } + + async findByIdForUpdate(x: Executor, id: string): Promise { + const [delivery] = await x.select().from(deliveries).where(eq(deliveries.id, id)).for("update"); + return delivery; + } + + async reEmitRetry( + x: Executor, + args: { delivery: DeliveryRow; tenantId: string; eventId: string; payload: DeliveryRow["payload"] }, + ): Promise { + const { delivery, tenantId, eventId, payload } = args; + await x.insert(outbox).values({ + id: eventId, + tenantId, + aggregateType: "delivery", + aggregateId: delivery.id, + eventType: String((delivery.payload as Record).type ?? "unknown"), + payload, + }); + const [updated] = await x + .update(deliveries) + .set({ + status: "pending", + attemptCount: 0, + payload, + eventId, + nextRetryAt: null, + lastError: null, + responseCode: null, + }) + .where(eq(deliveries.id, delivery.id)) + .returning(); + return updated; + } +} +``` + +- [ ] **Step 3: Rewrite the service** (`deliveries.service.ts`) — full replacement + +```ts +import { randomUUID } from "node:crypto"; +import { ConflictException, Inject, Injectable, NotFoundException } from "@nestjs/common"; +import { Pool } from "pg"; +import { withTenant } from "@eventform/db"; +import { API_POOL } from "../db/db.module"; +import { ListDeliveriesQuery } from "./deliveries.schemas"; +import { DELIVERY_REPOSITORY, DeliveryRepository } from "./deliveries.repository"; + +@Injectable() +export class DeliveriesService { + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(DELIVERY_REPOSITORY) private readonly repo: DeliveryRepository, + ) {} + + list(tenantId: string, query: ListDeliveriesQuery) { + return withTenant(this.pool, tenantId, (db) => this.repo.list(db, tenantId, query)); + } + + getWithAttempts(tenantId: string, id: string) { + return withTenant(this.pool, tenantId, async (db) => { + const delivery = await this.repo.findById(db, id); + if (!delivery) { + throw new NotFoundException("delivery not found"); + } + const attempts = await this.repo.listAttempts(db, id); + return { ...delivery, attempts }; + }); + } + + /** Manual retry: failed-only, FOR UPDATE, reset budget, re-emit through the outbox. */ + retry(tenantId: string, id: string) { + return withTenant(this.pool, tenantId, async (db) => { + const delivery = await this.repo.findByIdForUpdate(db, id); + if (!delivery) { + throw new NotFoundException("delivery not found"); + } + if (delivery.status !== "failed") { + throw new ConflictException("only failed deliveries can be retried"); + } + const eventId = randomUUID(); + const payload = { ...delivery.payload, eventId, attempt: 1 }; + return this.repo.reEmitRetry(db, { delivery, tenantId, eventId, payload }); + }); + } +} +``` + +- [ ] **Step 4: Provide the repo in the module** (`deliveries.module.ts`) + +```ts +import { DELIVERY_REPOSITORY } from "./deliveries.repository"; +import { DrizzleDeliveryRepository } from "./deliveries.repository.drizzle"; +// ... +providers: [ + DeliveriesService, + { provide: DELIVERY_REPOSITORY, useClass: DrizzleDeliveryRepository }, +], +``` + +- [ ] **Step 5: Run the API suite** + +Run: `pnpm --filter @eventform/api test` +Expected: PASS (retry e2e: failed → pending, new outbox row). + +- [ ] **Step 6: Commit** + +```bash +git add apps/api/src/deliveries/ +git commit -m "refactor(api): extract DeliveryRepository port + Drizzle adapter" +``` + +--- + +## Task 5: FormRepository + +**Files:** +- Create: `apps/api/src/forms/forms.repository.ts` +- Create: `apps/api/src/forms/forms.repository.drizzle.ts` +- Modify: `apps/api/src/forms/forms.service.ts` +- Modify: `apps/api/src/forms/forms.module.ts` + +- [ ] **Step 1: Create the port** (`forms.repository.ts`) + +```ts +import type { Executor } from "@eventform/db"; +import { formFields, forms } from "@eventform/db"; + +export const FORM_REPOSITORY = "FORM_REPOSITORY"; + +export type FormRow = typeof forms.$inferSelect; +export type FormFieldRow = typeof formFields.$inferSelect; + +export interface NewFormField { + formId: string; + tenantId: string; + type: FormFieldRow["type"]; + label: string; + options: string[] | null; + required: boolean; + position: number; +} + +export interface FormRepository { + insert(x: Executor, values: { tenantId: string; title: string; publicSlug: string }): Promise; + listByTenant(x: Executor): Promise; + findById(x: Executor, formId: string): Promise; + findByIdForUpdate(x: Executor, formId: string): Promise; + updateTitle(x: Executor, formId: string, title: string): Promise; + setStatus(x: Executor, formId: string, status: FormRow["status"]): Promise; + remove(x: Executor, formId: string): Promise; + listFields(x: Executor, formId: string): Promise; + deleteFields(x: Executor, formId: string): Promise; + insertFields(x: Executor, rows: NewFormField[]): Promise; +} +``` + +- [ ] **Step 2: Create the adapter** (`forms.repository.drizzle.ts`) + +```ts +import { Injectable } from "@nestjs/common"; +import { asc, eq } from "drizzle-orm"; +import { formFields, forms, type Executor } from "@eventform/db"; +import { FormFieldRow, FormRepository, FormRow, NewFormField } from "./forms.repository"; + +@Injectable() +export class DrizzleFormRepository implements FormRepository { + async insert( + x: Executor, + values: { tenantId: string; title: string; publicSlug: string }, + ): Promise { + const [form] = await x.insert(forms).values(values).returning(); + return form; + } + + listByTenant(x: Executor): Promise { + return x.select().from(forms).orderBy(asc(forms.createdAt)); + } + + async findById(x: Executor, formId: string): Promise { + const [form] = await x.select().from(forms).where(eq(forms.id, formId)); + return form; + } + + async findByIdForUpdate(x: Executor, formId: string): Promise { + const [form] = await x.select().from(forms).where(eq(forms.id, formId)).for("update"); + return form; + } + + async updateTitle(x: Executor, formId: string, title: string): Promise { + const [form] = await x.update(forms).set({ title }).where(eq(forms.id, formId)).returning(); + return form; + } + + async setStatus( + x: Executor, + formId: string, + status: FormRow["status"], + ): Promise { + const [form] = await x.update(forms).set({ status }).where(eq(forms.id, formId)).returning(); + return form; + } + + async remove(x: Executor, formId: string): Promise { + await x.delete(forms).where(eq(forms.id, formId)); // fields cascade + } + + listFields(x: Executor, formId: string): Promise { + return x + .select() + .from(formFields) + .where(eq(formFields.formId, formId)) + .orderBy(asc(formFields.position)); + } + + async deleteFields(x: Executor, formId: string): Promise { + await x.delete(formFields).where(eq(formFields.formId, formId)); + } + + insertFields(x: Executor, rows: NewFormField[]): Promise { + return x.insert(formFields).values(rows).returning(); + } +} +``` + +Note: `publish` reads fields without the `position` ordering in the original (`db.select().from(formFields).where(...)` with no orderBy). Reusing `listFields` (which adds an ORDER BY) is behavior-equivalent for the `fields.length === 0` check — the order is irrelevant to a count. Acceptable. + +- [ ] **Step 3: Rewrite the service** (`forms.service.ts`) — full replacement + +```ts +import { randomBytes } from "node:crypto"; +import { ConflictException, Inject, Injectable, NotFoundException } from "@nestjs/common"; +import { Pool } from "pg"; +import { withTenant } from "@eventform/db"; +import { API_POOL } from "../db/db.module"; +import { CreateFormDto, ReplaceFieldsDto } from "./forms.schemas"; +import { FORM_REPOSITORY, FormRepository } from "./forms.repository"; + +@Injectable() +export class FormsService { + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(FORM_REPOSITORY) private readonly repo: FormRepository, + ) {} + + create(tenantId: string, dto: CreateFormDto) { + return withTenant(this.pool, tenantId, (db) => + this.repo.insert(db, { + tenantId, + title: dto.title, + publicSlug: randomBytes(6).toString("base64url"), + }), + ); + } + + list(tenantId: string) { + return withTenant(this.pool, tenantId, (db) => this.repo.listByTenant(db)); + } + + async getWithFields(tenantId: string, formId: string) { + return withTenant(this.pool, tenantId, async (db) => { + const form = await this.repo.findById(db, formId); + if (!form) { + throw new NotFoundException("form not found"); + } + const fields = await this.repo.listFields(db, formId); + return { ...form, fields }; + }); + } + + async updateTitle(tenantId: string, formId: string, dto: CreateFormDto) { + return withTenant(this.pool, tenantId, async (db) => { + const form = await this.repo.updateTitle(db, formId, dto.title); + if (!form) { + throw new NotFoundException("form not found"); + } + return form; + }); + } + + async remove(tenantId: string, formId: string) { + return withTenant(this.pool, tenantId, async (db) => { + const form = await this.repo.findByIdForUpdate(db, formId); + if (!form) { + throw new NotFoundException("form not found"); + } + if (form.status !== "draft") { + throw new ConflictException("published forms cannot be deleted"); + } + await this.repo.remove(db, formId); + }); + } + + async replaceFields(tenantId: string, formId: string, dto: ReplaceFieldsDto) { + return withTenant(this.pool, tenantId, async (db) => { + const form = await this.repo.findByIdForUpdate(db, formId); + if (!form) { + throw new NotFoundException("form not found"); + } + if (form.status !== "draft") { + throw new ConflictException("published forms cannot be edited"); + } + await this.repo.deleteFields(db, formId); + return this.repo.insertFields( + db, + dto.fields.map((f, position) => ({ + formId, + tenantId, + type: f.type, + label: f.label, + options: f.options ?? null, + required: f.required, + position, + })), + ); + }); + } + + async publish(tenantId: string, formId: string) { + return withTenant(this.pool, tenantId, async (db) => { + const form = await this.repo.findByIdForUpdate(db, formId); + if (!form) { + throw new NotFoundException("form not found"); + } + if (form.status !== "draft") { + throw new ConflictException("form is already published"); + } + const fields = await this.repo.listFields(db, formId); + if (fields.length === 0) { + throw new ConflictException("cannot publish a form without fields"); + } + const updated = await this.repo.setStatus(db, formId, "published"); + return updated; + }); + } +} +``` + +- [ ] **Step 4: Provide the repo in the module** (`forms.module.ts`) + +```ts +import { FORM_REPOSITORY } from "./forms.repository"; +import { DrizzleFormRepository } from "./forms.repository.drizzle"; +// ... +providers: [ + FormsService, + { provide: FORM_REPOSITORY, useClass: DrizzleFormRepository }, +], +``` + +- [ ] **Step 5: Run the API suite** + +Run: `pnpm --filter @eventform/api test` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add apps/api/src/forms/ +git commit -m "refactor(api): extract FormRepository port + Drizzle adapter" +``` + +--- + +## Task 6: Public path — FormRepository (anonymous read) + OutboxRepository, the atomic write + +**Files:** +- Create: `apps/api/src/public/public.repository.ts` +- Create: `apps/api/src/public/public.repository.drizzle.ts` +- Modify: `apps/api/src/public/public.service.ts` +- Modify: `apps/api/src/public/public.module.ts` + +This task carries the two riskiest pieces: the **anonymous published-form read** (pool executor, no tenant set, RLS public policy) and the **atomic submit write** (submission + deliveries + outbox under one `withTenant`). Run tests with extra care. + +- [ ] **Step 1: Create the port** (`public.repository.ts`) + +```ts +import type { Executor } from "@eventform/db"; +import type { SubmissionReceivedEvent } from "@eventform/shared"; + +export const PUBLIC_REPOSITORY = "PUBLIC_REPOSITORY"; + +export interface PublicFormRecord { + id: string; + tenantId: string; + title: string; + slug: string; + fields: { + id: string; + type: "text" | "multiple_choice"; + label: string; + options: string[] | null; + required: boolean; + position: number; + }[]; +} + +export interface DeliveryWrite { + deliveryId: string; + endpointId: string; + eventId: string; + payload: SubmissionReceivedEvent; +} + +export interface PublicRepository { + /** + * Anonymous read: runs on a POOL executor (no tenant set) so the RLS + * public-read policies scope to published forms. Returns null if not found. + */ + findPublishedFormBySlug(x: Executor, slug: string): Promise; + /** Active endpoints for the tenant — used to fan out deliveries. */ + listActiveEndpointIds(x: Executor, tenantId: string): Promise; + /** Insert the submission and return its id. */ + insertSubmission( + x: Executor, + row: { formId: string; tenantId: string; answers: Record; sourceIp: string | undefined; submittedAt: Date }, + ): Promise; + /** Insert one delivery + its outbox row (called once per active endpoint, same tx). */ + insertDeliveryWithOutbox(x: Executor, tenantId: string, write: DeliveryWrite): Promise; +} +``` + +- [ ] **Step 2: Create the adapter** (`public.repository.drizzle.ts`) + +```ts +import { Injectable } from "@nestjs/common"; +import { and, eq } from "drizzle-orm"; +import { deliveries, endpoints, outbox, submissions, type Executor } from "@eventform/db"; +import { DeliveryWrite, PublicFormRecord, PublicRepository } from "./public.repository"; + +@Injectable() +export class DrizzlePublicRepository implements PublicRepository { + async findPublishedFormBySlug(x: Executor, slug: string): Promise { + // Raw SQL preserves the exact anonymous read the service used (RLS public + // policies apply because no app.tenant_id is set on a pool executor). + const session = x.$client; // node-postgres Pool/Client behind the Drizzle handle + const form = await session.query( + `SELECT id, tenant_id, title, public_slug FROM forms WHERE public_slug = $1`, + [slug], + ); + if (form.rowCount !== 1) { + return null; + } + const fields = await session.query( + `SELECT id, type, label, options, required, position + FROM form_fields WHERE form_id = $1 ORDER BY position`, + [form.rows[0].id], + ); + return { + id: form.rows[0].id, + tenantId: form.rows[0].tenant_id, + title: form.rows[0].title, + slug: form.rows[0].public_slug, + fields: fields.rows, + }; + } + + async listActiveEndpointIds(x: Executor, tenantId: string): Promise { + const rows = await x + .select({ id: endpoints.id }) + .from(endpoints) + .where(and(eq(endpoints.tenantId, tenantId), eq(endpoints.active, true))); + return rows.map((r) => r.id); + } + + async insertSubmission( + x: Executor, + row: { formId: string; tenantId: string; answers: Record; sourceIp: string | undefined; submittedAt: Date }, + ): Promise { + const [submission] = await x.insert(submissions).values(row).returning(); + return submission.id; + } + + async insertDeliveryWithOutbox(x: Executor, tenantId: string, write: DeliveryWrite): Promise { + await x.insert(deliveries).values({ + id: write.deliveryId, + tenantId, + endpointId: write.endpointId, + payload: write.payload, + eventId: write.eventId, + }); + await x.insert(outbox).values({ + id: write.eventId, + tenantId, + aggregateType: "delivery", + aggregateId: write.deliveryId, + eventType: "submission.received", + payload: write.payload, + }); + } +} +``` + +Note on `x.$client`: Drizzle's node-postgres handle exposes the underlying pool/client as `$client`. For the anonymous read we want raw SQL on the pool (exactly as the original service did). Pass the pool-bound executor from `poolExecutor(pool)`; `$client` is then the `Pool`. If `$client` typing is awkward, the service may instead inject `API_POOL` and pass the `Pool` directly to a `findPublishedFormBySlug(pool, slug)` overload — but prefer `$client` to keep the port uniform on `Executor`. Verify `x.$client.query` exists at build time (Step 4); if not, fall back to injecting `API_POOL` into the adapter and using it for this one method. + +- [ ] **Step 3: Rewrite the service** (`public.service.ts`) — full replacement + +```ts +import { randomUUID } from "node:crypto"; +import { Inject, Injectable, NotFoundException } from "@nestjs/common"; +import { Pool } from "pg"; +import { poolExecutor, withTenant } from "@eventform/db"; +import type { SubmissionReceivedEvent } from "@eventform/shared"; +import { API_POOL } from "../db/db.module"; +import { PUBLIC_REPOSITORY, PublicFormRecord, PublicRepository } from "./public.repository"; + +export interface PublicField { + id: string; + type: "text" | "multiple_choice"; + label: string; + options: string[] | null; + required: boolean; + position: number; +} + +export interface PublicForm { + id: string; + title: string; + slug: string; + fields: PublicField[]; +} + +export interface ResolvedForm extends PublicForm { + tenantId: string; +} + +@Injectable() +export class PublicService { + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(PUBLIC_REPOSITORY) private readonly repo: PublicRepository, + ) {} + + /** Anonymous read — RLS public-read policies scope to published forms. */ + async resolvePublishedForm(slug: string): Promise { + const record = await this.repo.findPublishedFormBySlug(poolExecutor(this.pool), slug); + if (!record) { + throw new NotFoundException("form not found"); + } + return record as PublicFormRecord & ResolvedForm; + } + + toPublicForm(resolved: ResolvedForm): PublicForm { + const { tenantId: _omitted, ...pub } = resolved; + return pub; + } + + async submit( + form: ResolvedForm, + answers: Record, + sourceIp: string | undefined, + ): Promise<{ submissionId: string }> { + const submittedAt = new Date(); + return withTenant(this.pool, form.tenantId, async (db) => { + const submissionId = await this.repo.insertSubmission(db, { + formId: form.id, + tenantId: form.tenantId, + answers, + sourceIp, + submittedAt, + }); + + const endpointIds = await this.repo.listActiveEndpointIds(db, form.tenantId); + for (const endpointId of endpointIds) { + const deliveryId = randomUUID(); + const eventId = randomUUID(); + const payload: SubmissionReceivedEvent = { + eventId, + type: "submission.received", + attempt: 1, + tenantId: form.tenantId, + formId: form.id, + formTitle: form.title, + submissionId, + endpointId, + deliveryId, + answers, + submittedAt: submittedAt.toISOString(), + }; + await this.repo.insertDeliveryWithOutbox(db, form.tenantId, { + deliveryId, + endpointId, + eventId, + payload, + }); + } + return { submissionId }; + }); + } +} +``` + +Note: `resolvePublishedForm` returns the record; the `PublicFormRecord` shape already matches `ResolvedForm` (id/tenantId/title/slug/fields), so the cast is a no-op at runtime. Keep the `PublicField`/`PublicForm`/`ResolvedForm` interfaces in the service since the controller imports them. + +- [ ] **Step 4: Provide the repo + build to check `$client`** (`public.module.ts`) + +```ts +import { PUBLIC_REPOSITORY } from "./public.repository"; +import { DrizzlePublicRepository } from "./public.repository.drizzle"; +// ... +providers: [ + PublicService, + { provide: PUBLIC_REPOSITORY, useClass: DrizzlePublicRepository }, +], +``` + +Run: `pnpm --filter @eventform/api build` +Expected: PASS. If `x.$client.query` fails to typecheck, apply the fallback from Step 2's note (inject `API_POOL` into `DrizzlePublicRepository`, give `findPublishedFormBySlug` the pool from the field instead of `x.$client`), then rebuild. + +- [ ] **Step 5: Run the API suite (atomic write + anonymous read are covered here)** + +Run: `pnpm --filter @eventform/api test` +Expected: PASS — public submit fans out deliveries+outbox atomically; anonymous form fetch still works; the outbox-rollback test (revoking INSERT on outbox rolls back the whole submission) still passes. + +- [ ] **Step 6: Commit** + +```bash +git add apps/api/src/public/ +git commit -m "refactor(api): extract PublicRepository (anonymous read + atomic outbox write)" +``` + +--- + +## Task 7: TenantRepository + +**Files:** +- Create: `apps/api/src/auth/tenants.repository.ts` +- Create: `apps/api/src/auth/tenants.repository.drizzle.ts` +- Modify: `apps/api/src/auth/tenants.service.ts` +- Modify: `apps/api/src/auth/auth.module.ts` + +- [ ] **Step 1: Create the port** (`tenants.repository.ts`) + +```ts +import type { Executor } from "@eventform/db"; + +export const TENANT_REPOSITORY = "TENANT_REPOSITORY"; + +export interface Tenant { + id: string; + name: string; +} + +export interface TenantRepository { + findBySub(x: Executor, sub: string): Promise; + /** INSERT ... ON CONFLICT DO UPDATE — race-safe first-login provisioning. */ + insertBySub(x: Executor, sub: string): Promise; + updateName(x: Executor, tenantId: string, name: string): Promise; +} +``` + +- [ ] **Step 2: Create the adapter** (`tenants.repository.drizzle.ts`) + +Tenant provisioning runs on the pool (no tenant context — it's pre-tenant), so the adapter uses raw SQL via `x.$client` to match today's behavior exactly. + +```ts +import { Injectable } from "@nestjs/common"; +import type { Executor } from "@eventform/db"; +import { Tenant, TenantRepository } from "./tenants.repository"; + +@Injectable() +export class DrizzleTenantRepository implements TenantRepository { + async findBySub(x: Executor, sub: string): Promise { + const res = await x.$client.query("SELECT id, name FROM tenants WHERE cognito_sub = $1", [sub]); + return res.rowCount === 1 ? (res.rows[0] as Tenant) : undefined; + } + + async insertBySub(x: Executor, sub: string): Promise { + const res = await x.$client.query( + `INSERT INTO tenants (name, cognito_sub) VALUES ($1, $2) + ON CONFLICT (cognito_sub) DO UPDATE SET cognito_sub = EXCLUDED.cognito_sub + RETURNING id, name`, + [sub, sub], + ); + return res.rows[0] as Tenant; + } + + async updateName(x: Executor, tenantId: string, name: string): Promise { + const res = await x.$client.query( + "UPDATE tenants SET name = $2 WHERE id = $1 RETURNING id, name", + [tenantId, name], + ); + return res.rows[0] as Tenant; + } +} +``` + +- [ ] **Step 3: Rewrite the service** (`tenants.service.ts`) — full replacement + +```ts +import { Inject, Injectable } from "@nestjs/common"; +import { Pool } from "pg"; +import { poolExecutor } from "@eventform/db"; +import { API_POOL } from "../db/db.module"; +import { TENANT_REPOSITORY, Tenant, TenantRepository } from "./tenants.repository"; + +export type { Tenant } from "./tenants.repository"; + +@Injectable() +export class TenantsService { + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(TENANT_REPOSITORY) private readonly repo: TenantRepository, + ) {} + + async findOrCreateBySub(sub: string): Promise { + const x = poolExecutor(this.pool); + const existing = await this.repo.findBySub(x, sub); + if (existing) { + return existing; + } + return this.repo.insertBySub(x, sub); + } + + /** Display-name update (e.g. from the Google ID token after first login). */ + async updateName(tenantId: string, name: string): Promise { + return this.repo.updateName(poolExecutor(this.pool), tenantId, name); + } +} +``` + +Note: the original `Tenant` interface lived in `tenants.service.ts`; it now lives in the port and is re-exported here so existing importers (e.g. `me.controller.ts`, `current-tenant.decorator.ts`) keep working. Verify importers resolve in Step 5. + +- [ ] **Step 4: Provide the repo in the module** (`auth.module.ts`) + +```ts +import { TENANT_REPOSITORY } from "./tenants.repository"; +import { DrizzleTenantRepository } from "./tenants.repository.drizzle"; +// ... +providers: [ + // ...existing (TenantsService, TOKEN_VERIFIER, guards)... + { provide: TENANT_REPOSITORY, useClass: DrizzleTenantRepository }, +], +``` + +- [ ] **Step 5: Run the API suite** + +Run: `pnpm --filter @eventform/api test` +Expected: PASS (auth/me/tenant-provisioning e2e). + +- [ ] **Step 6: Commit** + +```bash +git add apps/api/src/auth/ +git commit -m "refactor(api): extract TenantRepository port + Drizzle adapter" +``` + +--- + +## Task 8: Final sweep — verify the boundary holds + +**Files:** none (verification only). + +- [ ] **Step 1: No service imports drizzle-orm or pg's Pool for queries** + +Run: `grep -rnE "from \"drizzle-orm\"|drizzle\(" apps/api/src --include="*.service.ts"` +Expected: **no matches** (services no longer import drizzle-orm). `Pool` may still be imported in services that own `withTenant`/`poolExecutor` — that's expected (the service owns transaction policy). Drizzle imports should appear only in `*.repository.drizzle.ts`. + +Run: `grep -rlnE "from \"drizzle-orm\"" apps/api/src` +Expected: only `*.repository.drizzle.ts` files. + +- [ ] **Step 2: Full build + both suites** + +Run: `pnpm --filter @eventform/db build && pnpm --filter @eventform/api build` +Expected: PASS. + +Run: `pnpm --filter @eventform/db test && pnpm --filter @eventform/api test` +Expected: PASS (all green — no behavior change). + +- [ ] **Step 3: Playwright smoke (end-to-end backstop)** + +Bring up the stack per README local quickstart, then: +Run: `pnpm --filter @eventform/web exec playwright test` +Expected: PASS (sign-in → build → publish → submit → delivered). + +- [ ] **Step 4: Commit (if the sweep required any fixes)** + +```bash +git add -A +git commit -m "refactor(api): repository-layer boundary sweep" +``` + +--- + +## Self-Review Notes (author) + +- **Spec coverage:** every spec component maps to a task — `endpoints`(T2), `submissions`(T3), `deliveries`(T4), `forms`(T5), `public`/outbox + anonymous read(T6), `tenants`(T7); the `Executor`/`poolExecutor` foundation(T1); the no-drizzle-in-services guardrail verified(T8). Both atomic paths (submit T6, manual-retry T4) covered. +- **Type consistency:** DI token names (`*_REPOSITORY`), row types (`$inferSelect` aliases), and `Executor` are used identically across port/adapter/service in each task. +- **Risk flagged inline:** `x.$client` for raw-SQL anonymous reads (T6/T7) has a typed fallback (inject `API_POOL`) if it doesn't typecheck. From 9a54f7fdb69ea9dad1f4b4f74cf5be41c77d0f2b Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:20:55 +0530 Subject: [PATCH 03/11] feat(db): add Executor type + poolExecutor for the repository layer --- packages/db/src/client.ts | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 1ddc18d..dff200f 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -4,6 +4,28 @@ import * as schema from "./schema"; export type Db = NodePgDatabase; +/** + * A query executor handed to repositories. It is just a Drizzle handle: + * - tx-bound (from withTenant) for atomic, tenant-scoped work, or + * - pool-bound (from poolExecutor) for anonymous reads under RLS public policies. + * Repositories never open their own connection; the caller chooses the executor. + */ +export type Executor = Db; + +let pooled: Db | undefined; + +/** + * A pool-bound Drizzle handle for non-transactional / anonymous reads (no + * `SET LOCAL app.tenant_id`). Memoized per process. Each query borrows a pool + * connection. Do NOT use for multi-statement atomic work — use withTenant. + */ +export function poolExecutor(pool: Pool): Db { + if (!pooled) { + pooled = drizzle(pool, { schema }); + } + return pooled; +} + export function createPool(connectionString: string): Pool { return new Pool({ connectionString, max: 10 }); } From c012dc60f2b84983c56573d6ee9521e1e44ee458 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:25:44 +0530 Subject: [PATCH 04/11] refactor(api): extract EndpointRepository port + Drizzle adapter --- apps/api/src/endpoints/endpoints.module.ts | 7 +- .../endpoints/endpoints.repository.drizzle.ts | 71 +++++++++++++++++++ .../api/src/endpoints/endpoints.repository.ts | 29 ++++++++ apps/api/src/endpoints/endpoints.service.ts | 59 +++++---------- 4 files changed, 124 insertions(+), 42 deletions(-) create mode 100644 apps/api/src/endpoints/endpoints.repository.drizzle.ts create mode 100644 apps/api/src/endpoints/endpoints.repository.ts diff --git a/apps/api/src/endpoints/endpoints.module.ts b/apps/api/src/endpoints/endpoints.module.ts index 2101996..64523db 100644 --- a/apps/api/src/endpoints/endpoints.module.ts +++ b/apps/api/src/endpoints/endpoints.module.ts @@ -1,9 +1,14 @@ import { Module } from "@nestjs/common"; import { EndpointsController } from "./endpoints.controller"; import { EndpointsService } from "./endpoints.service"; +import { ENDPOINT_REPOSITORY } from "./endpoints.repository"; +import { DrizzleEndpointRepository } from "./endpoints.repository.drizzle"; @Module({ controllers: [EndpointsController], - providers: [EndpointsService], + providers: [ + EndpointsService, + { provide: ENDPOINT_REPOSITORY, useClass: DrizzleEndpointRepository }, + ], }) export class EndpointsModule {} diff --git a/apps/api/src/endpoints/endpoints.repository.drizzle.ts b/apps/api/src/endpoints/endpoints.repository.drizzle.ts new file mode 100644 index 0000000..6f30953 --- /dev/null +++ b/apps/api/src/endpoints/endpoints.repository.drizzle.ts @@ -0,0 +1,71 @@ +import { Injectable } from "@nestjs/common"; +import { and, asc, count, eq } from "drizzle-orm"; +import { endpoints, type Executor } from "@eventform/db"; +import { EndpointRepository, EndpointRow } from "./endpoints.repository"; + +@Injectable() +export class DrizzleEndpointRepository implements EndpointRepository { + async countByTenant(x: Executor, tenantId: string): Promise { + const [{ value }] = await x + .select({ value: count() }) + .from(endpoints) + .where(eq(endpoints.tenantId, tenantId)); + return value; + } + + async insert( + x: Executor, + values: { tenantId: string; name: string; url: string; secretCiphertext: string }, + ): Promise { + const [created] = await x.insert(endpoints).values(values).returning(); + return created; + } + + listByTenant(x: Executor): Promise { + return x.select().from(endpoints).orderBy(asc(endpoints.createdAt)); + } + + async findById(x: Executor, id: string, tenantId: string): Promise { + const [found] = await x + .select() + .from(endpoints) + .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))); + return found; + } + + async update( + x: Executor, + id: string, + tenantId: string, + patch: Partial<{ name: string; url: string; active: boolean }>, + ): Promise { + const [updated] = await x + .update(endpoints) + .set(patch) + .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))) + .returning(); + return updated; + } + + async updateSecret( + x: Executor, + id: string, + tenantId: string, + secretCiphertext: string, + ): Promise { + const [updated] = await x + .update(endpoints) + .set({ secretCiphertext }) + .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))) + .returning(); + return updated; + } + + async remove(x: Executor, id: string, tenantId: string): Promise { + const [removed] = await x + .delete(endpoints) + .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))) + .returning(); + return removed; + } +} diff --git a/apps/api/src/endpoints/endpoints.repository.ts b/apps/api/src/endpoints/endpoints.repository.ts new file mode 100644 index 0000000..2b0107c --- /dev/null +++ b/apps/api/src/endpoints/endpoints.repository.ts @@ -0,0 +1,29 @@ +import type { Executor } from "@eventform/db"; +import { endpoints } from "@eventform/db"; + +export const ENDPOINT_REPOSITORY = "ENDPOINT_REPOSITORY"; + +export type EndpointRow = typeof endpoints.$inferSelect; + +export interface EndpointRepository { + countByTenant(x: Executor, tenantId: string): Promise; + insert( + x: Executor, + values: { tenantId: string; name: string; url: string; secretCiphertext: string }, + ): Promise; + listByTenant(x: Executor): Promise; + findById(x: Executor, id: string, tenantId: string): Promise; + update( + x: Executor, + id: string, + tenantId: string, + patch: Partial<{ name: string; url: string; active: boolean }>, + ): Promise; + updateSecret( + x: Executor, + id: string, + tenantId: string, + secretCiphertext: string, + ): Promise; + remove(x: Executor, id: string, tenantId: string): Promise; +} diff --git a/apps/api/src/endpoints/endpoints.service.ts b/apps/api/src/endpoints/endpoints.service.ts index 8cf0231..26026ba 100644 --- a/apps/api/src/endpoints/endpoints.service.ts +++ b/apps/api/src/endpoints/endpoints.service.ts @@ -1,12 +1,10 @@ import { ConflictException, Inject, Injectable, NotFoundException } from "@nestjs/common"; import { Pool } from "pg"; -import { and, asc, count, eq } from "drizzle-orm"; -import { endpoints, withTenant } from "@eventform/db"; +import { withTenant } from "@eventform/db"; import { generateEndpointSecret, SecretCipher } from "@eventform/shared"; import { API_POOL, SECRET_CIPHER } from "../db/db.module"; import { CreateEndpointDto, UpdateEndpointDto } from "./endpoints.schemas"; - -type EndpointRow = typeof endpoints.$inferSelect; +import { ENDPOINT_REPOSITORY, EndpointRepository, EndpointRow } from "./endpoints.repository"; function publicView(row: EndpointRow) { const { secretCiphertext: _omitted, ...rest } = row; @@ -18,6 +16,7 @@ export class EndpointsService { constructor( @Inject(API_POOL) private readonly pool: Pool, @Inject(SECRET_CIPHER) private readonly cipher: SecretCipher, + @Inject(ENDPOINT_REPOSITORY) private readonly repo: EndpointRepository, ) {} async create(tenantId: string, dto: CreateEndpointDto) { @@ -25,41 +24,26 @@ export class EndpointsService { const secretCiphertext = await this.cipher.encrypt(secret, tenantId); const row = await withTenant(this.pool, tenantId, async (db) => { // Cap endpoints at 20 per tenant. - // NOTE: a loop-of-20-creates e2e would be slow; the cap is documented here - // and verified by code review. A future unit-ish e2e can monkeypatch the - // count query if needed. - const [{ value: existing }] = await db - .select({ value: count() }) - .from(endpoints) - .where(eq(endpoints.tenantId, tenantId)); + const existing = await this.repo.countByTenant(db, tenantId); if (existing >= 20) { throw new ConflictException("endpoint limit reached (20)"); } - const [created] = await db - .insert(endpoints) - .values({ tenantId, name: dto.name, url: dto.url, secretCiphertext }) - .returning(); - return created; + return this.repo.insert(db, { tenantId, name: dto.name, url: dto.url, secretCiphertext }); }); return { ...publicView(row), secret }; } list(tenantId: string) { return withTenant(this.pool, tenantId, async (db) => { - const rows = await db.select().from(endpoints).orderBy(asc(endpoints.createdAt)); + const rows = await this.repo.listByTenant(db); return rows.map(publicView); }); } async update(tenantId: string, id: string, dto: UpdateEndpointDto) { - const row = await withTenant(this.pool, tenantId, async (db) => { - const [updated] = await db - .update(endpoints) - .set(dto) - .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))) - .returning(); - return updated; - }); + const row = await withTenant(this.pool, tenantId, (db) => + this.repo.update(db, id, tenantId, dto), + ); if (!row) { throw new NotFoundException("endpoint not found"); } @@ -67,20 +51,18 @@ export class EndpointsService { } async remove(tenantId: string, id: string) { - const removed = await withTenant(this.pool, tenantId, async (db) => { - const rows = await db.delete(endpoints).where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))).returning(); - return rows[0]; - }); + const removed = await withTenant(this.pool, tenantId, (db) => + this.repo.remove(db, id, tenantId), + ); if (!removed) { throw new NotFoundException("endpoint not found"); } } async revealSecret(tenantId: string, id: string) { - const row = await withTenant(this.pool, tenantId, async (db) => { - const [found] = await db.select().from(endpoints).where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))); - return found; - }); + const row = await withTenant(this.pool, tenantId, (db) => + this.repo.findById(db, id, tenantId), + ); if (!row) { throw new NotFoundException("endpoint not found"); } @@ -91,14 +73,9 @@ export class EndpointsService { async rotateSecret(tenantId: string, id: string) { const secret = generateEndpointSecret(); const secretCiphertext = await this.cipher.encrypt(secret, tenantId); - const row = await withTenant(this.pool, tenantId, async (db) => { - const [updated] = await db - .update(endpoints) - .set({ secretCiphertext }) - .where(and(eq(endpoints.id, id), eq(endpoints.tenantId, tenantId))) - .returning(); - return updated; - }); + const row = await withTenant(this.pool, tenantId, (db) => + this.repo.updateSecret(db, id, tenantId, secretCiphertext), + ); if (!row) { throw new NotFoundException("endpoint not found"); } From 6eebf64622e510a1a00319455a71199a48796514 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:28:12 +0530 Subject: [PATCH 05/11] refactor(api): extract SubmissionRepository port + Drizzle adapter --- .../api/src/submissions/submissions.module.ts | 7 +++- .../submissions.repository.drizzle.ts | 40 +++++++++++++++++++ .../src/submissions/submissions.repository.ts | 23 +++++++++++ .../src/submissions/submissions.service.ts | 34 +++++----------- 4 files changed, 78 insertions(+), 26 deletions(-) create mode 100644 apps/api/src/submissions/submissions.repository.drizzle.ts create mode 100644 apps/api/src/submissions/submissions.repository.ts diff --git a/apps/api/src/submissions/submissions.module.ts b/apps/api/src/submissions/submissions.module.ts index 0b9f523..f0d8946 100644 --- a/apps/api/src/submissions/submissions.module.ts +++ b/apps/api/src/submissions/submissions.module.ts @@ -1,9 +1,14 @@ import { Module } from "@nestjs/common"; import { AllSubmissionsController, SubmissionsController } from "./submissions.controller"; import { SubmissionsService } from "./submissions.service"; +import { SUBMISSION_REPOSITORY } from "./submissions.repository"; +import { DrizzleSubmissionRepository } from "./submissions.repository.drizzle"; @Module({ controllers: [SubmissionsController, AllSubmissionsController], - providers: [SubmissionsService], + providers: [ + SubmissionsService, + { provide: SUBMISSION_REPOSITORY, useClass: DrizzleSubmissionRepository }, + ], }) export class SubmissionsModule {} diff --git a/apps/api/src/submissions/submissions.repository.drizzle.ts b/apps/api/src/submissions/submissions.repository.drizzle.ts new file mode 100644 index 0000000..fd2b5b8 --- /dev/null +++ b/apps/api/src/submissions/submissions.repository.drizzle.ts @@ -0,0 +1,40 @@ +import { Injectable } from "@nestjs/common"; +import { desc, eq } from "drizzle-orm"; +import { forms, submissions, type Executor } from "@eventform/db"; +import { + SubmissionListItem, + SubmissionRepository, + SubmissionRow, +} from "./submissions.repository"; + +@Injectable() +export class DrizzleSubmissionRepository implements SubmissionRepository { + listAll(x: Executor): Promise { + return x + .select({ + id: submissions.id, + formId: submissions.formId, + formTitle: forms.title, + answers: submissions.answers, + submittedAt: submissions.submittedAt, + sourceIp: submissions.sourceIp, + }) + .from(submissions) + .innerJoin(forms, eq(forms.id, submissions.formId)) + .orderBy(desc(submissions.submittedAt)) + .limit(200); + } + + async formExists(x: Executor, formId: string): Promise { + const [form] = await x.select({ id: forms.id }).from(forms).where(eq(forms.id, formId)); + return Boolean(form); + } + + listForForm(x: Executor, formId: string): Promise { + return x + .select() + .from(submissions) + .where(eq(submissions.formId, formId)) + .orderBy(desc(submissions.submittedAt)); + } +} diff --git a/apps/api/src/submissions/submissions.repository.ts b/apps/api/src/submissions/submissions.repository.ts new file mode 100644 index 0000000..c98aad7 --- /dev/null +++ b/apps/api/src/submissions/submissions.repository.ts @@ -0,0 +1,23 @@ +import type { Executor } from "@eventform/db"; +import { submissions } from "@eventform/db"; + +export const SUBMISSION_REPOSITORY = "SUBMISSION_REPOSITORY"; + +export type SubmissionRow = typeof submissions.$inferSelect; + +export interface SubmissionListItem { + id: string; + formId: string; + formTitle: string; + answers: SubmissionRow["answers"]; + submittedAt: SubmissionRow["submittedAt"]; + sourceIp: SubmissionRow["sourceIp"]; +} + +export interface SubmissionRepository { + /** All responses across the tenant's forms, newest first (latest 200). */ + listAll(x: Executor): Promise; + /** Returns undefined if the form doesn't exist (under the current tenant/RLS). */ + formExists(x: Executor, formId: string): Promise; + listForForm(x: Executor, formId: string): Promise; +} diff --git a/apps/api/src/submissions/submissions.service.ts b/apps/api/src/submissions/submissions.service.ts index 6858343..a863d13 100644 --- a/apps/api/src/submissions/submissions.service.ts +++ b/apps/api/src/submissions/submissions.service.ts @@ -1,43 +1,27 @@ import { Inject, Injectable, NotFoundException } from "@nestjs/common"; import { Pool } from "pg"; -import { desc, eq } from "drizzle-orm"; -import { forms, submissions, withTenant } from "@eventform/db"; +import { withTenant } from "@eventform/db"; import { API_POOL } from "../db/db.module"; +import { SUBMISSION_REPOSITORY, SubmissionRepository } from "./submissions.repository"; @Injectable() export class SubmissionsService { - constructor(@Inject(API_POOL) private readonly pool: Pool) {} + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(SUBMISSION_REPOSITORY) private readonly repo: SubmissionRepository, + ) {} /** All responses across the tenant's forms, newest first (latest 200). */ listAll(tenantId: string) { - return withTenant(this.pool, tenantId, (db) => - db - .select({ - id: submissions.id, - formId: submissions.formId, - formTitle: forms.title, - answers: submissions.answers, - submittedAt: submissions.submittedAt, - sourceIp: submissions.sourceIp, - }) - .from(submissions) - .innerJoin(forms, eq(forms.id, submissions.formId)) - .orderBy(desc(submissions.submittedAt)) - .limit(200), - ); + return withTenant(this.pool, tenantId, (db) => this.repo.listAll(db)); } listForForm(tenantId: string, formId: string) { return withTenant(this.pool, tenantId, async (db) => { - const [form] = await db.select().from(forms).where(eq(forms.id, formId)); - if (!form) { + if (!(await this.repo.formExists(db, formId))) { throw new NotFoundException("form not found"); } - return db - .select() - .from(submissions) - .where(eq(submissions.formId, formId)) - .orderBy(desc(submissions.submittedAt)); + return this.repo.listForForm(db, formId); }); } } From 0dcb34ca5f3c6a5b0e5ef476e73a21ee679b3e34 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:30:54 +0530 Subject: [PATCH 06/11] refactor(api): extract DeliveryRepository port + Drizzle adapter --- apps/api/src/deliveries/deliveries.module.ts | 7 +- .../deliveries.repository.drizzle.ts | 89 +++++++++++++++++++ .../src/deliveries/deliveries.repository.ts | 38 ++++++++ apps/api/src/deliveries/deliveries.service.ts | 83 +++-------------- 4 files changed, 144 insertions(+), 73 deletions(-) create mode 100644 apps/api/src/deliveries/deliveries.repository.drizzle.ts create mode 100644 apps/api/src/deliveries/deliveries.repository.ts diff --git a/apps/api/src/deliveries/deliveries.module.ts b/apps/api/src/deliveries/deliveries.module.ts index b462ba4..56f3746 100644 --- a/apps/api/src/deliveries/deliveries.module.ts +++ b/apps/api/src/deliveries/deliveries.module.ts @@ -1,9 +1,14 @@ import { Module } from "@nestjs/common"; import { DeliveriesController } from "./deliveries.controller"; import { DeliveriesService } from "./deliveries.service"; +import { DELIVERY_REPOSITORY } from "./deliveries.repository"; +import { DrizzleDeliveryRepository } from "./deliveries.repository.drizzle"; @Module({ controllers: [DeliveriesController], - providers: [DeliveriesService], + providers: [ + DeliveriesService, + { provide: DELIVERY_REPOSITORY, useClass: DrizzleDeliveryRepository }, + ], }) export class DeliveriesModule {} diff --git a/apps/api/src/deliveries/deliveries.repository.drizzle.ts b/apps/api/src/deliveries/deliveries.repository.drizzle.ts new file mode 100644 index 0000000..8d61cd8 --- /dev/null +++ b/apps/api/src/deliveries/deliveries.repository.drizzle.ts @@ -0,0 +1,89 @@ +import { Injectable } from "@nestjs/common"; +import { and, asc, desc, eq, SQL } from "drizzle-orm"; +import { deliveries, deliveryAttempts, endpoints, outbox, type Executor } from "@eventform/db"; +import type { ListDeliveriesQuery } from "./deliveries.schemas"; +import { + DeliveryAttemptRow, + DeliveryListItem, + DeliveryRepository, + DeliveryRow, +} from "./deliveries.repository"; + +@Injectable() +export class DrizzleDeliveryRepository implements DeliveryRepository { + list(x: Executor, tenantId: string, query: ListDeliveriesQuery): Promise { + const conditions: SQL[] = [eq(deliveries.tenantId, tenantId)]; + if (query.status) { + conditions.push(eq(deliveries.status, query.status)); + } + if (query.endpointId) { + conditions.push(eq(deliveries.endpointId, query.endpointId)); + } + return x + .select({ + id: deliveries.id, + endpointId: deliveries.endpointId, + endpointName: endpoints.name, + payload: deliveries.payload, + status: deliveries.status, + attemptCount: deliveries.attemptCount, + nextRetryAt: deliveries.nextRetryAt, + lastError: deliveries.lastError, + responseCode: deliveries.responseCode, + deliveredAt: deliveries.deliveredAt, + createdAt: deliveries.createdAt, + }) + .from(deliveries) + .innerJoin(endpoints, eq(endpoints.id, deliveries.endpointId)) + .where(and(...conditions)) + .orderBy(desc(deliveries.createdAt)) + .limit(200); + } + + async findById(x: Executor, id: string): Promise { + const [delivery] = await x.select().from(deliveries).where(eq(deliveries.id, id)); + return delivery; + } + + listAttempts(x: Executor, deliveryId: string): Promise { + return x + .select() + .from(deliveryAttempts) + .where(eq(deliveryAttempts.deliveryId, deliveryId)) + .orderBy(asc(deliveryAttempts.attemptNo)); + } + + async findByIdForUpdate(x: Executor, id: string): Promise { + const [delivery] = await x.select().from(deliveries).where(eq(deliveries.id, id)).for("update"); + return delivery; + } + + async reEmitRetry( + x: Executor, + args: { delivery: DeliveryRow; tenantId: string; eventId: string; payload: DeliveryRow["payload"] }, + ): Promise { + const { delivery, tenantId, eventId, payload } = args; + await x.insert(outbox).values({ + id: eventId, + tenantId, + aggregateType: "delivery", + aggregateId: delivery.id, + eventType: String((delivery.payload as Record).type ?? "unknown"), + payload, + }); + const [updated] = await x + .update(deliveries) + .set({ + status: "pending", + attemptCount: 0, + payload, + eventId, + nextRetryAt: null, + lastError: null, + responseCode: null, + }) + .where(eq(deliveries.id, delivery.id)) + .returning(); + return updated; + } +} diff --git a/apps/api/src/deliveries/deliveries.repository.ts b/apps/api/src/deliveries/deliveries.repository.ts new file mode 100644 index 0000000..750d8a6 --- /dev/null +++ b/apps/api/src/deliveries/deliveries.repository.ts @@ -0,0 +1,38 @@ +import type { Executor } from "@eventform/db"; +import { deliveries, deliveryAttempts } from "@eventform/db"; +import type { ListDeliveriesQuery } from "./deliveries.schemas"; + +export const DELIVERY_REPOSITORY = "DELIVERY_REPOSITORY"; + +export type DeliveryRow = typeof deliveries.$inferSelect; +export type DeliveryAttemptRow = typeof deliveryAttempts.$inferSelect; + +export interface DeliveryListItem { + id: string; + endpointId: string; + endpointName: string; + payload: DeliveryRow["payload"]; + status: DeliveryRow["status"]; + attemptCount: DeliveryRow["attemptCount"]; + nextRetryAt: DeliveryRow["nextRetryAt"]; + lastError: DeliveryRow["lastError"]; + responseCode: DeliveryRow["responseCode"]; + deliveredAt: DeliveryRow["deliveredAt"]; + createdAt: DeliveryRow["createdAt"]; +} + +export interface DeliveryRepository { + list(x: Executor, tenantId: string, query: ListDeliveriesQuery): Promise; + findById(x: Executor, id: string): Promise; + listAttempts(x: Executor, deliveryId: string): Promise; + /** SELECT ... FOR UPDATE — used to lock the row before a manual retry. */ + findByIdForUpdate(x: Executor, id: string): Promise; + /** + * Manual retry, one atomic unit: insert a fresh outbox row from the stored + * payload and reset the delivery to pending. Both writes run on `x`. + */ + reEmitRetry( + x: Executor, + args: { delivery: DeliveryRow; tenantId: string; eventId: string; payload: DeliveryRow["payload"] }, + ): Promise; +} diff --git a/apps/api/src/deliveries/deliveries.service.ts b/apps/api/src/deliveries/deliveries.service.ts index 9009ef1..19a5e75 100644 --- a/apps/api/src/deliveries/deliveries.service.ts +++ b/apps/api/src/deliveries/deliveries.service.ts @@ -1,61 +1,29 @@ import { randomUUID } from "node:crypto"; import { ConflictException, Inject, Injectable, NotFoundException } from "@nestjs/common"; import { Pool } from "pg"; -import { and, asc, desc, eq, SQL } from "drizzle-orm"; -import { deliveries, deliveryAttempts, endpoints, outbox, withTenant } from "@eventform/db"; +import { withTenant } from "@eventform/db"; import { API_POOL } from "../db/db.module"; import { ListDeliveriesQuery } from "./deliveries.schemas"; +import { DELIVERY_REPOSITORY, DeliveryRepository } from "./deliveries.repository"; @Injectable() export class DeliveriesService { - constructor(@Inject(API_POOL) private readonly pool: Pool) {} + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(DELIVERY_REPOSITORY) private readonly repo: DeliveryRepository, + ) {} list(tenantId: string, query: ListDeliveriesQuery) { - return withTenant(this.pool, tenantId, async (db) => { - const conditions: SQL[] = [eq(deliveries.tenantId, tenantId)]; - if (query.status) { - conditions.push(eq(deliveries.status, query.status)); - } - if (query.endpointId) { - conditions.push(eq(deliveries.endpointId, query.endpointId)); - } - const rows = await db - .select({ - id: deliveries.id, - endpointId: deliveries.endpointId, - endpointName: endpoints.name, - payload: deliveries.payload, - status: deliveries.status, - attemptCount: deliveries.attemptCount, - nextRetryAt: deliveries.nextRetryAt, - lastError: deliveries.lastError, - responseCode: deliveries.responseCode, - deliveredAt: deliveries.deliveredAt, - createdAt: deliveries.createdAt, - }) - .from(deliveries) - .innerJoin(endpoints, eq(endpoints.id, deliveries.endpointId)) - .where(and(...conditions)) - .orderBy(desc(deliveries.createdAt)) - .limit(200); - return rows; - }); + return withTenant(this.pool, tenantId, (db) => this.repo.list(db, tenantId, query)); } getWithAttempts(tenantId: string, id: string) { return withTenant(this.pool, tenantId, async (db) => { - const [delivery] = await db.select().from(deliveries).where(eq(deliveries.id, id)); + const delivery = await this.repo.findById(db, id); if (!delivery) { throw new NotFoundException("delivery not found"); } - const attempts = await db - .select() - .from(deliveryAttempts) - .where(eq(deliveryAttempts.deliveryId, id)) - .orderBy(asc(deliveryAttempts.attemptNo)); - // delivery.payload is the event body as last emitted — durable on the - // row itself (outbox rows are pruned, so they can't serve historical - // reads), so no joins back into producer tables are needed. + const attempts = await this.repo.listAttempts(db, id); return { ...delivery, attempts }; }); } @@ -63,45 +31,16 @@ export class DeliveriesService { /** Manual retry: failed-only, FOR UPDATE, reset budget, re-emit through the outbox. */ retry(tenantId: string, id: string) { return withTenant(this.pool, tenantId, async (db) => { - const [delivery] = await db - .select() - .from(deliveries) - .where(eq(deliveries.id, id)) - .for("update"); + const delivery = await this.repo.findByIdForUpdate(db, id); if (!delivery) { throw new NotFoundException("delivery not found"); } if (delivery.status !== "failed") { throw new ConflictException("only failed deliveries can be retried"); } - - // Re-emit from the stored payload. The machinery only rewrites the - // envelope fields it owns (eventId, attempt); the body stays whatever - // the producer originally supplied. const eventId = randomUUID(); const payload = { ...delivery.payload, eventId, attempt: 1 }; - await db.insert(outbox).values({ - id: eventId, - tenantId, - aggregateType: "delivery", - aggregateId: delivery.id, - eventType: String(delivery.payload.type ?? "unknown"), - payload, - }); - const [updated] = await db - .update(deliveries) - .set({ - status: "pending", - attemptCount: 0, - payload, - eventId, - nextRetryAt: null, - lastError: null, - responseCode: null, - }) - .where(eq(deliveries.id, id)) - .returning(); - return updated; + return this.repo.reEmitRetry(db, { delivery, tenantId, eventId, payload }); }); } } From eac4f886aa08ac4e62e8d5bb6e53a6d5ba2b88c8 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:33:29 +0530 Subject: [PATCH 07/11] refactor(api): extract FormRepository port + Drizzle adapter --- apps/api/src/forms/forms.module.ts | 7 +- .../api/src/forms/forms.repository.drizzle.ts | 63 +++++++++++++ apps/api/src/forms/forms.repository.ts | 30 +++++++ apps/api/src/forms/forms.service.ts | 90 ++++++++----------- 4 files changed, 135 insertions(+), 55 deletions(-) create mode 100644 apps/api/src/forms/forms.repository.drizzle.ts create mode 100644 apps/api/src/forms/forms.repository.ts diff --git a/apps/api/src/forms/forms.module.ts b/apps/api/src/forms/forms.module.ts index be7d133..86201a5 100644 --- a/apps/api/src/forms/forms.module.ts +++ b/apps/api/src/forms/forms.module.ts @@ -1,9 +1,14 @@ import { Module } from "@nestjs/common"; import { FormsController } from "./forms.controller"; import { FormsService } from "./forms.service"; +import { FORM_REPOSITORY } from "./forms.repository"; +import { DrizzleFormRepository } from "./forms.repository.drizzle"; @Module({ controllers: [FormsController], - providers: [FormsService], + providers: [ + FormsService, + { provide: FORM_REPOSITORY, useClass: DrizzleFormRepository }, + ], }) export class FormsModule {} diff --git a/apps/api/src/forms/forms.repository.drizzle.ts b/apps/api/src/forms/forms.repository.drizzle.ts new file mode 100644 index 0000000..6f8e65a --- /dev/null +++ b/apps/api/src/forms/forms.repository.drizzle.ts @@ -0,0 +1,63 @@ +import { Injectable } from "@nestjs/common"; +import { asc, eq } from "drizzle-orm"; +import { formFields, forms, type Executor } from "@eventform/db"; +import { FormFieldRow, FormRepository, FormRow, NewFormField } from "./forms.repository"; + +@Injectable() +export class DrizzleFormRepository implements FormRepository { + async insert( + x: Executor, + values: { tenantId: string; title: string; publicSlug: string }, + ): Promise { + const [form] = await x.insert(forms).values(values).returning(); + return form; + } + + listByTenant(x: Executor): Promise { + return x.select().from(forms).orderBy(asc(forms.createdAt)); + } + + async findById(x: Executor, formId: string): Promise { + const [form] = await x.select().from(forms).where(eq(forms.id, formId)); + return form; + } + + async findByIdForUpdate(x: Executor, formId: string): Promise { + const [form] = await x.select().from(forms).where(eq(forms.id, formId)).for("update"); + return form; + } + + async updateTitle(x: Executor, formId: string, title: string): Promise { + const [form] = await x.update(forms).set({ title }).where(eq(forms.id, formId)).returning(); + return form; + } + + async setStatus( + x: Executor, + formId: string, + status: FormRow["status"], + ): Promise { + const [form] = await x.update(forms).set({ status }).where(eq(forms.id, formId)).returning(); + return form; + } + + async remove(x: Executor, formId: string): Promise { + await x.delete(forms).where(eq(forms.id, formId)); // fields cascade + } + + listFields(x: Executor, formId: string): Promise { + return x + .select() + .from(formFields) + .where(eq(formFields.formId, formId)) + .orderBy(asc(formFields.position)); + } + + async deleteFields(x: Executor, formId: string): Promise { + await x.delete(formFields).where(eq(formFields.formId, formId)); + } + + insertFields(x: Executor, rows: NewFormField[]): Promise { + return x.insert(formFields).values(rows).returning(); + } +} diff --git a/apps/api/src/forms/forms.repository.ts b/apps/api/src/forms/forms.repository.ts new file mode 100644 index 0000000..3dc9b95 --- /dev/null +++ b/apps/api/src/forms/forms.repository.ts @@ -0,0 +1,30 @@ +import type { Executor } from "@eventform/db"; +import { formFields, forms } from "@eventform/db"; + +export const FORM_REPOSITORY = "FORM_REPOSITORY"; + +export type FormRow = typeof forms.$inferSelect; +export type FormFieldRow = typeof formFields.$inferSelect; + +export interface NewFormField { + formId: string; + tenantId: string; + type: FormFieldRow["type"]; + label: string; + options: string[] | null; + required: boolean; + position: number; +} + +export interface FormRepository { + insert(x: Executor, values: { tenantId: string; title: string; publicSlug: string }): Promise; + listByTenant(x: Executor): Promise; + findById(x: Executor, formId: string): Promise; + findByIdForUpdate(x: Executor, formId: string): Promise; + updateTitle(x: Executor, formId: string, title: string): Promise; + setStatus(x: Executor, formId: string, status: FormRow["status"]): Promise; + remove(x: Executor, formId: string): Promise; + listFields(x: Executor, formId: string): Promise; + deleteFields(x: Executor, formId: string): Promise; + insertFields(x: Executor, rows: NewFormField[]): Promise; +} diff --git a/apps/api/src/forms/forms.service.ts b/apps/api/src/forms/forms.service.ts index a0944a5..dac4c72 100644 --- a/apps/api/src/forms/forms.service.ts +++ b/apps/api/src/forms/forms.service.ts @@ -1,57 +1,46 @@ import { randomBytes } from "node:crypto"; import { ConflictException, Inject, Injectable, NotFoundException } from "@nestjs/common"; import { Pool } from "pg"; -import { asc, eq } from "drizzle-orm"; -import { formFields, forms, withTenant } from "@eventform/db"; +import { withTenant } from "@eventform/db"; import { API_POOL } from "../db/db.module"; import { CreateFormDto, ReplaceFieldsDto } from "./forms.schemas"; +import { FORM_REPOSITORY, FormRepository } from "./forms.repository"; @Injectable() export class FormsService { - constructor(@Inject(API_POOL) private readonly pool: Pool) {} + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(FORM_REPOSITORY) private readonly repo: FormRepository, + ) {} create(tenantId: string, dto: CreateFormDto) { - return withTenant(this.pool, tenantId, async (db) => { - const [form] = await db - .insert(forms) - .values({ - tenantId, - title: dto.title, - publicSlug: randomBytes(6).toString("base64url"), - }) - .returning(); - return form; - }); + return withTenant(this.pool, tenantId, (db) => + this.repo.insert(db, { + tenantId, + title: dto.title, + publicSlug: randomBytes(6).toString("base64url"), + }), + ); } list(tenantId: string) { - return withTenant(this.pool, tenantId, (db) => - db.select().from(forms).orderBy(asc(forms.createdAt)), - ); + return withTenant(this.pool, tenantId, (db) => this.repo.listByTenant(db)); } async getWithFields(tenantId: string, formId: string) { return withTenant(this.pool, tenantId, async (db) => { - const [form] = await db.select().from(forms).where(eq(forms.id, formId)); + const form = await this.repo.findById(db, formId); if (!form) { throw new NotFoundException("form not found"); } - const fields = await db - .select() - .from(formFields) - .where(eq(formFields.formId, formId)) - .orderBy(asc(formFields.position)); + const fields = await this.repo.listFields(db, formId); return { ...form, fields }; }); } async updateTitle(tenantId: string, formId: string, dto: CreateFormDto) { return withTenant(this.pool, tenantId, async (db) => { - const [form] = await db - .update(forms) - .set({ title: dto.title }) - .where(eq(forms.id, formId)) - .returning(); + const form = await this.repo.updateTitle(db, formId, dto.title); if (!form) { throw new NotFoundException("form not found"); } @@ -61,63 +50,56 @@ export class FormsService { async remove(tenantId: string, formId: string) { return withTenant(this.pool, tenantId, async (db) => { - const [form] = await db.select().from(forms).where(eq(forms.id, formId)).for("update"); + const form = await this.repo.findByIdForUpdate(db, formId); if (!form) { throw new NotFoundException("form not found"); } if (form.status !== "draft") { throw new ConflictException("published forms cannot be deleted"); } - await db.delete(forms).where(eq(forms.id, formId)); // fields cascade + await this.repo.remove(db, formId); }); } async replaceFields(tenantId: string, formId: string, dto: ReplaceFieldsDto) { return withTenant(this.pool, tenantId, async (db) => { - const [form] = await db.select().from(forms).where(eq(forms.id, formId)).for("update"); + const form = await this.repo.findByIdForUpdate(db, formId); if (!form) { throw new NotFoundException("form not found"); } if (form.status !== "draft") { throw new ConflictException("published forms cannot be edited"); } - await db.delete(formFields).where(eq(formFields.formId, formId)); - const rows = await db - .insert(formFields) - .values( - dto.fields.map((f, position) => ({ - formId, - tenantId, - type: f.type, - label: f.label, - options: f.options ?? null, - required: f.required, - position, - })), - ) - .returning(); - return rows; + await this.repo.deleteFields(db, formId); + return this.repo.insertFields( + db, + dto.fields.map((f, position) => ({ + formId, + tenantId, + type: f.type, + label: f.label, + options: f.options ?? null, + required: f.required, + position, + })), + ); }); } async publish(tenantId: string, formId: string) { return withTenant(this.pool, tenantId, async (db) => { - const [form] = await db.select().from(forms).where(eq(forms.id, formId)).for("update"); + const form = await this.repo.findByIdForUpdate(db, formId); if (!form) { throw new NotFoundException("form not found"); } if (form.status !== "draft") { throw new ConflictException("form is already published"); } - const fields = await db.select().from(formFields).where(eq(formFields.formId, formId)); + const fields = await this.repo.listFields(db, formId); if (fields.length === 0) { throw new ConflictException("cannot publish a form without fields"); } - const [updated] = await db - .update(forms) - .set({ status: "published" }) - .where(eq(forms.id, formId)) - .returning(); + const updated = await this.repo.setStatus(db, formId, "published"); return updated; }); } From bc2fe08a7e4eff1ca8b6a02e5aac593410b93f0a Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:37:06 +0530 Subject: [PATCH 08/11] refactor(api): extract PublicRepository (anonymous read + atomic outbox write) --- apps/api/src/public/public.module.ts | 8 +- .../src/public/public.repository.drizzle.ts | 72 ++++++++++++++++++ apps/api/src/public/public.repository.ts | 43 +++++++++++ apps/api/src/public/public.service.ts | 76 ++++++------------- 4 files changed, 146 insertions(+), 53 deletions(-) create mode 100644 apps/api/src/public/public.repository.drizzle.ts create mode 100644 apps/api/src/public/public.repository.ts diff --git a/apps/api/src/public/public.module.ts b/apps/api/src/public/public.module.ts index 970e8e4..3a2484a 100644 --- a/apps/api/src/public/public.module.ts +++ b/apps/api/src/public/public.module.ts @@ -2,10 +2,16 @@ import { Module } from "@nestjs/common"; import { PublicController } from "./public.controller"; import { AnswersValidationInterceptor } from "./answers-validation.interceptor"; import { PublicService } from "./public.service"; +import { PUBLIC_REPOSITORY } from "./public.repository"; +import { DrizzlePublicRepository } from "./public.repository.drizzle"; @Module({ controllers: [PublicController], - providers: [PublicService, AnswersValidationInterceptor], + providers: [ + PublicService, + AnswersValidationInterceptor, + { provide: PUBLIC_REPOSITORY, useClass: DrizzlePublicRepository }, + ], exports: [PublicService], }) export class PublicModule {} diff --git a/apps/api/src/public/public.repository.drizzle.ts b/apps/api/src/public/public.repository.drizzle.ts new file mode 100644 index 0000000..6fe0626 --- /dev/null +++ b/apps/api/src/public/public.repository.drizzle.ts @@ -0,0 +1,72 @@ +import { Inject, Injectable } from "@nestjs/common"; +import { and, eq } from "drizzle-orm"; +import { Pool } from "pg"; +import { deliveries, endpoints, outbox, submissions, type Executor } from "@eventform/db"; +import { API_POOL } from "../db/db.module"; +import { DeliveryWrite, PublicFormRecord, PublicRepository } from "./public.repository"; + +@Injectable() +export class DrizzlePublicRepository implements PublicRepository { + constructor(@Inject(API_POOL) private readonly pool: Pool) {} + + async findPublishedFormBySlug(_x: Executor, slug: string): Promise { + // Raw SQL preserves the exact anonymous read the service used (RLS public + // policies apply because no app.tenant_id is set on a pool executor). + // The pool executor passed in (_x) is pool-bound; using the underlying pool + // directly keeps the read off any tenant transaction. + const session = this.pool; + const form = await session.query( + `SELECT id, tenant_id, title, public_slug FROM forms WHERE public_slug = $1`, + [slug], + ); + if (form.rowCount !== 1) { + return null; + } + const fields = await session.query( + `SELECT id, type, label, options, required, position + FROM form_fields WHERE form_id = $1 ORDER BY position`, + [form.rows[0].id], + ); + return { + id: form.rows[0].id, + tenantId: form.rows[0].tenant_id, + title: form.rows[0].title, + slug: form.rows[0].public_slug, + fields: fields.rows, + }; + } + + async listActiveEndpointIds(x: Executor, tenantId: string): Promise { + const rows = await x + .select({ id: endpoints.id }) + .from(endpoints) + .where(and(eq(endpoints.tenantId, tenantId), eq(endpoints.active, true))); + return rows.map((r) => r.id); + } + + async insertSubmission( + x: Executor, + row: { formId: string; tenantId: string; answers: Record; sourceIp: string | undefined; submittedAt: Date }, + ): Promise { + const [submission] = await x.insert(submissions).values(row).returning(); + return submission.id; + } + + async insertDeliveryWithOutbox(x: Executor, tenantId: string, write: DeliveryWrite): Promise { + await x.insert(deliveries).values({ + id: write.deliveryId, + tenantId, + endpointId: write.endpointId, + payload: write.payload, + eventId: write.eventId, + }); + await x.insert(outbox).values({ + id: write.eventId, + tenantId, + aggregateType: "delivery", + aggregateId: write.deliveryId, + eventType: "submission.received", + payload: write.payload, + }); + } +} diff --git a/apps/api/src/public/public.repository.ts b/apps/api/src/public/public.repository.ts new file mode 100644 index 0000000..83f0a05 --- /dev/null +++ b/apps/api/src/public/public.repository.ts @@ -0,0 +1,43 @@ +import type { Executor } from "@eventform/db"; +import type { SubmissionReceivedEvent } from "@eventform/shared"; + +export const PUBLIC_REPOSITORY = "PUBLIC_REPOSITORY"; + +export interface PublicFormRecord { + id: string; + tenantId: string; + title: string; + slug: string; + fields: { + id: string; + type: "text" | "multiple_choice"; + label: string; + options: string[] | null; + required: boolean; + position: number; + }[]; +} + +export interface DeliveryWrite { + deliveryId: string; + endpointId: string; + eventId: string; + payload: SubmissionReceivedEvent; +} + +export interface PublicRepository { + /** + * Anonymous read: runs on a POOL executor (no tenant set) so the RLS + * public-read policies scope to published forms. Returns null if not found. + */ + findPublishedFormBySlug(x: Executor, slug: string): Promise; + /** Active endpoints for the tenant — used to fan out deliveries. */ + listActiveEndpointIds(x: Executor, tenantId: string): Promise; + /** Insert the submission and return its id. */ + insertSubmission( + x: Executor, + row: { formId: string; tenantId: string; answers: Record; sourceIp: string | undefined; submittedAt: Date }, + ): Promise; + /** Insert one delivery + its outbox row (called once per active endpoint, same tx). */ + insertDeliveryWithOutbox(x: Executor, tenantId: string, write: DeliveryWrite): Promise; +} diff --git a/apps/api/src/public/public.service.ts b/apps/api/src/public/public.service.ts index 7a59d4a..cf78191 100644 --- a/apps/api/src/public/public.service.ts +++ b/apps/api/src/public/public.service.ts @@ -1,10 +1,10 @@ import { randomUUID } from "node:crypto"; import { Inject, Injectable, NotFoundException } from "@nestjs/common"; import { Pool } from "pg"; -import { and, eq } from "drizzle-orm"; -import { deliveries, endpoints, outbox, submissions, withTenant } from "@eventform/db"; +import { poolExecutor, withTenant } from "@eventform/db"; import type { SubmissionReceivedEvent } from "@eventform/shared"; import { API_POOL } from "../db/db.module"; +import { PUBLIC_REPOSITORY, PublicFormRecord, PublicRepository } from "./public.repository"; export interface PublicField { id: string; @@ -22,36 +22,24 @@ export interface PublicForm { fields: PublicField[]; } -/** Internal shape — includes tenantId for the submit path; never returned by controllers. */ export interface ResolvedForm extends PublicForm { tenantId: string; } @Injectable() export class PublicService { - constructor(@Inject(API_POOL) private readonly pool: Pool) {} + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(PUBLIC_REPOSITORY) private readonly repo: PublicRepository, + ) {} /** Anonymous read — RLS public-read policies scope to published forms. */ async resolvePublishedForm(slug: string): Promise { - const form = await this.pool.query( - `SELECT id, tenant_id, title, public_slug FROM forms WHERE public_slug = $1`, - [slug], - ); - if (form.rowCount !== 1) { + const record = await this.repo.findPublishedFormBySlug(poolExecutor(this.pool), slug); + if (!record) { throw new NotFoundException("form not found"); } - const fields = await this.pool.query( - `SELECT id, type, label, options, required, position - FROM form_fields WHERE form_id = $1 ORDER BY position`, - [form.rows[0].id], - ); - return { - id: form.rows[0].id, - tenantId: form.rows[0].tenant_id, - title: form.rows[0].title, - slug: form.rows[0].public_slug, - fields: fields.rows, - }; + return record as PublicFormRecord & ResolvedForm; } toPublicForm(resolved: ResolvedForm): PublicForm { @@ -59,30 +47,23 @@ export class PublicService { return pub; } - /** - * Pure persistence: input validation (400s) happens upstream in the zod - * pipe + AnswersValidationInterceptor. This method only writes the atomic - * submission + deliveries + outbox set for an already-validated form. - */ async submit( form: ResolvedForm, answers: Record, sourceIp: string | undefined, ): Promise<{ submissionId: string }> { const submittedAt = new Date(); - return withTenant(this.pool, form.tenantId, async (db) => { - const [submission] = await db - .insert(submissions) - .values({ formId: form.id, tenantId: form.tenantId, answers, sourceIp, submittedAt }) - .returning(); - - const activeEndpoints = await db - .select() - .from(endpoints) - .where(and(eq(endpoints.tenantId, form.tenantId), eq(endpoints.active, true))); + const submissionId = await this.repo.insertSubmission(db, { + formId: form.id, + tenantId: form.tenantId, + answers, + sourceIp, + submittedAt, + }); - for (const endpoint of activeEndpoints) { + const endpointIds = await this.repo.listActiveEndpointIds(db, form.tenantId); + for (const endpointId of endpointIds) { const deliveryId = randomUUID(); const eventId = randomUUID(); const payload: SubmissionReceivedEvent = { @@ -92,29 +73,20 @@ export class PublicService { tenantId: form.tenantId, formId: form.id, formTitle: form.title, - submissionId: submission.id, - endpointId: endpoint.id, + submissionId, + endpointId, deliveryId, answers, submittedAt: submittedAt.toISOString(), }; - await db.insert(deliveries).values({ - id: deliveryId, - tenantId: form.tenantId, - endpointId: endpoint.id, - payload, + await this.repo.insertDeliveryWithOutbox(db, form.tenantId, { + deliveryId, + endpointId, eventId, - }); - await db.insert(outbox).values({ - id: eventId, - tenantId: form.tenantId, - aggregateType: "delivery", - aggregateId: deliveryId, - eventType: "submission.received", payload, }); } - return { submissionId: submission.id }; + return { submissionId }; }); } } From 3385dc009242ada7d1eb65959b88bde28153b2d8 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:39:47 +0530 Subject: [PATCH 09/11] refactor(api): extract TenantRepository port + Drizzle adapter --- apps/api/src/auth/auth.module.ts | 3 ++ .../src/auth/tenants.repository.drizzle.ts | 33 ++++++++++++++++ apps/api/src/auth/tenants.repository.ts | 15 +++++++ apps/api/src/auth/tenants.service.ts | 39 +++++++------------ 4 files changed, 64 insertions(+), 26 deletions(-) create mode 100644 apps/api/src/auth/tenants.repository.drizzle.ts create mode 100644 apps/api/src/auth/tenants.repository.ts diff --git a/apps/api/src/auth/auth.module.ts b/apps/api/src/auth/auth.module.ts index a6d3e18..85fa835 100644 --- a/apps/api/src/auth/auth.module.ts +++ b/apps/api/src/auth/auth.module.ts @@ -7,11 +7,14 @@ import { DevTokenVerifier } from "./dev-token-verifier"; import { MeController } from "./me.controller"; import { TenantsService } from "./tenants.service"; import { TOKEN_VERIFIER } from "./token-verifier"; +import { TENANT_REPOSITORY } from "./tenants.repository"; +import { DrizzleTenantRepository } from "./tenants.repository.drizzle"; @Module({ controllers: [MeController], providers: [ TenantsService, + { provide: TENANT_REPOSITORY, useClass: DrizzleTenantRepository }, { provide: TOKEN_VERIFIER, useFactory: () => { diff --git a/apps/api/src/auth/tenants.repository.drizzle.ts b/apps/api/src/auth/tenants.repository.drizzle.ts new file mode 100644 index 0000000..ad55402 --- /dev/null +++ b/apps/api/src/auth/tenants.repository.drizzle.ts @@ -0,0 +1,33 @@ +import { Inject, Injectable } from "@nestjs/common"; +import { Pool } from "pg"; +import type { Executor } from "@eventform/db"; +import { API_POOL } from "../db/db.module"; +import { Tenant, TenantRepository } from "./tenants.repository"; + +@Injectable() +export class DrizzleTenantRepository implements TenantRepository { + constructor(@Inject(API_POOL) private readonly pool: Pool) {} + + async findBySub(_x: Executor, sub: string): Promise { + const res = await this.pool.query("SELECT id, name FROM tenants WHERE cognito_sub = $1", [sub]); + return res.rowCount === 1 ? (res.rows[0] as Tenant) : undefined; + } + + async insertBySub(_x: Executor, sub: string): Promise { + const res = await this.pool.query( + `INSERT INTO tenants (name, cognito_sub) VALUES ($1, $2) + ON CONFLICT (cognito_sub) DO UPDATE SET cognito_sub = EXCLUDED.cognito_sub + RETURNING id, name`, + [sub, sub], + ); + return res.rows[0] as Tenant; + } + + async updateName(_x: Executor, tenantId: string, name: string): Promise { + const res = await this.pool.query( + "UPDATE tenants SET name = $2 WHERE id = $1 RETURNING id, name", + [tenantId, name], + ); + return res.rows[0] as Tenant; + } +} diff --git a/apps/api/src/auth/tenants.repository.ts b/apps/api/src/auth/tenants.repository.ts new file mode 100644 index 0000000..3794813 --- /dev/null +++ b/apps/api/src/auth/tenants.repository.ts @@ -0,0 +1,15 @@ +import type { Executor } from "@eventform/db"; + +export const TENANT_REPOSITORY = "TENANT_REPOSITORY"; + +export interface Tenant { + id: string; + name: string; +} + +export interface TenantRepository { + findBySub(x: Executor, sub: string): Promise; + /** INSERT ... ON CONFLICT DO UPDATE — race-safe first-login provisioning. */ + insertBySub(x: Executor, sub: string): Promise; + updateName(x: Executor, tenantId: string, name: string): Promise; +} diff --git a/apps/api/src/auth/tenants.service.ts b/apps/api/src/auth/tenants.service.ts index 5a4949b..36c8d8e 100644 --- a/apps/api/src/auth/tenants.service.ts +++ b/apps/api/src/auth/tenants.service.ts @@ -1,42 +1,29 @@ import { Inject, Injectable } from "@nestjs/common"; import { Pool } from "pg"; +import { poolExecutor } from "@eventform/db"; import { API_POOL } from "../db/db.module"; +import { TENANT_REPOSITORY, Tenant, TenantRepository } from "./tenants.repository"; -export interface Tenant { - id: string; - name: string; -} +export type { Tenant } from "./tenants.repository"; @Injectable() export class TenantsService { - constructor(@Inject(API_POOL) private readonly pool: Pool) {} + constructor( + @Inject(API_POOL) private readonly pool: Pool, + @Inject(TENANT_REPOSITORY) private readonly repo: TenantRepository, + ) {} async findOrCreateBySub(sub: string): Promise { - const existing = await this.pool.query( - "SELECT id, name FROM tenants WHERE cognito_sub = $1", - [sub], - ); - if (existing.rowCount === 1) { - return existing.rows[0]; + const x = poolExecutor(this.pool); + const existing = await this.repo.findBySub(x, sub); + if (existing) { + return existing; } - // INSERT with a no-op DO UPDATE guards the concurrent-first-login race: - // if two requests arrive simultaneously for a new sub, the loser of the - // INSERT still returns a row rather than silently discarding it. - const res = await this.pool.query( - `INSERT INTO tenants (name, cognito_sub) VALUES ($1, $2) - ON CONFLICT (cognito_sub) DO UPDATE SET cognito_sub = EXCLUDED.cognito_sub - RETURNING id, name`, - [sub, sub], - ); - return res.rows[0]; + return this.repo.insertBySub(x, sub); } /** Display-name update (e.g. from the Google ID token after first login). */ async updateName(tenantId: string, name: string): Promise { - const res = await this.pool.query( - "UPDATE tenants SET name = $2 WHERE id = $1 RETURNING id, name", - [tenantId, name], - ); - return res.rows[0]; + return this.repo.updateName(poolExecutor(this.pool), tenantId, name); } } From 1656dcd33c38d0b239f40948ef841969890b5408 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 14:47:22 +0530 Subject: [PATCH 10/11] refactor(api): drop unused Executor param from pool-only repository methods --- apps/api/src/auth/tenants.repository.drizzle.ts | 7 +++---- apps/api/src/auth/tenants.repository.ts | 8 +++----- apps/api/src/auth/tenants.service.ts | 11 +++-------- apps/api/src/public/public.repository.drizzle.ts | 7 +++---- apps/api/src/public/public.repository.ts | 4 ++-- apps/api/src/public/public.service.ts | 4 ++-- 6 files changed, 16 insertions(+), 25 deletions(-) diff --git a/apps/api/src/auth/tenants.repository.drizzle.ts b/apps/api/src/auth/tenants.repository.drizzle.ts index ad55402..bacdfc4 100644 --- a/apps/api/src/auth/tenants.repository.drizzle.ts +++ b/apps/api/src/auth/tenants.repository.drizzle.ts @@ -1,6 +1,5 @@ import { Inject, Injectable } from "@nestjs/common"; import { Pool } from "pg"; -import type { Executor } from "@eventform/db"; import { API_POOL } from "../db/db.module"; import { Tenant, TenantRepository } from "./tenants.repository"; @@ -8,12 +7,12 @@ import { Tenant, TenantRepository } from "./tenants.repository"; export class DrizzleTenantRepository implements TenantRepository { constructor(@Inject(API_POOL) private readonly pool: Pool) {} - async findBySub(_x: Executor, sub: string): Promise { + async findBySub(sub: string): Promise { const res = await this.pool.query("SELECT id, name FROM tenants WHERE cognito_sub = $1", [sub]); return res.rowCount === 1 ? (res.rows[0] as Tenant) : undefined; } - async insertBySub(_x: Executor, sub: string): Promise { + async insertBySub(sub: string): Promise { const res = await this.pool.query( `INSERT INTO tenants (name, cognito_sub) VALUES ($1, $2) ON CONFLICT (cognito_sub) DO UPDATE SET cognito_sub = EXCLUDED.cognito_sub @@ -23,7 +22,7 @@ export class DrizzleTenantRepository implements TenantRepository { return res.rows[0] as Tenant; } - async updateName(_x: Executor, tenantId: string, name: string): Promise { + async updateName(tenantId: string, name: string): Promise { const res = await this.pool.query( "UPDATE tenants SET name = $2 WHERE id = $1 RETURNING id, name", [tenantId, name], diff --git a/apps/api/src/auth/tenants.repository.ts b/apps/api/src/auth/tenants.repository.ts index 3794813..57ba6f7 100644 --- a/apps/api/src/auth/tenants.repository.ts +++ b/apps/api/src/auth/tenants.repository.ts @@ -1,5 +1,3 @@ -import type { Executor } from "@eventform/db"; - export const TENANT_REPOSITORY = "TENANT_REPOSITORY"; export interface Tenant { @@ -8,8 +6,8 @@ export interface Tenant { } export interface TenantRepository { - findBySub(x: Executor, sub: string): Promise; + findBySub(sub: string): Promise; /** INSERT ... ON CONFLICT DO UPDATE — race-safe first-login provisioning. */ - insertBySub(x: Executor, sub: string): Promise; - updateName(x: Executor, tenantId: string, name: string): Promise; + insertBySub(sub: string): Promise; + updateName(tenantId: string, name: string): Promise; } diff --git a/apps/api/src/auth/tenants.service.ts b/apps/api/src/auth/tenants.service.ts index 36c8d8e..2942351 100644 --- a/apps/api/src/auth/tenants.service.ts +++ b/apps/api/src/auth/tenants.service.ts @@ -1,7 +1,4 @@ import { Inject, Injectable } from "@nestjs/common"; -import { Pool } from "pg"; -import { poolExecutor } from "@eventform/db"; -import { API_POOL } from "../db/db.module"; import { TENANT_REPOSITORY, Tenant, TenantRepository } from "./tenants.repository"; export type { Tenant } from "./tenants.repository"; @@ -9,21 +6,19 @@ export type { Tenant } from "./tenants.repository"; @Injectable() export class TenantsService { constructor( - @Inject(API_POOL) private readonly pool: Pool, @Inject(TENANT_REPOSITORY) private readonly repo: TenantRepository, ) {} async findOrCreateBySub(sub: string): Promise { - const x = poolExecutor(this.pool); - const existing = await this.repo.findBySub(x, sub); + const existing = await this.repo.findBySub(sub); if (existing) { return existing; } - return this.repo.insertBySub(x, sub); + return this.repo.insertBySub(sub); } /** Display-name update (e.g. from the Google ID token after first login). */ async updateName(tenantId: string, name: string): Promise { - return this.repo.updateName(poolExecutor(this.pool), tenantId, name); + return this.repo.updateName(tenantId, name); } } diff --git a/apps/api/src/public/public.repository.drizzle.ts b/apps/api/src/public/public.repository.drizzle.ts index 6fe0626..e714d86 100644 --- a/apps/api/src/public/public.repository.drizzle.ts +++ b/apps/api/src/public/public.repository.drizzle.ts @@ -9,11 +9,10 @@ import { DeliveryWrite, PublicFormRecord, PublicRepository } from "./public.repo export class DrizzlePublicRepository implements PublicRepository { constructor(@Inject(API_POOL) private readonly pool: Pool) {} - async findPublishedFormBySlug(_x: Executor, slug: string): Promise { + async findPublishedFormBySlug(slug: string): Promise { // Raw SQL preserves the exact anonymous read the service used (RLS public - // policies apply because no app.tenant_id is set on a pool executor). - // The pool executor passed in (_x) is pool-bound; using the underlying pool - // directly keeps the read off any tenant transaction. + // policies apply because no app.tenant_id is set). This method always runs + // on the pool (no tenant context) — no Executor param is accepted. const session = this.pool; const form = await session.query( `SELECT id, tenant_id, title, public_slug FROM forms WHERE public_slug = $1`, diff --git a/apps/api/src/public/public.repository.ts b/apps/api/src/public/public.repository.ts index 83f0a05..be0e07d 100644 --- a/apps/api/src/public/public.repository.ts +++ b/apps/api/src/public/public.repository.ts @@ -27,10 +27,10 @@ export interface DeliveryWrite { export interface PublicRepository { /** - * Anonymous read: runs on a POOL executor (no tenant set) so the RLS + * Anonymous read: always runs on the pool (no tenant context) so the RLS * public-read policies scope to published forms. Returns null if not found. */ - findPublishedFormBySlug(x: Executor, slug: string): Promise; + findPublishedFormBySlug(slug: string): Promise; /** Active endpoints for the tenant — used to fan out deliveries. */ listActiveEndpointIds(x: Executor, tenantId: string): Promise; /** Insert the submission and return its id. */ diff --git a/apps/api/src/public/public.service.ts b/apps/api/src/public/public.service.ts index cf78191..5c2acba 100644 --- a/apps/api/src/public/public.service.ts +++ b/apps/api/src/public/public.service.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { Inject, Injectable, NotFoundException } from "@nestjs/common"; import { Pool } from "pg"; -import { poolExecutor, withTenant } from "@eventform/db"; +import { withTenant } from "@eventform/db"; import type { SubmissionReceivedEvent } from "@eventform/shared"; import { API_POOL } from "../db/db.module"; import { PUBLIC_REPOSITORY, PublicFormRecord, PublicRepository } from "./public.repository"; @@ -35,7 +35,7 @@ export class PublicService { /** Anonymous read — RLS public-read policies scope to published forms. */ async resolvePublishedForm(slug: string): Promise { - const record = await this.repo.findPublishedFormBySlug(poolExecutor(this.pool), slug); + const record = await this.repo.findPublishedFormBySlug(slug); if (!record) { throw new NotFoundException("form not found"); } From 1a365e74e093f2589a5f7d5def63e65e98cfb392 Mon Sep 17 00:00:00 2001 From: murugu21 Date: Tue, 16 Jun 2026 15:20:03 +0530 Subject: [PATCH 11/11] =?UTF-8?q?refactor(api):=20minimal=20domain=20layer?= =?UTF-8?q?=20=E2=80=94=20pure=20event/rule=20functions=20+=20value=20obje?= =?UTF-8?q?ct?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract the genuinely-pure domain bits into a framework-free apps/api/src/domain/ (no NestJS, no Drizzle): buildSubmissionReceivedEvent/rebuildRetryEvent (outbox payloads), canRetryDelivery (retry rule), and an EndpointSecret value object. Services call these instead of inlining; they keep owning transactions + HTTP errors. Adds pure unit tests (nothing to mock). No behavior change. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/api/src/deliveries/deliveries.service.ts | 6 +- apps/api/src/domain/delivery-rules.ts | 12 ++++ apps/api/src/domain/endpoint-secret.ts | 21 ++++++ apps/api/src/domain/submission-event.ts | 58 +++++++++++++++ apps/api/src/endpoints/endpoints.service.ts | 15 ++-- apps/api/src/public/public.service.ts | 15 ++-- apps/api/test/domain.test.ts | 71 +++++++++++++++++++ 7 files changed, 178 insertions(+), 20 deletions(-) create mode 100644 apps/api/src/domain/delivery-rules.ts create mode 100644 apps/api/src/domain/endpoint-secret.ts create mode 100644 apps/api/src/domain/submission-event.ts create mode 100644 apps/api/test/domain.test.ts diff --git a/apps/api/src/deliveries/deliveries.service.ts b/apps/api/src/deliveries/deliveries.service.ts index 19a5e75..d5b24f2 100644 --- a/apps/api/src/deliveries/deliveries.service.ts +++ b/apps/api/src/deliveries/deliveries.service.ts @@ -3,6 +3,8 @@ import { ConflictException, Inject, Injectable, NotFoundException } from "@nestj import { Pool } from "pg"; import { withTenant } from "@eventform/db"; import { API_POOL } from "../db/db.module"; +import { canRetryDelivery } from "../domain/delivery-rules"; +import { rebuildRetryEvent } from "../domain/submission-event"; import { ListDeliveriesQuery } from "./deliveries.schemas"; import { DELIVERY_REPOSITORY, DeliveryRepository } from "./deliveries.repository"; @@ -35,11 +37,11 @@ export class DeliveriesService { if (!delivery) { throw new NotFoundException("delivery not found"); } - if (delivery.status !== "failed") { + if (!canRetryDelivery(delivery.status)) { throw new ConflictException("only failed deliveries can be retried"); } const eventId = randomUUID(); - const payload = { ...delivery.payload, eventId, attempt: 1 }; + const payload = rebuildRetryEvent(delivery.payload, eventId); return this.repo.reEmitRetry(db, { delivery, tenantId, eventId, payload }); }); } diff --git a/apps/api/src/domain/delivery-rules.ts b/apps/api/src/domain/delivery-rules.ts new file mode 100644 index 0000000..ff4467e --- /dev/null +++ b/apps/api/src/domain/delivery-rules.ts @@ -0,0 +1,12 @@ +/** + * Delivery lifecycle rules (framework-free, pure). + */ + +/** + * Only a `failed` delivery may be manually retried — `pending`/`retrying` are + * still in flight and `delivered` is terminal-success. Accepts the raw status + * string so the domain stays decoupled from the persistence row type. + */ +export function canRetryDelivery(status: string): boolean { + return status === "failed"; +} diff --git a/apps/api/src/domain/endpoint-secret.ts b/apps/api/src/domain/endpoint-secret.ts new file mode 100644 index 0000000..22cc0ba --- /dev/null +++ b/apps/api/src/domain/endpoint-secret.ts @@ -0,0 +1,21 @@ +import { generateEndpointSecret } from "@eventform/shared"; + +/** + * Webhook signing secret as a value object: an immutable `whsec_`-prefixed + * token. Constructing it only via `generate()` keeps "this string is a webhook + * secret" a named concept at call sites rather than a bare `string`. It wraps + * the shared generator so the format has a single source of truth. + */ +export class EndpointSecret { + static readonly PREFIX = "whsec_"; + + private constructor(public readonly value: string) {} + + static generate(): EndpointSecret { + return new EndpointSecret(generateEndpointSecret()); + } + + toString(): string { + return this.value; + } +} diff --git a/apps/api/src/domain/submission-event.ts b/apps/api/src/domain/submission-event.ts new file mode 100644 index 0000000..08cc069 --- /dev/null +++ b/apps/api/src/domain/submission-event.ts @@ -0,0 +1,58 @@ +import { randomUUID } from "node:crypto"; +import type { SubmissionReceivedEvent } from "@eventform/shared"; + +/** + * Domain layer (framework-free): the rules that define EventForm's events and + * delivery behaviour, expressed as pure functions over plain values. No NestJS, + * no Drizzle — services call these and keep owning transactions + HTTP errors. + */ + +export interface SubmissionEventInput { + tenantId: string; + formId: string; + formTitle: string; + submissionId: string; + endpointId: string; + answers: Record; + submittedAt: Date; +} + +/** + * Construct a fresh `submission.received` event for one (submission, endpoint) + * pair. Mints the event id and delivery id (so a submission fanning out to N + * endpoints yields N distinct events), and stamps attempt = 1. + */ +export function buildSubmissionReceivedEvent(input: SubmissionEventInput): { + deliveryId: string; + eventId: string; + payload: SubmissionReceivedEvent; +} { + const deliveryId = randomUUID(); + const eventId = randomUUID(); + const payload: SubmissionReceivedEvent = { + eventId, + type: "submission.received", + attempt: 1, + tenantId: input.tenantId, + formId: input.formId, + formTitle: input.formTitle, + submissionId: input.submissionId, + endpointId: input.endpointId, + deliveryId, + answers: input.answers, + submittedAt: input.submittedAt.toISOString(), + }; + return { deliveryId, eventId, payload }; +} + +/** + * Manual-retry envelope: re-emit the same stored body with a fresh event id and + * attempt reset to 1. The machinery only rewrites the two envelope fields it + * owns (`eventId`, `attempt`); the producer-supplied body is preserved. + */ +export function rebuildRetryEvent( + payload: T, + eventId: string, +): T & { eventId: string; attempt: number } { + return { ...payload, eventId, attempt: 1 }; +} diff --git a/apps/api/src/endpoints/endpoints.service.ts b/apps/api/src/endpoints/endpoints.service.ts index 26026ba..87ac8ea 100644 --- a/apps/api/src/endpoints/endpoints.service.ts +++ b/apps/api/src/endpoints/endpoints.service.ts @@ -1,8 +1,9 @@ import { ConflictException, Inject, Injectable, NotFoundException } from "@nestjs/common"; import { Pool } from "pg"; import { withTenant } from "@eventform/db"; -import { generateEndpointSecret, SecretCipher } from "@eventform/shared"; +import { SecretCipher } from "@eventform/shared"; import { API_POOL, SECRET_CIPHER } from "../db/db.module"; +import { EndpointSecret } from "../domain/endpoint-secret"; import { CreateEndpointDto, UpdateEndpointDto } from "./endpoints.schemas"; import { ENDPOINT_REPOSITORY, EndpointRepository, EndpointRow } from "./endpoints.repository"; @@ -20,8 +21,8 @@ export class EndpointsService { ) {} async create(tenantId: string, dto: CreateEndpointDto) { - const secret = generateEndpointSecret(); - const secretCiphertext = await this.cipher.encrypt(secret, tenantId); + const secret = EndpointSecret.generate(); + const secretCiphertext = await this.cipher.encrypt(secret.value, tenantId); const row = await withTenant(this.pool, tenantId, async (db) => { // Cap endpoints at 20 per tenant. const existing = await this.repo.countByTenant(db, tenantId); @@ -30,7 +31,7 @@ export class EndpointsService { } return this.repo.insert(db, { tenantId, name: dto.name, url: dto.url, secretCiphertext }); }); - return { ...publicView(row), secret }; + return { ...publicView(row), secret: secret.value }; } list(tenantId: string) { @@ -71,14 +72,14 @@ export class EndpointsService { } async rotateSecret(tenantId: string, id: string) { - const secret = generateEndpointSecret(); - const secretCiphertext = await this.cipher.encrypt(secret, tenantId); + const secret = EndpointSecret.generate(); + const secretCiphertext = await this.cipher.encrypt(secret.value, tenantId); const row = await withTenant(this.pool, tenantId, (db) => this.repo.updateSecret(db, id, tenantId, secretCiphertext), ); if (!row) { throw new NotFoundException("endpoint not found"); } - return { ...publicView(row), secret }; + return { ...publicView(row), secret: secret.value }; } } diff --git a/apps/api/src/public/public.service.ts b/apps/api/src/public/public.service.ts index 5c2acba..dfae93d 100644 --- a/apps/api/src/public/public.service.ts +++ b/apps/api/src/public/public.service.ts @@ -1,9 +1,8 @@ -import { randomUUID } from "node:crypto"; import { Inject, Injectable, NotFoundException } from "@nestjs/common"; import { Pool } from "pg"; import { withTenant } from "@eventform/db"; -import type { SubmissionReceivedEvent } from "@eventform/shared"; import { API_POOL } from "../db/db.module"; +import { buildSubmissionReceivedEvent } from "../domain/submission-event"; import { PUBLIC_REPOSITORY, PublicFormRecord, PublicRepository } from "./public.repository"; export interface PublicField { @@ -64,21 +63,15 @@ export class PublicService { const endpointIds = await this.repo.listActiveEndpointIds(db, form.tenantId); for (const endpointId of endpointIds) { - const deliveryId = randomUUID(); - const eventId = randomUUID(); - const payload: SubmissionReceivedEvent = { - eventId, - type: "submission.received", - attempt: 1, + const { deliveryId, eventId, payload } = buildSubmissionReceivedEvent({ tenantId: form.tenantId, formId: form.id, formTitle: form.title, submissionId, endpointId, - deliveryId, answers, - submittedAt: submittedAt.toISOString(), - }; + submittedAt, + }); await this.repo.insertDeliveryWithOutbox(db, form.tenantId, { deliveryId, endpointId, diff --git a/apps/api/test/domain.test.ts b/apps/api/test/domain.test.ts new file mode 100644 index 0000000..b66e1ee --- /dev/null +++ b/apps/api/test/domain.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, it } from "vitest"; +import { canRetryDelivery } from "../src/domain/delivery-rules"; +import { + buildSubmissionReceivedEvent, + rebuildRetryEvent, +} from "../src/domain/submission-event"; +import { EndpointSecret } from "../src/domain/endpoint-secret"; + +// Pure domain logic — no infrastructure, nothing to mock. + +describe("canRetryDelivery", () => { + it("allows retry only for failed deliveries", () => { + expect(canRetryDelivery("failed")).toBe(true); + for (const status of ["pending", "retrying", "delivered"]) { + expect(canRetryDelivery(status)).toBe(false); + } + }); +}); + +describe("buildSubmissionReceivedEvent", () => { + const input = { + tenantId: "t1", + formId: "f1", + formTitle: "Title", + submissionId: "s1", + endpointId: "e1", + answers: { name: "alice" }, + submittedAt: new Date("2026-01-01T00:00:00.000Z"), + }; + + it("builds a submission.received payload with fresh ids and attempt 1", () => { + const { deliveryId, eventId, payload } = buildSubmissionReceivedEvent(input); + expect(payload.type).toBe("submission.received"); + expect(payload.attempt).toBe(1); + expect(payload.eventId).toBe(eventId); + expect(payload.deliveryId).toBe(deliveryId); + expect(payload.endpointId).toBe("e1"); + expect(payload.submittedAt).toBe("2026-01-01T00:00:00.000Z"); + expect(payload.answers).toEqual({ name: "alice" }); + }); + + it("mints distinct event and delivery ids per call", () => { + const a = buildSubmissionReceivedEvent(input); + const b = buildSubmissionReceivedEvent(input); + expect(a.eventId).not.toBe(b.eventId); + expect(a.deliveryId).not.toBe(b.deliveryId); + }); +}); + +describe("rebuildRetryEvent", () => { + it("keeps the body, swaps the eventId, and resets attempt to 1", () => { + const original = { eventId: "old", attempt: 3, type: "submission.received", body: "x" }; + const rebuilt = rebuildRetryEvent(original, "new"); + expect(rebuilt.eventId).toBe("new"); + expect(rebuilt.attempt).toBe(1); + expect(rebuilt.type).toBe("submission.received"); + expect(rebuilt.body).toBe("x"); + }); +}); + +describe("EndpointSecret", () => { + it("generates a whsec_-prefixed secret and round-trips via value/toString", () => { + const secret = EndpointSecret.generate(); + expect(secret.value.startsWith(EndpointSecret.PREFIX)).toBe(true); + expect(secret.toString()).toBe(secret.value); + }); + + it("generates unique values", () => { + expect(EndpointSecret.generate().value).not.toBe(EndpointSecret.generate().value); + }); +});