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

import { ConvexError, v } from "convex/values";

import type { Doc, Id } from "./_generated/dataModel";
import { mutation, query } from "./_generated/server";
import { requireOrganizationMember } from "./authz";

export type ConversationMessageStatus = "admitting" | "admitted" | "failed";

interface ConversationMessageView {
  readonly _id: Id<"conversationMessages">;
  readonly _creationTime: number;
  readonly organizationId: Id<"organizations">;
  readonly conversationId: string;
  readonly messageId: string;
  readonly submissionId: string | null;
  readonly clientRequestId: string;
  readonly role: "user";
  readonly rawText: string;
  readonly status: ConversationMessageStatus;
  readonly createdAt: number;
}

const toView = (doc: Doc<"conversationMessages">): ConversationMessageView => ({
  _creationTime: doc._creationTime,
  _id: doc._id,
  clientRequestId: doc.clientRequestId,
  conversationId: doc.conversationId,
  createdAt: doc.createdAt,
  messageId: doc.messageId,
  organizationId: doc.organizationId,
  rawText: doc.rawText,
  role: doc.role,
  status: doc.status,
  submissionId: doc.submissionId ?? null,
});

/**
 * The global Zopu conversation uses the organization ID as its conversation
 * identity (and agent instance ID). This resolves the conversation ID for the
 * authenticated user's current personal organization, so callers never supply
 * tenancy themselves.
 */
const resolveConversationId = (organizationId: Id<"organizations">): string =>
  organizationId;

/**
 * Idempotently begin a user message at the conversation ingress. If a row with
 * the same (organizationId, clientRequestId) already exists, return it instead
 * of creating a duplicate — Flue retries that replay the same request id must
 * not produce duplicate evidence. The raw text is stored verbatim; it is never
 * trimmed, collapsed, or rewritten.
 *
 * Authorization: the authenticated identity must be a member of the target
 * organization. The conversation ID, message ID, and timestamp are all
 * generated server-side; none are trusted from the caller.
 */
export const beginUserMessage = mutation({
  args: {
    organizationId: v.id("organizations"),
    clientRequestId: v.string(),
    rawText: v.string(),
  },
  handler: async (ctx, args): Promise<ConversationMessageView> => {
    await requireOrganizationMember(ctx, args.organizationId);
    const existing = await ctx.db
      .query("conversationMessages")
      .withIndex("by_organization_and_request", (q) =>
        q
          .eq("organizationId", args.organizationId)
          .eq("clientRequestId", args.clientRequestId)
      )
      .unique();
    if (existing) {
      return toView(existing);
    }
    const createdAt = Date.now();
    const messageId = crypto.randomUUID();
    const conversationId = resolveConversationId(args.organizationId);
    const rowId = await ctx.db.insert("conversationMessages", {
      clientRequestId: args.clientRequestId,
      conversationId,
      createdAt,
      messageId,
      organizationId: args.organizationId,
      rawText: args.rawText,
      role: "user",
      status: "admitting",
    });
    const created = await ctx.db.get(rowId);
    if (!created) {
      throw new Error("Failed to read created conversation message");
    }
    return toView(created);
  },
});

/**
 * Mark a begun message as admitted after Flue accepts the submission. The
 * submission id links the normalized evidence to the Flue transcript. Only an
 * `admitting` message may transition to `admitted`; an already-admitted or
 * failed message is returned unchanged so a retry is idempotent.
 */
export const markAdmitted = mutation({
  args: {
    organizationId: v.id("organizations"),
    clientRequestId: v.string(),
    submissionId: v.string(),
  },
  handler: async (ctx, args): Promise<ConversationMessageView> => {
    await requireOrganizationMember(ctx, args.organizationId);
    const existing = await ctx.db
      .query("conversationMessages")
      .withIndex("by_organization_and_request", (q) =>
        q
          .eq("organizationId", args.organizationId)
          .eq("clientRequestId", args.clientRequestId)
      )
      .unique();
    if (!existing) {
      throw new ConvexError("Conversation message not found");
    }
    if (existing.status === "admitting") {
      await ctx.db.patch(existing._id, {
        status: "admitted",
        submissionId: args.submissionId,
      });
    }
    const updated = await ctx.db.get(existing._id);
    if (!updated) {
      throw new Error("Failed to read updated conversation message");
    }
    return toView(updated);
  },
});

/**
 * Mark a begun message as failed when Flue admission rejects it. Failed
 * messages remain as evidence with an explicit status; they are never silently
 * deleted. Only an `admitting` message may transition to `failed`; an
 * already-admitted message is never overwritten (admission won), and an
 * already-failed message is returned unchanged.
 */
export const markFailed = mutation({
  args: {
    organizationId: v.id("organizations"),
    clientRequestId: v.string(),
  },
  handler: async (ctx, args): Promise<ConversationMessageView> => {
    await requireOrganizationMember(ctx, args.organizationId);
    const existing = await ctx.db
      .query("conversationMessages")
      .withIndex("by_organization_and_request", (q) =>
        q
          .eq("organizationId", args.organizationId)
          .eq("clientRequestId", args.clientRequestId)
      )
      .unique();
    if (!existing) {
      throw new ConvexError("Conversation message not found");
    }
    if (existing.status === "admitting") {
      await ctx.db.patch(existing._id, { status: "failed" });
    }
    const updated = await ctx.db.get(existing._id);
    if (!updated) {
      throw new Error("Failed to read updated conversation message");
    }
    return toView(updated);
  },
});

/**
 * Read a single message by its client request id. Authorization requires
 * membership in the message's organization; cross-organization observation is
 * denied. Returns null when no message matches so observers cannot probe for
 * another organization's request ids.
 */
export const getByRequest = query({
  args: {
    organizationId: v.id("organizations"),
    clientRequestId: v.string(),
  },
  handler: async (ctx, args): Promise<ConversationMessageView | null> => {
    await requireOrganizationMember(ctx, args.organizationId);
    const existing = await ctx.db
      .query("conversationMessages")
      .withIndex("by_organization_and_request", (q) =>
        q
          .eq("organizationId", args.organizationId)
          .eq("clientRequestId", args.clientRequestId)
      )
      .unique();
    return existing ? toView(existing) : null;
  },
});

/**
 * List the user messages for the authenticated user's current organization
 * conversation, newest first. This is the authorized observation path used by
 * GET/SSE routes and product clients. Cross-organization access is denied.
 */
export const listForCurrentOrganization = query({
  args: {
    organizationId: v.id("organizations"),
  },
  handler: async (ctx, args): Promise<ConversationMessageView[]> => {
    await requireOrganizationMember(ctx, args.organizationId);
    const conversationId = resolveConversationId(args.organizationId);
    const rows = await ctx.db
      .query("conversationMessages")
      .withIndex("by_organization_and_message", (q) =>
        q
          .eq("organizationId", args.organizationId)
          .eq("conversationId", conversationId)
      )
      .order("desc")
      .take(100);
    return rows.map(toView);
  },
});