|
|
@@ -0,0 +1,354 @@
|
|
|
+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();
|
|
|
+ }
|
|
|
+}
|