diff --git a/AGENTS.md b/AGENTS.md index 3724ef2..f75b287 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -56,9 +56,11 @@ 作品分析：左侧原“竞品分析”统一为“作品分析”，页面通过“竞品账号”“我的账号”两个 TAB 严格区分作品来源；各 TAB 独立保留筛选与分页。支持所属账号、发布时间和互动阈值筛选，以及发布时间、点赞、评论、分享、收藏、24h 点赞增量升降序；自有作品额外支持观看量，不展示“时间核验状态”筛选。默认最新发布在前，条件变化回到第一页，筛选与排序作用于全部已采集作品后再分页；缺失指标显示“—”，排序放末尾、阈值筛选不将其当作 0。未采集作品不参与分析。 -账号事件监听：仅支持抖音，自有账号默认关闭，在“我的账号”的“监听状态”列逐个开启，未登录环境不能开启。互动通知使用网页通知列表只读轮询，不再依赖私有 WebSocket SDK；每页 50 条，is_mark_read=0，不标记已读，原始 JSON 在 Python 解码以保留 64 位 ID。只记录点赞、评论、关注、转发，私信在独立收件箱同步与展示，不并入事件聚合，不自动互动。监听依赖已登录且运行中的浏览器；按账号固定页面并核验 UID，支持多标签页。分组读取位置、开启边界和事件在同一事务保存，提交成功后确认，重复投递去重；开关代次拒绝关闭期间的旧投递；首次开启、重新开启和服务重启均补齐平台仍可返回的全部历史和漏收记录，包括关闭期间通知，正常运行约每 5 分钟完整核对一次。开启时间只用于历史标记，不作为丢弃依据，补入保留平台原时间；不按头部位置或 100 页上限截断完整核对。开启设置不等于接收正常，状态应展示实际读取结果、最后成功时间及错误或可能遗漏。左侧“事件聚合”只展示当前已开启账号的已接收互动通知，按发生时间倒序，缺少发生时间置末尾，同时间按事件 ID 倒序，支持账号、类型、接收时间范围筛选和分页，每 5 秒刷新；关闭后保留历史但隐藏，重新开启后可查看。缺失发生时间不能用接收时间冒充；真实验收必须核对平台、数据库和页面同一通知，不把读取成功或平台合并后的重复互动当作新事件通过。 +账号事件监听：仅支持抖音，自有账号默认关闭，在“我的账号”的“监听状态”列逐个开启，未登录环境不能开启。互动通知复用已验证的抖音网页 NoticeFrontier 实时信号，不另建抖音私有协议客户端；实时信号立即读取详情，历史通知列表仅用于完整核对补漏。网关与后台在原端口通过 `/v1/channel` WS 共享连接推送多账号事件，保存成功后确认，断线重订阅并补送，不保留 HTTP GET 事件轮询回退。历史列表每页 50 条，is_mark_read=0，不标记已读，原始 JSON 在 Python 解码以保留 64 位 ID。只记录点赞、评论、关注、转发，私信在独立收件箱同步与展示，不并入事件聚合，不自动互动。监听依赖已登录且运行中的浏览器；按账号固定页面并核验 UID，支持多标签页。分组读取位置、开启边界和事件在同一事务保存，提交成功后确认，重复投递去重；开关代次拒绝关闭期间的旧投递；首次开启、重新开启和服务重启均补齐平台仍可返回的全部历史和漏收记录，包括关闭期间通知，正常运行约每 5 分钟完整核对一次。开启时间只用于历史标记，不作为丢弃依据，补入保留平台原时间；不按头部位置或 100 页上限截断完整核对。开启设置不等于接收正常，状态应展示实际读取结果、最后成功时间及错误或可能遗漏。左侧“事件聚合”只展示当前已开启账号的已接收互动通知，按发生时间倒序，缺少发生时间置末尾，同时间按事件 ID 倒序，支持账号、类型、接收时间范围筛选和分页，复用现有 creator SSE 通知即时刷新并明确显示断线；关闭后保留历史但隐藏，重新开启后可查看。缺失发生时间不能用接收时间冒充；真实验收必须核对平台、数据库和页面同一通知，不把读取成功或平台合并后的重复互动当作新事件通过。 -事件聚合资料：互动用户仅展示真实昵称，不展示 UID，以平台返回的 secUID 链接主页，不使用数字 UID 猜测主页；列表仅展示发生时间，不展示接收时间列，但保留按接收时间计算的范围筛选。事件类型使用彩色 Tag 区分：点赞粉色、评论蓝色、关注紫色、转发橙色；来源使用彩色 Tag 区分：同步金色、通知青色，保留文字说明。“账号”列置于第一列，列表及筛选选项仅显示昵称。内容列限定在表格宽度内，长文本自动换行并直接完整展示，不截断、不折叠、不提供“展开”入口，不得撑宽页面。对应作品展示 48px 小封面并链接抖音作品详情，图文使用 note 地址；缺封面保留作品入口，缺资料明确标示。通知中的作品封面复用 `<作者 UID>/<作品 ID>.<图片扩展名>` 本地缓存，不创建占位账号或作品，不回退展示远程图片。历史核对可补齐已保存事件的用户与作品资料，但只更新同一互动 UID、同一作品 ID 的资料，不改原事件时间、内容、历史标记或去重规则；封面下载失败单独记录和展示，不能阻断通知保存，也不能伪装成已缓存。 +网关通信方向：首期仅迁移事件推送与确认，其他 HTTP 业务通信保留；后续全部迁移到同一个 WS 通道，包括控制、状态及文件传输，每项迁移完成即删除对应 HTTP 路由，不保留兼容回退，最终仅保留标准 WS 握手入口、不增加端口。规划见 `docs/realtime-event-push-plan.md`。 + +事件聚合资料：互动用户仅展示真实昵称，不展示 UID，以平台返回的 secUID 链接主页，不使用数字 UID 猜测主页；列表仅展示发生时间，不展示接收时间列，但保留按接收时间计算的范围筛选。事件类型使用彩色 Tag 区分：点赞粉色、评论蓝色、关注紫色、转发橙色；来源使用彩色 Tag 区分：同步金色、通知青色，保留文字说明。“账号”列置于第一列，列表及筛选选项仅显示昵称。内容列限定在表格宽度内，长文本自动换行并直接完整展示，不截断、不折叠、不提供“展开”入口，不得撑宽页面。对应作品展示 48px 小封面并链接抖音作品详情，图文使用 note 地址；缺封面保留作品入口，缺资料明确标示。通知中的作品封面复用 `<作者 UID>/<作品 ID>.<图片扩展名>` 本地缓存，不创建占位账号或作品，不回退展示远程图片。历史核对可补齐已保存事件的用户与作品资料，但只更新同一互动 UID、同一作品 ID 的资料，不改原事件时间、内容、历史标记或去重规则；封面下载在事件保存后独立处理，不阻塞 WS 接收、保存或确认；只更新同一作品的封面错误并通知页面，不改变事件内容、时间、历史标记或检查点。封面下载失败单独记录和展示，不能阻断通知保存，也不能伪装成已缓存。 私信管理：仅支持已开启监听的自有抖音账号，独立聊天标签共享原浏览器资料，不干扰采集或主页面登录。展示聊天客户端已加载会话的最近 50 条消息，不宣称完整历史；只支持手动发送文字，不自动回复、不群发。聊天界面采用 Ant Design X Conversations、Bubble.List 和 Sender：左侧合并联系人并明确所属账号，同一联系人在不同账号下为独立会话，草稿按账号与联系人隔离。会话按最近一条有真实时间的收发消息倒序排列，无真实时间置末尾，不以保存时间或待发送请求的创建时间冒充；同时间按账号 ID、联系人 UID 固定排列，后端分页与前端合并使用相同规则。延迟返回的旧摘要不覆盖新摘要，刷新或加载更多不改变当前选中会话。会话列表与消息区联系人展示聊天客户端专用资料接口返回的真实昵称，支持普通用户与 AI 分身，不用普通主页接口代替、不用 UID 或会话标题占位；平台昵称本身为“用户＋数字”时保留原值，未获取昵称显示“昵称未获取”。昵称独立于新消息同步，已有会话可补齐及更新昵称；核验资料 UID 与会话对方一致，读取失败明确显示错误但不阻断消息保存，不改发送对象或聊天记录。专用标签从已登录的个人中心“消息”入口进入，沿用现有登录，不增加私信扫码；直接 /chat 的二维码不能作为账号未登录的依据。仅以聊天客户端的实际 UID 与已绑定 UID 一致确认私信身份，不创建账号或改变绑定。发送请求先持久化并按请求 ID 去重；超时、重启中断及 SDK 网络错误 1008 标记“结果未确认”，不自动重发，不伪造送达或已读。同步记录须持久化后推进检查点；关闭监听阻止旧代次写入。 diff --git a/browser_gateway/platform/notice_polling.py b/browser_gateway/platform/notice_polling.py index 570c5db..13cfc32 100644 --- a/browser_gateway/platform/notice_polling.py +++ b/browser_gateway/platform/notice_polling.py @@ -41,6 +41,7 @@ class NoticePollingSession: checkpoints = {} if checkpoints is None else checkpoints validate_checkpoint(boundary_at, checkpoints) self.browser, self.alias, self.uid = browser, alias, uid + self.session_id = uuid.uuid4().hex identity = browser.identity(alias, uid) self.boundary_at = boundary_at or identity.get('platform_now') self.boundary = boundary_time(self.boundary_at) @@ -48,6 +49,13 @@ class NoticePollingSession: self.checkpoints = dict(checkpoints) self.queue = deque() self.lock = threading.RLock() + self.capacity = threading.Condition(self.lock) + self.max_pending = 10000 + self.error = '' + self._checkpoint_dependencies = {} + self._stop = threading.Event() + self._changed = lambda: None + self._threads = [] self.stopped = False self.last_full_scan_at = None @@ -80,6 +88,8 @@ class NoticePollingSession: for group in GROUPS: cursor, found, head = 0, False, None while True: + if self._stop.is_set(): + return body = self._page(group, cursor) pages += 1 for raw in body['notice_list_v2']: @@ -131,7 +141,88 @@ class NoticePollingSession: LOG.info('notification scan alias=%s uid=%s mode=%s pages=%s events=%s ignored=%s gap_groups=%s', self.alias, self.uid, 'history_reconcile' if full_scan else 'incremental', pages, len(notices), dict(ignored), gaps) - self.queue.extend(result) + with self.lock: + self._checkpoint_dependencies[result[-1]['delivery_id']] = {d['delivery_id'] for d in result[:-1]} + # Historical backfill must leave room for live notifications. + history_limit = self.max_pending - min(500, self.max_pending // 2) + batch_size = min(100, history_limit) + for offset in range(0, len(result), batch_size): + batch = result[offset:offset + batch_size] + with self.capacity: + while len(self.queue) + len(batch) > history_limit and not self.stopped: + self.capacity.wait() + if self.stopped: + return + self.queue.extend(batch) + self._changed() + + def publish_notices(self, raws, received_at=None): + received_at = received_at or datetime.now(timezone.utc).isoformat() + result = [] + for raw in raws: + if not isinstance(raw, dict): + raise DouyinError('notification detail must be an object') + if not any(raw.get(k) for k in ('comment', 'follow', 'digg', 'share')): + LOG.info('non-interaction notification skipped alias=%s notice_id=%s', self.alias, raw.get('nid_str') or raw.get('nid')) + continue + if _notice_id(raw.get('user_id')) != self.uid: + raise DouyinError('notification identity changed') + timestamp = raw.get('create_time') + if type(timestamp) not in (int, float) or not math.isfinite(timestamp) or timestamp <= 0: + raise DouyinError('notification timestamp invalid') + result.append({'kind': 'notice', 'delivery_id': uuid.uuid4().hex, 'realtime_signal': True, + 'notice': normalize_notice(raw, received_at), + 'baseline': timestamp <= self.boundary.timestamp()}) + with self.lock: + if self.stopped: + raise DouyinError('notification stopped') + if len(self.queue) + len(result) > self.max_pending: + raise DouyinError('notification queue full; live delivery blocked') + # Live notices precede history that has not yet been sent. + self.queue.extendleft(reversed(result)) + if result: + LOG.info('notification signal alias=%s uid=%s received_at=%s events=%s', self.alias, self.uid, received_at, len(result)) + self._changed() + + def start(self, changed): + from .notice_signal import NoticeSignal + self._changed = changed + signal = NoticeSignal(self.browser, self.alias, self.uid) + def failed(exc): + if self._stop.is_set(): + return + with self.lock: + self.error = str(exc) + LOG.exception('notification receiver failed alias=%s uid=%s', self.alias, self.uid) + changed() + def receive(): + try: + while not self._stop.is_set(): + raws = signal.receive() + if raws: + self.publish_notices(raws, signal.received_at) + except Exception as exc: + failed(exc) + finally: + try: + signal.close() + except Exception: + LOG.exception('notification bridge cleanup failed alias=%s', self.alias) + def reconcile(): + try: + while not self._stop.is_set(): + self._scan() + # A new reconciliation cannot advance a previous unconfirmed one. + with self.capacity: + self.capacity.wait_for(lambda: self.stopped or not self._checkpoint_dependencies) + if self._stop.wait(HISTORY_RECONCILE_SECONDS): + break + except Exception as exc: + failed(exc) + for target in (receive, reconcile): + thread = threading.Thread(target=target, name='notice-' + self.alias, daemon=True) + self._threads.append(thread) + thread.start() def poll(self, limit, wait_seconds): # Pacing is owned by the control plane, outside the browser alias lock. @@ -151,18 +242,25 @@ class NoticePollingSession: with self.lock: for item in self.queue: if item['delivery_id'] in ids and item['kind'] == 'checkpoint': - if any(d['delivery_id'] not in ids for d in self.queue): + required = self._checkpoint_dependencies.get(item['delivery_id'], set()) + if any(d['delivery_id'] in required and d['delivery_id'] not in ids for d in self.queue): raise DouyinError('notification checkpoint acknowledged before its notices') + self._checkpoint_dependencies.pop(item['delivery_id'], None) self.checkpoints = dict(item['checkpoints']) self.initial = False if item['history_reconciled']: self.last_full_scan_at = time.monotonic() self.queue = deque(item for item in self.queue if item['delivery_id'] not in ids) + self.capacity.notify_all() + self._changed() def stop(self): - with self.lock: + self._stop.set() + with self.capacity: self.stopped = True self.queue.clear() + self.capacity.notify_all() + self._changed() class SubscriptionManager: @@ -170,17 +268,36 @@ class SubscriptionManager: self.browser = browser self._lock = threading.RLock() self._items = {} + self._lifecycle_locks = {} + self.changed = threading.Condition() + self.version = 0 - def start(self, alias, uid, boundary_at=None, checkpoints=None): + def notify(self): + with self.changed: + self.version += 1 + self.changed.notify_all() + + def _lifecycle_lock(self, alias): with self._lock: - previous = self._items.get(alias) + return self._lifecycle_locks.setdefault(alias, threading.RLock()) + + def start(self, alias, uid, boundary_at=None, checkpoints=None, generation=None): + # Browser initialization is per-account, never under the shared registry lock. + with self._lifecycle_lock(alias): + with self._lock: + previous = self._items.get(alias) if previous: previous.stop() - self._items.pop(alias) + with self._lock: + self._items.pop(alias, None) browser = self.browser._notification_browser(alias) item = NoticePollingSession(browser, alias, uid, boundary_at, checkpoints) - self._items[alias] = item - return {'connected': True, 'alias': alias, 'uid': uid, 'mode': 'notice_polling'} + item.generation = generation + item.start(self.notify) + with self._lock: + self._items[alias] = item + self.notify() + return {'connected': True, 'alias': alias, 'uid': uid, 'mode': 'notice_realtime', 'session_id': item.session_id} def _get(self, alias): with self._lock: @@ -195,12 +312,19 @@ class SubscriptionManager: def ack(self, alias, delivery_ids): self._get(alias).ack(delivery_ids) - def stop(self, alias): - with self._lock: - item = self._items.get(alias) + def stop(self, alias, generation=None, session_id=None): + with self._lifecycle_lock(alias): + with self._lock: + item = self._items.get(alias) + if item: + if generation is not None and item.generation != generation: + raise DouyinError('stale notification listener generation') + if session_id is not None and item.session_id != session_id: + raise DouyinError('stale notification listener session') if item: item.stop() - self._items.pop(alias) + with self._lock: + self._items.pop(alias, None) def close(self): with self._lock: diff --git a/browser_gateway/platform/notice_signal.py b/browser_gateway/platform/notice_signal.py new file mode 100644 index 0000000..af63843 --- /dev/null +++ b/browser_gateway/platform/notice_signal.py @@ -0,0 +1,116 @@ +"""Read-only bridge to the notification socket owned by the Douyin page.""" +import json +import logging +import uuid +from datetime import datetime, timezone + +LOG = logging.getLogger('browser-gateway') +from .douyin import DouyinError + +INSTALL = r"""(async()=>{ +const uid=UID,key=KEY,alias=ALIAS; +if(location.origin!=='https://www.douyin.com')throw Error('notification origin changed'); +const chunks=window.webpackChunkdouyin_web;if(!chunks)throw Error('notification runtime missing'); +let req;chunks.push([['notice-'+key],{},r=>{req=r}]);chunks.pop(); +const entries=Object.entries(req.m); +const entry=entries.find(([,f])=>String(f).includes('NOTICE_PUSH_EVENT_NAMES:function')); +const codec=entries.find(([,f])=>String(f).includes('.decodedFrame=')&&String(f).includes('.encodeFrame=')); +if(!entry||!codec)throw Error('notification SDK changed'); +const C=req(entry[0]).NoticeFrontier,decode=req(codec[0]).decodedFrame,f=C.frontierInstance; +if(!f||String(f._options.deviceID)!==uid)throw Error('notification identity changed'); +const registry=window.__creator_hub_notice_sessions||(window.__creator_hub_notice_sessions=Object.create(null)); +for(const [oldKey,old] of Object.entries(registry)){if(old.alias===alias){old.dispose();delete window[oldKey];delete registry[oldKey]}} +const s={queue:[],error:null,wake:null,C,f,uid,alias}; +const emit=e=>{if(s.queue.length>=1000)s.error='notification signal queue full';else s.queue.push(e);if(s.wake)s.wake()}; +const message=e=>{try{const v=decode(new Uint8Array(e.data));if(v.service===20313||v.service===20003)emit({kind:'push',signal_at:new Date().toISOString(),service:v.service,payload:new TextDecoder().decode(v.payload)})}catch(e){s.error='notification frame decode failed: '+String(e.message||e);if(s.wake)s.wake()}}; +const close=()=>emit({kind:'close'}); +f.addEventListener('message',message);f.addEventListener('close',close); +s.dispose=()=>{f.removeEventListener('message',message);f.removeEventListener('close',close);if(s.wake)s.wake()};window[key]=s;registry[key]=s; +if(f.readyState!==f.OPEN)await new Promise(resolve=>{const done=()=>{clearTimeout(timer);f.removeEventListener('open',done);resolve()};const timer=setTimeout(done,5000);f.addEventListener('open',done)}); +if(f.readyState!==f.OPEN||C.frontierInstance!==f){s.dispose();delete window[key];delete registry[key];throw Error('notification socket not connected')}; +return JSON.stringify({connected:true});})()""" +WAIT = r"""(async()=>{const s=window[KEY]; +if(!s||s.C.frontierInstance!==s.f||String(s.f._options.deviceID)!==s.uid)throw Error('notification identity or page changed'); +if(!s.queue.length&&!s.error)await new Promise(resolve=>{const done=()=>{clearTimeout(timer);s.wake=null;resolve()};const timer=setTimeout(done,2000);s.wake=done}); +if(s.error)throw Error(s.error);if(s.f.readyState!==s.f.OPEN)throw Error('notification socket disconnected'); +return JSON.stringify(s.queue.splice(0));})()""" + + +def notice_ids(event): + payload = json.loads(event['payload']) + if not isinstance(payload, dict): + raise DouyinError('notification push shape invalid') + if event['service'] == 20313: + notices = payload.get('notices', []) + if not isinstance(notices, list) or any(not isinstance(n, dict) or not isinstance(n.get('effect_groups', []), list) for n in notices): + raise DouyinError('notification push list invalid') + items = [n for n in notices if {str(g) for g in n.get('effect_groups', [])} & {'700','960','961'}] + elif event['service'] == 20003 and payload.get('notice_type') in (45,31,9009,9002,514,9067): + items = [payload] + else: + return [] + ids = [n.get('notice_id_str') for n in items] + if any(not isinstance(n, str) or not n.isascii() or not n.isdecimal() for n in ids): + raise DouyinError('notification push ID invalid') + return list(dict.fromkeys(ids)) + + +def detail_script(ids): + return r"""(async ids=>{ +const chunks=window.webpackChunkdouyin_web;let req;chunks.push([['notice-detail-'+Date.now()],{},r=>{req=r}]);chunks.pop(); +const entry=Object.entries(req.m).find(([,f])=>String(f).includes('getNoticeDetail:function')); +const sdk=entry&&req(entry[0]),client=window.axiosInstance; +if(!sdk||!client?.interceptors?.response)throw Error('notification detail SDK unavailable'); +const params={notice_ids:JSON.stringify(ids),is_mark_read:0};let raw,timer; +const observer=client.interceptors.response.use(r=>{const p=r.config?.params||{},u=new URL(r.config?.url||'',location.origin),xhr=r.request; +if(u.origin===location.origin&&u.pathname==='/aweme/v1/web/notice/detail/'&&p.notice_ids===params.notice_ids&&Number(p.is_mark_read)===0&&xhr&&(!xhr.responseType||xhr.responseType==='text')&&typeof xhr.responseText==='string')raw={status:r.status,body:xhr.responseText};return r}); +try{await Promise.race([sdk.getNoticeDetail(params),new Promise((_,reject)=>{timer=setTimeout(()=>reject(Error('notification detail timeout')),10000)})])} +finally{clearTimeout(timer);client.interceptors.response.eject(observer)} +if(!raw)throw Error('notification raw detail missing');return JSON.stringify(raw);})(IDS)""".replace('IDS', json.dumps(ids)) + + +class NoticeSignal: + def __init__(self, browser, alias, uid): + self.browser, self.alias, self.uid = browser, alias, uid + self.key = '__creator_notice_' + uuid.uuid4().hex + self.connection = browser._connect(alias) + try: + self.evaluate(INSTALL.replace('UID', json.dumps(uid)).replace('KEY', json.dumps(self.key)).replace('ALIAS', json.dumps(alias))) + except Exception: + self.connection.close() + raise + + def evaluate(self, expression): + raw = self.connection.evaluate(expression) + if not isinstance(raw, str): + raise DouyinError('notification bridge must return raw JSON') + return json.loads(raw) + + def receive(self): + events = self.evaluate(WAIT.replace('KEY', json.dumps(self.key))) + if not isinstance(events, list): + raise DouyinError('notification bridge result invalid') + ids = list(dict.fromkeys(n for event in events if event.get('kind') == 'push' for n in notice_ids(event))) + if any(event.get('kind') == 'close' for event in events): + raise DouyinError('notification socket disconnected') + if not ids: + return [] + self.received_at = datetime.now(timezone.utc).isoformat() + LOG.info('notification signal received alias=%s uid=%s signal_at=%s gateway_received_at=%s notice_ids=%s', + self.alias, self.uid, next((e.get('signal_at') for e in events if e.get('kind') == 'push'), None), self.received_at, ids) + response = self.evaluate(detail_script(ids)) + body = json.loads(response['body']) + raws = body.get('notice_list_v2') + if response['status'] != 200 or body.get('status_code') != 0 or not isinstance(raws, list): + raise DouyinError('notification detail request failed: HTTP %s, platform status %s' % (response['status'], body.get('status_code'))) + found = {str(n.get('nid_str') or n.get('nid')) for n in raws if isinstance(n, dict)} + if found != set(ids): + raise DouyinError('notification pushed detail incomplete: ' + ','.join(ids)) + return raws + + def close(self): + # Dispose page callbacks before closing; cleanup errors are observable. + try: + self.evaluate('(async()=>{const s=window['+json.dumps(self.key)+'];if(s){s.dispose();delete window['+json.dumps(self.key)+'];if(window.__creator_hub_notice_sessions)delete window.__creator_hub_notice_sessions['+json.dumps(self.key)+']}return "null"})()') + finally: + self.connection.close() diff --git a/browser_gateway/server/event_channel.py b/browser_gateway/server/event_channel.py new file mode 100644 index 0000000..209a89c --- /dev/null +++ b/browser_gateway/server/event_channel.py @@ -0,0 +1,135 @@ +"""One shared-port WebSocket carrying event subscriptions and acknowledgements.""" +import json +import logging +import threading +import time +from simple_websocket import Server, ConnectionClosed +from ..platform.douyin import DouyinError + +LOG = logging.getLogger('browser-gateway') + + +def serve_channel(handler): + from .http import RequestError + if handler.headers.get('Upgrade', '').lower() != 'websocket': + raise RequestError('event channel requires WebSocket upgrade', 400) + environ = {'werkzeug.socket': handler.connection} + # simple-websocket's documented socket environment adapter owns framing and handshake. + for name, value in handler.headers.items(): + environ['HTTP_' + name.upper().replace('-', '_')] = value + handler.connection.settimeout(None) + handler.close_connection = True + handler.websocket_handoff = True + ws = Server(environ, ping_interval=15, max_message_size=1024 * 1024) + handler.close_connection = True + manager = handler.server.gateway.subscriptions + gateway = handler.server.gateway + stopped = threading.Event() + lock = threading.RLock() + writing = threading.Lock() + subscriptions = {} + + def send(value): + with writing: + ws.send(json.dumps(value, ensure_ascii=False, separators=(',', ':'))) + + def close(): + try: + ws.close() + except ConnectionClosed: + LOG.debug('event WS already closed') + + def sender(): + try: + while not stopped.is_set(): + with manager.changed: + version = manager.version + with lock: + for alias, state in list(subscriptions.items()): + sub, item, sent = state + try: + current = manager._get(alias) + error = item.error or ('event subscription replaced or stopped' if current is not item or item.stopped else '') + except DouyinError as exc: + error = str(exc) + if error: + send({'type': 'error', 'subscription': sub, 'alias': alias, 'error': error}) + subscriptions.pop(alias) + continue + if sent: + continue + pending = [d for d in item.pending() if d['delivery_id'] not in sent][:100] + if pending: + started = time.monotonic() + send({'type': 'deliveries', 'subscription': sub, 'alias': alias, 'deliveries': pending}) + sent.update({d['delivery_id']: started for d in pending}) + LOG.info('event WS delivered alias=%s subscription=%s count=%s', alias, sub, len(pending)) + with manager.changed: + manager.changed.wait_for(lambda: stopped.is_set() or manager.version != version, timeout=15) + except Exception: + if not stopped.is_set(): + LOG.exception('event WS delivery failed') + try: + send({'type': 'error', 'error': 'event channel delivery failed'}) + except ConnectionClosed: + LOG.info('event WS disconnected during error reporting') + close() + stopped.set() + + thread = threading.Thread(target=sender, name='event-ws-delivery', daemon=True) + thread.start() + try: + while not stopped.is_set(): + raw = ws.receive() + if not isinstance(raw, str): + raise DouyinError('event channel requires text JSON') + value = json.loads(raw) + if not isinstance(value, dict) or value.get('type') not in ('subscribe', 'ack', 'unsubscribe'): + raise DouyinError('event channel message invalid') + alias, sub = value.get('alias'), value.get('subscription') + if not isinstance(alias, str) or not isinstance(sub, str) or not sub or len(sub) > 128: + raise DouyinError('event channel subscription invalid') + try: + # This is a metadata snapshot check, not a browser operation. + # Do not wait on the browser-operation lock in the shared receiver. + if value['type'] == 'subscribe': + gateway._require_douyin_generation(alias, value) + with lock: + if value['type'] == 'subscribe': + item = manager._get(alias) + if item.uid != value.get('uid') or item.stopped or getattr(item, 'generation', None) != value.get('listener_generation') or item.session_id != value.get('session_id'): + raise DouyinError('event channel identity changed') + subscriptions[alias] = (sub, item, {}) + send({'type': 'subscribed', 'alias': alias, 'subscription': sub}) + else: + state = subscriptions.get(alias) + if not state or state[0] != sub or manager._get(alias) is not state[1]: + raise DouyinError('stale event channel subscription') + if value['type'] == 'unsubscribe': + subscriptions.pop(alias) + else: + ids = value.get('delivery_ids') + if not isinstance(ids, list) or not ids or any(not isinstance(i, str) or i not in state[2] for i in ids): + raise DouyinError('event acknowledgement invalid') + state[1].ack(ids) + elapsed_ms = round((time.monotonic() - min(state[2][i] for i in ids)) * 1000, 3) + for delivery_id in ids: + state[2].pop(delivery_id) + LOG.info('event WS acknowledged alias=%s count=%s send_to_persist_ack_ms=%s', alias, len(ids), elapsed_ms) + except (DouyinError, RequestError) as exc: + LOG.warning('event WS subscription rejected alias=%s subscription=%s reason=%s', alias, sub, exc) + send({'type': 'error', 'subscription': sub, 'alias': alias, 'error': str(exc)}) + manager.notify() + except ConnectionClosed: + LOG.info('event WS disconnected subscriptions=%s', len(subscriptions)) + except Exception as exc: + LOG.exception('event WS request failed') + try: + send({'type': 'error', 'error': str(exc)}) + except ConnectionClosed: + LOG.info('event WS disconnected during error reporting') + finally: + stopped.set() + manager.notify() + thread.join(timeout=2) + close() diff --git a/browser_gateway/server/http.py b/browser_gateway/server/http.py index acff5eb..db1bbc9 100644 --- a/browser_gateway/server/http.py +++ b/browser_gateway/server/http.py @@ -512,7 +512,7 @@ class Gateway: self._require_douyin_generation(alias, input) # Polling-session initialization performs read-only page requests. # It must not occupy the manual-send slot while waiting on them. - result = self.subscriptions.start(alias, expected_uid, input.get("boundary_at"), input.get("notice_checkpoints", {})) + result = self.subscriptions.start(alias, expected_uid, input.get("boundary_at"), input.get("notice_checkpoints", {}), input.get('listener_generation')) self._require_douyin_generation(alias, input) return result except DouyinError as exc: @@ -522,37 +522,18 @@ class Gateway: ) raise RequestError(f"Douyin notification polling could not start: {exc}") from exc - def poll_douyin_events(self, alias: str, input: dict, query: dict) -> list[dict]: - if not valid_douyin_generation(input): - raise RequestError("invalid Douyin event request", 400) - try: - limit = int(query.get("limit", ["50"])[0]) - wait = float(query.get("wait", ["0"])[0]) - except (IndexError, TypeError, ValueError) as exc: - raise RequestError("invalid event poll options", 400) from exc - if not 1 <= limit <= 100 or not 0 <= wait <= 30: - raise RequestError("invalid event poll options", 400) - try: - with self._alias_lock(alias): - self._require_douyin_generation(alias, input) - acknowledgements = query.get("ack", []) - delivery_ids = [value for raw in acknowledgements for value in raw.split(",")] - self.subscriptions.ack(alias, delivery_ids) - return self.subscriptions.poll(alias, limit, 0) - except DouyinError as exc: - LOG.warning( - "Douyin event listener poll failed alias=%s reason=%s", - alias, - exc, - ) - raise RequestError(f"Douyin notification polling failed: {exc}") from exc - def stop_douyin_events(self, alias: str, input: dict) -> None: if not valid_douyin_generation(input): raise RequestError("invalid Douyin event request", 400) with self._alias_lock(alias): self._require_douyin_generation(alias, input) - self.subscriptions.stop(alias) + if input.get('stop_all') is True: + self.subscriptions.stop(alias) + else: + session_id = input.get('session_id') + if not isinstance(session_id, str) or not session_id: + raise RequestError('event listener session_id required', 400) + self.subscriptions.stop(alias, input.get('listener_generation'), session_id) class GatewayHTTPServer(ThreadingHTTPServer): daemon_threads = True @@ -635,9 +616,11 @@ class GatewayHandler(BaseHTTPRequestHandler): self._respond(401, json_bytes({"error": "gateway token rejected"})) return try: - needs_body = method in {"POST", "DELETE"} or ( - method == "GET" and parsed.path.endswith("/douyin/events") - ) + if parsed.path == '/v1/channel' and method == 'GET': + from .event_channel import serve_channel + serve_channel(self) + return + needs_body = method in {"POST", "DELETE"} body = self._body() if needs_body else {} result = self._route(method, parsed.path, parse_qs(parsed.query), body) if result is None: @@ -653,6 +636,10 @@ class GatewayHandler(BaseHTTPRequestHandler): self._handle_exception(parsed.path, exc) def _handle_exception(self, path: str, exc: Exception) -> None: + if getattr(self, 'websocket_handoff', False): + LOG.exception('gateway WebSocket failed after handoff', extra={'path': path}) + self.close_connection = True + return if isinstance(exc, RequestError): payload = {"error": str(exc)} if exc.network_id: @@ -721,8 +708,6 @@ class GatewayHandler(BaseHTTPRequestHandler): if action == "events": if method == "POST": return gateway.start_douyin_events(alias, body) - if method == "GET": - return gateway.poll_douyin_events(alias, body, query) if method == "DELETE": gateway.stop_douyin_events(alias, body) return None diff --git a/browser_gateway/test_event_channel.py b/browser_gateway/test_event_channel.py new file mode 100644 index 0000000..2b3e80b --- /dev/null +++ b/browser_gateway/test_event_channel.py @@ -0,0 +1,95 @@ +import json +import threading +import unittest +import urllib.request +import urllib.error +from types import SimpleNamespace +from unittest.mock import Mock +import websocket +from browser_gateway.server.http import GatewayHTTPServer, GatewayHandler +from browser_gateway.platform.notice_polling import SubscriptionManager, NoticePollingSession +from browser_gateway.test_notice_polling import BOUNDARY, notice + +class EventChannelTest(unittest.TestCase): + def setUp(self): + self.manager = SubscriptionManager(Mock()) + self.sessions = {} + for alias in ('one','two'): + item = NoticePollingSession(Mock(), alias, '123', BOUNDARY, {}) + self.sessions[alias] = item + self.manager._items[alias] = item + item._changed = self.manager.notify + gateway = SimpleNamespace(token='test', subscriptions=self.manager, + _alias_lock=lambda alias:threading.RLock(), _require_douyin_generation=lambda alias,value:None, + runtimes=SimpleNamespace(close=lambda:None)) + self.server = GatewayHTTPServer(('127.0.0.1',0), gateway) + self.server.gateway = gateway + self.thread = threading.Thread(target=self.server.serve_forever,daemon=True); self.thread.start() + self.ws = websocket.create_connection('ws://127.0.0.1:%s/v1/channel'%self.server.server_port, + header=['Authorization: Bearer test'],timeout=2) + def tearDown(self): + self.ws.close();self.server.shutdown();self.server.server_close();self.thread.join(2) + def send(self,kind,alias='one',**values): + self.ws.send(json.dumps({'type':kind,'alias':alias,'subscription':alias,'uid':'123','session_id':self.sessions[alias].session_id,**values})) + def read(self):return json.loads(self.ws.recv()) + def test_shared_port_two_accounts_live_delivery_and_ack(self): + for alias in ('one','two'): + self.send('subscribe',alias);self.assertEqual(self.read()['type'],'subscribed') + self.sessions['two'].publish_notices([notice('9007199254740993123')]) + message=self.read();self.assertEqual(message['alias'],'two') + delivery=message['deliveries'][0] + self.assertEqual(delivery['notice']['event_key'],'9007199254740993123') + self.send('ack','two',delivery_ids=[delivery['delivery_id']]) + self.send('unsubscribe','one');self.send('subscribe','one');self.assertEqual(self.read()['type'],'subscribed') + self.assertEqual(self.sessions['two'].pending(),[]) + def test_unconfirmed_batch_is_replayed_after_reconnect(self): + self.send('subscribe');self.read();self.sessions['one'].publish_notices([notice('12')]);first=self.read() + self.ws.close() + self.ws=websocket.create_connection('ws://127.0.0.1:%s/v1/channel'%self.server.server_port, + header=['Authorization: Bearer test'],timeout=2) + self.send('subscribe');self.read();second=self.read() + self.assertEqual(first['deliveries'],second['deliveries']) + def test_plain_http_request_requires_upgrade(self): + request=urllib.request.Request('http://127.0.0.1:%s/v1/channel'%self.server.server_port,headers={'Authorization':'Bearer test'}) + with self.assertRaises(urllib.error.HTTPError) as failure: + urllib.request.urlopen(request,timeout=2) + self.assertEqual(failure.exception.code,400) + + def test_busy_browser_operation_does_not_delay_channel_subscription(self): + lock = threading.Lock();lock.acquire() + self.server.gateway._alias_lock = lambda alias:lock + try: + self.send('subscribe');self.assertEqual(self.read()['type'],'subscribed') + finally: + lock.release() + + def test_waiting_channel_does_not_hold_alias_lock(self): + lock = threading.Lock() + self.server.gateway._alias_lock = lambda alias:lock + self.send('subscribe');self.read() + self.assertTrue(lock.acquire(timeout=.1));lock.release() + + def test_one_stopped_account_does_not_disconnect_another(self): + self.send('subscribe','one');self.read();self.send('subscribe','two');self.read() + self.sessions['one'].stop() + self.assertEqual(self.read()['type'],'error') + self.sessions['two'].publish_notices([notice('22')]) + self.assertEqual(self.read()['alias'],'two') + + def test_old_session_is_rejected_after_same_generation_restart(self): + self.send('subscribe',session_id='previous-session') + self.assertEqual(self.read()['type'],'error') + + def test_uid_mismatch_is_rejected_without_disconnect_for_other_accounts(self): + self.send('subscribe','two');self.assertEqual(self.read()['type'],'subscribed') + self.send('subscribe',uid='999') + self.assertEqual(self.read()['type'],'error') + self.sessions['two'].publish_notices([notice('23')]) + self.assertEqual(self.read()['alias'],'two') + + def test_invalid_ack_is_reported_not_silently_removed(self): + self.send('subscribe');self.read();self.sessions['one'].publish_notices([notice('1')]);self.read() + self.send('ack',delivery_ids=['not-delivered']);self.assertEqual(self.read()['type'],'error') + self.assertEqual(len(self.sessions['one'].pending()),1) + +if __name__=='__main__':unittest.main() diff --git a/browser_gateway/test_event_listener.py b/browser_gateway/test_event_listener.py index 081ab36..c2285a3 100644 --- a/browser_gateway/test_event_listener.py +++ b/browser_gateway/test_event_listener.py @@ -56,7 +56,7 @@ class EventListenerTests(unittest.TestCase): manager.start("browser", "123") self.assertIs(manager._items["browser"], subscription) - def test_gateway_listener_contract_uses_post_get_delete(self): + def test_gateway_listener_contract_removes_http_event_polling(self): gateway = Mock() gateway.start_douyin_events.return_value = {"connected": True, "alias": "browser", "uid": "123"} gateway.poll_douyin_events.return_value = [{"kind": "open", "delivery_id": "1"}] @@ -69,8 +69,9 @@ class EventListenerTests(unittest.TestCase): payload = {"expected_uid": "123", "binding_version": 7, "runtime_id": "runtime", "network_id": "network", "network_exit_id": ""} self.assertTrue(handler._route("POST", route, {}, payload)["connected"]) query = {"ack": ["1"], "wait": ["5"], "limit": ["100"]} - self.assertIsInstance(handler._route("GET", route, query, payload), list) - gateway.poll_douyin_events.assert_called_once_with("browser", payload, query) + with self.assertRaisesRegex(Exception, 'not found'): + handler._route("GET", route, query, payload) + gateway.poll_douyin_events.assert_not_called() self.assertIsNone(handler._route("DELETE", route, {}, payload)) diff --git a/browser_gateway/test_event_push.py b/browser_gateway/test_event_push.py new file mode 100644 index 0000000..8544e35 --- /dev/null +++ b/browser_gateway/test_event_push.py @@ -0,0 +1,115 @@ +import unittest +import threading +from unittest.mock import Mock, patch +from browser_gateway.platform.notice_polling import NoticePollingSession +from browser_gateway.test_notice_polling import page, notice, BOUNDARY + +class EventPushTest(unittest.TestCase): + def session(self): + browser = Mock() + browser.identity.return_value = {'status': 'verified'} + browser.get.return_value = page([notice('21')]) + return NoticePollingSession(browser, 'safe', '123', BOUNDARY, {}) + + def test_stale_stop_cannot_stop_a_new_receiver_of_the_same_generation(self): + from browser_gateway.platform.notice_polling import SubscriptionManager + manager=SubscriptionManager(Mock());session=self.session();session.generation='same' + manager._items['safe']=session + with self.assertRaisesRegex(Exception,'stale notification listener session'): + manager.stop('safe','same','previous-session') + self.assertFalse(session.stopped) + manager.stop('safe','same',session.session_id);self.assertTrue(session.stopped) + + @patch('browser_gateway.platform.notice_polling.NoticePollingSession') + def test_one_account_start_does_not_block_other_account_registry_reads(self, session_class): + from browser_gateway.platform.notice_polling import SubscriptionManager + manager=SubscriptionManager(Mock());manager._items['other']=object() + entered,release,read=threading.Event(),threading.Event(),threading.Event() + new=session_class.return_value;new.session_id='new';new.start.side_effect=lambda changed:(entered.set(),release.wait(1)) + starter=threading.Thread(target=lambda:manager.start('safe','123',BOUNDARY,{}),daemon=True);starter.start() + try: + self.assertTrue(entered.wait(1)) + reader=threading.Thread(target=lambda:(manager._get('other'),read.set()),daemon=True);reader.start() + self.assertTrue(read.wait(.1),'account startup held the shared registry lock') + finally: + release.set();starter.join(1);reader.join(1) + + def test_signal_is_queued_without_history_scan(self): + session = self.session() + session.publish_notices([notice('9007199254740993123')]) + self.assertEqual(session.pending()[0]['notice']['event_key'], '9007199254740993123') + session.browser.get.assert_not_called() + + def test_later_realtime_notice_does_not_block_history_checkpoint(self): + session = self.session() + history = session.poll(100, 0) + session.publish_notices([notice('99')]) + session.ack([item['delivery_id'] for item in history]) + self.assertEqual(session.checkpoints, {'700':'21','960':'21','961':'21'}) + self.assertEqual([item['notice']['event_key'] for item in session.pending()], ['99']) + + def test_bad_identity_does_not_queue_or_confirm(self): + session = self.session() + raw = notice('1'); raw['user_id'] = '999' + with self.assertRaisesRegex(Exception, 'identity'): + session.publish_notices([raw]) + self.assertEqual(session.pending(), []) + + def test_history_backpressure_reserves_room_for_live_notifications(self): + session = self.session(); session.max_pending = 4 + changed = threading.Event(); session._changed = changed.set + thread = threading.Thread(target=session._scan, daemon=True); thread.start() + try: + self.assertTrue(changed.wait(1)) + history = session.pending(); self.assertEqual(len(history),2) + session.publish_notices([notice('99')]) + self.assertEqual(session.pending()[0]['notice']['event_key'],'99') + session.ack([d['delivery_id'] for d in history]) + session.ack([d['delivery_id'] for d in session.pending() if d.get('notice',{}).get('event_key')=='99']) + thread.join(1); self.assertFalse(thread.is_alive()) + session.ack([d['delivery_id'] for d in session.pending()]) + self.assertEqual(session.checkpoints, {'700':'21','960':'21','961':'21'}) + finally: + session.stop();thread.join(1) + + @patch('browser_gateway.platform.notice_signal.NoticeSignal') + def test_receiver_continues_while_history_is_blocked_and_stops_cleanly(self, bridge): + session = self.session();started = threading.Event();received = threading.Event() + signal = bridge.return_value;signal.received_at = BOUNDARY + first = True + def receive(): + nonlocal first + if first: + first=False;started.wait(1);return [notice('99')] + session._stop.wait();return [] + signal.receive.side_effect=receive + session._scan=lambda:(started.set(),session._stop.wait()) + session.start(received.set) + try: + self.assertTrue(received.wait(1));self.assertEqual(session.pending()[0]['notice']['event_key'],'99') + finally: + session.stop() + for thread in session._threads:thread.join(1);self.assertFalse(thread.is_alive()) + signal.close.assert_called_once() + + @patch('browser_gateway.platform.notice_signal.NoticeSignal') + def test_receiver_error_is_visible_without_fabricating_notifications(self, bridge): + session = self.session();changed = threading.Event() + bridge.return_value.receive.side_effect=RuntimeError('signal failed') + session._scan=lambda:None + session.start(changed.set) + try: + self.assertTrue(changed.wait(1));self.assertEqual(session.error,'signal failed');self.assertEqual(session.pending(),[]) + finally: + session.stop() + for thread in session._threads:thread.join(1) + + def test_overload_is_explicit_and_does_not_drop_pending_events(self): + session = self.session() + session.max_pending = 1 + session.publish_notices([notice('1')]) + with self.assertRaisesRegex(Exception, 'queue'): + session.publish_notices([notice('2')]) + self.assertEqual(session.pending()[0]['notice']['event_key'], '1') + +if __name__ == '__main__': unittest.main() diff --git a/browser_gateway/test_gateway.py b/browser_gateway/test_gateway.py index 956e8a8..5e14f26 100644 --- a/browser_gateway/test_gateway.py +++ b/browser_gateway/test_gateway.py @@ -297,9 +297,8 @@ class GatewayValidationTests(unittest.TestCase): {"status": "succeeded"}, ) gateway.douyin_message_history.assert_called_once_with("safe", {}) - self.assertEqual( - handler._route("GET", "/v1/browsers/safe/douyin/events", {}, {}), [] - ) + with self.assertRaises(RequestError): + handler._route("GET", "/v1/browsers/safe/douyin/events", {}, {}) handler._route("POST", "/v1/browsers/safe/douyin/events", {}, {}) handler._route("DELETE", "/v1/browsers/safe/douyin/events", {}, {}) with self.assertRaises(RequestError): @@ -319,7 +318,7 @@ class GatewayValidationTests(unittest.TestCase): handler._handle_exception("/v1", ValueError("bad")) self.assertEqual(handler._respond.call_count, 2) - def test_get_event_route_reads_generation_body(self) -> None: + def test_removed_get_event_route_does_not_read_generation_body(self) -> None: handler = gateway_module.GatewayHandler.__new__(gateway_module.GatewayHandler) cast(Any, handler).path = "/v1/browsers/safe/douyin/events?wait=1" cast(Any, handler).headers = {"Content-Length": "67"} @@ -334,7 +333,7 @@ class GatewayValidationTests(unittest.TestCase): "GET", "/v1/browsers/safe/douyin/events", {"wait": ["1"]}, - {"binding_version": 1, "runtime_id": "runtime", "network_id": "network"}, + {}, ) def test_validation_boundaries(self) -> None: @@ -545,11 +544,7 @@ class GatewayBusinessMethodTests(unittest.TestCase): ), {"status": "started"}, ) - self.assertEqual( - self.gateway.poll_douyin_events("safe", self.generation, {"limit": ["1"], "wait": ["0"]}), - [], - ) - self.gateway.stop_douyin_events("safe", self.generation) + self.gateway.stop_douyin_events("safe", {**self.generation, 'session_id': 'test-session'}) def test_http_info_route_and_action_error_ownership(self) -> None: handler = gateway_module.GatewayHandler.__new__(gateway_module.GatewayHandler) diff --git a/browser_gateway/test_notice_polling.py b/browser_gateway/test_notice_polling.py index 2f95a5b..fea9c7c 100644 --- a/browser_gateway/test_notice_polling.py +++ b/browser_gateway/test_notice_polling.py @@ -172,7 +172,8 @@ class NoticePollingTests(unittest.TestCase): with self.subTest(boundary=boundary, checkpoints=checkpoints), self.assertRaises(DouyinError): NoticePollingSession(self.browser(), 'safe', '123', boundary, checkpoints) - def test_stop_prevents_further_reads_and_manager_restarts_from_persisted_state(self): + @patch('browser_gateway.platform.notice_polling.NoticePollingSession.start') + def test_stop_prevents_further_reads_and_manager_restarts_from_persisted_state(self, start): browser = self.browser() manager = SubscriptionManager(browser) self.assertTrue(manager.start('safe', '123', BOUNDARY, {'960': '9'})['connected']) diff --git a/browser_gateway/test_notice_signal.py b/browser_gateway/test_notice_signal.py new file mode 100644 index 0000000..1c4a7b7 --- /dev/null +++ b/browser_gateway/test_notice_signal.py @@ -0,0 +1,42 @@ +import json +import unittest +from unittest.mock import Mock +from browser_gateway.platform.notice_signal import NoticeSignal, notice_ids, detail_script +from browser_gateway.platform.douyin import DouyinError + +class NoticeSignalTest(unittest.TestCase): + def test_ids_remain_exact_and_unknown_groups_are_ignored(self): + event={'service':20313,'payload':json.dumps({'notices':[{'effect_groups':[700],'notice_id_str':'9007199254740993123'},{'effect_groups':[3],'notice_id_str':'7'}]})} + self.assertEqual(notice_ids(event),['9007199254740993123']) + self.assertEqual(notice_ids({'service':20003,'payload':'{"notice_type":45,"notice_id_str":"8"}'}),['8']) + self.assertEqual(notice_ids({'service':20003,'payload':'{"notice_type":0}'}),[]) + def test_bad_payload_or_id_fails(self): + for payload in ('[]','{"notices":"bad"}','{"notices":[{"effect_groups":[700],"notice_id_str":2}]}'): + with self.assertRaises(DouyinError):notice_ids({'service':20313,'payload':payload}) + def bridge(self, responses): + browser=Mock();connection=browser._connect.return_value + connection.evaluate.side_effect=[json.dumps(v) for v in responses] + return NoticeSignal(browser,'safe','123'),connection + def test_empty_heartbeat_does_not_fetch_notification_api(self): + bridge,conn=self.bridge([{'connected':True},[],None]);self.assertEqual(bridge.receive(),[]) + bridge.close();self.assertEqual(conn.evaluate.call_count,3);conn.close.assert_called_once() + def test_details_are_decoded_in_python_with_no_numeric_rounding(self): + nid=9007199254740993123 + push={'kind':'push','service':20313,'payload':json.dumps({'notices':[{'effect_groups':[700],'notice_id_str':str(nid)}]})} + bridge,conn=self.bridge([{'connected':True},[push],{'status':200,'body':json.dumps({'status_code':0,'notice_list_v2':[{'nid':nid}]})},None]) + self.assertEqual(bridge.receive()[0]['nid'],nid);bridge.close() + self.assertIn('is_mark_read:0',detail_script([str(nid)]));self.assertIn('eject(observer)',detail_script([str(nid)])) + def test_closed_signal_and_incomplete_details_fail_explicitly(self): + bridge,conn=self.bridge([{'connected':True},[{'kind':'close'}]]) + with self.assertRaisesRegex(DouyinError,'disconnected'):bridge.receive() + conn.close() + push={'kind':'push','service':20313,'payload':'{"notices":[{"effect_groups":[700],"notice_id_str":"9"}]}'} + bridge,conn=self.bridge([{'connected':True},[push],{'status':200,'body':'{"status_code":0,"notice_list_v2":[]}'}]) + with self.assertRaisesRegex(DouyinError,'incomplete'):bridge.receive() + conn.close() + def test_install_failure_closes_connection(self): + browser=Mock();browser._connect.return_value.evaluate.return_value=42 + with self.assertRaisesRegex(DouyinError,'raw JSON'):NoticeSignal(browser,'safe','123') + browser._connect.return_value.close.assert_called_once() + +if __name__=='__main__':unittest.main() diff --git a/browser_gateway/test_private_messages.py b/browser_gateway/test_private_messages.py index 045402f..1245baa 100644 --- a/browser_gateway/test_private_messages.py +++ b/browser_gateway/test_private_messages.py @@ -74,16 +74,6 @@ assert.equal((await vm.runInNewContext(SOURCE,context)).ready,true);assert.equal for action in ['private-login-qr','private-login-verify']: with self.assertRaises(RequestError):handler._route('POST','/v1/browsers/account/douyin/'+action,{}, {'expected_uid':'123'}) - def test_listener_pacing_does_not_wait_while_holding_alias_lock(self): - gateway = object.__new__(Gateway) - gateway._alias_lock = lambda alias: threading.Lock() - gateway._require_douyin_generation = Mock() - gateway.subscriptions = Mock() - gateway.subscriptions.poll.return_value = [] - payload = {'runtime_id':'a'*64, 'network_id':'native-'+'b'*32, 'binding_version':1} - self.assertEqual(gateway.poll_douyin_events('account', payload, {'wait':['25']}), []) - gateway.subscriptions.poll.assert_called_once_with('account', 50, 0) - def test_listener_restart_validates_checkpoint_before_touching_browser(self): gateway = object.__new__(Gateway) gateway.subscriptions = Mock() diff --git a/go.mod b/go.mod index 9f962a4..a52e81d 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26 require ( github.com/gofiber/fiber/v3 v3.5.0 + github.com/gorilla/websocket v1.5.3 github.com/jackc/pgx/v5 v5.10.0 github.com/sirupsen/logrus v1.10.0 github.com/spf13/cobra v1.10.2 diff --git a/go.sum b/go.sum index f13eae1..a038df9 100644 --- a/go.sum +++ b/go.sum @@ -22,6 +22,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= diff --git a/internal/controlplane/api/event_channel.go b/internal/controlplane/api/event_channel.go new file mode 100644 index 0000000..71a85eb --- /dev/null +++ b/internal/controlplane/api/event_channel.go @@ -0,0 +1,232 @@ +package api + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/url" + "strings" + "sync" + "sync/atomic" + "time" + + "git.ipao.vip/rogee/creator-hub/internal/creator" + "git.ipao.vip/rogee/creator-hub/internal/environment" + "github.com/gorilla/websocket" +) + +type channelMessage struct { + Type string `json:"type"` + Subscription string `json:"subscription"` + Error string `json:"error"` + Deliveries []creator.ListenerDelivery `json:"deliveries"` +} +type eventSubscription struct { + hub *gatewayEventChannel + id, alias string + messages chan channelMessage +} +type gatewayEventChannel struct { + target environment.Gateway + once sync.Once + connectErr error + conn *websocket.Conn + write sync.Mutex + mu sync.Mutex + subs map[string]*eventSubscription + done chan struct{} + finish sync.Once +} + +var eventChannels = struct { + sync.Mutex + items map[string]*gatewayEventChannel +}{items: map[string]*gatewayEventChannel{}} +var subscriptionSequence atomic.Uint64 + +func acquireEventSubscription(ctx context.Context, target environment.Gateway, params map[string]any) (*eventSubscription, error) { + key := target.Endpoint + "\n" + target.Token + eventChannels.Lock() + hub := eventChannels.items[key] + if hub != nil { + select { + case <-hub.done: + hub = nil + default: + } + } + if hub == nil { + hub = &gatewayEventChannel{target: target, subs: map[string]*eventSubscription{}, done: make(chan struct{})} + eventChannels.items[key] = hub + } + alias, _ := params["alias"].(string) + sub := &eventSubscription{hub: hub, id: fmt.Sprint(subscriptionSequence.Add(1)), alias: alias, messages: make(chan channelMessage, 2)} + // Reserve the subscription before releasing the pool lock, so closing the + // previous last account cannot retire a channel another account is joining. + hub.mu.Lock() + hub.subs[sub.id] = sub + hub.mu.Unlock() + eventChannels.Unlock() + hub.once.Do(func() { + u, err := url.Parse(strings.TrimRight(target.Endpoint, "/") + "/v1/channel") + if err != nil { + hub.connectErr = err + hub.fail(err) + return + } + switch u.Scheme { + case "http": + u.Scheme = "ws" + case "https": + u.Scheme = "wss" + default: + hub.connectErr = errors.New("invalid gateway scheme") + hub.fail(hub.connectErr) + return + } + headers := http.Header{"Authorization": []string{"Bearer " + target.Token}} + dialer := websocket.Dialer{HandshakeTimeout: 10 * time.Second} + hub.conn, _, hub.connectErr = dialer.DialContext(ctx, u.String(), headers) + if hub.connectErr != nil { + hub.fail(hub.connectErr) + return + } + hub.conn.SetReadLimit(16 << 20) + hub.conn.SetPingHandler(func(data string) error { + _ = hub.conn.SetReadDeadline(time.Now().Add(45 * time.Second)) + return hub.conn.WriteControl(websocket.PongMessage, []byte(data), time.Now().Add(5*time.Second)) + }) + go hub.receive() + }) + if hub.connectErr != nil { + sub.close() + return nil, fmt.Errorf("gateway WS connection: %w", hub.connectErr) + } + value := make(map[string]any, len(params)+2) + for k, v := range params { + value[k] = v + } + value["type"] = "subscribe" + value["subscription"] = sub.id + if err := hub.send(value); err != nil { + sub.close() + return nil, err + } + select { + case message := <-sub.messages: + if message.Type != "subscribed" { + sub.close() + return nil, fmt.Errorf("gateway WS subscription: %s", message.Error) + } + return sub, nil + case <-hub.done: + sub.close() + return nil, errors.New("gateway WS disconnected") + case <-ctx.Done(): + sub.close() + return nil, ctx.Err() + } +} +func (h *gatewayEventChannel) send(value any) error { + h.write.Lock() + defer h.write.Unlock() + select { + case <-h.done: + return errors.New("gateway WS disconnected") + default: + } + if err := h.conn.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil { + return err + } + if err := h.conn.WriteJSON(value); err != nil { + h.fail(err) + return err + } + return nil +} +func (h *gatewayEventChannel) receive() { + for { + _ = h.conn.SetReadDeadline(time.Now().Add(45 * time.Second)) + var message channelMessage + if err := h.conn.ReadJSON(&message); err != nil { + h.fail(err) + return + } + if message.Type != "subscribed" && message.Type != "deliveries" && message.Type != "error" { + h.fail(errors.New("invalid gateway WS message")) + return + } + if message.Type == "error" && message.Subscription == "" { + h.fail(errors.New(message.Error)) + return + } + h.mu.Lock() + sub := h.subs[message.Subscription] + h.mu.Unlock() + if sub == nil { + continue + } // A closed generation cannot receive delayed replies. + select { + case sub.messages <- message: + default: + h.fail(errors.New("gateway WS subscription queue full")) + return + } + } +} +func (h *gatewayEventChannel) fail(err error) { + h.finish.Do(func() { + close(h.done) + if h.conn != nil { + _ = h.conn.Close() + } + h.mu.Lock() + defer h.mu.Unlock() + for _, sub := range h.subs { + select { + case sub.messages <- channelMessage{Type: "error", Error: err.Error()}: + default: + } + } + }) +} +func (s *eventSubscription) poll(ctx context.Context, acks []string) ([]creator.ListenerDelivery, error) { + if len(acks) > 0 { + if err := s.hub.send(map[string]any{"type": "ack", "alias": s.alias, "subscription": s.id, "delivery_ids": acks}); err != nil { + return nil, err + } + } + select { + case message := <-s.messages: + if message.Type != "deliveries" || len(message.Deliveries) == 0 { + return nil, fmt.Errorf("gateway event delivery: %s", message.Error) + } + for _, d := range message.Deliveries { + if err := d.Validate(); err != nil { + return nil, err + } + } + return message.Deliveries, nil + case <-s.hub.done: + return nil, errors.New("gateway WS disconnected") + case <-ctx.Done(): + return nil, ctx.Err() + } +} +func (s *eventSubscription) close() { + _ = s.hub.send(map[string]any{"type": "unsubscribe", "alias": s.alias, "subscription": s.id}) + key := s.hub.target.Endpoint + "\n" + s.hub.target.Token + eventChannels.Lock() + s.hub.mu.Lock() + delete(s.hub.subs, s.id) + empty := len(s.hub.subs) == 0 + if empty && eventChannels.items[key] == s.hub { + delete(eventChannels.items, key) + } + s.hub.mu.Unlock() + eventChannels.Unlock() + if empty { + s.hub.fail(errors.New("gateway event channel stopped")) + } +} diff --git a/internal/controlplane/api/event_channel_test.go b/internal/controlplane/api/event_channel_test.go new file mode 100644 index 0000000..5efa453 --- /dev/null +++ b/internal/controlplane/api/event_channel_test.go @@ -0,0 +1,179 @@ +package api + +import ( + "context" + "errors" + "git.ipao.vip/rogee/creator-hub/internal/environment" + "github.com/gorilla/websocket" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func TestEventChannelSharesConnectionAndAcknowledges(t *testing.T) { + var connections atomic.Int32 + acks := make(chan struct{}, 2) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/channel" || r.Header.Get("Authorization") != "Bearer test" { + t.Errorf("unexpected channel request") + w.WriteHeader(400) + return + } + c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if err != nil { + t.Error(err) + return + } + defer c.Close() + connections.Add(1) + for { + var v map[string]any + if c.ReadJSON(&v) != nil { + return + } + switch v["type"] { + case "subscribe": + _ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": v["subscription"]}) + _ = c.WriteJSON(map[string]any{"type": "deliveries", "subscription": v["subscription"], "deliveries": []any{map[string]any{"kind": "open", "delivery_id": v["alias"]}}}) + case "ack": + acks <- struct{}{} + } + } + })) + defer server.Close() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + target := environment.Gateway{Endpoint: server.URL, Token: "test"} + first, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "a"}) + if err != nil { + t.Fatal(err) + } + defer first.close() + second, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "b"}) + if err != nil { + t.Fatal(err) + } + defer second.close() + if connections.Load() != 1 { + t.Fatalf("connections=%d", connections.Load()) + } + items, err := first.poll(ctx, nil) + if err != nil || items[0].DeliveryID != "a" { + t.Fatalf("%+v %v", items, err) + } + items, err = second.poll(ctx, nil) + if err != nil || items[0].DeliveryID != "b" { + t.Fatalf("%+v %v", items, err) + } + short, c := context.WithTimeout(ctx, 30*time.Millisecond) + defer c() + _, _ = first.poll(short, []string{"a"}) + select { + case <-acks: + case <-ctx.Done(): + t.Fatal("ACK not received") + } +} +func TestEventChannelRejectsHandshakeAndSubscriptionTimeout(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(503) })) + defer server.Close() + for _, endpoint := range []string{server.URL, "invalid://gateway", "http://%"} { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + _, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: endpoint}, map[string]any{"alias": "a"}) + cancel() + if err == nil { + t.Fatalf("invalid connection accepted: %s", endpoint) + } + } + waiting := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, e := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if e != nil { + return + } + defer c.Close() + for { + var value map[string]any + if c.ReadJSON(&value) != nil { + return + } + } + })) + defer waiting.Close() + ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond) + defer cancel() + _, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: waiting.URL}, map[string]any{"alias": "a"}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("subscription wait did not honor cancellation: %v", err) + } +} + +func TestEventChannelReportsDisconnect(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, e := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if e != nil { + return + } + defer c.Close() + var v map[string]any + _ = c.ReadJSON(&v) + _ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": v["subscription"]}) + })) + defer server.Close() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + sub, err := acquireEventSubscription(ctx, environment.Gateway{Endpoint: server.URL}, map[string]any{"alias": "a"}) + if err == nil { + defer sub.close() + _, err = sub.poll(ctx, nil) + } + if err == nil { + t.Fatal("disconnect hidden") + } +} + +func TestEventChannelClosingLastAccountDoesNotRetireJoiningAccount(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if err != nil { + return + } + defer c.Close() + for { + var value map[string]any + if c.ReadJSON(&value) != nil { + return + } + if value["type"] == "subscribe" { + _ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": value["subscription"]}) + } + } + })) + defer server.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + target := environment.Gateway{Endpoint: server.URL} + previous, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "a"}) + if err != nil { + t.Fatal(err) + } + defer func() { previous.close() }() + for i := 0; i < 50; i++ { + type result struct { + sub *eventSubscription + err error + } + joined := make(chan result, 1) + go func() { + sub, err := acquireEventSubscription(ctx, target, map[string]any{"alias": "b"}) + joined <- result{sub, err} + }() + previous.close() + next := <-joined + if next.err != nil { + t.Fatalf("joining account interrupted at iteration %d: %v", i, next.err) + } + previous = next.sub + } +} diff --git a/internal/controlplane/api/event_cover_worker.go b/internal/controlplane/api/event_cover_worker.go new file mode 100644 index 0000000..45ce31a --- /dev/null +++ b/internal/controlplane/api/event_cover_worker.go @@ -0,0 +1,86 @@ +package api + +import ( + "context" + "errors" + + "git.ipao.vip/rogee/creator-hub/internal/creator" + "github.com/sirupsen/logrus" +) + +type listenerCoverWriter interface { + UpdateListenerEventCovers(context.Context, string, string, []creator.ListenerDelivery) error +} +type listenerCoverEnricher interface { + enrich(context.Context, []creator.ListenerDelivery) ([]creator.ListenerDelivery, error) +} + +const coverQueueFullReason = "封面待处理队列已满，等待历史核对重试" + +// The account receiver is the sole producer: prepare before persistence, submit +// after commit. Queue overload is recorded in that same transaction, never by +// a second synchronous database write on the event acknowledgement path. +func startListenerCoverWorker(ctx context.Context, store listenerWriter, accountID, generation string, session listenerSession) func([]creator.ListenerDelivery) func() { + noop := func() {} + writer, ok := store.(listenerCoverWriter) + enricher, has := session.(listenerCoverEnricher) + if !ok || !has { + return func([]creator.ListenerDelivery) func() { return noop } + } + jobs := make(chan []creator.ListenerDelivery, 32) + log := logrus.WithFields(logrus.Fields{"account_id": accountID, "generation": generation}) + save := func(items []creator.ListenerDelivery) { + if err := writer.UpdateListenerEventCovers(ctx, accountID, generation, items); err != nil { + if !errors.Is(err, creator.ErrConflict) && ctx.Err() == nil { + log.WithError(err).Error("persist event cover result failed") + } + return + } + publishEventChanges(ctx) + } + go func() { + for { + select { + case <-ctx.Done(): + return + case items := <-jobs: + result, err := enricher.enrich(ctx, items) + if err != nil { + log.WithError(err).Error("event cover enrichment failed") + for i := range items { + items[i].Notice.WorkCoverError = err.Error() + } + save(items) + } else { + save(result) + } + } + } + }() + return func(items []creator.ListenerDelivery) func() { + pending := make([]creator.ListenerDelivery, 0, len(items)) + for _, d := range items { + if d.Notice != nil && d.Notice.WorkID != "" && d.Notice.WorkCoverURL != "" { + copy := *d.Notice + d.Notice = © + pending = append(pending, d) + } + } + if len(pending) == 0 { + return noop + } + if len(jobs) == cap(jobs) { + log.Error("event cover queue full; recording overload with notification persistence") + for i, d := range items { + if d.Notice != nil && d.Notice.WorkID != "" && d.Notice.WorkCoverURL != "" { + copy := *d.Notice + copy.WorkCoverError = coverQueueFullReason + items[i].Notice = © + } + } + return noop + } + // No other producer can fill this capacity between prepare and commit. + return func() { jobs <- pending } + } +} diff --git a/internal/controlplane/api/event_cover_worker_test.go b/internal/controlplane/api/event_cover_worker_test.go new file mode 100644 index 0000000..c0b91d2 --- /dev/null +++ b/internal/controlplane/api/event_cover_worker_test.go @@ -0,0 +1,107 @@ +package api + +import ( + "context" + "errors" + "git.ipao.vip/rogee/creator-hub/internal/creator" + "testing" + "time" +) + +type coverWriterStub struct { + listenerWriterStub + saved chan []creator.ListenerDelivery + err error +} + +func (s *coverWriterStub) UpdateListenerEventCovers(_ context.Context, _, _ string, items []creator.ListenerDelivery) error { + s.saved <- items + return s.err +} + +type coverEnricherStub struct { + listenerSessionStub + entered chan struct{} + release chan struct{} + err error +} + +func (s *coverEnricherStub) enrich(ctx context.Context, items []creator.ListenerDelivery) ([]creator.ListenerDelivery, error) { + select { + case s.entered <- struct{}{}: + default: + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-s.release: + } + for i := range items { + items[i].Notice.WorkCoverError = "cached result" + } + return items, s.err +} +func coverDelivery() []creator.ListenerDelivery { + return []creator.ListenerDelivery{{Kind: "notice", DeliveryID: "cover", Notice: &creator.ListenerNotice{EventKey: "cover", EventType: "like", WorkID: "9", WorkAuthorUID: "1", WorkCoverURL: "https://example.com/cover"}}} +} +func TestEventCoverWorkerDoesNotBlockReceptionAndReportsOverload(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + writer := &coverWriterStub{saved: make(chan []creator.ListenerDelivery, 64)} + session := &coverEnricherStub{entered: make(chan struct{}, 1), release: make(chan struct{})} + queue := startListenerCoverWorker(ctx, writer, "account", "generation", session) + original := coverDelivery() + queue(original)() + select { + case <-session.entered: + case <-time.After(time.Second): + t.Fatal("cover worker did not start") + } + for i := 0; i < 32; i++ { + queue(coverDelivery())() + } + overloaded := coverDelivery() + queue(overloaded)() + if overloaded[0].Notice.WorkCoverError != coverQueueFullReason { + t.Fatal("overload must be included in notification persistence") + } + select { + case <-writer.saved: + t.Fatal("overload performed a synchronous cover-result write") + default: + } + if original[0].Notice.WorkCoverError != "" { + t.Fatal("delivery was mutated before persistence") + } + close(session.release) + select { + case saved := <-writer.saved: + if saved[0].Notice.WorkCoverError != "cached result" { + t.Fatal("result missing") + } + case <-time.After(time.Second): + t.Fatal("cover not saved") + } +} +func TestEventCoverFailureIsRecordedAndDoesNotChangeListenerStatus(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + release := make(chan struct{}) + close(release) + writer := &coverWriterStub{saved: make(chan []creator.ListenerDelivery, 4), err: creator.ErrConflict} + session := &coverEnricherStub{entered: make(chan struct{}, 1), release: release, err: errors.New("cover unavailable")} + queue := startListenerCoverWorker(ctx, writer, "account", "generation", session) + queue(nil)() + queue(coverDelivery())() + select { + case saved := <-writer.saved: + if saved[0].Notice.WorkCoverError != "cover unavailable" { + t.Fatal("error hidden") + } + case <-time.After(time.Second): + t.Fatal("error not saved") + } + if writer.batches != 0 { + t.Fatal("cover failure changed listener health") + } +} diff --git a/internal/controlplane/api/event_listener.go b/internal/controlplane/api/event_listener.go index cc182ff..7821c4a 100644 --- a/internal/controlplane/api/event_listener.go +++ b/internal/controlplane/api/event_listener.go @@ -28,6 +28,7 @@ func registerAccountEventRoutes(app *fiber.App, store *creator.Store) { return creatorError(c, err) } logrus.WithFields(logrus.Fields{"account_id": state.AccountID, "generation": state.Generation, "enabled": state.Enabled}).Info("account event listener setting changed") + creatorUpdates.publish() return c.JSON(state) }) app.Get("/api/creator/events", func(c fiber.Ctx) error { diff --git a/internal/controlplane/api/event_listener_lifecycle_test.go b/internal/controlplane/api/event_listener_lifecycle_test.go index 044c2b9..959d024 100644 --- a/internal/controlplane/api/event_listener_lifecycle_test.go +++ b/internal/controlplane/api/event_listener_lifecycle_test.go @@ -5,9 +5,9 @@ import ( "encoding/json" "errors" "fmt" + "github.com/gorilla/websocket" "net/http" "net/http/httptest" - "net/url" "strings" "sync" "testing" @@ -131,7 +131,7 @@ func TestGatewayListenerSessionProtocol(t *testing.T) { case http.MethodDelete: w.WriteHeader(http.StatusNoContent) case http.MethodPost: - w.Write([]byte(`{"connected":true,"alias":"browser","uid":"123"}`)) + w.Write([]byte(`{"connected":true,"alias":"browser","uid":"123","session_id":"test-session"}`)) case http.MethodGet: w.Write([]byte(`[{"delivery_id":"1","kind":"open"}]`)) default: @@ -144,22 +144,13 @@ func TestGatewayListenerSessionProtocol(t *testing.T) { if err != nil { t.Fatal(err) } - if items, err := session.Start(context.Background()); err != nil || len(items) != 0 { + if items, err := session.request(context.Background(), "start"); err != nil || len(items) != 0 { t.Fatalf("start=%+v %v", items, err) } start := <-calls if start["method"] != http.MethodPost || start["expected_uid"] != "123" || start["binding_version"] != float64(7) || start["runtime_id"] != "runtime" { t.Fatalf("start=%+v", start) } - session.pollInterval = 0 - if _, err := session.Poll(context.Background(), []string{"1"}); err != nil { - t.Fatal(err) - } - poll := <-calls - query := poll["query"].(url.Values) - if poll["method"] != http.MethodGet || len(query["ack"]) != 1 || query.Get("ack") != "1" || query.Get("wait") != "0" || query.Get("limit") != "100" { - t.Fatalf("poll=%+v", poll) - } if err := session.Stop(context.Background()); err != nil { t.Fatal(err) } @@ -167,7 +158,7 @@ func TestGatewayListenerSessionProtocol(t *testing.T) { if stop["method"] != http.MethodDelete { t.Fatalf("stop=%+v", stop) } - for i := 0; i < 3; i++ { + for i := 0; i < 2; i++ { select { case aliases := <-accounts.lockCalls: if len(aliases) != 1 || aliases[0] != "browser" { @@ -285,6 +276,28 @@ func TestListenerReconcilerHonorsEnableDisableAndRestart(t *testing.T) { var starts, stops int var pending string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/channel" { + c, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if err != nil { + t.Error(err) + return + } + defer c.Close() + var value map[string]any + if c.ReadJSON(&value) != nil { + return + } + _ = c.WriteJSON(map[string]any{"type": "subscribed", "subscription": value["subscription"]}) + mu.Lock() + delivery := pending + mu.Unlock() + _ = c.WriteJSON(map[string]any{"type": "deliveries", "subscription": value["subscription"], "deliveries": []creator.ListenerDelivery{{Kind: "open", DeliveryID: delivery}}}) + for { + if c.ReadJSON(&value) != nil { + return + } + } + } w.Header().Set("Content-Type", "application/json") switch { case r.Method == http.MethodPost: @@ -292,7 +305,7 @@ func TestListenerReconcilerHonorsEnableDisableAndRestart(t *testing.T) { starts++ pending = fmt.Sprintf("open-%d", starts) mu.Unlock() - w.Write([]byte(`{"connected":true,"alias":"browser","uid":"123"}`)) + w.Write([]byte(`{"connected":true,"alias":"browser","uid":"123","session_id":"test-session"}`)) case r.Method == http.MethodDelete: mu.Lock() stops++ diff --git a/internal/controlplane/api/event_listener_worker.go b/internal/controlplane/api/event_listener_worker.go index b1b5bd4..5b54cf4 100644 --- a/internal/controlplane/api/event_listener_worker.go +++ b/internal/controlplane/api/event_listener_worker.go @@ -39,14 +39,17 @@ type gatewayListenerSession struct { target hub.Gateway environment hub.EnvironmentContext uid string + generation string + sessionID string + enabled bool resources browserAliasResourceLocker boundaryAt *time.Time checkpoints map[string]string - pollInterval time.Duration coverFiles eventCoverFiles + subscription *eventSubscription } -func (s gatewayListenerSession) request(ctx context.Context, action string, ack []string) ([]creator.ListenerDelivery, error) { +func (s *gatewayListenerSession) request(ctx context.Context, action string) ([]creator.ListenerDelivery, error) { if s.resources == nil { return nil, errors.New("event listener browser alias resource locker unavailable") } @@ -57,6 +60,11 @@ func (s gatewayListenerSession) request(ctx context.Context, action string, ack defer unlock() payload := gatewayGenerationPayload(s.environment) + payload["listener_generation"] = s.generation + payload["session_id"] = s.sessionID + if action == "stop" && !s.enabled { + payload["stop_all"] = true + } path := "/v1/browsers/" + url.PathEscape(s.environment.Alias) + "/douyin/events" method, expectedStatus := http.MethodPost, http.StatusOK switch action { @@ -68,13 +76,6 @@ func (s gatewayListenerSession) request(ctx context.Context, action string, ack checkpoints = map[string]string{} } payload["notice_checkpoints"] = checkpoints - case "poll": - method = http.MethodGet - query := url.Values{"limit": {"100"}, "wait": {"0"}} - for _, id := range ack { - query.Add("ack", id) - } - path += "?" + query.Encode() case "stop": method, expectedStatus = http.MethodDelete, http.StatusNoContent default: @@ -95,40 +96,43 @@ func (s gatewayListenerSession) request(ctx context.Context, action string, ack Connected bool `json:"connected"` Alias string `json:"alias"` UID string `json:"uid"` + SessionID string `json:"session_id"` } if err := json.Unmarshal(body, &response); err != nil { return nil, fmt.Errorf("event listener start response: %w", err) } - if !response.Connected || response.Alias != s.environment.Alias || response.UID != s.uid { + s.sessionID = response.SessionID + if !response.Connected || response.Alias != s.environment.Alias || response.UID != s.uid || response.SessionID == "" { return nil, errors.New("event listener start response identity or connection mismatch") } return nil, nil } - var items []creator.ListenerDelivery - if err := json.Unmarshal(body, &items); err != nil { - return nil, fmt.Errorf("event listener poll response: %w", err) - } - if items == nil { - return nil, errors.New("event listener poll response must be an array") - } - for _, d := range items { - if err := d.Validate(); err != nil { - return nil, fmt.Errorf("event listener invalid %s delivery: %w", d.Kind, err) - } - } - return items, nil + return nil, creator.ErrInvalid } -func (s gatewayListenerSession) Start(ctx context.Context) ([]creator.ListenerDelivery, error) { - return s.request(ctx, "start", nil) -} -func (s gatewayListenerSession) Poll(ctx context.Context, ack []string) ([]creator.ListenerDelivery, error) { - if s.pollInterval > 0 && !waitListener(ctx, s.pollInterval) { - return nil, ctx.Err() - } - items, err := s.request(ctx, "poll", ack) - if err != nil { +func (s *gatewayListenerSession) Start(ctx context.Context) ([]creator.ListenerDelivery, error) { + if _, err := s.request(ctx, "start"); err != nil { return nil, err } + params := gatewayGenerationPayload(s.environment) + params["alias"], params["uid"] = s.environment.Alias, s.uid + var err error + params["listener_generation"] = s.generation + params["session_id"] = s.sessionID + subscribeCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + s.subscription, err = acquireEventSubscription(subscribeCtx, s.target, params) + if err != nil { + cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer cleanupCancel() + _, cleanupErr := s.request(cleanupCtx, "stop") + return nil, errors.Join(err, cleanupErr) + } + return nil, nil +} +func (s *gatewayListenerSession) Poll(ctx context.Context, ack []string) ([]creator.ListenerDelivery, error) { + return s.subscription.poll(ctx, ack) +} +func (s *gatewayListenerSession) enrich(ctx context.Context, items []creator.ListenerDelivery) ([]creator.ListenerDelivery, error) { if s.coverFiles != nil { browser := creatorGatewayBrowser{gateway: s.target, environment: s.environment} fetch := func(ctx context.Context, target string) (string, []byte, error) { @@ -149,8 +153,11 @@ func (s gatewayListenerSession) Poll(ctx context.Context, ack []string) ([]creat } return items, nil } -func (s gatewayListenerSession) Stop(ctx context.Context) error { - _, err := s.request(ctx, "stop", nil) +func (s *gatewayListenerSession) Stop(ctx context.Context) error { + if s.subscription != nil { + s.subscription.close() + } + _, err := s.request(ctx, "stop") return err } @@ -168,6 +175,9 @@ func runListenerSession(ctx context.Context, store listenerWriter, accountID, ge } err = errors.Join(err, stopErr) }() + coverCtx, stopCovers := context.WithCancel(ctx) + defer stopCovers() + queueCovers := startListenerCoverWorker(coverCtx, store, accountID, generation, session) healthStatus, healthReason := "starting", "" for { if err := ctx.Err(); err != nil { @@ -189,6 +199,7 @@ func runListenerSession(ctx context.Context, store listenerWriter, accountID, ge } } } + submitCovers := queueCovers(items) persistenceFailed := false for { err := store.RecordListenerDeliveries(ctx, accountID, generation, items) @@ -216,6 +227,18 @@ func runListenerSession(ctx context.Context, store listenerWriter, accountID, ge return err } } + savedAt := time.Now().UTC() + for _, item := range items { + if item.Notice != nil { + fields := logrus.Fields{"account_id": accountID, "generation": generation, "delivery_id": item.DeliveryID, "event_key": item.Notice.EventKey, "realtime_signal": item.RealtimeSignal, "gateway_received_at": item.Notice.GatewayReceivedAt, "saved_at": savedAt.Format(time.RFC3339Nano)} + if item.Notice.GatewayReceivedAt != nil { + fields["gateway_to_save_clock_delta_ms"] = savedAt.Sub(*item.Notice.GatewayReceivedAt).Milliseconds() + } + logrus.WithFields(fields).Info("listener event committed") + } + } + publishEventChanges(ctx) + submitCovers() logrus.WithFields(logrus.Fields{"account_id": accountID, "generation": generation, "delivery_count": len(items), "last_delivery_id": items[len(items)-1].DeliveryID}).Debug("account event deliveries committed") for _, d := range items { ack = append(ack, d.DeliveryID) @@ -259,8 +282,8 @@ func listenerSessionForAccount(ctx context.Context, store listenerStore, account return gatewayListenerSession{}, err } coverFiles, _ := store.(eventCoverFiles) - return gatewayListenerSession{target: target, environment: environment, uid: profile.PlatformAccountKey, coverFiles: coverFiles, resources: accounts, - boundaryAt: state.BoundaryAt, checkpoints: state.Checkpoints, pollInterval: 5 * time.Second}, nil + return gatewayListenerSession{target: target, environment: environment, uid: profile.PlatformAccountKey, generation: state.Generation, enabled: state.Enabled, coverFiles: coverFiles, resources: accounts, + boundaryAt: state.BoundaryAt, checkpoints: state.Checkpoints}, nil } func runAccountEventListener(ctx context.Context, store listenerStore, accounts HubStore, state creator.ListenerState) { @@ -268,9 +291,13 @@ func runAccountEventListener(ctx context.Context, store listenerStore, accounts update := func(status, reason string) { writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) defer cancel() - if err := store.UpdateListenerStatus(writeCtx, state.AccountID, state.Generation, status, reason); err != nil && !errors.Is(err, creator.ErrConflict) { + err := store.UpdateListenerStatus(writeCtx, state.AccountID, state.Generation, status, reason) + if err != nil && !errors.Is(err, creator.ErrConflict) { log.WithError(err).Error("persist event listener status failed") } + if err == nil { + publishEventChanges(writeCtx) + } } for ctx.Err() == nil { // Reload the last committed checkpoint after any failed request or restart. @@ -313,7 +340,7 @@ func runAccountEventListener(ctx context.Context, store listenerStore, accounts if err == nil { if state.Enabled { log.Info("account event listener starting") - err = runListenerSession(ctx, store, state.AccountID, state.Generation, session) + err = runListenerSession(ctx, store, state.AccountID, state.Generation, &session) } else { err = session.Stop(ctx) if err == nil { @@ -332,8 +359,8 @@ func runAccountEventListener(ctx context.Context, store listenerStore, accounts return } update("error", err.Error()) - log.WithError(err).Error("account event listener failed; retry in 10 seconds") - if !waitListener(ctx, 10*time.Second) { + log.WithError(err).Error("account event listener failed; retry in 2 seconds") + if !waitListener(ctx, 2*time.Second) { return } } diff --git a/internal/controlplane/api/event_push_commit_test.go b/internal/controlplane/api/event_push_commit_test.go new file mode 100644 index 0000000..f130a66 --- /dev/null +++ b/internal/controlplane/api/event_push_commit_test.go @@ -0,0 +1,55 @@ +package api + +import ( + "context" + "git.ipao.vip/rogee/creator-hub/internal/creator" + "testing" + "time" +) + +type gatedListenerWriter struct { + listenerWriterStub + entered, release chan struct{} +} + +func (s *gatedListenerWriter) RecordListenerDeliveries(ctx context.Context, id, generation string, items []creator.ListenerDelivery) error { + close(s.entered) + select { + case <-ctx.Done(): + return ctx.Err() + case <-s.release: + } + return s.listenerWriterStub.RecordListenerDeliveries(ctx, id, generation, items) +} +func TestListenerPushIsPublishedOnlyAfterSuccessfulCommit(t *testing.T) { + updates, remove := creatorUpdates.subscribe() + defer remove() + writer := &gatedListenerWriter{entered: make(chan struct{}), release: make(chan struct{})} + session := &listenerSessionStub{initial: []creator.ListenerDelivery{{Kind: "open", DeliveryID: "saved"}}, pollErr: creator.ErrConflict} + result := make(chan error, 1) + go func() { result <- runListenerSession(context.Background(), writer, "account", "generation", session) }() + select { + case <-writer.entered: + case <-time.After(time.Second): + t.Fatal("persistence not started") + } + select { + case <-updates: + t.Fatal("published before commit") + default: + } + close(writer.release) + select { + case <-result: + case <-time.After(time.Second): + t.Fatal("worker did not finish") + } + select { + case <-updates: + case <-time.After(100 * time.Millisecond): + t.Fatal("successful commit did not notify page") + } + if len(session.ack) != 1 || session.ack[0] != "saved" { + t.Fatal("saved delivery not acknowledged") + } +} diff --git a/internal/controlplane/api/event_updates.go b/internal/controlplane/api/event_updates.go new file mode 100644 index 0000000..071a651 --- /dev/null +++ b/internal/controlplane/api/event_updates.go @@ -0,0 +1,6 @@ +package api + +import "context" + +// Reuse the existing creator SSE feed; the database remains the source of truth. +func publishEventChanges(context.Context) { creatorUpdates.publish() } diff --git a/internal/creator/event_cover_update.go b/internal/creator/event_cover_update.go new file mode 100644 index 0000000..0e02797 --- /dev/null +++ b/internal/creator/event_cover_update.go @@ -0,0 +1,37 @@ +package creator + +import "context" + +// Cover completion cannot rewrite an event, a delivery cursor, or its history flag. +func (s *Store) UpdateListenerEventCovers(ctx context.Context, accountID, generation string, items []ListenerDelivery) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + var id int64 + var enabled bool + var current string + err = tx.QueryRowContext(ctx, `SELECT l.account_id,l.enabled,l.generation FROM creator_account_listener l + JOIN social_account a ON a.id=l.account_id WHERE a.account_id=$1 FOR UPDATE OF l`, accountID).Scan(&id, &enabled, ¤t) + if err != nil { + return rowError(err) + } + if !enabled || current != generation { + return ErrConflict + } + for _, d := range items { + if d.Notice == nil || d.Notice.WorkID == "" { + continue + } + n := d.Notice + if err := d.Validate(); err != nil { + return err + } + _, err = tx.ExecContext(ctx, `UPDATE creator_account_event SET work_cover_error=$1 WHERE account_id=$2 AND event_key=$3 AND work_id=$4 AND work_author_uid=$5`, n.WorkCoverError, id, n.EventKey, n.WorkID, n.WorkAuthorUID) + if err != nil { + return err + } + } + return tx.Commit() +} diff --git a/internal/creator/event_cover_update_test.go b/internal/creator/event_cover_update_test.go new file mode 100644 index 0000000..4f3f32e --- /dev/null +++ b/internal/creator/event_cover_update_test.go @@ -0,0 +1,55 @@ +package creator + +import ( + "errors" + "fmt" + "testing" + "time" +) + +func TestUpdateListenerEventCoversPreservesNoticeAndRejectsOldGeneration(t *testing.T) { + store, accounts, ctx := openCreatorIntegrationStore(t) + id := createIntegrationAccount(t, ctx, accounts, fmt.Sprint(time.Now().UnixNano())) + profile, err := store.GetAccountProfile(ctx, id) + if err != nil { + t.Fatal(err) + } + if _, err = store.RecordVerifiedLoginResult(ctx, id, profile.PlatformAccountKey); err != nil { + t.Fatal(err) + } + state, err := store.SetListenerEnabled(ctx, id, true) + if err != nil { + t.Fatal(err) + } + at := time.Now().UTC().Add(-time.Hour).Truncate(time.Second) + delivery := ListenerDelivery{Kind: "notice", DeliveryID: "original", Baseline: true, Notice: &ListenerNotice{EventKey: "notice", EventType: "like", MessageText: "original", InteractorUID: "22", WorkID: "99", WorkAuthorUID: "123", PlatformEventAt: &at}} + if err = store.RecordListenerDeliveries(ctx, id, state.Generation, []ListenerDelivery{delivery}); err != nil { + t.Fatal(err) + } + delivery.Notice.WorkCoverError = "download failed" + if err = store.UpdateListenerEventCovers(ctx, id, state.Generation, []ListenerDelivery{delivery}); err != nil { + t.Fatal(err) + } + page, err := store.ListAggregatedEventsPage(ctx, EventAggregationFilter{AccountID: id}, 1, 10) + if err != nil { + t.Fatal(err) + } + if len(page.Data) != 1 || page.Data[0].WorkCoverError != "download failed" || page.Data[0].MessageText != "original" || !page.Data[0].Baseline || !page.Data[0].PlatformEventAt.Equal(at) { + t.Fatalf("event rewritten: %+v", page.Data) + } + delivery.Notice.WorkID = "100" + delivery.Notice.WorkCoverError = "wrong work" + if err = store.UpdateListenerEventCovers(ctx, id, state.Generation, []ListenerDelivery{delivery}); err != nil { + t.Fatal(err) + } + var coverError string + if err = store.db.QueryRowContext(ctx, `SELECT work_cover_error FROM creator_account_event WHERE account_id=(SELECT id FROM social_account WHERE account_id=$1) AND event_key='notice'`, id).Scan(&coverError); err != nil || coverError != "download failed" { + t.Fatalf("wrong work updated: %q %v", coverError, err) + } + if _, err = store.SetListenerEnabled(ctx, id, false); err != nil { + t.Fatal(err) + } + if err = store.UpdateListenerEventCovers(ctx, id, state.Generation, []ListenerDelivery{delivery}); !errors.Is(err, ErrConflict) { + t.Fatalf("old generation accepted: %v", err) + } +} diff --git a/internal/creator/event_listener.go b/internal/creator/event_listener.go index f1fa398..de4d50a 100644 --- a/internal/creator/event_listener.go +++ b/internal/creator/event_listener.go @@ -31,14 +31,15 @@ type ListenerNotice struct { } type ListenerDelivery struct { - DeliveryID string `json:"delivery_id"` - Kind string `json:"kind"` - BoundaryAt *time.Time `json:"boundary_at"` - Reason string `json:"reason"` - Baseline bool `json:"baseline"` - Notice *ListenerNotice `json:"notice"` - Checkpoints map[string]string `json:"checkpoints,omitempty"` - CheckedAt *time.Time `json:"checked_at,omitempty"` + DeliveryID string `json:"delivery_id"` + Kind string `json:"kind"` + BoundaryAt *time.Time `json:"boundary_at"` + Reason string `json:"reason"` + Baseline bool `json:"baseline"` + RealtimeSignal bool `json:"realtime_signal,omitempty"` + Notice *ListenerNotice `json:"notice"` + Checkpoints map[string]string `json:"checkpoints,omitempty"` + CheckedAt *time.Time `json:"checked_at,omitempty"` } func validEventType(value string) bool { diff --git a/requirements-gateway.lock b/requirements-gateway.lock index 2f738ef..964e23e 100644 --- a/requirements-gateway.lock +++ b/requirements-gateway.lock @@ -1 +1,4 @@ websocket-client==1.9.0 +simple-websocket==1.1.0 +wsproto==1.3.2 +h11==0.16.0 diff --git a/web/src/pages/creator/events/index.tsx b/web/src/pages/creator/events/index.tsx index 6f0e02a..629a725 100644 --- a/web/src/pages/creator/events/index.tsx +++ b/web/src/pages/creator/events/index.tsx @@ -2,7 +2,7 @@ import { useEffect, useState } from 'react'; import { Alert, Avatar, Button, Card, Flex, Select, Table, Tag, Tooltip, Typography } from 'antd'; import { ReloadOutlined, VideoCameraOutlined } from '@ant-design/icons'; import type { ColumnsType } from 'antd/es/table'; -import { creatorGet } from '@/services/api'; +import { creatorGet, creatorSubscribe } from '@/services/api'; import { usePageState } from '@/hooks/usePageState'; import { conflictMessage, dateTime } from '@/utils/helpers'; import { tablePagination, useTableSticky } from '@/utils/table'; @@ -75,17 +75,25 @@ export default function EventAggregationPage() { const [pending, setPending] = useState(true); const [error, setError] = useState(null); const [revision, setRevision] = useState(0); + const [streamStatus, setStreamStatus] = useState<'connecting' | 'connected' | 'disconnected'>('connecting'); const sticky = useTableSticky(); useEffect(() => { let active = true; let timer: ReturnType; + let busy = false; + let queued = false; + const controller = new AbortController(); + setStreamStatus('connecting'); const query = eventQuery(filters, pagination.page, pagination.pageSize); setPending(true); setError(null); setEvents([]); setTotal(0); async function load() { + if (!active) return; + if (busy) { queued = true; return; } + busy = true; try { const [page, profiles] = await Promise.all([ creatorGet(`/creator/events?${query}`), creatorGet('/creator/accounts'), @@ -100,14 +108,32 @@ export default function EventAggregationPage() { console.error('读取账号事件失败', { query, error: cause }); if (active) setError(cause); } finally { + busy = false; if (active) { setPending(false); - timer = setTimeout(load, 5000); + if (queued) { queued = false; void load(); } } } } + async function connect() { + try { + await creatorSubscribe('/creator/updates', () => { if (active) void load(); }, controller.signal, + (status) => { + if (!active) return; + setStreamStatus(status); + if (status === 'connected') void load(); + }); + } catch (cause) { + if (active) console.error('事件实时更新连接失败', { error: cause }); + } + if (active) { + setStreamStatus('disconnected'); + timer = setTimeout(connect, 2000); + } + } void load(); - return () => { active = false; clearTimeout(timer); }; + void connect(); + return () => { active = false; controller.abort(); clearTimeout(timer); }; }, [filters.accountId, filters.eventType, filters.hours, pagination.page, pagination.pageSize, revision]); function updateFilters(next: Partial) { @@ -118,7 +144,7 @@ export default function EventAggregationPage() { return ( - 共 {total} 条事件 · 每 5 秒刷新 + 共 {total} 条事件 · {streamStatus === 'connected' ? '实时更新' : streamStatus === 'connecting' ? '正在连接实时更新' : '实时更新已断开'} @@ -133,6 +159,7 @@ export default function EventAggregationPage() { { value: 24, label: '最近 1 天' }, { value: 72, label: '最近 3 天' }, { value: 168, label: '最近 7 天' }]} /> 仅展示已开启监听账号的事件；停止监听的账号不在此列表展示，历史记录仍保留。时间范围按接收时间计算，仅记录事件，不执行自动互动。私信在独立收件箱展示。 + {streamStatus === 'disconnected' ? : null} {error ? : null} rowKey="id" columns={columns} dataSource={events} loading={pending} sticky={sticky} tableLayout="fixed" diff --git a/web/tests/event-aggregation-page.test.cjs b/web/tests/event-aggregation-page.test.cjs index 9ec9025..44208df 100644 --- a/web/tests/event-aggregation-page.test.cjs +++ b/web/tests/event-aggregation-page.test.cjs @@ -9,7 +9,7 @@ const React = require('react'); function transpile(file) { return ts.transpileModule(readFileSync(file,'utf8'), {compilerOptions:{module:ts.ModuleKind.CommonJS,jsx:ts.JsxEmit.ReactJSX,target:ts.ScriptTarget.ES2020}}).outputText; } function harness(options={}) { - const states=[],effects=[],queued=[],calls=[]; + const states=[],effects=[],queued=[],calls=[],streams=[]; let cursor=0; const antd={}; for(const name of ['Alert','Avatar','Button','Card','Flex','Select','Table','Tag','Tooltip']){antd[name]=()=>null;antd[name].displayName=name;} @@ -29,11 +29,14 @@ function harness(options={}) { if(id==='@/hooks/usePageState')return loadPageState(loaded.require('react'),{storage:options.storage}); if(id==='@/utils/table')return{useTableSticky:()=>false,tablePagination:()=>({})}; if(id==='@/utils/helpers')return{dateTime:value=>`time:${value}`,conflictMessage:(cause,fallback)=>cause.message||fallback}; - if(id==='@/services/api')return{creatorGet:async path=>{calls.push(path);if(options.get)return options.get(path);if(path==='/creator/accounts')return options.accounts||[];return options.page||{data:[],total:0};}}; + if(id==='@/services/api')return{creatorGet:async path=>{calls.push(path);if(options.get)return options.get(path);if(path==='/creator/accounts')return options.accounts||[];return options.page||{data:[],total:0};},creatorSubscribe:(path,onMessage,signal,onStatus)=>{ + streams.push({path,onMessage,signal,onStatus});onStatus?.('connected'); + return new Promise(resolve=>signal.addEventListener('abort',resolve,{once:true})); + }}; return require(id); }; loaded._compile(transpile(file),file); - return{antd,calls,render(){cursor=0;return loaded.exports.default();},async effects(){while(queued.length)queued.shift()();for(let i=0;i<8;i++)await Promise.resolve();},dispose(){for(const effect of effects)effect?.cleanup?.();}}; + return{antd,calls,streams,push(){streams.at(-1).onMessage('data: {}');},render(){cursor=0;return loaded.exports.default();},async effects(){while(queued.length)queued.shift()();for(let i=0;i<8;i++)await Promise.resolve();},dispose(){for(const effect of effects)effect?.cleanup?.();}}; } function all(node,predicate){const result=[];function visit(value){if(Array.isArray(value))return value.forEach(visit);if(!React.isValidElement(value))return;if(predicate(value))result.push(value);visit(value.props.children);}visit(node);return result;} function text(node){if(node==null)return'';if(typeof node==='string'||typeof node==='number')return String(node);if(Array.isArray(node))return node.map(text).join('');return React.isValidElement(node)?text(node.props.children):'';} @@ -85,7 +88,17 @@ test('only enabled accounts are offered, event filters are server-wide and chang table(h,tree).props.pagination.onChange(3,50);h.render();await h.effects();tree=h.render(); assert.equal(table(h,tree).props.pagination.current,1);assert.equal(table(h,tree).props.pagination.pageSize,50); all(tree,node=>node.type===h.antd.Button)[0].props.onClick();h.render();await h.effects(); - assert.equal(h.calls.filter(p=>p.startsWith('/creator/events')).length,7); + assert.ok(h.calls.filter(p=>p.startsWith('/creator/events')).length>=7); + assert.ok(h.streams.every(s=>s.path==='/creator/updates')); + const before=h.calls.length;h.push();await h.effects();assert.ok(h.calls.length>before); + assert.equal(table(h,h.render()).props.pagination.current,1); +}); + +test('stream disconnect is visible and cleanup aborts the subscription',async t=>{ + const h=harness();t.after(()=>h.dispose());h.render();await h.effects(); + h.streams.at(-1).onStatus('disconnected'); + assert.ok(all(h.render(),n=>n.type===h.antd.Alert&&n.props.title==='实时更新连接已断开').length); + const signal=h.streams.at(-1).signal;h.dispose();assert.equal(signal.aborted,true); }); test('event contents distinguish missing data, non-text messages and baseline history',async t=>{ diff --git a/web/tests/notice-signal.test.cjs b/web/tests/notice-signal.test.cjs new file mode 100644 index 0000000..e1b6122 --- /dev/null +++ b/web/tests/notice-signal.test.cjs @@ -0,0 +1,39 @@ +const test=require('node:test');const assert=require('node:assert/strict');const fs=require('node:fs');const vm=require('node:vm'); +const source=fs.readFileSync('../browser_gateway/platform/notice_signal.py','utf8'); +const install=source.match(/INSTALL = r"""([\s\S]*?)"""/)[1];const wait=source.match(/WAIT = r"""([\s\S]*?)"""/)[1]; +function browser(){ + const listeners=new Map(),timers=new Map();let counter=0; + const f={_options:{deviceID:'123'},OPEN:1,readyState:1,addEventListener(k,fn){if(!listeners.has(k))listeners.set(k,new Set());listeners.get(k).add(fn);},removeEventListener(k,fn){listeners.get(k)?.delete(fn);}}; + const C={frontierInstance:f};const modules={sdk:{NoticeFrontier:C},codec:{decodedFrame:data=>({service:20313,payload:data})}}; + const req=id=>modules[id];req.m={sdk:function(){/*NOTICE_PUSH_EVENT_NAMES:function*/},codec:function(){/* .decodedFrame= .encodeFrame= */}}; + const chunks=[];chunks.push=function(v){v[2](req);return Array.prototype.push.call(this,v)}; + const window={webpackChunkdouyin_web:chunks}; + const context=vm.createContext({window,location:{origin:'https://www.douyin.com'},Uint8Array,TextDecoder,JSON,Date,Error, + setTimeout(fn){const id=++counter;timers.set(id,fn);return id;},clearTimeout(id){timers.delete(id);}}); + const evaluate=s=>vm.runInContext(s.replaceAll('KEY','"bridge"').replaceAll('UID','"123"').replaceAll('ALIAS','"account"'),context); + return{window,f,C,timers,evaluate,emit(k,value){for(const fn of listeners.get(k)||[])fn(value)},listeners}; +} +test('native notification wakes a pending receiver without waiting for heartbeat',async()=>{ + const b=browser();assert.equal(JSON.parse(await b.evaluate(install)).connected,true); + const pending=b.evaluate(wait);assert.equal(b.timers.size,1); + const payload='{"notices":[{"notice_id_str":"9007199254740993123","effect_groups":[700]}]}'; + b.emit('message',{data:Buffer.from(payload)}); + const result=JSON.parse(await pending);assert.equal(result[0].payload,payload);assert.equal(b.timers.size,0); +}); +test('connection replacement and identity change fail explicitly',async()=>{ + const b=browser();await b.evaluate(install);b.C.frontierInstance={};await assert.rejects(b.evaluate(wait),/identity or page changed/); + const c=browser();await c.evaluate(install);c.f._options.deviceID='456';await assert.rejects(c.evaluate(wait),/identity or page changed/); +}); +test('signal overload reports an error rather than silently dropping notifications',async()=>{ + const b=browser();await b.evaluate(install); + for(let i=0;i<1001;i++)b.emit('message',{data:Buffer.from('{}')}); + await assert.rejects(b.evaluate(wait),/queue full/); +}); +test('reinstall removes stale callbacks instead of accumulating page listeners',async()=>{ + const b=browser();await b.evaluate(install);await b.evaluate(install); + assert.equal(b.listeners.get('message').size,1);assert.equal(b.listeners.get('close').size,1); +}); +test('disposal removes page listeners and wakes an idle receiver',async()=>{ + const b=browser();await b.evaluate(install);const pending=b.evaluate(wait);b.window.bridge.dispose();await pending; + assert.equal(b.listeners.get('message').size,0);assert.equal(b.listeners.get('close').size,0); +});