refactor(api): repository layer + minimal domain extraction

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: () => {

apps/api/src/auth/tenants.repository.drizzle.ts

@@ -0,0 +1,32 @@
+import { Inject, Injectable } from "@nestjs/common";
+import { Pool } from "pg";
+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(sub: string): Promise<Tenant | undefined> {
+    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(sub: string): Promise<Tenant> {
+    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(tenantId: string, name: string): Promise<Tenant> {
+    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;
+  }
+}

apps/api/src/auth/tenants.repository.ts

@@ -0,0 +1,13 @@
+export const TENANT_REPOSITORY = "TENANT_REPOSITORY";
+
+export interface Tenant {
+  id: string;
+  name: string;
+}
+
+export interface TenantRepository {
+  findBySub(sub: string): Promise<Tenant | undefined>;
+  /** INSERT ... ON CONFLICT DO UPDATE — race-safe first-login provisioning. */
+  insertBySub(sub: string): Promise<Tenant>;
+  updateName(tenantId: string, name: string): Promise<Tenant>;
+}

apps/api/src/auth/tenants.service.ts

@@ -1,42 +1,24 @@
 import { Inject, Injectable } from "@nestjs/common";
-import { Pool } from "pg";
-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(TENANT_REPOSITORY) private readonly repo: TenantRepository,
+  ) {}
 
   async findOrCreateBySub(sub: string): Promise<Tenant> {
-    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 existing = await this.repo.findBySub(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(sub);
   }
 
   /** Display-name update (e.g. from the Google ID token after first login). */
   async updateName(tenantId: string, name: string): Promise<Tenant> {
-    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(tenantId, name);
   }
 }

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 {}

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<DeliveryListItem[]> {
+    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<DeliveryRow | undefined> {
+    const [delivery] = await x.select().from(deliveries).where(eq(deliveries.id, id));
+    return delivery;
+  }
+
+  listAttempts(x: Executor, deliveryId: string): Promise<DeliveryAttemptRow[]> {
+    return x
+      .select()
+      .from(deliveryAttempts)
+      .where(eq(deliveryAttempts.deliveryId, deliveryId))
+      .orderBy(asc(deliveryAttempts.attemptNo));
+  }
+
+  async findByIdForUpdate(x: Executor, id: string): Promise<DeliveryRow | undefined> {
+    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<DeliveryRow> {
+    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<string, unknown>).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;
+  }
+}

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<DeliveryListItem[]>;
+  findById(x: Executor, id: string): Promise<DeliveryRow | undefined>;
+  listAttempts(x: Executor, deliveryId: string): Promise<DeliveryAttemptRow[]>;
+  /** SELECT ... FOR UPDATE — used to lock the row before a manual retry. */
+  findByIdForUpdate(x: Executor, id: string): Promise<DeliveryRow | undefined>;
+  /**
+   * 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<DeliveryRow>;
+}

apps/api/src/deliveries/deliveries.service.ts

@@ -1,107 +1,48 @@
 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 { canRetryDelivery } from "../domain/delivery-rules";
+import { rebuildRetryEvent } from "../domain/submission-event";
 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 };
     });
   }
 
   /** 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") {
+      if (!canRetryDelivery(delivery.status)) {
         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;
+      const payload = rebuildRetryEvent(delivery.payload, eventId);
+      return this.repo.reEmitRetry(db, { delivery, tenantId, eventId, payload });
     });
   }
 }

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";
+}