chajia/workers/rpa/job-handler.ts

import { Prisma } from "@prisma/client";
import { QUOTE_VALIDITY_MS } from "@/lib/constants/quote";
import { CAPTCHA_PAUSE_MINUTES } from "@/lib/constants/rpa";
import { prepareMotherShipStorageQuotes } from "@/modules/quote/quote-completeness";
import { detectDeviation } from "@/modules/alert/deviation-detector";
import { writeAlert } from "@/modules/alert/service";
import {
  closeCircuit,
  isCircuitOpen,
  recordRpaFailure,
  recordRpaSuccess,
} from "@/modules/cache/circuit-breaker";
import { overwriteL1, setL2, setL3 } from "@/modules/cache/redis-cache";
import { prisma } from "@/lib/prisma";
import { applyMarkupToQuotes, getMarkupRule } from "@/modules/pricing/engine";
import { getConfidenceScore } from "@/modules/quote/confidence";
import { handleRpaFailure } from "@/modules/quote/fallback-orchestrator";
import type { QuoteJobData } from "@/modules/quote/rpa-queue";
import { NO_RETRY_CODES, RpaError } from "@/modules/rpa/errors";
import { isEntryErrorCode } from "@/modules/rpa/error-mapper";
import {
  cargoToQuoteRequest,
  quoteItemsToRawTiers,
} from "@/modules/rpa/quote-mapper";
import { recordPostDone, safeRecord } from "@/modules/metrics/collector";
import { recordQuoteQueryOutcome } from "@/modules/quote/query-log";
import { mothershipProvider } from "@/workers/rpa/mothership";
import {
  heartbeat,
  isWorkerPaused,
  pauseWorker,
} from "@/workers/rpa/worker-state";

async function writeRpaAlert(
  alertType: "RPA_CAPTCHA" | "STRUCT_CHANGE" | "SESSION_EXPIRED",
  quoteId: string,
  cargoHash: string,
  detail: Record<string, unknown>,
): Promise<void> {
  await writeAlert(alertType, { quoteId, cargoHash, detail });
}

async function completeQuoteSuccess(
  data: QuoteJobData,
  rawQuotes: ReturnType<typeof quoteItemsToRawTiers>,
): Promise<void> {
  const uiQuotes = prepareMotherShipStorageQuotes(rawQuotes);
  const markupRule = await getMarkupRule(data.customerId);
  const markedQuotes = applyMarkupToQuotes(uiQuotes, markupRule);
  const validUntil = new Date(Date.now() + QUOTE_VALIDITY_MS);
  const cachePayload = { quotes: uiQuotes };

  await setL2(data.cargoHash, cachePayload);
  await setL3(data.cargoHash, cachePayload);

  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: markedQuotes as unknown as Prisma.InputJsonValue,
      validUntil,
      errorCode: null,
    },
  });

  await detectDeviation(data.cargoHash, uiQuotes, data.quoteId);

  const doneResponse = {
    quote_id: data.quoteId,
    status: "done" as const,
    source_type: "rpa" as const,
    is_realtime: true,
  };
  await overwriteL1(data.requestId, {
    quoteId: data.quoteId,
    response: doneResponse,
  });

  await recordQuoteQueryOutcome(data.quoteId, "success", {
    sourceType: "rpa",
    tierCount: markedQuotes.length,
  });
}

/** 单条 RPA job 处理（task-029） */
export async function processQuoteJob(
  data: QuoteJobData,
  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 {
    const result = await mothershipProvider.getQuote(
      cargoToQuoteRequest(data.cargo),
    );
    const rawQuotes = quoteItemsToRawTiers(result.items);
    await completeQuoteSuccess(data, rawQuotes);
    safeRecord(() =>
      recordPostDone({
        quoteId: data.quoteId,
        sourceType: "rpa",
        isRealtime: true,
      }),
    );
    await recordRpaSuccess();
    await closeCircuit();
  } catch (error) {
    if (error instanceof RpaError) {
      if (error.pauseWorker) {
        await pauseWorker(workerId, CAPTCHA_PAUSE_MINUTES);
        await writeRpaAlert("RPA_CAPTCHA", data.quoteId, data.cargoHash, {
          worker_id: workerId,
          message: error.message,
        });
      } else if (error.code === "STRUCT_CHANGE") {
        await writeRpaAlert("STRUCT_CHANGE", data.quoteId, data.cargoHash, {
          message: error.message,
        });
      } else if (error.code === "QUOTE_ENTRY_UNAVAILABLE") {
        await writeRpaAlert("STRUCT_CHANGE", data.quoteId, data.cargoHash, {
          message: error.message,
          entry_error: "QUOTE_ENTRY_UNAVAILABLE",
        });
      } else if (error.code === "SESSION_EXPIRED") {
        await writeRpaAlert("SESSION_EXPIRED", data.quoteId, data.cargoHash, {
          message: error.message,
        });
      }
    }

    const shouldCountFailure = !(
      error instanceof RpaError &&
      (NO_RETRY_CODES.has(error.code) || isEntryErrorCode(error.code))
    );
    if (shouldCountFailure) {
      await recordRpaFailure();
    }

    const isFinalAttempt = options?.isFinalAttempt ?? true;
    const canRetry =
      error instanceof RpaError &&
      error.retryable &&
      !isEntryErrorCode(error.code) &&
      !isFinalAttempt;

    if (canRetry) {
      throw error;
    }

    await handleRpaFailure(
      data.quoteId,
      data.cargoHash,
      data.customerId,
      error,
    );

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

    throw error;
  }
}