chajia/modules/address/candidates-queue.ts

import { randomUUID } from "node:crypto";
import {
  ADDRESS_CANDIDATES_POLL_MS,
  ADDRESS_CANDIDATES_RESULT_PREFIX,
  ADDRESS_CANDIDATES_RESULT_TTL_SECONDS,
} from "@/lib/constants/address-candidates-queue";
import { ADDRESS_CANDIDATES_TIMEOUT_MS } from "@/lib/constants/rpa";
import { ensureNativeInfraIfNeeded } from "@/lib/infra/ensure-native-infra";
import {
  ensureRedisReady,
  getRedis,
  resetRedisClient,
} from "@/lib/redis";
import { isParkedSessionEnabled } from "@/lib/rpa/env";
import { spawnFetchMothershipAddressCandidates } from "@/lib/rpa/spawn-address-candidates";
import { isAnyRpaWorkerHealthy } from "@/lib/rpa/worker-health";
import type {
  AddressLookupInput,
  MothershipCandidatesResult,
} from "@/modules/address/types";
import { RpaError, type RpaErrorCode } from "@/modules/rpa/errors";
import { addAddressCandidatesJob } from "@/workers/rpa/queue";
import type { AddressCandidatesJobData } from "@/modules/address/candidates-queue-types";

type StoredCandidatesSuccess = {
  ok: true;
  data: MothershipCandidatesResult;
};

type StoredCandidatesFailure = {
  ok: false;
  code: RpaErrorCode;
  message: string;
  retryable?: boolean;
};

type StoredCandidatesResult = StoredCandidatesSuccess | StoredCandidatesFailure;

function resultKey(jobId: string): string {
  return `${ADDRESS_CANDIDATES_RESULT_PREFIX}${jobId}`;
}

async function ensureCandidatesRedisReady(): Promise<void> {
  try {
    await ensureRedisReady(10, 800);
  } catch {
    ensureNativeInfraIfNeeded();
    resetRedisClient();
    await ensureRedisReady(15, 1_000);
  }
}

/** 快速探测是否可走 BullMQ（Redis + Worker 心跳均可用） */
async function canUseCandidatesQueue(): Promise<boolean> {
  try {
    await ensureRedisReady(5, 500);
    return await isAnyRpaWorkerHealthy();
  } catch {
    return false;
  }
}

export async function saveAddressCandidatesResult(
  jobId: string,
  payload: StoredCandidatesResult,
): Promise<void> {
  await ensureCandidatesRedisReady();
  await getRedis().setex(
    resultKey(jobId),
    ADDRESS_CANDIDATES_RESULT_TTL_SECONDS,
    JSON.stringify(payload),
  );
}

async function readAddressCandidatesResult(
  jobId: string,
): Promise<StoredCandidatesResult | null> {
  const raw = await getRedis().get(resultKey(jobId));
  if (!raw) {
    return null;
  }
  return JSON.parse(raw) as StoredCandidatesResult;
}

async function requestViaSpawn(input: {
  pickup: AddressLookupInput;
  delivery: AddressLookupInput;
}): Promise<MothershipCandidatesResult> {
  return spawnFetchMothershipAddressCandidates(input);
}

/** 经 RPA Worker 执行地址候选；Worker/队列不可用时回退 spawn 子进程 */
export async function requestMothershipCandidatesViaWorker(input: {
  pickup: AddressLookupInput;
  delivery: AddressLookupInput;
}): Promise<MothershipCandidatesResult> {
  if (process.env.RPA_QUEUE_ENABLED === "false") {
    return requestViaSpawn(input);
  }

  const parkedSessionsEnabled = isParkedSessionEnabled();
  const useQueue = await canUseCandidatesQueue();
  if (!useQueue) {
    if (parkedSessionsEnabled) {
      throw new RpaError(
        "PAGE_LOAD_TIMEOUT",
        "RPA Worker 未运行，无法使用单会话加速。请执行 npm run worker:rpa 或 npm run dev:start",
        { retryable: true },
      );
    }
    console.warn(
      "[candidates] Redis/Worker 不可用，改用子进程执行地址联想",
    );
    return requestViaSpawn(input);
  }

  await ensureCandidatesRedisReady();

  const jobId = randomUUID();
  await addAddressCandidatesJob({
    jobId,
    pickup: input.pickup,
    delivery: input.delivery,
  });

  const deadline = Date.now() + ADDRESS_CANDIDATES_TIMEOUT_MS;
  while (Date.now() < deadline) {
    try {
      const stored = await readAddressCandidatesResult(jobId);
      if (stored?.ok) {
        return stored.data;
      }
      if (stored && !stored.ok) {
        throw new RpaError(stored.code, stored.message, {
          retryable: stored.retryable ?? false,
        });
      }
    } catch (error) {
      if (error instanceof RpaError) {
        throw error;
      }
      ensureNativeInfraIfNeeded();
      resetRedisClient();
      await ensureCandidatesRedisReady();
    }
    await new Promise((resolve) => setTimeout(resolve, ADDRESS_CANDIDATES_POLL_MS));
  }

  throw new RpaError(
    "PAGE_LOAD_TIMEOUT",
    "MotherShip 地址联想超时，请确认 RPA Worker 与 Redis 已启动（npm run dev:start）",
    { retryable: true },
  );
}