bot-28/public/static/js/realtime-probe.js

export class ProbeError extends Error {
  constructor(message, status = 0) { super(message); this.status = status; }
}

export function apiAddress(value) {
  let url;
  try { url = new URL(value.trim()); } catch { throw new ProbeError('请填写完整 API 地址，例如 https://api.example.com'); }
  if (!['http:', 'https:'].includes(url.protocol) || url.username || url.password || url.search || url.hash) {
    throw new ProbeError('API 地址只允许 http/https 域名和路径，不要包含账号、查询参数或 #。');
  }
  return url.href.replace(/\/+$/, '');
}

export function apiEndpoint(base, path) {
  if (typeof path !== 'string' || !path.startsWith('/') || path.startsWith('//') || path.includes('\\')) {
    throw new ProbeError('接口路径无效，请检查后端配置。');
  }
  const url = new URL(`${base}${path}`);
  if (url.origin !== new URL(base).origin) throw new ProbeError('授权接口必须与填写的 API 地址同源。');
  return url.href;
}

export function redact(value, token = '', depth = 0) {
  if (depth > 8) return '[内容层级过深]';
  if (typeof value === 'string') {
    return (token ? value.split(token).join('[已隐藏 token]') : value).replace(/Bearer\s+[^\s"<>]+/gi, 'Bearer [已隐藏]');
  }
  if (Array.isArray(value)) return value.slice(0, 100).map((v) => redact(v, token, depth + 1));
  if (value && typeof value === 'object') {
    return Object.fromEntries(Object.entries(value).map(([key, v]) => [key,
      /authorization|password|secret|token|ticket/i.test(key) ? '[已隐藏]' : redact(v, token, depth + 1)]));
  }
  return value;
}

export async function requestJSON(url, { method = 'GET', token, body, signal } = {}) {
  let response;
  try {
    response = await fetch(url, {
      method, signal, credentials: 'omit', cache: 'no-store', redirect: 'error',
      headers: { Accept: 'application/json', Authorization: `Bearer ${token}`,
        ...(body ? { 'Content-Type': 'application/x-www-form-urlencoded' } : {}) },
      ...(body ? { body: new URLSearchParams(body) } : {}),
    });
  } catch (error) {
    if (error.name === 'AbortError') throw error;
    throw new ProbeError('请求未能完成：请检查 API 地址、HTTPS 证书和跨域配置。');
  }
  let data;
  try { data = await response.json(); } catch {
    throw new ProbeError(`接口未返回 JSON（HTTP ${response.status}），请检查 API 地址和站点转发配置。`, response.status);
  }
  if (!response.ok) throw new ProbeError(data.msg || data.message || `HTTP ${response.status}`, response.status);
  return data;
}

const EVENTS = ['order.created', 'order.updated', 'order.deleted', 'user.created', 'user.online', 'user.offline', 'stats.updated'];

export class RealtimeProbe {
  constructor({ Echo, Pusher, request = requestJSON, onChange = () => {}, onLog = () => {}, onEvent = () => {} }) {
    Object.assign(this, { Echo, Pusher, request, onChange, onLog, onEvent });
    this.epoch = 0;
    this.session = 0;
    this.controllers = new Set();
    this.channels = new Map();
    this.seen = new Set();
    this.state = this.emptyState();
    this.token = '';
  }

  emptyState() {
    return { connection: 'disconnected', api: 'idle', config: null, socketId: '',
      liveCount: 0, replayCount: 0, duplicateCount: 0, lastSeq: 0, expired: false, channels: [] };
  }

  emit() {
    this.state.channels = [...this.channels.values()].map(({ name, status, checkpoint, liveSeq, pendingAfter, replaying }) =>
      ({ name, status, checkpoint, liveSeq, pendingAfter, replaying }));
    this.onChange(this.state);
  }

  log(level, message, details) {
    this.onLog({ level, message: redact(message, this.token), details: redact(details, this.token), time: Date.now() });
  }

  invalidate() {
    this.disconnect({ quiet: true });
    this.session++;
    this.token = '';
    this.channels.clear();
    this.seen.clear();
    this.state = this.emptyState();
    this.emit();
  }

  clearCounts() {
    this.state.liveCount = this.state.replayCount = this.state.duplicateCount = 0;
    this.emit();
  }

  async get(path, options = {}) {
    const controller = new AbortController();
    this.controllers.add(controller);
    const timeout = setTimeout(() => controller.abort(), 15000);
    try {
      return await this.request(apiEndpoint(this.base, path), { token: this.token, signal: controller.signal, ...options });
    } finally {
      clearTimeout(timeout);
      this.controllers.delete(controller);
    }
  }

  disconnect({ quiet = false } = {}) {
    this.epoch++;
    for (const channel of this.channels.values()) {
      channel.pendingAfter = Math.min(channel.pendingAfter ?? channel.checkpoint, channel.checkpoint);
      channel.request = null;
      channel.replaying = false;
      channel.status = 'idle';
    }
    for (const controller of this.controllers) controller.abort();
    this.controllers.clear();
    const echo = this.echo;
    this.echo = null;
    echo?.disconnect();
    this.state.connection = 'disconnected';
    this.state.socketId = '';
    if (!quiet) this.log('info', '已断开连接；续传点已保留，可再次连接验证断线补发。');
    this.emit();
  }

  async configure(base, token) {
    this.disconnect({ quiet: true });
    this.session++;
    const session = this.session;
    this.channels.clear();
    this.seen.clear();
    this.token = '';
    this.state = this.emptyState();
    this.state.api = 'loading';
    this.emit();
    this.log('info', '正在读取后台实时推送配置。');
    try {
      this.base = apiAddress(base);
      this.token = token.trim().replace(/^Bearer\s+/i, '').trim();
      if (!this.token || /\s/.test(this.token)) throw new ProbeError('请填写有效的后台管理员 token，可带 Bearer 前缀。');
      const response = await this.get('/admin/realtime/config');
      if (session !== this.session) return null;
      if (response.code !== 0) throw new ProbeError(response.msg || '后台未返回成功状态，请检查 token 或推送开关。');
      const config = response.data;
      if (!config || typeof config.key !== 'string' || !config.key || typeof config.ws_host !== 'string' || !config.ws_host
          || !Array.isArray(config.channels) || !config.channels.every((c) => typeof c === 'string')
          || !Number.isInteger(config.ws_port) || config.ws_port < 1 || config.ws_port > 65535
          || typeof config.force_tls !== 'boolean' || !Number.isSafeInteger(config.last_seq) || config.last_seq < 0) {
        throw new ProbeError('连接参数不完整，请检查 REVERB_APP_KEY、ADMIN_WS_PUBLIC_HOST / PORT / SCHEME。');
      }
      const wsURL = new URL(`${config.force_tls ? 'wss' : 'ws'}://${config.ws_host}:${config.ws_port}`);
      if (wsURL.username || wsURL.password || wsURL.pathname !== '/' || wsURL.search || wsURL.hash) {
        throw new ProbeError('WebSocket host 应只填写域名，不带协议或路径。');
      }
      apiEndpoint(this.base, config.auth_endpoint);
      this.state.config = {
        key: config.key, ws_host: config.ws_host, ws_port: config.ws_port, force_tls: config.force_tls,
        auth_endpoint: config.auth_endpoint, last_seq: config.last_seq,
        session_expires_at: config.session_expires_at || null,
        channels: [...new Set(config.channels)],
      };
      this.state.lastSeq = config.last_seq;
      this.state.api = 'ok';
      this.log('success', `配置读取成功，当前账号可订阅 ${this.state.config.channels.length} 个频道。`);
      this.emit();
      return this.state.config;
    } catch (error) {
      if (session !== this.session) return null;
      this.state.api = 'error';
      this.log('error', error.name === 'AbortError' ? '配置请求已取消或超时，请重试。' : error.message);
      this.emit();
      throw error;
    }
  }

  connect(names) {
    if (!this.state.config || !this.token) throw new ProbeError('请先获取连接配置。');
    if (this.state.expired) throw new ProbeError('登录已过期，请更新 token 后重新获取配置。');
    if (typeof this.Echo !== 'function' || typeof this.Pusher !== 'function') throw new ProbeError('本地客户端文件加载失败，请检查 static/vendor/realtime 文件是否已上传。');
    if (globalThis.location?.protocol === 'https:' && !this.state.config.force_tls) {
      throw new ProbeError('当前页面是 HTTPS，但后端配置了 ws。请将 ADMIN_WS_PUBLIC_SCHEME 配置为 https。');
    }
    names = [...new Set(names)].filter((name) => this.state.config.channels.includes(name));
    if (!names.length) throw new ProbeError('请至少选择一个有权限的频道。');
    this.disconnect({ quiet: true });
    const epoch = this.epoch;
    const c = this.state.config;
    this.state.connection = 'connecting';
    this.echo = new this.Echo({
      broadcaster: 'reverb', Pusher: this.Pusher, key: c.key,
      wsHost: c.ws_host, wsPort: c.ws_port, wssPort: c.ws_port,
      forceTLS: c.force_tls, enabledTransports: ['ws', 'wss'], enableStats: false, disableStats: true,
      authorizer: (channel) => ({ authorize: (socketId, callback) => this.authorize(channel.name, socketId, callback, epoch) }),
    });
    const echo = this.echo;
    echo.connector.pusher.connection.bind('state_change', ({ current }) => {
      if (this.echo !== echo || epoch !== this.epoch) return;
      this.state.connection = current;
      if (current === 'connected') {
        this.state.socketId = echo.socketId() || '';
        this.log('success', 'WebSocket 握手成功。接下来检查频道授权。');
      } else {
        for (const channel of this.channels.values()) {
          channel.pendingAfter = Math.min(channel.pendingAfter ?? channel.checkpoint, channel.checkpoint);
          channel.request = null;
          channel.replaying = false;
          channel.status = 'idle';
        }
        this.log(current === 'failed' || current === 'unavailable' ? 'error' : 'info', `连接状态：${current}`);
      }
      this.emit();
    });
    echo.connector.pusher.connection.bind('error', (error) => {
      if (this.echo === echo) this.log('error', 'WebSocket 连接错误，请检查域名、Nginx、证书和 Reverb 进程。', error);
    });
    this.subscribe(names);
    this.emit();
  }

  async authorize(channelName, socketId, callback, epoch) {
    const name = channelName.replace(/^private-/, '');
    const channel = this.channels.get(name);
    if (channel) channel.status = 'authorizing';
    this.emit();
    try {
      const result = await this.get(this.state.config.auth_endpoint, { method: 'POST', body: { socket_id: socketId, channel_name: channelName } });
      if (epoch !== this.epoch || this.echo?.socketId() !== socketId) return;
      if (typeof result.auth !== 'string' || !result.auth) throw new ProbeError('授权接口缺少 auth 字段，请检查请求是否到达正确的 API。');
      callback(null, result);
    } catch (error) {
      if (epoch !== this.epoch || this.echo?.socketId() !== socketId) return;
      if (channel) channel.status = error.status === 403 ? 'denied' : 'error';
      const message = error.status === 401 ? '登录已过期或 token 无效，请重新登录并更新 token。'
        : error.status === 403 ? '频道授权被拒绝，请检查账号权限或后台推送开关。' : error.message;
      this.log('error', `${name}：${message}`);
      callback(error, null);
      if (error.status === 401) { this.state.expired = true; this.disconnect({ quiet: true }); }
      this.emit();
    }
  }

  subscribe(names) {
    if (!this.echo) throw new ProbeError('请先连接 WebSocket。');
    const allowed = this.state.config.channels;
    const wanted = new Set(names.filter((name) => allowed.includes(name)));
    for (const [name, channel] of this.channels) {
      if (!wanted.has(name)) {
        this.echo.leave(name);
        channel.active = false;
        channel.binding = null;
        channel.status = 'idle'; channel.request = null; channel.replaying = false;
      }
    }
    const echo = this.echo;
    const epoch = this.epoch;
    for (const name of wanted) {
      let channel = this.channels.get(name);
      if (channel?.binding?.echo === echo && channel.active) continue;
      if (!channel) {
        channel = { name, checkpoint: this.state.config.last_seq, liveSeq: 0, pendingAfter: this.state.config.last_seq };
        this.channels.set(name, channel);
      }
      channel.status = 'subscribing';
      channel.active = true;
      const sub = echo.private(name);
      const binding = { echo, sub };
      channel.binding = binding;
      EVENTS.forEach((event) => sub.listen(`.${event}`, (message) => {
        if (this.echo !== echo || epoch !== this.epoch || !channel.active || channel.binding !== binding) return;
        if (message.channel !== name) { this.log('warning', '消息频道与订阅频道不一致，未计入检测结果。'); return; }
        channel.liveSeq = Math.max(channel.liveSeq, message.seq || 0);
        if (channel.pendingAfter === null) channel.checkpoint = Math.max(channel.checkpoint, channel.liveSeq);
        this.receive(message, 'live');
      }));
      sub.subscribed(() => {
        if (this.echo !== echo || epoch !== this.epoch || !channel.active || channel.binding !== binding) return;
        channel.status = 'subscribed';
        this.log('success', `${name} 订阅成功。`);
        this.emit();
        this.replay([name]);
      });
      sub.error((error) => {
        if (this.echo !== echo || epoch !== this.epoch || !channel.active || channel.binding !== binding) return;
        channel.status = error.status === 403 ? 'denied' : 'error';
        this.emit();
      });
    }
    this.emit();
  }

  async replay(names, after = null) {
    if (!this.state.config) throw new ProbeError('请先获取配置。');
    if (after !== null && (!Number.isSafeInteger(after) || after < 0)) throw new ProbeError('补发起点必须是大于等于 0 的整数。');
    const jobs = names.map(async (name) => {
      const channel = this.channels.get(name);
      if (!channel || channel.replaying) return;
      const request = {};
      channel.request = request;
      channel.replaying = true;
      const session = this.session;
      const epoch = this.epoch;
      const current = () => this.session === session && this.epoch === epoch && channel.request === request;
      const from = after ?? channel.pendingAfter ?? channel.checkpoint;
      this.emit();
      try {
        const query = new URLSearchParams({ after_seq: String(from), channels: name });
        const response = await this.get(`/admin/realtime/replay?${query}`);
        if (!current()) return;
        if (response.code !== 0) throw new ProbeError(response.msg || '补发请求失败');
        const data = response.data;
        if (!data || !Number.isSafeInteger(data.last_seq) || !Array.isArray(data.events)) throw new ProbeError('补发响应格式不正确。');
        if (data.resync) {
          this.log('warning', `${name} 需要重新加载业务列表（${data.reason || 'history_unavailable'}）。本页继续检查实时消息。`);
        } else {
          for (const event of data.events) if (event.channel === name) this.receive(event, 'replay');
          this.log('success', `${name} 补发检查完成：起点 ${from}，返回 ${data.events.length} 条。`);
        }
        channel.checkpoint = Math.max(data.last_seq, channel.liveSeq);
        channel.pendingAfter = null;
      } catch (error) {
        if (!current()) return;
        this.log('error', `${name} 补发失败：${error.name === 'AbortError' ? '请求超时或已取消' : error.message}`);
      } finally {
        if (current()) { channel.replaying = false; channel.request = null; this.emit(); }
      }
    });
    await Promise.all(jobs);
  }

  receive(event, source) {
    if (!event || !Number.isSafeInteger(event.seq) || event.seq <= 0 || !this.state.config.channels.includes(event.channel)) {
      this.log('warning', '收到格式不符合预期的事件，未计入检测结果。');
      return;
    }
    const key = `${event.seq}:${event.channel}`;
    const duplicate = this.seen.has(key);
    if (duplicate) this.state.duplicateCount++;
    else {
      this.seen.add(key);
      if (this.seen.size > 5000) this.seen.delete(this.seen.values().next().value);
    }
    if (source === 'live') this.state.liveCount++;
    else this.state.replayCount++;
    this.state.lastSeq = Math.max(this.state.lastSeq, event.seq);
    this.onEvent({ source, duplicate, time: Date.now(), event: redact(event, this.token) });
    this.emit();
  }
}