chajia/lib/redis.ts

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<boolean> {
  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<void> {
  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<void> {
  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<T>(
  fn: (client: Redis) => Promise<T>,
): Promise<T | null> {
  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;
}