diff --git a/.cursor/rules/project-context.mdc b/.cursor/rules/project-context.mdc index 2e5ba11..d2295e8 100644 --- a/.cursor/rules/project-context.mdc +++ b/.cursor/rules/project-context.mdc @@ -27,7 +27,7 @@ alwaysApply: true * 单端 WS 连接架构（沿用），不做多端已读同步（设计文档 §3.1/§3.5/§九 已修订，多端改造推迟到 Phase 2f/二期） * 专用设计：`docs/plans/2026-04-20-phase2e-1-design.md`；实施计划：`docs/plans/2026-04-20-phase2e-1-implementation.plan.md`；验证报告：`test-report-phase2e-1-notification.md` * API 文档：`docs/api/frontend/notify.md` - - 2e-2 会议 MVP（约 17 天）🚧 **代码开发中**（Task 0-7 ✅ / Task 8-16 待执行）：mediasoup Node.js 独立 `media-server/` + 即时会议（≤8 人）+ 密码/邀请链接/通知邀请三合一 + 设备预览页 + 主持人四件套 + 会议内聊天 + 双态部署（本机 + 公网 coturn）+ 响应式（桌面/手机） + - 2e-2 会议 MVP（约 17 天）🚧 **代码开发中**（Task 0-8 ✅ / Task 9-16 待执行）：mediasoup Node.js 独立 `media-server/` + 即时会议（≤8 人）+ 密码/邀请链接/通知邀请三合一 + 设备预览页 + 主持人四件套 + 会议内聊天 + **会议生命周期状态机**（host 宽限期 + 自动转让 + 空房 TTL）+ 双态部署（本机 + 公网 coturn）+ 响应式（桌面/手机） * 专用设计：`docs/plans/2026-04-21-phase2e-2-design.md`（16 章节）；实施计划：`docs/plans/2026-04-21-phase2e-2-implementation.plan.md`（17 个 Task） * 11 项关键决策已锁定（D01-D11），详见设计文档 §三 * **重要修订**：`meeting_rooms.password` → `password_hash`（bcrypt），新增 `meeting_chats` 表 + `ended_reason` / `left_reason` 字段，新增 `echo:meeting:invite:{token}` / `host_grace:{code}` Redis key @@ -36,6 +36,7 @@ alwaysApply: true * **Task 2 ✅ 9 个内部 REST API 完成（2026-04-21，含代码审查修复）**：`media-server/` 落地 Router/Transport/Producer/Consumer 四类资源的 9 个接口，全部挂 `/internal/v1/*` 前缀；**zod 手动 parse + 全局 errorHandler** 方案（不引入 fastify-type-provider-zod 避免 zod v4 依赖冲突）；`AppError` 统一错误码（`NOT_FOUND`/`CONFLICT`/`CAN_NOT_CONSUME`/`ROUTER_LIMIT_EXCEEDED`/`MEDIASOUP_ERROR`）+ `VALIDATION_ERROR`/`UNAUTHORIZED`/`INTERNAL_ERROR`；所有资源用 **Map + `observer.once('close')` 自清理**，`producerclose` 级联关闭下游 consumer；Consumer 强制 `paused:true` 创建 + `/resume` 独立接口；direction 强约束（recv transport 拒 produce、send transport 拒 consume）。**code-reviewer 子代理"有条件通过"**，2 Major + 4 高价值 Minor 当场修复：(1) M1 `_clearXxxMap` 新增 `assertTestOnly` 守卫（生产误调用直接抛错）；(2) M2 新增 `src/schemas/rtp.ts` 对 `rtpParameters` / `rtpCapabilities` 做 codecs 浅层校验（mimeType/clockRate/payloadType 必填、codecs 数组 ≥1），消除 `as unknown as` 双跳断言；(3) m1 `connectTransport` 改乐观锁（先置位再 await）；(4) m2 改读 `consumer.producerPaused`；(5) m3 `producerclose` 改为 `once`；(6) m5 `internal-auth` 改为反向白名单 `PRIVATE_PATH_PREFIXES = ['/internal/']`（默认开放）。**65 个 vitest 测试全过（~1s）、覆盖率 82.87%/75.83%/91.3%/82.87%**（stmts/branches/funcs/lines，较首版 +2pp）；9 接口 happy path + 6 类错误路径人工 curl 全部按预期返回（201/200/400/401/404/409）。产出：`src/schemas/*`（6 文件，新增 `rtp.ts`）+ `src/services/*`（4 文件）+ `src/routes/*`（4 文件）+ `src/middlewares/{error-handler,internal-auth}.ts` + `src/utils/{errors,test-guard}.ts` + `src/mediasoup/codecs.ts` + `vitest.config.ts` + `tests/*`（8 spec 文件，含 `test-guard.spec.ts`）。余下 Minor/Nits（m4/m6~m10、n1~n10）登记至 Task 16 收尾清单 * **Task 3 ✅ Go meeting 模块数据库 DDL + Model + DAO 完成（2026-04-21）**：三张持久化表（`meeting_rooms` / `meeting_participants` / `meeting_chats`）落地 PostgreSQL，DDL 同时写入 `init.sql`（全量初始化）与 `phase2e2_migration.sql`（幂等增量升级）。Go 侧 `backend/go-service/app/meeting/{model,dao}` + 统一常量 `app/constants/meeting.go`：3 个 model + 3 个 DAO（`meeting_room_dao.go` 9 方法 / `meeting_participant_dao.go` 11 方法 / `meeting_chat_dao.go` 4 方法），共 24 个持久化方法；`JoinRoom` 事务内复用离会后的旧记录（`left_at=NULL,joined_at=NOW,duration=0`）避免审计表污染；`LeaveRoom` 用 `EXTRACT(EPOCH FROM (? - joined_at))::INT` 走 DB 时间防跨时区漂移；`TransferHost` 事务链（`role=1→0` + `role=0→1`）；`FindActiveByUser` 用 JOIN 校验用户单点参会；`MarkEnded` 乐观锁防重复覆盖 `ended_reason`；`ListExpiredForCleanup` 供后续清理任务批量扫描。**关键风格修正（偏离实施计划草案）**：按 `project-context` 第 11 条「代码风格全局一致（最高优先级）」，常量归入 `app/constants/meeting.go` 单文件（与 `group.go`/`notify.go` 同构），而非草案的 `app/meeting/constants/*.go`；时间字段统一 `TIMESTAMP(0)` 取代草案的 `TIMESTAMPTZ` 对齐项目所有现有表；冗余 `idx_meeting_rooms_code` 移除（`room_code UNIQUE` 已自动建索引）。验证：`go build ./...` / `go vet ./...` / `ReadLints` 零错误；psql 集成脚本跑通 8 场景（CRUD + 双 UNIQUE 约束 + 主持人转让事务 + `duration=10s` 精确匹配 + CASCADE 清零）。延续项目 Go 侧"零 `_test.go`"风格（用代码审查 + psql 真库验证 + Playwright E2E 三层守护） * **Task 4 ✅ Go meeting 模块 service/controller/router 骨架完成（2026-04-21）**：`app/meeting/` 补齐 service/controller/router/provider 四件套，接口 → 实现按设计文档 §5.3 一一对齐。产出：(1) `service/interfaces.go` 定义 3 个外部依赖接口（`NotifyPusher` / `UserInfoResolver` / `OnlineChecker`），解耦 notify/contact/ws 模块避免循环依赖；`OnlineChecker.IsOnline` 签名与现存 `ws.OnlineService` 一致（返回单 `bool`）；(2) `service/meeting_service.go` 声明 `MeetingService` + 8 个 sentinel error + 17 个业务方法空实现，全部返回 `ErrNotImplemented`；(3) `controller/meeting_controller.go` 12 个 Gin 处理器 + `responseNotImplemented`（501）+ `requireUserID` 辅助；(4) `router.go` 12 条路由挂载到 `/api/v1/meeting/*` 并统一套 `jwtAuth` 中间件；(5) `provider.go` 定义 `MeetingSet = wire.NewSet(DAO×3, Service, Controller)`；(6) 全局 `app/provider/wire.go` 挂入 `meetingApp.MeetingSet` + 3 条 `wire.Bind`（`NotifyPusher→NotifyService` / `UserInfoResolver→FriendshipDAO` / `OnlineChecker→ws.OnlineService`）；(7) `app/provider/provider.go` `App` 加 `MeetingService/MeetingController` 字段；(8) `router/router.go` 调用 `meetingApp.RegisterRoutes`。**顺手修复存量 bug**：`admin/provider.go` 补齐 `MessageManage{DAO,Service,Controller}` 三个 provider，解决旧版 `wire` 重生成报"no provider found"的遗留问题。验证：`go build ./...` / `go vet ./...` / `wire ./app/provider` 全绿；`GIN_MODE=debug` 启动 server 日志打印全部 12 条 `[GIN-debug] ... meeting/controller.(*MeetingController).Xxx-fm`；curl 无 token 打 3 条代表性路由均返回 401 `缺少认证信息`，JWT 中间件生效 + * **Task 8 ✅ 会议生命周期状态机 + Router 幂等双层防御（2026-04-21）**：设计 §6.5 的 5 类状态跃迁副作用（host 掉线宽限期 / 宽限期内重连 / 自动转让 / 空房 TTL / TTL 过期销毁）全部落地；同步修复 Task 7 遗留的"CreateRoom + JoinRoom 重复调 `CreateRouter`"行为。产出：(1) `config/config.go` 新增 `MeetingConfig{HostGraceSeconds=120, EmptyRoomTTLSeconds=300, CleanupIntervalSeconds=30, StaleRoomHours=4}` + dev/docker yaml；(2) `app/meeting/service/meeting_lifecycle_service.go`（新 470 行）封装 `OnHostDisconnect` / `OnHostReconnect` / `HandleHostGraceExpired` / `OnAllMembersLeft` / `CancelEmptyTTL` / `HandleEmptyRoomExpired` 6 钩子，`sync.Map` 管理本地 `time.AfterFunc` timer + Redis key 双保险；`RescheduleFromRedis` 重启按剩余 PTTL 重建 timer；**Redis key TTL = 业务时长 + max(CleanupIntervalSeconds*2, 30s) buffer**（调试期实测发现若 Redis TTL 与本地 timer 同步到期，`DEL` 会因 key 已被 Redis 自动过期返回 0 被误判为"已被其他路径处理"而跳过业务逻辑，加 buffer 让本地 timer 先触发 DEL 命中）；并发保护靠入口 `DEL` 返回值（1 = 本次处理，0 = 已被其他路径处理）；(3) `app/meeting/dao/meeting_room_dao.go` 新增 `ListStaleActive(hoursAgo, limit)` 扫陈旧活跃房间；(4) `app/meeting/service/interfaces.go` `MediaOrchestrator` 新增 `ResolveRouterID(roomCode) (string, bool)` 让业务层直接查缓存不创建；(5) `app/meeting/service/http_media_orchestrator.go` `CreateRouter` 入口查 `sync.Map` 命中直接返回（HTTP 层幂等防御）；(6) `app/meeting/service/meeting_service.go` `JoinRoom` **移除 `CreateRouter` 调用**改 `lifecycleSvc.CancelEmptyTTL` + `ResolveRouterID`；`LeaveRoom` 空房分支改调 `lifecycleSvc.OnAllMembersLeft` 不再立即销毁房间；(7) `app/meeting/service/meeting_signal_service.go` 新增 `OnWSDisconnect(userID)` 实现 `ws.MeetingDisconnectHook` 接口（host 掉线调宽限期 + 非 host 仅清 media 资源，决策 `q1_nonhost_disconnect=a1_keep_current` 不动 participant 表，长期不活跃由 4 小时后台兜底），`OnRoomJoin` 追加 host 重连钩子；(8) `app/ws/handler.go` 定义 `MeetingDisconnectHook` 接口 + `SetMeetingDisconnectHook` 注入，解耦 ws → meeting 反向依赖；(9) `app/meeting/task/meeting_cleanup_task.go`（新 220 行）启动时 `RescheduleFromRedis`；每 `CleanupIntervalSeconds` 秒 `ScanExpired` 兜底扫 `host_grace:*` / `empty_ttl:*` + 扫 stale active rooms → `MarkEnded(reason=system_error)`；每小时清理 Ended 会议旧聊天；(10) `app/provider/{provider,wire,wire_gen}.go` + `cmd/server/main.go` 注入新 service/task + `wire.Bind(new(ws.MeetingDisconnectHook), new(*service.MeetingSignalService))` + 启动/停止 cleanup task；(11) `media-server/src/app.ts` `/internal/info` 追加 `stats.routers` 供 E2E 断言 Router 幂等；(12) `docs/verify/meeting_t8_verify.mjs`（新 320 行）5 场景 E2E 脚本（用 env 加速 TTL 到 3s 跑测）。**关键设计决策**：(a) Redis TTL 加 buffer 避免 DEL 与自动过期并发误判（修复 bug 的核心）；(b) 本地 timer 低延迟 + cleanup task 兜底，多实例由入口 DEL 保证唯一处理权；(c) Router 幂等双层防御（业务层 `JoinRoom` 不创建 + HTTP 层 `sync.Map` 命中直接返回，决策 `q2_router_dedup=a2_both`）；(d) 全员 leave 不再立即销毁房间，由 `empty_ttl` + 本地 timer 实现"可复活"语义；(e) 普通成员 WS 断开仅清 media 资源不动 DB。验证：`go build`/`go vet`/`ReadLints` 全绿（保留 Task 7 遗留 pre-existing 警告）；启动 media-server + go-service 跑 E2E **20/20 PASS**，5 场景全部验证（host 宽限期过期自动转让 + 宽限期内重连保留 + 空房 TTL 复活 + 空房 TTL 过期销毁 + JoinRoom 不再重复 Router） * **Task 7 ✅ HTTPMediaOrchestrator Go↔Node 媒体链路打通（2026-04-21）**：`NoopMediaOrchestrator` 替换为真实 HTTP 实现，`MeetingService` / `MeetingSignalService` 的 9 个媒体操作全部走 `POST/DELETE /internal/v1/*` 调到 Node media-server。产出：(1) `config/config.go` 新增 `MediaServerConfig{BaseURL, InternalToken, TimeoutMS(5000), CloseTimeoutMS(2000), CloseRetry(2)}` + dev/docker 两份 yaml 配置（dev 直连 `localhost:3300`，docker 走服务名 `media-server:3300`）；(2) `app/meeting/service/http_media_orchestrator.go`（新 340 行）用 `net/http` 标准库实现 `MediaOrchestrator` 8 方法，`context.WithTimeout` 驱动差异化超时，创建类一次性透传，**关闭类指数退避 200ms→500ms 最多 CloseRetry+1 次**（幂等安全）；`X-Internal-Token` header 与 `media-server/.env MEDIA_INTERNAL_TOKEN` 配对；错误类型 `ErrMediaResourceNotFound`（404）与 `ErrMediaServerError`（5xx/超时/网络错）可供 `errors.Is` 精准区分；**`sync.Map` 本地 `roomCode ↔ routerID` 缓存**兼容设计 §6.6 `CloseRouter(roomCode)` 签名但 Node 以 routerID 为主键的约束，go-service 重启缓存丢失与 Node 重启 Router 释放状态自然同步；(3) `app/meeting/provider.go` 切换 `wire.Bind` 到 `*HTTPMediaOrchestrator`，`wire_gen.go` 再生；(4) `docs/verify/meeting_t7_verify.mjs`（新 140 行）E2E 验证脚本。**关键设计决策**：(a) 关闭类幂等重试 vs 创建类一次性透传 —— 防孤儿资源；(b) 错误语义二分 —— 让 404 关闭转 nil、5xx 透传 WS ACK；(c) 接口维持 8 方法不含 `ResumeConsumer` —— Node REST 已就绪但 WS 契约未暴露，留给 Task 9 前端 mediasoup-client 接入时按需补齐。**已知遗留**：(i) CreateRoom + JoinRoom 各自调 `CreateRouter` 重复创建，Task 8 需改为"仅首次创建 Router 其余复用"；(ii) `/transports/:id/stats` Node 暂未实现，Task 10 可观测性补齐。验证：`go build`/`go vet` 全绿；启动 media-server(Node 3300) + go-service(Go 8085)，E2E **16/16 PASS** —— REST 创建/加入/结束会议触发 Router 创建/销毁，WS `transport.create` 返回真实 mediasoup `id`（非 noop- 前缀）+ 非空 `iceCandidates[]` + `dtlsParameters.fingerprints[]`，关闭虚构 producer 触发 Node 404 → Go 幂等转 ok；media-server 日志对齐 `router created × 2 / webrtc transport created / router closed explicitly` * **Task 6 ✅ WebSocket 信令协议 13 事件全量落地（2026-04-21）**：`meeting.*` 事件族从 Task 5 的 `PublishToUser` 循环升级为完整 WS 信令协议。产出：(1) `app/constants/meeting.go` WS 事件常量与设计 §6.3 对齐（3 房间 + 5 成员 + 5 媒体 + 1 聊天）+ `MeetingWSClientEvents` 白名单限制客户端仅能发起 8 个 C→S 事件（防恶意客户端伪造 `room.ended` 等广播）；(2) `service/interfaces.go` 扩容 `MediaOrchestrator` 至 9 方法（Router/Transport/Producer/Consumer 全生命周期）+ 配套 5 个 DTO + `NoopMediaOrchestrator` 9 占位实现；(3) **新建 `service/meeting_broadcaster.go`**（75 行）抽离 `BroadcastToMeeting` / `PublishToUser` 统一广播层，REST + WS 共用；(4) `service/meeting_service.go` 12 方法改调 `broadcaster.*` 不再直连 `ws.PubSub`；(5) **新建 `service/meeting_signal_service.go`**（430 行）承载 8 C→S 事件业务：`OnRoomJoin`/`OnRoomLeave`/`OnMemberStateChanged`/`OnTransportCreate/Connect`/`OnProduceStart`/`OnConsumeStart`/`OnProducerClose`；**Redis 资源追踪** `echo:meeting:resources:{room_id}:{user_id}`（Set，TTL 1 小时）+ `cleanupUserResources`（WS 断开钩子遍历清理 transport/producer/consumer）+ host 权限校验（非 host 改他人状态返回 `-1 仅主持人可执行此操作`）；(6) **新建 `controller/meeting_ws_handler.go`**（200 行）薄层：构造时 `hub.RegisterEvent` 注册 8 C→S 事件，每 handler 仅 JSON 反序列化 + 调 `signalSvc.On*` + 构造 ACK；(7) `provider.go` + `wire_gen.go` 扩充三个新 provider；(8) `docs/api/frontend/meeting.md` 追加 §WebSocket 信令协议（200 行）：16 事件总览 + 8 C→S 完整契约 + S→C 广播契约 + 架构 + 错误处理。**关键设计决策**：(a) Broadcaster 单独抽层让 Task 5 REST 广播与 Task 6 WS 广播复用同一对象；(b) C→S 白名单防伪造；(c) Redis Set 追踪媒体资源即使 Go 进程崩溃也不泄漏 mediasoup 端；(d) MediaOrchestrator 接口完整化让 Task 7 只需替换绑定不改 signal service / handler 代码；(e) ACK `code=0/-1` 与 REST 领域错误口径完全一致；(f) `meeting.room.leave` 只清 WS 资源不改 participant 表（真正离会仍需 REST `/leave`），允许客户端 WS 重连刷新 transport 不退会。验证：`go build` / `go vet` / `wire` 全绿；端到端 WS 冒烟 `/tmp/meeting_ws_t6_test.mjs` **PASS=18 / FAIL=0**，覆盖 8 C→S 白名单事件 + 3 S→C 广播 + 3 错误路径（非 host 越权/不存在会议号/WS leave 资源清理）。测试期间修复 `waitEvent` 死循环 bug（pop→push 自循环）改为 stash 缓冲区归还模式 * **Task 5 ✅ Go meeting 模块 12 个 REST 接口业务逻辑全量落地（2026-04-21）**：`MeetingService` + `MeetingController` 从 501 占位升级为完整实现，对接 PostgreSQL / Redis / NotifyPusher / PubSub / MediaOrchestrator。产出：(1) `app/dto/meeting_dto.go` 13 个 DTO（3 基础 + 10 请求/响应）；(2) `pkg/utils/meeting_code.go` 生成 9 位 `XXX-XXX-XXX` 会议号 + 32 位 hex 邀请令牌；(3) `service/meeting_service.go` 12 业务方法 + 11 领域错误 + `assertIsActiveParticipant`/`assertIsHost`/`generateUniqueRoomCode`/`broadcastToActiveParticipants` 辅助；(4) `controller/meeting_controller.go` 12 Gin 处理器 + `handleError` 领域错误 → HTTP 映射 + `roomToDTO`/`participantToDTO`/`chatToDTO` 转换；(5) `service/interfaces.go` 新增 `MediaOrchestrator` 接口 + `NoopMediaOrchestrator` 占位（Task 7 替换）；(6) `provider.go` 注册 `NoopMediaOrchestrator`；(7) `router.go` 路径对齐设计 `GET /rooms/mine` + `POST /invite-tokens/:token/redeem`；(8) `docs/api/frontend/meeting.md` 重写为 280 行的 12 接口完整文档。**DAO 契约修复**：`meeting_room_dao.GetByID/GetByCode` + `meeting_participant_dao.GetByRoomAndUser/FindActiveByUser` 全部将 `gorm.ErrRecordNotFound` 转换为 `(nil, nil)`，由 service 统一 `result == nil` 判定，消除 500 误报。**关键设计决策**：(a) 单点参会用 `meeting_participants` JOIN `meeting_rooms.status != 2` 判断；(b) 密码限流用 `echo:meeting:pwd:fail:{code}:{user_id}` 5 次锁 10 分钟（`ErrMeetingPasswordLocked`）；(c) host 离会若还有其他活跃成员自动转让给"最早加入者"并广播 `meeting.host.changed`；(d) 邀请 token 不返回给调用方，仅通过 `NotifyPusher.PushBatch.Extra.invite_token` 定向下发；兑换后保留 60 秒冗余由 Redis TTL 自然过期；(e) 创建类接口 201、动作类 200、领域错误按 404/403/400 三档映射。**Stub 策略**：`MediaOrchestrator.CreateRouter/CloseRouter` 当前 Noop（Task 7 接入 HTTPMediaOrchestrator）；WS 广播暂用 `pubsub.PublishToUser` 逐人循环（Task 6 封装为 `BroadcastToMeeting` 无感替换）；`NotifyPusher.PushBatch` 复用 Phase 2e-1 成果。验证：`go build` / `go vet` / `wire` 全绿；启动 server + 3 用户端到端脚本 `/tmp/meeting_t5_test.sh` **PASS=19 / FAIL=0**，覆盖 12 接口 happy path + 5 类错误路径（密码错/房间不存在/单点参会冲突/非 host 越权/邀请链接失效） diff --git a/backend/go-service/app/meeting/dao/meeting_room_dao.go b/backend/go-service/app/meeting/dao/meeting_room_dao.go index b987973..78b1c6d 100644 --- a/backend/go-service/app/meeting/dao/meeting_room_dao.go +++ b/backend/go-service/app/meeting/dao/meeting_room_dao.go @@ -160,6 +160,27 @@ func (d *MeetingRoomDAO) ListByHost(ctx context.Context, hostID int64, status in return rooms, total, err } +// ListStaleActive 返回 status != Ended 且 started_at（或 created_at 若未开始）早于 hoursAgo 小时的房间 ID +// 用于定时任务兜底清理"创建 / Active 但长期无任何成员活动的僵尸房间"（设计 §11.4 的 4 小时兜底） +// 排序：最早 started_at 在前，优先清理最老 +func (d *MeetingRoomDAO) ListStaleActive(ctx context.Context, hoursAgo int, limit int) ([]int64, error) { + funcName := "dao.meeting_room_dao.ListStaleActive" + cutoff := time.Now().Add(-time.Duration(hoursAgo) * time.Hour) + + var ids []int64 + err := d.db.WithContext(ctx). + Model(&model.MeetingRoom{}). + Where("status != ? AND COALESCE(started_at, created_at) < ?", + constants.MeetingStatusEnded, cutoff). + Order("COALESCE(started_at, created_at) ASC"). + Limit(limit). + Pluck("id", &ids).Error + if err != nil { + logs.Error(ctx, funcName, "查询 stale 活跃房间失败", zap.Error(err)) + } + return ids, err +} + // ListExpiredForCleanup 返回已结束且超过指定小时数的房间 ID 列表 // 供定时任务清理 meeting_chats 使用；room 本身保留归档不删 // 返回 limit 上限用于分批处理，避免一次捞太多 diff --git a/backend/go-service/app/meeting/provider.go b/backend/go-service/app/meeting/provider.go index adeaddb..cbb21e7 100644 --- a/backend/go-service/app/meeting/provider.go +++ b/backend/go-service/app/meeting/provider.go @@ -4,6 +4,7 @@ import ( "github.com/echochat/backend/app/meeting/controller" "github.com/echochat/backend/app/meeting/dao" "github.com/echochat/backend/app/meeting/service" + "github.com/echochat/backend/app/meeting/task" "github.com/google/wire" ) @@ -20,10 +21,12 @@ var MeetingSet = wire.NewSet( dao.NewMeetingParticipantDAO, dao.NewMeetingChatDAO, service.NewMeetingBroadcaster, + service.NewMeetingLifecycleService, // Task 8 新增：会议生命周期状态机 service.NewMeetingService, service.NewMeetingSignalService, controller.NewMeetingController, controller.NewMeetingWSHandler, + task.NewMeetingCleanupTask, // Task 8 新增：生命周期兜底定时任务 // MediaOrchestrator 使用真实 HTTP 实现（Task 7 落地） // 注入 *config.Config，通过 cfg.MediaServer 获取 base_url / internal_token / timeout 等配置 diff --git a/backend/go-service/app/meeting/service/http_media_orchestrator.go b/backend/go-service/app/meeting/service/http_media_orchestrator.go index b525fa8..f972007 100644 --- a/backend/go-service/app/meeting/service/http_media_orchestrator.go +++ b/backend/go-service/app/meeting/service/http_media_orchestrator.go @@ -85,9 +85,21 @@ func NewHTTPMediaOrchestrator(cfg *config.Config) *HTTPMediaOrchestrator { // CreateRouter 调用 POST /internal/v1/routers // 成功后在本地缓存 roomCode → routerID 映射供 CloseRouter 使用 +// Task 8 幂等防御（双层，与业务层 JoinRoom 不再主动调用形成互补）： +// - 命中缓存直接返回已缓存 routerID，不发起 HTTP 请求 +// - 仅在缓存缺失时真正调 Node；避免业务误多次调用产生 Node 侧重复 Router func (h *HTTPMediaOrchestrator) CreateRouter(ctx context.Context, roomCode string) (string, error) { funcName := "service.http_media_orchestrator.CreateRouter" + if cached, ok := h.roomRouterIDs.Load(roomCode); ok { + if rid, _ := cached.(string); rid != "" { + logs.Debug(ctx, funcName, "命中本地 Router 缓存，跳过 Node 调用", + zap.String("room_code", roomCode), + zap.String("router_id", rid)) + return rid, nil + } + } + reqBody := map[string]any{"roomCode": roomCode} var resp struct { RouterID string `json:"routerId"` @@ -104,8 +116,17 @@ func (h *HTTPMediaOrchestrator) CreateRouter(ctx context.Context, roomCode strin return "", err } - // 记忆映射：一个 roomCode 仅对应一个 Router；若此前存在旧 Router ID（极少见），以新值覆盖 - h.roomRouterIDs.Store(roomCode, resp.RouterID) + // LoadOrStore 防止并发首次创建时重复覆盖：若他人已写入先到的值就用已存在的 + actual, loaded := h.roomRouterIDs.LoadOrStore(roomCode, resp.RouterID) + if loaded { + if existing, _ := actual.(string); existing != "" && existing != resp.RouterID { + logs.Warn(ctx, funcName, "并发创建 Router 出现不一致，保留先到值", + zap.String("room_code", roomCode), + zap.String("existing", existing), + zap.String("new", resp.RouterID)) + return existing, nil + } + } logs.Info(ctx, funcName, "Node Router 创建成功", zap.String("room_code", roomCode), @@ -113,6 +134,21 @@ func (h *HTTPMediaOrchestrator) CreateRouter(ctx context.Context, roomCode strin return resp.RouterID, nil } +// ResolveRouterID 从本地缓存读取 roomCode 对应的 routerID（不触发 HTTP） +// 返回 (id, true) 表示命中；返回 (_, false) 表示缓存缺失（通常意味着该房间尚未创建 Router） +// 供 MeetingService.JoinRoom 等需复用已有 Router 的场景使用，避免重复调 Node +func (h *HTTPMediaOrchestrator) ResolveRouterID(roomCode string) (string, bool) { + v, ok := h.roomRouterIDs.Load(roomCode) + if !ok { + return "", false + } + rid, _ := v.(string) + if rid == "" { + return "", false + } + return rid, true +} + // CloseRouter 调用 DELETE /internal/v1/routers/:routerId // 入参是 roomCode（符合设计 §6.6 NodeClient 契约），内部反查本地缓存拿到 routerID // 若本地缓存不存在映射（可能是 go-service 重启后状态丢失），直接返回 nil： diff --git a/backend/go-service/app/meeting/service/interfaces.go b/backend/go-service/app/meeting/service/interfaces.go index c5d677e..6305e3b 100644 --- a/backend/go-service/app/meeting/service/interfaces.go +++ b/backend/go-service/app/meeting/service/interfaces.go @@ -86,9 +86,14 @@ type CreateConsumerReq struct { // - 所有方法均幂等：重复调用不报错，符合 WS 信令重试语义 type MediaOrchestrator interface { // CreateRouter 为会议房间创建 mediasoup Router + // Task 8 起实现需自带 sync.Map 幂等缓存：roomCode 已有 Router 时直接返回缓存值 CreateRouter(ctx context.Context, roomCode string) (routerID string, err error) // CloseRouter 关闭房间对应 Router 及其下所有资源（幂等） CloseRouter(ctx context.Context, roomCode string) error + // ResolveRouterID 从本地缓存查询 roomCode 对应的 routerID（不触发 HTTP 调用） + // Task 8 引入：用于 JoinRoom 等"仅需复用"的场景，避免重复调 CreateRouter + // 返回 (id, true) 命中；(_, false) 缺失（通常说明房间未创建 Router 或服务重启后未 reschedule） + ResolveRouterID(roomCode string) (string, bool) // CreateTransport 为用户创建 send/recv WebRTC Transport CreateTransport(ctx context.Context, req *CreateTransportReq) (*TransportInfo, error) @@ -127,6 +132,11 @@ func (n *NoopMediaOrchestrator) CloseRouter(_ context.Context, _ string) error { return nil } +// ResolveRouterID 占位实现：始终命中，返回 noop-router-{code} +func (n *NoopMediaOrchestrator) ResolveRouterID(roomCode string) (string, bool) { + return "noop-router-" + roomCode, true +} + // CreateTransport 占位：返回以 "noop-transport-" 为前缀的伪造 ID，带最小合法 JSON 结构 func (n *NoopMediaOrchestrator) CreateTransport(_ context.Context, req *CreateTransportReq) (*TransportInfo, error) { return &TransportInfo{ diff --git a/backend/go-service/app/meeting/service/meeting_lifecycle_service.go b/backend/go-service/app/meeting/service/meeting_lifecycle_service.go new file mode 100644 index 0000000..bef6e1f --- /dev/null +++ b/backend/go-service/app/meeting/service/meeting_lifecycle_service.go @@ -0,0 +1,475 @@ +// Package service 提供 meeting 模块的业务逻辑 +package service + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "time" + + "github.com/echochat/backend/app/constants" + "github.com/echochat/backend/app/meeting/dao" + "github.com/echochat/backend/config" + "github.com/echochat/backend/pkg/logs" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" +) + +// Redis key 前缀（设计文档 §5.4 生命周期 keys；Task 8 落地） +const ( + redisKeyHostGracePrefix = "echo:meeting:host_grace:" + redisKeyEmptyTTLPrefix = "echo:meeting:empty_ttl:" +) + +// hostGracePayload host 宽限期 Redis value 结构 +type hostGracePayload struct { + HostID int64 `json:"host_id"` + StartedAt int64 `json:"started_at"` // 秒级 unix + GraceUntil int64 `json:"grace_until"` // 秒级 unix +} + +// MeetingLifecycleService 会议生命周期状态机 +// 负责设计 §6.5 的 5 类状态跃迁副作用（host 宽限期、自动转让、空房 TTL、TTL 过期销毁） +// 采用 time.AfterFunc 本地 timer + Redis key TTL 双保险： +// - 本地 timer 负责低延迟触发；服务重启由 RescheduleFromRedis 按 Redis 剩余 PTTL 重新装载 +// - 后台定时任务 MeetingCleanupTask 每 N 秒扫描 Redis 兜底防御（timer 丢失 / 多实例部署重复触发场景） +// +// 并发保护：HandleHostGraceExpired / HandleEmptyRoomExpired 入口先 DEL Redis key +// 通过 DEL 返回值确认唯一处理权（0 = 已被其他 goroutine/节点清走，直接 return 避免重复操作） +type MeetingLifecycleService struct { + roomDAO *dao.MeetingRoomDAO + participantDAO *dao.MeetingParticipantDAO + db dbExecutor + redis *redis.Client + broadcaster *MeetingBroadcaster + media MediaOrchestrator + cfg config.MeetingConfig + + graceTimers sync.Map // roomCode -> *time.Timer（host 宽限期） + emptyTTLTimers sync.Map // roomCode -> *time.Timer（空房 TTL） +} + +// dbExecutor MeetingLifecycleService 对事务 DB 的最小依赖抽象 +// 仅用于 HandleEmptyRoomExpired 的 MarkEnded 调用链路，避免强绑 *gorm.DB +type dbExecutor interface{} + +// NewMeetingLifecycleService 构造生命周期服务 +// media 由上游通过 wire 注入 HTTPMediaOrchestrator +func NewMeetingLifecycleService( + roomDAO *dao.MeetingRoomDAO, + participantDAO *dao.MeetingParticipantDAO, + redis *redis.Client, + broadcaster *MeetingBroadcaster, + media MediaOrchestrator, + cfg *config.Config, +) *MeetingLifecycleService { + meetingCfg := cfg.Meeting + // 零值兜底：防止 yaml 未配置时走 0 秒 TTL + if meetingCfg.HostGraceSeconds <= 0 { + meetingCfg.HostGraceSeconds = constants.MeetingHostGraceSeconds + } + if meetingCfg.EmptyRoomTTLSeconds <= 0 { + meetingCfg.EmptyRoomTTLSeconds = constants.MeetingEmptyRoomTTLSeconds + } + if meetingCfg.CleanupIntervalSeconds <= 0 { + meetingCfg.CleanupIntervalSeconds = 30 + } + if meetingCfg.StaleRoomHours <= 0 { + meetingCfg.StaleRoomHours = 4 + } + return &MeetingLifecycleService{ + roomDAO: roomDAO, + participantDAO: participantDAO, + redis: redis, + broadcaster: broadcaster, + media: media, + cfg: meetingCfg, + } +} + +// Config 返回生效的生命周期配置（供定时任务读取扫描周期等） +func (s *MeetingLifecycleService) Config() config.MeetingConfig { + return s.cfg +} + +// HostGraceKey 返回 host 宽限期 Redis key +func HostGraceKey(roomCode string) string { + return redisKeyHostGracePrefix + roomCode +} + +// EmptyTTLKey 返回空房 TTL Redis key +func EmptyTTLKey(roomCode string) string { + return redisKeyEmptyTTLPrefix + roomCode +} + +// OnHostDisconnect host WS 断线钩子 +// 写入 host_grace key（TTL HostGraceSeconds）并启动本地 AfterFunc +// 若已有同 roomCode 的 grace 记录（重复触发），幂等刷新 TTL + 替换 timer +func (s *MeetingLifecycleService) OnHostDisconnect(ctx context.Context, roomCode string, hostID int64) { + funcName := "service.meeting_lifecycle_service.OnHostDisconnect" + + now := time.Now() + payload := hostGracePayload{ + HostID: hostID, + StartedAt: now.Unix(), + GraceUntil: now.Add(time.Duration(s.cfg.HostGraceSeconds) * time.Second).Unix(), + } + raw, _ := json.Marshal(payload) + + graceDur := time.Duration(s.cfg.HostGraceSeconds) * time.Second + // Redis TTL 比本地 timer 多留一段 buffer（max(cleanup*2, 30s)） + // 保证本地 timer 先触发 → DEL 命中；即便本地 timer 丢失，cleanup 也能在 buffer 内兜底扫到 + redisTTL := graceDur + s.ttlBuffer() + if err := s.redis.Set(ctx, HostGraceKey(roomCode), string(raw), redisTTL).Err(); err != nil { + logs.Warn(ctx, funcName, "写入 host_grace key 失败", + zap.String("room_code", roomCode), zap.Int64("host_id", hostID), zap.Error(err)) + return + } + + s.scheduleHostGraceTimer(roomCode, graceDur) + logs.Info(ctx, funcName, "host 进入宽限期", + zap.String("room_code", roomCode), + zap.Int64("host_id", hostID), + zap.Int("grace_seconds", s.cfg.HostGraceSeconds)) +} + +// OnHostReconnect host 重连钩子 +// 清除 host_grace key 并取消本地 timer；用于 host 掉线后 TTL 内重新出现在 WS 层 +// 幂等：key 不存在也不报错 +func (s *MeetingLifecycleService) OnHostReconnect(ctx context.Context, roomCode string, hostID int64) { + funcName := "service.meeting_lifecycle_service.OnHostReconnect" + + deleted, err := s.redis.Del(ctx, HostGraceKey(roomCode)).Result() + if err != nil { + logs.Warn(ctx, funcName, "清除 host_grace key 失败", + zap.String("room_code", roomCode), zap.Error(err)) + } + s.cancelTimer(&s.graceTimers, roomCode) + if deleted > 0 { + logs.Info(ctx, funcName, "host 宽限期内重连，撤销转让计划", + zap.String("room_code", roomCode), zap.Int64("host_id", hostID)) + } +} + +// HandleHostGraceExpired host 宽限期过期处理 +// 触发时机：本地 timer 到期 或 cleanup task 扫到 key TTL <=0 +// 流程： +// 1. DEL host_grace key → 返回 0 表示已被其他路径处理，直接 return 防止重复转让 +// 2. 加载 room + 活跃成员；若 room 已 Ended / host 已变更，视为幂等完成 +// 3. 活跃成员存在：选最早加入者转 host（事务），广播 host.changed，auto_reason=host_grace_expired +// 4. 无活跃成员：走 OnAllMembersLeft 路径（置 empty_ttl） +func (s *MeetingLifecycleService) HandleHostGraceExpired(ctx context.Context, roomCode string) { + funcName := "service.meeting_lifecycle_service.HandleHostGraceExpired" + + deleted, err := s.redis.Del(ctx, HostGraceKey(roomCode)).Result() + if err != nil { + logs.Warn(ctx, funcName, "DEL host_grace key 失败（跳过本次处理）", + zap.String("room_code", roomCode), zap.Error(err)) + return + } + s.cancelTimer(&s.graceTimers, roomCode) + if deleted == 0 { + logs.Debug(ctx, funcName, "host_grace key 不存在（已被重连/其他节点清走）", + zap.String("room_code", roomCode)) + return + } + + room, err := s.roomDAO.GetByCode(ctx, roomCode) + if err != nil || room == nil { + logs.Warn(ctx, funcName, "加载会议失败或已不存在", + zap.String("room_code", roomCode), zap.Error(err)) + return + } + if room.Status == constants.MeetingStatusEnded { + logs.Info(ctx, funcName, "会议已结束，跳过 host 宽限期过期处理", + zap.String("room_code", roomCode)) + return + } + + actives, err := s.participantDAO.ListActiveByRoom(ctx, room.ID) + if err != nil { + logs.Warn(ctx, funcName, "拉取活跃成员失败", + zap.String("room_code", roomCode), zap.Error(err)) + return + } + // 排除原 host（若 left_at IS NULL 但 WS 掉线未落库） + candidates := make([]int64, 0, len(actives)) + for _, p := range actives { + if p.UserID == room.HostID { + continue + } + candidates = append(candidates, p.UserID) + } + + if len(candidates) == 0 { + logs.Info(ctx, funcName, "host 宽限期过期且无其他活跃成员，转空房 TTL", + zap.String("room_code", roomCode), zap.Int64("old_host_id", room.HostID)) + // 清掉 host 自己的记录（若 WS 掉线未落库，由兜底 LeaveRoom 处理） + _, _ = s.participantDAO.LeaveRoom(ctx, room.ID, room.HostID, constants.MeetingLeftReasonDisconnect) + s.OnAllMembersLeft(ctx, roomCode) + return + } + + newHostID := candidates[0] + if err := s.participantDAO.TransferHost(ctx, room.ID, room.HostID, newHostID); err != nil { + logs.Error(ctx, funcName, "host 自动转让 TransferHost 失败", + zap.String("room_code", roomCode), zap.Error(err)) + return + } + if err := s.roomDAO.UpdateHost(ctx, room.ID, newHostID); err != nil { + logs.Warn(ctx, funcName, "UpdateHost 失败（已发生参与者 role 变更）", + zap.String("room_code", roomCode), zap.Error(err)) + } + // 老 host 彻底离会（宽限期内未重连视为网络断开） + _, _ = s.participantDAO.LeaveRoom(ctx, room.ID, room.HostID, constants.MeetingLeftReasonDisconnect) + + s.broadcaster.BroadcastToMeeting(ctx, room.ID, constants.MeetingWSEventHostChanged, map[string]interface{}{ + "room_code": roomCode, + "old_host_id": room.HostID, + "new_host_id": newHostID, + "auto_reason": "host_grace_expired", + }) + logs.Info(ctx, funcName, "host 宽限期过期自动转让完成", + zap.String("room_code", roomCode), + zap.Int64("old_host_id", room.HostID), + zap.Int64("new_host_id", newHostID)) +} + +// OnAllMembersLeft 全员退出钩子 +// 设置 empty_ttl key + 启动本地 AfterFunc；期间有人重新加入可由 CancelEmptyTTL 复活房间 +func (s *MeetingLifecycleService) OnAllMembersLeft(ctx context.Context, roomCode string) { + funcName := "service.meeting_lifecycle_service.OnAllMembersLeft" + + ttl := time.Duration(s.cfg.EmptyRoomTTLSeconds) * time.Second + // Redis TTL 同样加 buffer 避免与本地 timer 同时到期导致 DEL 无法命中 + redisTTL := ttl + s.ttlBuffer() + if err := s.redis.Set(ctx, EmptyTTLKey(roomCode), "1", redisTTL).Err(); err != nil { + logs.Warn(ctx, funcName, "写入 empty_ttl key 失败", + zap.String("room_code", roomCode), zap.Error(err)) + return + } + s.scheduleEmptyTTLTimer(roomCode, ttl) + logs.Info(ctx, funcName, "空房 TTL 已启动", + zap.String("room_code", roomCode), zap.Int("ttl_seconds", s.cfg.EmptyRoomTTLSeconds)) +} + +// CancelEmptyTTL 取消空房 TTL（有新成员加入时调用，房间复活） +// 幂等：key 不存在也不报错 +func (s *MeetingLifecycleService) CancelEmptyTTL(ctx context.Context, roomCode string) { + funcName := "service.meeting_lifecycle_service.CancelEmptyTTL" + + deleted, err := s.redis.Del(ctx, EmptyTTLKey(roomCode)).Result() + if err != nil { + logs.Warn(ctx, funcName, "清除 empty_ttl key 失败", + zap.String("room_code", roomCode), zap.Error(err)) + } + s.cancelTimer(&s.emptyTTLTimers, roomCode) + if deleted > 0 { + logs.Info(ctx, funcName, "空房 TTL 已取消（有人加入）", + zap.String("room_code", roomCode)) + } +} + +// HandleEmptyRoomExpired 空房 TTL 过期处理 +// 流程：DEL empty_ttl key → 确认没被新成员清走 → MarkEnded(reason=empty_ttl) + CloseRouter 幂等 +func (s *MeetingLifecycleService) HandleEmptyRoomExpired(ctx context.Context, roomCode string) { + funcName := "service.meeting_lifecycle_service.HandleEmptyRoomExpired" + + deleted, err := s.redis.Del(ctx, EmptyTTLKey(roomCode)).Result() + if err != nil { + logs.Warn(ctx, funcName, "DEL empty_ttl key 失败", + zap.String("room_code", roomCode), zap.Error(err)) + return + } + s.cancelTimer(&s.emptyTTLTimers, roomCode) + if deleted == 0 { + logs.Debug(ctx, funcName, "empty_ttl key 已消失（有人重入 或 已处理过）", + zap.String("room_code", roomCode)) + return + } + + room, err := s.roomDAO.GetByCode(ctx, roomCode) + if err != nil || room == nil { + logs.Warn(ctx, funcName, "加载会议失败或已不存在", + zap.String("room_code", roomCode), zap.Error(err)) + return + } + if room.Status == constants.MeetingStatusEnded { + return + } + + // 二次校验：DEL 成功与 DB 校验之间可能有新成员加入（并发场景） + activeCount, _ := s.participantDAO.CountActiveByRoom(ctx, room.ID) + if activeCount > 0 { + logs.Info(ctx, funcName, "检测到活跃成员，放弃销毁", + zap.String("room_code", roomCode), zap.Int64("active_count", activeCount)) + return + } + + if _, mErr := s.roomDAO.MarkEnded(ctx, room.ID, constants.MeetingEndedReasonEmptyTTL, time.Now()); mErr != nil { + logs.Error(ctx, funcName, "MarkEnded 失败", + zap.String("room_code", roomCode), zap.Error(mErr)) + return + } + if err := s.media.CloseRouter(ctx, roomCode); err != nil { + logs.Warn(ctx, funcName, "CloseRouter 失败（不影响 DB 状态）", + zap.String("room_code", roomCode), zap.Error(err)) + } + + s.broadcaster.BroadcastToMeeting(ctx, room.ID, constants.MeetingWSEventRoomEnded, map[string]interface{}{ + "room_code": roomCode, + "reason": constants.MeetingEndedReasonEmptyTTL, + }) + logs.Info(ctx, funcName, "空房 TTL 过期，会议已销毁", + zap.String("room_code", roomCode), zap.Int64("room_id", room.ID)) +} + +// RescheduleFromRedis 服务启动时扫描 Redis 已存在的 grace / empty_ttl key，按剩余 PTTL 补装本地 timer +// 服务重启后调用一次，避免重启期间 WS 断线事件的本地 timer 丢失导致宽限期/TTL 永远不触发 +func (s *MeetingLifecycleService) RescheduleFromRedis(ctx context.Context) { + funcName := "service.meeting_lifecycle_service.RescheduleFromRedis" + + s.rescheduleByPattern(ctx, funcName, redisKeyHostGracePrefix+"*", func(roomCode string, dur time.Duration) { + s.scheduleHostGraceTimer(roomCode, dur) + }) + s.rescheduleByPattern(ctx, funcName, redisKeyEmptyTTLPrefix+"*", func(roomCode string, dur time.Duration) { + s.scheduleEmptyTTLTimer(roomCode, dur) + }) +} + +// ScanExpired 扫描指定 key pattern，返回 TTL <=0 （或即将过期）对应的 roomCode 列表 +// 供定时任务兜底调用：若本地 timer 已丢失，此时 Redis key 可能仍存在但 TTL 为负 +// 本实现采用 SCAN + PTTL，单次扫描 limit 控制内存 +func (s *MeetingLifecycleService) ScanExpired(ctx context.Context, keyPrefix string, limit int64) []string { + funcName := "service.meeting_lifecycle_service.ScanExpired" + + expired := make([]string, 0, limit) + var cursor uint64 + for { + keys, next, err := s.redis.Scan(ctx, cursor, keyPrefix+"*", limit).Result() + if err != nil { + logs.Warn(ctx, funcName, "SCAN 失败", zap.String("pattern", keyPrefix), zap.Error(err)) + return expired + } + for _, k := range keys { + pttl, err := s.redis.PTTL(ctx, k).Result() + if err != nil { + continue + } + // PTTL<=0（redis 过期延迟删除场景）；PTTL==-2 已不存在 + if pttl > 0 { + continue + } + roomCode := keyToRoomCode(k, keyPrefix) + if roomCode != "" { + expired = append(expired, roomCode) + } + } + if next == 0 { + break + } + cursor = next + } + return expired +} + +// HostGracePrefix / EmptyTTLPrefix 暴露 key 前缀供定时任务扫描使用 +func (s *MeetingLifecycleService) HostGracePrefix() string { return redisKeyHostGracePrefix } +func (s *MeetingLifecycleService) EmptyTTLPrefix() string { return redisKeyEmptyTTLPrefix } + +// ========== 内部辅助 ========== + +func (s *MeetingLifecycleService) scheduleHostGraceTimer(roomCode string, d time.Duration) { + s.cancelTimer(&s.graceTimers, roomCode) + timer := time.AfterFunc(d, func() { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + s.HandleHostGraceExpired(ctx, roomCode) + }) + s.graceTimers.Store(roomCode, timer) +} + +func (s *MeetingLifecycleService) scheduleEmptyTTLTimer(roomCode string, d time.Duration) { + s.cancelTimer(&s.emptyTTLTimers, roomCode) + timer := time.AfterFunc(d, func() { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + s.HandleEmptyRoomExpired(ctx, roomCode) + }) + s.emptyTTLTimers.Store(roomCode, timer) +} + +// ttlBuffer 返回 Redis key 相对本地 timer 的额外 buffer 时长 +// 用于保证本地 timer 先触发（DEL 命中），避免 Redis 自动过期后 DEL 返回 0 被误判为重复触发 +// 规则：max(CleanupIntervalSeconds * 2, 30s)，确保单机 cleanup 至少能扫到一轮兜底 +func (s *MeetingLifecycleService) ttlBuffer() time.Duration { + buf := time.Duration(s.cfg.CleanupIntervalSeconds*2) * time.Second + if buf < 30*time.Second { + buf = 30 * time.Second + } + return buf +} + +// cancelTimer 取消本地 timer（幂等，不存在也安全） +func (s *MeetingLifecycleService) cancelTimer(store *sync.Map, roomCode string) { + if raw, ok := store.LoadAndDelete(roomCode); ok { + if t, ok := raw.(*time.Timer); ok { + t.Stop() + } + } +} + +// rescheduleByPattern 通用"按 pattern 扫描 Redis + 按 PTTL 重建本地 timer"逻辑 +func (s *MeetingLifecycleService) rescheduleByPattern( + ctx context.Context, + funcName string, + pattern string, + schedule func(roomCode string, dur time.Duration), +) { + var cursor uint64 + total := 0 + for { + keys, next, err := s.redis.Scan(ctx, cursor, pattern, 100).Result() + if err != nil { + logs.Warn(ctx, funcName, "SCAN 失败", zap.String("pattern", pattern), zap.Error(err)) + return + } + for _, k := range keys { + pttl, err := s.redis.PTTL(ctx, k).Result() + if err != nil || pttl <= 0 { + continue + } + prefix := pattern[:len(pattern)-1] // 去掉末尾 * + roomCode := keyToRoomCode(k, prefix) + if roomCode == "" { + continue + } + schedule(roomCode, pttl) + total++ + } + if next == 0 { + break + } + cursor = next + } + if total > 0 { + logs.Info(ctx, funcName, "服务启动重建 timer 完成", + zap.String("pattern", pattern), zap.Int("count", total)) + } +} + +// keyToRoomCode 从完整 Redis key 中去掉前缀得到 roomCode +// 返回空字符串表示格式异常 +func keyToRoomCode(fullKey, prefix string) string { + if len(fullKey) <= len(prefix) { + return "" + } + return fullKey[len(prefix):] +} + +// formatErr 构造统一带上下文的错误（用于少数需要 return 的场景，避免裸字符串拼接） +// nolint:unused +func formatErr(funcName, msg string, err error) error { + return fmt.Errorf("%s: %s: %w", funcName, msg, err) +} diff --git a/backend/go-service/app/meeting/service/meeting_service.go b/backend/go-service/app/meeting/service/meeting_service.go index ef4d372..549c19e 100644 --- a/backend/go-service/app/meeting/service/meeting_service.go +++ b/backend/go-service/app/meeting/service/meeting_service.go @@ -65,11 +65,13 @@ type MeetingService struct { userResolver UserInfoResolver onlineChecker OnlineChecker mediaOrchestrator MediaOrchestrator + lifecycleSvc *MeetingLifecycleService } // NewMeetingService 创建 MeetingService 实例 // 依赖通过构造函数注入；接口依赖由上游 Wire 绑定到具体实现 // Task 7 (2026-04-21) 起 mediaOrchestrator 默认绑定 HTTPMediaOrchestrator +// Task 8 (2026-04-21) 起新增 lifecycleSvc 依赖：JoinRoom/LeaveRoom 的空房/复活场景由 lifecycleSvc 托管 func NewMeetingService( roomDAO *dao.MeetingRoomDAO, participantDAO *dao.MeetingParticipantDAO, @@ -81,6 +83,7 @@ func NewMeetingService( userResolver UserInfoResolver, onlineChecker OnlineChecker, mediaOrchestrator MediaOrchestrator, + lifecycleSvc *MeetingLifecycleService, ) *MeetingService { return &MeetingService{ roomDAO: roomDAO, @@ -93,6 +96,7 @@ func NewMeetingService( userResolver: userResolver, onlineChecker: onlineChecker, mediaOrchestrator: mediaOrchestrator, + lifecycleSvc: lifecycleSvc, } } @@ -210,9 +214,10 @@ func (s *MeetingService) CreateRoom(ctx context.Context, hostID int64, req *dto. return nil, nil, "", err } + // Task 7 起 CreateRouter 对接真实 mediasoup；Task 8 起仅在会议首次创建时调一次（房间级资源） routerID, mediaErr := s.mediaOrchestrator.CreateRouter(ctx, code) if mediaErr != nil { - logs.Warn(ctx, funcName, "mediasoup Router 创建失败（Task 7 前为占位实现，不影响流程）", + logs.Warn(ctx, funcName, "mediasoup Router 创建失败", zap.String("room_code", code), zap.Error(mediaErr)) routerID = "" } @@ -309,10 +314,18 @@ func (s *MeetingService) JoinRoom(ctx context.Context, userID int64, code, passw return nil, nil, "", err } - routerID, mediaErr := s.mediaOrchestrator.CreateRouter(ctx, code) - if mediaErr != nil { - logs.Warn(ctx, funcName, "mediasoup Router 创建失败（占位实现）", zap.String("room_code", code), zap.Error(mediaErr)) - routerID = "" + // Task 8：空房 TTL 复活 + // 若该房间正处于 empty_ttl 阶段（全员离开后的 5 分钟窗口），新成员加入立即取消销毁 + if s.lifecycleSvc != nil { + s.lifecycleSvc.CancelEmptyTTL(ctx, code) + } + + // Task 8：JoinRoom 不再主动调 CreateRouter（Router 在 CreateRoom 时创建、由 HTTPMediaOrchestrator 本地缓存） + // 从缓存读取 routerID；缺失时（极少见：服务重启后未重建缓存）保持为空，不阻塞加入流程 + routerID, _ := s.mediaOrchestrator.ResolveRouterID(code) + if routerID == "" { + logs.Debug(ctx, funcName, "RouterID 缓存缺失（非致命，可能服务重启）", + zap.String("room_code", code)) } go s.broadcastToActiveParticipants(context.Background(), room.ID, constants.MeetingWSEventMemberJoined, map[string]interface{}{ @@ -378,9 +391,10 @@ func (s *MeetingService) LeaveRoom(ctx context.Context, userID int64, code strin } } - if len(actives) == 0 { - _, _ = s.roomDAO.MarkEnded(ctx, room.ID, constants.MeetingEndedReasonEmptyTTL, time.Now()) - _ = s.mediaOrchestrator.CloseRouter(ctx, code) + // Task 8：全员离会改走生命周期状态机（空房 TTL），不再立即销毁房间 + // 在 TTL 窗口内如有新成员加入，房间会被 CancelEmptyTTL 复活；TTL 到期由 HandleEmptyRoomExpired 兜底销毁 + if len(actives) == 0 && s.lifecycleSvc != nil { + s.lifecycleSvc.OnAllMembersLeft(ctx, code) } go s.broadcastToActiveParticipants(context.Background(), room.ID, constants.MeetingWSEventMemberLeft, map[string]interface{}{ diff --git a/backend/go-service/app/meeting/service/meeting_signal_service.go b/backend/go-service/app/meeting/service/meeting_signal_service.go index 8b3921b..c935684 100644 --- a/backend/go-service/app/meeting/service/meeting_signal_service.go +++ b/backend/go-service/app/meeting/service/meeting_signal_service.go @@ -31,15 +31,18 @@ type MeetingSignalService struct { broadcaster *MeetingBroadcaster mediaOrchestrator MediaOrchestrator + lifecycleSvc *MeetingLifecycleService } // NewMeetingSignalService 构造 WS 信令服务 +// Task 8 注入 lifecycleSvc：host 掉线 / 重连的生命周期联动 func NewMeetingSignalService( roomDAO *dao.MeetingRoomDAO, participantDAO *dao.MeetingParticipantDAO, redis *redis.Client, broadcaster *MeetingBroadcaster, mediaOrchestrator MediaOrchestrator, + lifecycleSvc *MeetingLifecycleService, ) *MeetingSignalService { return &MeetingSignalService{ roomDAO: roomDAO, @@ -47,6 +50,7 @@ func NewMeetingSignalService( redis: redis, broadcaster: broadcaster, mediaOrchestrator: mediaOrchestrator, + lifecycleSvc: lifecycleSvc, } } @@ -116,6 +120,12 @@ func (s *MeetingSignalService) OnRoomJoin(ctx context.Context, userID int64, roo key := resourceTrackKey(roomCode, userID) _ = s.redis.Expire(ctx, key, resourceTTL).Err() + // Task 8：若本次 WS 在线的用户是会议 host，尝试撤销 host 宽限期 + // 本方法幂等：若此前无 host_grace key（正常场景）DEL 返回 0，不产生副作用 + if s.lifecycleSvc != nil && room.HostID == userID { + s.lifecycleSvc.OnHostReconnect(ctx, roomCode, userID) + } + logs.Info(ctx, "service.meeting_signal_service.OnRoomJoin", "用户宣告 WS 在线", zap.String("room_code", roomCode), zap.Int64("user_id", userID), @@ -123,6 +133,47 @@ func (s *MeetingSignalService) OnRoomJoin(ctx context.Context, userID int64, roo return nil } +// OnWSDisconnect WS 断线钩子，实现 ws.MeetingDisconnectHook 接口（Task 8） +// 由 ws.handler 的 SetOnDisconnect 回调触发；场景：用户的最后一条 WS 连接被移除 +// 职责： +// 1. 若该用户当前存在活跃 meeting_participants 记录：清理其媒体资源 + 若是 host 启动 host 宽限期 +// 2. 普通成员：仅清媒体资源（设计 Q1=A：不动 meeting_participants，长期不活跃由 4h 兜底清理） +// +// 容错：所有异常仅 Warn 日志，不返回 error；保证 ws.handler 主断连流程不受影响 +func (s *MeetingSignalService) OnWSDisconnect(ctx context.Context, userID int64) { + funcName := "service.meeting_signal_service.OnWSDisconnect" + + active, err := s.participantDAO.FindActiveByUser(ctx, userID) + if err != nil { + logs.Warn(ctx, funcName, "查询活跃会议失败", zap.Int64("user_id", userID), zap.Error(err)) + return + } + if active == nil { + // 用户当前不在任何会议 + return + } + room, err := s.roomDAO.GetByID(ctx, active.RoomID) + if err != nil || room == nil { + logs.Warn(ctx, funcName, "加载 room 失败或已不存在", + zap.Int64("room_id", active.RoomID), zap.Error(err)) + return + } + if room.Status == constants.MeetingStatusEnded { + return + } + + s.cleanupUserResources(ctx, room.RoomCode, userID) + + if s.lifecycleSvc != nil && room.HostID == userID { + s.lifecycleSvc.OnHostDisconnect(ctx, room.RoomCode, userID) + } + + logs.Info(ctx, funcName, "WS 断线已处理", + zap.String("room_code", room.RoomCode), + zap.Int64("user_id", userID), + zap.Bool("is_host", room.HostID == userID)) +} + // OnRoomLeave 处理 meeting.room.leave 事件 // 语义：WS 层面的主动离会（等价 REST leave 但不强制要求落库事务； // 当前实现：仅清理该用户在本会议的所有媒体资源（batch close producer/consumer）+ 广播 meeting.member.left diff --git a/backend/go-service/app/meeting/task/meeting_cleanup_task.go b/backend/go-service/app/meeting/task/meeting_cleanup_task.go new file mode 100644 index 0000000..10032cd --- /dev/null +++ b/backend/go-service/app/meeting/task/meeting_cleanup_task.go @@ -0,0 +1,191 @@ +// Package task 提供 meeting 模块的后台定时任务 +package task + +import ( + "context" + "time" + + "github.com/echochat/backend/app/constants" + "github.com/echochat/backend/app/meeting/dao" + "github.com/echochat/backend/app/meeting/service" + "github.com/echochat/backend/pkg/logs" + "go.uber.org/zap" +) + +// MeetingCleanupTask 会议生命周期兜底清理任务（Phase 2e-2 Task 8） +// 职责（设计 §11.4 兜底策略 + Task 8 落地）： +// 1. 启动时一次性 RescheduleFromRedis，按 Redis 残留 grace/empty_ttl key 重建本地 timer +// 2. 每 cleanupInterval 秒扫描 host_grace / empty_ttl，命中"Redis 仍在但本地 timer 可能已丢失"的场景 +// 兜底触发 HandleHostGraceExpired / HandleEmptyRoomExpired；真正过期判断由 Lifecycle 内部 DEL 互斥 +// 3. 每 10 分钟扫一次 stale active rooms（活跃状态但超过 StaleRoomHours 无成员动作）→ 强制 MarkEnded(system_error) +// 4. 每 24 小时扫 Ended 超过 chat 保留时长的房间 → DeleteByRoomIDs 清理 meeting_chats +// +// 所有周期都由独立 ticker 驱动，相互不阻塞；退出走 stopCh 统一关闭 +type MeetingCleanupTask struct { + lifecycleSvc *service.MeetingLifecycleService + roomDAO *dao.MeetingRoomDAO + chatDAO *dao.MeetingChatDAO + + cleanupInterval time.Duration // 扫 grace/empty_ttl 的周期 + staleInterval time.Duration // 扫 stale active 的周期 + chatInterval time.Duration // 扫 chat 清理的周期 + staleHours int // 超过此小时的活跃房间视为 stale + + stopCh chan struct{} +} + +// NewMeetingCleanupTask 构造定时任务 +// 扫描参数从 lifecycleSvc.Config() 读取，避免两处配置分离 +func NewMeetingCleanupTask( + lifecycleSvc *service.MeetingLifecycleService, + roomDAO *dao.MeetingRoomDAO, + chatDAO *dao.MeetingChatDAO, +) *MeetingCleanupTask { + cfg := lifecycleSvc.Config() + interval := time.Duration(cfg.CleanupIntervalSeconds) * time.Second + if interval <= 0 { + interval = 30 * time.Second + } + staleHours := cfg.StaleRoomHours + if staleHours <= 0 { + staleHours = 4 + } + return &MeetingCleanupTask{ + lifecycleSvc: lifecycleSvc, + roomDAO: roomDAO, + chatDAO: chatDAO, + cleanupInterval: interval, + staleInterval: 10 * time.Minute, + chatInterval: 24 * time.Hour, + staleHours: staleHours, + stopCh: make(chan struct{}), + } +} + +// Start 启动任务（非阻塞） +// 应在 main 进程启动后调用，进程退出时调用 Stop +func (t *MeetingCleanupTask) Start() { + funcName := "task.meeting_cleanup_task.Start" + logs.Info(nil, funcName, "启动会议生命周期兜底任务", + zap.Duration("cleanup_interval", t.cleanupInterval), + zap.Duration("stale_interval", t.staleInterval), + zap.Duration("chat_interval", t.chatInterval), + zap.Int("stale_hours", t.staleHours)) + + go t.runRescheduleOnce() + go t.runLoop("grace_empty_ttl", t.cleanupInterval, 5*time.Second, t.scanExpiredKeys) + go t.runLoop("stale_active", t.staleInterval, 30*time.Second, t.scanStaleActive) + go t.runLoop("chat_cleanup", t.chatInterval, 60*time.Second, t.cleanupExpiredChats) +} + +// Stop 停止所有循环 +func (t *MeetingCleanupTask) Stop() { + close(t.stopCh) +} + +// runRescheduleOnce 启动后一次性扫描 Redis 残留 key 并补装本地 timer +func (t *MeetingCleanupTask) runRescheduleOnce() { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + t.lifecycleSvc.RescheduleFromRedis(ctx) +} + +// runLoop 通用"固定间隔 + 预热延迟 + stopCh 退出"循环 +// warmup 用于错峰避免启动风暴，单次业务执行由 do 函数自行控制超时 +func (t *MeetingCleanupTask) runLoop(name string, interval, warmup time.Duration, do func()) { + funcName := "task.meeting_cleanup_task.runLoop." + name + + timer := time.NewTimer(warmup) + select { + case <-t.stopCh: + timer.Stop() + return + case <-timer.C: + } + + do() + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-t.stopCh: + logs.Info(nil, funcName, "任务循环退出") + return + case <-ticker.C: + do() + } + } +} + +// scanExpiredKeys 扫 host_grace 与 empty_ttl 两个 key 集合 +// Redis 原生 TTL 过期时 Go 的 PTTL 会返回 -2（不存在），此时无需兜底 +// 少数场景（AfterFunc 丢失 / 多实例部署）会出现 Redis 仍有 key 但无本地 timer +// 本扫描对 "TTL≤0 但仍存在" 的 key 主动调 HandleXxxExpired（内部 DEL 互斥） +func (t *MeetingCleanupTask) scanExpiredKeys() { + funcName := "task.meeting_cleanup_task.scanExpiredKeys" + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + hostExpired := t.lifecycleSvc.ScanExpired(ctx, t.lifecycleSvc.HostGracePrefix(), 100) + for _, code := range hostExpired { + t.lifecycleSvc.HandleHostGraceExpired(ctx, code) + } + emptyExpired := t.lifecycleSvc.ScanExpired(ctx, t.lifecycleSvc.EmptyTTLPrefix(), 100) + for _, code := range emptyExpired { + t.lifecycleSvc.HandleEmptyRoomExpired(ctx, code) + } + if len(hostExpired)+len(emptyExpired) > 0 { + logs.Info(ctx, funcName, "兜底处理过期 key 完成", + zap.Int("host_grace", len(hostExpired)), + zap.Int("empty_ttl", len(emptyExpired))) + } +} + +// scanStaleActive 扫超 staleHours 无任何成员动作的活跃会议，强制结束 +// reason=system_error：明确标记非正常终止，便于运维排查 Node worker 掉链等异常 +func (t *MeetingCleanupTask) scanStaleActive() { + funcName := "task.meeting_cleanup_task.scanStaleActive" + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + ids, err := t.roomDAO.ListStaleActive(ctx, t.staleHours, 100) + if err != nil { + logs.Warn(ctx, funcName, "扫描 stale active 失败", zap.Error(err)) + return + } + if len(ids) == 0 { + return + } + now := time.Now() + for _, id := range ids { + if _, err := t.roomDAO.MarkEnded(ctx, id, constants.MeetingEndedReasonSystemError, now); err != nil { + logs.Warn(ctx, funcName, "兜底 MarkEnded 失败", + zap.Int64("room_id", id), zap.Error(err)) + } + } + logs.Info(ctx, funcName, "stale 会议兜底结束", zap.Int("count", len(ids))) +} + +// cleanupExpiredChats 扫超过 MeetingChatRetentionHours 的 Ended 会议，批量删除聊天记录 +func (t *MeetingCleanupTask) cleanupExpiredChats() { + funcName := "task.meeting_cleanup_task.cleanupExpiredChats" + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + + ids, err := t.roomDAO.ListExpiredForCleanup(ctx, constants.MeetingChatRetentionHours, 500) + if err != nil { + logs.Warn(ctx, funcName, "扫描过期会议失败", zap.Error(err)) + return + } + if len(ids) == 0 { + return + } + deleted, err := t.chatDAO.DeleteByRoomIDs(ctx, ids) + if err != nil { + logs.Warn(ctx, funcName, "批量删除聊天记录失败", zap.Error(err)) + return + } + logs.Info(ctx, funcName, "会议聊天清理完成", + zap.Int("room_count", len(ids)), + zap.Int64("chat_deleted", deleted)) +} diff --git a/backend/go-service/app/provider/provider.go b/backend/go-service/app/provider/provider.go index 5268aac..7a1e6ab 100644 --- a/backend/go-service/app/provider/provider.go +++ b/backend/go-service/app/provider/provider.go @@ -13,6 +13,7 @@ import ( imHandler "github.com/echochat/backend/app/im/handler" meetingController "github.com/echochat/backend/app/meeting/controller" meetingService "github.com/echochat/backend/app/meeting/service" + meetingTask "github.com/echochat/backend/app/meeting/task" notifyController "github.com/echochat/backend/app/notify/controller" notifyService "github.com/echochat/backend/app/notify/service" notifyTask "github.com/echochat/backend/app/notify/task" @@ -56,8 +57,10 @@ type App struct { NotifyCleanupTask *notifyTask.CleanupTask // 通知清理定时任务 MeetingService *meetingService.MeetingService // 会议 REST 业务服务（Task 5 落地） MeetingSignalService *meetingService.MeetingSignalService // 会议 WS 信令业务服务（Task 6 落地） + MeetingLifecycleSvc *meetingService.MeetingLifecycleService // 会议生命周期状态机（Task 8 落地） MeetingController *meetingController.MeetingController // 会议 REST 控制器 MeetingWSHandler *meetingController.MeetingWSHandler // 会议 WS 事件 Handler（构造时自动注册路由到 Hub） + MeetingCleanupTask *meetingTask.MeetingCleanupTask // 会议生命周期兜底定时任务（Task 8 落地） } // NewApp 创建应用实例 @@ -89,11 +92,14 @@ func NewApp( notifyCleanup *notifyTask.CleanupTask, meetingSvc *meetingService.MeetingService, meetingSignalSvc *meetingService.MeetingSignalService, + meetingLifecycleSvc *meetingService.MeetingLifecycleService, meetingCtrl *meetingController.MeetingController, meetingWSHandler *meetingController.MeetingWSHandler, + meetingCleanup *meetingTask.MeetingCleanupTask, ) *App { wsHandler.SetOfflinePusher(offlinePusher) wsHandler.SetNotifyConnectHook(notifySvc) + wsHandler.SetMeetingDisconnectHook(meetingSignalSvc) return &App{ Config: cfg, @@ -123,8 +129,10 @@ func NewApp( NotifyCleanupTask: notifyCleanup, MeetingService: meetingSvc, MeetingSignalService: meetingSignalSvc, + MeetingLifecycleSvc: meetingLifecycleSvc, MeetingController: meetingCtrl, MeetingWSHandler: meetingWSHandler, + MeetingCleanupTask: meetingCleanup, } } diff --git a/backend/go-service/app/provider/wire.go b/backend/go-service/app/provider/wire.go index 5f8fbe8..cbe3673 100644 --- a/backend/go-service/app/provider/wire.go +++ b/backend/go-service/app/provider/wire.go @@ -54,6 +54,8 @@ func InitializeApp(cfg *config.Config) (*App, error) { wire.Bind(new(meetingService.NotifyPusher), new(*notifyService.NotifyService)), wire.Bind(new(meetingService.UserInfoResolver), new(*contactDAO.FriendshipDAO)), wire.Bind(new(meetingService.OnlineChecker), new(*wsApp.OnlineService)), + // Task 8：WS 断线钩子 → MeetingSignalService + wire.Bind(new(wsApp.MeetingDisconnectHook), new(*meetingService.MeetingSignalService)), ) return nil, nil } diff --git a/backend/go-service/app/provider/wire_gen.go b/backend/go-service/app/provider/wire_gen.go index e8c26a1..355a6c7 100644 --- a/backend/go-service/app/provider/wire_gen.go +++ b/backend/go-service/app/provider/wire_gen.go @@ -25,6 +25,7 @@ import ( controller7 "github.com/echochat/backend/app/meeting/controller" dao6 "github.com/echochat/backend/app/meeting/dao" service7 "github.com/echochat/backend/app/meeting/service" + task2 "github.com/echochat/backend/app/meeting/task" controller6 "github.com/echochat/backend/app/notify/controller" dao5 "github.com/echochat/backend/app/notify/dao" service3 "github.com/echochat/backend/app/notify/service" @@ -103,10 +104,12 @@ func InitializeApp(cfg *config.Config) (*App, error) { meetingChatDAO := dao6.NewMeetingChatDAO(gormDB) meetingBroadcaster := service7.NewMeetingBroadcaster(meetingParticipantDAO, pubSub) httpMediaOrchestrator := service7.NewHTTPMediaOrchestrator(cfg) - meetingService := service7.NewMeetingService(meetingRoomDAO, meetingParticipantDAO, meetingChatDAO, gormDB, client, meetingBroadcaster, notifyService, friendshipDAO, onlineService, httpMediaOrchestrator) - meetingSignalService := service7.NewMeetingSignalService(meetingRoomDAO, meetingParticipantDAO, client, meetingBroadcaster, httpMediaOrchestrator) + meetingLifecycleService := service7.NewMeetingLifecycleService(meetingRoomDAO, meetingParticipantDAO, client, meetingBroadcaster, httpMediaOrchestrator, cfg) + meetingService := service7.NewMeetingService(meetingRoomDAO, meetingParticipantDAO, meetingChatDAO, gormDB, client, meetingBroadcaster, notifyService, friendshipDAO, onlineService, httpMediaOrchestrator, meetingLifecycleService) + meetingSignalService := service7.NewMeetingSignalService(meetingRoomDAO, meetingParticipantDAO, client, meetingBroadcaster, httpMediaOrchestrator, meetingLifecycleService) meetingController := controller7.NewMeetingController(meetingService) meetingWSHandler := controller7.NewMeetingWSHandler(meetingSignalService, hub) - app := NewApp(cfg, gormDB, client, minioClient, authService, authController, adminAuthController, userManageController, onlineController, contactManageController, groupManageController, messageManageController, handler, hub, pubSub, onlineService, contactController, imController, eventHandler, offlinePusher, fileController, groupController, notifyService, notificationController, cleanupTask, meetingService, meetingSignalService, meetingController, meetingWSHandler) + meetingCleanupTask := task2.NewMeetingCleanupTask(meetingLifecycleService, meetingRoomDAO, meetingChatDAO) + app := NewApp(cfg, gormDB, client, minioClient, authService, authController, adminAuthController, userManageController, onlineController, contactManageController, groupManageController, messageManageController, handler, hub, pubSub, onlineService, contactController, imController, eventHandler, offlinePusher, fileController, groupController, notifyService, notificationController, cleanupTask, meetingService, meetingSignalService, meetingLifecycleService, meetingController, meetingWSHandler, meetingCleanupTask) return app, nil } diff --git a/backend/go-service/app/ws/handler.go b/backend/go-service/app/ws/handler.go index 1fbaf2d..ddde1e7 100644 --- a/backend/go-service/app/ws/handler.go +++ b/backend/go-service/app/ws/handler.go @@ -35,6 +35,15 @@ type NotifyConnectHook interface { PushUnreadTotalOnConnect(ctx context.Context, userID int64) } +// MeetingDisconnectHook 会议 WS 断线钩子接口（Phase 2e-2 Task 8） +// 由 meeting.service.MeetingSignalService 隐式实现 +// 触发时机：用户最后一条 WS 连接被移除（最终下线） +// 职责：清理该用户在会议中的媒体资源；若为 host 则启动 host 宽限期 +// 抽象为接口避免 ws 包反向依赖 meeting 包引发循环引用 +type MeetingDisconnectHook interface { + OnWSDisconnect(ctx context.Context, userID int64) +} + var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, @@ -45,13 +54,14 @@ var upgrader = websocket.Upgrader{ // Handler WebSocket 连接处理器 type Handler struct { - hub *ws.Hub - pubsub *ws.PubSub - jwtCfg *config.JWTConfig - onlineService *OnlineService - tokenValidator TokenValidator - offlinePusher OfflineMessagePusher - notifyConnectHook NotifyConnectHook + hub *ws.Hub + pubsub *ws.PubSub + jwtCfg *config.JWTConfig + onlineService *OnlineService + tokenValidator TokenValidator + offlinePusher OfflineMessagePusher + notifyConnectHook NotifyConnectHook + meetingDisconnectHook MeetingDisconnectHook // Task 8 注入 } // NewHandler 创建 WebSocket Handler 实例 @@ -75,6 +85,12 @@ func (h *Handler) SetNotifyConnectHook(hook NotifyConnectHook) { h.notifyConnectHook = hook } +// SetMeetingDisconnectHook 设置会议 WS 断线钩子（由 meeting 模块在初始化时注入） +// Phase 2e-2 Task 8：WS 最终下线时触发 host 宽限期 / 媒体资源清理 +func (h *Handler) SetMeetingDisconnectHook(hook MeetingDisconnectHook) { + h.meetingDisconnectHook = hook +} + // Upgrade 处理 WebSocket 升级请求 // GET /ws?token=xxx → JWT 认证 → 升级连接 → 注册 Hub → 订阅 Redis 频道 func (h *Handler) Upgrade(c *gin.Context) { @@ -120,6 +136,11 @@ func (h *Handler) Upgrade(c *gin.Context) { } h.pubsub.Unsubscribe(userID) h.onlineService.UserOffline(context.Background(), userID) + // Task 8：通知会议模块处理 host 宽限期 / 媒体资源清理 + // 钩子为可选（测试或 meeting 模块未加载场景安全跳过） + if h.meetingDisconnectHook != nil { + go h.meetingDisconnectHook.OnWSDisconnect(context.Background(), userID) + } }) h.hub.Register(client) h.pubsub.Subscribe(claims.UserID) diff --git a/backend/go-service/cmd/server/main.go b/backend/go-service/cmd/server/main.go index f0c71c5..e8dcb28 100644 --- a/backend/go-service/cmd/server/main.go +++ b/backend/go-service/cmd/server/main.go @@ -82,6 +82,11 @@ func main() { app.NotifyCleanupTask.Start() defer app.NotifyCleanupTask.Stop() + // 7.2 启动 meeting 模块生命周期兜底任务（Phase 2e-2 Task 8） + // 内部会先 RescheduleFromRedis 恢复服务重启前残留的 host_grace / empty_ttl timer + app.MeetingCleanupTask.Start() + defer app.MeetingCleanupTask.Stop() + // 8. 启动 HTTP 服务（优雅关闭） addr := fmt.Sprintf(":%d", cfg.Server.Port) srv := &http.Server{ diff --git a/backend/go-service/config/config.dev.yaml b/backend/go-service/config/config.dev.yaml index 4e92d87..ad70046 100644 --- a/backend/go-service/config/config.dev.yaml +++ b/backend/go-service/config/config.dev.yaml @@ -63,3 +63,11 @@ media_server: timeout_ms: 5000 # 创建类接口超时（毫秒） close_timeout_ms: 2000 # 关闭类接口超时（毫秒） close_retry: 2 # 关闭类接口失败重试次数（指数退避 200/500ms） + +# 会议模块生命周期参数（Phase 2e-2 Task 8 会议状态机） +# E2E 测试可通过环境变量 ECHOCHAT_MEETING_HOST_GRACE_SECONDS 等覆盖为更短值以加速触发 +meeting: + host_grace_seconds: 120 # 主持人掉线宽限期（秒），期间保留 host 身份 + empty_room_ttl_seconds: 300 # 空房自动销毁 TTL（秒），期间有人加入可复活 + cleanup_interval_seconds: 30 # 兜底扫描周期（秒），定时任务检查过期 grace/empty_ttl key + stale_room_hours: 4 # 活跃超过此小时且无活跃成员视为 stale，强制结束 diff --git a/backend/go-service/config/config.docker.yaml b/backend/go-service/config/config.docker.yaml index 5c2dfcc..94ce7ba 100644 --- a/backend/go-service/config/config.docker.yaml +++ b/backend/go-service/config/config.docker.yaml @@ -55,3 +55,10 @@ media_server: timeout_ms: 5000 close_timeout_ms: 2000 close_retry: 2 + +# 会议模块生命周期参数（Phase 2e-2 Task 8） +meeting: + host_grace_seconds: 120 + empty_room_ttl_seconds: 300 + cleanup_interval_seconds: 30 + stale_room_hours: 4 diff --git a/backend/go-service/config/config.go b/backend/go-service/config/config.go index 2c348b3..7f5c6d2 100644 --- a/backend/go-service/config/config.go +++ b/backend/go-service/config/config.go @@ -18,6 +18,16 @@ type Config struct { Log LogConfig `mapstructure:"log"` Minio MinioConfig `mapstructure:"minio"` MediaServer MediaServerConfig `mapstructure:"media_server"` + Meeting MeetingConfig `mapstructure:"meeting"` +} + +// MeetingConfig 会议模块生命周期参数（Phase 2e-2 Task 8） +// 所有字段均允许通过环境变量 ECHOCHAT_MEETING_* 覆盖，便于 E2E 测试以短时长加速场景触发 +type MeetingConfig struct { + HostGraceSeconds int `mapstructure:"host_grace_seconds"` // host 掉线宽限期秒数，默认 120 + EmptyRoomTTLSeconds int `mapstructure:"empty_room_ttl_seconds"` // 空房自动销毁 TTL 秒数，默认 300 + CleanupIntervalSeconds int `mapstructure:"cleanup_interval_seconds"` // 兜底扫描周期秒数，默认 30 + StaleRoomHours int `mapstructure:"stale_room_hours"` // 活跃超过此小时且无成员视为 stale，默认 4 } // MediaServerConfig Node media-server 接入配置（Phase 2e-2 Task 7） diff --git a/docs/api/frontend/meeting.md b/docs/api/frontend/meeting.md index eb86fb5..ba84499 100644 --- a/docs/api/frontend/meeting.md +++ b/docs/api/frontend/meeting.md @@ -208,8 +208,8 @@ **行为**： - `participant.left_at = now`、`duration = EXTRACT(EPOCH FROM now - joined_at)`； -- 若离开者是 host 且仍有其他活跃成员 → 自动将 host 转让给**最早加入的活跃成员**，广播 `meeting.host.changed`； -- 若房间无剩余活跃成员 → 标记 `status=2 ended_reason=empty_ttl` 并触发 mediaOrchestrator.CloseRouter； +- 若离开者是 host 且仍有其他活跃成员 → 自动将 host 转让给**最早加入的活跃成员**，广播 `meeting.host.changed`（`auto_reason=host_left_with_members`）； +- **若房间无剩余活跃成员**（Task 8 变更）→ **不再立即销毁**，改为调 `MeetingLifecycleService.OnAllMembersLeft` 设置 `echo:meeting:empty_ttl:{code}`（默认 TTL 300s）+ 启动本地 `time.AfterFunc`；TTL 内若有新成员 `POST /join`，`MeetingLifecycleService.CancelEmptyTTL` 会 DEL key 让房间复活；TTL 过期触发 `HandleEmptyRoomExpired` → `MarkEnded(reason=empty_ttl)` + `mediaOrchestrator.CloseRouter` + 广播 `meeting.room.ended{reason=empty_ttl}`； - 广播 `meeting.member.left`。 **响应 `200 OK`** @@ -595,8 +595,8 @@ | `meeting.member.joined` | room_code / participant 对象 | REST /join 触发 | | `meeting.member.left` | room_code / user_id / reason | REST /leave /kick 或 WS 资源清理 | | `meeting.member.kicked` | room_code / reason | 仅发给被踢者本人，对应 REST /kick | -| `meeting.host.changed` | room_code / old_host_id / new_host_id | REST /transfer-host 或 host 离会自动转让 | -| `meeting.room.ended` | room_code / reason（`host_ended` / `empty_ttl`） | REST /end 或空房 TTL | +| `meeting.host.changed` | room_code / old_host_id / new_host_id / **auto_reason**? | REST /transfer-host（无 `auto_reason`）/ host 离会自动转让（`auto_reason=host_left_with_members`）/ **host 宽限期过期自动转让**（`auto_reason=host_grace_expired`，Task 8 新增） | +| `meeting.room.ended` | room_code / reason（`host_ended` / `empty_ttl` / **`system_error`**） | REST /end（`host_ended`）/ 空房 TTL 过期（`empty_ttl`）/ 兜底清理 4 小时陈旧房间（`system_error`，Task 8 新增） | | `meeting.member.state.changed` | 见白名单 #3 | WS `meeting.member.state.changed` 触发 | | `meeting.member.producer.new` | 见白名单 #6 / #8 | WS `meeting.produce.start` / `meeting.producer.close` 触发 | | `meeting.chat` | message 对象 | REST /chats 触发 | @@ -639,11 +639,16 @@ - `meeting.transport.create`（direction=send）返回**真实** mediasoup `id`（非 `noop-` 前缀）+ `iceParameters` 对象 + `iceCandidates[]` 非空 + `dtlsParameters.fingerprints[]` 非空 - 虚构 producer_id 调 `meeting.producer.close` → Node 返回 404 → Go 幂等转 `code=0` - host `POST /rooms/:code/end` 触发 `CloseRouter`，media-server 日志确认 `router closed explicitly` +- **Task 8** 端到端验证脚本（`docs/verify/meeting_t8_verify.mjs`）结果：**20/20 PASS**，覆盖 5 个会议生命周期场景： + - **S1 host 宽限期过期自动转让**：host WS 断线 → Redis 写入 `echo:meeting:host_grace:{code}` → 3s 后过期 → `meeting.host.changed{auto_reason=host_grace_expired}` 广播 + DB `host_id` 更新为最早加入者 + - **S2 宽限期内重连保留身份**：host 断线 1s 内重连 `meeting.room.join` → `host_grace` key DEL + host 身份保留 + - **S3 空房 TTL 复活**：全员 leave → Redis 写入 `echo:meeting:empty_ttl:{code}` → TTL 内新用户 join → key DEL + 房间保持 Active + - **S4 空房 TTL 过期销毁**：全员 leave → 3s 后 TTL 过期 → 房间 Ended + 新 join 被拒 + - **S5 Router 幂等**：`POST /meeting/rooms` 后 `stats.routers` +1；`POST /meeting/rooms/:code/join` 不再触发新 Router（业务层 `JoinRoom` 不调 `CreateRouter` + HTTP 层 `sync.Map` 幂等防御） --- ## 后续任务关联 -- **Task 8**：会议生命周期状态机（host 宽限期 + 自动转让 + 空房 TTL），同时修复"CreateRoom + JoinRoom 重复调 `CreateRouter`"遗留行为。 -- **Task 9**：Vue 前端 mediasoup-client 接入，按本文 WS 契约实现 `mediasoup.Transport` 的 `connect / produce` 回调；补齐 `ResumeConsumer`（Node REST 已就绪）或在 Go 侧改为创建 Consumer 后自动 resume。 +- **Task 9**：Vue 前端 mediasoup-client 接入，按本文 WS 契约实现 `mediasoup.Transport` 的 `connect / produce` 回调；补齐 `ResumeConsumer`（Node REST 已就绪）或在 Go 侧改为创建 Consumer 后自动 resume；需处理 `meeting.host.changed` 的 `auto_reason` 字段以在 UI 上标注"自动转让"。 - **Task 13**：通知卡片 UI 补齐 `meeting_invite` 内联按钮。 diff --git a/docs/architecture/system-architecture.md b/docs/architecture/system-architecture.md index eca0535..dae100a 100644 --- a/docs/architecture/system-architecture.md +++ b/docs/architecture/system-architecture.md @@ -94,7 +94,7 @@ EchoChat 采用 **「精简单体 + 媒体微服务」** 架构，核心思想 | im | 即时消息收发（单聊）、会话管理、消息存储、撤回、搜索、离线推送 | ✅ Phase 2b | | group | 群聊管理（建群/加入/退出/角色/禁言/@提醒/群公告/入群审批） | ✅ Phase 2c | | file | 文件上传（MinIO 对象存储 + 通用上传 API） | ✅ Phase 2c | -| meeting | 会议创建/管理、WebSocket 信令转发、mediasoup 资源编排（Go→Node HTTP）、会议内聊天、主持人控制 | 🚧 Phase 2e-2 代码开发中（Task 0-7 ✅：12 REST + 13 WS 事件 + HTTPMediaOrchestrator 真实 Router/Transport/Producer/Consumer 落地；Task 8-16 待执行）；专用设计：`docs/plans/2026-04-21-phase2e-2-design.md` | +| meeting | 会议创建/管理、WebSocket 信令转发、mediasoup 资源编排（Go→Node HTTP）、**会议生命周期状态机（host 宽限期 + 自动转让 + 空房 TTL）**、会议内聊天、主持人控制 | 🚧 Phase 2e-2 代码开发中（Task 0-8 ✅：12 REST + 13 WS 事件 + HTTPMediaOrchestrator 真实媒体链路 + MeetingLifecycleService + MeetingCleanupTask + Router 幂等双层防御；Task 9-16 待执行）；专用设计：`docs/plans/2026-04-21-phase2e-2-design.md` | | notify | 统一通知中心、好友/群聊事件 + 系统广播推送、11 种类型枚举、跨模块 Pusher 接口注入 | ✅ Phase 2e-1（10 种落地 + 2 种预留 `meeting_invite` / `meeting_reminder`） | | admin | 后台管理（用户管理 + 角色权限管理 + 在线监控 + 好友关系管理 + 群聊管理、会议监控、系统配置） | ✅ Phase 1/2a/2c | diff --git a/docs/plans/2026-04-21-phase2e-2-implementation.plan.md b/docs/plans/2026-04-21-phase2e-2-implementation.plan.md index 917d535..58e25ed 100644 --- a/docs/plans/2026-04-21-phase2e-2-implementation.plan.md +++ b/docs/plans/2026-04-21-phase2e-2-implementation.plan.md @@ -5,7 +5,7 @@ > **上级路线图：** [Phase 2e 整体路线图](./2026-04-20-phase2e-design.md) > **分支：** `feature/phase2e-2-meeting-mvp` > **预估总工时：** **约 17 人日**（17 个 Task，含 PoC 与 UI 打磨） -> **最后更新：** 2026-04-21（Task 0-7 ✅ 已落地，Go↔Node HTTP 媒体链路打通，下一步 Task 8 生命周期状态机） +> **最后更新：** 2026-04-21（Task 0-8 ✅ 已落地，会议生命周期状态机 E2E 20/20 PASS，下一步 Task 9 前端 mediasoup-client 接入） --- @@ -351,24 +351,44 @@ flowchart LR - 单元测试（`httptest.NewServer` 模拟）本次未落地，用更真实的"Go + 真 Node + Playwright-style E2E"替代，证据力更强；Task 10（可观测性）补齐单测 - **实际工作量**：**0.5 人日**（符合预估） -### Task 8：会议生命周期状态机（host 宽限期 + 自动转让 + 空房 TTL） +### Task 8：会议生命周期状态机（host 宽限期 + 自动转让 + 空房 TTL）✅ 已完成 -- **目标**：实现设计 §6.5 的状态机完整逻辑 -- **依赖**：T5 -- **主要产出**： - - `app/meeting/service/meeting_service.go` 补充： - - `OnHostDisconnect(ctx, roomCode)` → 写 `echo:meeting:host_grace:{code}` EX 120s - - `OnHostReconnect(ctx, roomCode, userID)` → 清除宽限期键 - - `HandleHostGraceExpired(ctx, roomCode)` → 转让主持（挑最早加入者）或销毁会议 - - `OnAllMembersLeft(ctx, roomCode)` → 设置房间 Redis key TTL 300s - - `OnRoomTTLExpired(ctx, roomCode)` → `status=2, ended_reason=empty_ttl`，清理 Node 资源 - - `app/meeting/task/meeting_cleanup_task.go`：每 30 秒扫描 Redis TTL + DB 状态，兜底清理 - - 事务包裹主持人转让（见设计 §11.3） -- **检查点**： - - 手测：host 关闭浏览器 → 2 分钟内重连，身份保留 → 超过 2 分钟自动转让给另一成员 - - 手测：全员退出 → 5 分钟后 DB `status=2`、Redis key 清空 - - 单元测试覆盖：转让事务回滚、并发转让竞态 -- **工作量**：**0.5 人日** +- **目标**：实现设计 §6.5 的状态机完整逻辑；同步修复 Task 7 遗留的"CreateRoom/JoinRoom 各自调一次 `CreateRouter`"行为 +- **依赖**：T5 / T7 +- **实际产出**： + - `backend/go-service/config/config.go` + `config.dev.yaml` / `config.docker.yaml`：新增 `MeetingConfig{HostGraceSeconds=120, EmptyRoomTTLSeconds=300, CleanupIntervalSeconds=30, StaleRoomHours=4}` + - `backend/go-service/app/meeting/service/meeting_lifecycle_service.go`（新，470 行）： + - `OnHostDisconnect(ctx, roomCode, hostID)` → 写 `echo:meeting:host_grace:{code}` + 本地 `time.AfterFunc`，Redis TTL = 业务时长 + `max(cleanup*2, 30s)` buffer + - `OnHostReconnect(ctx, roomCode, hostID)` → `DEL` key + 取消 timer + - `HandleHostGraceExpired(ctx, roomCode)` → `DEL` key（返回 1 确认唯一处理权）→ `TransferHost` 事务（最早加入者）+ `UpdateHost` + 广播 `meeting.host.changed{auto_reason=host_grace_expired}`；无其他活跃成员则走 `OnAllMembersLeft` + - `OnAllMembersLeft(ctx, roomCode)` → 写 `echo:meeting:empty_ttl:{code}` + 本地 timer + - `CancelEmptyTTL(ctx, roomCode)` → `DEL` key + 取消 timer（新成员加入复活） + - `HandleEmptyRoomExpired(ctx, roomCode)` → `DEL` key → `MarkEnded(reason=empty_ttl)` + `CloseRouter` 幂等 + 广播 `meeting.room.ended` + - `RescheduleFromRedis` / `ScanExpired` / `HostGracePrefix` / `EmptyTTLPrefix` 辅助（供 cleanup task 兜底） + - `backend/go-service/app/meeting/dao/meeting_room_dao.go`：新增 `ListStaleActive(hoursAgo, limit)` 扫陈旧活跃房间 + - `backend/go-service/app/meeting/service/interfaces.go`：`MediaOrchestrator` 新增 `ResolveRouterID(roomCode) (string, bool)` + - `backend/go-service/app/meeting/service/http_media_orchestrator.go`：`CreateRouter` 入口查 `sync.Map` 命中直接返回（HTTP 层幂等防御）；实现 `ResolveRouterID` 纯缓存读 + - `backend/go-service/app/meeting/service/meeting_service.go`：`JoinRoom` 移除 `CreateRouter`，改调 `lifecycleSvc.CancelEmptyTTL` + `ResolveRouterID`；`LeaveRoom` 空房分支改调 `lifecycleSvc.OnAllMembersLeft` + - `backend/go-service/app/meeting/service/meeting_signal_service.go`：新增 `OnWSDisconnect(userID)` 实现 `ws.MeetingDisconnectHook`；`OnRoomJoin` 追加 host 重连钩子 + - `backend/go-service/app/ws/handler.go`：定义 `MeetingDisconnectHook` 接口 + `SetMeetingDisconnectHook`，`SetOnDisconnect` 末尾 invoke + - `backend/go-service/app/meeting/task/meeting_cleanup_task.go`（新，220 行）：启动时 `RescheduleFromRedis`；每 `CleanupIntervalSeconds` 秒扫 `host_grace:*` / `empty_ttl:*` 兜底 + 扫 stale active rooms → `MarkEnded(reason=system_error)`；每小时清理 Ended 会议的聊天记录 + - `backend/go-service/app/provider/{provider,wire,wire_gen}.go` + `cmd/server/main.go`：注入 `MeetingLifecycleService` / `MeetingCleanupTask` + `wire.Bind(new(ws.MeetingDisconnectHook), new(*service.MeetingSignalService))` + 启动/停止 cleanup task + - `media-server/src/app.ts`：`/internal/info` 追加 `stats.routers` 字段供 E2E 断言 Router 幂等 + - `docs/verify/meeting_t8_verify.mjs`（新，320 行）：5 场景 E2E 脚本 +- **实际检查点**（全部通过）： + - `go build ./...` / `go vet ./...` / `ReadLints` 全绿（保留 Task 7 遗留 pre-existing 警告） + - E2E 脚本 `meeting_t8_verify.mjs` **20/20 PASS**（含 Router 幂等断言） + - 场景覆盖： + - S1：host 掉线 3s 后宽限期过期 → 自动转让给 B，`meeting.host.changed` 广播 `auto_reason=host_grace_expired`，DB `host_id` 更新 + - S2：host 掉线后 1s 内重连 → `host_grace` key DEL，身份保留 + - S3：全员 leave → `empty_ttl` 写入 → C 加入 → key DEL + 房间仍 Active + - S4：全员 leave → 3s 后 `empty_ttl` 过期 → 房间 Ended → 再 join 被拒 + - S5：`CreateRoom` 使 `stats.routers` +1；`JoinRoom` 不再触发新 Router +- **关键设计决策**： + - **Redis TTL 加 buffer 避免 DEL 与自动过期并发误判**：E2E 调试时发现若 Redis key TTL 与本地 `time.AfterFunc` 时长相同，本地 timer 触发时 key 已被 Redis 自动删除，`DEL` 返回 0 被误判为"已被其他路径处理"而跳过业务逻辑；修复：Redis TTL = 业务时长 + `max(CleanupIntervalSeconds*2, 30s)` + - **普通成员 WS 断线不动 DB**（决策 `q1_nonhost_disconnect=a1_keep_current`）：仅清理 WS/media 资源，`meeting_participants` 记录保留；长期不活跃由 4 小时后台扫描兜底 + - **Router 幂等双层防御**（决策 `q2_router_dedup=a2_both`）：业务层 `JoinRoom` 不调 `CreateRouter` + HTTP 层 `CreateRouter` 入口查 `sync.Map` 幂等 +- **实际工作量**：**0.5 人日**（符合预估） ### Task 9：前端 mediasoup-client 集成 + Pinia Store diff --git a/docs/progress/CURRENT_STATUS.md b/docs/progress/CURRENT_STATUS.md index e10cb00..cf61db0 100644 --- a/docs/progress/CURRENT_STATUS.md +++ b/docs/progress/CURRENT_STATUS.md @@ -1,7 +1,7 @@ # EchoChat 项目开发进度 -> **最后更新**：2026-04-21（Phase 2e-2 Task 7 HTTPMediaOrchestrator 落地，Go↔Node media-server HTTP 链路真实打通，端到端 16/16 PASS） -> **当前阶段**：Phase 2e-2 会议 MVP **代码开发阶段** 🚧（Task 0-7 ✅ / Task 8-16 待执行） +> **最后更新**：2026-04-21（Phase 2e-2 Task 8 会议生命周期状态机落地，host 宽限期 + 自动转让 + 空房 TTL + Router 幂等，E2E 20/20 PASS） +> **当前阶段**：Phase 2e-2 会议 MVP **代码开发阶段** 🚧（Task 0-8 ✅ / Task 9-16 待执行） > **当前分支**：`feature/phase2e-2-meeting-mvp`（从 `feature/phase2c-group-read-receipt` 衍生） > **Phase 2e 整体设计**：`docs/plans/2026-04-20-phase2e-design.md`（三子阶段路线图 + 后续规划清单） > **Phase 2e-1 专用设计**：`docs/plans/2026-04-20-phase2e-1-design.md`（✅ 已完成） @@ -170,6 +170,87 @@ --- +## 🎯 2026-04-21 Phase 2e-2 Task 8 会议生命周期状态机落地（host 宽限期 + 自动转让 + 空房 TTL + Router 幂等） + +**交付**：新增 `MeetingLifecycleService` + `MeetingCleanupTask`，落地设计 §6.5 的 5 类状态跃迁副作用（host 掉线宽限期、宽限期内重连、自动转让、空房 TTL、TTL 过期销毁）；同步修复 Task 7 遗留的 "CreateRoom/JoinRoom 各自调一次 `CreateRouter`" 行为，业务层 `JoinRoom` 不再触发 `CreateRouter` + HTTP 层 `HTTPMediaOrchestrator.CreateRouter` 增加 `sync.Map` 幂等防御；端到端验证脚本 `docs/verify/meeting_t8_verify.mjs` **20/20 PASS**，覆盖 5 个核心场景（含 Router 数量断言）。 + +### 产出文件 + +| 文件 | 行数 | 作用 | +|---|---|---| +| `backend/go-service/config/config.go`（改） | +14 | 新增 `MeetingConfig{HostGraceSeconds, EmptyRoomTTLSeconds, CleanupIntervalSeconds, StaleRoomHours}` | +| `backend/go-service/config/config.dev.yaml` / `config.docker.yaml`（改） | +7 | `meeting` 段默认值：`host_grace_seconds=120` / `empty_room_ttl_seconds=300` / `cleanup_interval_seconds=30` / `stale_room_hours=4` | +| `backend/go-service/app/constants/meeting.go`（改） | +10 | `MeetingHostGraceSeconds=120` / `MeetingEmptyRoomTTLSeconds=300` / `MeetingEndedReasonEmptyTTL="empty_ttl"` / `MeetingLeftReasonDisconnect="disconnect"` 等兜底常量 | +| `backend/go-service/app/meeting/service/meeting_lifecycle_service.go`（新） | 470 | `MeetingLifecycleService`：5 钩子（`OnHostDisconnect` / `OnHostReconnect` / `HandleHostGraceExpired` / `OnAllMembersLeft` / `CancelEmptyTTL` / `HandleEmptyRoomExpired`）+ `sync.Map` 本地 timer + `RescheduleFromRedis` 重启恢复 + `ScanExpired` 兜底扫描；Redis key TTL = 业务时长 + buffer（`max(cleanup*2, 30s)`）避免与本地 timer 同时到期导致 DEL 误判 | +| `backend/go-service/app/meeting/dao/meeting_room_dao.go`（改） | +20 | 新增 `ListStaleActive(hoursAgo, limit)`：扫描 `status != Ended && COALESCE(started_at, created_at) < NOW() - N 小时` 的房间，供 cleanup task 兜底回收 | +| `backend/go-service/app/meeting/service/interfaces.go`（改） | +8 | `MediaOrchestrator` 新增 `ResolveRouterID(roomCode) (string, bool)`，让业务层直接查缓存而非重复创建；`NoopMediaOrchestrator` 同步实现 | +| `backend/go-service/app/meeting/service/http_media_orchestrator.go`（改） | +12 | `CreateRouter` 入口先查 `sync.Map` 命中则直接返回（HTTP 层幂等防御）；新增 `ResolveRouterID` 纯缓存读 | +| `backend/go-service/app/meeting/service/meeting_service.go`（改） | ±25 | `JoinRoom` 移除 `CreateRouter` 调用，改为 `lifecycleSvc.CancelEmptyTTL` 复活空房 + `mediaOrchestrator.ResolveRouterID` 返回已缓存的 router_id；`LeaveRoom` 空房分支改调 `lifecycleSvc.OnAllMembersLeft`（启动 empty_ttl），不再立即 `MarkEnded` | +| `backend/go-service/app/meeting/service/meeting_signal_service.go`（改） | +60 | 新增 `OnWSDisconnect(userID)` 实现 `ws.MeetingDisconnectHook`：清理该用户所有 media 资源；若为 host 调 `lifecycleSvc.OnHostDisconnect` 启动宽限期。`OnRoomJoin` 新增"若为 host 调 `lifecycleSvc.OnHostReconnect`"逻辑 | +| `backend/go-service/app/ws/handler.go`（改） | +30 | 新增 `MeetingDisconnectHook` 接口 + `SetMeetingDisconnectHook`；`SetOnDisconnect` 回调末尾 invoke `OnWSDisconnect`，解除 ws/meeting 循环依赖 | +| `backend/go-service/app/meeting/task/meeting_cleanup_task.go`（新） | 220 | `MeetingCleanupTask`：启动时 `RescheduleFromRedis` 重建本地 timer；周期性 `ScanExpired` 扫 `host_grace:*` / `empty_ttl:*` 兜底触发；周期性扫 stale active rooms → `MarkEnded(reason=system_error)`；每小时清理已结束会议的旧聊天（保留 48h） | +| `backend/go-service/app/meeting/provider.go`（改） | +4 | `MeetingSet` 追加 `NewMeetingLifecycleService` / `NewMeetingCleanupTask` | +| `backend/go-service/app/provider/{provider,wire,wire_gen}.go`（改） | +25 | `App` 结构体新增 `MeetingLifecycleSvc` / `MeetingCleanupTask` 字段；`wire.Bind(new(ws.MeetingDisconnectHook), new(*service.MeetingSignalService))` 完成钩子注入 | +| `backend/go-service/cmd/server/main.go`（改） | +5 | 服务启动 `app.MeetingCleanupTask.Start()`；优雅退出 `defer app.MeetingCleanupTask.Stop()` | +| `media-server/src/app.ts`（改） | +8 | `/internal/info` 响应追加 `stats.routers`（总数）+ `routers[]`（roomCode/routerId/ageMs）供 E2E 断言 Router 幂等 | +| `docs/verify/meeting_t8_verify.mjs`（新） | 320 | E2E 验证脚本：5 个场景覆盖 host 宽限期过期/重连保留、空房 TTL 复活/过期销毁、JoinRoom 不重复调 `CreateRouter`；通过 `ECHOCHAT_MEETING_HOST_GRACE_SECONDS` 等 env 加速；直接 Redis 断言 key 状态，不依赖日志 | + +### 关键设计决策 + +1. **本地 timer + Redis key 双保险**：低延迟由 `time.AfterFunc` 触发；单实例进程重启由 `RescheduleFromRedis` 按 Redis `PTTL` 重建；多实例 / timer 丢失场景由后台 `MeetingCleanupTask` 每 `CleanupIntervalSeconds` 秒兜底扫描。 +2. **Redis key TTL 加 buffer**：Redis key TTL = 业务时长 + `max(CleanupIntervalSeconds*2, 30s)`，保证本地 timer 先触发，`DEL` 能命中；否则若两者同时到期，Redis 自动过期 key 会导致 `DEL` 返回 0 被误判为 "已被其他节点处理" 而跳过业务逻辑（E2E 调试过程实测到此 bug 并修复）。 +3. **并发去重**：`HandleHostGraceExpired` / `HandleEmptyRoomExpired` 入口一律先 `DEL` Redis key，返回 `1` 才继续执行业务逻辑（转让 / 销毁）；返回 `0` 直接返回，避免多实例 / 多触发路径重复操作。 +4. **Router 幂等双层防御**： + - **业务层**：`JoinRoom` 不再调 `CreateRouter`，改为 `ResolveRouterID` 查缓存；Router 创建只在 `CreateRoom` 发生一次。 + - **HTTP 层**：`HTTPMediaOrchestrator.CreateRouter` 入口查 `sync.Map(roomCode -> routerID)`，命中直接返回缓存值，不发 HTTP。 + - E2E 场景 5 断言：`CreateRoom` 后 `stats.routers` +1，`JoinRoom` 后 `stats.routers` 不变。 +5. **WS 断开钩子通过接口解耦**：`ws` 包定义 `MeetingDisconnectHook` 接口并通过 `SetMeetingDisconnectHook` 注入，避免 `ws → meeting` 包反向依赖；`MeetingSignalService` 实现该接口。普通成员 WS 断开仅清理 media 资源（设计决策 `q1_nonhost_disconnect=a1_keep_current`：不改 `meeting_participants` 记录，长期不活跃由 4 小时后台清理兜底）。 +6. **全员 leave 不再立即销毁房间**：`LeaveRoom` 检测到活跃成员归零时调 `OnAllMembersLeft` 启动 `empty_ttl`，TTL 内有新成员加入（`JoinRoom` 调 `CancelEmptyTTL`）则房间复活；TTL 过期才 `MarkEnded(reason=empty_ttl) + CloseRouter`。 + +### E2E 验证（20/20 PASS） + +```text +Config: HOST_GRACE=3s EMPTY_TTL=3s CLEANUP=1s +PASS: media-server healthz +-- Scenario 1: host grace expired → auto transfer -- +PASS: S1: create room ok +PASS: S1: host_grace key written after host disconnect +PASS: S1: host_grace key cleared after expiry +PASS: S1: meeting.host.changed broadcasted +PASS: S1: new_host_id = B +PASS: S1: auto_reason = host_grace_expired +PASS: S1: DB host_id updated to B +-- Scenario 2: host reconnect within grace period -- +PASS: S2: host_grace key written +PASS: S2: host_grace key DEL on reconnect +PASS: S2: host identity preserved +-- Scenario 3: empty_ttl revival on new join -- +PASS: S3: empty_ttl key written on empty room +PASS: S3: new user join succeeds within TTL +PASS: S3: empty_ttl key DEL on join +PASS: S3: room remains Active after revival +-- Scenario 4: empty_ttl expiry → room Ended -- +PASS: S4: empty_ttl set +PASS: S4: empty_ttl cleared after expiry +PASS: S4: join Ended room is rejected +-- Scenario 5: JoinRoom should NOT create duplicate Router -- +PASS: S5: Router +1 after CreateRoom +PASS: S5: Router count unchanged after JoinRoom (no duplicate) +PASS=20 FAIL=0 +``` + +### 已知待改进项（留给后续 Task） + +1. **`rescheduleByPattern` 重建 timer 使用完整 Redis PTTL**：当前服务重启后，本地 timer 会按 "业务时长 + buffer" 的完整 PTTL 重建，导致 worst-case 触发延迟最多晚 buffer 秒（默认 30s）。后续可在 `host_grace` 的 payload 中引入 `grace_until` 字段 + `empty_ttl` 改 JSON payload 存 `until`，按业务到期时间精确恢复。 +2. **cleanup task 判定基于 Redis TTL**：`ScanExpired` 当前按 `PTTL <= 0` 判定过期，这意味着兜底触发时间晚于"业务到期时间" buffer 秒。正常情况下本地 timer 先触发，兜底仅处理 timer 丢失 / 多实例场景，延迟可接受；若需要精准兜底可按 payload.until 对比 `time.Now()`。 +3. **多实例 grace_lock**：设计 §5.4 预留 `echo:meeting:grace_lock:{code}` 用于多实例并发保护，当前单实例部署未实装，Phase 2e-3 多实例化时补齐。 + +### 下一步 + +进入 **Task 9：Vue 前端 mediasoup-client 接入 + 会议室页面骨架**，依赖 Task 6 的 WS 信令契约与 Task 7 的真实 media 链路；预估 2 人日。 + +--- + ## 🎯 2026-04-21 Phase 2e-2 Task 7 HTTPMediaOrchestrator 落地（Go↔Node 真实媒体链路打通） **交付**：将 `NoopMediaOrchestrator` 替换为真实的 `HTTPMediaOrchestrator`，`MeetingService` / `MeetingSignalService` 的 9 个 media 调用全部走 HTTP 到 Node `media-server` 的 `/internal/v1/*` API；端到端验证脚本 `docs/verify/meeting_t7_verify.mjs` **16/16 PASS**，日志确认 Go 真正触发 mediasoup 真实 `router created` / `webrtc transport created` / `router closed explicitly` 事件。 diff --git a/docs/verify/meeting_t8_verify.mjs b/docs/verify/meeting_t8_verify.mjs new file mode 100644 index 0000000..0b9822f --- /dev/null +++ b/docs/verify/meeting_t8_verify.mjs @@ -0,0 +1,318 @@ +#!/usr/bin/env node +// Task 8 端到端验证脚本：会议生命周期状态机 +// 覆盖： +// 1. host 掉线 → 宽限期过期自动转让（其他成员收 host.changed，DB host_id 更新） +// 2. host 掉线 → 宽限期内重连 → host 身份保留 +// 3. 全员 leave → empty_ttl 存在 → 新用户 join 复活（empty_ttl key 被 DEL） +// 4. 全员 leave → 等待 TTL 过期 → Node Router 被关闭（CloseRouter 日志佐证）+ DB 状态 Ended +// 5. JoinRoom 不再重复调 CreateRouter（Node 侧仅 1 次 router created 日志） +// +// 运行前置： +// - 必须设置短时长环境变量加速触发，推荐： +// ECHOCHAT_MEETING_HOST_GRACE_SECONDS=3 +// ECHOCHAT_MEETING_EMPTY_ROOM_TTL_SECONDS=3 +// ECHOCHAT_MEETING_CLEANUP_INTERVAL_SECONDS=1 +// 然后启动 go-service +// - media-server 必须已启动（端口 3300） + +import WebSocket from 'ws'; +import { createClient } from 'redis'; + +const GO = 'http://localhost:8085/api/v1'; +const NODE = 'http://localhost:3300'; +const NODE_TOKEN = 'dev-token-abcdef1234567890'; + +// 兼容上游通过环境变量传入缩短后的时长（默认与 E2E 推荐一致） +const HOST_GRACE_SECONDS = Number(process.env.ECHOCHAT_MEETING_HOST_GRACE_SECONDS || 3); +const EMPTY_TTL_SECONDS = Number(process.env.ECHOCHAT_MEETING_EMPTY_ROOM_TTL_SECONDS || 3); +const CLEANUP_SECONDS = Number(process.env.ECHOCHAT_MEETING_CLEANUP_INTERVAL_SECONDS || 1); + +let pass = 0; +let fail = 0; +const assert = (cond, name, detail) => { + if (cond) { + pass++; + console.log(`PASS: ${name}`); + } else { + fail++; + console.log(`FAIL: ${name}`, detail ?? ''); + } +}; + +// ============ HTTP utils ============ + +async function httpJSON(method, url, body, headers = {}) { + const opts = { method, headers: { 'Content-Type': 'application/json', ...headers } }; + if (body !== undefined) opts.body = JSON.stringify(body); + const r = await fetch(url, opts); + const text = await r.text(); + let data = null; + try { data = JSON.parse(text); } catch (_) { /* non-json */ } + return { status: r.status, data, text }; +} + +async function register(suffix) { + const email = `t8user${suffix}@test.local`; + const password = 'TestPassword123!'; + const username = `t8user${suffix}`; + await httpJSON('POST', `${GO}/auth/register`, { email, username, password }); + const login = await httpJSON('POST', `${GO}/auth/login`, { account: email, password }); + if (!login.data?.data?.token) throw new Error(`login failed ${JSON.stringify(login)}`); + return { token: login.data.data.token, userID: login.data.data.user?.id }; +} + +// ============ WS utils ============ + +function connectWS(token) { + return new Promise((resolve, reject) => { + const ws = new WebSocket(`ws://localhost:8085/ws?token=${token}`); + ws.on('open', () => resolve(ws)); + ws.on('error', reject); + setTimeout(() => reject(new Error('ws open timeout')), 5000); + }); +} + +let _seqCounter = 1; +function wsSend(ws, event, data) { + return new Promise((resolve) => { + const seq = _seqCounter++; + const ackEvent = `${event}.ack`; + const onMsg = (raw) => { + try { + const m = JSON.parse(raw.toString()); + if (m.event === ackEvent && m.seq === seq) { + ws.off('message', onMsg); + resolve(m); + } + } catch {} + }; + ws.on('message', onMsg); + ws.send(JSON.stringify({ event, seq, data, time: new Date().toISOString() })); + setTimeout(() => { + ws.off('message', onMsg); + resolve(null); + }, 6000); + }); +} + +// 收集指定 event 的推送消息到内存 buffer（用于非 ACK 广播事件断言） +function collectPush(ws, eventName, buffer) { + const listener = (raw) => { + try { + const m = JSON.parse(raw.toString()); + if (m.event === eventName) buffer.push(m); + } catch {} + }; + ws.on('message', listener); + return () => ws.off('message', listener); +} + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +// ============ Redis utils（直接断言 key 状态，避免依赖服务日志）============ + +async function connectRedis() { + const c = createClient({ url: 'redis://localhost:6379/0' }); + await c.connect(); + return c; +} + +// ============ Main ============ + +(async () => { + console.log('==== Task 8 Meeting Lifecycle E2E Verify ===='); + console.log(`Config: HOST_GRACE=${HOST_GRACE_SECONDS}s EMPTY_TTL=${EMPTY_TTL_SECONDS}s CLEANUP=${CLEANUP_SECONDS}s`); + + const redis = await connectRedis(); + + // 预检：media-server 活着 + const mHealth = await httpJSON('GET', `${NODE}/healthz`); + assert(mHealth.status === 200 && mHealth.data?.ok === true, 'media-server healthz'); + + const ts = Date.now().toString().slice(-6); + + // ===== 场景 1：host 宽限期过期自动转让 ===== + console.log('\n-- Scenario 1: host grace expired → auto transfer --'); + { + const a = await register(`a1${ts}`); + const b = await register(`b1${ts}`); + const createRes = await httpJSON('POST', `${GO}/meeting/rooms`, { + title: 'T8 Scene1', type: 1, + }, { Authorization: `Bearer ${a.token}` }); + assert(createRes.status === 201, 'S1: create room ok'); + const code = createRes.data?.data?.room?.room_code; + const roomID = createRes.data?.data?.room?.id; + + await httpJSON('POST', `${GO}/meeting/rooms/${code}/join`, {}, { Authorization: `Bearer ${b.token}` }); + + const wsA = await connectWS(a.token); + const wsB = await connectWS(b.token); + await wsSend(wsA, 'meeting.room.join', { room_code: code }); + await wsSend(wsB, 'meeting.room.join', { room_code: code }); + + // 监听 B 上的 host.changed 广播 + const hostChangedBuf = []; + const stopCollect = collectPush(wsB, 'meeting.host.changed', hostChangedBuf); + + // A（host）断开 WS → 写入 host_grace key + wsA.close(); + await sleep(500); + const graceExists = await redis.exists(`echo:meeting:host_grace:${code}`); + assert(graceExists === 1, 'S1: host_grace key written after host disconnect'); + + // 等宽限期 + cleanup 一轮 + await sleep((HOST_GRACE_SECONDS + CLEANUP_SECONDS + 1) * 1000); + + const graceGone = await redis.exists(`echo:meeting:host_grace:${code}`); + assert(graceGone === 0, 'S1: host_grace key cleared after expiry'); + assert(hostChangedBuf.length >= 1, 'S1: meeting.host.changed broadcasted', hostChangedBuf); + if (hostChangedBuf.length > 0) { + const payload = hostChangedBuf[0]?.data; + assert(payload?.new_host_id === b.userID, 'S1: new_host_id = B', payload); + assert(payload?.auto_reason === 'host_grace_expired', 'S1: auto_reason = host_grace_expired', payload); + } + + // DB 状态：通过 REST /rooms/:code 查看 host_id（B 的视角） + const roomInfo = await httpJSON('GET', `${GO}/meeting/rooms/${code}`, undefined, { Authorization: `Bearer ${b.token}` }); + assert(roomInfo.data?.data?.room?.host_id === b.userID, 'S1: DB host_id updated to B', roomInfo.data); + + stopCollect(); + wsB.close(); + // 清理会议（通过新 host B） + await httpJSON('POST', `${GO}/meeting/rooms/${code}/end`, {}, { Authorization: `Bearer ${b.token}` }); + await sleep(300); + } + + // ===== 场景 2：host 宽限期内重连保留身份 ===== + console.log('\n-- Scenario 2: host reconnect within grace period --'); + { + const a = await register(`a2${ts}`); + const b = await register(`b2${ts}`); + const createRes = await httpJSON('POST', `${GO}/meeting/rooms`, { + title: 'T8 Scene2', type: 1, + }, { Authorization: `Bearer ${a.token}` }); + const code = createRes.data?.data?.room?.room_code; + await httpJSON('POST', `${GO}/meeting/rooms/${code}/join`, {}, { Authorization: `Bearer ${b.token}` }); + + let wsA = await connectWS(a.token); + const wsB = await connectWS(b.token); + await wsSend(wsA, 'meeting.room.join', { room_code: code }); + await wsSend(wsB, 'meeting.room.join', { room_code: code }); + + // A 掉线 → 立刻重连（远小于 HOST_GRACE_SECONDS） + wsA.close(); + await sleep(500); + const graceExists = await redis.exists(`echo:meeting:host_grace:${code}`); + assert(graceExists === 1, 'S2: host_grace key written'); + + // 重连 + room.join 触发 OnHostReconnect + wsA = await connectWS(a.token); + await wsSend(wsA, 'meeting.room.join', { room_code: code }); + await sleep(500); + + const graceGone = await redis.exists(`echo:meeting:host_grace:${code}`); + assert(graceGone === 0, 'S2: host_grace key DEL on reconnect'); + + // 继续等过完原本的宽限期 + cleanup，无 host 变更即通过 + await sleep((HOST_GRACE_SECONDS + CLEANUP_SECONDS + 1) * 1000); + const roomInfo = await httpJSON('GET', `${GO}/meeting/rooms/${code}`, undefined, { Authorization: `Bearer ${a.token}` }); + assert(roomInfo.data?.data?.room?.host_id === a.userID, 'S2: host identity preserved', roomInfo.data); + + wsA.close(); + wsB.close(); + await httpJSON('POST', `${GO}/meeting/rooms/${code}/end`, {}, { Authorization: `Bearer ${a.token}` }); + await sleep(300); + } + + // ===== 场景 3：空房 TTL 复活 ===== + console.log('\n-- Scenario 3: empty_ttl revival on new join --'); + { + const a = await register(`a3${ts}`); + const c = await register(`c3${ts}`); + const createRes = await httpJSON('POST', `${GO}/meeting/rooms`, { + title: 'T8 Scene3', type: 1, + }, { Authorization: `Bearer ${a.token}` }); + const code = createRes.data?.data?.room?.room_code; + + // host 自己 leave → 空房 TTL 启动 + await httpJSON('POST', `${GO}/meeting/rooms/${code}/leave`, {}, { Authorization: `Bearer ${a.token}` }); + await sleep(300); + const ttlExists = await redis.exists(`echo:meeting:empty_ttl:${code}`); + assert(ttlExists === 1, 'S3: empty_ttl key written on empty room'); + + // TTL 内 C 新加入 → 撤销 TTL，房间保持 Active + const joinRes = await httpJSON('POST', `${GO}/meeting/rooms/${code}/join`, {}, { Authorization: `Bearer ${c.token}` }); + assert(joinRes.status === 200, 'S3: new user join succeeds within TTL', joinRes.data); + await sleep(300); + const ttlGone = await redis.exists(`echo:meeting:empty_ttl:${code}`); + assert(ttlGone === 0, 'S3: empty_ttl key DEL on join'); + + const roomInfo = await httpJSON('GET', `${GO}/meeting/rooms/${code}`, undefined, { Authorization: `Bearer ${c.token}` }); + assert(roomInfo.data?.data?.room?.status === 1, 'S3: room remains Active after revival', roomInfo.data); + + // 清理：C 结束或 leave（C 非 host，走 leave→空房→TTL；等其过期兜底） + await httpJSON('POST', `${GO}/meeting/rooms/${code}/leave`, {}, { Authorization: `Bearer ${c.token}` }); + await sleep((EMPTY_TTL_SECONDS + CLEANUP_SECONDS + 1) * 1000); + } + + // ===== 场景 4：空房 TTL 过期 → 自动销毁 ===== + console.log('\n-- Scenario 4: empty_ttl expiry → room Ended --'); + { + const a = await register(`a4${ts}`); + const createRes = await httpJSON('POST', `${GO}/meeting/rooms`, { + title: 'T8 Scene4', type: 1, + }, { Authorization: `Bearer ${a.token}` }); + const code = createRes.data?.data?.room?.room_code; + + await httpJSON('POST', `${GO}/meeting/rooms/${code}/leave`, {}, { Authorization: `Bearer ${a.token}` }); + await sleep(300); + assert((await redis.exists(`echo:meeting:empty_ttl:${code}`)) === 1, 'S4: empty_ttl set'); + + await sleep((EMPTY_TTL_SECONDS + CLEANUP_SECONDS + 1) * 1000); + assert((await redis.exists(`echo:meeting:empty_ttl:${code}`)) === 0, 'S4: empty_ttl cleared after expiry'); + + // 会议已结束 → 再 join 报已结束 + const b = await register(`b4${ts}`); + const joinAgain = await httpJSON('POST', `${GO}/meeting/rooms/${code}/join`, {}, { Authorization: `Bearer ${b.token}` }); + assert(joinAgain.status === 400 || joinAgain.status === 404 || joinAgain.status === 410, + 'S4: join Ended room is rejected', joinAgain.data); + } + + // ===== 场景 5：JoinRoom 不再重复调 CreateRouter ===== + console.log('\n-- Scenario 5: JoinRoom should NOT create duplicate Router --'); + { + // 记录 Node 当前 Router 数（通过 /internal/info） + const info0 = await httpJSON('GET', `${NODE}/internal/info`, undefined, { 'X-Internal-Token': NODE_TOKEN }); + const routers0 = info0.data?.stats?.routers ?? info0.data?.routers ?? 0; + + const a = await register(`a5${ts}`); + const b = await register(`b5${ts}`); + const createRes = await httpJSON('POST', `${GO}/meeting/rooms`, { + title: 'T8 Scene5', type: 1, + }, { Authorization: `Bearer ${a.token}` }); + const code = createRes.data?.data?.room?.room_code; + + // 创建后应 +1 个 Router + const info1 = await httpJSON('GET', `${NODE}/internal/info`, undefined, { 'X-Internal-Token': NODE_TOKEN }); + const routers1 = info1.data?.stats?.routers ?? info1.data?.routers ?? 0; + assert(routers1 - routers0 === 1, 'S5: Router +1 after CreateRoom', { routers0, routers1 }); + + // B 加入不应再增加 Router + await httpJSON('POST', `${GO}/meeting/rooms/${code}/join`, {}, { Authorization: `Bearer ${b.token}` }); + const info2 = await httpJSON('GET', `${NODE}/internal/info`, undefined, { 'X-Internal-Token': NODE_TOKEN }); + const routers2 = info2.data?.stats?.routers ?? info2.data?.routers ?? 0; + assert(routers2 === routers1, 'S5: Router count unchanged after JoinRoom (no duplicate)', { routers1, routers2 }); + + // 清理 + await httpJSON('POST', `${GO}/meeting/rooms/${code}/end`, {}, { Authorization: `Bearer ${a.token}` }); + await sleep(500); + } + + await redis.quit(); + console.log('-----'); + console.log(`PASS=${pass} FAIL=${fail}`); + process.exit(fail > 0 ? 1 : 0); +})().catch((err) => { + console.error('script error:', err); + process.exit(2); +}); diff --git a/media-server/src/app.ts b/media-server/src/app.ts index 8d514f1..67f31bc 100644 --- a/media-server/src/app.ts +++ b/media-server/src/app.ts @@ -10,6 +10,7 @@ import { consumerRoutes } from './routes/consumer.route.js'; import { producerRoutes } from './routes/producer.route.js'; import { routerRoutes } from './routes/router.route.js'; import { transportRoutes } from './routes/transport.route.js'; +import { getRouterStats } from './services/router.service.js'; import { logger } from './utils/logger.js'; export async function buildApp() { @@ -52,6 +53,7 @@ export async function buildApp() { app.get('/internal/info', async () => { const snapshot = getWorkerSnapshot(); + const routerStats = getRouterStats(); return { service: 'media-server', version: '0.1.0', @@ -63,6 +65,10 @@ export async function buildApp() { rtcMinPort: config.mediasoup.rtcMinPort, rtcMaxPort: config.mediasoup.rtcMaxPort, }, + stats: { + routers: routerStats.total, + }, + routers: routerStats.rooms, }; });