import type { MessageBlock, ThreadMessage, ThreadMessagePage } from "@rakazo/contracts"; import { isPeerReceiptBlocks } from "@rakazo/core"; 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, before: number | undefined, pageSize: number, around?: { messageId?: string; seq?: number }, includePeerRuns = false, includePeerReceipts = false, ): Promise { if (around) { let targetSeq = around.seq; if (targetSeq === undefined && around.messageId) { const row = await prisma.message.findFirst({ where: { id: around.messageId, threadId }, select: { seq: true }, }); targetSeq = row?.seq; } if (targetSeq !== undefined) { const half = Math.floor(pageSize / 2); const minSeq = Math.max(0, targetSeq - half); const maxSeq = targetSeq + half; const rows = await prisma.message.findMany({ where: { threadId, seq: { gte: minSeq, lte: maxSeq } }, orderBy: { seq: "asc" }, take: pageSize, }); const first = rows[0]; const hasOlder = first ? (await prisma.message.count({ where: { threadId, seq: { lt: first.seq } } })) > 0 : false; // Peer text/activity stays out of the normal transcript (including the // around target). Receipts remain via withoutPeerRunMessages; full peer // history belongs in the bot-messages overlay (includePeerRuns). const messages = includePeerRuns ? rows : await withoutPeerRunMessages(prisma, rows); return { threadId, messages: messages.map(toThreadMessage), olderCursor: hasOlder ? (first?.seq ?? null) : null, }; } } if (includePeerRuns) { const rows = await prisma.message.findMany({ where: { threadId, ...(before === undefined ? {} : { seq: { lt: before } }), }, orderBy: { seq: "desc" }, take: pageSize + 1, }); 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 = 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, pageSize: number, ): Promise { const pages: ThreadMessage[][] = []; let before: number | undefined; do { const page = await loadMessagePage(prisma, threadId, before, pageSize, undefined, true); pages.push(page.messages); before = page.olderCursor ?? undefined; } while (before !== undefined); return pages.reverse().flat(); } async function withoutPeerRunMessages( prisma: MessageDb, rows: T[], ): Promise { const runIds = [...new Set(rows.flatMap((row) => (row.runId ? [row.runId] : [])))]; if (runIds.length === 0) return rows; const peerRuns = await prisma.run.findMany({ where: { id: { in: runIds }, trigger: "bot_message" }, select: { id: true }, }); const peerRunIds = new Set(peerRuns.map((run) => run.id)); return rows.filter((row) => { if (!row.runId || !peerRunIds.has(row.runId)) return true; const blocks = row.blocks as MessageBlock[]; return blocks.some((block) => PEER_VISIBLE_BLOCK_KINDS.includes(block.kind)); }); } export async function isPeerRun( prisma: MessageDb, runId: string | undefined, cache: Map>, ): Promise { if (!runId) return false; let peerRun = cache.get(runId); if (!peerRun) { peerRun = prisma.run .findUnique({ where: { id: runId }, select: { trigger: true } }) .then((run) => run?.trigger === "bot_message"); cache.set(runId, peerRun); } return peerRun; } /** Peer-run SSE events that must still reach an open thread (terminals, waits, receipts, asks, text). */ export function shouldForwardPeerThreadEvent(event: { type: string; payload: { blocks?: unknown }; }): boolean { if ( event.type === "run.completed" || event.type === "run.failed" || event.type === "run.cancelled" || event.type === "run.waiting_input" || event.type === "computer.takeover.requested" ) { return true; } if (event.type !== "thread.message.created" && event.type !== "thread.message.updated") { return false; } const blocks = event.payload.blocks; return ( Array.isArray(blocks) && blocks.some( (block) => !!block && typeof block === "object" && "kind" in block && typeof block.kind === "string" && PEER_VISIBLE_BLOCK_KINDS.includes(block.kind), ) ); } function toThreadMessage(row: { id: string; threadId: string; seq: number; role: string; blocks: Prisma.JsonValue; botId: string | null; replyToMessageId: string | null; replyQuote: string | null; runId: string | null; createdAt: Date; }): ThreadMessage { return { id: row.id, threadId: row.threadId, seq: row.seq, role: row.role as ThreadMessage["role"], blocks: row.blocks as ThreadMessage["blocks"], botId: row.botId ?? undefined, replyToMessageId: row.replyToMessageId ?? undefined, replyQuote: row.replyQuote ?? undefined, runId: row.runId ?? undefined, createdAt: row.createdAt.toISOString(), }; }