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

import { v } from "convex/values";

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

/**
 * The system actors used by the coordinator and agent tooling. The existing
 * `events.actor` union already permits `{ kind: "system", service }` and
 * `{ kind: "tool", tool }`; these constants name the canonical services.
 */
export const SYSTEM_SERVICE_PROJECT_SETUP = "project-setup";
export const SYSTEM_SERVICE_AGENT_GATEWAY = "agent-gateway";

interface AppendInput {
  readonly actor:
    | { readonly kind: "system"; readonly service: string }
    | { readonly kind: "tool"; readonly tool: string };
  readonly artifactIds?: readonly Id<"artifacts">[];
  readonly causationId?: Id<"events">;
  readonly correlationId: string;
  readonly idempotencyKey: string;
  readonly organizationId: Id<"organizations">;
  readonly payload: unknown;
  readonly projectId?: Id<"projects">;
  readonly replyToEventId?: Id<"events">;
  readonly scopeKind: "global" | "work";
  readonly threadId?: Id<"threads">;
  readonly type: string;
  readonly visibility: "timeline" | "compact" | "internal";
  readonly workId?: Id<"works">;
}

/**
 * Append an immutable system/tool event with exact-once semantics: an event
 * with the same `(organizationId, idempotencyKey)` is returned unchanged on
 * retry. This is the shared primitive every coordinator phase and agent tool
 * writes through.
 */
const appendEventOnce = async (
  ctx: MutationCtx,
  input: AppendInput
): Promise<Id<"events">> => {
  const existing = await ctx.db
    .query("events")
    .withIndex("by_organizationId_and_idempotencyKey", (q) =>
      q
        .eq("organizationId", input.organizationId)
        .eq("idempotencyKey", input.idempotencyKey)
    )
    .unique();
  if (existing) {
    return existing._id;
  }
  const now = Date.now();
  return await ctx.db.insert("events", {
    actor: input.actor,
    artifactIds: input.artifactIds ? [...input.artifactIds] : undefined,
    causationId: input.causationId,
    correlationId: input.correlationId,
    idempotencyKey: input.idempotencyKey,
    occurredAt: now,
    organizationId: input.organizationId,
    payload: input.payload,
    projectId: input.projectId,
    recordedAt: now,
    replyToEventId: input.replyToEventId,
    scopeKind: input.scopeKind,
    threadId: input.threadId,
    type: input.type,
    visibility: input.visibility,
    workId: input.workId,
  });
};

export const appendSystemEvent = internalMutation({
  args: {
    actorService: v.string(),
    artifactIds: v.optional(v.array(v.id("artifacts"))),
    causationId: v.optional(v.id("events")),
    correlationId: v.string(),
    idempotencyKey: v.string(),
    organizationId: v.id("organizations"),
    payload: v.any(),
    projectId: v.optional(v.id("projects")),
    replyToEventId: v.optional(v.id("events")),
    scopeKind: v.union(v.literal("global"), v.literal("work")),
    threadId: v.optional(v.id("threads")),
    type: v.string(),
    visibility: v.union(
      v.literal("timeline"),
      v.literal("compact"),
      v.literal("internal")
    ),
    workId: v.optional(v.id("works")),
  },
  handler: (ctx, args): Promise<Id<"events">> =>
    appendEventOnce(ctx, {
      actor: { kind: "system", service: args.actorService },
      artifactIds: args.artifactIds,
      causationId: args.causationId,
      correlationId: args.correlationId,
      idempotencyKey: args.idempotencyKey,
      organizationId: args.organizationId,
      payload: args.payload,
      projectId: args.projectId,
      replyToEventId: args.replyToEventId,
      scopeKind: args.scopeKind,
      threadId: args.threadId,
      type: args.type,
      visibility: args.visibility,
      workId: args.workId,
    }),
});

export const appendToolEvent = internalMutation({
  args: {
    actorTool: v.string(),
    artifactIds: v.optional(v.array(v.id("artifacts"))),
    causationId: v.optional(v.id("events")),
    correlationId: v.string(),
    idempotencyKey: v.string(),
    organizationId: v.id("organizations"),
    payload: v.any(),
    projectId: v.optional(v.id("projects")),
    replyToEventId: v.optional(v.id("events")),
    scopeKind: v.union(v.literal("global"), v.literal("work")),
    threadId: v.optional(v.id("threads")),
    type: v.string(),
    visibility: v.union(
      v.literal("timeline"),
      v.literal("compact"),
      v.literal("internal")
    ),
    workId: v.optional(v.id("works")),
  },
  handler: (ctx, args): Promise<Id<"events">> =>
    appendEventOnce(ctx, {
      actor: { kind: "tool", tool: args.actorTool },
      artifactIds: args.artifactIds,
      causationId: args.causationId,
      correlationId: args.correlationId,
      idempotencyKey: args.idempotencyKey,
      organizationId: args.organizationId,
      payload: args.payload,
      projectId: args.projectId,
      replyToEventId: args.replyToEventId,
      scopeKind: args.scopeKind,
      threadId: args.threadId,
      type: args.type,
      visibility: args.visibility,
      workId: args.workId,
    }),
});

export { appendEventOnce };