yuban-server/src/http-app.mjs

import { createHash, randomUUID } from 'node:crypto';
import { AppError, publicError } from './errors.mjs';
import { publicJob } from './job-store.mjs';
import { sameOwner } from './auth.mjs';
import { validateRemoteAudioUrl } from './url-security.mjs';

function sendJson(response, status, body) {
  const payload = JSON.stringify(body);
  response.writeHead(status, {
    'Content-Type': 'application/json; charset=utf-8',
    'Content-Length': Buffer.byteLength(payload),
    'Cache-Control': 'no-store',
    'X-Content-Type-Options': 'nosniff',
  });
  response.end(payload);
}

async function readJson(request, maxBytes = 32 * 1024) {
  const chunks = [];
  let size = 0;
  for await (const chunk of request) {
    size += chunk.length;
    if (size > maxBytes) {
      throw new AppError('请求体过大', { code: 'REQUEST_TOO_LARGE', status: 413 });
    }
    chunks.push(chunk);
  }
  if (!chunks.length) return {};
  try {
    return JSON.parse(Buffer.concat(chunks).toString('utf8'));
  } catch {
    throw new AppError('请求体必须是 JSON', { code: 'INVALID_JSON', status: 400 });
  }
}

function normalizedRequest(body) {
  const durationMs = Number(body?.durationMs);
  const roleNum = Number(body?.roleNum);
  const roleType = Number(body?.roleType);
  return {
    audioUrl: String(body?.audioUrl || '').trim(),
    chatVoiceId: String(body?.chatVoiceId || '').trim().slice(0, 64) || null,
    durationMs: Number.isFinite(durationMs) && durationMs > 0 ? Math.round(durationMs) : null,
    roleType: [0, 1].includes(roleType) ? roleType : 1,
    roleNum: Number.isFinite(roleNum) ? Math.max(0, Math.min(10, Math.floor(roleNum))) : 0,
  };
}

function idempotencyKey(ownerKey, requestId) {
  const value = String(requestId || '').trim();
  if (!value) return null;
  return createHash('sha256').update(`${ownerKey}:${value.slice(0, 200)}`).digest('hex');
}

function routeJobId(pathname) {
  const match = /^\/recording-transcription\/jobs\/([0-9a-f-]+)(?:\/(cancel|retry))?$/.exec(
    pathname,
  );
  return match ? { id: match[1], action: match[2] || 'get' } : null;
}

function configureCors(request, response, config) {
  const origin = String(request.headers.origin || '').trim();
  if (!origin) return;
  if (!config.corsAllowedOrigins.includes(origin)) {
    throw new AppError('当前网页来源不允许访问此服务', {
      code: 'ORIGIN_NOT_ALLOWED',
      status: 403,
    });
  }
  response.setHeader('Access-Control-Allow-Origin', origin);
  response.setHeader('Vary', 'Origin');
  response.setHeader('Access-Control-Allow-Headers', 'Authorization, Content-Type, Idempotency-Key');
  response.setHeader('Access-Control-Allow-Methods', 'GET, POST, OPTIONS');
  response.setHeader('Access-Control-Max-Age', '600');
}

export function createHttpHandler({
  config,
  store,
  worker,
  authenticator,
  persistence = { enabled: false, async validateAccess() {} },
  validateAudioUrl = validateRemoteAudioUrl,
}) {
  const publicView = job =>
    publicJob(job, {
      queuePosition:
        typeof worker.queuePosition === 'function' ? worker.queuePosition(job.id) : null,
    });

  return async function handle(request, response) {
    const requestId = randomUUID();
    response.setHeader('X-Request-Id', requestId);
    try {
      configureCors(request, response, config);
      if (request.method === 'OPTIONS') {
        response.writeHead(204);
        response.end();
        return;
      }
      const url = new URL(request.url || '/', 'http://localhost');
      if (request.method === 'GET' && url.pathname === '/health') {
        sendJson(response, 200, {
          success: true,
          service: 'yuban-server',
          status: 'ok',
          version: config.serviceVersion || 'unknown',
          gitSha: config.gitSha || null,
          workerReady: true,
          storeReady: true,
          capabilities: ['recording-transcription'],
          time: new Date().toISOString(),
        });
        return;
      }

      const identity = await authenticator.authenticate(request);
      if (request.method === 'POST' && url.pathname === '/recording-transcription/jobs') {
        const body = await readJson(request);
        const input = normalizedRequest(body);
        await validateAudioUrl(input.audioUrl, config.allowedAudioHosts);
        await persistence.validateAccess(input.chatVoiceId, request.headers.authorization);
        const key = idempotencyKey(
          identity.ownerKey,
          request.headers['idempotency-key'] || body.requestId,
        );
        const existing = key ? store.findByIdempotency(key) : null;
        if (
          existing &&
          !['failed', 'cancelled'].includes(existing.status) &&
          sameOwner(existing.ownerKey, identity.ownerKey)
        ) {
          sendJson(response, 202, { success: true, reused: true, job: publicView(existing) });
          return;
        }

        const ownerJobs = store.listByOwner(identity.ownerKey);
        const activeJobs = ownerJobs.filter(job => ['queued', 'running'].includes(job.status));
        if (activeJobs.length >= config.maxActiveJobsPerUser) {
          throw new AppError('当前正在处理的长录音任务过多，请等待已有任务完成', {
            code: 'TOO_MANY_ACTIVE_JOBS',
            status: 429,
            retryable: true,
          });
        }
        if (ownerJobs.length >= config.maxRetainedJobsPerUser) {
          throw new AppError('临时任务数量已达上限，请稍后再试', {
            code: 'JOB_LIMIT_REACHED',
            status: 429,
            retryable: true,
          });
        }

        const now = new Date();
        const job = await store.create({
          id: randomUUID(),
          ownerKey: identity.ownerKey,
          idempotencyKey: key,
          status: 'queued',
          stage: 'queued',
          progress: 0,
          attempt: 0,
          runAttempt: 0,
          request: input,
          createdAt: now.toISOString(),
          updatedAt: now.toISOString(),
          expiresAt: new Date(now.getTime() + config.jobRetentionMs).toISOString(),
        });
        worker.enqueue(job.id);
        console.info('[yuban-server] transcription job created', {
          requestId,
          jobId: job.id,
          durationMs: input.durationMs,
        });
        sendJson(response, 202, { success: true, reused: false, job: publicView(job) });
        return;
      }

      const jobRoute = routeJobId(url.pathname);
      if (jobRoute && request.method === 'GET' && jobRoute.action === 'get') {
        const job = store.get(jobRoute.id);
        if (!job || !sameOwner(job.ownerKey, identity.ownerKey)) {
          throw new AppError('转写任务不存在', { code: 'JOB_NOT_FOUND', status: 404 });
        }
        sendJson(response, 200, { success: true, job: publicView(job) });
        return;
      }
      if (jobRoute && request.method === 'POST' && jobRoute.action === 'cancel') {
        const job = store.get(jobRoute.id);
        if (!job || !sameOwner(job.ownerKey, identity.ownerKey)) {
          throw new AppError('转写任务不存在', { code: 'JOB_NOT_FOUND', status: 404 });
        }
        const cancelled = await worker.cancel(job.id);
        sendJson(response, 200, { success: true, job: publicView(cancelled) });
        return;
      }
      if (jobRoute && request.method === 'POST' && jobRoute.action === 'retry') {
        const job = store.get(jobRoute.id);
        if (!job || !sameOwner(job.ownerKey, identity.ownerKey)) {
          throw new AppError('转写任务不存在', { code: 'JOB_NOT_FOUND', status: 404 });
        }
        const activeJobs = store
          .listByOwner(identity.ownerKey)
          .filter(item => ['queued', 'running', 'retry_wait'].includes(item.status));
        if (activeJobs.length >= config.maxActiveJobsPerUser) {
          throw new AppError('当前正在处理的长录音任务过多，请等待已有任务完成', {
            code: 'TOO_MANY_ACTIVE_JOBS',
            status: 429,
            retryable: true,
          });
        }
        const retried = await worker.retry(job.id);
        sendJson(response, 202, { success: true, reused: false, job: publicView(retried) });
        return;
      }
      throw new AppError('接口不存在', { code: 'NOT_FOUND', status: 404 });
    } catch (error) {
      const result = publicError(error);
      if (!(error instanceof AppError) || error.status >= 500) {
        console.error('[yuban-server] request failed', {
          requestId,
          method: request.method,
          path: request.url,
          code: error?.code || 'INTERNAL_ERROR',
          message: error?.message || 'unknown',
        });
      }
      if (!response.headersSent) sendJson(response, result.status, result.body);
      else response.end();
    }
  };
}