webdev-pipeline/convex/runs.ts

import { v } from "convex/values";

import { normalizeListLimit } from "./domain";
import { mutation, query } from "./_generated/server";

const runType = v.union(
  v.literal("campaign"),
  v.literal("lead_discovery"),
  v.literal("audit"),
  v.literal("outreach"),
  v.literal("lifecycle"),
);
const runStatus = v.union(
  v.literal("pending"),
  v.literal("running"),
  v.literal("succeeded"),
  v.literal("failed"),
  v.literal("canceled"),
);
const eventLevel = v.union(
  v.literal("info"),
  v.literal("warning"),
  v.literal("error"),
);

export const create = mutation({
  args: {
    type: runType,
    campaignId: v.optional(v.id("campaigns")),
    leadId: v.optional(v.id("leads")),
    auditId: v.optional(v.id("audits")),
    status: v.optional(runStatus),
    currentStep: v.optional(v.string()),
  },
  handler: async (ctx, args) => {
    const now = Date.now();

    return await ctx.db.insert("agentRuns", {
      ...args,
      status: args.status ?? "pending",
      counters: {
        leadsFound: 0,
        leadsCreated: 0,
        auditsCreated: 0,
        outreachPrepared: 0,
        errors: 0,
      },
      createdAt: now,
      updatedAt: now,
    });
  },
});

export const updateStatus = mutation({
  args: {
    id: v.id("agentRuns"),
    status: runStatus,
    currentStep: v.optional(v.string()),
    errorSummary: v.optional(v.string()),
  },
  handler: async (ctx, args) => {
    const now = Date.now();
    const patch: {
      status: typeof args.status;
      updatedAt: number;
      currentStep?: string;
      errorSummary?: string;
      startedAt?: number;
      finishedAt?: number;
    } = {
      status: args.status,
      updatedAt: now,
    };

    if (args.currentStep !== undefined) {
      patch.currentStep = args.currentStep;
    }
    if (args.errorSummary !== undefined) {
      patch.errorSummary = args.errorSummary;
    }
    if (args.status === "running") {
      patch.startedAt = now;
    }
    if (
      args.status === "succeeded" ||
      args.status === "failed" ||
      args.status === "canceled"
    ) {
      patch.finishedAt = now;
    }

    await ctx.db.patch(args.id, patch);
    return args.id;
  },
});

export const list = query({
  args: {
    status: v.optional(runStatus),
    type: v.optional(runType),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const limit = normalizeListLimit(args.limit);

    if (args.type && args.status) {
      const type = args.type;
      const status = args.status;

      return await ctx.db
        .query("agentRuns")
        .withIndex("by_type_and_status", (q) =>
          q.eq("type", type).eq("status", status),
        )
        .order("desc")
        .take(limit);
    }

    if (args.status) {
      const status = args.status;

      return await ctx.db
        .query("agentRuns")
        .withIndex("by_status", (q) => q.eq("status", status))
        .order("desc")
        .take(limit);
    }

    return await ctx.db.query("agentRuns").order("desc").take(limit);
  },
});

export const appendEvent = mutation({
  args: {
    runId: v.id("agentRuns"),
    level: eventLevel,
    message: v.string(),
    details: v.optional(
      v.array(
        v.object({
          label: v.string(),
          value: v.string(),
          source: v.optional(v.string()),
        }),
      ),
    ),
  },
  handler: async (ctx, args) => {
    return await ctx.db.insert("agentRunEvents", {
      ...args,
      createdAt: Date.now(),
    });
  },
});

export const listEvents = query({
  args: {
    runId: v.id("agentRuns"),
    limit: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const limit = normalizeListLimit(args.limit);

    return await ctx.db
      .query("agentRunEvents")
      .withIndex("by_runId_and_createdAt", (q) => q.eq("runId", args.runId))
      .order("desc")
      .take(limit);
  },
});