nexmoe--vidbee/packages/task-queue/test/orchestrator.test.ts

/**
 * Integration tests for TaskQueueAPI using a fake executor that scripts
 * spawn/progress/finish events deterministically.
 */
import { describe, expect, it } from 'vitest'

import { TaskQueueAPI } from '../src/api'
import type {
  Executor,
  ExecutorEvents,
  ExecutorRun
} from '../src/executor'
import { MemoryPersistAdapter } from '../src/persist'
import type { TaskQueueEvent } from '../src/events'
import type { ClassifiedError } from '../src/types'

interface ScriptStep {
  type:
    | 'spawn'
    | 'progress'
    | 'success'
    | 'errorRetryable'
    | 'errorFatal'
    | 'cancelled'
}

function makeFakeExecutor(scriptByUrl: Map<string, ScriptStep[]>): Executor {
  let pidCounter = 1000
  return {
    run(ctx, events: ExecutorEvents): ExecutorRun {
      const script = scriptByUrl.get(ctx.input.url) ?? [
        { type: 'spawn' },
        { type: 'success' }
      ]
      const pid = ++pidCounter
      let cancelled = false
      // Emit synchronously so tests don't have to await microtasks.
      queueMicrotask(() => {
        for (const step of script) {
          if (cancelled) {
            events.onFinish({
              taskId: ctx.taskId,
              attemptId: ctx.attemptId,
              result: { type: 'cancelled' },
              closedAt: 1,
              stdoutTail: '',
              stderrTail: ''
            })
            return
          }
          switch (step.type) {
            case 'spawn':
              events.onSpawn({
                taskId: ctx.taskId,
                attemptId: ctx.attemptId,
                pid,
                pidStartedAt: 1,
                kind: 'yt-dlp',
                spawnedAt: 1
              })
              break
            case 'progress':
              events.onProgress({
                taskId: ctx.taskId,
                attemptId: ctx.attemptId,
                progress: {
                  percent: 0.5,
                  bytesDownloaded: 10,
                  bytesTotal: 20,
                  speedBps: 100,
                  etaMs: 100,
                  ticks: 1
                },
                enteredProcessing: false
              })
              break
            case 'success':
              events.onFinish({
                taskId: ctx.taskId,
                attemptId: ctx.attemptId,
                result: {
                  type: 'success',
                  output: {
                    filePath: '/fake/file.mp4',
                    size: 1234,
                    durationMs: null,
                    sha256: null
                  }
                },
                closedAt: 1,
                stdoutTail: '',
                stderrTail: ''
              })
              break
            case 'errorRetryable': {
              const err: ClassifiedError = {
                category: 'network-transient',
                exitCode: 1,
                rawMessage: 'ECONNRESET',
                uiMessageKey: 'task.error.networkTransient',
                uiActionHints: ['retry'],
                retryable: true,
                suggestedRetryAfterMs: null
              }
              events.onFinish({
                taskId: ctx.taskId,
                attemptId: ctx.attemptId,
                result: { type: 'error', error: err, exitCode: 1 },
                closedAt: 1,
                stdoutTail: '',
                stderrTail: 'ECONNRESET on socket'
              })
              break
            }
            case 'errorFatal': {
              const err: ClassifiedError = {
                category: 'auth-required',
                exitCode: 1,
                rawMessage: 'Sign in to confirm',
                uiMessageKey: 'task.error.authRequired',
                uiActionHints: ['import-cookies'],
                retryable: false,
                suggestedRetryAfterMs: null
              }
              events.onFinish({
                taskId: ctx.taskId,
                attemptId: ctx.attemptId,
                result: { type: 'error', error: err, exitCode: 1 },
                closedAt: 1,
                stdoutTail: '',
                stderrTail: 'Sign in to confirm'
              })
              break
            }
            case 'cancelled':
              events.onFinish({
                taskId: ctx.taskId,
                attemptId: ctx.attemptId,
                result: { type: 'cancelled' },
                closedAt: 1,
                stdoutTail: '',
                stderrTail: ''
              })
              break
          }
        }
      })
      return {
        cancel: async () => {
          cancelled = true
        },
        pause: async () => {
          cancelled = true
        }
      }
    }
  }
}

async function flush(ms = 50): Promise<void> {
  await new Promise((r) => setTimeout(r, ms))
}

describe('TaskQueueAPI orchestrator', () => {
  it('happy path: queued → running → completed', async () => {
    const api = new TaskQueueAPI({
      persist: new MemoryPersistAdapter(),
      executor: makeFakeExecutor(new Map()),
      filePresent: () => true,
      maxConcurrency: 2
    })
    await api.start()
    const events: TaskQueueEvent[] = []
    api.subscribe((e) => events.push(e))
    const { id } = await api.add({
      input: { url: 'https://example.com/v', kind: 'video' }
    })
    await flush()
    const t = api.get(id)!
    expect(t.status).toBe('completed')
    expect(t.output?.size).toBe(1234)
    const transitions = events
      .filter((e) => e.type === 'transition')
      .map((e) => (e as { to: string }).to)
    expect(transitions).toEqual(['queued', 'running', 'completed'])
  })

  it('fatal error → failed (no retry)', async () => {
    const api = new TaskQueueAPI({
      persist: new MemoryPersistAdapter(),
      executor: makeFakeExecutor(
        new Map([
          [
            'https://e.com/x',
            [{ type: 'spawn' }, { type: 'errorFatal' }]
          ]
        ])
      ),
      filePresent: () => true
    })
    await api.start()
    const { id } = await api.add({
      input: { url: 'https://e.com/x', kind: 'video' }
    })
    await flush()
    const t = api.get(id)!
    expect(t.status).toBe('failed')
    expect(t.lastError?.category).toBe('auth-required')
  })

  it('retryable error → retry-scheduled then queued on tick', async () => {
    let now = 1_000_000
    const ticks: number[] = []
    const api = new TaskQueueAPI({
      persist: new MemoryPersistAdapter(),
      executor: makeFakeExecutor(
        new Map([
          [
            'https://e.com/x',
            // First attempt: fail. Subsequent attempts: succeed.
            [{ type: 'spawn' }, { type: 'errorRetryable' }]
          ]
        ])
      ),
      filePresent: () => true,
      clock: () => now,
      setTimer: (fn, ms) => {
        ticks.push(ms)
        // Force timer to fire on next macrotask using real setTimeout(0).
        return setTimeout(fn, 0)
      },
      clearTimer: (h) => clearTimeout(h as ReturnType<typeof setTimeout>),
      rng: () => 0 // deterministic backoff = 0
    })
    await api.start()
    const { id } = await api.add({
      input: { url: 'https://e.com/x', kind: 'video' }
    })
    await flush()
    const t = api.get(id)!
    // After first attempt fails the orchestrator goes retry-scheduled then
    // due to rng=0 the backoff is 0, so the tick fires almost immediately.
    // The fake executor still has only one script (errorRetryable), so the
    // second attempt also fails → another retry. Eventually the task hits
    // the same script. We verify *either* retry-scheduled or further along
    // the retry path, but at minimum that the task is not stuck queued and
    // that the retry timer fired at least once.
    expect(['retry-scheduled', 'queued', 'running', 'failed']).toContain(t.status)
    expect(ticks.length).toBeGreaterThan(0)
  })

  it('cancel on running issues SIGTERM and lands in cancelled', async () => {
    let resolveSpawn: (() => void) | null = null
    const spawnSeen = new Promise<void>((r) => {
      resolveSpawn = r
    })
    const executor: Executor = {
      run(ctx, events) {
        let cancelled = false
        queueMicrotask(() => {
          events.onSpawn({
            taskId: ctx.taskId,
            attemptId: ctx.attemptId,
            pid: 1234,
            pidStartedAt: 1,
            kind: 'yt-dlp',
            spawnedAt: 1
          })
          resolveSpawn?.()
        })
        return {
          cancel: async () => {
            cancelled = true
            queueMicrotask(() =>
              events.onFinish({
                taskId: ctx.taskId,
                attemptId: ctx.attemptId,
                result: { type: 'cancelled' },
                closedAt: 1,
                stdoutTail: '',
                stderrTail: ''
              })
            )
          },
          pause: async () => {
            void cancelled
          }
        }
      }
    }
    const api = new TaskQueueAPI({
      persist: new MemoryPersistAdapter(),
      executor,
      filePresent: () => true
    })
    await api.start()
    const { id } = await api.add({
      input: { url: 'https://e.com/c', kind: 'video' }
    })
    await spawnSeen
    await flush(0)
    await api.cancel(id, 'user')
    await flush()
    expect(api.get(id)!.status).toBe('cancelled')
    expect(api.stats().running).toBe(0)
  })

  it('processing → completed guard fails when file missing/empty', async () => {
    const api = new TaskQueueAPI({
      persist: new MemoryPersistAdapter(),
      executor: makeFakeExecutor(new Map()),
      filePresent: () => false // pretend the file disappeared
    })
    await api.start()
    const { id } = await api.add({
      input: { url: 'https://e.com/v', kind: 'video' }
    })
    await flush()
    const t = api.get(id)!
    expect(t.status).toBe('failed')
    expect(t.lastError?.category).toBe('output-missing')
  })

  it('crash recovery: paused/recovered tasks survive restart', async () => {
    const persist = new MemoryPersistAdapter()
    let api = new TaskQueueAPI({
      persist,
      executor: makeFakeExecutor(new Map()),
      filePresent: () => true
    })
    await api.start()
    const { id } = await api.add({
      input: { url: 'https://e.com/r', kind: 'video' }
    })
    await flush()
    expect(api.get(id)!.status).toBe('completed')

    // Spin up a brand-new orchestrator on the same persist adapter.
    api = new TaskQueueAPI({
      persist,
      executor: makeFakeExecutor(new Map()),
      filePresent: () => true
    })
    await api.start()
    expect(api.get(id)!.status).toBe('completed')
  })

  // Regression: NEX-124 review screenshot showed two list rows for one
  // pasted URL — the renderer's optimistic placeholder (renderer-id) and
  // the kernel-generated row (random UUID) coexisted because add() ignored
  // the caller-supplied id. Honoring `req.id` makes them merge into one.
  it('honors caller-supplied id and is idempotent on re-add', async () => {
    const api = new TaskQueueAPI({
      persist: new MemoryPersistAdapter(),
      executor: makeFakeExecutor(new Map()),
      filePresent: () => true
    })
    await api.start()
    const callerId = 'renderer-optimistic-id-123'
    const first = await api.add({
      id: callerId,
      input: { url: 'https://example.com/a', kind: 'video' }
    })
    expect(first.id).toBe(callerId)
    expect(api.get(callerId)).toBeDefined()

    // Second add with the same id is a no-op (idempotent), not a duplicate.
    const second = await api.add({
      id: callerId,
      input: { url: 'https://example.com/a', kind: 'video' }
    })
    expect(second.id).toBe(callerId)
    const all = api.list({ limit: 100, cursor: null }).tasks
    expect(all.filter((t) => t.id === callerId)).toHaveLength(1)
  })
})