evOo-local/apps/api/src/thread-message-pages.ts

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<ThreadMessagePage> {
  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<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,
  pageSize: number,
): Promise<ThreadMessage[]> {
  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<T extends { runId: string | null; blocks: Prisma.JsonValue }>(
  prisma: MessageDb,
  rows: T[],
): Promise<T[]> {
  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<string, Promise<boolean>>,
): Promise<boolean> {
  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(),
  };
}