ack: async () => {
  // Ack callbacks must be idempotent because receive pipelines may revisit completed stages.
  if (ctx.ackState === "acked") {
    return;
  }
  await params.onAck?.();
  ctx.ackState = "acked";
  ctx.ackedAt = Date.now();
  delete ctx.nackErrorMessage;
},

nack: async (error) => {
  await params.onNack?.(error);      // 多次调用会重复触发回调
  ctx.ackState = "nacked";
  ctx.nackErrorMessage = normalizeAckErrorMessage(error);
},

const ctx = createMessageReceiveContext({
  id: "test",
  channel: "test",
  message: {},
  onNack: () => console.log("nack called"),  // 会被调用两次
});
await ctx.nack(new Error("first"));
await ctx.nack(new Error("second"));  // 第二次调用，onNack 再次触发

nack: async (error) => {
  if (ctx.ackState !== "pending") {
    return;
  }
  await params.onNack?.(error);
  ctx.ackState = "nacked";
  ctx.nackErrorMessage = normalizeAckErrorMessage(error);
},