aigen/server/utils/jobs.ts

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

export interface JobEvent {
  type: string
  message?: string
  progress?: number
  step?: number
  maxStep?: number
  node?: string | null
  filename?: string
  subfolder?: string
  mediaType?: string
  error?: string
  elapsedMs?: number
}

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 }
  error?: string
  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 emitJob(job: Job, event: JobEvent) {
  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)
}