chajia/lib/flock/exclusive-lock.ts

/**
 * Flock RPA 本机互斥锁 — 防止 embed Worker 与 batch 探针同时开浏览器打同一出口 IP
 */
import fs from "node:fs";
import path from "node:path";

const LOCK_DIR = path.join(process.cwd(), ".rpa");
const LOCK_PATH = path.join(LOCK_DIR, "flock-rpa.lock");

export type FlockLockInfo = {
  owner: string;
  pid: number;
  at: string;
};

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function isPidAlive(pid: number): boolean {
  if (!Number.isFinite(pid) || pid <= 0) {
    return false;
  }
  try {
    process.kill(pid, 0);
    return true;
  } catch {
    return false;
  }
}

export function readFlockExclusiveLock(): FlockLockInfo | null {
  try {
    if (!fs.existsSync(LOCK_PATH)) {
      return null;
    }
    const raw = JSON.parse(fs.readFileSync(LOCK_PATH, "utf8")) as FlockLockInfo;
    if (!raw?.owner || !raw?.pid) {
      return null;
    }
    return raw;
  } catch {
    return null;
  }
}

/** 锁是否被其他存活进程占用 */
export function isFlockExclusiveBusy(ignoreOwner?: string): boolean {
  const info = readFlockExclusiveLock();
  if (!info) {
    return false;
  }
  if (!isPidAlive(info.pid)) {
    try {
      fs.unlinkSync(LOCK_PATH);
    } catch {
      /* ignore */
    }
    return false;
  }
  if (ignoreOwner && info.owner === ignoreOwner && info.pid === process.pid) {
    return false;
  }
  return info.pid !== process.pid;
}

export function getFlockExclusiveBusyMessage(): string | null {
  if (!isFlockExclusiveBusy()) {
    return null;
  }
  const info = readFlockExclusiveLock();
  const who = info?.owner === "batch" ? "本地批量探针" : "其他 Flock 查价任务";
  return `Flock 官网访问正被${who}占用（串行防互踩），请待其结束后再试`;
}

/** 仅当本地 batch 探针占锁时用于网站入队拒绝（Worker 占锁时仍可入队排队） */
export function getFlockBatchExclusiveBusyMessage(): string | null {
  const info = readFlockExclusiveLock();
  if (!info || info.owner !== "batch") {
    return null;
  }
  if (!isPidAlive(info.pid)) {
    try {
      fs.unlinkSync(LOCK_PATH);
    } catch {
      /* ignore */
    }
    return null;
  }
  return "Flock 官网访问正被本地批量探针占用（串行防互踩），请待探针结束后再在网站查询";
}

function tryAcquireOnce(owner: string): boolean {
  fs.mkdirSync(LOCK_DIR, { recursive: true });
  const stale = readFlockExclusiveLock();
  if (stale && !isPidAlive(stale.pid)) {
    try {
      fs.unlinkSync(LOCK_PATH);
    } catch {
      /* ignore */
    }
  }
  try {
    const fd = fs.openSync(LOCK_PATH, "wx");
    const payload: FlockLockInfo = {
      owner,
      pid: process.pid,
      at: new Date().toISOString(),
    };
    fs.writeFileSync(fd, JSON.stringify(payload), "utf8");
    fs.closeSync(fd);
    return true;
  } catch (error) {
    const code =
      error && typeof error === "object" && "code" in error
        ? String((error as { code: unknown }).code)
        : "";
    if (code === "EEXIST") {
      const again = readFlockExclusiveLock();
      if (again && again.pid === process.pid) {
        return true;
      }
      if (again && !isPidAlive(again.pid)) {
        try {
          fs.unlinkSync(LOCK_PATH);
        } catch {
          /* ignore */
        }
        return tryAcquireOnce(owner);
      }
      return false;
    }
    throw error;
  }
}

export function releaseFlockExclusiveLock(owner: string): void {
  const info = readFlockExclusiveLock();
  if (!info) {
    return;
  }
  if (info.pid !== process.pid && info.owner !== owner) {
    return;
  }
  try {
    fs.unlinkSync(LOCK_PATH);
  } catch {
    /* ignore */
  }
}

/** 等待获取本机唯一 Flock 浏览器锁 */
export async function withFlockExclusiveLock<T>(
  owner: string,
  fn: () => Promise<T>,
  options?: { waitMs?: number; pollMs?: number },
): Promise<T> {
  const waitMs = options?.waitMs ?? 180_000;
  const pollMs = options?.pollMs ?? 1_000;
  const deadline = Date.now() + waitMs;

  while (!tryAcquireOnce(owner)) {
    if (Date.now() >= deadline) {
      throw new Error(
        getFlockExclusiveBusyMessage() ??
          "Flock 官网访问繁忙，请稍后重试",
      );
    }
    const info = readFlockExclusiveLock();
    console.warn(
      `[flock-lock] 等待互斥锁 owner=${info?.owner ?? "?"} pid=${info?.pid ?? "?"}…`,
    );
    await sleep(pollMs);
  }

  console.log(`[flock-lock] 已获取互斥锁 owner=${owner} pid=${process.pid}`);
  try {
    return await fn();
  } finally {
    releaseFlockExclusiveLock(owner);
    console.log(`[flock-lock] 已释放互斥锁 owner=${owner}`);
  }
}

/** 失败是否不应 BullMQ 重试（上限/拒号重试只会加重 IP 风控） */
export function isFlockNonRetryableFailure(message?: string): boolean {
  if (!message) {
    return false;
  }
  return (
    /flock_quote_limit|thank you for your interest|报价.*上限|quote\s*limit/i.test(
      message,
    ) ||
    /flock_account_rejected|issue creating your account/i.test(message) ||
    /flock_account_exists|already (have|registered)/i.test(message) ||
    /flock_login_failed|session_expired|未配置 FLOCK_LOGIN_/i.test(message) ||
    /互斥锁|正被.*占用|防互踩/i.test(message) ||
    /FLOCK_RATES_PENDING|still looking for the best rates|约需\s*15\s*分钟/i.test(
      message,
    )
  );
}