From 818e483f47cee679607cd952d314379435cfaf5a Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 4 Oct 2026 01:18:31 +0000 Subject: [PATCH] perf(api): filter peer-run activity in the transcript page query The transcript loaded a raw page, dropped hidden peer-run rows in memory and rescanned when nothing visible was left, so a long delegated stretch cost two queries per hidden page. The page query now joins runs and excludes hidden peer activity in Postgres, so every page is filled with visible rows. The kept block kinds live in one list shared by paging, the in-memory filter and live event forwarding. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01YTJRkJrXAS9e3aEFcFYTy8 --- .../src/thread-message-pages.postgres.test.ts | 143 ++++++++++++++++ apps/api/src/thread-message-pages.test.ts | 162 ++---------------- apps/api/src/thread-message-pages.ts | 96 +++++++---- apps/api/src/thread-target.test.ts | 24 ++- packages/testkit/src/cli/harness.ts | 1 + 5 files changed, 235 insertions(+), 191 deletions(-) create mode 100644 apps/api/src/thread-message-pages.postgres.test.ts diff --git a/apps/api/src/thread-message-pages.postgres.test.ts b/apps/api/src/thread-message-pages.postgres.test.ts new file mode 100644 index 0000000..b5b11e7 --- /dev/null +++ b/apps/api/src/thread-message-pages.postgres.test.ts @@ -0,0 +1,143 @@ +import type { MessageBlock } from "@rakazo/contracts"; +import type { PrismaClient } from "@rakazo/db"; +import { createDb, provisionMessagingIdentity } from "@rakazo/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { loadMessagePage } from "./thread-message-pages.js"; + +const databaseUrl = process.env.DATABASE_URL; +const describePostgres = + process.env.VERIFY_DATABASE && databaseUrl ? describe.sequential : describe.skip; + +describePostgres("loadMessagePage (PostgreSQL)", () => { + let prisma: PrismaClient; + let close: () => Promise; + let identity: { userId: string; spaceId: string; botId: string; threadId: string }; + let userRunId: string; + let peerRunId: string; + let seq = 0; + + beforeAll(async () => { + const db = createDb(databaseUrl!); + prisma = db.prisma; + close = async () => { + await db.prisma.$disconnect(); + await db.pool.end(); + }; + identity = await provisionMessagingIdentity( + prisma, + { provider: "sendblue", address: `+1555${Date.now().toString().slice(-7)}` }, + { signupsEnabled: undefined, signupAllowlist: undefined }, + ); + const { spaceId, botId, threadId, userId } = identity; + const owner = { spaceId, botId, threadId, userId }; + const run = async (trigger: string) => { + const task = await prisma.task.create({ + data: { ...owner, prompt: "fixture", status: "completed" }, + }); + const created = await prisma.run.create({ + data: { ...owner, taskId: task.id, status: "completed", trigger }, + }); + return created.id; + }; + userRunId = await run("user_message"); + peerRunId = await run("bot_message"); + seq = + ( + await prisma.message.aggregate({ + where: { threadId: identity.threadId }, + _max: { seq: true }, + }) + )._max.seq ?? 0; + }); + + afterAll(async () => { + if (identity) { + await prisma.messagingIdentity.deleteMany({ where: { userId: identity.userId } }); + await prisma.organization.deleteMany({ where: { id: identity.spaceId } }); + await prisma.user.deleteMany({ where: { id: identity.userId } }); + } + await close?.(); + }); + + async function message(id: string, runId: string | null, blocks: MessageBlock[]) { + seq += 1; + await prisma.message.create({ + data: { + id: `${identity.threadId}-${id}`, + threadId: identity.threadId, + seq, + role: "bot", + runId, + blocks, + }, + }); + return seq; + } + + const ids = (page: Awaited>) => + page.messages.map((row) => row.id.slice(identity.threadId.length + 1)); + + it("fills a page with visible rows across a long stretch of hidden peer activity", async () => { + await message("older", userRunId, [{ kind: "text", text: "Older answer" }]); + await message("newer", userRunId, [{ kind: "text", text: "Newer answer" }]); + for (let index = 0; index < 30; index += 1) { + await message(`peer-steps-${index}`, peerRunId, [ + { kind: "steps", steps: [{ label: "Working", count: 1 }] }, + ]); + } + await message("peer-receipt", peerRunId, [ + { kind: "bot_message_received", fromBotId: "bot-2", fromBotName: "Coder", text: "Done." }, + ]); + await message("peer-reply", peerRunId, [{ kind: "text", text: "Peer summary" }]); + await message("orphan-run", "run-that-no-longer-exists", [{ kind: "text", text: "Kept" }]); + + const first = await loadMessagePage(prisma, identity.threadId, undefined, 3); + expect(ids(first)).toEqual(["peer-receipt", "peer-reply", "orphan-run"]); + + const second = await loadMessagePage(prisma, identity.threadId, first.olderCursor!, 2); + expect(ids(second)).toEqual(["older", "newer"]); + + const peerView = await loadMessagePage( + prisma, + identity.threadId, + undefined, + 3, + undefined, + true, + ); + expect(ids(peerView)).toEqual(["peer-receipt", "peer-reply", "orphan-run"]); + const peerOlder = await loadMessagePage( + prisma, + identity.threadId, + peerView.olderCursor!, + 1, + undefined, + true, + ); + expect(ids(peerOlder)).toEqual(["peer-steps-29"]); + }); + + it("skips a receipt-only page for web but returns it when receipts are displayed", async () => { + await message("before-receipts", userRunId, [{ kind: "text", text: "Answer" }]); + await message("sent", userRunId, [ + { kind: "bot_message_sent", toBotId: "bot-2", toBotName: "Coder", text: "Check this." }, + ]); + await message("received", peerRunId, [ + { kind: "bot_message_received", fromBotId: "bot-2", fromBotName: "Coder", text: "Done." }, + ]); + + const web = await loadMessagePage(prisma, identity.threadId, undefined, 2); + expect(ids(web)).toEqual(["orphan-run", "before-receipts"]); + + const mobile = await loadMessagePage( + prisma, + identity.threadId, + undefined, + 2, + undefined, + false, + true, + ); + expect(ids(mobile)).toEqual(["sent", "received"]); + }); +}); diff --git a/apps/api/src/thread-message-pages.test.ts b/apps/api/src/thread-message-pages.test.ts index e290385..6a19d63 100644 --- a/apps/api/src/thread-message-pages.test.ts +++ b/apps/api/src/thread-message-pages.test.ts @@ -43,98 +43,6 @@ describe("thread message pages", () => { ).toBe(false); }); - it("keeps peer receipt rows when filtering peer-run output from pages", async () => { - const findMany = vi.fn(async () => [ - { - id: "message-reply", - threadId: "thread-1", - seq: 3, - role: "bot", - blocks: [{ kind: "text", text: "Echoed peer reply" }], - botId: "bot-1", - replyToMessageId: null, - runId: "run-peer", - createdAt: new Date("2026-08-16T00:00:03.000Z"), - }, - { - id: "message-received", - threadId: "thread-1", - seq: 2, - role: "user", - blocks: [ - { - kind: "bot_message_received", - fromBotId: "bot-2", - fromBotName: "Coder", - text: "Done.", - }, - ], - botId: null, - replyToMessageId: null, - runId: "run-peer", - createdAt: new Date("2026-08-16T00:00:02.000Z"), - }, - { - id: "message-user", - threadId: "thread-1", - seq: 1, - role: "bot", - blocks: [{ kind: "text", text: "Visible answer" }], - botId: "bot-1", - replyToMessageId: null, - runId: "run-user", - createdAt: new Date("2026-08-16T00:00:01.000Z"), - }, - ]); - const prisma = { - message: { findMany }, - run: { findMany: vi.fn(async () => [{ id: "run-peer" }]) }, - } as unknown as PrismaClient; - - const page = await loadMessagePage(prisma, "thread-1", undefined, 3); - - expect(page.messages.map((message) => message.id)).toEqual([ - "message-user", - "message-received", - "message-reply", - ]); - }); - - it("filters peer-run activity when its receipt is outside the loaded page", async () => { - const findMany = vi.fn(async () => [ - { - id: "message-peer", - threadId: "thread-1", - seq: 2, - role: "bot", - blocks: [{ kind: "steps", steps: [{ label: "Echoed peer reply", count: 1 }] }], - botId: "bot-1", - replyToMessageId: null, - runId: "run-peer", - createdAt: new Date("2026-08-16T00:00:02.000Z"), - }, - { - id: "message-user", - threadId: "thread-1", - seq: 1, - role: "bot", - blocks: [{ kind: "text", text: "Visible answer" }], - botId: "bot-1", - replyToMessageId: null, - runId: "run-user", - createdAt: new Date("2026-08-16T00:00:01.000Z"), - }, - ]); - const prisma = { - message: { findMany }, - run: { findMany: vi.fn(async () => [{ id: "run-peer" }]) }, - } as unknown as PrismaClient; - - const page = await loadMessagePage(prisma, "thread-1", undefined, 2); - - expect(page.messages.map((message) => message.id)).toEqual(["message-user"]); - }); - it("keeps peer text around-page targets and omits peer activity", async () => { const findMany = vi.fn(async () => [ { @@ -281,45 +189,6 @@ describe("thread message pages", () => { expect(page.messages.map((message) => message.id)).toEqual(["message-peer"]); }); - it("scans past a page containing only hidden peer-run activity", async () => { - const row = (seq: number, runId: string, kind: "text" | "steps" = "text") => ({ - id: `message-${seq}`, - threadId: "thread-1", - seq, - role: "bot", - blocks: - kind === "steps" - ? [{ kind: "steps", steps: [{ label: String(seq), count: 1 }] }] - : [{ kind: "text", text: String(seq) }], - botId: "bot-1", - replyToMessageId: null, - runId, - createdAt: new Date("2026-08-16T00:00:00.000Z"), - }); - const findMany = vi - .fn() - .mockResolvedValueOnce([ - row(4, "run-peer", "steps"), - row(3, "run-peer", "steps"), - row(2, "run-peer", "steps"), - ]) - .mockResolvedValueOnce([row(1, "run-user")]); - const prisma = { - message: { findMany }, - run: { - findMany: vi - .fn() - .mockResolvedValueOnce([{ id: "run-peer" }]) - .mockResolvedValueOnce([]), - }, - } as unknown as PrismaClient; - - const page = await loadMessagePage(prisma, "thread-1", undefined, 2); - - expect(page.messages.map((message) => message.id)).toEqual(["message-1"]); - expect(findMany).toHaveBeenCalledTimes(2); - }); - it("scans past a receipt-only page so web can reach older user-visible rows", async () => { const receiptRows = [ { @@ -383,21 +252,13 @@ describe("thread message pages", () => { createdAt: new Date("2026-08-16T00:00:01.000Z"), }, ]; - const findMany = vi.fn().mockResolvedValueOnce(receiptRows).mockResolvedValueOnce(olderRows); - const prisma = { - message: { findMany }, - run: { - findMany: vi - .fn() - .mockResolvedValueOnce([{ id: "run-peer" }]) - .mockResolvedValueOnce([]), - }, - } as unknown as PrismaClient; + const queryRaw = vi.fn().mockResolvedValueOnce(receiptRows).mockResolvedValueOnce(olderRows); + const prisma = { $queryRaw: queryRaw } as unknown as PrismaClient; const page = await loadMessagePage(prisma, "thread-1", undefined, 2); expect(page.messages.map((message) => message.id)).toEqual(["message-user"]); - expect(findMany).toHaveBeenCalledTimes(2); + expect(queryRaw).toHaveBeenCalledTimes(2); }); it("returns a receipt-only page when the client displays peer receipts", async () => { @@ -419,11 +280,8 @@ describe("thread message pages", () => { runId: "run-peer", createdAt: new Date(`2026-08-16T00:00:0${seq}.000Z`), }); - const findMany = vi.fn(async () => [receipt(3), receipt(2), receipt(1)]); - const prisma = { - message: { findMany }, - run: { findMany: vi.fn(async () => [{ id: "run-peer" }]) }, - } as unknown as PrismaClient; + const queryRaw = vi.fn(async () => [receipt(3), receipt(2), receipt(1)]); + const prisma = { $queryRaw: queryRaw } as unknown as PrismaClient; const page = await loadMessagePage(prisma, "thread-1", undefined, 2, undefined, false, true); @@ -432,10 +290,10 @@ describe("thread message pages", () => { "message-receipt-3", ]); expect(page.olderCursor).toBe(2); - expect(findMany).toHaveBeenCalledTimes(1); + expect(queryRaw).toHaveBeenCalledTimes(1); }); - it("queries before the cursor and returns an ascending bounded page", async () => { + it("queries the bot messages view before the cursor as an ascending bounded page", async () => { const findMany = vi.fn(async () => [5, 4, 3].map((seq) => ({ id: `message-${seq}`, @@ -449,7 +307,7 @@ describe("thread message pages", () => { ); const prisma = { message: { findMany } } as unknown as PrismaClient; - const page = await loadMessagePage(prisma, "thread-1", 6, 2); + const page = await loadMessagePage(prisma, "thread-1", 6, 2, undefined, true); expect(findMany).toHaveBeenCalledWith({ where: { threadId: "thread-1", seq: { lt: 6 } }, @@ -461,7 +319,7 @@ describe("thread message pages", () => { }); it("ends pagination when the database returns no lookahead row", async () => { - const findMany = vi.fn(async () => [ + const queryRaw = vi.fn(async () => [ { id: "message-0", threadId: "thread-1", @@ -472,7 +330,7 @@ describe("thread message pages", () => { createdAt: new Date("2026-08-16T00:00:00.000Z"), }, ]); - const prisma = { message: { findMany } } as unknown as PrismaClient; + const prisma = { $queryRaw: queryRaw } as unknown as PrismaClient; const page = await loadMessagePage(prisma, "thread-1", 1, 2); diff --git a/apps/api/src/thread-message-pages.ts b/apps/api/src/thread-message-pages.ts index cbec6ba..4d4fb20 100644 --- a/apps/api/src/thread-message-pages.ts +++ b/apps/api/src/thread-message-pages.ts @@ -1,9 +1,18 @@ import type { MessageBlock, ThreadMessage, ThreadMessagePage } from "@rakazo/contracts"; import { isPeerReceiptBlocks } from "@rakazo/core"; -import type { Prisma, PrismaClient } from "@rakazo/db"; +import type { PrismaClient } from "@rakazo/db"; +import { Prisma } from "@rakazo/db"; type MessageDb = PrismaClient | Prisma.TransactionClient; +/** Peer-run rows that stay in the normal transcript: receipts, ask cards and the bot's own reply. */ +const PEER_VISIBLE_BLOCK_KINDS: readonly string[] = [ + "bot_message_sent", + "bot_message_received", + "ask", + "text", +]; + export async function loadMessagePage( prisma: MessageDb, threadId: string, @@ -47,37 +56,69 @@ export async function loadMessagePage( } } - let cursor = before; - while (true) { + if (includePeerRuns) { const rows = await prisma.message.findMany({ where: { threadId, - ...(cursor === undefined ? {} : { seq: { lt: cursor } }), + ...(before === undefined ? {} : { seq: { lt: before } }), }, orderBy: { seq: "desc" }, take: pageSize + 1, }); - const hasOlder = rows.length > pageSize; - const pageRows = rows.slice(0, pageSize).reverse(); - const visibleRows = includePeerRuns ? pageRows : await withoutPeerRunMessages(prisma, pageRows); + return toPage(threadId, rows, pageSize); + } + + let cursor = before; + while (true) { + const rows = await visibleMessageRows(prisma, threadId, cursor, pageSize + 1); + const page = toPage(threadId, rows, pageSize); // Web hides receipts client-side, so its receipt-only pages keep scanning. // Mobile explicitly retains them and must receive each page for pagination. - const hasSubstantive = visibleRows.some( - (row) => !isPeerReceiptBlocks(row.blocks as MessageBlock[]), - ); - if (hasSubstantive || includePeerReceipts || !hasOlder || includePeerRuns) { - return { - threadId, - messages: visibleRows.map(toThreadMessage), - olderCursor: hasOlder ? (pageRows[0]?.seq ?? null) : null, - }; - } - // TODO: only rescan when a raw page is entirely peer output. Consider a run relation if - // long peer-only histories make this path hot. - cursor = pageRows[0]?.seq; + const hasSubstantive = page.messages.some((message) => !isPeerReceiptBlocks(message.blocks)); + if (hasSubstantive || includePeerReceipts || page.olderCursor === null) return page; + cursor = page.olderCursor; } } +type MessageRow = Parameters[0]; + +function toPage(threadId: string, rows: MessageRow[], pageSize: number): ThreadMessagePage { + const pageRows = rows.slice(0, pageSize).reverse(); + return { + threadId, + messages: pageRows.map(toThreadMessage), + olderCursor: rows.length > pageSize ? (pageRows[0]?.seq ?? null) : null, + }; +} + +/** + * Newest-first transcript rows with peer-run activity excluded in the database, so a long + * peer-only stretch costs one query instead of a rescan per hidden page. + */ +function visibleMessageRows( + prisma: MessageDb, + threadId: string, + before: number | undefined, + limit: number, +): Promise { + return prisma.$queryRaw` + SELECT m.id, m."threadId", m.seq, m.role, m.blocks, m."botId", m."replyToMessageId", + m."replyQuote", m."runId", m."createdAt" + FROM messages m + LEFT JOIN runs r ON r.id = m."runId" + WHERE m."threadId" = ${threadId} + ${before === undefined ? Prisma.empty : Prisma.sql`AND m.seq < ${before}`} + AND (r.trigger IS DISTINCT FROM 'bot_message' OR ${Prisma.join( + PEER_VISIBLE_BLOCK_KINDS.map( + (kind) => Prisma.sql`m.blocks @> ${JSON.stringify([{ kind }])}::jsonb`, + ), + " OR ", + )}) + ORDER BY m.seq DESC + LIMIT ${limit} + `; +} + export async function loadAllMessages( prisma: PrismaClient, threadId: string, @@ -106,15 +147,8 @@ async function withoutPeerRunMessages run.id)); return rows.filter((row) => { if (!row.runId || !peerRunIds.has(row.runId)) return true; - // Keep peer receipts (chips), ask cards, and the bot's own text reply. const blocks = row.blocks as MessageBlock[]; - return blocks.some( - (block) => - block.kind === "bot_message_sent" || - block.kind === "bot_message_received" || - block.kind === "ask" || - block.kind === "text", - ); + return blocks.some((block) => PEER_VISIBLE_BLOCK_KINDS.includes(block.kind)); }); } @@ -159,10 +193,8 @@ export function shouldForwardPeerThreadEvent(event: { !!block && typeof block === "object" && "kind" in block && - (block.kind === "bot_message_received" || - block.kind === "bot_message_sent" || - block.kind === "ask" || - block.kind === "text"), + typeof block.kind === "string" && + PEER_VISIBLE_BLOCK_KINDS.includes(block.kind), ) ); } diff --git a/apps/api/src/thread-target.test.ts b/apps/api/src/thread-target.test.ts index 8658532..199eec9 100644 --- a/apps/api/src/thread-target.test.ts +++ b/apps/api/src/thread-target.test.ts @@ -183,7 +183,7 @@ describe("threadSnapshot", () => { }, ]); const tx = { - $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), + $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue({ seq: 4 }), @@ -203,7 +203,10 @@ describe("threadSnapshot", () => { const snapshot = await threadSnapshot({ prisma }, target); - expect(tx.$queryRaw).toHaveBeenCalledOnce(); + expect(tx.$queryRaw).toHaveBeenCalledWith( + expect.arrayContaining([expect.stringContaining("FOR SHARE")]), + "thread-1", + ); expect(findManyEvents).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ @@ -245,7 +248,7 @@ describe("threadSnapshot", () => { const findManyEvents = vi.fn(); const findFirstRun = botRunFindFirst([run]); const tx = { - $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), + $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), @@ -330,7 +333,7 @@ describe("threadSnapshot", () => { prisma: { $transaction: vi.fn(async (callback: (client: unknown) => unknown) => callback({ - $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), + $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), @@ -385,7 +388,7 @@ describe("threadSnapshot", () => { }; const findFirstRun = botRunFindFirst([failed, completed]); const tx = { - $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), + $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), @@ -421,7 +424,7 @@ describe("threadSnapshot", () => { const findManyEvents = vi.fn(); const findFirstRun = botRunFindFirst([]); const tx = { - $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), + $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), @@ -925,9 +928,16 @@ function groupRunFindMany(input: { active?: unknown[]; terminals?: unknown[] }) ); } +/** The snapshot's thread lock finds the thread; its message page query finds no rows. */ +function threadQueries() { + return vi.fn(async (sql: TemplateStringsArray) => + sql.join("").includes("FOR SHARE") ? [{ id: "thread-1" }] : [], + ); +} + function groupPrisma(findManyRuns: ReturnType) { const tx = { - $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), + $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), diff --git a/packages/testkit/src/cli/harness.ts b/packages/testkit/src/cli/harness.ts index 27415a4..03067b7 100644 --- a/packages/testkit/src/cli/harness.ts +++ b/packages/testkit/src/cli/harness.ts @@ -110,6 +110,7 @@ async function main() { "packages/db/src/main-agent.postgres.test.ts", "packages/db/src/user-profile.postgres.test.ts", "packages/db/src/messaging.postgres.test.ts", + "apps/api/src/thread-message-pages.postgres.test.ts", "packages/memory/src/commit.postgres.test.ts", "packages/adapters/src/wakeup.postgres.test.ts", "packages/adapters/src/realtime.postgres.test.ts",