evOo-local/apps/api/src/inbox.ts

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; a space rarely holds 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<MessageBlock, { kind: "ask" }>;
type ComputerBlock = Extract<MessageBlock, { kind: "computer" }>;
type InboxBot = { id: string; name: string; color: string };

export interface InboxRunRow {
  id: 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 {
  createdAt: Date;
  bot: InboxBot | null;
  /** The newest bot message, if any. */
  messages: Array<Omit<InboxMessageRow, "runId">>;
}

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<T>(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 main conversation already shows its own asks, and an
 * archived group or an external conversation has no thread target to answer through.
 */
function askTarget(thread: InboxRunRow["thread"]): InboxThreadTarget | null {
  if (thread.groupId) {
    if (!thread.group || thread.group.kind === "main" || thread.group.archivedAt) 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.
 */
export function buildInboxItems(input: {
  runs: InboxRunRow[];
  /** Ask and takeover messages of those runs, newest first. */
  messages: InboxMessageRow[];
  unreadThreads: InboxUnreadThreadRow[];
}): InboxItem[] {
  const messagesByRun = new Map<string, InboxMessageRow[]>();
  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);
      const pending = target ? newest(messages, pendingAsk) : undefined;
      if (!target || !pending) continue;
      waiting.push({
        kind: isApprovalAskBlock(pending.block) ? "approval" : "ask",
        bot: run.bot,
        target,
        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 },
        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,
      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 current space's inbox: only the actor's own bots, runs and threads. */
export async function listInbox(prisma: PrismaClient, actor: Actor): Promise<InboxItem[]> {
  const ownBot = { userId: actor.userId, archivedAt: null };
  const [runs, unreadThreads] = await Promise.all([
    prisma.run.findMany({
      where: {
        spaceId: actor.spaceId,
        userId: actor.userId,
        status: { in: ["waiting_input", "waiting_takeover"] },
        bot: ownBot,
        thread: { userId: actor.userId },
      },
      select: {
        id: 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: actor.spaceId,
        userId: actor.userId,
        unread: true,
        bot: { ...ownBot, kind: "specialist" },
      },
      select: {
        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 });
}