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

import type { Id } from "./_generated/dataModel";
import type { MutationCtx } from "./_generated/server";

interface ProjectionContext {
  readonly assistantMessageId: Id<"conversationMessages"> | null;
  readonly turnId: Id<"conversationTurns"> | null;
}

const parseStringField = (record: unknown, key: string): string | undefined => {
  if (typeof record !== "object" || record === null) {
    return undefined;
  }
  const value = (record as Record<string, unknown>)[key];
  return typeof value === "string" ? value : undefined;
};

const resolveContext = async (
  ctx: MutationCtx,
  submissionId: string | undefined
): Promise<ProjectionContext> => {
  if (submissionId === undefined) {
    return { assistantMessageId: null, turnId: null };
  }
  const turn = await ctx.db
    .query("conversationTurns")
    .withIndex("by_submissionId", (q) => q.eq("submissionId", submissionId))
    .unique();
  if (turn === null) {
    return { assistantMessageId: null, turnId: null };
  }
  const assistant = await ctx.db
    .query("conversationMessages")
    .withIndex("by_turnId_and_role", (q) =>
      q.eq("turnId", turn._id).eq("role", "assistant")
    )
    .unique();
  return {
    assistantMessageId: assistant?._id ?? null,
    turnId: turn._id,
  };
};

const terminalError = (record: unknown): string => {
  if (typeof record !== "object" || record === null) {
    return "Flue submission failed";
  }
  const value = (record as Record<string, unknown>).error;
  if (typeof value === "string") {
    return value.slice(0, 2000);
  }
  if (typeof value === "object" && value !== null) {
    const { message } = value as Record<string, unknown>;
    if (typeof message === "string") {
      return message.slice(0, 2000);
    }
  }
  return "Flue submission failed";
};

/**
 * Project the small product view from Flue's canonical conversation stream.
 * The raw stream remains authoritative; this function only updates the existing
 * assistant message and turn row that the web already reads.
 */
export const projectConversationRecords = async (
  ctx: MutationCtx,
  recordsJson: string,
  submissionId: string | undefined
): Promise<void> => {
  let records: unknown[];
  try {
    const parsed = JSON.parse(recordsJson) as unknown;
    if (!Array.isArray(parsed)) {
      return;
    }
    records = parsed;
  } catch {
    return;
  }

  const projection = await resolveContext(ctx, submissionId);
  if (projection.turnId === null) {
    return;
  }

  for (const record of records) {
    const type = parseStringField(record, "type");
    if (type === "assistant_text_delta" && projection.assistantMessageId) {
      const delta = parseStringField(record, "delta");
      if (delta !== undefined) {
        const assistant = await ctx.db.get(projection.assistantMessageId);
        if (assistant !== null) {
          await ctx.db.patch(assistant._id, {
            content: `${assistant.content}${delta}`,
          });
        }
      }
      continue;
    }

    if (type !== "submission_settled") {
      continue;
    }

    const turn = await ctx.db.get(projection.turnId);
    if (
      turn === null ||
      turn.status === "completed" ||
      turn.status === "failed" ||
      turn.status === "aborted"
    ) {
      continue;
    }

    const outcome = parseStringField(record, "outcome");
    if (outcome === "completed" && projection.assistantMessageId) {
      const result =
        typeof record === "object" && record !== null
          ? (record as Record<string, unknown>).result
          : undefined;
      const text =
        typeof result === "object" && result !== null
          ? (result as Record<string, unknown>).text
          : undefined;
      if (typeof text === "string") {
        await ctx.db.patch(projection.assistantMessageId, { content: text });
      }
    }

    let status: "aborted" | "completed" | "failed" = "failed";
    if (outcome === "completed") {
      status = "completed";
    } else if (outcome === "aborted") {
      status = "aborted";
    }

    await ctx.db.patch(turn._id, {
      completedAt: Date.now(),
      error: outcome === "failed" ? terminalError(record) : undefined,
      leaseExpiresAt: undefined,
      leaseOwner: undefined,
      status,
    });
  }
};