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

import { ORPCError } from "@orpc/server";
import type { JobPublisher } from "@rakazo/adapter-kit";
import { type ConnectorRegistry, getCoachRhythm } from "@rakazo/adapters";
import type { Actor, MessageBlock, OnboardingStep } from "@rakazo/contracts";
import { CHOICE_TEXT_ANSWER_ID } from "@rakazo/contracts";
import { featuredConnectorProvidersMatch } from "@rakazo/core";
import {
  appendEventInTransaction,
  createThreadMessageInTransaction,
  ensureMainConversation,
  IsolationError,
  type Prisma,
  type PrismaClient,
  type ThreadEvents,
} from "@rakazo/db";
import {
  type ChoiceBlock,
  closingNote,
  FOCUS_OPTIONS,
  type FocusOption,
  focusCard,
  type IntakeAnswer,
  type IntakeContext,
  intakeContext,
  intakeStep,
  nextIntakeStep,
  readIntakeAnswer,
  secondaryOption,
} from "./onboarding-intake.js";

/**
 * Conversational onboarding, posted deterministically into a bot's thread; no model tokens are
 * spent. The main bot asks the onboarding questions (`onboarding-intake.ts`) in the main
 * conversation, one card after each answer. Any other bot gets one focus card that suggests
 * apps the user authorizes inline. Focus never renames a bot.
 */

type OnboardingDeps = {
  prisma: PrismaClient;
  events: ThreadEvents;
  connectors: ConnectorRegistry;
  jobs: JobPublisher;
};

type Target = { spaceId: string; botId: string; threadId: string };

const APP_DESCRIPTIONS: Record<string, string> = {
  slack: "Search, read, and send messages.",
  gmail: "Search, read, draft, and send email.",
  googlecalendar: "Search events and schedule meetings.",
  notion: "Search and edit pages and databases.",
  googledocs: "Draft and edit documents.",
  hackernews: "Search stories and discussions.",
};

async function requireBotThread(deps: OnboardingDeps, actor: Actor, botId: string) {
  const bot = await deps.prisma.bot.findFirst({
    where: { id: botId, spaceId: actor.spaceId, userId: actor.userId },
    include: { thread: true },
  });
  if (!bot?.thread) throw new IsolationError();
  if (bot.kind === "main") {
    const conversation = await ensureMainConversation(deps.prisma, actor.userId);
    return { bot, thread: { id: conversation.threadId } };
  }
  return { bot, thread: bot.thread };
}

/**
 * Serializes card writes on a thread. NO KEY UPDATE still lets an answer's effects insert rows
 * that reference the thread from another connection (the coach routines do).
 */
async function lockThread(tx: Prisma.TransactionClient, threadId: string): Promise<void> {
  await tx.$executeRaw`SELECT id FROM threads WHERE id = ${threadId} FOR NO KEY UPDATE`;
}

async function postInTransaction(
  tx: Prisma.TransactionClient,
  target: Target,
  blocks: MessageBlock[],
): Promise<{ seq: number }> {
  const message = await createThreadMessageInTransaction(tx, {
    threadId: target.threadId,
    role: "bot",
    blocks,
    botId: target.botId,
  });
  return appendEventInTransaction(tx, {
    spaceId: target.spaceId,
    threadId: target.threadId,
    botId: target.botId,
    type: "thread.message.created",
    payload: { messageId: message.id, role: "bot", blocks },
  });
}

async function updateBlocksInTransaction(
  tx: Prisma.TransactionClient,
  target: Target,
  messageId: string,
  blocks: MessageBlock[],
): Promise<{ seq: number }> {
  await tx.message.update({ where: { id: messageId }, data: { blocks } });
  return appendEventInTransaction(tx, {
    spaceId: target.spaceId,
    threadId: target.threadId,
    botId: target.botId,
    type: "thread.message.updated",
    payload: { messageId, role: "bot", blocks },
  });
}

async function updateBlocks(
  deps: OnboardingDeps,
  target: Target,
  messageId: string,
  blocks: MessageBlock[],
): Promise<void> {
  const event = await deps.prisma.$transaction((tx: Prisma.TransactionClient) =>
    updateBlocksInTransaction(tx, target, messageId, blocks),
  );
  await deps.events.notify(target.threadId, event.seq);
}

/** Sentinel answerId for a focus card the user dismissed without choosing. */
export const FOCUS_DISMISSED_ANSWER_ID = "_dismissed";

/** Checks that the caller may onboard this bot's thread; fresh threads start empty. */
export async function startOnboarding(
  deps: OnboardingDeps,
  actor: Actor,
  botId: string,
): Promise<void> {
  await requireBotThread(deps, actor, botId);
}

function isChoice(block: MessageBlock): block is ChoiceBlock {
  return block.kind === "choice";
}

function pendingChoice(blocks: MessageBlock[]): ChoiceBlock | undefined {
  return blocks.find((block): block is ChoiceBlock => isChoice(block) && !block.answerId);
}

/** The newest card in the thread that still waits for an answer, however long ago it came. */
async function findPendingCard(client: PrismaClient | Prisma.TransactionClient, threadId: string) {
  const messages = await client.message.findMany({
    where: { threadId, role: "bot", blocks: { array_contains: [{ kind: "choice" }] } },
    select: { id: true, blocks: true },
    orderBy: { seq: "desc" },
    take: 20,
  });
  for (const message of messages) {
    const blocks = message.blocks as MessageBlock[];
    const card = pendingChoice(blocks);
    if (card) return { messageId: message.id, blocks, card };
  }
  return null;
}

/**
 * Posts the first card once. The main bot starts the onboarding questions, one set per user;
 * any other bot posts its focus card only while its thread is still idle.
 */
export async function promptFocus(
  deps: OnboardingDeps,
  actor: Actor,
  botId: string,
): Promise<void> {
  const { bot, thread } = await requireBotThread(deps, actor, botId);
  const target = { spaceId: actor.spaceId, botId: bot.id, threadId: thread.id };
  const main = bot.kind === "main";
  // Check + insert + event in one transaction so concurrent callers cannot duplicate cards,
  // and a concurrent user send cannot publish first.
  const committed = await deps.prisma.$transaction(async (tx: Prisma.TransactionClient) => {
    await lockThread(tx, thread.id);
    if (main && (await tx.userOnboarding.findUnique({ where: { userId: actor.userId } }))) {
      return null;
    }
    const recent = await tx.message.findMany({
      where: { threadId: thread.id },
      select: { role: true, blocks: true },
    });
    if (!main && recent.some((message) => message.role === "user")) return null;
    if (recent.some((message) => (message.blocks as MessageBlock[]).some(isChoice))) return null;
    if (!main) return postInTransaction(tx, target, [focusCard()]);
    const first = intakeStep("focus");
    await tx.userOnboarding.create({ data: { userId: actor.userId, step: first.id } });
    return postInTransaction(tx, target, [
      first.card(await loadIntakeContext(tx, actor.userId, new Date())),
    ]);
  });
  if (committed) await deps.events.notify(target.threadId, committed.seq);
}

/** Dismisses the pending card. An onboarding question counts as skipped, so the next follows. */
export async function dismissFocus(
  deps: OnboardingDeps,
  actor: Actor,
  botId: string,
): Promise<void> {
  const { bot, thread } = await requireBotThread(deps, actor, botId);
  const target = { spaceId: actor.spaceId, botId: bot.id, threadId: thread.id };
  const pending = await findPendingCard(deps.prisma, thread.id);
  const skip = pending?.card.step ? secondaryOption(pending.card) : undefined;
  if (pending?.card.step && skip) {
    await chooseFocus(deps, actor, botId, { step: pending.card.step, optionId: skip.id });
    return;
  }
  const event = await deps.prisma.$transaction(async (tx: Prisma.TransactionClient) => {
    await lockThread(tx, thread.id);
    const current = await findPendingCard(tx, thread.id);
    if (!current || current.card.step) return null;
    const blocks = current.blocks.map((block) =>
      block === current.card ? { ...current.card, answerId: FOCUS_DISMISSED_ANSWER_ID } : block,
    );
    return updateBlocksInTransaction(tx, target, current.messageId, blocks);
  });
  if (event) await deps.events.notify(target.threadId, event.seq);
}

export interface ChooseInput {
  step?: OnboardingStep;
  optionId: string;
  text?: string;
  canEnablePush?: boolean;
}

/**
 * Answers the pending card. The answer is checked against that card, a stale answer (another
 * card is pending by now) is ignored, and everything that follows is committed together with
 * it: the focus card's app suggestions, then the next onboarding question or the closing line.
 */
export async function chooseFocus(
  deps: OnboardingDeps,
  actor: Actor,
  botId: string,
  input: ChooseInput,
): Promise<void> {
  const { bot, thread } = await requireBotThread(deps, actor, botId);
  const target = { spaceId: actor.spaceId, botId: bot.id, threadId: thread.id };
  const pending = await findPendingCard(deps.prisma, thread.id);
  if (!pending || pending.card.step !== input.step) return;
  const answer = readIntakeAnswer(pending.card, input.optionId, input.text);
  if (!answer) throw new ORPCError("BAD_REQUEST", { message: "This card has no such answer" });
  const focus =
    !pending.card.step || pending.card.step === "focus"
      ? FOCUS_OPTIONS.find((option) => option.id === answer.option?.id)
      : undefined;
  // Fetched before the transaction: the connector catalog can take seconds.
  const followUp = focus ? await focusFollowUp(deps, actor, bot.id, focus) : [];

  const event = await deps.prisma.$transaction(async (tx: Prisma.TransactionClient) => {
    await lockThread(tx, thread.id);
    const current = await findPendingCard(tx, thread.id);
    if (current?.messageId !== pending.messageId || current.card.step !== input.step) return null;
    const onboarding = current.card.step
      ? await tx.userOnboarding.findUnique({ where: { userId: actor.userId } })
      : null;
    if (current.card.step) {
      if (onboarding?.step !== current.card.step) return null;
      await intakeStep(current.card.step).save(
        { prisma: deps.prisma, jobs: deps.jobs, tx, userId: actor.userId },
        answer,
      );
    }
    const answered = answeredCard(current.card, answer);
    const blocks = current.blocks.map((block) => (block === current.card ? answered : block));
    let last = await updateBlocksInTransaction(tx, target, current.messageId, blocks);
    for (const message of followUp) last = await postInTransaction(tx, target, message);
    if (current.card.step) {
      last = await askNext(deps, tx, target, actor.userId, current.card.step, input.canEnablePush);
    }
    return last;
  });
  if (event) await deps.events.notify(target.threadId, event.seq);
}

function answeredCard(card: ChoiceBlock, answer: IntakeAnswer): ChoiceBlock {
  return answer.option
    ? { ...card, answerId: answer.option.id }
    : { ...card, answerId: CHOICE_TEXT_ANSWER_ID, answerText: answer.text };
}

async function loadIntakeContext(
  tx: Prisma.TransactionClient,
  userId: string,
  now: Date,
): Promise<IntakeContext> {
  const profile = await tx.userProfile.findUnique({
    where: { userId },
    select: { locale: true, timeZone: true },
  });
  return intakeContext(profile, now);
}

/** Posts the next onboarding question, or the closing line once every step is answered. */
async function askNext(
  deps: OnboardingDeps,
  tx: Prisma.TransactionClient,
  target: Target,
  userId: string,
  answeredStep: OnboardingStep,
  canEnablePush = false,
): Promise<{ seq: number }> {
  const now = new Date();
  const context = await loadIntakeContext(tx, userId, now);
  const next = nextIntakeStep(answeredStep, { canEnablePush });
  if (next) {
    await tx.userOnboarding.update({ where: { userId }, data: { step: next.id } });
    return postInTransaction(tx, target, [next.card(context)]);
  }
  // The rhythm step saved its answer outside this transaction, so the routines are readable.
  const rhythm = await getCoachRhythm(deps.prisma, userId);
  await tx.userOnboarding.update({ where: { userId }, data: { step: null, completedAt: now } });
  return postInTransaction(tx, target, [closingNote(rhythm, context)]);
}

/** The apps worth connecting for a focus, as cards; they need no words around them. */
async function focusFollowUp(
  deps: OnboardingDeps,
  actor: Actor,
  botId: string,
  option: FocusOption,
): Promise<MessageBlock[][]> {
  const providers = deps.connectors.managedProviders();
  const catalog = (
    await Promise.all(
      providers.map((provider) =>
        provider
          .catalog({
            operationId: "onboarding.choose",
            traceId: "onboarding.choose",
            spaceId: actor.spaceId,
            userId: actor.userId,
            botId,
            signal: AbortSignal.timeout(15_000),
          })
          .catch(() => []),
      ),
    )
  ).flat();
  const cards = option.apps.flatMap((slug) => {
    const entry = catalog.find(
      (item) =>
        item.slug.toLowerCase() === slug.toLowerCase() ||
        featuredConnectorProvidersMatch(item.slug, slug),
    );
    if (!entry) return [];
    return [
      {
        kind: "app_connect" as const,
        connectorId: entry.connectorId ?? "composio",
        provider: entry.slug,
        name: entry.name,
        description: APP_DESCRIPTIONS[slug] ?? "",
        logo: entry.logo ?? null,
        status: entry.connected ? ("connected" as const) : ("pending" as const),
      },
    ];
  });
  return cards.length > 0 ? [cards] : [];
}

export async function markAppConnected(
  deps: OnboardingDeps,
  actor: Actor,
  botId: string,
  provider: string,
  connectorId = "composio",
): Promise<void> {
  const { bot, thread } = await requireBotThread(deps, actor, botId);
  const target = { spaceId: actor.spaceId, botId: bot.id, threadId: thread.id };
  const messages = await deps.prisma.message.findMany({
    where: { threadId: thread.id },
    select: { id: true, blocks: true },
    orderBy: { createdAt: "asc" },
    take: 100,
  });
  for (const message of messages) {
    const blocks = message.blocks as MessageBlock[];
    if (
      !blocks.some(
        (block) =>
          block.kind === "app_connect" &&
          (block.connectorId ?? "composio") === connectorId &&
          featuredConnectorProvidersMatch(block.provider, provider) &&
          block.status !== "connected",
      )
    )
      continue;
    const next = blocks.map((block) =>
      block.kind === "app_connect" &&
      (block.connectorId ?? "composio") === connectorId &&
      featuredConnectorProvidersMatch(block.provider, provider)
        ? { ...block, status: "connected" as const }
        : block,
    );
    await updateBlocks(deps, target, message.id, next);
  }
}