102 lines
4.1 KiB
TypeScript
102 lines
4.1 KiB
TypeScript
// 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 })
|
|
},
|
|
})
|
|
},
|
|
})
|