import os from 'node:os'; import { z } from 'zod'; import { buildQueueDefinitions, parseQueueSpec, type WorkerQueueDefinition } from './queues'; const envSchema = z.object({ NODE_ENV: z.enum(['development', 'test', 'production']).default('development'), LOG_LEVEL: z.enum(['fatal', 'error', 'warn', 'info', 'debug', 'trace']).default('info'), REDIS_URL: z .string() .min(1) .refine( (value) => value.startsWith('redis://') || value.startsWith('rediss://'), 'REDIS_URL 必须是 redis:// 连接串', ), DATABASE_URL: z .string() .min(1) .refine( (value) => value.startsWith('postgresql://') || value.startsWith('postgres://'), 'DATABASE_URL 必须是 postgresql:// 连接串', ), QUEUE: z.string().min(1).default('all'), WORKER_CONCURRENCY: z.coerce.number().int().min(1).max(64).optional(), WORKER_ID: z.string().min(1).max(128).optional(), }); export interface WorkerConfig { nodeEnv: 'development' | 'test' | 'production'; logLevel: 'fatal' | 'error' | 'warn' | 'info' | 'debug' | 'trace'; redisUrl: string; databaseUrl: string; queueSpec: string; queues: WorkerQueueDefinition[]; workerId: string; } function defaultWorkerId(): string { return `${os.hostname()}-${String(process.pid)}`; } /** * 加载并校验 worker 环境变量: * - QUEUE:`all` 或逗号分隔的队列组名(script/image/video/export/verify)/ 队列名(如 media:verify); * - WORKER_CONCURRENCY:覆盖全部队列并发;缺省取 TASK_QUEUES 中每个队列的定义值; * - WORKER_ID:默认 hostname + pid,用于心跳与日志定位。 */ export function loadWorkerConfig(source: NodeJS.ProcessEnv = process.env): WorkerConfig { const result = envSchema.safeParse(source); if (!result.success) { const details = result.error.issues .map((issue) => `${issue.path.join('.') || '(root)'}: ${issue.message}`) .join('; '); throw new Error(`Worker 环境变量校验失败:${details}`); } const env = result.data; let queues: WorkerQueueDefinition[]; try { queues = buildQueueDefinitions(parseQueueSpec(env.QUEUE), env.WORKER_CONCURRENCY); } catch (error) { throw new Error( `QUEUE 配置非法(当前值 ${env.QUEUE}):${error instanceof Error ? error.message : String(error)}`, ); } return { nodeEnv: env.NODE_ENV, logLevel: env.LOG_LEVEL, redisUrl: env.REDIS_URL, databaseUrl: env.DATABASE_URL, queueSpec: env.QUEUE, queues, workerId: env.WORKER_ID ?? defaultWorkerId(), }; }