bot-28/tests/Regression/admin_realtime_client.mjs

import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import { test } from 'node:test';

// 执行交付文档中的实际参考类，避免测试一份与交付内容不同的实现。
const document = readFileSync(new URL('../../docs/总后台/后台实时推送-前端接入.md', import.meta.url), 'utf8');
const block = [...document.matchAll(/```js\n([\s\S]*?)\n```/g)]
  .map((match) => match[1]).find((code) => code.includes('export class AdminRealtime'));
const source = block.slice(block.indexOf('export class AdminRealtime'), block.indexOf('// 体彩订单列表页'))
  .replace('export class', 'class');
const A = 'admin.orders.withdraw';
const B = 'admin.orders.sport';
const flush = () => new Promise((resolve) => setImmediate(resolve));

function deferred() {
  let resolve;
  const promise = new Promise((done) => { resolve = done; });
  return { promise, resolve };
}

class Subscription {
  listeners = new Map();
  listen(name, callback) { this.listeners.set(name, callback); return this; }
  subscribed(callback) { this.ready = callback; return this; }
  error(callback) { this.onError = callback; return this; }
  emit(event) { this.listeners.get(`.${event.event}`)?.(event); }
}

class Echo {
  channels = new Map();
  connector = { pusher: { connection: { bind: (name, callback) => { this.changeState = callback; } } } };
  private(channel) {
    if (!this.channels.has(channel)) this.channels.set(channel, new Subscription());
    return this.channels.get(channel);
  }
  leave(channel) { this.channels.delete(channel); }
  disconnect() { this.changeState?.({ current: 'disconnected' }); }
}

async function setup() {
  const requests = [];
  const http = { get(url, options) {
    if (url.endsWith('/config')) return Promise.resolve({ code: 0, data: { channels: [A, B], last_seq: 0 } });
    const response = deferred();
    requests.push({ ...response, ...options.params });
    return response.promise;
  } };
  const Client = new Function('Echo', 'Pusher', 'API_BASE', `${source}\nreturn AdminRealtime;`)(Echo, {}, '');
  const client = new Client(http, 'fixture-token', () => {});
  await client.start();
  return { client, echo: client.echo, requests };
}

const event = (seq, action = 'updated', channel = A, key = 88) => ({
  seq, channel, action, event: `order.${action}`, key: { id: key }, data: action === 'deleted' ? null : { id: key, version: seq },
});
const response = (events, lastSeq = 10) => ({ code: 0, data: { resync: false, last_seq: lastSeq, events } });
const watch = (client, channel, callback, reload = async () => {}) => client.watch(channel,
  ['order.created', 'order.updated', 'order.deleted'], callback, reload);

test('older replay cannot resurrect a row deleted by a newer live event', async () => {
  const { client, echo, requests } = await setup();
  const rows = new Map();
  watch(client, A, (message) => {
    if (message.action === 'deleted') rows.delete(message.key.id);
    else rows.set(message.key.id, message.data);
  });
  echo.private(A).ready();
  echo.private(A).emit(event(2, 'deleted'));
  requests[0].resolve(response([event(1, 'created')]));
  await flush();
  assert.equal(rows.has(88), false);
  client.deliver(event(1, 'created'));
  assert.equal(rows.has(88), false, 'retain the deletion watermark after replay completes');
  echo.private(A).emit(event(3, 'created'));
  assert.equal(rows.get(88).version, 3, 'a genuinely newer creation is still applied');
});

test('old HTTP completion cannot clear the new connection resume point', async () => {
  const { client, echo, requests } = await setup();
  const received = [];
  watch(client, A, (message) => received.push(message.seq));
  echo.private(A).ready();
  echo.changeState({ current: 'unavailable' });
  requests[0].resolve(response([event(1)], 1));
  await flush();
  assert.deepEqual(received, [], 'discard the response belonging to the disconnected generation');
  echo.private(A).ready();
  assert.equal(requests.length, 2);
  assert.equal(requests[1].after_seq, 0);
  requests[1].resolve(response([event(2)], 2));
  await flush();
  assert.deepEqual(received, [2]);
});

test('unwatch and rewatch isolate pending responses for the same channel', async () => {
  const { client, echo, requests } = await setup();
  const old = [];
  const current = [];
  watch(client, A, (message) => old.push(message.seq));
  echo.private(A).ready();
  client.unwatch(A);
  watch(client, A, (message) => current.push(message.seq));
  echo.private(A).ready();
  requests[0].resolve(response([event(1)], 1));
  await flush();
  requests[1].resolve(response([event(2)], 2));
  await flush();
  assert.deepEqual(old, []);
  assert.deepEqual(current, [2]);
});

test('channels catch up independently even when another response has a greater last_seq', async () => {
  const { client, echo, requests } = await setup();
  const received = [];
  for (const channel of [A, B]) {
    watch(client, channel, (message) => received.push([message.channel, message.seq]));
    echo.private(channel).ready();
  }
  requests[0].resolve(response([event(90)], 100));
  await flush();
  requests[1].resolve(response([event(10, 'updated', B)], 100));
  await flush();
  assert.deepEqual(received, [[A, 90], [B, 10]]);
  assert.deepEqual(requests.map((r) => r.after_seq), [0, 0]);
});

test('resync buffers live messages until the HTTP list reload finishes', async () => {
  const { client, echo, requests } = await setup();
  const list = deferred();
  const received = [];
  watch(client, A, (message) => received.push(message.seq), () => list.promise);
  echo.private(A).ready();
  requests[0].resolve({ code: 0, data: { resync: true, last_seq: 5, events: [] } });
  await flush();
  echo.private(A).emit(event(5));
  echo.private(A).emit(event(6));
  assert.deepEqual(received, []);
  list.resolve();
  await flush();
  assert.deepEqual(received, [6]);
});

test('stop invalidates pending responses and removes previous session state', async () => {
  const { client, echo, requests } = await setup();
  const received = [];
  watch(client, A, (message) => received.push(message.seq));
  echo.private(A).ready();
  client.stop();
  requests[0].resolve(response([event(1)]));
  await flush();
  assert.deepEqual(received, []);
  assert.equal(client.pending.size, 0);
  assert.equal(client.latest.size, 0);
});