evOo-local/packages/adapter-kit/src/background-jobs.ts

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<BackgroundJobPayloads[Name]> };

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<void> {
  const job = parseBackgroundJob(name, payload);
  const handler = handlers[job.name] as (payload: typeof job.payload) => Promise<void>;
  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 } : {}),
  };
}