zopu-code/packages/backend/convex/conversationMessages.test.ts

import { convexTest } from "convex-test";
import { anyApi, makeFunctionReference } from "convex/server";
import { describe, expect, test } from "vitest";

import schema from "./schema";

declare global {
  interface ImportMeta {
    readonly glob: (pattern: string) => Record<string, () => Promise<unknown>>;
  }
}

const modules = import.meta.glob("./**/*.ts");
const api = anyApi;
const identityA = { tokenIdentifier: "https://convex.test|user-a" };
const identityB = { tokenIdentifier: "https://convex.test|user-b" };
const newTest = () => convexTest({ modules, schema });
const markProcessingRef = makeFunctionReference<
  "mutation",
  { turnId: string; attempt: number; leaseOwner: string },
  boolean
>("conversationMessages:markProcessing");
const completeTurnRef = makeFunctionReference<
  "mutation",
  {
    turnId: string;
    attempt: number;
    leaseOwner: string;
    submissionId: string;
    text: string;
  },
  boolean
>("conversationMessages:completeTurn");
const failTurnRef = makeFunctionReference<
  "mutation",
  {
    turnId: string;
    attempt: number;
    leaseOwner: string;
    error: string;
    retry: boolean;
  },
  boolean
>("conversationMessages:failTurn");

const ensureOrg = async (
  t: ReturnType<typeof newTest>,
  identity: { readonly tokenIdentifier: string }
) =>
  await t
    .withIdentity(identity)
    .mutation(api.organizations.ensurePersonalOrganization, {});

describe("conversationMessages", () => {
  test("queues one durable user turn and one assistant placeholder", async () => {
    const t = newTest();
    const organization = await ensureOrg(t, identityA);

    await t.withIdentity(identityA).mutation(api.conversationMessages.send, {
      clientRequestId: "request-1",
      images: [],
      organizationId: organization._id,
      rawText: "  Keep this exact text.  ",
    });

    const rows = await t
      .withIdentity(identityA)
      .query(api.conversationMessages.listForCurrentOrganization, {
        organizationId: organization._id,
      });
    expect(rows).toHaveLength(2);
    expect(rows[0]).toMatchObject({
      rawText: "  Keep this exact text.  ",
      role: "user",
      status: "queued",
    });
    expect(rows[1]).toMatchObject({
      rawText: "",
      role: "assistant",
      status: "queued",
    });
  });

  test("is idempotent by client request id", async () => {
    const t = newTest();
    const organization = await ensureOrg(t, identityA);
    const input = {
      clientRequestId: "request-duplicate",
      images: [],
      organizationId: organization._id,
      rawText: "First payload wins",
    };

    const first = await t
      .withIdentity(identityA)
      .mutation(api.conversationMessages.send, input);
    const second = await t
      .withIdentity(identityA)
      .mutation(api.conversationMessages.send, {
        ...input,
        rawText: "Ignored duplicate payload",
      });

    expect(second).toEqual(first);
    const rows = await t
      .withIdentity(identityA)
      .query(api.conversationMessages.listForCurrentOrganization, {
        organizationId: organization._id,
      });
    expect(rows).toHaveLength(2);
    expect(rows[0]?.rawText).toBe("First payload wins");
  });

  test("rejects unauthenticated and cross-organization access", async () => {
    const t = newTest();
    const organization = await ensureOrg(t, identityA);
    await ensureOrg(t, identityB);
    const input = {
      clientRequestId: "request-private",
      images: [],
      organizationId: organization._id,
      rawText: "Private",
    };

    await expect(
      t.mutation(api.conversationMessages.send, input)
    ).rejects.toThrow(/Authentication required/u);
    await expect(
      t.withIdentity(identityB).mutation(api.conversationMessages.send, input)
    ).rejects.toThrow(/Organization membership required/u);
    await expect(
      t
        .withIdentity(identityB)
        .query(api.conversationMessages.listForCurrentOrganization, {
          organizationId: organization._id,
        })
    ).rejects.toThrow(/Organization membership required/u);
  });

  test("fences stale turn attempts from overwriting a retry", async () => {
    const t = newTest();
    const organization = await ensureOrg(t, identityA);
    const sent = await t
      .withIdentity(identityA)
      .mutation(api.conversationMessages.send, {
        clientRequestId: "request-fenced",
        images: [],
        organizationId: organization._id,
        rawText: "Keep only the current response",
      });

    expect(
      await t.mutation(markProcessingRef, {
        attempt: 1,
        leaseOwner: "worker-1",
        turnId: sent.turnId,
      })
    ).toBe(true);
    expect(
      await t.mutation(failTurnRef, {
        attempt: 1,
        error: "retry",
        leaseOwner: "worker-1",
        retry: true,
        turnId: sent.turnId,
      })
    ).toBe(true);
    expect(
      await t.mutation(completeTurnRef, {
        attempt: 1,
        leaseOwner: "worker-1",
        submissionId: "stale",
        text: "stale response",
        turnId: sent.turnId,
      })
    ).toBe(false);
    expect(
      await t.mutation(markProcessingRef, {
        attempt: 2,
        leaseOwner: "worker-2",
        turnId: sent.turnId,
      })
    ).toBe(true);
    expect(
      await t.mutation(completeTurnRef, {
        attempt: 2,
        leaseOwner: "worker-2",
        submissionId: "current",
        text: "current response",
        turnId: sent.turnId,
      })
    ).toBe(true);

    const messages = await t
      .withIdentity(identityA)
      .query(api.conversationMessages.listForCurrentOrganization, {
        organizationId: organization._id,
      });
    expect(messages[1]?.rawText).toBe("current response");
  });
});