aigen/server/utils/jobs.ts

export type JobStatus = 'queued' | 'uploading' | 'running' | 'complete' | 'error' | 'cancelled' | 'deferred'

export interface JobEvent {
  type: string
  message?: string
  progress?: number
  step?: number
  maxStep?: number
  node?: string | null
  filename?: string
  subfolder?: string
  mediaType?: string
  clipId?: string
  hideThumbnail?: boolean
  error?: string
  elapsedMs?: number
  busy?: boolean
  queueRunning?: number
  queuePending?: number
  folderLocked?: boolean
  jobId?: string
  draftId?: string
}

export interface Job {
  id: string
  promptId?: string
  clientId: string
  status: JobStatus
  message: string
  progress: number
  step: number
  maxStep: number
  startedAt: number
  video?: { filename: string; subfolder: string; type: string }
  clipId?: string
  hideThumbnail?: boolean
  library?: {
    ownerKey: string
    folderId: string
    hideThumbnail: boolean
    hideInput?: boolean
    folderLocked?: boolean
    name?: string
    prompt: string
    aspect: string
    width: number
    height: number
    steps: number
    turbo: boolean
    seed: number
    cfg?: number
    fps?: number
    samplerName?: string
    scheduler?: string
    thumb?: Buffer
    imageName?: string
    imageSubfolder?: string
    stillId?: string
    stillFilename?: string
    duration?: number
    sound?: boolean
    draftId?: string
    extendTmpDir?: string
    extendPart1Path?: string
    extendSourceClipId?: string
  }
  error?: string
  socketReady?: boolean
  events: JobEvent[]
  listeners: Set<(event: JobEvent) => void>
}

const jobs = new Map<string, Job>()
const MAX_JOBS = 40

export function createJob(): Job {
  const job: Job = {
    id: crypto.randomUUID(),
    clientId: crypto.randomUUID(),
    status: 'queued',
    message: 'Queued',
    progress: 0,
    step: 0,
    maxStep: 0,
    startedAt: Date.now(),
    events: [],
    listeners: new Set()
  }
  jobs.set(job.id, job)
  while (jobs.size > MAX_JOBS) {
    const oldest = jobs.keys().next().value
    if (oldest) jobs.delete(oldest)
  }
  return job
}

export function getJob(id: string) {
  return jobs.get(id)
}

export function listJobs() {
  return [...jobs.values()]
}

export function jobSnapshot(job: Job) {
  return {
    jobId: job.id,
    type: 'snapshot' as const,
    status: job.status,
    message: job.message,
    progress: job.progress,
    step: job.step,
    maxStep: job.maxStep,
    promptId: job.promptId,
    elapsedMs: Date.now() - job.startedAt,
    filename: job.library?.folderLocked ? undefined : job.video?.filename,
    subfolder: job.library?.folderLocked ? undefined : job.video?.subfolder,
    mediaType: job.library?.folderLocked ? undefined : job.video?.type,
    clipId: job.clipId,
    hideThumbnail: job.hideThumbnail,
    error: job.error,
    folderLocked: job.library?.folderLocked,
    draftId: job.library?.draftId
  }
}

export function emitJob(job: Job, event: JobEvent) {
  event.jobId = job.id
  if (event.message) job.message = event.message
  if (typeof event.progress === 'number') job.progress = event.progress
  if (typeof event.step === 'number') job.step = event.step
  if (typeof event.maxStep === 'number') job.maxStep = event.maxStep
  event.elapsedMs = Date.now() - job.startedAt
  job.events.push(event)
  if (job.events.length > 200) job.events.shift()
  for (const listener of job.listeners) {
    try {
      listener(event)
    } catch {
      // ignore broken SSE listeners
    }
  }
}

export function subscribeJob(job: Job, listener: (event: JobEvent) => void) {
  job.listeners.add(listener)
  return () => job.listeners.delete(listener)
}