Send it twice

it("a retry WITHOUT a key duplicates the message — journey 4's failure, staged", async () => {
  const channel = await repo.createChannel("idem-no-key", "public");
  await repo.sendMessage(channel.id, { text: "B2, north ramp" });
  // The ack was lost; the client cannot know. It retries:
  await repo.sendMessage(channel.id, { text: "B2, north ramp" });
  const rows = await repo.listMessagesRaw(channel.id);
  // Two rows, seq 1 and 2, identical text. The dispatcher reads it twice.
  expect(rows.filter((m) => m.text === "B2, north ramp")).toHaveLength(2);
});

CREATE UNIQUE INDEX "messages_idem" ON "messages"
  USING btree ("channel_id","idempotency_key")
  WHERE "messages"."idempotency_key" IS NOT NULL;   -- DR-03

 import { z } from "zod";
 
 // The send body (chapter 2.2). FR-MSG-01 fixes the limits: text up to
 // 8,000 characters, metadata up to 4 KB of JSON — the length check lands
 // with FR-EMJ-02's code-point counting in the emoji chapter; today the
 // character bound is the honest approximation, recorded as such.
 export const sendMessageBodySchema = z.strictObject({
   text: z.string().min(1).max(8000),
   metadata: z.record(z.string(), z.unknown()).optional(),
+  // Chapter 2.3 (FR-MSG-04): the client's idempotency key — minted at send
+  // time (FR-SDK-06), optional because server-originated messages may not
+  // carry one. The partial unique index (DR-03) ignores NULLs.
+  idempotency_key: z.string().uuid().optional(),
 });
 
 export type SendMessageBody = z.infer<typeof sendMessageBodySchema>;

 import { Injectable, NotFoundException } from "@nestjs/common";
 
 import {
   ChannelNotFoundError,
   Repository,
   type MessageRow,
 } from "../db/repository";
 import type { SendMessageBody } from "./messages.schema";
 
-// The thin layer between HTTP and the repository (chapter 2.2). It owns
+// The thin layer between HTTP and the repository (chapters 2.2 + 2.3). It
-// exactly one thing today: turning the layer's domain error into the
+// owns two things: turning the layer's domain error into the wire's 404,
-// wire's 404 — same non-answer a foreign tenant's id has received since
-// 2.1, now wearing an HTTP status. The filter from 1.4 shapes the body.
+// and translating the repository's `duplicate: true` into FR-MSG-04's
+// "201-equivalent semantics" — the retry returns the original message,
+// indistinguishable from a fresh send (the `duplicate` flag stays internal).
 @Injectable()
 export class MessagesService {
   constructor(private readonly repo: Repository) {}
 
-  async send(channelId: string, body: SendMessageBody): Promise<MessageRow> {
+  async send(
+    channelId: string,
+    body: SendMessageBody,
+  ): Promise<MessageRow> {
     try {
-      return await this.repo.sendMessage(channelId, body);
+      const result = await this.repo.sendMessage(channelId, {
+        text: body.text,
+        metadata: body.metadata,
+        ...(body.idempotency_key != null && {
+          idempotencyKey: body.idempotency_key,
+        }),
+      });
+      // The internal duplicate flag never reaches the wire: the client
+      // sees the same body whether this was the original send or the
+      // retry that recovered it (FR-MSG-04's 201-equivalent semantics).
+      return {
+        id: result.id,
+        channel_id: result.channel_id,
+        seq: result.seq,
+        text: result.text,
+        created_at: result.created_at,
+      };
     } catch (error) {
       if (error instanceof ChannelNotFoundError) {
         // A CONSTANT message: echoing the id back would make the foreign-id
         // answer differ from the missing-id answer, and "different" is
         // itself a disclosure (FR-TEN-05).
         throw new NotFoundException("channel not found");
       }
       throw error;
     }
   }
 }

 import { randomUUID } from "node:crypto";
 
 import { and, asc, eq, sql } from "drizzle-orm";
 
 import type { Db } from "./client";
 import { channels, members, messages, users } from "./schema";
 
 // The repository layer — the ONE place data access lives (ADR-04's single
 // writer, constitution I). Two surfaces with a bright line between them:
 //
 //   createEnvironment — the ADMIN surface. It creates tenants, so it is the
 //   only operation here that is not tenant-scoped. It also inserts a stub
 //   application row to satisfy environments' NOT NULL foreign key (recorded
 //   decision: the real application lifecycle belongs to Part 3).
 //
 //   Repository — everything else. The constructor REQUIRES an
 //   environment_id; every query is scoped by it HERE, in one home — never
 //   at call sites. Cross-tenant reads return null/empty: no data, and no
 //   reveal that the foreign id even exists (FR-TEN-05).
 //
 // Drizzle is the query engine inside this layer (ADR-16): queries keep
 // their SQL shape and gain end-to-end types. Where the builder falls short,
 // a raw SQL island is permitted — inside the layer, never outside it.
 //
 // All primary keys are generated app-side (crypto.randomUUID) — the SAD's
 // SQL declares no id defaults, and the migration adds none.
 
 export interface Environment {
   id: string;
   kind: "development" | "production";
 }
 
 export async function createEnvironment(
   db: Db,
   { name, kind = "development" }: { name: string; kind?: Environment["kind"] },
 ): Promise<Environment> {
   const applicationId = randomUUID();
   const environmentId = randomUUID();
   // The admin surface writes through the same Db handle but carries no
   // tenant scope — it is the operation that MINTS the scope.
   await db.execute(
     sql`INSERT INTO applications (id, name) VALUES (${applicationId}, ${name})`,
   );
   await db.execute(
     sql`INSERT INTO environments (id, application_id, kind, signing_secret)
         VALUES (${environmentId}, ${applicationId}, ${kind}, ${randomUUID()})`,
   );
   return { id: environmentId, kind };
 }
 
 export interface UserRow {
   id: string;
   external_id: string;
   display_name: string | null;
 }
 
 export interface ChannelRow {
   id: string;
   external_id: string;
   type: "public" | "private";
   name: string | null;
 }
 
 export interface MessageRow {
   id: string;
   channel_id: string;
   seq: number;
   text: string | null;
   created_at: string;
+  /** Chapter 2.3 (FR-MSG-04): true when a retry was recognised by the
+   * idempotency index and the ORIGINAL message was returned instead of
+   * a new insert. The service layer uses this to decide response shape. */
+  duplicate?: boolean;
 }
 
 /** Thrown when a channel id resolves to nothing IN THIS TENANT — which,
  * from the caller's side, is indistinguishable from "does not exist"
  * (FR-TEN-05: no data, and no reveal that the foreign id exists). The
  * layer stays framework-free; the service turns this into the wire's
  * 404 (constitution I's isolation, EIR-API-04's envelope). */
 export class ChannelNotFoundError extends Error {
   constructor(public readonly channelId: string) {
     super(`channel not found: ${channelId}`);
     this.name = "ChannelNotFoundError";
   }
+}
+
+/** Timestamps cross the wire as RFC 3339 strings (constitution: UTC,
+ * millisecond precision) — the driver hands back a Date or a string
+ * depending on the column and the query shape. */
+function toIso(value: Date | string): string {
+  return value instanceof Date ? value.toISOString() : String(value);
 }
 
 export class Repository {
   // Constructor parameter properties — the shorthand chapter 1.4 released
   // for this service when ADR-15 spent erasableSyntaxOnly on decorator
   // metadata. The guarantee still holds in the gateway and every package.
   constructor(
     private readonly db: Db,
     private readonly environmentId: string,
   ) {}
 
   async createUser(externalId: string, displayName?: string): Promise<UserRow> {
     const id = randomUUID();
     await this.db.insert(users).values({
       id,
       environmentId: this.environmentId,
       externalId,
       displayName: displayName ?? null,
     });
     return { id, external_id: externalId, display_name: displayName ?? null };
   }
 
   async getUserByExternalId(externalId: string): Promise<UserRow | null> {
     const rows = await this.db
       .select({
         id: users.id,
         external_id: users.externalId,
         display_name: users.displayName,
       })
       .from(users)
       .where(
         and(
           eq(users.environmentId, this.environmentId),
           eq(users.externalId, externalId),
         ),
       );
     return rows[0] ?? null;
   }
 
   async createChannel(
     externalId: string,
     type: ChannelRow["type"],
     name?: string,
   ): Promise<ChannelRow> {
     const id = randomUUID();
     await this.db.insert(channels).values({
       id,
       environmentId: this.environmentId,
       externalId,
       type,
       name: name ?? null,
     });
     return { id, external_id: externalId, type, name: name ?? null };
   }
 
   async getChannelByExternalId(externalId: string): Promise<ChannelRow | null> {
     const rows = await this.db
       .select({
         id: channels.id,
         external_id: channels.externalId,
         type: sql<ChannelRow["type"]>`${channels.type}`,
         name: channels.name,
       })
       .from(channels)
       .where(
         and(
           eq(channels.environmentId, this.environmentId),
           eq(channels.externalId, externalId),
         ),
       );
     return rows[0] ?? null;
   }
 
   async listChannels(): Promise<ChannelRow[]> {
     return this.db
       .select({
         id: channels.id,
         external_id: channels.externalId,
         type: sql<ChannelRow["type"]>`${channels.type}`,
         name: channels.name,
       })
       .from(channels)
       .where(eq(channels.environmentId, this.environmentId))
       .orderBy(asc(channels.externalId));
   }
 
   /** Membership joins live in channel-land, so the tenant scope rides the
    * channel: the double-scoped SELECT below is what makes a foreign channel
    * id useless. INSERT ... SELECT is where the builder falls short — this
    * is the layer's one raw SQL island, permitted by ADR-16 and kept inside
    * the wall like everything else. */
   async addMember(channelId: string, userId: string): Promise<boolean> {
     const result = await this.db.execute(
       sql`INSERT INTO members (channel_id, user_id)
           SELECT c.id, u.id FROM channels c, users u
           WHERE c.id = ${channelId} AND c.environment_id = ${this.environmentId}
             AND u.id = ${userId} AND u.environment_id = ${this.environmentId}`,
     );
     return (result.rowCount ?? 0) > 0;
   }
 
   async listMembers(channelId: string): Promise<string[]> {
     const rows = await this.db
       .select({ user_id: members.userId })
       .from(members)
       .innerJoin(channels, eq(channels.id, members.channelId))
       .where(
         and(
           eq(members.channelId, channelId),
           eq(channels.environmentId, this.environmentId),
         ),
       )
       .orderBy(asc(members.joinedAt));
     return rows.map((r) => r.user_id);
   }
 
   async channelsForUser(userId: string): Promise<string[]> {
     const rows = await this.db
       .select({ channel_id: members.channelId })
       .from(members)
       .innerJoin(users, eq(users.id, members.userId))
       .where(
         and(
           eq(members.userId, userId),
           eq(users.environmentId, this.environmentId),
         ),
       );
     return rows.map((r) => r.channel_id);
   }
 
-  /** The write path (chapter 2.2): sequence assignment under the channel
+  /** The write path (chapters 2.2 + 2.3): sequence assignment under the
-   * row lock (ADR-03). The transaction IS the ordering guarantee: the
+   * channel row lock (ADR-03), with idempotency enforcement via the
+   * partial unique index (DR-03). The transaction IS the ordering
-   * lock serialises assignment per channel, and the ack that matters
+   * guarantee: the lock serialises assignment per channel, and the ack
-   * happens only after commit (FR-MSG-05).
+   * that matters happens only after commit (FR-MSG-05).
+   *
+   * When an idempotency key is present and conflicts with an existing
+   * message, the insert is skipped (ON CONFLICT DO NOTHING), the channel's
+   * sequence is left untouched, and the ORIGINAL message is returned with
+   * `duplicate: true` — FR-MSG-04's "201-equivalent semantics".
    */
   async sendMessage(
     channelId: string,
     {
       userId,
       text,
       metadata,
-    }: { userId?: string; text: string; metadata?: unknown },
+      idempotencyKey,
+    }: {
+      userId?: string;
+      text: string;
+      metadata?: unknown;
+      idempotencyKey?: string;
+    },
   ): Promise<MessageRow> {
     return this.db.transaction(async (tx) => {
       const [channel] = await tx
         .select({ id: channels.id, lastSequence: channels.lastSequence })
         .from(channels)
         .where(
           and(
             eq(channels.id, channelId),
             eq(channels.environmentId, this.environmentId),
           ),
         )
         .for("update");
       if (!channel) throw new ChannelNotFoundError(channelId);
       const seq = channel.lastSequence + 1;
       const id = randomUUID();
+
-      await tx
-        .update(channels)
-        .set({ lastSequence: seq })
-        .where(eq(channels.id, channel.id));
-      await tx.insert(messages).values({
+      const insert = tx.insert(messages).values({
         id,
         channelId: channel.id,
         sequence: seq,
         userId: userId ?? null,
         text,
         metadata: metadata ?? {},
+        idempotencyKey: idempotencyKey ?? null,
       });
+
+      // The conflict clause is attached ONLY when a key is present, and it
+      // names the partial index explicitly. A bare ON CONFLICT DO NOTHING
+      // would absorb every constraint on the table — including DR-01's
+      // UNIQUE (channel_id, sequence), whose loud failure is 2.2's safety
+      // net. A keyless send therefore carries no conflict clause at all.
+      const inserted = await (
+        idempotencyKey
+          ? insert.onConflictDoNothing({
+              target: [messages.channelId, messages.idempotencyKey],
+              where: sql`${messages.idempotencyKey} IS NOT NULL`,
+            })
+          : insert
+      ).returning({ id: messages.id, createdAt: messages.createdAt });
+
+      if (inserted.length === 0) {
+        // The key has been here before. Return the ORIGINAL message — the
+        // retry gets the same answer the lost ack carried (FR-MSG-04) —
+        // and leave last_sequence alone: a recognised duplicate wrote
+        // nothing, so it consumes nothing.
+        return {
+          ...(await this.getMessageByIdempotencyKey(
+            tx,
+            channel.id,
+            idempotencyKey!,
+          )),
+          duplicate: true,
+        };
+      }
+
+      // The sequence is spent only by a message that actually landed.
+      await tx
+        .update(channels)
+        .set({ lastSequence: seq })
+        .where(eq(channels.id, channel.id));
+
       return {
         id,
         channel_id: channel.id,
         seq,
         text,
-        created_at: new Date().toISOString(),
+        created_at: toIso(inserted[0]!.createdAt),
       };
     });
   }
+
+  /** Fetch a message by its idempotency key within a channel — the
+   * recovery leg of 2.3's duplicate-recognised path. The channel join
+   * carries the tenant scope: every query in this layer answers only for
+   * its own environment, private helpers included (constitution I). */
+  private async getMessageByIdempotencyKey(
+    tx: Db,
+    channelId: string,
+    idempotencyKey: string,
+  ): Promise<MessageRow> {
+    const [row] = await tx
+      .select({
+        id: messages.id,
+        channel_id: messages.channelId,
+        seq: messages.sequence,
+        text: messages.text,
+        created_at: messages.createdAt,
+      })
+      .from(messages)
+      .innerJoin(channels, eq(channels.id, messages.channelId))
+      .where(
+        and(
+          eq(messages.channelId, channelId),
+          eq(messages.idempotencyKey, idempotencyKey),
+          eq(channels.environmentId, this.environmentId),
+        ),
+      );
+    // The row MUST exist: this method is only reached when the insert
+    // conflicted on the idempotency index, so the key is already there.
+    if (!row) {
+      throw new Error(
+        `idempotency key ${idempotencyKey} conflicted but its message is missing — index inconsistency`,
+      );
+    }
+    return { ...row, created_at: toIso(row.created_at) };
+  }
+
+  /** Every message in the channel, ordered by sequence — tenant-scoped
+   * like everything else here. DECISION (chapter 2.3): this exists for
+   * the idempotency suite's row counts; 2.4 replaces it with the real
+   * paginated read, and this method retires with that chapter. */
+  async listMessagesRaw(
+    channelId: string,
+  ): Promise<{ id: string; text: string | null; seq: number }[]> {
+    return this.db
+      .select({
+        id: messages.id,
+        text: messages.text,
+        seq: messages.sequence,
+      })
+      .from(messages)
+      .innerJoin(channels, eq(channels.id, messages.channelId))
+      .where(
+        and(
+          eq(messages.channelId, channelId),
+          eq(channels.environmentId, this.environmentId),
+        ),
+      )
+      .orderBy(asc(messages.sequence));
+  }
 }

Error: idempotency key undefined conflicted but its message is missing …

 import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { sql } from "drizzle-orm";
 
 import { createDb, createPool, DEFAULT_DATABASE_URL, type Db } from "./client";
 import { migrate } from "./migrate";
 import { createEnvironment, Repository, type Environment } from "./repository";
 
 // The isolation suite: attack the repository with FOREIGN tenant ids and
 // prove the leak inexpressible (FR-TEN-05, NFR-SEC-09, constitution I).
 // Requires the compose Postgres — this file is *.itest.ts precisely so the
 // Docker-free unit lane never collects it.
 
 // Guardrail: integration tests run against the LOCAL compose stack only.
 const url = new URL(process.env.DATABASE_URL ?? DEFAULT_DATABASE_URL);
 if (!["localhost", "127.0.0.1"].includes(url.hostname)) {
   throw new Error(
     `integration tests refuse non-local databases (got host "${url.hostname}") — never point this suite at a shared database`,
   );
 }
 
 const pool = createPool();
 const db: Db = createDb(pool);
 let envA: Environment;
 let envB: Environment;
 let repoA: Repository;
 let repoB: Repository;
 
 beforeAll(async () => {
   await migrate(pool);
   // Deterministic ground WITHOUT a truncate: this suite mints its own
   // environments, and 2.1 proved no other environment's rows are visible
   // through them. Isolation buys parallel-safe test files for free — see
   // the note below.
   envA = await createEnvironment(db, { name: "tenant-a" });
   envB = await createEnvironment(db, { name: "tenant-b" });
   repoA = new Repository(db, envA.id);
   repoB = new Repository(db, envB.id);
 });
 
 afterAll(async () => {
   await pool.end();
 });
 
 describe("tenant isolation is structural (FR-TEN-05)", () => {
   it("a foreign external_id resolves to nothing — not even an existence hint", async () => {
     await repoA.createUser("tuan", "Tuan");
     expect(await repoA.getUserByExternalId("tuan")).not.toBeNull();
     expect(await repoB.getUserByExternalId("tuan")).toBeNull();
   });
 
   it("channel reads and lists are scoped by construction", async () => {
     await repoA.createChannel("support", "public", "Support");
     expect(await repoB.getChannelByExternalId("support")).toBeNull();
     expect(await repoB.listChannels()).toEqual([]);
     expect((await repoA.listChannels()).map((c) => c.external_id)).toContain(
       "support",
     );
   });
 
   it("membership writes with foreign ids affect zero rows", async () => {
     const user = await repoA.getUserByExternalId("tuan");
     const channel = await repoA.getChannelByExternalId("support");
     expect(await repoA.addMember(channel!.id, user!.id)).toBe(true);
     // B holds A's REAL ids — and still cannot write or read through them.
     expect(await repoB.addMember(channel!.id, user!.id)).toBe(false);
     expect(await repoB.listMembers(channel!.id)).toEqual([]);
     expect(await repoB.channelsForUser(user!.id)).toEqual([]);
     expect(await repoA.listMembers(channel!.id)).toEqual([user!.id]);
   });
 
   it("uniqueness is per-tenant (DR-02): both tenants may own the same external_id", async () => {
     await expect(
       repoB.createUser("tuan", "A different Tuan"),
     ).resolves.toBeTruthy();
     await expect(repoA.createUser("tuan", "Duplicate in A")).rejects.toThrow();
   });
 });
 
 describe("sequence assignment is serialised per channel (ADR-03)", () => {
   it("two concurrent sends never interleave", async () => {
     const channel = await repoA.createChannel("ordering", "public");
     const [a, b] = await Promise.all([
       repoA.sendMessage(channel.id, { text: "first writer" }),
       repoA.sendMessage(channel.id, { text: "second writer" }),
     ]);
     // Two sends, two DISTINCT consecutive sequence numbers — always.
     expect(new Set([a.seq, b.seq]).size).toBe(2);
     expect(Math.abs(a.seq - b.seq)).toBe(1);
   });
 });
+
+describe("idempotency must not disarm DR-01 (chapter 2.3)", () => {
+  it("a keyless send still fails loudly on a sequence collision", async () => {
+    const channel = await repoA.createChannel("dr01-guard", "public");
+    await repoA.sendMessage(channel.id, { text: "first" });
+    // Rewind the counter so the next keyless send reuses seq 1. The
+    // conflict clause must NOT swallow this: DR-01's unique constraint is
+    // 2.2's safety net, and idempotency has no business disarming it.
+    await db.execute(
+      sql`UPDATE channels SET last_sequence = 0 WHERE id = ${channel.id}`,
+    );
+    await expect(
+      repoA.sendMessage(channel.id, { text: "collides" }),
+    ).rejects.toThrow();
+    // And nothing landed: the failed insert wrote no row.
+    const rows = await repoA.listMessagesRaw(channel.id);
+    expect(rows).toHaveLength(1);
+  });
+});

KEY=$(uuidgen)
curl -s -X POST localhost:4000/v1/channels/$CHANNEL/messages \
  -H "X-Relay-Environment: $ENV" -H "content-type: application/json" \
  -d "{\"text\":\"B2, north ramp\",\"idempotency_key\":\"$KEY\"}"
# → seq 42. Now the "retry" — identical request:
curl -s -X POST localhost:4000/v1/channels/$CHANNEL/messages \
  -H "X-Relay-Environment: $ENV" -H "content-type: application/json" \
  -d "{\"text\":\"B2, north ramp\",\"idempotency_key\":\"$KEY\"}"
# → seq 42 again — the SAME message, recognised, not re-created

on conflict ("channel_id","idempotency_key")
  where "messages"."idempotency_key" IS NOT NULL
  do nothing