chajia/workers/rpa/flock-job-handler.ts

import { Prisma } from "@prisma/client";
import { QUOTE_VALIDITY_MS } from "@/lib/constants/quote";
import { getFlockMinQuotes } from "@/lib/flock/env";
import { isFlockNonRetryableFailure } from "@/lib/flock/exclusive-lock";
import { prisma } from "@/lib/prisma";
import { getMarkupRule } from "@/modules/pricing/engine";
import { getConfidenceScore } from "@/modules/quote/confidence";
import { handleRpaFailure } from "@/modules/quote/fallback-orchestrator";
import type { FlockQuoteJobData } from "@/modules/flock/rpa-queue";
import { buildFlockStoredQuotes } from "@/modules/flock/quote-storage";
import { RpaError } from "@/modules/rpa/errors";
import {
  isProviderLoginFailureMessage,
  PROVIDER_LOGIN_FAILED_USER_MESSAGE,
} from "@/modules/rpa/provider-login-message";
import { recordPostDone, safeRecord } from "@/modules/metrics/collector";
import {
  closeCircuit,
  isCircuitOpen,
  recordRpaFailure,
  recordRpaSuccess,
} from "@/modules/cache/circuit-breaker";
import { runFlockQuoteRpa } from "@/workers/rpa/flock/run-quote";
import { heartbeat, isWorkerPaused } from "@/workers/rpa/worker-state";
import {
  buildFlockProgressPayload,
  mapFlockErrorToStage,
  parseFlockProgress,
  shouldOverwriteFlockFailureStage,
  type FlockRpaStage,
} from "@/lib/flock/rpa-progress";

async function writeFlockRpaStage(
  quoteId: string,
  stage: FlockRpaStage,
): Promise<void> {
  try {
    await prisma.quoteRecord.update({
      where: { quoteId },
      data: {
        quotesJson: buildFlockProgressPayload(stage) as unknown as Prisma.InputJsonValue,
      },
    });
  } catch (error) {
    console.warn(
      `[flock-job] 写进度失败 stage=${stage}:`,
      error instanceof Error ? error.message : error,
    );
  }
}

async function readFlockProgressStage(
  quoteId: string,
): Promise<string | null> {
  try {
    const row = await prisma.quoteRecord.findUnique({
      where: { quoteId },
      select: { quotesJson: true },
    });
    return parseFlockProgress(row?.quotesJson)?.stage ?? null;
  } catch {
    return null;
  }
}

export async function processFlockQuoteJob(
  data: FlockQuoteJobData,
  workerId: string,
  options?: { isFinalAttempt?: boolean },
): Promise<void> {
  await heartbeat(workerId);

  if (await isCircuitOpen()) {
    throw new RpaError("PAGE_LOAD_TIMEOUT", "RPA 熔断中，暂缓消费", {
      retryable: true,
    });
  }

  if (await isWorkerPaused(workerId)) {
    throw new RpaError("RPA_CAPTCHA", "Worker 已暂停", { retryable: true });
  }

  try {
    // 每轮（含 BullMQ 重试）从互斥锁阶段开始；勿在重试时把阶段回退逻辑搞乱——这里确实要重新抢锁
    await writeFlockRpaStage(data.quoteId, "lock");
    const result = await runFlockQuoteRpa(data.input, {
      customerId: data.customerId,
      onStage: (stage) => writeFlockRpaStage(data.quoteId, stage),
    });
    const minQuotes = getFlockMinQuotes();
    if (!result.ok || result.quotes.length < minQuotes) {
      const msg = result.errorMessage ?? "Flock 报价提取失败";
      if (isProviderLoginFailureMessage(msg)) {
        throw new RpaError(
          "PROVIDER_LOGIN_FAILED",
          PROVIDER_LOGIN_FAILED_USER_MESSAGE,
          { retryable: false },
        );
      }
      // 上限/拒号/互斥忙：禁止 BullMQ 二次空烧（否则加重 IP 风控且拖慢 embed）
      throw new RpaError("RPA_DATA_INVALID", msg, {
        retryable: !isFlockNonRetryableFailure(msg),
      });
    }

    const markupRule = await getMarkupRule(data.customerId);
    const storedQuotes = buildFlockStoredQuotes(
      result.quotes,
      markupRule,
      result.reference,
    );
    const validUntil = new Date(Date.now() + QUOTE_VALIDITY_MS);

    await prisma.quoteRecord.update({
      where: { quoteId: data.quoteId },
      data: {
        status: "done",
        sourceType: "rpa",
        isRealtime: true,
        confidenceScore: getConfidenceScore("rpa"),
        markupPercent:
          markupRule.type === "percent" ? markupRule.percent : 0,
        quotesJson: storedQuotes as unknown as Prisma.InputJsonValue,
        validUntil,
        errorCode: null,
        errorMessage: null,
      },
    });

    safeRecord(() =>
      recordPostDone({
        quoteId: data.quoteId,
        sourceType: "rpa",
        isRealtime: true,
      }),
    );
    await recordRpaSuccess();
    await closeCircuit();
  } catch (error) {
    await recordRpaFailure();

    const rawMsg = error instanceof Error ? error.message : String(error);
    const mapped = mapFlockErrorToStage(rawMsg);
    const currentStage = await readFlockProgressStage(data.quoteId);
    if (shouldOverwriteFlockFailureStage(currentStage, mapped)) {
      await writeFlockRpaStage(data.quoteId, mapped.stage);
    } else if (currentStage) {
      console.warn(
        `[flock-job] 保留进度 stage=${currentStage}（不因未匹配错误回退到 ${mapped.stage}）err=${rawMsg.slice(0, 120)}`,
      );
    }

    const isFinalAttempt = options?.isFinalAttempt ?? true;
    const canRetry =
      error instanceof RpaError && error.retryable && !isFinalAttempt;

    if (canRetry) throw error;

    await handleRpaFailure(
      data.quoteId,
      data.cargoHash,
      data.customerId,
      error,
      { skipStaleFallback: true },
    );

    if (error instanceof RpaError && !error.retryable) return;
    throw error;
  }
}