# Phase 7 · Live Monitor 迁移 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development or superpowers:executing-plans > > **本 plan 特殊性：** 这不是从零写的功能，而是**从 catclaw 项目迁移**生产验证过的代码。catclaw 路径：`/Users/luzhipeng/projects/web_claude_code_mac_native/web/server/tiktok/` **Goal:** 把 catclaw 已经在跑的 Live Monitor 模块（live-monitor.ts + live-scan-scheduler.ts + 相关 collectors + DB schema）迁移到 Pawcast 独立项目，重新组织到 `packages/core/live-monitor/`，对接新的 IPC contract，让 Pawcast 在 macOS 上能 1:1 复现 catclaw 的监控能力。 **Architecture:** 沿用 catclaw 的核心数据流（30s scheduler → connectRoom → live-events 入表 → session 聚合统计），但：① 改用我们 monorepo 的 `@pawcast/db` repository 模式；② 走 `@pawcast/ipc-contract` 的 Zod schema；③ 用 `worker_threads` 跑高并发房间避免阻塞主线程（catclaw 现在主线程跑，8 房并发会有偶发卡顿）。 **Tech Stack:** tiktok-live-connector + Electron BrowserWindow（替代 Playwright）+ bun:sqlite + worker_threads + Zod **Project root:** `/Users/luzhipeng/projects/pawcast` **Source ref:** `/Users/luzhipeng/projects/web_claude_code_mac_native` --- ## 迁移逻辑（不是重写） **catclaw 现有产物（约 1500 行核心代码）：** | catclaw 路径 | 行数 | Pawcast 落点 | |---|---|---| | `web/server/tiktok/live-monitor.ts` | 678 | `packages/core/src/tiktok-engine/live-monitor.ts` | | `web/server/tiktok/live-scan-scheduler.ts` | ~150 | `packages/core/src/tiktok-engine/live-scheduler.ts` | | `web/server/tiktok/collectors/pk-collector.ts` | 56 | `packages/core/src/tiktok-engine/pk-collector.ts` | | `web/server/tiktok/database.ts`（监控相关 ~400 行） | 400 | `packages/db/src/repositories/{live-sessions,live-events,fans,pk-battles}.ts` | **核心改动**： 1. 拆分 monolithic `database.ts` 为 4 个 repository（DRY 原则） 2. live-monitor.ts 的事件分发改为 worker_threads（每房一 worker） 3. 把 `bridge.executeJS` 包装成 `playwright-engine` package（Phase 2 已有）的薄封装 --- ## File Structure（本 phase 创建/迁移） ``` packages/core/src/tiktok-engine/ ├── live-monitor.ts ★ 迁移 + 改造 (RoomConnection 类) ├── live-monitor-worker.ts ★ 新增 (worker thread 入口) ├── live-scheduler.ts ★ 迁移 ├── pk-collector.ts ★ 迁移 ├── event-extractor.ts ★ 抽取 (catclaw 的 extractUserFields) ├── recorder.ts ★ 新增 (JSONL 录制器) ├── manager.ts ★ 新增 (多 worker 协调) └── types.ts ★ 类型定义 (GiftEvent / ChatEvent / 24 种事件) packages/db/src/repositories/ ├── live-sessions.ts ★ 新增 ├── live-events.ts ★ 新增 ├── fans.ts ★ 新增 (送礼者档案) └── pk-battles.ts ★ 新增 packages/db/src/migrations/ └── 002-live-monitor.sql ★ 补充表 (tk_live_sessions / tk_live_events / tk_fans / tk_pk_battles 改名为 live_sessions / live_events / fans / pk_battles) packages/ipc-contract/src/ └── live-monitor.ts ★ 新增 (MonitorContract Zod schema) apps/desktop/src/pages/Monitor/ ├── index.tsx ★ 重写 (从 PageShell 占位 → 真实 UI) ├── components/ │ ├── RoomGrid.tsx │ ├── RoomCard.tsx │ ├── RoomDetail.tsx │ ├── PKTracker.tsx │ ├── EventStream.tsx │ └── HeatMap.tsx ├── hooks/ │ └── useLiveRooms.ts └── stores/ └── monitorStore.ts ``` --- ## 准入条件 - ✅ Phase 1 已 merge（package 骨架 + IPC + DB 基础已就位） - ✅ Phase 2 已 merge（tiktok-live-connector 封装 + Electron BrowserWindow bridge 框架已就位） --- ## Task 1: 迁移 live-events 相关表 schema **Files:** - Create: `packages/db/src/migrations/002-live-monitor.sql` - [ ] **Step 1: 写 migration** 参考 catclaw `database.ts` 中的相关表定义。注意**重命名**去掉 `tk_` 前缀（Pawcast 项目本身就是 TikTok 主题，前缀冗余）： ```sql -- 002-live-monitor: 直播监控相关表 -- 房间元信息（监控配置） CREATE TABLE IF NOT EXISTS room_meta ( streamer_username TEXT PRIMARY KEY, streamer_nickname TEXT, streamer_avatar_url TEXT, auto_monitor INTEGER NOT NULL DEFAULT 0, -- 0/1 monitor_status TEXT, -- 监听中/连接中/待开播/连接失败/已暂停 last_connect_attempt_at TEXT, -- ISO timestamp last_connect_success_at TEXT, last_event_at INTEGER, -- ms group_name TEXT, -- 核心团员/候补/观察中 notes TEXT, updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_room_meta_auto ON room_meta(auto_monitor); -- 直播会话（一场一行） CREATE TABLE IF NOT EXISTS live_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, room_id TEXT NOT NULL, streamer_username TEXT NOT NULL, streamer_nickname TEXT, streamer_avatar_url TEXT, streamer_follower_count INTEGER, streamer_verified INTEGER, anchor_level INTEGER, room_title TEXT, room_category TEXT, room_cover_url TEXT, room_description TEXT, connected_at INTEGER NOT NULL, -- ms disconnected_at INTEGER, is_active INTEGER NOT NULL DEFAULT 1, -- 聚合统计 total_events INTEGER DEFAULT 0, total_chat INTEGER DEFAULT 0, total_gift_count INTEGER DEFAULT 0, total_gift_value INTEGER DEFAULT 0, total_likes INTEGER DEFAULT 0, total_members INTEGER DEFAULT 0, total_follows INTEGER DEFAULT 0, total_shares INTEGER DEFAULT 0, peak_viewers INTEGER DEFAULT 0, avg_viewers INTEGER DEFAULT 0, duration_sec INTEGER DEFAULT 0, diamond_per_hour REAL DEFAULT 0, efficiency REAL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_live_sessions_room ON live_sessions(room_id); CREATE INDEX IF NOT EXISTS idx_live_sessions_username ON live_sessions(streamer_username); CREATE INDEX IF NOT EXISTS idx_live_sessions_connected_at ON live_sessions(connected_at); CREATE INDEX IF NOT EXISTS idx_live_sessions_active ON live_sessions(is_active); -- 直播事件流（24 种事件 + 用户字段） CREATE TABLE IF NOT EXISTS live_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id INTEGER NOT NULL REFERENCES live_sessions(id), room_id TEXT NOT NULL, streamer_username TEXT, event_type TEXT NOT NULL, -- chat/gift/member/like/follow/share/battle/... ts INTEGER NOT NULL, -- ms -- 用户字段 user_uid TEXT, user_username TEXT, user_nickname TEXT, user_avatar_url TEXT, user_verified INTEGER, user_fan_level INTEGER, user_role TEXT, -- subscriber/moderator/follower/visitor user_is_follower INTEGER, user_follower_count INTEGER, user_following_count INTEGER, -- 礼物字段 gift_id TEXT, gift_name TEXT, gift_value INTEGER, gift_count INTEGER, -- 计数字段 viewer_count INTEGER, like_count INTEGER, -- 弹幕原文 chat_text TEXT, -- 原始 payload（兜底，调试用） raw_json TEXT ); CREATE INDEX IF NOT EXISTS idx_live_events_session ON live_events(session_id); CREATE INDEX IF NOT EXISTS idx_live_events_type ON live_events(event_type); CREATE INDEX IF NOT EXISTS idx_live_events_ts ON live_events(ts); CREATE INDEX IF NOT EXISTS idx_live_events_user ON live_events(user_uid); -- 送礼者档案（聚合 live_events 中的用户） CREATE TABLE IF NOT EXISTS fans ( uid TEXT PRIMARY KEY, username TEXT, nickname TEXT, avatar_url TEXT, verified INTEGER, total_gift_value INTEGER DEFAULT 0, total_gift_count INTEGER DEFAULT 0, total_chats INTEGER DEFAULT 0, rooms_seen TEXT, -- JSON array of room_id fan_level_max INTEGER, first_seen INTEGER NOT NULL, last_seen INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_fans_total_gift_value ON fans(total_gift_value); -- PK 战役 CREATE TABLE IF NOT EXISTS pk_battles ( id INTEGER PRIMARY KEY AUTOINCREMENT, battle_id TEXT, my_session_id INTEGER REFERENCES live_sessions(id), my_streamer TEXT NOT NULL, other_room TEXT NOT NULL, other_streamer TEXT, start_at INTEGER NOT NULL, end_at INTEGER, my_score INTEGER DEFAULT 0, other_score INTEGER DEFAULT 0, winner TEXT, -- mine/other/draw punish_type TEXT, meta_json TEXT ); CREATE INDEX IF NOT EXISTS idx_pk_battles_session ON pk_battles(my_session_id); -- 告警规则 CREATE TABLE IF NOT EXISTS alert_rules ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, rule_type TEXT NOT NULL, -- gift_rate / pk_lag / viewer_drop / offline / big_gift config_json TEXT NOT NULL, -- 阈值参数 enabled INTEGER DEFAULT 1, updated_at INTEGER NOT NULL ); ``` - [ ] **Step 2: 跑 runMigrations 验证 schema 创建成功** ```typescript // packages/db/src/migrations/runner.test.ts 加 case test('runMigrations 002 creates room_meta + live_sessions + live_events + fans + pk_battles + alert_rules', () => { const db = openMemoryDb(); runMigrations(db); const tables = (db.prepare("SELECT name FROM sqlite_master WHERE type='table'").all() as any[]).map(r => r.name); ['room_meta', 'live_sessions', 'live_events', 'fans', 'pk_battles', 'alert_rules'].forEach(t => { expect(tables).toContain(t); }); }); ``` - [ ] **Step 3: Commit** ```bash git add packages/db/src/migrations/002-live-monitor.sql git commit -m "feat(db): add migration 002 for live-monitor tables" ``` --- ## Task 2: LiveSessionsRepo + LiveEventsRepo **Files:** - Create: `packages/db/src/repositories/live-sessions.ts` - Create: `packages/db/src/repositories/live-events.ts` - Test: 同名 .test.ts - [ ] **Step 1: `live-sessions.ts`** —— 参考 catclaw `database.ts:1300-1450` ```typescript import type { Database } from 'bun:sqlite'; export interface LiveSessionRow { id: number; room_id: string; streamer_username: string; streamer_nickname?: string; streamer_avatar_url?: string; connected_at: number; disconnected_at?: number; is_active: 0 | 1; total_events: number; total_chat: number; total_gift_count: number; total_gift_value: number; total_likes: number; peak_viewers: number; avg_viewers: number; duration_sec: number; diamond_per_hour: number; // ... full set per migration } export interface NewLiveSession { room_id: string; streamer_username: string; streamer_nickname?: string; streamer_avatar_url?: string; connected_at: number; } export class LiveSessionsRepo { constructor(private db: Database) {} /** * 开启 session。先查 10 分钟内同 room 的 session，有则复用；否则新建。 * 这是从 catclaw live-monitor.ts:225-260 迁移过来的核心逻辑。 */ startOrReuse(s: NewLiveSession, reuseWindowMs = 600_000): { id: number; reused: boolean } { const recent = this.db .prepare( `SELECT id FROM live_sessions WHERE room_id = ? AND streamer_username = ? AND disconnected_at IS NOT NULL AND disconnected_at > ? ORDER BY disconnected_at DESC LIMIT 1`, ) .get(s.room_id, s.streamer_username, Date.now() - reuseWindowMs) as { id: number } | undefined; if (recent) { this.db.prepare(`UPDATE live_sessions SET is_active = 1, disconnected_at = NULL WHERE id = ?`) .run(recent.id); return { id: recent.id, reused: true }; } const result = this.db .prepare( `INSERT INTO live_sessions (room_id, streamer_username, streamer_nickname, streamer_avatar_url, connected_at, is_active) VALUES (?, ?, ?, ?, ?, 1)`, ) .run(s.room_id, s.streamer_username, s.streamer_nickname ?? null, s.streamer_avatar_url ?? null, s.connected_at); return { id: Number(result.lastInsertRowid), reused: false }; } close(sessionId: number, ts = Date.now()): void { this.db .prepare( `UPDATE live_sessions SET disconnected_at = ?, is_active = 0, duration_sec = (? - connected_at) / 1000 WHERE id = ?`, ) .run(ts, ts, sessionId); } /** * 增量更新 session 聚合统计。每个事件到达时调用一次。 * 迁移自 catclaw database.ts:1321-1336 */ incrementStats( sessionId: number, eventType: string, giftValue = 0, giftCount = 0, viewerCount = 0, likeIncrement = 0, ): void { const updates: string[] = ['total_events = total_events + 1']; if (eventType === 'chat') updates.push('total_chat = total_chat + 1'); if (eventType === 'gift') { updates.push(`total_gift_value = total_gift_value + ${giftValue * giftCount}`); updates.push(`total_gift_count = total_gift_count + 1`); } if (eventType === 'follow') updates.push('total_follows = total_follows + 1'); if (eventType === 'share') updates.push('total_shares = total_shares + 1'); if (likeIncrement > 0) updates.push(`total_likes = total_likes + ${likeIncrement}`); if (viewerCount > 0) updates.push(`peak_viewers = MAX(peak_viewers, ${viewerCount})`); this.db.exec(`UPDATE live_sessions SET ${updates.join(',')} WHERE id = ${sessionId}`); } /** * 完整重算（定期 + session 结束时调用） * 迁移自 catclaw database.ts updateLiveSessionStats() */ recompute(sessionId: number): void { this.db.exec(` UPDATE live_sessions SET total_chat = (SELECT COUNT(*) FROM live_events WHERE session_id = ${sessionId} AND event_type = 'chat'), total_gift_count = (SELECT COUNT(*) FROM live_events WHERE session_id = ${sessionId} AND event_type = 'gift'), total_gift_value = COALESCE((SELECT SUM(gift_value * gift_count) FROM live_events WHERE session_id = ${sessionId} AND event_type = 'gift'), 0), peak_viewers = COALESCE((SELECT MAX(viewer_count) FROM live_events WHERE session_id = ${sessionId} AND viewer_count > 0), 0), avg_viewers = COALESCE((SELECT AVG(viewer_count) FROM live_events WHERE session_id = ${sessionId} AND viewer_count > 0), 0), duration_sec = (COALESCE(disconnected_at, ?) - connected_at) / 1000, diamond_per_hour = CASE WHEN duration_sec > 0 THEN total_gift_value * 3600.0 / duration_sec ELSE 0 END WHERE id = ${sessionId} `, Date.now()); } byId(id: number): LiveSessionRow | null { return (this.db.prepare('SELECT * FROM live_sessions WHERE id = ?').get(id) as LiveSessionRow) ?? null; } listActive(): LiveSessionRow[] { return this.db.prepare('SELECT * FROM live_sessions WHERE is_active = 1').all() as LiveSessionRow[]; } listRecent(limit = 50): LiveSessionRow[] { return this.db .prepare('SELECT * FROM live_sessions ORDER BY connected_at DESC LIMIT ?') .all(limit) as LiveSessionRow[]; } } ``` - [ ] **Step 2: 写 unit test** （`live-sessions.test.ts`）—— 重点测： ```typescript import { test, expect, beforeEach } from 'bun:test'; import { openMemoryDb, runMigrations, LiveSessionsRepo } from '../index'; let db: any, repo: LiveSessionsRepo; beforeEach(() => { db = openMemoryDb(); runMigrations(db); repo = new LiveSessionsRepo(db); }); test('startOrReuse: 首次创建新 session', () => { const r = repo.startOrReuse({ room_id: 'r1', streamer_username: '@yoochan', connected_at: Date.now() }); expect(r.reused).toBe(false); expect(r.id).toBeGreaterThan(0); }); test('startOrReuse: 10 分钟内复用旧 session', () => { const t1 = Date.now(); const r1 = repo.startOrReuse({ room_id: 'r1', streamer_username: '@yoochan', connected_at: t1 }); repo.close(r1.id, t1 + 1000); const r2 = repo.startOrReuse({ room_id: 'r1', streamer_username: '@yoochan', connected_at: t1 + 60_000 }); expect(r2.reused).toBe(true); expect(r2.id).toBe(r1.id); }); test('startOrReuse: 超 10 分钟新建 session', () => { const t1 = Date.now() - 700_000; const r1 = repo.startOrReuse({ room_id: 'r1', streamer_username: '@y', connected_at: t1 }); repo.close(r1.id, t1 + 1000); const r2 = repo.startOrReuse({ room_id: 'r1', streamer_username: '@y', connected_at: Date.now() }); expect(r2.reused).toBe(false); expect(r2.id).not.toBe(r1.id); }); test('incrementStats: gift 增量', () => { const { id } = repo.startOrReuse({ room_id: 'r', streamer_username: '@u', connected_at: Date.now() }); repo.incrementStats(id, 'gift', 100, 5, 1000); const row = repo.byId(id)!; expect(row.total_gift_count).toBe(1); expect(row.total_gift_value).toBe(500); expect(row.peak_viewers).toBe(1000); }); test('close 设置 is_active=0 + 计算 duration', () => { const t = Date.now(); const { id } = repo.startOrReuse({ room_id: 'r', streamer_username: '@u', connected_at: t }); repo.close(id, t + 60_000); const row = repo.byId(id)!; expect(row.is_active).toBe(0); expect(row.duration_sec).toBe(60); }); ``` - [ ] **Step 3: 写 `live-events.ts`** （类似结构） ```typescript import type { Database } from 'bun:sqlite'; export interface LiveEventRow { id: number; session_id: number; room_id: string; event_type: string; ts: number; user_uid?: string; user_username?: string; user_nickname?: string; gift_id?: string; gift_name?: string; gift_value?: number; gift_count?: number; viewer_count?: number; like_count?: number; chat_text?: string; raw_json?: string; } export interface NewLiveEvent { session_id: number; room_id: string; event_type: string; ts: number; user_uid?: string; user_username?: string; user_nickname?: string; user_avatar_url?: string; user_fan_level?: number; gift_id?: string; gift_name?: string; gift_value?: number; gift_count?: number; viewer_count?: number; like_count?: number; chat_text?: string; raw_json?: string; } export class LiveEventsRepo { constructor(private db: Database) {} insert(ev: NewLiveEvent): number { const result = this.db.prepare(` INSERT INTO live_events (session_id, room_id, event_type, ts, user_uid, user_username, user_nickname, user_avatar_url, user_fan_level, gift_id, gift_name, gift_value, gift_count, viewer_count, like_count, chat_text, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `).run( ev.session_id, ev.room_id, ev.event_type, ev.ts, ev.user_uid ?? null, ev.user_username ?? null, ev.user_nickname ?? null, ev.user_avatar_url ?? null, ev.user_fan_level ?? null, ev.gift_id ?? null, ev.gift_name ?? null, ev.gift_value ?? null, ev.gift_count ?? null, ev.viewer_count ?? null, ev.like_count ?? null, ev.chat_text ?? null, ev.raw_json ?? null, ); return Number(result.lastInsertRowid); } insertBatch(events: NewLiveEvent[]): number { const insert = this.db.prepare(`INSERT INTO live_events (...) VALUES (...)`); const tx = this.db.transaction((evs: NewLiveEvent[]) => { let count = 0; for (const ev of evs) { // same fields as above count++; } return count; }); return tx(events); } bySession(sessionId: number, eventType?: string, limit = 500): LiveEventRow[] { if (eventType) { return this.db.prepare( 'SELECT * FROM live_events WHERE session_id = ? AND event_type = ? ORDER BY ts DESC LIMIT ?', ).all(sessionId, eventType, limit) as LiveEventRow[]; } return this.db.prepare( 'SELECT * FROM live_events WHERE session_id = ? ORDER BY ts DESC LIMIT ?', ).all(sessionId, limit) as LiveEventRow[]; } giftRanking(sessionId: number, top = 10): { gift_name: string; total: number }[] { return this.db.prepare(` SELECT gift_name, SUM(gift_value * gift_count) as total FROM live_events WHERE session_id = ? AND event_type = 'gift' GROUP BY gift_name ORDER BY total DESC LIMIT ? `).all(sessionId, top) as any[]; } topGifters(sessionId: number, top = 10): { user_username: string; total: number }[] { return this.db.prepare(` SELECT user_username, user_nickname, SUM(gift_value * gift_count) as total FROM live_events WHERE session_id = ? AND event_type = 'gift' AND user_username IS NOT NULL GROUP BY user_username ORDER BY total DESC LIMIT ? `).all(sessionId, top) as any[]; } } ``` - [ ] **Step 4: 跑 test**：`bun test packages/db`，期望 ≥10 个 test 通过 - [ ] **Step 5: Commit** ```bash git commit -m "feat(db): add LiveSessionsRepo + LiveEventsRepo with reuse window logic" ``` --- ## Task 3: Event Extractor（提取 user 字段） **Files:** - Create: `packages/core/src/tiktok-engine/event-extractor.ts` - Test: 同名 .test.ts **迁移源**：catclaw `live-monitor.ts:17-64` 的 `extractUserFields` 函数 - [ ] **Step 1: 写 event-extractor.ts** ```typescript /** * 从 tiktok-live-connector 的 raw event 提取标准化的 user 字段。 * 迁移自 catclaw web/server/tiktok/live-monitor.ts:17-64 * * 关键逻辑： * - badgeList 扫描 fan-club / privilege-icon / gradient-image 类型徽章 → fan_level * - fansClub.data.userBadges 提供 fan-level 数字 * - role 推断顺序: subscriber > moderator > follower > visitor */ export interface ExtractedUserFields { user_uid: string | null; user_username: string | null; user_nickname: string | null; user_avatar_url: string | null; user_verified: 0 | 1; user_fan_level: number | null; user_role: 'subscriber' | 'moderator' | 'follower' | 'visitor' | null; user_is_follower: 0 | 1 | null; user_follower_count: number | null; user_following_count: number | null; } export function extractUserFields(rawEvent: any): ExtractedUserFields { const u = rawEvent?.user ?? {}; const fanLevel = u.fansClub?.data?.userBadges?.[0]?.badgeSceneType ?? u.badgeList?.find((b: any) => b.type === 'fans_club')?.imageBadge?.level ?? null; const roles: string[] = []; if (u.isSubscriber) roles.push('subscriber'); if (u.isModerator) roles.push('moderator'); if (u.followInfo?.followStatus === 1) roles.push('follower'); const role = roles[0] ?? 'visitor'; return { user_uid: u.userId?.toString() ?? null, user_username: u.uniqueId ?? null, user_nickname: u.nickname ?? null, user_avatar_url: u.profilePicture?.urls?.[0] ?? u.avatarThumb?.urls?.[0] ?? null, user_verified: u.isVerified ? 1 : 0, user_fan_level: fanLevel, user_role: role as any, user_is_follower: u.followInfo?.followStatus === 1 ? 1 : 0, user_follower_count: u.followInfo?.followerCount ?? null, user_following_count: u.followInfo?.followingCount ?? null, }; } ``` - [ ] **Step 2: 写 test 用 fixture（直接复用 catclaw 录制的 raw event）** ```typescript import { test, expect } from 'bun:test'; import { extractUserFields } from './event-extractor'; test('extracts uid/username/nickname/avatar from a typical gift event', () => { const raw = { user: { userId: '12345', uniqueId: 'taro_pop', nickname: 'Taro', profilePicture: { urls: ['https://p16.tiktokcdn.com/foo.jpg'] }, isVerified: false, followInfo: { followStatus: 1, followerCount: 100, followingCount: 50 }, badgeList: [{ type: 'fans_club', imageBadge: { level: 12 } }], }, }; const r = extractUserFields(raw); expect(r.user_uid).toBe('12345'); expect(r.user_username).toBe('taro_pop'); expect(r.user_fan_level).toBe(12); expect(r.user_role).toBe('follower'); expect(r.user_is_follower).toBe(1); }); test('handles missing user gracefully', () => { const r = extractUserFields({}); expect(r.user_uid).toBeNull(); expect(r.user_role).toBe('visitor'); }); ``` - [ ] **Step 3: Commit** ```bash git commit -m "feat(tiktok-engine): port extractUserFields from catclaw" ``` --- ## Task 4: RoomConnection 类（迁移 + 改造 live-monitor.ts） **Files:** - Create: `packages/core/src/tiktok-engine/types.ts` - Create: `packages/core/src/tiktok-engine/live-monitor.ts` **迁移源**：catclaw `live-monitor.ts:79-678` - [ ] **Step 1: 写 `types.ts`** ```typescript export type EventType = | 'chat' | 'gift' | 'member' | 'like' | 'follow' | 'share' | 'subscribe' | 'roomInfo' | 'roomUserSequence' | 'social' | 'questionNew' | 'goalUpdate' | 'caption' | 'imDelete' | 'envelope' | 'rankUpdate' | 'rankText' | 'liveIntro' | 'pollAction' | 'detect' | 'linkMicBattle' | 'linkMicArmies' | 'liveEnd' | 'connected' | 'disconnected'; /** 完整 24 种事件，迁移自 catclaw live-monitor.ts:79-121 EVENT_MAP */ export const EVENT_MAP: Record = { WebcastChatMessage: 'chat', WebcastGiftMessage: 'gift', WebcastMemberMessage: 'member', WebcastLikeMessage: 'like', WebcastSocialMessage: 'social', WebcastSubNotifyMessage: 'subscribe', WebcastQuestionNewMessage: 'questionNew', WebcastRoomMessage: 'roomInfo', WebcastRoomUserSeqMessage: 'roomUserSequence', WebcastGoalUpdateMessage: 'goalUpdate', WebcastCaptionMessage: 'caption', WebcastImDeleteMessage: 'imDelete', WebcastEnvelopeMessage: 'envelope', WebcastRankUpdateMessage: 'rankUpdate', WebcastRankTextMessage: 'rankText', WebcastLiveIntroMessage: 'liveIntro', WebcastPollMessage: 'pollAction', WebcastDetectMessage: 'detect', WebcastLinkMicBattle: 'linkMicBattle', WebcastLinkMicArmies: 'linkMicArmies', WebcastControlMessage: 'liveEnd', }; export interface RoomInfo { roomId: string; uniqueId: string; nickname?: string; avatarUrl?: string; followerCount?: number; isVerified?: boolean; anchorLevel?: number; title?: string; category?: string; coverUrl?: string; } export interface NormalizedEvent { type: EventType; ts: number; room_id: string; raw: unknown; // 原始 payload 兜底 user?: ReturnType; gift?: { id: string; name: string; value: number; count: number }; chat?: { text: string }; viewer_count?: number; like_count?: number; } ``` - [ ] **Step 2: 写 `live-monitor.ts`** （RoomConnection 类） ```typescript import { TikTokLiveConnection, WebcastEvent } from 'tiktok-live-connector'; import { logger } from '@pawcast/core'; import { extractUserFields } from './event-extractor'; import { EVENT_MAP, type RoomInfo, type NormalizedEvent, type EventType } from './types'; export interface RoomConnectionOptions { uniqueId: string; /** 自定义 sign provider（Electron BrowserWindow）;catclaw 用了一个 hidden window 抓 X-Bogus 等签名 */ signFetcher?: (url: string) => Promise>; } export type EventHandler = (ev: NormalizedEvent) => void; export class RoomConnection { private conn: TikTokLiveConnection; private handlers = new Set(); private connectedAt: number | null = null; private retryCount = 0; private readonly MAX_RETRY = 5; private readonly BACKOFF = [3000, 6000, 12000, 24000, 48000]; public roomInfo: RoomInfo | null = null; constructor(public readonly opts: RoomConnectionOptions) { this.conn = new TikTokLiveConnection(opts.uniqueId, { // …配置：用 opts.signFetcher 注入签名（Phase 2 已封装好） }); this.bindEvents(); } private bindEvents(): void { for (const [tikName, normalized] of Object.entries(EVENT_MAP)) { this.conn.on(tikName as keyof typeof WebcastEvent, (data: any) => this.dispatch(normalized as EventType, data), ); } this.conn.on('disconnected', () => this.handleDisconnect()); } private dispatch(type: EventType, raw: any): void { const event: NormalizedEvent = { type, ts: Date.now(), room_id: this.roomInfo?.roomId ?? '', raw, user: extractUserFields(raw), }; if (type === 'gift') { event.gift = { id: raw.giftId?.toString() ?? '', name: raw.giftDetails?.giftName ?? '', value: Number(raw.giftDetails?.diamondCount ?? 0), count: Number(raw.repeatCount ?? 1), }; } else if (type === 'chat') { event.chat = { text: raw.comment ?? '' }; } else if (type === 'roomUserSequence') { event.viewer_count = Number(raw.totalUser ?? 0); } else if (type === 'like') { event.like_count = Number(raw.totalLikeCount ?? raw.likeCount ?? 0); } for (const h of this.handlers) h(event); } async connect(): Promise { const state = await this.conn.connect(); this.connectedAt = Date.now(); this.retryCount = 0; this.roomInfo = { roomId: state.roomId.toString(), uniqueId: this.opts.uniqueId, nickname: state.roomInfo?.owner?.nickname, avatarUrl: state.roomInfo?.owner?.avatarThumb?.urls?.[0], followerCount: state.roomInfo?.owner?.followInfo?.followerCount, isVerified: state.roomInfo?.owner?.isVerified, title: state.roomInfo?.title, category: state.roomInfo?.category, }; logger.info('RoomConnection', 'connected', { uniqueId: this.opts.uniqueId, roomId: this.roomInfo.roomId }); this.dispatch('connected', { roomId: this.roomInfo.roomId }); return this.roomInfo; } private async handleDisconnect(): Promise { this.dispatch('disconnected', {}); if (this.retryCount >= this.MAX_RETRY) { logger.error('RoomConnection', 'max retries reached, giving up', { uniqueId: this.opts.uniqueId }); return; } const wait = this.BACKOFF[this.retryCount] ?? 60000; this.retryCount++; logger.warn('RoomConnection', `reconnect in ${wait}ms`, { attempt: this.retryCount }); setTimeout(() => this.connect().catch((e) => logger.error('RoomConnection', 'reconnect failed', e)), wait); } disconnect(): void { this.conn.disconnect(); } on(handler: EventHandler): () => void { this.handlers.add(handler); return () => this.handlers.delete(handler); } } ``` - [ ] **Step 3: 写 unit test**（用 mock 的 TikTokLiveConnection） ```typescript import { test, expect, mock } from 'bun:test'; import { RoomConnection } from './live-monitor'; // Mock tiktok-live-connector mock.module('tiktok-live-connector', () => ({ TikTokLiveConnection: class { on() {} connect() { return Promise.resolve({ roomId: '12345', roomInfo: { owner: { nickname: 'Test' } } }); } disconnect() {} }, })); test('connect resolves and sets roomInfo', async () => { const c = new RoomConnection({ uniqueId: '@test' }); const info = await c.connect(); expect(info.roomId).toBe('12345'); }); test('subscribe to events via on()', async () => { const c = new RoomConnection({ uniqueId: '@test' }); const handler = mock(); c.on(handler); await c.connect(); expect(handler).toHaveBeenCalled(); // emits 'connected' synthetic event }); ``` - [ ] **Step 4: Commit** ```bash git commit -m "feat(tiktok-engine): port RoomConnection with auto-reconnect (5x exp backoff)" ``` --- ## Task 5: Live Scheduler（30s 巡检 + 90s backoff） **Files:** - Create: `packages/core/src/tiktok-engine/live-scheduler.ts` - Test: 同名 **迁移源**：catclaw `live-scan-scheduler.ts:1-150` - [ ] **Step 1: 写 live-scheduler.ts** ```typescript import { logger } from '@pawcast/core'; import type { Database } from 'bun:sqlite'; export interface SchedulerOptions { intervalMs?: number; // 默认 30000 attemptBackoffMs?: number; // 默认 90000 onConnect: (username: string) => Promise; isConnected: (username: string) => boolean; } export class LiveScheduler { private timer: ReturnType | null = null; private readonly intervalMs: number; private readonly backoffMs: number; constructor(private db: Database, private opts: SchedulerOptions) { this.intervalMs = opts.intervalMs ?? 30_000; this.backoffMs = opts.attemptBackoffMs ?? 90_000; } start(): void { if (this.timer) return; this.timer = setInterval(() => this.runOnce().catch((e) => logger.error('LiveScheduler', 'tick failed', e)), this.intervalMs); logger.info('LiveScheduler', 'started', { intervalMs: this.intervalMs }); } stop(): void { if (this.timer) clearInterval(this.timer); this.timer = null; } async runOnce(): Promise { const rooms = this.db.prepare(`SELECT * FROM room_meta WHERE auto_monitor = 1`).all() as any[]; for (const room of rooms) { if (this.opts.isConnected(room.streamer_username)) continue; const lastAttempt = room.last_connect_attempt_at ? Date.parse(room.last_connect_attempt_at) : 0; if (Date.now() - lastAttempt < this.backoffMs) continue; const now = new Date().toISOString(); this.db.prepare(`UPDATE room_meta SET monitor_status = ?, last_connect_attempt_at = ? WHERE streamer_username = ?`) .run('连接中', now, room.streamer_username); try { await this.opts.onConnect(room.streamer_username); this.db.prepare(`UPDATE room_meta SET monitor_status = ?, last_connect_success_at = ? WHERE streamer_username = ?`) .run('监听中', now, room.streamer_username); } catch (err: any) { const isOffline = err?.message?.includes('not live') || err?.message?.includes('offline'); const status = isOffline ? '待开播' : '连接失败'; this.db.prepare(`UPDATE room_meta SET monitor_status = ? WHERE streamer_username = ?`) .run(status, room.streamer_username); } } } } ``` - [ ] **Step 2: 写 test** ```typescript import { test, expect, beforeEach, mock } from 'bun:test'; import { openMemoryDb, runMigrations } from '@pawcast/db'; import { LiveScheduler } from './live-scheduler'; let db: any; beforeEach(() => { db = openMemoryDb(); runMigrations(db); }); test('runOnce skips already connected rooms', async () => { db.prepare(`INSERT INTO room_meta (streamer_username, auto_monitor, updated_at) VALUES (?, 1, ?)`) .run('@yoochan', Date.now()); const onConnect = mock(async () => {}); const sched = new LiveScheduler(db, { onConnect, isConnected: () => true }); await sched.runOnce(); expect(onConnect).not.toHaveBeenCalled(); }); test('runOnce respects 90s backoff', async () => { db.prepare(`INSERT INTO room_meta (streamer_username, auto_monitor, last_connect_attempt_at, updated_at) VALUES (?, 1, ?, ?)`) .run('@yoochan', new Date().toISOString(), Date.now()); const onConnect = mock(async () => {}); const sched = new LiveScheduler(db, { onConnect, isConnected: () => false }); await sched.runOnce(); expect(onConnect).not.toHaveBeenCalled(); // 因为刚刚 attempted }); test('runOnce calls onConnect for due rooms and updates status', async () => { db.prepare(`INSERT INTO room_meta (streamer_username, auto_monitor, updated_at) VALUES (?, 1, ?)`) .run('@yoochan', Date.now()); const onConnect = mock(async () => {}); const sched = new LiveScheduler(db, { onConnect, isConnected: () => false }); await sched.runOnce(); expect(onConnect).toHaveBeenCalledWith('@yoochan'); const row = db.prepare(`SELECT monitor_status FROM room_meta WHERE streamer_username = ?`) .get('@yoochan') as any; expect(row.monitor_status).toBe('监听中'); }); ``` - [ ] **Step 3: Commit** ```bash git commit -m "feat(tiktok-engine): port LiveScheduler with 30s tick + 90s attempt backoff" ``` --- ## Task 6: Manager（多房协调 + worker thread 隔离） **Files:** - Create: `packages/core/src/tiktok-engine/manager.ts` - Test: 同名 为避免 8 房并发解析事件阻塞主线程，每个房间一个 worker thread。 - [ ] **Step 1: 写 manager.ts** ```typescript import { logger } from '@pawcast/core'; import { Worker } from 'node:worker_threads'; import type { NormalizedEvent } from './types'; export interface ManagerOptions { maxConcurrent?: number; // 默认 16 workerPath: string; // worker 入口绝对路径（编译后的 .js） onEvent: (uniqueId: string, ev: NormalizedEvent) => void; } interface RoomEntry { uniqueId: string; worker: Worker; connectedAt: number; } export class LiveMonitorManager { private rooms = new Map(); private readonly maxConcurrent: number; constructor(private opts: ManagerOptions) { this.maxConcurrent = opts.maxConcurrent ?? 16; } async connect(uniqueId: string): Promise { if (this.rooms.has(uniqueId)) { logger.info('Manager', 'already connected', { uniqueId }); return; } if (this.rooms.size >= this.maxConcurrent) { throw new Error(`max concurrent rooms reached (${this.maxConcurrent})`); } const worker = new Worker(this.opts.workerPath, { workerData: { uniqueId } }); worker.on('message', (msg: { type: 'event'; ev: NormalizedEvent } | { type: 'error'; error: string }) => { if (msg.type === 'event') { this.opts.onEvent(uniqueId, msg.ev); } else if (msg.type === 'error') { logger.error('Manager', 'worker error', { uniqueId, error: msg.error }); } }); worker.on('exit', () => { this.rooms.delete(uniqueId); logger.info('Manager', 'worker exited', { uniqueId }); }); this.rooms.set(uniqueId, { uniqueId, worker, connectedAt: Date.now() }); } async disconnect(uniqueId: string): Promise { const entry = this.rooms.get(uniqueId); if (!entry) return; entry.worker.postMessage({ cmd: 'disconnect' }); await entry.worker.terminate(); this.rooms.delete(uniqueId); } list(): string[] { return Array.from(this.rooms.keys()); } isConnected(uniqueId: string): boolean { return this.rooms.has(uniqueId); } async stopAll(): Promise { await Promise.all([...this.rooms.keys()].map((u) => this.disconnect(u))); } } ``` - [ ] **Step 2: 写 worker 入口 `live-monitor-worker.ts`** ```typescript import { workerData, parentPort } from 'node:worker_threads'; import { RoomConnection } from './live-monitor'; const { uniqueId } = workerData as { uniqueId: string }; const conn = new RoomConnection({ uniqueId }); conn.on((ev) => parentPort?.postMessage({ type: 'event', ev })); parentPort?.on('message', (msg: { cmd: 'disconnect' }) => { if (msg.cmd === 'disconnect') { conn.disconnect(); process.exit(0); } }); conn.connect().catch((err) => { parentPort?.postMessage({ type: 'error', error: err.message }); process.exit(1); }); ``` - [ ] **Step 3: Commit** ```bash git commit -m "feat(tiktok-engine): add LiveMonitorManager with worker thread per room" ``` --- ## Task 7: IPC Contract for Monitor **Files:** - Create: `packages/ipc-contract/src/live-monitor.ts` - [ ] **Step 1: 写 schema** ```typescript import { z } from 'zod'; const Method = (input: I, output: O) => ({ input, output }) as { input: I; output: O }; export const RoomMetaSchema = z.object({ streamer_username: z.string(), streamer_nickname: z.string().nullable(), streamer_avatar_url: z.string().nullable(), auto_monitor: z.union([z.literal(0), z.literal(1)]), monitor_status: z.string().nullable(), group_name: z.string().nullable(), notes: z.string().nullable(), }); export const LiveSessionSchema = z.object({ id: z.number(), room_id: z.string(), streamer_username: z.string(), connected_at: z.number(), disconnected_at: z.number().nullable(), is_active: z.union([z.literal(0), z.literal(1)]), total_gift_value: z.number(), total_chat: z.number(), peak_viewers: z.number(), duration_sec: z.number(), diamond_per_hour: z.number(), }); export const MonitorContract = { 'monitor.listRooms': Method(z.void(), z.array(RoomMetaSchema)), 'monitor.addRoom': Method( z.object({ streamer_username: z.string(), group_name: z.string().optional() }), z.object({ ok: z.literal(true) }), ), 'monitor.removeRoom': Method(z.object({ streamer_username: z.string() }), z.object({ ok: z.literal(true) })), 'monitor.setAutoMonitor': Method( z.object({ streamer_username: z.string(), enabled: z.boolean() }), z.object({ ok: z.literal(true) }), ), 'monitor.listActiveSessions': Method(z.void(), z.array(LiveSessionSchema)), 'monitor.getSessionDetail': Method( z.object({ session_id: z.number() }), z.object({ session: LiveSessionSchema, gifts: z.array(z.any()), chats: z.array(z.any()) }), ), 'monitor.connectNow': Method(z.object({ streamer_username: z.string() }), z.object({ ok: z.literal(true) })), } as const; export const MonitorEvents = { 'monitor.event': z.object({ uniqueId: z.string(), type: z.string(), ts: z.number(), payload: z.any(), }), 'monitor.session_started': z.object({ session_id: z.number(), uniqueId: z.string() }), 'monitor.session_ended': z.object({ session_id: z.number(), uniqueId: z.string() }), 'monitor.alert': z.object({ rule: z.string(), uniqueId: z.string(), message: z.string(), }), } as const; ``` - [ ] **Step 2: 注册到根 contract** —— `packages/ipc-contract/src/index.ts` 加 export - [ ] **Step 3: Commit** ```bash git commit -m "feat(ipc-contract): add MonitorContract + Monitor events" ``` --- ## Task 8-15（剩余任务，简洁列出） 由于 Phase 7 体量极大，剩余 task 按主题归纳列出（每个 task 内部步骤同样按 TDD 节奏走）： - [ ] **Task 8**: PK Collector（迁移 catclaw `pk-collector.ts:1-56`，订阅 `linkMicBattle` 事件，写 `pk_battles` 表） - [ ] **Task 9**: Recorder（每个 room 一个 JSONL 文件，事件流原样落盘到 `~/.pawcast/recordings/`） - [ ] **Task 10**: AlertEngine（5 内置规则 + 自定义 DSL，监听 `monitor.event` 并匹配阈值） - [ ] **Task 11**: 主进程 IPC 路由 wire-up（实例化 Manager + Scheduler + Recorder，绑定 IPC handler） - [ ] **Task 12**: Renderer Monitor 主页 (`pages/Monitor/index.tsx`)，4×2 房间网格 + 右侧栏（PK / 速率柱图 / 事件流），使用 `useLiveQuery` hook 自动刷新 - [ ] **Task 13**: RoomDetail 弹窗（4 Tab：Now / Top Gifters / Events / Members） - [ ] **Task 14**: PKTracker 详情弹窗（双方对照 + 比分时间线 + 胜率预测） - [ ] **Task 15**: 集成 e2e 测试：用录制的 JSONL 灌假事件流，断言 UI 渲染 + DB 写入正确 --- ## Phase 7 准出清单 - [ ] `bun test packages/core/src/tiktok-engine/` 全部通过（≥30 个测试） - [ ] `bun test packages/db/src/repositories/live-*.test.ts` 全部通过 - [ ] 实际连接一个真实房间（用 testing 账号）能看到事件流入 DB - [ ] 同时连 8 个房间，CPU < 30% / RAM < 800MB（M1 Mac） - [ ] Renderer Monitor Tab 显示房间网格 + 实时观众/礼物数字跳动 - [ ] PK 战役自动识别 + 比分时间线显示 - [ ] 事件流右侧栏滚动刷新 - [ ] session 自动 close + duration_sec 计算正确 --- ## Self-Review - ✅ Spec coverage: live-monitor 设计稿描述的功能全部映射到 task（8 房网格 / 单房详情 / PK 追踪 / 告警 / 跨房分析） - ✅ catclaw 现有功能 1:1 迁移（30s 调度 + 90s backoff + 10min reuse + 5x retry + 24 事件类型） - ✅ Type consistency: `RoomConnection` / `LiveMonitorManager` / `LiveScheduler` 命名贯穿 - ⚠️ design HTML 中的"PK 详情"功能 catclaw 未实装，本 plan 在 Task 8 实装它（按设计稿的 schema） - ⚠️ design HTML 中的"跨房间 Heatmap"功能本 phase 暂不实装，留 v0.2 ## 与 catclaw 的差异点（迁移时需特别留意） | 点 | catclaw | Pawcast 改造 | |---|---|---| | 进程模型 | 主线程跑所有 WS | **worker thread 每房一个**（避免阻塞 IPC） | | DB schema | `tk_*` 前缀 | **去前缀**（项目本身就是 TikTok 主题） | | 浏览器签名 | Electron BrowserWindow + bridge | 复用 catclaw 同一架构（Phase 2 已封装） | | 事件存储 | flushEvents 批量写（catclaw 256 batch） | **同步实现**（防止丢失），future 改 batch | | 重连策略 | 5 次指数退避 | **保持一致**（已验证生产稳定） | | Session reuse | 10 分钟同 room 复用 | **保持一致** |