chajia/workers/scheduler/timeout-sweeper.ts

import { config } from "dotenv";

config();

import { Prisma } from "@prisma/client";
import { writeAlert } from "@/modules/alert/service";
import {
  QUOTE_TIMEOUT_MS,
  SWEEPER_INTERVAL_MS,
} from "@/lib/constants/quote";
import { prisma } from "@/lib/prisma";
import { getL3 } from "@/modules/cache/redis-cache";
import { getConfidenceScore } from "@/modules/quote/confidence";
import {
  applyMarkupToQuotes,
  getMarkupRule,
  type RawQuoteTier,
} from "@/modules/pricing/engine";
import { prepareMotherShipStorageQuotes } from "@/modules/quote/quote-completeness";
import { recordPostDone, safeRecord } from "@/modules/metrics/collector";
import { formatQuoteErrorMessage } from "@/modules/quote/quote-error-messages";
import { recordQuoteQueryOutcome } from "@/modules/quote/query-log";
import type { L2CachePayload } from "@/modules/quote/types";
import { getQuoteRpaQueue } from "@/workers/rpa/queue";

const TIMEOUT_ERROR = `processing 超过 ${Math.round(QUOTE_TIMEOUT_MS / 1_000)} 秒未完成`;

async function writeTimeoutAlert(
  quoteId: string,
  cargoHash: string,
  alertType: "STALE_FALLBACK" | "RPA_FAILED",
  detail: Record<string, unknown>,
): Promise<void> {
  await writeAlert(alertType, { quoteId, cargoHash, detail });
}

/** 单条 processing 超时记录处理：优先 L3 降级，否则 QUOTE_TIMEOUT */
export async function sweepTimedOutQuote(quoteId: string): Promise<void> {
  const record = await prisma.quoteRecord.findFirst({
    where: { quoteId, status: "processing", isDeleted: false },
  });
  if (!record) {
    return;
  }

  const elapsed = Date.now() - record.createdAt.getTime();
  if (elapsed < QUOTE_TIMEOUT_MS) {
    return;
  }

  const l3 = await getL3<L2CachePayload>(record.cargoHash);

  if (l3?.quotes?.length) {
    let uiQuotes: RawQuoteTier[];
    try {
      uiQuotes = prepareMotherShipStorageQuotes(l3.quotes as RawQuoteTier[]);
    } catch {
      await markQuoteTimeout(record.quoteId, record.cargoHash);
      return;
    }

    const markupRule = await getMarkupRule(record.customerId);
    const markedQuotes = applyMarkupToQuotes(uiQuotes, markupRule);
    const validUntil = new Date(Date.now() + 3 * 60 * 1_000);

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

    await writeTimeoutAlert(record.quoteId, record.cargoHash, "STALE_FALLBACK", {
      reason: TIMEOUT_ERROR,
      customer_id: record.customerId,
    });
    await recordQuoteQueryOutcome(record.quoteId, "stale", {
      failureReason: TIMEOUT_ERROR,
      sourceType: "stale",
      tierCount: markedQuotes.length,
    });
    safeRecord(() =>
      recordPostDone({
        quoteId: record.quoteId,
        sourceType: "stale",
        isRealtime: false,
      }),
    );
    return;
  }

  await markQuoteTimeout(record.quoteId, record.cargoHash);
}

async function cancelQuoteRpaJob(quoteId: string): Promise<void> {
  try {
    const job = await getQuoteRpaQueue().getJob(quoteId);
    if (!job) {
      return;
    }
    const state = await job.getState();
    if (state === "active") {
      await job.moveToFailed(new Error("QUOTE_TIMEOUT sweeper"), "0", true);
    } else if (state === "waiting" || state === "delayed") {
      await job.remove();
    }
  } catch (error) {
    console.warn(
      `[timeout-sweeper] 清理队列 job 失败 quote_id=${quoteId}:`,
      error instanceof Error ? error.message : error,
    );
  }
}

async function markQuoteTimeout(
  quoteId: string,
  cargoHash: string,
): Promise<void> {
  await cancelQuoteRpaJob(quoteId);
  await prisma.quoteRecord.update({
    where: { quoteId },
    data: {
      status: "failed",
      errorCode: "QUOTE_TIMEOUT",
      errorMessage: formatQuoteErrorMessage("QUOTE_TIMEOUT"),
      isRealtime: false,
    },
  });
  await writeTimeoutAlert(quoteId, cargoHash, "RPA_FAILED", {
    reason: TIMEOUT_ERROR,
  });
  await recordQuoteQueryOutcome(quoteId, "failed", {
    failureReason: formatQuoteErrorMessage("QUOTE_TIMEOUT"),
    errorCode: "QUOTE_TIMEOUT",
  });
}

/** 扫描所有超时 processing 记录 */
export async function sweepTimedOutQuotes(): Promise<number> {
  const cutoff = new Date(Date.now() - QUOTE_TIMEOUT_MS);
  const staleRecords = await prisma.quoteRecord.findMany({
    where: {
      status: "processing",
      isDeleted: false,
      createdAt: { lt: cutoff },
    },
    select: { quoteId: true },
    take: 50,
  });

  if (staleRecords.length === 0) {
    console.log(
      `[timeout-sweeper] 扫描完成，无超时记录（阈值 ${QUOTE_TIMEOUT_MS / 1000}s）`,
    );
    return 0;
  }

  console.log(
    `[timeout-sweeper] 发现 ${staleRecords.length} 条超时 processing，开始处理...`,
  );

  for (const { quoteId } of staleRecords) {
    await sweepTimedOutQuote(quoteId);
    console.log(`[timeout-sweeper] 已处理 quote_id=${quoteId}`);
  }

  return staleRecords.length;
}

/** 启动定时扫描（每 10s），启动时立即执行首轮 */
export function startTimeoutSweeper(): NodeJS.Timeout {
  sweepTimedOutQuotes().catch((error) => {
    console.error("[timeout-sweeper] 首轮扫描异常:", error);
  });

  const timer = setInterval(() => {
    sweepTimedOutQuotes().catch((error) => {
      console.error("[timeout-sweeper] 扫描异常:", error);
    });
  }, SWEEPER_INTERVAL_MS);

  return timer;
}

async function runOnce(): Promise<void> {
  console.log("[timeout-sweeper] 单次扫描模式");
  const count = await sweepTimedOutQuotes();
  console.log(`[timeout-sweeper] 单次扫描结束，处理 ${count} 条`);
  await prisma.$disconnect();
}

if (require.main === module) {
  const once = process.argv.includes("--once");

  if (!process.env.DATABASE_URL) {
    console.error("[timeout-sweeper] DATABASE_URL 未配置，请检查 .env");
    process.exit(1);
  }

  console.log("[timeout-sweeper] 启动，间隔", SWEEPER_INTERVAL_MS, "ms");

  if (once) {
    runOnce()
      .catch((error) => {
        console.error("[timeout-sweeper] 异常:", error);
        process.exit(1);
      })
      .finally(() => process.exit(0));
  } else {
    startTimeoutSweeper();
  }
}