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",