evOo-local/apps/api/src/thread-cards.postgres.test.ts

import type { JobPublisher } from "@rakazo/adapter-kit";
import type { Actor, CardBlock, MessageBlock } from "@rakazo/contracts";
import { buildCardFromTool, buildSuggestionCard } from "@rakazo/core";
import { approvalEffectKey } from "@rakazo/core/node/approval-effect-key";
import type { PrismaClient, ThreadEvents } from "@rakazo/db";
import { createDb, provisionMessagingIdentity } from "@rakazo/db";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import { actOnThreadCard } from "./thread-cards.js";
import type { ThreadTarget } from "./thread-target.js";
import { resolveThreadTarget } from "./thread-target.js";

const databaseUrl = process.env.DATABASE_URL;
const describePostgres =
  process.env.VERIFY_DATABASE && databaseUrl ? describe.sequential : describe.skip;

describePostgres("actOnThreadCard (PostgreSQL)", () => {
  let prisma: PrismaClient;
  let close: () => Promise<void>;
  let identity: { userId: string; spaceId: string; botId: string; threadId: string };
  let actor: Actor;
  let target: ThreadTarget;
  const notify = vi.fn(async () => undefined);
  const enqueue = vi.fn(async () => undefined);
  let deps: { prisma: PrismaClient; events: ThreadEvents; jobs: JobPublisher };
  let seq = 0;
  let nonce = 0;

  beforeAll(async () => {
    const db = createDb(databaseUrl!);
    prisma = db.prisma;
    close = async () => {
      await db.prisma.$disconnect();
      await db.pool.end();
    };
    identity = await provisionMessagingIdentity(
      prisma,
      { provider: "sendblue", address: `+1556${Date.now().toString().slice(-7)}` },
      { signupsEnabled: undefined, signupAllowlist: undefined },
    );
    actor = {
      userId: identity.userId,
      spaceId: identity.spaceId,
      email: "card@example.com",
      isDeploymentOwner: false,
    };
    deps = {
      prisma,
      events: { notify } as unknown as ThreadEvents,
      jobs: {
        enqueue,
        cancel: vi.fn(async () => undefined),
        close: vi.fn(async () => undefined),
      } as unknown as JobPublisher,
    };
    target = await resolveThreadTarget(prisma, actor, { botId: identity.botId });
    seq =
      (
        await prisma.message.aggregate({
          where: { threadId: identity.threadId },
          _max: { seq: true },
        })
      )._max.seq ?? 0;
  });

  afterAll(async () => {
    if (identity) {
      await prisma.messagingIdentity.deleteMany({ where: { userId: identity.userId } });
      await prisma.organization.deleteMany({ where: { id: identity.spaceId } });
      await prisma.user.deleteMany({ where: { id: identity.userId } });
    }
    await close?.();
  });

  async function postCard(args: Parameters<typeof buildCardFromTool>[0]) {
    const built = buildCardFromTool(args, `card${++seq}`);
    if (!("ok" in built)) throw new Error(built.error);
    const message = await prisma.message.create({
      data: {
        threadId: identity.threadId,
        seq,
        role: "bot",
        botId: identity.botId,
        blocks: [built.block],
      },
    });
    return { messageId: message.id, cardId: built.block.id! };
  }

  async function card(messageId: string) {
    const row = await prisma.message.findUniqueOrThrow({ where: { id: messageId } });
    return (row.blocks as MessageBlock[])[0] as CardBlock;
  }

  const act = (input: Omit<Parameters<typeof actOnThreadCard>[3], "clientNonce">, again?: string) =>
    actOnThreadCard(deps, actor, target, { ...input, clientNonce: again ?? `nonce-${++nonce}` });

  it("ticks a checklist item without starting a run", async () => {
    const posted = await postCard({ title: "Shopping", checklist: ["Milk", "Eggs"] });
    await act({
      ...posted,
      act: { type: "toggle", sectionId: "checklist", itemId: "i2", checked: true },
    });
    const updated = await card(posted.messageId);
    expect(updated.sections?.[1]).toMatchObject({
      items: [{ id: "i1" }, { id: "i2", checked: true }],
    });
    expect(await prisma.message.count({ where: { replyToMessageId: posted.messageId } })).toBe(0);
    expect(
      await prisma.event.count({
        where: {
          threadId: identity.threadId,
          type: "thread.message.updated",
          payload: { path: ["messageId"], equals: posted.messageId },
        },
      }),
    ).toBe(1);
  });

  it("answers a choice once and sends it to the bot as the user's reply", async () => {
    const posted = await postCard({ title: "Dinner", options: ["Pasta", "Curry"] });
    const pick = {
      ...posted,
      act: { type: "select" as const, sectionId: "select", optionId: "o2" },
    };
    await act(pick, "pick-curry");
    // A retried tap is a no-op rather than an error.
    await act(pick, "pick-curry");
    expect((await card(posted.messageId)).answer).toEqual({ text: "Curry" });
    const replies = await prisma.message.findMany({
      where: { replyToMessageId: posted.messageId },
      select: { role: true, blocks: true, runId: true },
    });
    expect(replies).toHaveLength(1);
    expect(replies[0]).toMatchObject({ role: "user", blocks: [{ kind: "text", text: "Curry" }] });
    expect(replies[0]?.runId).toBeTruthy();
    await expect(
      act({ ...posted, act: { type: "select", sectionId: "select", optionId: "o1" } }),
    ).rejects.toThrow("already answered");
  });

  async function postSuggestion(createdAt?: Date) {
    const task = await prisma.task.create({
      data: {
        spaceId: identity.spaceId,
        botId: identity.botId,
        threadId: identity.threadId,
        userId: identity.userId,
        prompt: "I should move more",
        status: "completed",
      },
    });
    const run = await prisma.run.create({
      data: {
        spaceId: identity.spaceId,
        botId: identity.botId,
        threadId: identity.threadId,
        taskId: task.id,
        userId: identity.userId,
        status: "completed",
        trigger: "user",
      },
    });
    const call = {
      tool: "schedule_create",
      args: { name: "Morning walk", prompt: "Take a walk.", cron: "0 8 * * *" },
    };
    const effect = await prisma.externalEffect.create({
      data: {
        spaceId: identity.spaceId,
        runId: run.id,
        kind: "suggestion",
        idempotencyKey: `suggestion:${run.id}:test`,
        status: "proposed",
        request: { ...call, botId: identity.botId, threadId: identity.threadId, subject: "x" },
        ...(createdAt ? { createdAt } : {}),
      },
    });
    const block = buildSuggestionCard({ id: effect.id, effectId: effect.id, ...call });
    const message = await prisma.message.create({
      data: {
        threadId: identity.threadId,
        seq: ++seq,
        role: "bot",
        botId: identity.botId,
        blocks: [block],
      },
    });
    await prisma.externalEffect.update({
      where: { id: effect.id },
      data: {
        request: {
          ...call,
          botId: identity.botId,
          threadId: identity.threadId,
          subject: "x",
          card: { threadId: identity.threadId, messageId: message.id },
        },
      },
    });
    return { messageId: message.id, cardId: effect.id, effectId: effect.id, ...call };
  }

  it("runs an accepted suggestion as an approved run, once", async () => {
    const posted = await postSuggestion();
    enqueue.mockClear();
    const accept = {
      messageId: posted.messageId,
      cardId: posted.cardId,
      act: { type: "action" as const, actionId: "accept" },
    };
    await act(accept, "accept-walk");
    await act(accept, "accept-walk");
    expect((await card(posted.messageId)).suggestion?.status).toBe("running");
    const effect = await prisma.externalEffect.findUniqueOrThrow({
      where: { id: posted.effectId },
    });
    expect(effect.status).toBe("accepted");
    const runId = (effect.result as { runId: string }).runId;
    const run = await prisma.run.findUniqueOrThrow({ where: { id: runId } });
    expect(run).toMatchObject({ trigger: "suggestion", botId: identity.botId, status: "queued" });
    const approved = await prisma.externalEffect.findMany({ where: { runId } });
    expect(approved).toEqual([
      expect.objectContaining({
        kind: "schedule_create",
        status: "approved",
        idempotencyKey: approvalEffectKey(runId, "schedule_create", posted.args),
      }),
    ]);
    expect(enqueue).toHaveBeenCalledTimes(1);
    // No reply goes to the bot; the card itself carries the outcome.
    expect(await prisma.message.count({ where: { replyToMessageId: posted.messageId } })).toBe(0);
    await expect(act({ ...accept, act: { type: "action", actionId: "dismiss" } })).rejects.toThrow(
      "already settled",
    );
  });

  it("dismisses a suggestion without running anything", async () => {
    const posted = await postSuggestion();
    await act({
      messageId: posted.messageId,
      cardId: posted.cardId,
      act: { type: "action", actionId: "dismiss" },
    });
    expect((await card(posted.messageId)).suggestion?.status).toBe("dismissed");
    const effect = await prisma.externalEffect.findUniqueOrThrow({
      where: { id: posted.effectId },
    });
    expect(effect).toMatchObject({ status: "dismissed" });
    expect((effect.result as { runId: unknown }).runId).toBeNull();
  });

  it("lets an old suggestion expire instead of running it", async () => {
    const posted = await postSuggestion(new Date(Date.now() - 8 * 24 * 60 * 60 * 1000));
    await act({
      messageId: posted.messageId,
      cardId: posted.cardId,
      act: { type: "action", actionId: "accept" },
    });
    expect((await card(posted.messageId)).suggestion?.status).toBe("expired");
    expect(
      await prisma.externalEffect.findUniqueOrThrow({ where: { id: posted.effectId } }),
    ).toMatchObject({ status: "expired", result: null });
  });

  it("keeps other threads' cards out of reach", async () => {
    const posted = await postCard({ title: "Elsewhere", checklist: ["One"] });
    await expect(
      act({
        ...posted,
        cardId: "not-this-card",
        act: { type: "toggle", sectionId: "checklist", itemId: "i1", checked: true },
      }),
    ).rejects.toThrow();
  });
});