hn-hlw-service/hlw/automation/index.js

const { ObjectId } = require("mongodb");
const { randomUUID } = require("crypto");
const {
  AUTOMATION_TYPES,
  getTypeConfig,
  normalizeAutomationConfig,
  randomInteger,
} = require("./config");
const { decryptOrderFields } = require("../consult-order/format");
const tencentIM = require("../tencent-im");

const COLLECTION = "hlw-automation-task";
const TASK_STATUS = {
  PENDING: "PENDING",
  RUNNING: "RUNNING",
  SUCCEEDED: "SUCCEEDED",
  CANCELLED: "CANCELLED",
  FAILED: "FAILED",
};
const ERROR_SOURCES = {
  SYSTEM: "SYSTEM",
  HIS: "HIS",
};
const LEASE_MS = 2 * 60 * 1000;
const POLL_INTERVAL_MS = 1000;
const RECONCILE_INTERVAL_MS = 30 * 1000;
const MAX_TASKS_PER_TICK = 10;
const MAX_RETRY_DELAY_MS = 10 * 60 * 1000;
const WORKER_ID = `${process.pid}:${randomUUID()}`;

let workerDb = null;
let pollTimer = null;
let reconcileTimer = null;
let pollIsRunning = false;
let reconcileIsRunning = false;

class AutomationError extends Error {
  constructor(message, { retryable = false, source = ERROR_SOURCES.SYSTEM } = {}) {
    super(message);
    this.name = "AutomationError";
    this.retryable = retryable;
    this.source = source === ERROR_SOURCES.HIS ? ERROR_SOURCES.HIS : ERROR_SOURCES.SYSTEM;
  }
}

function getTaskKey(type, orderId, rxId) {
  if (type === AUTOMATION_TYPES.PASS_RX) {
    return `${type}:${rxId}`;
  }
  return `${type}:${orderId}`;
}

async function getConfig(db, corpId) {
  const config = await db.collection("hlw-config").findOne({ corpId });
  return normalizeAutomationConfig(config || {});
}

function getBusinessExpiry(order, fallbackMs = 30 * 60 * 1000) {
  if (order && Number.isFinite(order.expireTime) && order.expireTime > Date.now()) {
    return order.expireTime;
  }
  return Date.now() + fallbackMs;
}

function calculateDueAt(type, config, baseTime = Date.now(), random = Math.random) {
  const { minSeconds, maxSeconds } = getTypeConfig(config, type);
  const seconds = randomInteger(minSeconds, maxSeconds, random);
  let dueAt = baseTime + seconds * 1000;

  if (type === AUTOMATION_TYPES.OPEN_RX) {
    const firstSubmitRxIntervel =
      Number.isInteger(config.firstSubmitRxIntervel) && config.firstSubmitRxIntervel > 0
        ? config.firstSubmitRxIntervel
        : 0;
    dueAt = Math.max(dueAt, baseTime + firstSubmitRxIntervel * 1000);
  }

  if (dueAt < Date.now()) {
    dueAt = Date.now() + randomInteger(0, 2, random) * 1000;
  }
  return dueAt;
}

async function enqueueTask(
  db,
  { type, corpId, orderId, rxId = "", baseTime = Date.now(), expiresAt }
) {
  if (!corpId || !orderId || !Object.values(AUTOMATION_TYPES).includes(type)) {
    return { success: false, message: "自动化任务参数错误" };
  }
  if (type === AUTOMATION_TYPES.PASS_RX && !rxId) {
    return { success: false, message: "自动审方任务缺少处方ID" };
  }

  const config = await getConfig(db, corpId);
  const stageConfig = getTypeConfig(config, type);
  if (!stageConfig.enabled) {
    return { success: false, skipped: true, message: "自动化配置未开启" };
  }

  const now = Date.now();
  const taskKey = getTaskKey(type, orderId, rxId);
  const dueAt = calculateDueAt(type, config, baseTime);
  const task = {
    taskKey,
    type,
    corpId,
    orderId,
    rxId,
    status: TASK_STATUS.PENDING,
    dueAt,
    nextRunAt: dueAt,
    expiresAt: Number.isFinite(expiresAt) ? expiresAt : now + 30 * 60 * 1000,
    attempts: 0,
    leaseUntil: 0,
    leaseOwner: "",
    lastError: "",
    createdAt: now,
    updatedAt: now,
  };

  const reactivated = await db.collection(COLLECTION).updateOne(
    { taskKey, status: TASK_STATUS.CANCELLED },
    {
      $set: {
        ...task,
        createdAt: now,
      },
      $unset: {
        completedAt: "",
        result: "",
      },
    }
  );
  if (reactivated.modifiedCount === 1) {
    return {
      success: true,
      created: true,
      reactivated: true,
      taskKey,
      dueAt,
    };
  }

  let result;
  try {
    result = await db.collection(COLLECTION).updateOne(
      { taskKey },
      { $setOnInsert: task },
      { upsert: true }
    );
  } catch (error) {
    if (error && error.code === 11000) {
      return { success: true, created: false, taskKey, dueAt };
    }
    throw error;
  }
  return {
    success: true,
    created: result.upsertedCount === 1,
    taskKey,
    dueAt,
  };
}

async function scheduleAutoAccept({ db, orderId, corpId, baseTime, expiresAt }) {
  const eligible = await db.collection("consult-order").findOne(
    { orderId, corpId, orderSource: "ALIPAY_MINI" },
    { projection: { _id: 1 } }
  );
  if (!eligible) {
    return { success: false, skipped: true, message: "非支付宝小程序订单" };
  }
  return enqueueTask(db, {
    type: AUTOMATION_TYPES.ACCEPT,
    orderId,
    corpId,
    baseTime,
    expiresAt,
  });
}

async function scheduleAutoOpenRx({ db, orderId, corpId, baseTime, expiresAt }) {
  const eligible = await db.collection("consult-order").findOne(
    { orderId, corpId, orderSource: "ALIPAY_MINI" },
    { projection: { _id: 1 } }
  );
  if (!eligible) {
    return { success: false, skipped: true, message: "非支付宝小程序订单" };
  }
  return enqueueTask(db, {
    type: AUTOMATION_TYPES.OPEN_RX,
    orderId,
    corpId,
    baseTime,
    expiresAt,
  });
}

async function scheduleAutoPassRx({
  db,
  orderId,
  rxId,
  corpId,
  baseTime,
  expiresAt,
}) {
  const eligible = await db.collection("consult-order").findOne(
    { orderId, corpId, orderSource: "ALIPAY_MINI" },
    { projection: { _id: 1 } }
  );
  if (!eligible) {
    return { success: false, skipped: true, message: "非支付宝小程序订单" };
  }
  return enqueueTask(db, {
    type: AUTOMATION_TYPES.PASS_RX,
    orderId,
    rxId: rxId && rxId.toString(),
    corpId,
    baseTime,
    expiresAt,
  });
}

async function ensureIndexes(db) {
  const collection = db.collection(COLLECTION);
  await collection.createIndex({ taskKey: 1 }, { unique: true });
  await collection.createIndex({ status: 1, nextRunAt: 1, leaseUntil: 1 });
  await collection.createIndex({ orderId: 1, type: 1 });
  await collection.createIndex({ completedAt: 1 });
}

async function claimNextTask(db, now = Date.now()) {
  const result = await db.collection(COLLECTION).findOneAndUpdate(
    {
      $or: [
        {
          status: TASK_STATUS.PENDING,
          nextRunAt: { $lte: now },
        },
        {
          status: TASK_STATUS.RUNNING,
          leaseUntil: { $lte: now },
        },
      ],
    },
    {
      $set: {
        status: TASK_STATUS.RUNNING,
        leaseUntil: now + LEASE_MS,
        leaseOwner: WORKER_ID,
        updatedAt: now,
      },
      $inc: { attempts: 1 },
    },
    {
      sort: { nextRunAt: 1, createdAt: 1 },
      returnDocument: "after",
    }
  );
  return result && result.value ? result.value : result;
}

async function markTask(db, task, status, extra = {}) {
  const now = Date.now();
  const terminal = [
    TASK_STATUS.SUCCEEDED,
    TASK_STATUS.CANCELLED,
    TASK_STATUS.FAILED,
  ].includes(status);
  await db.collection(COLLECTION).updateOne(
    {
      _id: task._id,
      status: TASK_STATUS.RUNNING,
      leaseOwner: WORKER_ID,
    },
    {
      $set: {
        status,
        leaseUntil: 0,
        leaseOwner: "",
        updatedAt: now,
        ...(terminal ? { completedAt: now } : {}),
        ...extra,
      },
    }
  );
}

async function renewLease(db, task) {
  await db.collection(COLLECTION).updateOne(
    {
      _id: task._id,
      status: TASK_STATUS.RUNNING,
      leaseOwner: WORKER_ID,
    },
    {
      $set: {
        leaseUntil: Date.now() + LEASE_MS,
        updatedAt: Date.now(),
      },
    }
  );
}

function getRetryDelay(attempts) {
  const exponent = Math.max(0, Math.min(attempts - 1, 10));
  const base = Math.min(5000 * 2 ** exponent, MAX_RETRY_DELAY_MS);
  return Math.min(base + Math.floor(Math.random() * 3000), MAX_RETRY_DELAY_MS);
}

async function retryTask(db, task, error) {
  const now = Date.now();
  if (!Number.isFinite(task.expiresAt) || task.expiresAt <= now) {
    await markTask(db, task, TASK_STATUS.FAILED, {
      lastError: `业务已过期: ${error.message}`,
    });
    return;
  }
  const nextRunAt = Math.min(now + getRetryDelay(task.attempts || 1), task.expiresAt);
  await markTask(db, task, TASK_STATUS.PENDING, {
    nextRunAt,
    lastError: error.message,
  });
}

function isSingleAttemptTask(type) {
  return [AUTOMATION_TYPES.OPEN_RX, AUTOMATION_TYPES.PASS_RX].includes(type);
}

function isHisUploadFailure(message) {
  return typeof message === "string" && /^\s*\[his\]/i.test(message);
}

function cleanHisUploadFailureMessage(message) {
  if (typeof message !== "string") return "";
  return message
    .replace(/^\s*\[his\]\s*/i, "")
    .replace(/^处方上传(?:失败|错误)\s*[:：]\s*/, "")
    .trim();
}

function getFailureNotificationExt(error) {
  const errorSource =
    error && error.source === ERROR_SOURCES.HIS
      ? ERROR_SOURCES.HIS
      : ERROR_SOURCES.SYSTEM;
  const rawReason =
    error && typeof error.message === "string" ? error.message.trim() : "";
  const reason =
    errorSource === ERROR_SOURCES.HIS
      ? `HIS上传错误：${cleanHisUploadFailureMessage(rawReason) || "未知错误"}`
      : `系统错误：${rawReason || "未知错误"}`;
  return { errorSource, reason };
}

async function sendFailureNotification(db, task, error) {
  let doctorCode = "";

  try {
    const order = await db.collection("consult-order").findOne(
      { orderId: task.orderId, corpId: task.corpId },
      { projection: { doctorCode: 1 } }
    );
    doctorCode = order && order.doctorCode ? order.doctorCode : "";
  } catch (lookupError) {
    console.error(
      "[自动化任务] 查询失败通知接收人异常:",
      lookupError.message
    );
  }

  if (
    !doctorCode &&
    task.type === AUTOMATION_TYPES.PASS_RX &&
    ObjectId.isValid(task.rxId)
  ) {
    try {
      const rx = await db.collection("diagnostic-record").findOne(
        { _id: new ObjectId(task.rxId), orderId: task.orderId },
        { projection: { doctorCode: 1 } }
      );
      doctorCode = rx && rx.doctorCode ? rx.doctorCode : "";
    } catch (lookupError) {
      console.error(
        "[自动化任务] 查询处方失败通知接收人异常:",
        lookupError.message
      );
    }
  }

  if (!doctorCode) {
    console.error(
      `[自动化任务] 未找到订单 ${task.orderId} 的医生，无法发送失败通知`
    );
    return;
  }

  const ext = getFailureNotificationExt(error);
  try {
    const result = await tencentIM(
      {
        type: "sendSystemNotification",
        corpId: task.corpId,
        formAccount: task.orderId,
        toAccount: doctorCode,
        SyncOtherMachine: 1,
        msgBody: [
          {
            MsgType: "TIMCustomElem",
            MsgContent: {
              Data: "AUTORXFAIL",
              Desc: "notification",
              Ext: JSON.stringify(ext),
            },
          },
        ],
      },
      db
    );
    if (!result || !result.success) {
      console.error(
        `[自动化任务] 发送失败通知失败: ${result?.message || "未知错误"}`
      );
    }
  } catch (notificationError) {
    console.error(
      "[自动化任务] 发送失败通知异常:",
      notificationError.message
    );
  }
}

async function assertTaskEnabled(db, task) {
  const config = await getConfig(db, task.corpId);
  const { enabled } = getTypeConfig(config, task.type);
  if (!enabled) {
    await markTask(db, task, TASK_STATUS.CANCELLED, {
      lastError: "执行前检查发现自动化配置已关闭",
    });
    return null;
  }
  return config;
}

async function getAlipayOrder(db, orderId) {
  const raw = await db.collection("consult-order").findOne({ orderId });
  if (!raw) {
    throw new AutomationError("咨询订单不存在");
  }
  if (raw.orderSource !== "ALIPAY_MINI") {
    throw new AutomationError("仅支持支付宝小程序咨询订单");
  }
  return decryptOrderFields(raw);
}

async function handleAutoAccept(db, task) {
  const order = await getAlipayOrder(db, task.orderId);
  if (order.orderStatus === "processing") {
    await scheduleAutoOpenRx({
      db,
      orderId: order.orderId,
      corpId: order.corpId,
      baseTime: order.prescriptionStartTime || Date.now(),
      expiresAt: getBusinessExpiry(order),
    });
    return { orderId: order.orderId };
  }
  if (order.orderStatus !== "pending") {
    throw new AutomationError(`订单状态不支持自动接诊: ${order.orderStatus}`);
  }
  if (Number.isFinite(order.expireTime) && order.expireTime <= Date.now()) {
    throw new AutomationError("咨询订单已过期");
  }

  const consultOrder = require("../consult-order");
  const result = await consultOrder(
    {
      type: "acceptConsultOrder",
      orderId: order.orderId,
      corpId: order.corpId,
      doctorCode: order.doctorCode,
      operationSource: "AUTO",
    },
    db
  );
  if (!result || !result.success) {
    throw new AutomationError(result?.message || "自动接诊失败", { retryable: true });
  }
  return { orderId: order.orderId };
}

async function resolveDiagnosisList(db, order) {
  const result = [];
  const seen = new Set();
  const add = (code, name) => {
    const normalizedCode = typeof code === "string" ? code.trim() : "";
    const normalizedName = typeof name === "string" ? name.trim() : "";
    if (!normalizedCode || !normalizedName) return;
    const key = `${normalizedCode}:${normalizedName}`;
    if (!seen.has(key)) {
      result.push({ code: normalizedCode, name: normalizedName, desc: "" });
      seen.add(key);
    }
  };

  const medInfo = order.medInfo || {};
  add(medInfo.dise_codg, medInfo.dise_name);

  const diseaseNames = Array.isArray(order.diseases)
    ? order.diseases.filter((item) => typeof item === "string" && item.trim())
    : [];
  if (diseaseNames.length) {
    const diagnosisRecords = await db
      .collection("hlw-diagnosis")
      .find({ name: { $in: diseaseNames } }, { projection: { code: 1, name: 1 } })
      .toArray();
    for (const name of diseaseNames) {
      const match = diagnosisRecords.find((item) => item.name === name);
      if (!match || !match.code) {
        throw new AutomationError(`诊断“${name}”无法精确匹配诊断编码`);
      }
      add(match.code, match.name);
    }
  }

  // if (!result.length) {
  //   throw new AutomationError("订单缺少可用于开方的诊断编码");
  // }
  return result;
}

async function getMedicineConfig(db, corpId) {
  const records = await db
    .collection("hlw-config")
    .find(
      { group: `${corpId}-medicine-related` },
      { projection: { key: 1, list: 1 } }
    )
    .toArray();
  return records.reduce((map, item) => {
    if (item.key && Array.isArray(item.list)) {
      map[item.key] = item.list;
    }
    return map;
  }, {});
}

function matchConfig(list, code, name) {
  const items = Array.isArray(list) ? list : [];
  return (
    items.find((item) => code !== undefined && code !== "" && item.code == code) ||
    items.find((item) => name && item.name === name)
  );
}

async function resolvePrescriptionDrugs(db, order) {
  const orderDrugs = Array.isArray(order.drugs) ? order.drugs : [];
  if (!orderDrugs.length) {
    throw new AutomationError("订单缺少药品，无法自动开方");
  }

  const ids = orderDrugs
    .map((item) => (ObjectId.isValid(item._id) ? new ObjectId(item._id) : null))
    .filter(Boolean);
  if (ids.length !== orderDrugs.length) {
    throw new AutomationError("订单药品ID不完整");
  }

  const collectionName =
    order.consultType === "onlineMedicinePurchase" ? "online-drug-info" : "drug-info";
  const masterDrugs = await db
    .collection(collectionName)
    .find({ _id: { $in: ids }, onSale: true })
    .toArray();
  if (masterDrugs.length !== orderDrugs.length) {
    throw new AutomationError("订单中存在已下架或不存在的药品");
  }

  const medicineConfig = await getMedicineConfig(db, order.corpId);
  const prefix =
    order.consultType === "onlineMedicinePurchase" ? "online-medicine" : "store-medicine";
  const dosageUnitList = medicineConfig[`${prefix}-dosage-unit`] || [];
  const frequencyList = medicineConfig[`${prefix}-frequence`] || [];
  const usageList = medicineConfig[`${prefix}-administration`] || [];
  const unitList = medicineConfig["medicine-package-unit"] || [];

  return orderDrugs.map((requested) => {
    const master = masterDrugs.find(
      (item) => item._id.toString() === requested._id.toString()
    );
    const usage = matchConfig(
      usageList,
      requested.usageCode,
      requested.usageName || master.administration_method
    );
    const frequency = matchConfig(
      frequencyList,
      requested.frequencyCode,
      requested.frequencyName || master.freq
    );
    const dosageUnit = matchConfig(
      dosageUnitList,
      requested.dosage_unit_code,
      requested.dosage_unit || master.dosage_unit
    );
    const unit = matchConfig(unitList, requested.unit || master.unit, requested.unit || master.unit);
    const dosage = Number(requested.dosage);
    const quantity = Number(requested.quantity);
    const days = Number(master.days);

    if (!usage || !frequency || !dosageUnit || !unit) {
      throw new AutomationError(`药品“${master.name}”的用法用量配置无法匹配`);
    }
    if (!(dosage > 0) || !(quantity > 0) || !Number.isInteger(quantity)) {
      throw new AutomationError(`药品“${master.name}”的剂量或数量不正确`);
    }
    if (!(days > 0) || !Number.isInteger(days)) {
      throw new AutomationError(`药品“${master.name}”未维护有效的用药天数`);
    }
    if (!master.erpId || !master.insurance_code) {
      throw new AutomationError(`药品“${master.name}”缺少HIS所需编码`);
    }

    return {
      _id: master._id.toString(),
      erpId: master.erpId,
      dosage_form: master.dosage_form || "",
      days,
      dosage,
      dosage_unit: dosageUnit.name,
      dosage_unit_code: dosageUnit.code,
      drugName: master.name,
      specification: master.specification || "",
      frequencyCode: frequency.code,
      frequencyName: frequency.name,
      insurance_code: master.insurance_code,
      product_id: master.product_id,
      quantity,
      unit: unit.code,
      usageCode: usage.code,
      usageName: usage.name,
      package_amount: master.package_amount,
      recommended_quantity: master.recommended_quantity,
      forceSelfPay: master.forceSelfPay === "Y" ? "Y" : "N",
      limitUsageScope:
        typeof master.usage_restriction_desc === "string" &&
        master.usage_restriction_desc.trim()
          ? "Y"
          : "N",
    };
  });
}

async function buildAutoPrescriptionParams(db, order) {
  const doctor = await db.collection("hlw-doctor").findOne({
    corpId: order.corpId,
    doctorNo: order.doctorCode,
    job: "doctor",
  });
  if (!doctor) {
    throw new AutomationError("开方医生不存在");
  }
  if (doctor.onlineStatus !== "online") {
    throw new AutomationError("开方医生暂不在线", { retryable: true });
  }

  const [diagnosisList, drugs] = await Promise.all([
    resolveDiagnosisList(db, order),
    resolvePrescriptionDrugs(db, order),
  ]);
  const diseaseText = Array.isArray(order.diseases) ? order.diseases.join("，") : "";
  const complaint = [diseaseText, order.description]
    .filter((item) => typeof item === "string" && item.trim())
    .join(" ")
    .slice(0, 500);
  if (!complaint) {
    throw new AutomationError("订单缺少主诉和病情描述");
  }

  const config = await getConfig(db, order.corpId);
  const prescriptionType =
    order.consultType === "onlineMedicinePurchase"
      ? "onlineMedicinePurchase"
      : "storeMedicinePurchase";
  const medicinePurchaseRxDuration =
    Number.isInteger(config.medicinePurchaseRxDuration) &&
    config.medicinePurchaseRxDuration > 0
      ? config.medicinePurchaseRxDuration
      : 30;

  return {
    complaint,
    presentIllness:
      typeof order.pastHistoryStr === "string" ? order.pastHistoryStr : "",
    dispose: "",
    doctorCAUserId: doctor.signatureUrl || "",
    patientId: order.patientId,
    name: order.name,
    orderId: order.orderId,
    doctorCode: order.doctorCode,
    doctorName: order.doctorName,
    deptName: order.deptName,
    unitCode: order.unitCode,
    drugStoreNo: order.drugStoreNo,
    orderSource: order.orderSource,
    idCard: order.idCard,
    blhno: order.blhno,
    medOrgOrderNo: order.medorg_order_no,
    address: order.address,
    mobile: order.mobile,
    prescriptionType,
    pickUpType: order.pickUpType,
    expireTime:
      prescriptionType === "onlineMedicinePurchase"
        ? Date.now() + medicinePurchaseRxDuration * 60 * 1000
        : order.expireTime,
    diagnosisList,
    drugs,
  };
}

async function handleAutoOpenRx(db, task) {
  const order = await getAlipayOrder(db, task.orderId);
  const existing = await db.collection("diagnostic-record").findOne(
    {
      orderId: order.orderId,
      status: { $in: ["INIT", "PASS"] },
    },
    { projection: { _id: 1, status: 1, createTime: 1, expireTime: 1 } }
  );
  if (existing) {
    await scheduleAutoPassRx({
      db,
      orderId: order.orderId,
      rxId: existing._id,
      corpId: order.corpId,
      baseTime: existing.createTime || Date.now(),
      expiresAt: existing.expireTime || getBusinessExpiry(order),
    });
    return { orderId: order.orderId, rxId: existing._id.toString() };
  }

  if (order.orderStatus !== "processing") {
    throw new AutomationError(`订单状态不支持自动开方: ${order.orderStatus}`);
  }
  if (Number.isFinite(order.expireTime) && order.expireTime <= Date.now()) {
    throw new AutomationError("咨询订单已过期");
  }

  const params = await buildAutoPrescriptionParams(db, order);
  const diagnosticRecord = require("../diagnostic-record");
  const result = await diagnosticRecord(
    {
      type: "addConsultDiagnosis",
      corpId: order.corpId,
      params,
      automated: true,
      operationSource: "AUTO",
    },
    db
  );
  if (!result || !result.success) {
    throw new AutomationError(result?.message || "开方接口返回失败", {
      retryable: true,
    });
  }

  const rx = await db.collection("diagnostic-record").findOne(
    { orderId: order.orderId, status: { $in: ["INIT", "PASS"] } },
    { projection: { _id: 1, createTime: 1, expireTime: 1 } }
  );
  if (!rx) {
    throw new AutomationError("自动开方成功但未查询到处方", { retryable: true });
  }
  await scheduleAutoPassRx({
    db,
    orderId: order.orderId,
    rxId: rx._id,
    corpId: order.corpId,
    baseTime: rx.createTime || Date.now(),
    expiresAt: rx.expireTime || getBusinessExpiry(order),
  });
  return { orderId: order.orderId, rxId: rx._id.toString() };
}

async function handleAutoPassRx(db, task) {
  if (!ObjectId.isValid(task.rxId)) {
    throw new AutomationError("处方ID格式错误");
  }
  const rx = await db.collection("diagnostic-record").findOne({
    _id: new ObjectId(task.rxId),
    orderId: task.orderId,
  });
  if (!rx) {
    throw new AutomationError("处方不存在");
  }
  if (rx.status === "PASS") {
    return { orderId: task.orderId, rxId: task.rxId };
  }
  if (rx.status !== "INIT") {
    throw new AutomationError(`处方状态不支持自动审方: ${rx.status}`);
  }
  if (!rx.pharmacistNo) {
    throw new AutomationError("处方未分配审方药师", { retryable: true });
  }

  const pharmacist = await db.collection("hlw-doctor").findOne({
    corpId: task.corpId,
    doctorNo: rx.pharmacistNo,
    job: "pharmacist",
  });
  if (!pharmacist) {
    throw new AutomationError("审方药师不存在");
  }
  if (pharmacist.onlineStatus !== "online") {
    throw new AutomationError("审方药师暂不在线", { retryable: true });
  }

  const diagnosticRecord = require("../diagnostic-record");
  const result = await diagnosticRecord(
    {
      type: "auditDiagnosis",
      ids: [task.rxId],
      status: "PASS",
      pharmacistNo: rx.pharmacistNo,
      corpId: task.corpId,
      operationSource: "AUTO",
    },
    db
  );
  if (!result || !result.success) {
    const failure =
      Array.isArray(result?.failList) && result.failList[0]
        ? result.failList[0]
        : null;
    const detail = failure ? failure.message : result?.message;
    throw new AutomationError(detail || "审方接口返回失败", {
      retryable: true,
      source: isHisUploadFailure(detail)
        ? ERROR_SOURCES.HIS
        : ERROR_SOURCES.SYSTEM,
    });
  }
  return { orderId: task.orderId, rxId: task.rxId };
}

async function executeTask(db, task) {
  const config = await assertTaskEnabled(db, task);
  if (!config) return;

  let result;
  if (task.type === AUTOMATION_TYPES.ACCEPT) {
    result = await handleAutoAccept(db, task);
  } else if (task.type === AUTOMATION_TYPES.OPEN_RX) {
    result = await handleAutoOpenRx(db, task);
  } else if (task.type === AUTOMATION_TYPES.PASS_RX) {
    result = await handleAutoPassRx(db, task);
  } else {
    throw new AutomationError(`未知自动化任务类型: ${task.type}`);
  }

  await markTask(db, task, TASK_STATUS.SUCCEEDED, {
    lastError: "",
    result: result || {},
  });
}

async function processClaimedTask(db, task) {
  const heartbeat = setInterval(
    () =>
      renewLease(db, task).catch((error) => {
        console.error("[自动化任务] 续租失败:", error.message);
      }),
    Math.floor(LEASE_MS / 3)
  );
  if (typeof heartbeat.unref === "function") heartbeat.unref();
  try {
    await executeTask(db, task);
  } catch (error) {
    const automationError =
      error instanceof AutomationError
        ? error
        : new AutomationError(error.message || "自动化任务执行失败", {
            retryable: true,
          });
    if (isSingleAttemptTask(task.type)) {
      await markTask(db, task, TASK_STATUS.FAILED, {
        lastError: automationError.message,
      });
      await sendFailureNotification(db, task, automationError);
    } else if (automationError.retryable) {
      await retryTask(db, task, automationError);
    } else {
      await markTask(db, task, TASK_STATUS.FAILED, {
        lastError: automationError.message,
      });
    }
  } finally {
    clearInterval(heartbeat);
  }
}

async function runDueTasks(db = workerDb) {
  if (!db || pollIsRunning) return;
  pollIsRunning = true;
  try {
    const tasks = [];
    for (let i = 0; i < MAX_TASKS_PER_TICK; i += 1) {
      const task = await claimNextTask(db);
      if (!task) break;
      tasks.push(task);
    }
    await Promise.all(tasks.map((task) => processClaimedTask(db, task)));
  } catch (error) {
    console.error("[自动化任务] 执行器异常:", error.message);
  } finally {
    pollIsRunning = false;
  }
}

async function reconcileAutomationTasks(db = workerDb) {
  if (!db || reconcileIsRunning) return;
  reconcileIsRunning = true;
  try {
    const now = Date.now();
    const todayStart = new Date();
    todayStart.setHours(0, 0, 0, 0);
    const orders = await db
      .collection("consult-order")
      .find(
        {
          orderSource: "ALIPAY_MINI",
          orderStatus: { $in: ["pending", "processing"] },
          $or: [
            { expireTime: { $gt: now } },
            {
              expireTime: { $exists: false },
              createTime: { $gte: todayStart.getTime() },
            },
          ],
        },
        {
          projection: {
            orderId: 1,
            corpId: 1,
            orderStatus: 1,
            createTime: 1,
            prescriptionStartTime: 1,
            expireTime: 1,
          },
        }
      )
      .toArray();

    const corpIds = [...new Set(orders.map((item) => item.corpId).filter(Boolean))];
    const configs = await db
      .collection("hlw-config")
      .find({ corpId: { $in: corpIds } })
      .toArray();
    const configMap = new Map(
      configs.map((item) => [item.corpId, normalizeAutomationConfig(item)])
    );

    for (const order of orders) {
      const config = configMap.get(order.corpId) || normalizeAutomationConfig({});
      if (order.orderStatus === "pending" && config.autoAcceptOrder) {
        await scheduleAutoAccept({
          db,
          orderId: order.orderId,
          corpId: order.corpId,
          baseTime: order.createTime,
          expiresAt: order.expireTime,
        });
      }
      if (order.orderStatus === "processing" && config.autoOpenRx) {
        const record = await db.collection("diagnostic-record").findOne(
          { orderId: order.orderId, status: { $in: ["INIT", "PASS"] } },
          { projection: { _id: 1 } }
        );
        if (!record) {
          await scheduleAutoOpenRx({
            db,
            orderId: order.orderId,
            corpId: order.corpId,
            baseTime: order.prescriptionStartTime || order.createTime,
            expiresAt: order.expireTime,
          });
        }
      }
    }

    const pendingRxList = await db
      .collection("diagnostic-record")
      .find(
        {
          status: "INIT",
          orderSource: "ALIPAY_MINI",
          $or: [
            { expireTime: { $gt: now } },
            {
              expireTime: { $exists: false },
              createTime: { $gte: todayStart.getTime() },
            },
          ],
        },
        {
          projection: {
            _id: 1,
            orderId: 1,
            corpId: 1,
            createTime: 1,
            expireTime: 1,
          },
        }
      )
      .toArray();
    for (const rx of pendingRxList) {
      const config =
        configMap.get(rx.corpId) ||
        (await getConfig(db, rx.corpId));
      if (config.autoPassRx) {
        const order = orders.find((item) => item.orderId === rx.orderId);
        await scheduleAutoPassRx({
          db,
          orderId: rx.orderId,
          rxId: rx._id,
          corpId: rx.corpId,
          baseTime: rx.createTime,
          expiresAt: rx.expireTime || getBusinessExpiry(order),
        });
      }
    }
  } catch (error) {
    console.error("[自动化任务] 核对任务失败:", error.message);
  } finally {
    reconcileIsRunning = false;
  }
}

async function start(db) {
  return 
  if (pollTimer || reconcileTimer) return;
  workerDb = db;
  await ensureIndexes(db);
  await reconcileAutomationTasks(db);
  await runDueTasks(db);
  pollTimer = setInterval(() => runDueTasks(db), POLL_INTERVAL_MS);
  reconcileTimer = setInterval(
    () => reconcileAutomationTasks(db),
    RECONCILE_INTERVAL_MS
  );
  if (typeof pollTimer.unref === "function") pollTimer.unref();
  if (typeof reconcileTimer.unref === "function") reconcileTimer.unref();
  console.log("[自动化任务] 执行器已启动");
}

function stop() {
  if (pollTimer) clearInterval(pollTimer);
  if (reconcileTimer) clearInterval(reconcileTimer);
  pollTimer = null;
  reconcileTimer = null;
  workerDb = null;
  pollIsRunning = false;
  reconcileIsRunning = false;
}

module.exports = {
  COLLECTION,
  TASK_STATUS,
  AUTOMATION_TYPES,
  ERROR_SOURCES,
  AutomationError,
  getFailureNotificationExt,
  calculateDueAt,
  enqueueTask,
  scheduleAutoAccept,
  scheduleAutoOpenRx,
  scheduleAutoPassRx,
  buildAutoPrescriptionParams,
  claimNextTask,
  processClaimedTask,
  reconcileAutomationTasks,
  runDueTasks,
  start,
  stop,
};