import { z } from "zod"; import type { BackgroundJob, BackgroundJobHandlers, BackgroundJobName, BackgroundJobPayloads, } from "./types.js"; const payloadSchemas = { "run.continue": z.object({ runId: z.string().min(1) }), "routine.wakeup": z.object({ routineId: z.string().min(1), scheduledFor: z.string().datetime({ offset: true }), }), "computer.update": z.object({ updateId: z.string().min(1) }), "computer.sleep": z.object({ computerId: z.string().min(1) }), "computer.control-expire": z.object({ computerId: z.string().min(1), leaseId: z.string().min(1), }), "skill.teaching-expire": z.object({ skillId: z.string().min(1) }), "history.compact": z.object({ threadId: z.string().min(1) }), "messaging.deliver": z.object({ runId: z.string().min(1).optional() }), "cloud_agent.poll": z.object({ agentId: z.string().min(1) }), "profile.consolidate": z.object({ userId: z.string().min(1) }), "media.run": z.object({ jobId: z.string().min(1) }), "media.library.sync": z.object({ sourceId: z.string().min(1) }), } satisfies { [Name in BackgroundJobName]: z.ZodType }; export function parseBackgroundJob(name: string, payload: unknown): BackgroundJob { if (!(name in payloadSchemas)) throw new Error(`Unknown background job: ${name}`); const typedName = name as BackgroundJobName; const parsed = payloadSchemas[typedName].parse(payload); return { name: typedName, payload: parsed } as BackgroundJob; } export async function dispatchBackgroundJob( handlers: BackgroundJobHandlers, name: string, payload: unknown, ): Promise { const job = parseBackgroundJob(name, payload); const handler = handlers[job.name] as (payload: typeof job.payload) => Promise; await handler(job.payload); } export function runJobKey(runId: string): string { return `run:${runId}`; } export function routineJobKey(routineId: string): string { return `routine:${routineId}`; } export function computerSleepJobKey(computerId: string): string { return `computer.sleep:${computerId}`; } export function computerControlExpireJobKey(computerId: string, leaseId?: string): string { return leaseId ? `computer.control-expire:${computerId}:${leaseId}` : `computer.control-expire:${computerId}`; } export function skillTeachingExpireJobKey(skillId: string): string { return `skill.teaching-expire:${skillId}`; } export function runContinueJob(runId: string): BackgroundJob { return { name: "run.continue", payload: { runId }, replaceKey: runJobKey(runId), }; } export function routineWakeupJob(routineId: string, scheduledFor: Date): BackgroundJob { return { name: "routine.wakeup", payload: { routineId, scheduledFor: scheduledFor.toISOString() }, availableAt: scheduledFor, replaceKey: routineJobKey(routineId), }; } export function computerSleepJob(computerId: string, availableAt: Date): BackgroundJob { return { name: "computer.sleep", payload: { computerId }, availableAt, replaceKey: computerSleepJobKey(computerId), }; } export function computerControlExpireJob( computerId: string, leaseId: string, availableAt: Date, ): BackgroundJob { return { name: "computer.control-expire", payload: { computerId, leaseId }, availableAt, replaceKey: computerControlExpireJobKey(computerId, leaseId), }; } export function skillTeachingExpireJob(skillId: string, availableAt: Date): BackgroundJob { return { name: "skill.teaching-expire", payload: { skillId }, availableAt, replaceKey: skillTeachingExpireJobKey(skillId), }; } export function historyCompactJobKey(threadId: string): string { return `history.compact:${threadId}`; } export function messagingDeliverJob(runId?: string, availableAt?: Date): BackgroundJob { return { name: "messaging.deliver", payload: runId ? { runId } : {}, replaceKey: `messaging.deliver:${runId ?? "drain"}`, ...(availableAt ? { availableAt } : {}), }; } export function historyCompactJob(threadId: string): BackgroundJob { return { name: "history.compact", payload: { threadId }, replaceKey: historyCompactJobKey(threadId), }; } export function cloudAgentPollJobKey(agentId: string): string { return `cloud_agent.poll:${agentId}`; } export function cloudAgentPollJob( payload: BackgroundJobPayloads["cloud_agent.poll"], availableAt?: Date, ): BackgroundJob { return { name: "cloud_agent.poll", payload, replaceKey: cloudAgentPollJobKey(payload.agentId), ...(availableAt ? { availableAt } : {}), }; } export function profileConsolidateJobKey(userId: string): string { return `profile.consolidate:${userId}`; } export function profileConsolidateJob(userId: string): BackgroundJob { return { name: "profile.consolidate", payload: { userId }, replaceKey: profileConsolidateJobKey(userId), }; } /** Media jobs share one queue so a single GPU never runs two generations at once. */ export const MEDIA_JOB_QUEUE = "media"; export function mediaRunJobKey(jobId: string): string { return `media.run:${jobId}`; } export function mediaRunJob(jobId: string, availableAt?: Date): BackgroundJob { return { name: "media.run", payload: { jobId }, replaceKey: mediaRunJobKey(jobId), queue: MEDIA_JOB_QUEUE, ...(availableAt ? { availableAt } : {}), }; } /** Library syncs walk disks; one at a time keeps them from competing for the same drive. */ export const MEDIA_LIBRARY_JOB_QUEUE = "media-library"; export function mediaLibrarySyncJobKey(sourceId: string): string { return `media.library.sync:${sourceId}`; } export function mediaLibrarySyncJob(sourceId: string, availableAt?: Date): BackgroundJob { return { name: "media.library.sync", payload: { sourceId }, replaceKey: mediaLibrarySyncJobKey(sourceId), queue: MEDIA_LIBRARY_JOB_QUEUE, ...(availableAt ? { availableAt } : {}), }; }