quick-start.mjs

import assert from 'node:assert/strict';
import { randomUUID } from 'node:crypto';
import { setTimeout as delay } from 'node:timers/promises';
import { createBatchQueue } from 'queuebit';
import { DEMO_SNAPSHOT_ID, DEMO_SOURCE_ROWS, formatReceiptLine } from './receipt-domain.mjs';
import { defineReceiptTask } from './receipt-task.mjs';

const snapshots = new Map([[DEMO_SNAPSHOT_ID, DEMO_SOURCE_ROWS]]);
const receipts = new Map();
const notifications = new Map();
const queue = createBatchQueue({
  namespace: `receipt-demo-${randomUUID()}`,
  redis: { mode: 'url', url: process.argv[2] ?? 'redis://127.0.0.1:6379' },
  runtime: { mode: 'all', concurrency: 1, callbackConcurrency: 1 },
});

const repository = {
  /** @param {string} snapshotId @param {number} afterSourceId @param {number} limit @param {AbortSignal} signal */
  async readPage(snapshotId, afterSourceId, limit, signal) {
    signal.throwIfAborted();
    const rows = snapshots.get(snapshotId);
    if (!rows) throw new Error(`未知快照：${snapshotId}`);
    return rows.filter(row => row.sourceId > afterSourceId).slice(0, limit);
  },
};

const sink = {
  /** @param {string} key @param {import('./receipt-domain.mjs').Receipt} receipt @param {AbortSignal} signal */
  async putOnce(key, receipt, signal) {
    signal.throwIfAborted();
    const previous = receipts.get(key);
    if (previous && JSON.stringify(previous) !== JSON.stringify(receipt)) throw new Error(`回执幂等冲突：${key}`);
    if (!previous) {
      receipts.set(key, receipt);
      console.log(`生成回执：${formatReceiptLine(receipt)}`);
    }
  },
  /** @param {string} key @param {{eventId: string, snapshotId: string}} event @param {AbortSignal} signal */
  async completeOnce(key, event, signal) {
    signal.throwIfAborted();
    const previous = notifications.get(key);
    if (previous && previous.eventId !== event.eventId) throw new Error(`完成通知幂等冲突：${key}`);
    if (!previous) {
      notifications.set(key, event);
      console.log(`完成通知：${event.snapshotId}`);
    }
  },
};

const task = defineReceiptTask(queue, repository, sink);

try {
  await queue.ready();
  const input = { query: { snapshotId: DEMO_SNAPSHOT_ID }, idempotencyKey: DEMO_SNAPSHOT_ID };
  const started = await task.start(input);
  const duplicate = await task.start(input);
  assert.equal(duplicate.runId, started.runId);
  assert.equal(duplicate.created, false);
  console.log(`已提交：snapshot=${DEMO_SNAPSHOT_ID}，created=${started.created}`);
  console.log(`提交去重：${duplicate.created}，同一 runId=${duplicate.runId === started.runId}`);

  const deadline = Date.now() + 30_000;
  let run;
  do {
    run = await task.get(started.runId);
    if (!run) throw new Error('Run 不存在');
    if (run.status === 'failed' || run.status === 'cancelled') throw new Error(`Run 终止：${run.status}`);
    if (run.callbacks.deadLetters > 0) throw new Error('完成通知进入死信');
    if (run.status === 'success' && run.callbacks.delivered === 1) break;
    await delay(50);
  } while (Date.now() < deadline);

  assert.equal(run?.status, 'success', '等待 Run 成功超时');
  assert.equal(run.callbacks.delivered, 1, '等待通知交付超时');
  assert.equal(run.state?.afterSourceId, DEMO_SOURCE_ROWS.at(-1)?.sourceId);
  assert.equal(receipts.size, DEMO_SOURCE_ROWS.length);
  assert.equal(notifications.size, 1);
  console.log(`核对：${receipts.size} 条业务回执，${notifications.size} 条完成通知，状态 ${run.status}`);
} finally {
  const closed = await queue.close();
  console.log(`关闭：${closed.status}，timedOut=${closed.timedOut}`);
  assert.equal(closed.timedOut, false);
  assert.equal(closed.remainingExecutions + closed.remainingCallbacks, 0);
}