// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0
// The jobs dispatcher (spec §5.4: jobs are the Convex scheduler; the workpool component is this
// same dispatch under parallelism). A module's enqueue is a row plus a scheduled run of this
// mutation, inside the enqueueing transaction; this function re-reads the row, validates the
// payload against the module's declared schema, runs the handler with a JobContext, and retries
// or dead-letters.

import { Container, type JobContext } from "@conusai-kit/core"
import { DATA_SOURCE, JOB_QUEUE, SYSTEM_DATA } from "@conusai-kit/data"
import {
  createConvexDataSource,
  createConvexJobQueue,
  dispatchConvexJob,
  reclaimConvexActionJobs,
  settleConvexActionJob,
} from "@conusai-kit/data-convex"
import { executorFor } from "@conusai-kit/data-convex/runtime"
import { makeFunctionReference } from "convex/server"
import { v } from "convex/values"
import { internalMutation } from "./_generated/server"
import { convexEntities, registry } from "./lib/runtime"

export const run = internalMutation({
  args: { jobId: v.string(), inline: v.optional(v.any()) },
  handler: async (ctx, args) => {
    const data = createConvexDataSource({ entities: convexEntities, executor: executorFor(ctx) })
    const dispatch = makeFunctionReference<"mutation">("conusaiJobs:run")
    await dispatchConvexJob(
      ctx.db,
      { jobId: args.jobId, ...(args.inline ? { inline: args.inline as never } : {}) },
      {
        jobs: Object.fromEntries(Object.entries(registry.jobs).map(([name, { def }]) => [name, def])),
        schedule: async ({ delayMs, jobId, inline }) => {
          await ctx.scheduler.runAfter(delayMs, dispatch, { jobId, ...(inline ? { inline } : {}) })
        },
        run: async (job, payload, call) => {
          if (job === "documents.ocr") {
            if (call.scope.level !== "project") throw new Error("documents.ocr requires a project job scope")
            await ctx.scheduler.runAfter(0, makeFunctionReference<"action">("documentsActions:process"), {
              jobId: call.jobId,
              attempt: call.attempt,
              orgId: call.scope.orgId,
              workspaceId: call.scope.workspaceId,
              projectId: call.scope.projectId,
              payload,
            })
            return { deferred: true }
          }
          const def = registry.jobs[job]?.def
          if (!def) throw new Error(`no job named ${job} is registered`)
          const container = new Container()
          container.value(DATA_SOURCE, data)
          container.value(SYSTEM_DATA, data)
          container.value(
            JOB_QUEUE,
            createConvexJobQueue({
              db: ctx.db,
              schedule: ({ delayMs, jobId, inline }) =>
                ctx.scheduler
                  .runAfter(delayMs, dispatch, { jobId, ...(inline ? { inline } : {}) })
                  .then(() => undefined),
            }),
          )
          const context: JobContext = { container, correlationId: call.correlationId, attempt: call.attempt }
          await def.handler(payload, context)
        },
      },
    )
  },
})

/** An OCR action records its acknowledgement or retry back through the durable queue row. */
export const settleAction = internalMutation({
  args: {
    jobId: v.string(),
    attempt: v.number(),
    error: v.optional(v.string()),
    retryAfterSeconds: v.optional(v.number()),
  },
  handler: async (ctx, args) => {
    const dispatch = makeFunctionReference<"mutation">("conusaiJobs:run")
    await settleConvexActionJob(ctx.db as never, args, {
      schedule: async ({ delayMs, jobId }) => {
        await ctx.scheduler.runAfter(delayMs, dispatch, { jobId })
      },
    })
  },
})

/** Requeue an action job whose lease expired after the Convex action process died. */
export const reclaimActionJobs = internalMutation({
  args: {},
  handler: async (ctx) => {
    const dispatch = makeFunctionReference<"mutation">("conusaiJobs:run")
    await reclaimConvexActionJobs(ctx.db as never, {
      schedule: async ({ delayMs, jobId }) => {
        await ctx.scheduler.runAfter(delayMs, dispatch, { jobId })
      },
    })
  },
})