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) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YTJRkJrXAS9e3aEFcFYTy8
This commit is contained in:
Claude
parent
2c6b5c3d7b
commit
818e483f47
143
apps/api/src/thread-message-pages.postgres.test.ts
Normal file
143
apps/api/src/thread-message-pages.postgres.test.ts
Normal file
@ -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<void>;
|
||||
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<ReturnType<typeof loadMessagePage>>) =>
|
||||
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"]);
|
||||
});
|
||||
});
|
||||
@ -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);
|
||||
|
||||
|
||||
@ -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<typeof toThreadMessage>[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<MessageRow[]> {
|
||||
return prisma.$queryRaw<MessageRow[]>`
|
||||
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<T extends { runId: string | null; blocks:
|
||||
const peerRunIds = new Set(peerRuns.map((run) => 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),
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
@ -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<typeof groupRunFindMany>) {
|
||||
const tx = {
|
||||
$queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]),
|
||||
$queryRaw: threadQueries(),
|
||||
message: { findMany: vi.fn().mockResolvedValue([]) },
|
||||
event: {
|
||||
findFirst: vi.fn().mockResolvedValue(null),
|
||||
|
||||
@ -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",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user