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; 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<MessageBlock, { kind: "ask" }>;
type ComputerBlock = Extract<MessageBlock, { kind: "computer" }>;
type InboxBot = { id: string; name: string; color: string };
type InboxSpace = NonNullable<InboxItem["space"]>;

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<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. 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<string, InboxSpace>;
}): InboxItem[] {
  const spaceOf = (spaceId: string) => {
    const space = input.otherSpaces?.get(spaceId);
    return space ? { space } : {};
  };
  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,
        ...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<Map<string, InboxSpace>> {
  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<InboxItem[]> {
  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 });
}