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 {
  applyMarkupToFlockCarriersByTier,
  applyMarkupToFlockFlexibilityByTier,
  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";
import { flockHoldSessionId } from "@/lib/constants/flock-quote-hold";
import { createFlockQuoteHold, saveFlockHoldFlexibility, saveFlockHoldCarriers } from "@/lib/flock/flock-quote-hold-store";
import {
  canPersistParkedQuoteSession,
  hasParkedQuoteSession,
} from "@/workers/rpa/parked-quote-session";

/** 本 job 阶段耗时：startedAtMs / lastAtMs */
const flockStageTiming = new Map<
  string,
  { startedAtMs: number; lastAtMs: number }
>();

async function writeFlockRpaStage(
  quoteId: string,
  stage: FlockRpaStage,
): Promise<void> {
  try {
    const now = Date.now();
    let timing = flockStageTiming.get(quoteId);
    if (!timing) {
      timing = { startedAtMs: now, lastAtMs: now };
      flockStageTiming.set(quoteId, timing);
    }
    const elapsed_ms = now - timing.startedAtMs;
    const stage_ms = now - timing.lastAtMs;
    timing.lastAtMs = now;
    console.log(
      `[flock-job] stage=${stage} quote=${quoteId} elapsed_ms=${elapsed_ms} stage_ms=${stage_ms}`,
    );
    await prisma.quoteRecord.update({
      where: { quoteId },
      data: {
        quotesJson: buildFlockProgressPayload(stage, {
          elapsed_ms,
          stage_ms,
        }) 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 holdSessionId =
      data.input.formMode === "logged_in_quick" &&
      canPersistParkedQuoteSession()
        ? flockHoldSessionId(data.quoteId)
        : undefined;
    const result = await runFlockQuoteRpa(data.input, {
      customerId: data.customerId,
      onStage: (stage) => writeFlockRpaStage(data.quoteId, stage),
      holdSessionId,
    });
    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 quoteRecord = await prisma.quoteRecord.findUnique({
      where: { quoteId: data.quoteId },
      select: { businessCustomerId: true },
    });
    const markupRule = await getMarkupRule(
      data.customerId,
      quoteRecord?.businessCustomerId,
    );
    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,
      },
    });

    // 档内灵活价/承运商价与一级档位同样叠加加价（侧栏优先展示 rateUsd）
    const markedFlexibilityByTier = result.flexibilityByTier
      ? applyMarkupToFlockFlexibilityByTier(
          result.flexibilityByTier,
          markupRule,
        )
      : undefined;
    const markedCarriersByTier = result.carriersByTier
      ? applyMarkupToFlockCarriersByTier(result.carriersByTier, markupRule)
      : undefined;

    if (
      result.quoteSessionId &&
      hasParkedQuoteSession(result.quoteSessionId)
    ) {
      await createFlockQuoteHold({
        quoteId: data.quoteId,
        sessionId: result.quoteSessionId,
        customerId: data.customerId,
        flexibilityByTier: markedFlexibilityByTier,
        carriersByTier: markedCarriersByTier,
      });
      console.log(
        `[flock-job] quote-hold created quote=${data.quoteId} session=${result.quoteSessionId.slice(0, 12)}`,
      );
    } else if (
      (markedFlexibilityByTier &&
        Object.keys(markedFlexibilityByTier).length > 0) ||
      (markedCarriersByTier && Object.keys(markedCarriersByTier).length > 0)
    ) {
      // park 失败时仍落档内价，供前端弹窗首屏展示（二次拉价仅兜底）
      if (
        markedFlexibilityByTier &&
        Object.keys(markedFlexibilityByTier).length > 0
      ) {
        await saveFlockHoldFlexibility(
          data.quoteId,
          markedFlexibilityByTier,
        );
      }
      if (
        markedCarriersByTier &&
        Object.keys(markedCarriersByTier).length > 0
      ) {
        await saveFlockHoldCarriers(data.quoteId, markedCarriersByTier);
      }
      console.log(
        `[flock-job] pricing saved without hold quote=${data.quoteId} flex=${Object.keys(markedFlexibilityByTier ?? {}).join(",")} carriers=${Object.keys(markedCarriersByTier ?? {}).join(",")}`,
      );
    }

    safeRecord(() =>
      recordPostDone({
        quoteId: data.quoteId,
        sourceType: "rpa",
        isRealtime: true,
      }),
    );
    await recordRpaSuccess();
    await closeCircuit();
    flockStageTiming.delete(data.quoteId);
  } 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)}`,
      );
    }

    flockStageTiming.delete(data.quoteId);
    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,
        businessCustomerId: (
          await prisma.quoteRecord.findUnique({
            where: { quoteId: data.quoteId },
            select: { businessCustomerId: true },
          })
        )?.businessCustomerId,
      },
    );

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