Lịch sử biết phân trang

import { beforeAll, describe, expect, it } from "vitest";
import { desc, eq, sql } from "drizzle-orm";
 
import { createDb, createPool, DEFAULT_DATABASE_URL, type Db } from "./client";
import { migrate } from "./migrate";
import { createEnvironment, Repository } from "./repository";
import { messages } from "./schema";
 
// The staged failure for chapter 2.4 — offset pagination drifting under a
// moving feed.
//
// The offset read lives HERE, in the test, and not in the repository: the
// constitution says offset pagination "is not offered", and a method that
// exists is a method someone will call. Keeping the broken version in the
// suite means the demonstration stays runnable at the tag without the
// production layer ever shipping the thing the chapter argues against.
// (This file sits under src/db/ because that is where the lint rule
// permits the query engine — the same reason 2.3's DR-01 guard lives in
// repository.itest.ts.)
 
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 repo: Repository;
 
/** The pagination every framework hands you — correct for data that holds
 * still, which a chat feed never does. */
async function readByOffset(
  channelId: string,
  { offset, limit }: { offset: number; limit: number },
) {
  return db
    .select({ id: messages.id, seq: messages.sequence })
    .from(messages)
    .where(eq(messages.channelId, channelId))
    .orderBy(desc(messages.sequence))
    .limit(limit)
    .offset(offset);
}
 
beforeAll(async () => {
  await migrate(pool);
  const env = await createEnvironment(db, { name: "history-drift-itest" });
  repo = new Repository(db, env.id);
});
 
describe("offset pagination drifts under live inserts (chapter 2.4)", () => {
  it("serves rows the reader has already seen", async () => {
    const channel = await repo.createChannel("drift-repeat", "public");
    for (let i = 1; i <= 60; i += 1) {
      await repo.sendMessage(channel.id, { text: `m-${i}` });
    }
    const page1 = await readByOffset(channel.id, { offset: 0, limit: 50 });
    // The feed moves mid-scroll: three drivers type while page two loads.
    for (let i = 1; i <= 3; i += 1) {
      await repo.sendMessage(channel.id, { text: `live-${i}` });
    }
    const page2 = await readByOffset(channel.id, { offset: 50, limit: 50 });
 
    const seen = new Set(page1.map((m) => m.id));
    const repeats = page2.filter((m) => seen.has(m.id));
    // Three inserts, three repeats — the drift is exactly the shift.
    expect(repeats).toHaveLength(3);
  });
 
  it("hides rows the reader will never see, when the feed shrinks", async () => {
    const channel = await repo.createChannel("drift-gap", "public");
    for (let i = 1; i <= 60; i += 1) {
      await repo.sendMessage(channel.id, { text: `m-${i}` });
    }
    const page1 = await readByOffset(channel.id, { offset: 0, limit: 50 });
    // A moderator deletes a message the reader has ALREADY passed — one
    // from page one's range. Every row below it shifts up one position,
    // so page two's offset now starts one row too late.
    const victim = page1[25]!;
    await db.execute(sql`DELETE FROM messages WHERE id = ${victim.id}`);
 
    const page2 = await readByOffset(channel.id, { offset: 50, limit: 50 });
    const delivered = new Set([...page1, ...page2].map((m) => m.seq));
    const survivors = await db
      .select({ seq: messages.sequence })
      .from(messages)
      .where(eq(messages.channelId, channel.id));
    const missed = survivors.filter((m) => !delivered.has(m.seq));
    // A row that still exists was served on neither page. From the
    // reader's chair that is not "a little overlap" — it is data loss.
    expect(missed.length).toBeGreaterThan(0);
  });
});

-import { and, asc, eq, sql } from "drizzle-orm";
+import { and, asc, desc, eq, gt, lt, sql, type SQL } from "drizzle-orm";
 
 import type { Db } from "./client";
 import { channels, members, messages, users } from "./schema";
@@ -351,6 +351,60 @@ export class Repository {
     return { ...row, created_at: toIso(row.created_at) };
   }
 
+  /** History reads (chapter 2.4): one page of messages anchored to a
+   * sequence position, in either direction (FR-MSG-09), riding the
+   * messages_channel_seq index in its natural order. The channel join
+   * carries the tenant scope, so a foreign channel id pages nothing.
+   *
+   * Anchors are strictly EXCLUSIVE: the cursor names the last row the
+   * client already has. Inclusive comparisons would serve that row twice,
+   * once per page — offset drift rebuilt at a single row's scale.
+   */
+  async listMessages(
+    channelId: string,
+    {
+      beforeSeq,
+      afterSeq,
+      limit,
+    }: { beforeSeq?: number; afterSeq?: number; limit: number },
+  ): Promise<MessageRow[]> {
+    const columns = {
+      id: messages.id,
+      channel_id: messages.channelId,
+      seq: messages.sequence,
+      text: messages.text,
+      created_at: messages.createdAt,
+    };
+    const scoped = (extra?: SQL) =>
+      and(
+        eq(messages.channelId, channelId),
+        eq(channels.environmentId, this.environmentId),
+        ...(extra ? [extra] : []),
+      );
+    const rows = await (afterSeq === undefined
+      ? this.db
+          .select(columns)
+          .from(messages)
+          .innerJoin(channels, eq(channels.id, messages.channelId))
+          .where(
+            scoped(
+              beforeSeq === undefined
+                ? undefined
+                : lt(messages.sequence, beforeSeq),
+            ),
+          )
+          .orderBy(desc(messages.sequence))
+          .limit(limit)
+      : this.db
+          .select(columns)
+          .from(messages)
+          .innerJoin(channels, eq(channels.id, messages.channelId))
+          .where(scoped(gt(messages.sequence, afterSeq)))
+          .orderBy(asc(messages.sequence))
+          .limit(limit));
+    return rows.map((row) => ({ ...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

Limit  (cost=0.27..8.29 rows=1 width=24)
  ->  Index Scan Backward using messages_channel_id_sequence_unique on messages
        Index Cond: ((channel_id = '…'::uuid) AND (sequence < 100))

@@ -127,9 +127,10 @@ export const messages = pgTable(
     uniqueIndex("messages_idem")
       .on(t.channelId, t.idempotencyKey)
       .where(sql`${t.idempotencyKey} IS NOT NULL`),
-    // Hot-path index (SAD §6.3): history pagination as a pure
-    // index-order scan (FR-MSG-09).
-    index("messages_channel_seq").on(t.channelId, t.sequence.desc()),
+    // No dedicated (channel_id, sequence DESC) index: DR-01's unique
+    // constraint above already supplies that ordering, and Postgres walks
+    // it backward for newest-first pages. Chapter 2.4 measured it and
+    // migration 0002 dropped the redundant twin (SAD §6.3, amended).
   ],
 );

-- 0001 (chapter 2.4): GENERATED by drizzle-kit, then reviewed.
--
-- 0000 created messages_channel_seq on (channel_id, sequence DESC) for
-- history pagination, per SAD §6.3 as it stood. Measured at 50,000 rows,
-- the planner never used it: it walks DR-01's UNIQUE (channel_id, sequence)
-- BACKWARD instead, and dropping this index changed neither the plan nor
-- the cost estimate (0.41..5.04 either way). A btree is bidirectional, so a
-- DESC twin of an existing ASC index adds no ordering — only write
-- amplification on the send path and storage.
--
-- Review disposition: one statement, exactly the intended drop, nothing
-- else. Forward-only as always — 0000 stays true at its own tag, and this
-- file is how the correction travels (SAD §6.3 amended 2026-08-02, v1.1).
 
DROP INDEX "messages_channel_seq";

Index Scan Backward using messages_channel_id_sequence_unique
  cost=0.41..4.93  Index Cond: ((channel_id = '…') AND (sequence < 40000))

// Opaque history cursors (chapter 2.4). The token encodes a sequence
// position; clients treat it as a black box (constitution V: "List
// endpoints use opaque cursor pagination; offset pagination is not
// offered"). The structure inside may change at any time — a timestamp
// for retention-aware reads, a shard hint at 10x scale — and no client
// breaks, because no client ever saw inside.
 
const PREFIX = "s:";
 
export function encodeCursor(seq: number): string {
  return Buffer.from(`${PREFIX}${seq}`, "utf8").toString("base64url");
}
 
/** Decode a cursor, or null if the token is not one of ours. Callers turn
 * null into a 400 through the usual envelope — never a 500, and never a
 * silent fallback to "start from the top", which would quietly serve the
 * wrong page. */
export function decodeCursor(token: string): number | null {
  let raw: string;
  try {
    raw = Buffer.from(token, "base64url").toString("utf8");
  } catch {
    return null;
  }
  const match = /^s:(\d+)$/.exec(raw);
  if (!match) return null;
  const seq = Number(match[1]);
  return Number.isSafeInteger(seq) ? seq : null;
}

import { describe, expect, it } from "vitest";
 
import { decodeCursor, encodeCursor } from "./cursor";
 
// The codec is dull on purpose — but "dull" still has to round-trip and
// still has to refuse everything else, because a cursor that decodes to
// the wrong number serves the wrong page in silence.
describe("history cursors (chapter 2.4)", () => {
  it("round-trips a sequence position", () => {
    for (const seq of [0, 1, 42, 1_000_000]) {
      expect(decodeCursor(encodeCursor(seq))).toBe(seq);
    }
  });
 
  it("is opaque — the token does not read as its contents", () => {
    expect(encodeCursor(363)).not.toContain("363");
  });
 
  it("refuses anything it did not mint", () => {
    for (const junk of [
      "",
      "363",
      "not-base64!",
      Buffer.from("s:abc").toString("base64url"),
    ]) {
      expect(decodeCursor(junk)).toBeNull();
    }
  });
});

 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>;
+
+// The history query (chapter 2.4, FR-MSG-09): an opaque cursor, a
+// direction, and a page size capped at 200. `limit` CLAMPS rather than
+// rejects — a client asking for 500 gets 200 and a next_cursor, because
+// caps exist to protect the server and a clamp does that just as well
+// while leaving the client's loop logic alone.
+export const historyQuerySchema = z.strictObject({
+  cursor: z.string().min(1).optional(),
+  direction: z.enum(["older", "newer"]).default("older"),
+  limit: z.coerce.number().int().min(1).max(200).default(50),
+});
+
+export type HistoryQuery = z.infer<typeof historyQuerySchema>;

-import { Injectable, NotFoundException } from "@nestjs/common";
+import {
+  BadRequestException,
+  Injectable,
+  NotFoundException,
+} from "@nestjs/common";
 
 import {
   ChannelNotFoundError,
   Repository,
   type MessageRow,
 } from "../db/repository";
+import { decodeCursor, encodeCursor } from "./cursor";
-import type { SendMessageBody } from "./messages.schema";
+import type { HistoryQuery, SendMessageBody } from "./messages.schema";
 
 // The thin layer between HTTP and the repository (chapters 2.2 + 2.3). It
 // owns two things: turning the layer's domain error into the wire's 404,
 // 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 {
       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;
     }
   }
+
+  /** A page of history (chapter 2.4). The cursor is opaque coming in and
+   * going out; the service is the only place that knows it encodes a
+   * sequence. A cursor we did not mint is a 400, never a silent reset to
+   * the top — serving the wrong page quietly is worse than refusing. */
+  async history(
+    channelId: string,
+    { cursor, direction, limit }: HistoryQuery,
+  ): Promise<{
+    messages: MessageRow[];
+    next_cursor: string | null;
+    prev_cursor: string | null;
+  }> {
+    let anchor: number | undefined;
+    if (cursor !== undefined) {
+      const decoded = decodeCursor(cursor);
+      if (decoded === null) throw new BadRequestException("malformed cursor");
+      anchor = decoded;
+    }
+    const messages = await this.repo.listMessages(channelId, {
+      limit,
+      ...(direction === "newer"
+        ? { afterSeq: anchor ?? 0 }
+        : anchor === undefined
+          ? {}
+          : { beforeSeq: anchor }),
+    });
+    // Edge rows become the next anchors. A short page still yields a
+    // next_cursor: "no more yet" and "no more ever" are the same answer
+    // in a feed that keeps growing, and the client simply gets an empty
+    // page next time.
+    const first = messages[0];
+    const last = messages[messages.length - 1];
+    return {
+      messages,
+      next_cursor: last ? encodeCursor(last.seq) : null,
+      prev_cursor: first ? encodeCursor(first.seq) : null,
+    };
+  }
 }

-import { Body, Controller, Param, Post, UseGuards } from "@nestjs/common";
+import {
+  Body,
+  Controller,
+  Get,
+  Param,
+  Post,
+  Query,
+  UseGuards,
+} from "@nestjs/common";
 
 import { EnvironmentContextGuard } from "./environment-context.guard";
 import { MessagesService } from "./messages.service";
-import { sendMessageBodySchema } from "./messages.schema";
+import { historyQuerySchema, sendMessageBodySchema } from "./messages.schema";
 // `import type` is required, not stylistic: with isolatedModules and
 // emitDecoratorMetadata on (ADR-15's trade-off, chapter 1.4), a type used
 // in a decorated signature must be imported as a type or TS1272 refuses
 // to compile it.
-import type { SendMessageBody } from "./messages.schema";
+import type { HistoryQuery, SendMessageBody } from "./messages.schema";
 import { ZodValidationPipe } from "./zod-validation.pipe";
 
 // The api's first product endpoint (chapter 2.2). Validation is zod at the
 // boundary — the same schema family as @relay/protocol, so the REST body
 // and the WebSocket frame payload cannot drift (1.3's payoff, again).
 @Controller("v1/channels/:channelId/messages")
 @UseGuards(EnvironmentContextGuard)
 export class MessagesController {
   constructor(private readonly messages: MessagesService) {}
 
   @Post()
   async send(
     @Param("channelId") channelId: string,
     @Body(new ZodValidationPipe(sendMessageBodySchema)) body: SendMessageBody,
   ) {
     return this.messages.send(channelId, body);
   }
+
+  @Get()
+  async history(
+    @Param("channelId") channelId: string,
+    @Query(new ZodValidationPipe(historyQuerySchema)) query: HistoryQuery,
+  ) {
+    return this.messages.history(channelId, query);
+  }
 }

curl -s "localhost:4000/v1/channels/$CHANNEL/messages?limit=50" \
  -H "X-Relay-Environment: $ENV"
# → 50 newest messages + next_cursor
curl -s "localhost:4000/v1/channels/$CHANNEL/messages?limit=50&cursor=$NEXT" \
  -H "X-Relay-Environment: $ENV"
# → the NEXT 50, no repeats — send three messages between the calls and
#   verify the second page is identical either way