mnote/wolai-frontend/convex/jobs.ts

import { internalAction, internalMutation, internalQuery, mutation, query } from "./_generated/server";
import { v } from "convex/values";
import { nowIso } from "./_utils/time";
import { internal } from "./_generated/api";

export const get = query({
  args: { userId: v.string(), id: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return null;
    if (job.user_id !== args.userId) return null;
    return {
      id: job.id,
      type: job.type,
      status: job.status,
      payload: job.payload,
      result: job.result,
      error: job.error,
      created_at: job.created_at,
      updated_at: job.updated_at,
      started_at: job.started_at,
      finished_at: job.finished_at,
    };
  },
});

export const enqueueDemo = mutation({
  args: { userId: v.string(), id: v.string(), ms: v.optional(v.number()) },
  handler: async (ctx, args) => {
    const ts = nowIso();
    const payload = { ms: args.ms ?? 800 };
    await ctx.db.insert("jobs", {
      id: args.id,
      user_id: args.userId,
      type: "demo.sleep",
      status: "queued",
      payload,
      result: null,
      error: null,
      created_at: ts,
      updated_at: ts,
      started_at: null,
      finished_at: null,
    });

    // 说明：阶段 5 骨架——用 scheduler 触发内部 mutation，再由内部 action 执行耗时逻辑。
    await ctx.scheduler.runAfter(0, internal.jobs.start, { id: args.id });
    return { ok: true, id: args.id };
  },
});

export const start = internalMutation({
  args: { id: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return;
    if (job.status !== "queued") return;
    const ts = nowIso();
    await ctx.db.patch(job._id, { status: "running", started_at: ts, updated_at: ts });
    await ctx.scheduler.runAfter(0, internal.jobs.run, { id: args.id });
  },
});

export const run = internalAction({
  args: { id: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.runQuery(internal.jobs._getInternal, { id: args.id });
    if (!job) return;
    if (job.status !== "running") return;

    try {
      if (job.type === "demo.sleep") {
        const ms = typeof job.payload?.ms === "number" ? job.payload.ms : 800;
        await new Promise((r) => setTimeout(r, ms));
        await ctx.runMutation(internal.jobs.finishSuccess, {
          id: args.id,
          result: { ok: true, sleptMs: ms },
        });
        return;
      }

      throw new Error(`未知任务类型：${job.type}`);
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      await ctx.runMutation(internal.jobs.finishFailure, { id: args.id, error: message });
    }
  },
});

export const _getInternal = internalQuery({
  args: { id: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return null;
    return {
      id: job.id,
      type: job.type,
      status: job.status,
      payload: job.payload,
    };
  },
});

export const finishSuccess = internalMutation({
  args: { id: v.string(), result: v.any() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return;
    const ts = nowIso();
    await ctx.db.patch(job._id, {
      status: "succeeded",
      result: args.result,
      error: null,
      finished_at: ts,
      updated_at: ts,
    });
  },
});

export const finishFailure = internalMutation({
  args: { id: v.string(), error: v.string() },
  handler: async (ctx, args) => {
    const job = await ctx.db
      .query("jobs")
      .withIndex("by_job_id", (q) => q.eq("id", args.id))
      .first();
    if (!job) return;
    const ts = nowIso();
    await ctx.db.patch(job._id, {
      status: "failed",
      result: null,
      error: args.error,
      finished_at: ts,
      updated_at: ts,
    });
  },
});