import Redis from "ioredis"; import type { RedisOptions } from "ioredis"; import { execSync } from "node:child_process"; let redis: Redis | null = null; const REDIS_RETRY_DEFAULT_ATTEMPTS = 12; const REDIS_RETRY_DEFAULT_DELAY_MS = 600; const DEFAULT_WSL_DISTRO = "Ubuntu-22.04"; const REDIS_IO_OPTIONS: Pick< RedisOptions, "family" | "maxRetriesPerRequest" | "enableReadyCheck" | "retryStrategy" > = { /** WSL localhost 转发仅稳定支持 IPv4;避免 ::1 ECONNREFUSED */ family: 4, /** BullMQ + 长连接 worker 必须为 null,否则断线时心跳抛 MaxRetriesPerRequestError */ maxRetriesPerRequest: null, enableReadyCheck: true, retryStrategy: (times) => Math.min(times * 500, 5_000), }; /** BullMQ / ioredis 连接选项 */ export function getRedisConnectionOptions(): RedisOptions { const url = process.env.REDIS_URL; if (!url) { throw new Error("REDIS_URL 未配置"); } const parsed = new URL(url); return { ...REDIS_IO_OPTIONS, host: parsed.hostname, port: Number(parsed.port || 6379), ...(parsed.password ? { password: parsed.password } : {}), }; } export function resetRedisClient(): void { if (redis) { redis.disconnect(); redis = null; } } export function getRedis(): Redis { if (!redis) { const url = process.env.REDIS_URL; if (!url) { const error = new Error("REDIS_URL 未配置"); console.error("[redis] 连接失败:", error.message); throw error; } redis = new Redis(url, REDIS_IO_OPTIONS); redis.on("error", (err) => { console.error("[redis] 运行时错误:", err); }); } return redis; } function wakeWslRedisForward(): void { if (process.platform !== "win32") { return; } const distro = process.env.WSL_DISTRO?.trim() || DEFAULT_WSL_DISTRO; try { execSync(`wsl -d ${distro} -- redis-cli PING`, { stdio: "ignore", timeout: 8_000, windowsHide: true, }); } catch { /* 唤醒失败不阻断重试 */ } } /** WSL IP 失效时回退 127.0.0.1(.dev/infra.env 可能写入过期的 172.x) */ export function buildRedisCandidateUrls(primary?: string): string[] { const base = primary?.trim() || process.env.REDIS_URL?.trim(); if (!base) { return []; } const urls: string[] = [base]; try { const parsed = new URL(base); const port = parsed.port || "6379"; const host = parsed.hostname.toLowerCase(); if (host !== "127.0.0.1") { urls.push(`redis://127.0.0.1:${port}`); } if (host !== "localhost") { urls.push(`redis://localhost:${port}`); } } catch { /* 保留 base */ } return [...new Set(urls)]; } async function probeRedisUrl( url: string, connectTimeoutMs: number, ): Promise { const probe = new Redis(url, { ...REDIS_IO_OPTIONS, maxRetriesPerRequest: 1, connectTimeout: connectTimeoutMs, lazyConnect: true, }); try { await probe.connect(); const pong = await probe.ping(); return pong === "PONG"; } catch { return false; } finally { await probe.quit().catch(() => undefined); } } function adoptRedisUrl(url: string): void { if (process.env.REDIS_URL === url) { return; } process.env.REDIS_URL = url; resetRedisClient(); } /** 新建连接探针(不复用单例),用于验收前摇 */ export async function probeFreshRedisConnection(): Promise { const url = process.env.REDIS_URL; if (!url) { throw new Error("REDIS_URL 未配置"); } const probe = new Redis(url, { ...REDIS_IO_OPTIONS, maxRetriesPerRequest: 1, connectTimeout: 4_000, lazyConnect: true, }); try { await probe.connect(); const pong = await probe.ping(); if (pong !== "PONG") { throw new Error(`Redis ping=${pong}`); } } finally { await probe.quit().catch(() => undefined); } } /** worker 启动前等待 Redis 可达(WSL localhost 转发有冷启动延迟) */ export async function ensureRedisReady( maxAttempts = REDIS_RETRY_DEFAULT_ATTEMPTS, delayMs = REDIS_RETRY_DEFAULT_DELAY_MS, ): Promise { const candidates = buildRedisCandidateUrls(); if (candidates.length === 0) { throw new Error("REDIS_URL 未配置"); } let consecutiveRefused = 0; let lastError = "Redis 连接失败"; for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { for (const url of candidates) { const ok = await probeRedisUrl(url, 4_000); if (ok) { adoptRedisUrl(url); return; } } lastError = `Connection is closed.(${candidates.join(" / ")})`; consecutiveRefused += 1; if ( process.env.DEV_INFRA_MODE?.trim().toLowerCase() === "native" && (consecutiveRefused === 1 || consecutiveRefused % 2 === 0) ) { wakeWslRedisForward(); } if (attempt === maxAttempts) { throw new Error( `${lastError},已重试 ${maxAttempts} 次;请先 npm run dev:infra:start`, ); } const backoffMs = attempt <= 4 ? delayMs : attempt <= 8 ? delayMs * 2 : delayMs * 3; await new Promise((resolve) => setTimeout(resolve, backoffMs)); } } const BOUNDED_REDIS_CONNECT_MS = 3_000; /** API 热路径短连接:Redis 不可达时快速失败,避免 getRedis 单例无限重试拖慢请求 */ export async function withBoundedRedis( fn: (client: Redis) => Promise, ): Promise { const candidates = buildRedisCandidateUrls(); if (candidates.length === 0) { return null; } for (const url of candidates) { const client = new Redis(url, { family: 4, connectTimeout: BOUNDED_REDIS_CONNECT_MS, maxRetriesPerRequest: 1, lazyConnect: true, retryStrategy: () => null, }); try { await client.connect(); const result = await fn(client); adoptRedisUrl(url); return result; } catch { /* 尝试下一个候选 URL */ } finally { await client.quit().catch(() => undefined); } } return null; }