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

import { composeSignal, SignalValidationError } from "@code/primitives/signal";
import { ConvexError, v } from "convex/values";
import { Effect } from "effect";

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

const PROBLEM_STATEMENT = v.object({
  title: v.string(),
  summary: v.string(),
  desiredOutcome: v.string(),
  constraints: v.array(v.string()),
});

interface SignalSourceView {
  readonly _id: Id<"signalSources">;
  readonly _creationTime: number;
  readonly signalId: Id<"signals">;
  readonly organizationId: Id<"organizations">;
  readonly conversationId: string;
  readonly messageId: string;
  readonly ordinal: number;
  readonly rawTextSnapshot: string;
  readonly sourceCreatedAt: number;
  readonly submissionId: string | null;
}

interface SignalView {
  readonly _id: Id<"signals">;
  readonly _creationTime: number;
  readonly organizationId: Id<"organizations">;
  readonly projectId: Id<"projects"> | null;
  readonly conversationId: string;
  readonly problemStatement: {
    readonly title: string;
    readonly summary: string;
    readonly desiredOutcome: string;
    readonly constraints: readonly string[];
  };
  readonly processedByAgentName: string;
  readonly processedByAgentInstanceId: string;
  readonly createdAt: number;
}

interface ComposedSignal {
  readonly signal: SignalView;
  readonly sources: readonly SignalSourceView[];
}

/**
 * An admitted user conversation message presented to the agent as candidate
 * Signal evidence. The agent selects from these by `messageId`; raw text and
 * timestamps are copied server-side and never accepted from the agent.
 */
interface ConversationMessageSourceView {
  readonly messageId: string;
  readonly clientRequestId: string;
  readonly rawText: string;
  readonly submissionId: string | null;
  readonly createdAt: number;
}

const toSignalView = (doc: Doc<"signals">): SignalView => ({
  _creationTime: doc._creationTime,
  _id: doc._id,
  conversationId: doc.conversationId,
  createdAt: doc.createdAt,
  organizationId: doc.organizationId,
  problemStatement: doc.problemStatement,
  processedByAgentInstanceId: doc.processedByAgentInstanceId,
  processedByAgentName: doc.processedByAgentName,
  projectId: doc.projectId ?? null,
});

const toSourceView = (doc: Doc<"signalSources">): SignalSourceView => ({
  _creationTime: doc._creationTime,
  _id: doc._id,
  conversationId: doc.conversationId,
  messageId: doc.messageId,
  ordinal: doc.ordinal,
  organizationId: doc.organizationId,
  rawTextSnapshot: doc.rawTextSnapshot,
  signalId: doc.signalId,
  sourceCreatedAt: doc.sourceCreatedAt,
  submissionId: doc.submissionId ?? null,
});

const toConversationMessageSourceView = (
  doc: Doc<"conversationMessages">
): ConversationMessageSourceView => ({
  clientRequestId: doc.clientRequestId,
  createdAt: doc.createdAt,
  messageId: doc.messageId,
  rawText: doc.rawText,
  submissionId: doc.submissionId ?? null,
});

/**
 * Build a deterministic, collision-safe key from the organization, the
 * conversation, and the ordered message selection. The key is the JSON
 * encoding of `[organizationId, conversationId, ...messageIds]`, so no
 * delimiter inside arbitrary message IDs can ever cause a collision and the
 * caller's selection order is part of the identity.
 */
const buildSourceKey = (
  organizationId: Id<"organizations">,
  conversationId: string,
  messageIds: readonly string[]
): string => JSON.stringify([organizationId, conversationId, ...messageIds]);

/**
 * Resolve all selected conversation messages server-side. Each message must
 * belong to the same organization and conversation, be a `user` message, be
 * `admitted`, and appear exactly once in the ordered selection. Raw text,
 * timestamps, and submission ids are copied from the stored rows; the agent
 * never supplies them. Returns the messages in the caller's requested order.
 */
const resolveSources = async (
  ctx: MutationCtx,
  organizationId: Id<"organizations">,
  conversationId: string,
  messageIds: readonly string[]
): Promise<Doc<"conversationMessages">[]> => {
  if (messageIds.length === 0) {
    throw new ConvexError("At least one source message is required");
  }
  if (new Set(messageIds).size !== messageIds.length) {
    throw new ConvexError("Source message ids must be unique");
  }
  const resolved: Doc<"conversationMessages">[] = [];
  for (const messageId of messageIds) {
    const doc = await ctx.db
      .query("conversationMessages")
      .withIndex("by_organization_and_message", (q) =>
        q
          .eq("organizationId", organizationId)
          .eq("conversationId", conversationId)
          .eq("messageId", messageId)
      )
      .unique();
    if (!doc) {
      throw new ConvexError(`Source message not found: ${messageId}`);
    }
    if (doc.role !== "user") {
      throw new ConvexError(
        `Source message is not a user message: ${messageId}`
      );
    }
    if (doc.status !== "admitted") {
      throw new ConvexError(`Source message is not admitted: ${messageId}`);
    }
    resolved.push(doc);
  }
  return resolved;
};

/**
 * Resolve whether an optional project belongs to the same organization. Until
 * projects carry an explicit `organizationId`, a project is attachable to an
 * organization's signal only when the project owner is a member of that
 * organization. The caller is already proven to be a member, so this is a
 * security-preserving temporary mapping: an organization's project set is the
 * intersection of (caller is a member) and (project owner is a member), which
 * cannot weaken cross-tenant isolation.
 */
const authorizeProject = async (
  ctx: MutationCtx,
  organizationId: Id<"organizations">,
  projectId: Id<"projects">
): Promise<void> => {
  const project = await ctx.db.get(projectId);
  if (!project) {
    throw new ConvexError("Project not found");
  }
  const ownerMembership = await ctx.db
    .query("organizationMembers")
    .withIndex("by_organizationId_and_userId", (q) =>
      q.eq("organizationId", organizationId).eq("userId", project.ownerId)
    )
    .unique();
  if (!ownerMembership) {
    throw new ConvexError("Project does not belong to the target organization");
  }
};

/**
 * Atomically create one idempotent Signal from an ordered selection of user
 * messages plus a structured problem statement produced by the agent.
 *
 * Authorization: the authenticated identity must be a member of the target
 * organization. When a project is supplied it must belong to the same
 * organization. Every message id is resolved server-side from
 * `conversationMessages` under the same organization and conversation; raw
 * text, timestamps, and submission ids are copied from those rows and never
 * accepted from the agent.
 *
 * Idempotency: if a Signal with the same `(organizationId, sourceKey)` already
 * exists, it is returned as-is without creating duplicate sources. The whole
 * lookup-or-insert runs in one Convex mutation transaction, so retries cannot
 * produce duplicates.
 *
 * Returns the new (or existing) signal id.
 */
export const createFromMessages = mutation({
  args: {
    organizationId: v.id("organizations"),
    projectId: v.optional(v.id("projects")),
    conversationId: v.string(),
    messageIds: v.array(v.string()),
    problemStatement: PROBLEM_STATEMENT,
    processedBy: v.object({
      agentName: v.string(),
      agentInstanceId: v.string(),
    }),
  },
  handler: async (ctx, args): Promise<{ signalId: Id<"signals"> }> => {
    await requireOrganizationMember(ctx, args.organizationId);
    if (args.projectId) {
      await authorizeProject(ctx, args.organizationId, args.projectId);
    }

    // The global conversation id equals the organization id; the caller's
    // conversation id must match.
    if (args.conversationId !== args.organizationId) {
      throw new ConvexError("Conversation does not belong to organization");
    }

    const sourceKey = buildSourceKey(
      args.organizationId,
      args.conversationId,
      args.messageIds
    );

    const existing = await ctx.db
      .query("signals")
      .withIndex("by_organization_and_sourceKey", (q) =>
        q.eq("organizationId", args.organizationId).eq("sourceKey", sourceKey)
      )
      .unique();
    if (existing) {
      return { signalId: existing._id };
    }

    const sources = await resolveSources(
      ctx,
      args.organizationId,
      args.conversationId,
      args.messageIds
    );

    // Compose the pure Effect Signal before writes, using a prospective id.
    // `composeSignal` decodes an `unknown` input, so pass plain objects.
    const now = Date.now();
    const prospectiveId = crypto.randomUUID();
    const scope =
      args.projectId === undefined
        ? { _tag: "Organization" as const, organizationId: args.organizationId }
        : {
            _tag: "Project" as const,
            organizationId: args.organizationId,
            projectId: args.projectId,
          };

    const signal = await Effect.runPromise(
      composeSignal({
        createdAt: now,
        id: prospectiveId,
        problemStatement: args.problemStatement,
        processedBy: args.processedBy,
        scope,
        sourceMessages: sources.map((doc) => ({
          conversationId: doc.conversationId,
          createdAt: doc.createdAt,
          messageId: doc.messageId,
          rawText: doc.rawText,
        })),
      })
    ).catch((error: unknown) => {
      if (error instanceof SignalValidationError) {
        throw new ConvexError(`Invalid signal: ${error.message}`);
      }
      throw new ConvexError(
        `Invalid signal: ${error instanceof Error ? error.message : "unknown"}`
      );
    });

    const signalId = await ctx.db.insert("signals", {
      conversationId: args.conversationId,
      createdAt: now,
      organizationId: args.organizationId,
      problemStatement: {
        constraints: [...signal.problemStatement.constraints],
        desiredOutcome: signal.problemStatement.desiredOutcome,
        summary: signal.problemStatement.summary,
        title: signal.problemStatement.title,
      },
      processedByAgentInstanceId: args.processedBy.agentInstanceId,
      processedByAgentName: args.processedBy.agentName,
      ...(args.projectId === undefined ? {} : { projectId: args.projectId }),
      sourceKey,
    });

    await Promise.all(
      sources.map((doc, ordinal) =>
        ctx.db.insert("signalSources", {
          conversationId: doc.conversationId,
          messageId: doc.messageId,
          ordinal,
          organizationId: args.organizationId,
          rawTextSnapshot: doc.rawText,
          signalId,
          ...(doc.submissionId === undefined
            ? {}
            : { submissionId: doc.submissionId }),
          sourceCreatedAt: doc.createdAt,
        })
      )
    );

    return { signalId };
  },
});

/**
 * Load the composed source rows for a signal in ordinal order.
 */
const loadComposedSources = async (
  ctx: QueryCtx,
  signal: Doc<"signals">
): Promise<SignalSourceView[]> => {
  const rows = await ctx.db
    .query("signalSources")
    .withIndex("by_signal_and_ordinal", (q) => q.eq("signalId", signal._id))
    .collect();
  // `collect` does not guarantee index order across all backends; sort by
  // ordinal to be deterministic.
  rows.sort((a, b) => a.ordinal - b.ordinal);
  return rows.map(toSourceView);
};

/**
 * List the Signals for the authenticated user's organization, newest first.
 * Membership is required; cross-organization access is denied. Each entry
 * includes its composed source records in ordinal order.
 */
export const list = query({
  args: {
    organizationId: v.id("organizations"),
  },
  handler: async (ctx, args): Promise<ComposedSignal[]> => {
    await requireOrganizationMember(ctx, args.organizationId);
    const signals = await ctx.db
      .query("signals")
      .withIndex("by_organization_and_createdAt", (q) =>
        q.eq("organizationId", args.organizationId)
      )
      .order("desc")
      .take(50);
    return Promise.all(
      signals.map(async (signal) => ({
        signal: toSignalView(signal),
        sources: await loadComposedSources(ctx, signal),
      }))
    );
  },
});

/**
 * List the bounded set of admitted user messages in the organization's
 * conversation that have not yet been consumed as a Signal source, newest
 * first. These are the candidate messages an agent should select from when
 * composing a new Signal.
 *
 * Authorization: the authenticated identity must be a member of the target
 * organization. The conversation id equals the organization id; the caller
 * never supplies tenancy. Cross-organization access is denied.
 *
 * Bounded: capped at 100 rows to keep agent evidence selection tractable.
 */
export const listAdmittedUnused = query({
  args: {
    organizationId: v.id("organizations"),
  },
  handler: async (ctx, args): Promise<ConversationMessageSourceView[]> => {
    await requireOrganizationMember(ctx, args.organizationId);
    const conversationId = args.organizationId;

    const admitted = await ctx.db
      .query("conversationMessages")
      .withIndex("by_organization_and_message", (q) =>
        q
          .eq("organizationId", args.organizationId)
          .eq("conversationId", conversationId)
      )
      .order("desc")
      .take(100);

    const candidates = admitted.filter(
      (message) => message.role === "user" && message.status === "admitted"
    );

    const unused: ConversationMessageSourceView[] = [];
    for (const message of candidates) {
      const consumed = await ctx.db
        .query("signalSources")
        .withIndex("by_organization_and_message", (q) =>
          q
            .eq("organizationId", args.organizationId)
            .eq("conversationId", conversationId)
            .eq("messageId", message.messageId)
        )
        .first();
      if (!consumed) {
        unused.push(toConversationMessageSourceView(message));
      }
    }

    return unused;
  },
});

/**
 * Get one Signal by id with its composed sources. Membership in the signal's
 * organization is required; access to another organization's signal is denied.
 */
export const get = query({
  args: {
    signalId: v.id("signals"),
  },
  handler: async (ctx, args): Promise<ComposedSignal | null> => {
    const signal = await ctx.db.get(args.signalId);
    if (!signal) {
      return null;
    }
    await requireOrganizationMember(ctx, signal.organizationId);
    return {
      signal: toSignalView(signal),
      sources: await loadComposedSources(ctx, signal),
    };
  },
});