import type { Actor, InboxItem, InboxThreadTarget } from "@rakazo/contracts"; import { MessageBlock } from "@rakazo/contracts"; import { isApprovalAskBlock } from "@rakazo/core"; import type { PrismaClient } from "@rakazo/db"; import { promptSnippet } from "./runs.js"; /** Caps per query; the user's spaces rarely hold more waiting runs or unread threads than this. */ const INBOX_RUN_LIMIT = 50; const INBOX_UNREAD_LIMIT = 50; const INBOX_PREVIEW_LENGTH = 280; type AskBlock = Extract; type ComputerBlock = Extract; type InboxBot = { id: string; name: string; color: string }; type InboxSpace = NonNullable; export interface InboxRunRow { id: string; spaceId: string; status: string; updatedAt: Date; bot: InboxBot; thread: { botId: string | null; groupId: string | null; group: { kind: string; archivedAt: Date | null } | null; }; } export interface InboxMessageRow { id: string; runId: string | null; blocks: unknown; createdAt: Date; } export interface InboxUnreadThreadRow { spaceId: string; createdAt: Date; bot: InboxBot | null; /** The newest bot message, if any. */ messages: Array>; } function parseBlocks(blocks: unknown): MessageBlock[] { const parsed = MessageBlock.array().safeParse(blocks); return parsed.success ? parsed.data : []; } function pendingAsk(blocks: MessageBlock[]): AskBlock | undefined { return blocks.find( (block): block is AskBlock => block.kind === "ask" && block.status !== "answered", ); } function takeoverRequest(blocks: MessageBlock[]): ComputerBlock | undefined { return blocks.find((block): block is ComputerBlock => block.kind === "computer"); } /** The newest message whose blocks hold what `pick` looks for. */ function newest(messages: InboxMessageRow[], pick: (blocks: MessageBlock[]) => T | undefined) { for (const message of messages) { const block = pick(parseBlocks(message.blocks)); if (block) return { message, block }; } return undefined; } function previewText(blocks: unknown): string { for (const block of parseBlocks(blocks)) { if ("text" in block && typeof block.text === "string" && block.text.trim()) { return promptSnippet(block.text, INBOX_PREVIEW_LENGTH); } } return ""; } /** * Where an ask can be answered from. The open space's main conversation already shows its own * asks (a project's main conversation does not), and an archived group or an external * conversation has no thread target to answer through. */ function askTarget(thread: InboxRunRow["thread"], currentSpace: boolean): InboxThreadTarget | null { if (thread.groupId) { if (!thread.group || thread.group.archivedAt) return null; if (thread.group.kind === "main" && currentSpace) return null; return { groupId: thread.groupId }; } return thread.botId ? { botId: thread.botId } : null; } function targetKey(target: InboxThreadTarget): string { return "botId" in target ? `bot:${target.botId}` : `group:${target.groupId}`; } function byCreatedAt(left: InboxItem, right: InboxItem): number { return left.createdAt.localeCompare(right.createdAt); } /** * Turns waiting runs, their ask and takeover messages, and unread threads into inbox items: * everything that waits on the user first, then unread threads not already covered, each * oldest first. Items from another space name it. */ export function buildInboxItems(input: { runs: InboxRunRow[]; /** Ask and takeover messages of those runs, newest first. */ messages: InboxMessageRow[]; unreadThreads: InboxUnreadThreadRow[]; /** The user's other spaces by id; rows of any space not listed belong to the current one. */ otherSpaces?: ReadonlyMap; }): InboxItem[] { const spaceOf = (spaceId: string) => { const space = input.otherSpaces?.get(spaceId); return space ? { space } : {}; }; const messagesByRun = new Map(); for (const message of input.messages) { if (!message.runId) continue; const list = messagesByRun.get(message.runId) ?? []; list.push(message); messagesByRun.set(message.runId, list); } const waiting: InboxItem[] = []; for (const run of input.runs) { const messages = messagesByRun.get(run.id) ?? []; if (run.status === "waiting_input") { const target = askTarget(run.thread, !input.otherSpaces?.has(run.spaceId)); const pending = target ? newest(messages, pendingAsk) : undefined; if (!target || !pending) continue; waiting.push({ kind: isApprovalAskBlock(pending.block) ? "approval" : "ask", bot: run.bot, target, ...spaceOf(run.spaceId), runId: run.id, messageId: pending.message.id, ask: pending.block, createdAt: pending.message.createdAt.toISOString(), }); } else if (run.status === "waiting_takeover") { // Takeover controls live in the bot's own thread, so the item stays even for a run that // asked in the main conversation. const request = newest(messages, takeoverRequest); waiting.push({ kind: "takeover", bot: run.bot, target: { botId: run.bot.id }, ...spaceOf(run.spaceId), runId: run.id, messageId: request?.message.id ?? null, text: request?.block.text ?? "", createdAt: (request?.message.createdAt ?? run.updatedAt).toISOString(), }); } } const covered = new Set(waiting.map((item) => targetKey(item.target))); const unread: InboxItem[] = []; for (const thread of input.unreadThreads) { if (!thread.bot) continue; const target = { botId: thread.bot.id }; if (covered.has(targetKey(target))) continue; const latest = thread.messages[0]; unread.push({ kind: "unread", bot: thread.bot, target, ...spaceOf(thread.spaceId), messageId: latest?.id ?? null, text: previewText(latest?.blocks), createdAt: (latest?.createdAt ?? thread.createdAt).toISOString(), }); } return [...waiting.sort(byCreatedAt), ...unread.sort(byCreatedAt)]; } const botSelect = { id: true, name: true, color: true } as const; /** * The other spaces whose items the inbox shows: every space the user belongs to, except spaces * being deleted. */ async function otherInboxSpaces( prisma: PrismaClient, actor: Actor, ): Promise> { const memberships = await prisma.spaceMember.findMany({ where: { userId: actor.userId, spaceId: { not: actor.spaceId }, space: { deletingAt: null } }, select: { space: { select: { id: true, name: true } } }, }); return new Map(memberships.map(({ space }) => [space.id, space])); } /** * The inbox across the user's spaces: only the actor's own bots, runs and threads. Items are * answered and opened in their own space, where the thread procedures check membership and * ownership again. */ export async function listInbox(prisma: PrismaClient, actor: Actor): Promise { const otherSpaces = await otherInboxSpaces(prisma, actor); const spaceIds = [actor.spaceId, ...otherSpaces.keys()]; const ownBot = { userId: actor.userId, archivedAt: null }; const [runs, unreadThreads] = await Promise.all([ prisma.run.findMany({ where: { spaceId: { in: spaceIds }, userId: actor.userId, status: { in: ["waiting_input", "waiting_takeover"] }, bot: ownBot, thread: { userId: actor.userId }, }, select: { id: true, spaceId: true, status: true, threadId: true, updatedAt: true, bot: { select: botSelect }, thread: { select: { botId: true, groupId: true, group: { select: { kind: true, archivedAt: true } }, }, }, }, orderBy: [{ updatedAt: "asc" }, { id: "asc" }], take: INBOX_RUN_LIMIT, }), // Only specialists' own threads; group threads, the main conversation included, stay out. prisma.thread.findMany({ where: { spaceId: { in: spaceIds }, userId: actor.userId, unread: true, bot: { ...ownBot, kind: "specialist" }, }, select: { spaceId: true, createdAt: true, bot: { select: botSelect }, messages: { where: { role: "bot" }, orderBy: { seq: "desc" }, take: 1, select: { id: true, blocks: true, createdAt: true }, }, }, orderBy: [{ createdAt: "asc" }, { id: "asc" }], take: INBOX_UNREAD_LIMIT, }), ]); const messages = runs.length === 0 ? [] : await prisma.message.findMany({ where: { threadId: { in: [...new Set(runs.map((run) => run.threadId))] }, runId: { in: runs.map((run) => run.id) }, role: "bot", OR: [ { blocks: { array_contains: [{ kind: "ask" }] } }, { blocks: { array_contains: [{ kind: "computer" }] } }, ], }, orderBy: { seq: "desc" }, select: { id: true, runId: true, blocks: true, createdAt: true }, }); return buildInboxItems({ runs, messages, unreadThreads, otherSpaces }); }