yuban-server/test/transcription-worker.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtemp, rm } from 'node:fs/promises';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { AppError } from '../src/errors.mjs';
import { JobStore } from '../src/job-store.mjs';
import { TranscriptionWorker } from '../src/transcription-worker.mjs';

async function waitFor(predicate, timeoutMs = 2_000) {
  const startedAt = Date.now();
  while (Date.now() - startedAt < timeoutMs) {
    const value = predicate();
    if (value) return value;
    await new Promise(resolve => setTimeout(resolve, 5));
  }
  throw new Error('等待任务状态超时');
}

async function fixture(t, provider) {
  const directory = await mkdtemp(join(tmpdir(), 'yuban-worker-test-'));
  t.after(() => rm(directory, { recursive: true, force: true }));
  const store = new JobStore(directory, 86_400_000);
  await store.init();
  const config = {
    maxJobConcurrency: 1,
    maxJobAttempts: 3,
    retryBaseDelayMs: 5,
    jobRetentionMs: 86_400_000,
    jobTempDir: directory,
    iflytek: { pollTimeoutMs: 1_000 },
  };
  const media = {
    async downloadRemoteAudio() {},
    async transcodeToPcmWav() {},
    async probeDurationMs() { return 1_800_000; },
  };
  const worker = new TranscriptionWorker({
    config,
    store,
    provider,
    media,
    sleepFn: async () => {},
  });
  return { store, worker };
}

async function createJob(store, overrides = {}) {
  const now = new Date();
  return store.create({
    id: overrides.id || '00000000-0000-4000-8000-000000000001',
    ownerKey: 'owner',
    idempotencyKey: 'stable-key',
    status: 'queued',
    stage: 'queued',
    progress: 0,
    attempt: 0,
    runAttempt: 0,
    request: {
      audioUrl: 'https://file.yuban.co/recording.webm',
      durationMs: 1_800_000,
      roleType: 1,
      roleNum: 0,
    },
    createdAt: now.toISOString(),
    updatedAt: now.toISOString(),
    expiresAt: new Date(now.getTime() + 86_400_000).toISOString(),
    ...overrides,
  });
}

test('可重试错误进入 retry_wait，并由 Worker 自动完成下一次 attempt', async t => {
  let submitCount = 0;
  const provider = {
    async submit() {
      submitCount += 1;
      if (submitCount === 1) {
        throw new AppError('讯飞暂时不可用', {
          code: 'PROVIDER_TEMPORARY',
          status: 502,
          retryable: true,
        });
      }
      return { orderId: 'provider-order-1', estimateTimeMs: 1 };
    },
    async getResult() {
      return {
        status: 'completed',
        result: {
          text: '中断前后的完整逐字稿',
          segments: [{ text: '中断前后的完整逐字稿', startMs: 0, endMs: 1000, speakerId: '1' }],
        },
      };
    },
  };
  const { store, worker } = await fixture(t, provider);
  const job = await createJob(store);
  worker.enqueue(job.id);

  const completed = await waitFor(() => {
    const current = store.get(job.id);
    return current?.status === 'completed' ? current : null;
  });
  assert.equal(submitCount, 2);
  assert.equal(completed.attempt, 2);
  assert.equal(completed.result.text, '中断前后的完整逐字稿');
  assert.equal(completed.error, undefined);
});

test('人工重试失败任务会创建新的 jobId 并保留父任务', async t => {
  const provider = {
    async submit() { return { orderId: 'unused', estimateTimeMs: 1 }; },
    async getResult() { return { status: 'pending' }; },
  };
  const { store, worker } = await fixture(t, provider);
  const failed = await createJob(store, {
    status: 'failed',
    stage: 'failed',
    attempt: 3,
    runAttempt: 3,
    error: { code: 'NETWORK', message: '网络错误', retryable: true },
  });
  const retried = await worker.retry(failed.id);
  assert.notEqual(retried.id, failed.id);
  assert.equal(retried.parentJobId, failed.id);
  assert.equal(retried.status, 'queued');
  assert.equal(retried.attempt, 3);
  assert.equal(store.findByIdempotency('stable-key').id, retried.id);
  await worker.cancel(retried.id);
});