diff --git a/script/livekit-poc/mp-gateway/.env.example b/script/livekit-poc/mp-gateway/.env.example new file mode 100644 index 0000000..4a392a2 --- /dev/null +++ b/script/livekit-poc/mp-gateway/.env.example @@ -0,0 +1,42 @@ +# 复制为 .env 后填写；.env 已被 .gitignore 忽略，勿提交真实凭据 +# 必须与后端 yudao.im.rtc.api-key / yudao.im.rtc.api-secret 一致，网关用它校验后端签发的房间 Token +LIVEKIT_API_KEY=replace-with-api-key +LIVEKIT_API_SECRET=replace-with-secret-at-least-32-chars + +# 网关试验开关：默认 false，网关容器打印说明后退出 +MP_GATEWAY_ENABLED=false + +# 后端 webhook（host 网络下容器可直接访问宿主机回环） +MP_LIVEKIT_WEBHOOK_URL=http://127.0.0.1:48080/admin-api/im/livekit/webhook + +# 本机回环联调保持 127.0.0.1；真机联调改为手机可达的主机名 / IP，并同步放开对应监听与防火墙 +MP_LIVEKIT_BIND_ADDRESS=127.0.0.1 +MP_LIVEKIT_NODE_IP=127.0.0.1 +MP_INGRESS_RTMP_PUBLIC_HOST=127.0.0.1 +MP_SRS_PLAY_PUBLIC_HOST=127.0.0.1 +MP_GATEWAY_API_HOST=127.0.0.1 + +# 资源上限：每个小程序会话 1 路 Ingress 转码，每个远端 1 路 Egress 转码 +MP_GATEWAY_MAX_SESSIONS=4 +MP_GATEWAY_MAX_EGRESS=8 + +# 网关专用 SRS app（yudao-mp-gw 或 yudao-mp-gw-<后缀>），该 app 只能给网关使用 +# MP_GATEWAY_SRS_APP=yudao-mp-gw + +# 其它可调参数，含义与默认值见 README「4. 配置」；不配置即使用默认值 +# MP_GATEWAY_POLL_INTERVAL_MS=2000 +# MP_GATEWAY_LEASE_TTL_MS=20000 +# MP_GATEWAY_MAX_SESSION_MS=7200000 +# MP_GATEWAY_MAX_REMOTES_PER_SESSION=4 +# MP_GATEWAY_EGRESS_RETRY_MS=10000 +# MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS=20000 +# MP_GATEWAY_RELEASE_RETRY_MS=5000 +# MP_GATEWAY_CLEANUP_RETRY_MS=5000 +# MP_GATEWAY_REQUEST_TIMEOUT_MS=10000 +# MP_GATEWAY_CLOCK_SKEW_SEC=30 +# MP_GATEWAY_EGRESS_WIDTH=640 +# MP_GATEWAY_EGRESS_HEIGHT=360 +# MP_GATEWAY_EGRESS_FRAMERATE=15 +# MP_GATEWAY_EGRESS_VIDEO_KBPS=600 +# MP_GATEWAY_EGRESS_AUDIO_KBPS=64 +# MP_GATEWAY_EGRESS_KEYFRAME_SEC=1 diff --git a/script/livekit-poc/mp-gateway/.gitignore b/script/livekit-poc/mp-gateway/.gitignore new file mode 100644 index 0000000..4c49bd7 --- /dev/null +++ b/script/livekit-poc/mp-gateway/.gitignore @@ -0,0 +1 @@ +.env diff --git a/script/livekit-poc/mp-gateway/README.md b/script/livekit-poc/mp-gateway/README.md new file mode 100644 index 0000000..d695006 --- /dev/null +++ b/script/livekit-poc/mp-gateway/README.md @@ -0,0 +1,224 @@ +# 小程序 RTMP ↔ LiveKit 网关（试验） + +> **试验功能，默认关闭，勿用于生产。** 小程序音视频经 RTMP 转码中转，端到端延迟预计为**秒级**（通常 1~3 秒，弱网更高），明显差于 PC / H5 / App 的 WebRTC 直连；回声消除、资源成本、真机兼容性均未经验收。 + +## 1. 为什么需要网关 + +微信小程序不能运行 `livekit-client`（无 WebRTC），原生 `live-pusher` 只能推 RTMP，`live-player` 只能拉 RTMP / FLV。网关在不更换 LiveKit 的前提下把小程序接入同一个房间： + +``` +小程序 live-pusher ──RTMP──▶ LiveKit Ingress ──WebRTC──▶ LiveKit 房间 ◀──WebRTC── PC / H5 / App + │ + 每个远端参与者 1 路 Track Composite Egress + ▼ +小程序 live-player ◀──RTMP── SRS（推流、播放均回调网关鉴权）◀──RTMP──┘ +``` + +- 小程序自己的 Ingress 参与者 identity 即后端 Token 内的 userId（不加后缀），与 PC / App 同账号进房时沿用 LiveKit 同 identity 互踢；后端 webhook 的 joined / left 照常生效，**无需修改 Java 代码**。 +- 每个远端单独一路流，**从不混入小程序自己的声音**；视频屏幕共享优先、摄像头回退；Egress、Agent、SIP 等非业务参与者不转出。 +- 群通话一个远端一路 live-player，单会话默认最多 4 路，超出部分如实返回 `overflow`。 + +## 2. 目录 + +| 文件 | 说明 | +| --- | --- | +| `src/server.mjs` | 入口：小程序会话 API（8090）+ SRS 回调（8091，仅回环） | +| `src/session-manager.mjs` | 会话、租约、Ingress / Egress 生命周期、SRS 鉴权、残留回收 | +| `src/jwt.mjs` | 校验后端签发的房间 Token（HS256）、签发管理 Token | +| `src/livekit-api.mjs` | LiveKit Twirp API 最小客户端（无第三方依赖） | +| `src/config.mjs` | 环境变量配置，`MP_GATEWAY_ENABLED=true` 才允许启动 | +| `docker-compose.yml` | 独立试验环境：Redis + LiveKit + Ingress + Egress + SRS + 网关 | +| `srs.conf` | SRS 配置，HTTP 回调鉴权 | +| `.env.example` | 凭据与地址占位，复制为 `.env` 使用 | + +网关进程只使用 Node.js 22 内置模块（`http`、`crypto`、`fetch`），无需安装 npm 依赖；运行时依赖 LiveKit Server、Ingress、Egress 与 SRS，由同目录 `docker-compose.yml` 一并启动。 + +## 3. 本机回环联调 + +与上级目录 `../docker-compose.yml`（LiveKit-only PoC）**二选一**，两者都占用 7880 等端口。 + +```bash +cd script/livekit-poc/mp-gateway +docker compose -f ../docker-compose.yml down # 停掉 LiveKit-only PoC +cp .env.example .env # 填写与后端 yudao.im.rtc.api-key / api-secret 一致的凭据，MP_GATEWAY_ENABLED=true +docker compose up -d +docker compose logs -f mp-gateway # 看到「小程序网关已启动（试验功能）」 +curl -s http://127.0.0.1:8090/healthz # {"status":"ok"} +``` + +- 全部服务使用 `network_mode: host`。macOS 需 Docker Desktop 4.34+，并在 Settings → Resources → Network 勾选「Enable host networking」，否则容器能启动但端口不通。 +- 后端 `yudao.im.rtc.livekit-url` 仍为 `ws://127.0.0.1:7880`；webhook 默认回调 `http://127.0.0.1:48080/admin-api/im/livekit/webhook`。 +- 本机端口：LiveKit 7880 / 7881 / 7882udp，Ingress RTMP 1935、WHIP 18080、relay 19090、UDP 7885，Egress 健康检查 18082，SRS 1936，Redis 16379，网关 8090 / 8091。LiveKit HTTP、SRS、Redis、网关默认只监听回环；**Ingress RTMP 与 LiveKit 媒体端口会监听所有网卡**（推流靠一次性 stream key 鉴权），在不可信局域网中请配合防火墙。 + +### 3.1 用 FFmpeg 模拟小程序 + +1. 用 H5 或 PC 账号 A 发起通话，账号 B（模拟小程序端）取得同一房间的 Token（例如在 H5 发起 / 接听接口响应中取得 `token`，**只在本机终端使用，勿写日志或提交**）。 +2. 创建会话： + + ```bash + curl -s -X POST -H "Authorization: Bearer $ROOM_TOKEN" http://127.0.0.1:8090/mp-rtc/sessions + ``` + +3. 用返回的 `pushUrl` 推测试流（模拟 live-pusher）： + + ```bash + ffmpeg -re -f lavfi -i testsrc=size=640x360:rate=15 -f lavfi -i sine=frequency=440 \ + -c:v libx264 -preset veryfast -tune zerolatency -g 15 -c:a aac -ar 48000 -f flv "$PUSH_URL" + ``` + + 账号 A 的页面应出现 B 的测试画面与提示音；会话查询 `publishing` 变为 `true`。 +4. 每 2 秒查询会话（同时续租），待远端 A 的 `state` 为 `ready` 后，用 `playUrl` 拉流（模拟 live-player）： + + ```bash + curl -s -H "Authorization: Bearer $SESSION_TOKEN" http://127.0.0.1:8090/mp-rtc/sessions/$SESSION_ID + ffplay -fflags nobuffer -flags low_delay "$PLAY_URL" + ``` + +5. 反向校验：去掉 `playUrl` 的 `token` 参数或改错一位，SRS 应拒绝播放；停止查询超过 `leaseTtlMs`，或 `DELETE` 会话后，Ingress、Egress 应被释放（`docker compose logs mp-gateway` 有「会话已关闭」）。 + +## 4. 配置 + +| 变量 | 默认值 | 说明 | +| --- | --- | --- | +| `MP_GATEWAY_ENABLED` | 无 | 必须为 `true` 才启动 | +| `MP_GATEWAY_API_HOST` / `_PORT` | `127.0.0.1` / `8090` | 小程序 API；公网部署放在 HTTPS 反向代理之后 | +| `MP_GATEWAY_HOOK_HOST` / `_PORT` | `127.0.0.1` / `8091` | SRS 回调，只允许 SRS 访问 | +| `MP_GATEWAY_LIVEKIT_HTTP_URL` | `http://127.0.0.1:7880` | LiveKit API 地址 | +| `MP_GATEWAY_LIVEKIT_API_KEY` / `_SECRET` | 必填 | 与后端一致；secret 至少 32 位 | +| `MP_GATEWAY_SRS_PUBLISH_BASE_URL` | `rtmp://127.0.0.1:1936/yudao-mp-gw` | Egress 推到 SRS 的内部地址；app 必须为网关专用的 `yudao-mp-gw` 或 `yudao-mp-gw-<后缀>`，否则拒绝启动 | +| `MP_GATEWAY_SRS_PLAY_BASE_URL` | 同上 | 小程序拉流的外部地址，app 需与推流一致 | +| `MP_GATEWAY_POLL_INTERVAL_MS` | `2000` | 建议轮询间隔，也是网关同步房间的间隔 | +| `MP_GATEWAY_LEASE_TTL_MS` | `20000` | 租约，至少 3 倍轮询间隔；超时未查询即释放 | +| `MP_GATEWAY_MAX_SESSION_MS` | 2 小时 | 单会话最长存活，另受房间 Token `exp` 限制 | +| `MP_GATEWAY_MAX_SESSIONS` | `4` | 全局会话数（每个 1 路 Ingress 转码） | +| `MP_GATEWAY_MAX_REMOTES_PER_SESSION` | `4` | 单会话最多拉取的远端 | +| `MP_GATEWAY_MAX_EGRESS` | `8` | 全局 Egress 路数 | +| `MP_GATEWAY_EGRESS_RETRY_MS` | `10000` | 单路 Egress 失败后的冷却 | +| `MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS` | `20000` | Egress 启动后等待 SRS `on_publish` 的最长时间，超时停止并按 `failed` 冷却重试 | +| `MP_GATEWAY_CLEANUP_RETRY_MS` | `5000` | 启动回收失败后的重试间隔 | +| `MP_GATEWAY_RELEASE_RETRY_MS` | `5000` | Ingress / Egress 释放失败后的重试间隔；确认释放前继续占用配额 | +| `MP_GATEWAY_REQUEST_TIMEOUT_MS` | `10000` | 调用 LiveKit API 的超时 | +| `MP_GATEWAY_CLOCK_SKEW_SEC` | `30` | 校验房间 Token `nbf` / `exp` 允许的时钟偏差 | +| `MP_GATEWAY_EGRESS_WIDTH` / `HEIGHT` / `FRAMERATE` / `VIDEO_KBPS` / `AUDIO_KBPS` / `KEYFRAME_SEC` | 640 / 360 / 15 / 600 / 64 / 1 | Egress 编码（H264 Baseline + AAC） | + +上表是网关进程直接读取的环境变量，自行运行 `node src/server.mjs` 时全部按此配置。使用本目录 `docker-compose.yml` 时，网关容器的环境由 compose 组装，`.env` 中的变量并非全部原样转发： + +- **compose 固定、`.env` 不生效**：`MP_GATEWAY_API_PORT=8090`、`MP_GATEWAY_HOOK_HOST=127.0.0.1`、`MP_GATEWAY_HOOK_PORT=8091`、`MP_GATEWAY_LIVEKIT_HTTP_URL=http://127.0.0.1:7880`，与 `srs.conf` 及 compose 内各服务端口一一对应。 +- **改名转发**：LiveKit 凭据读取 `.env` 的 `LIVEKIT_API_KEY` / `LIVEKIT_API_SECRET`（同时供 LiveKit、Ingress、Egress 使用），写 `MP_GATEWAY_LIVEKIT_API_KEY` / `_SECRET` 不生效。 +- **由 compose 变量拼出**：`MP_GATEWAY_SRS_PUBLISH_BASE_URL` 固定为 `rtmp://127.0.0.1:1936/`，`MP_GATEWAY_SRS_PLAY_BASE_URL` 为 `rtmp://:1936/`；`` 取 `MP_GATEWAY_SRS_APP`（默认 `yudao-mp-gw`），`MP_SRS_PLAY_PUBLIC_HOST` 默认 `127.0.0.1`。 +- **原样转发**：`MP_GATEWAY_ENABLED`（默认 `false`）、`MP_GATEWAY_API_HOST`（默认 `127.0.0.1`），以及从 `MP_GATEWAY_POLL_INTERVAL_MS` 到 Egress 编码的其余可调参数，未配置即用上表默认值（见 `.env.example`）。 + +## 5. 会话 API + +所有响应 `Cache-Control: no-store`，错误体为 `{ "code": "...", "message": "..." }`。不支持 CORS（小程序 `uni.request` 不受 CORS 限制）。 + +### 5.1 创建会话 `POST /mp-rtc/sessions` + +- 请求头 `Authorization: Bearer <后端签发的房间 Token>`；请求体可为空（上限 4KB）。**不得**携带后端登录 accessToken。 +- 网关严格校验：HS256 签名、`iss` = apiKey、`exp` / `nbf`（允许 30 秒时钟偏差）、`sub` 为数字 userId、`video.roomJoin` / `canPublish` / `canSubscribe` 为 `true`、非 hidden / recorder。房间与身份**只取自 Token**。 +- 同一用户在同一房间已有未释放的会话或推流资源（含释放中、创建中）时返回 `409 session_busy`，不会替换正在推流的会话：替换已进房的 Ingress 会触发 `participant_left`，后端会据此结束私聊通话。原会话 `DELETE` 或租约到期、资源确认释放后才能再次创建；若该房间 + 用户存在创建结果未知的占位，则需重启网关才能再次创建（见第 7 节）。网关不会返回已有会话的凭证。 + +`201` 响应： + +```json +{ + "sessionId": "32 位十六进制", + "sessionToken": "会话凭证，仅本会话后续请求使用", + "userId": 1, + "pushUrl": "rtmp://host:1935/x/", + "pollIntervalMs": 2000, + "leaseTtlMs": 20000, + "expiresAt": 1760000000000 +} +``` + +错误：`401` Token 无效 / 过期，`403` 授权不足，`429` 会话数已满，`502` LiveKit / Ingress 不可用或未配置 `ingress.rtmp_base_url`，`409` 同一用户已有会话（`session_busy`）或会话在创建过程中被关闭（`session_closed`），`503` 网关正在退出（`shutting_down`）或启动回收尚未完成（`cleanup_pending`）。创建过程中客户端断开，网关会立即释放刚创建的资源。 + +### 5.2 查询并续租 `GET /mp-rtc/sessions/{sessionId}` + +请求头 `Authorization: Bearer `；会话不存在、已结束或凭证错误统一返回 `404`。 + +```json +{ + "sessionId": "...", + "status": "active", + "publishing": true, + "remotes": [ + { "userId": 2, "state": "ready", "ready": true, "playUrl": "rtmp://host:1936/yudao-mp-gw/?token=<拉流凭证>", "hasAudio": true, "hasVideo": true, "videoVisible": true } + ], + "error": null, + "leaseTtlMs": 20000, + "pollIntervalMs": 2000, + "expiresAt": 1760000000000 +} +``` + +- `publishing`：小程序推流已作为 LiveKit 参与者进房（Ingress 参与者已有轨道）。 +- `remotes[].state`： + - `ready`：SRS 已收到该远端的流，`playUrl` 可播放；只有此状态下 `playUrl` 非空。 + - `starting`：Egress 启动中。 + - `waiting`：远端已进房，暂无可转出的音视频轨道。 + - `overflow`：超出单会话路数上限，不会转出。 + - `capacity`：全局 Egress 已满，空出后自动重试。 + - `failed`：Egress 启动失败、超过 `MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS` 仍未推到 SRS，或中途结束时冷却后自动重试；创建结果未知（见「7. 生命周期与资源」）时不再为该远端新建转流，直到网关重启。具体原因（`egress_failed` / `egress_no_media` / `egress_ended` / 创建结果未知）只记录在网关日志。 +- 远端媒体组合变化（如开始 / 结束屏幕共享）时会换新的流名与 `playUrl`，客户端需按最新值重新播放。 +- `error`：最近一次同步房间失败为 `livekit_unavailable`，否则为 `null`。 + +### 5.3 结束会话 `DELETE /mp-rtc/sessions/{sessionId}` + +请求头同上，成功 `204`，释放 Ingress 与全部 Egress。 + +### 5.4 健康检查 `GET /healthz` + +启动回收完成后 `200 {"status":"ok"}`；完成前 `503 {"status":"cleanup_pending"}`。 + +## 6. SRS 鉴权与凭证 + +- 每路 Egress 使用新的随机流名、推流密钥（`secret`）、拉流凭证（`token`）。推流密钥与会话凭证只保存摘要；拉流凭证需要在会话查询中重新下发 `playUrl`，因此明文保存在网关进程内存中，会话关闭或远端流移除时随之删除。所有凭证都不落盘、不写日志。 +- `on_publish`：只允许当前存活会话中对应流名 + 正确 `secret` 推流；`on_play`：只允许正确 `token` 播放；网关不可用时 SRS 拒绝。 +- 日志只记录会话编号、房间、userId、资源编号，**不记录**房间 Token、会话凭证、stream key、推拉流 URL；LiveKit API 出错时只记录方法、HTTP 状态与 Twirp 错误码，不记录上游返回的错误原文。 + +## 7. 生命周期与资源 + +- 会话超过 `leaseTtlMs` 未查询、到达 Token `exp` 或 `MP_GATEWAY_MAX_SESSION_MS`、房间被销毁（曾进房之后）时自动释放。 +- 会话关闭或远端流移除时立即撤销推拉流授权与会话查询；对应的 Ingress / Egress 删除失败时保留在内存待释放列表中，按 `MP_GATEWAY_RELEASE_RETRY_MS` 重试（同时最多 4 个重试请求），**确认释放前继续计入会话数与 Egress 路数上限**，同一用户也不能新建会话。 +- 异步创建 Ingress / Egress 期间会话被关闭或远端已变化时，结果返回后立即转入待释放；创建中的资源同样计入上限。 +- `SIGINT` / `SIGTERM` 时拒绝新会话，关闭全部会话，等待进行中的创建返回并确认全部资源释放后以 0 退出；15 秒内未能全部释放则以非零状态退出，剩余资源由下次启动回收或按 7.2 手动处理。 +- **创建结果未知**：创建 Ingress / Egress 时超时、网络错误、HTTP 5xx（含反向代理 502 / 504），或成功响应缺少资源编号，LiveKit 侧都可能已建好资源而网关拿不到编号。网关按 fail-closed 处理：该次创建作为占位**永久计入**会话数 / Egress 路数上限，同一房间 + 用户不能再新建会话（`409 session_busy`），发现结果未知后不再为对应远端新建转流，避免不断叠加未知资源：若该远端换轨后的新转流在旧请求结果未知前已经启动，则继续按其真实状态显示和运行，否则显示 `failed`；旧的未知占位仍计入配额直到重启。占位只能**重启网关**后由启动回收（Ingress 按名称前缀、Egress 按专用 app）清除；存在占位时网关退出以非零状态结束。出现日志「创建结果未知」后请择机重启网关。注意 LiveKit 在 Egress 容量不足时也可能返回 5xx，同样会被保守占位、需要重启恢复，这是试验阶段的取舍；4xx 等明确拒绝的失败按未创建处理，沿用冷却重试。 +- **单实例限制**：会话保存在内存，多实例共用同一 LiveKit 会互相清理，暂不支持。 + +### 7.1 启动回收与专用 SRS app + +- LiveKit 返回的 EgressInfo 会把 RTMP 地址最后一段（流名 + 查询参数）脱敏为 `{首 3 位...末 3 位}`，例如 `rtmp://127.0.0.1:1936/yudao-mp-gw/{yud...xyz}`，只有 scheme / host / port / app 可信。因此**网关用专用 app 标识归属**：`yudao-mp-gw`（或 `yudao-mp-gw-<后缀>`）只能给网关使用。 +- 网关启动后先回收残留：删除名称前缀为 `yudao-mp-gw-` 的 Ingress；停止输出地址为「`MP_GATEWAY_SRS_PUBLISH_BASE_URL`/单段流名」且流名为完整网关流名或 `{yud...xxx}` 脱敏形式的活跃 Egress。其它 app（如 `live`）下的 Egress 不受影响。 +- 回收任一步失败都会按 `MP_GATEWAY_CLEANUP_RETRY_MS` 后台重试；完成前 `/healthz` 与创建会话返回 `503 cleanup_pending`，避免旧残留与新会话并存。 +- `srs.conf` 的 HTTP 回调对整个 vhost 生效，网关会拒绝非专用 app 与未知流名的推流 / 播放，所以 SRS 需**独占一个实例或 vhost**，不要与其它直播业务共用。 +- 已经建立的 RTMP 连接不会因网关重启而重新鉴权；残留 Egress 停止后其推流断开，播放端随之断流。 + +### 7.2 无法自动回收时的手动恢复 + +LiveKit 长时间不可用、回收一直失败时，网关会持续返回 `cleanup_pending`，不会自行放行。恢复步骤： + +1. `docker compose stop mp-gateway`； +2. 通过 LiveKit API（`ListIngress`）删除名称前缀为 `yudao-mp-gw-` 的 Ingress； +3. 通过 `ListEgress`（active）停止输出地址以 `MP_GATEWAY_SRS_PUBLISH_BASE_URL/` 开头的 Egress； +4. `docker compose restart srs`，断开 SRS 上残留的推流与播放连接； +5. `docker compose start mp-gateway`，确认 `/healthz` 返回 `ok`。 +- 成本估算（LiveKit 官方 CPU 估值）：每个小程序会话 1 路 RTMP Ingress 转码约 2 核，每个远端 1 路 Track Composite Egress 约 2 核。1 对 1 小程序通话约 4 核，Egress 超出 CPU 余量会被 LiveKit 拒绝并如实返回 `failed`。 + +## 8. 已知限制与待验证点 + +- **延迟**：RTMP 推流 + Ingress 转码 + Egress 转码 + RTMP 拉流，秒级延迟，不适合对实时性要求高的场景。 +- **回声**：小程序 live-pusher 与 live-player 是两个独立组件，系统回声消除效果未验证，试验阶段建议戴耳机。 +- **静音状态**：小程序关闭摄像头或麦克风时，Ingress 推流仍在，PC / App 看到的是黑屏或静音画面，而不是 LiveKit 的 muted 状态。 +- **后台 / 断网**：小程序进入后台或 RTMP 网络断开（live-pusher 内部重连耗尽，状态码 -1307）时推流中断，Ingress 参与者离房，1 对 1 通话会由后端 webhook 结束；小程序端也会在推流丢失超过时限后结束本次通话。不支持无缝恢复，需要重新发起。 +- **屏幕共享转出**：本地真实媒体验证中出现过屏幕共享的 Track Composite Egress 处于 `EGRESS_ACTIVE` 但始终未推到 SRS 的情况（摄像头转出正常），根因待排查；网关只做启动超时与 `failed` 如实返回，不静默回退摄像头。 +- 本地真实媒体验证（FFmpeg 模拟推流 + H5，非微信原生组件）已观察到 Egress 地址中的 `?secret=` 原样传给 SRS 并通过回调校验。 +- **待真实运行验证**：纯音频 Track Composite 推 RTMP；live-player 拉带 `?token=` 的 RTMP 地址；Ingress 对 live-pusher RTC 模式码流的兼容性。 + +## 9. 真机要求 + +- 小程序主体为企业，类目在 live-pusher / live-player 支持范围内，并在小程序后台「开发 → 开发管理 → 接口设置」开通「实时播放音视频流」「实时录制音视频流」。 +- 网关 API 部署在 HTTPS 域名后，并加入小程序「request 合法域名」。 +- 手机可达的地址：Ingress RTMP（`MP_INGRESS_RTMP_PUBLIC_HOST:1935`）、SRS（`MP_SRS_PLAY_PUBLIC_HOST:1936`，需把 `srs.conf` 的 `listen` 改为对外地址）；PC / H5 / App 照常需要访问 LiveKit 信令与媒体端口（`MP_LIVEKIT_BIND_ADDRESS`、`MP_LIVEKIT_NODE_IP`）。 +- 开发者工具不支持 live-pusher 推流，必须真机调试。 diff --git a/script/livekit-poc/mp-gateway/docker-compose.yml b/script/livekit-poc/mp-gateway/docker-compose.yml new file mode 100644 index 0000000..20b65eb --- /dev/null +++ b/script/livekit-poc/mp-gateway/docker-compose.yml @@ -0,0 +1,147 @@ +# 小程序 RTMP ↔ LiveKit 网关试验环境（默认关闭，勿用于生产） +# 与上级目录的 LiveKit-only PoC 二选一：两者都占用 7880 等端口，启动本环境前先 docker compose -f ../docker-compose.yml down +# +# 全部服务使用 host 网络：Ingress / Egress 作为 WebRTC 客户端连接 LiveKit 广播的 node_ip（127.0.0.1）， +# 端口映射模式下容器内的 127.0.0.1 不是 LiveKit；macOS 需 Docker Desktop 4.34+ 并开启 Settings → Resources → Network → Enable host networking +# +# 凭据通过同目录 .env 提供（参考 .env.example），必须与后端 yudao.im.rtc.api-key / api-secret 一致 +name: yudao-livekit-mp-gateway + +services: + redis: + image: redis:7.4-alpine + restart: unless-stopped + network_mode: host + # LiveKit / Ingress / Egress 共享同一 Redis 交换任务；只监听回环、不持久化 + command: ["redis-server", "--bind", "127.0.0.1", "--port", "16379", "--save", "", "--appendonly", "no"] + + livekit: + image: livekit/livekit-server:v1.13.10 + restart: unless-stopped + network_mode: host + depends_on: + - redis + environment: + LIVEKIT_CONFIG: | + port: 7880 + bind_addresses: + - "${MP_LIVEKIT_BIND_ADDRESS:-127.0.0.1}" + rtc: + tcp_port: 7881 + udp_port: 7882 + use_external_ip: false + node_ip: ${MP_LIVEKIT_NODE_IP:-127.0.0.1} + redis: + address: 127.0.0.1:16379 + keys: + ${LIVEKIT_API_KEY:?需在 .env 配置 LIVEKIT_API_KEY}: ${LIVEKIT_API_SECRET:?需在 .env 配置 LIVEKIT_API_SECRET} + webhook: + api_key: ${LIVEKIT_API_KEY} + urls: + - ${MP_LIVEKIT_WEBHOOK_URL:-http://127.0.0.1:48080/admin-api/im/livekit/webhook} + ingress: + rtmp_base_url: rtmp://${MP_INGRESS_RTMP_PUBLIC_HOST:-127.0.0.1}:1935/x + room: + empty_timeout: 300 + departure_timeout: 20 + logging: + level: info + + ingress: + image: livekit/ingress:v1.5.0 + restart: unless-stopped + network_mode: host + depends_on: + - redis + - livekit + environment: + INGRESS_CONFIG_BODY: | + api_key: ${LIVEKIT_API_KEY} + api_secret: ${LIVEKIT_API_SECRET} + ws_url: ws://127.0.0.1:7880 + redis: + address: 127.0.0.1:16379 + rtmp_port: 1935 + whip_port: 18080 + http_relay_port: 19090 + health_port: 18081 + rtc_config: + udp_port: 7885 + use_external_ip: false + logging: + level: info + + egress: + image: livekit/egress:v1.15.0 + restart: unless-stopped + network_mode: host + # 官方运行方式要求；Track Composite 不启动 Chrome，但保持与官方镜像文档一致 + cap_add: + - SYS_ADMIN + depends_on: + - redis + - livekit + environment: + EGRESS_CONFIG_BODY: | + api_key: ${LIVEKIT_API_KEY} + api_secret: ${LIVEKIT_API_SECRET} + ws_url: ws://127.0.0.1:7880 + redis: + address: 127.0.0.1:16379 + health_port: 18082 + template_port: 17980 + log_level: info + + srs: + image: ossrs/srs:6 + restart: unless-stopped + network_mode: host + volumes: + - ./srs.conf:/usr/local/srs/conf/mp-gateway.conf:ro + command: ["./objs/srs", "-c", "conf/mp-gateway.conf"] + + mp-gateway: + image: node:22-alpine + # 未显式开启时网关打印说明后退出，不自动重启 + restart: "no" + network_mode: host + depends_on: + - livekit + - ingress + - egress + - srs + working_dir: /app + volumes: + - ./src:/app/src:ro + command: ["node", "src/server.mjs"] + environment: + MP_GATEWAY_ENABLED: ${MP_GATEWAY_ENABLED:-false} + MP_GATEWAY_API_HOST: ${MP_GATEWAY_API_HOST:-127.0.0.1} + MP_GATEWAY_API_PORT: 8090 + MP_GATEWAY_HOOK_HOST: 127.0.0.1 + MP_GATEWAY_HOOK_PORT: 8091 + MP_GATEWAY_LIVEKIT_HTTP_URL: http://127.0.0.1:7880 + MP_GATEWAY_LIVEKIT_API_KEY: ${LIVEKIT_API_KEY} + MP_GATEWAY_LIVEKIT_API_SECRET: ${LIVEKIT_API_SECRET} + # SRS 主机与端口固定为本 compose 内的实例，只允许通过 MP_GATEWAY_SRS_APP 调整专用 app 后缀 + MP_GATEWAY_SRS_PUBLISH_BASE_URL: rtmp://127.0.0.1:1936/${MP_GATEWAY_SRS_APP:-yudao-mp-gw} + MP_GATEWAY_SRS_PLAY_BASE_URL: rtmp://${MP_SRS_PLAY_PUBLIC_HOST:-127.0.0.1}:1936/${MP_GATEWAY_SRS_APP:-yudao-mp-gw} + # 以下可调参数未配置时为空串，由网关按 README 默认值处理 + MP_GATEWAY_POLL_INTERVAL_MS: ${MP_GATEWAY_POLL_INTERVAL_MS:-} + MP_GATEWAY_LEASE_TTL_MS: ${MP_GATEWAY_LEASE_TTL_MS:-} + MP_GATEWAY_MAX_SESSION_MS: ${MP_GATEWAY_MAX_SESSION_MS:-} + MP_GATEWAY_MAX_SESSIONS: ${MP_GATEWAY_MAX_SESSIONS:-} + MP_GATEWAY_MAX_REMOTES_PER_SESSION: ${MP_GATEWAY_MAX_REMOTES_PER_SESSION:-} + MP_GATEWAY_MAX_EGRESS: ${MP_GATEWAY_MAX_EGRESS:-} + MP_GATEWAY_EGRESS_RETRY_MS: ${MP_GATEWAY_EGRESS_RETRY_MS:-} + MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS: ${MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS:-} + MP_GATEWAY_RELEASE_RETRY_MS: ${MP_GATEWAY_RELEASE_RETRY_MS:-} + MP_GATEWAY_CLEANUP_RETRY_MS: ${MP_GATEWAY_CLEANUP_RETRY_MS:-} + MP_GATEWAY_REQUEST_TIMEOUT_MS: ${MP_GATEWAY_REQUEST_TIMEOUT_MS:-} + MP_GATEWAY_CLOCK_SKEW_SEC: ${MP_GATEWAY_CLOCK_SKEW_SEC:-} + MP_GATEWAY_EGRESS_WIDTH: ${MP_GATEWAY_EGRESS_WIDTH:-} + MP_GATEWAY_EGRESS_HEIGHT: ${MP_GATEWAY_EGRESS_HEIGHT:-} + MP_GATEWAY_EGRESS_FRAMERATE: ${MP_GATEWAY_EGRESS_FRAMERATE:-} + MP_GATEWAY_EGRESS_VIDEO_KBPS: ${MP_GATEWAY_EGRESS_VIDEO_KBPS:-} + MP_GATEWAY_EGRESS_AUDIO_KBPS: ${MP_GATEWAY_EGRESS_AUDIO_KBPS:-} + MP_GATEWAY_EGRESS_KEYFRAME_SEC: ${MP_GATEWAY_EGRESS_KEYFRAME_SEC:-} diff --git a/script/livekit-poc/mp-gateway/package.json b/script/livekit-poc/mp-gateway/package.json new file mode 100644 index 0000000..30f8080 --- /dev/null +++ b/script/livekit-poc/mp-gateway/package.json @@ -0,0 +1,13 @@ +{ + "name": "yudao-im-mp-rtc-gateway", + "version": "0.1.0", + "private": true, + "description": "IM 微信小程序 RTMP ↔ LiveKit 网关试验：默认关闭，秒级延迟，仅供验证", + "type": "module", + "engines": { + "node": ">=22" + }, + "scripts": { + "start": "node src/server.mjs" + } +} diff --git a/script/livekit-poc/mp-gateway/src/config.mjs b/script/livekit-poc/mp-gateway/src/config.mjs new file mode 100644 index 0000000..04d6223 --- /dev/null +++ b/script/livekit-poc/mp-gateway/src/config.mjs @@ -0,0 +1,122 @@ +// 小程序 RTMP 网关配置：全部来自环境变量；密钥只保存在内存中，不打印、不回显 + +/** + * 网关专用 SRS app 命名空间 + *

+ * LiveKit 返回的 EgressInfo 会把 RTMP 地址最后一段（流名与查询参数）脱敏，只剩 scheme / host / port / app 可信， + * 启动回收据此识别本网关的残留 Egress，因此该 app 只能给网关使用 + */ +export const GATEWAY_SRS_APP_PATTERN = /^yudao-mp-gw(?:-[\w-]+)?$/ + +/** + * 读取整数配置 + * + * @param {Record} env 环境变量 + * @param {string} name 变量名 + * @param {number} fallback 默认值 + * @param {number} min 最小值 + * @param {number} max 最大值 + * @returns {number} 配置值 + */ +function readInt(env, name, fallback, min, max) { + const raw = env[name] + if (raw === undefined || raw === '') { + return fallback + } + const value = Number(raw) + if (!Number.isInteger(value) || value < min || value > max) { + throw new Error(`${name} 需为 ${min}~${max} 的整数`) + } + return value +} + +/** + * 读取 URL 配置并校验协议 + * + * @param {Record} env 环境变量 + * @param {string} name 变量名 + * @param {string} fallback 默认值 + * @param {string[]} protocols 允许的协议，如 ['http:'] + * @returns {string} 去掉末尾斜杠的 URL + */ +function readUrl(env, name, fallback, protocols) { + const raw = env[name] || fallback + let url + try { + url = new URL(raw) + } catch { + throw new Error(`${name} 不是合法 URL`) + } + if (!protocols.includes(url.protocol) || url.search || url.hash || url.username || url.password) { + throw new Error(`${name} 仅支持 ${protocols.join('/')} 且不能带查询参数或账号`) + } + return raw.replace(/\/+$/, '') +} + +/** + * 读取必填字符串配置 + * + * @param {Record} env 环境变量 + * @param {string} name 变量名 + * @param {number} minLength 最小长度 + * @returns {string} 配置值 + */ +function readRequired(env, name, minLength) { + const value = env[name] + if (!value || value.length < minLength) { + throw new Error(`${name} 未配置或长度不足 ${minLength}`) + } + return value +} + +/** + * 加载网关配置；未显式开启时拒绝启动 + * + * @param {Record} env 环境变量 + * @returns {Readonly>} 只读配置 + */ +export function loadConfig(env = process.env) { + if (env.MP_GATEWAY_ENABLED !== 'true') { + throw new Error('MP_GATEWAY_ENABLED 未设置为 true：小程序网关为默认关闭的试验功能') + } + const srsPublishBaseUrl = readUrl(env, 'MP_GATEWAY_SRS_PUBLISH_BASE_URL', 'rtmp://127.0.0.1:1936/yudao-mp-gw', ['rtmp:']) + const srsPlayBaseUrl = readUrl(env, 'MP_GATEWAY_SRS_PLAY_BASE_URL', 'rtmp://127.0.0.1:1936/yudao-mp-gw', ['rtmp:']) + const srsApp = new URL(srsPublishBaseUrl).pathname.replace(/^\/+/, '') + if (!GATEWAY_SRS_APP_PATTERN.test(srsApp) || new URL(srsPlayBaseUrl).pathname.replace(/^\/+/, '') !== srsApp) { + throw new Error('MP_GATEWAY_SRS_PUBLISH_BASE_URL 与 MP_GATEWAY_SRS_PLAY_BASE_URL 需使用同一个网关专用 app（yudao-mp-gw 或 yudao-mp-gw-<后缀>）') + } + const pollIntervalMs = readInt(env, 'MP_GATEWAY_POLL_INTERVAL_MS', 2000, 500, 10000) + return Object.freeze({ + apiHost: env.MP_GATEWAY_API_HOST || '127.0.0.1', // 小程序 API 监听地址；公网部署放在 HTTPS 反向代理之后 + apiPort: readInt(env, 'MP_GATEWAY_API_PORT', 8090, 1, 65535), // 小程序 API 端口 + hookHost: env.MP_GATEWAY_HOOK_HOST || '127.0.0.1', // SRS 回调监听地址，只允许 SRS 访问 + hookPort: readInt(env, 'MP_GATEWAY_HOOK_PORT', 8091, 1, 65535), // SRS 回调端口 + livekitHttpUrl: readUrl(env, 'MP_GATEWAY_LIVEKIT_HTTP_URL', 'http://127.0.0.1:7880', ['http:', 'https:']), // LiveKit Server API 地址，只取自部署配置 + apiKey: readRequired(env, 'MP_GATEWAY_LIVEKIT_API_KEY', 1), // 与后端签发房间 Token 的 apiKey 一致 + apiSecret: readRequired(env, 'MP_GATEWAY_LIVEKIT_API_SECRET', 32), // 与后端签发房间 Token 的 apiSecret 一致 + srsPublishBaseUrl, // Egress 推流到 SRS 的内部地址 + srsPlayBaseUrl, // 小程序 live-player 拉流的外部地址 + srsApp, // SRS app 名，回调时校验 + pollIntervalMs, // 建议小程序轮询间隔 + syncIntervalMs: pollIntervalMs, // 网关同步房间成员的间隔 + leaseTtlMs: readInt(env, 'MP_GATEWAY_LEASE_TTL_MS', 20000, pollIntervalMs * 3, 120000), // 会话租约：超过该时间未轮询即释放 + maxSessionMs: readInt(env, 'MP_GATEWAY_MAX_SESSION_MS', 2 * 60 * 60 * 1000, 60000, 24 * 60 * 60 * 1000), // 单会话最长存活时间 + maxSessions: readInt(env, 'MP_GATEWAY_MAX_SESSIONS', 4, 1, 100), // 全局最大小程序会话数（每个会话 1 路 Ingress 转码） + maxRemotesPerSession: readInt(env, 'MP_GATEWAY_MAX_REMOTES_PER_SESSION', 4, 1, 16), // 单会话最多拉取的远端路数 + maxEgress: readInt(env, 'MP_GATEWAY_MAX_EGRESS', 8, 1, 200), // 全局最大 Egress 路数 + egressRetryMs: readInt(env, 'MP_GATEWAY_EGRESS_RETRY_MS', 10000, 1000, 120000), // 单路 Egress 失败后的冷却时间 + egressPublishTimeoutMs: readInt(env, 'MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS', 20000, 5000, 120000), // Egress 启动后等待 SRS on_publish 的最长时间 + cleanupRetryMs: readInt(env, 'MP_GATEWAY_CLEANUP_RETRY_MS', 5000, 1000, 60000), // 启动回收失败后的重试间隔 + releaseRetryMs: readInt(env, 'MP_GATEWAY_RELEASE_RETRY_MS', 5000, 1000, 60000), // Ingress / Egress 释放失败后的重试间隔；确认释放前继续占用配额 + requestTimeoutMs: readInt(env, 'MP_GATEWAY_REQUEST_TIMEOUT_MS', 10000, 1000, 60000), // 调用 LiveKit API 超时 + clockSkewSec: readInt(env, 'MP_GATEWAY_CLOCK_SKEW_SEC', 30, 0, 300), // 校验 nbf / exp 允许的时钟偏差 + egressEncoding: Object.freeze({ // live-player 兼容的 H264 Baseline + AAC 编码参数 + width: readInt(env, 'MP_GATEWAY_EGRESS_WIDTH', 640, 160, 1920), + height: readInt(env, 'MP_GATEWAY_EGRESS_HEIGHT', 360, 120, 1080), + framerate: readInt(env, 'MP_GATEWAY_EGRESS_FRAMERATE', 15, 5, 30), + videoBitrate: readInt(env, 'MP_GATEWAY_EGRESS_VIDEO_KBPS', 600, 100, 4000), + audioBitrate: readInt(env, 'MP_GATEWAY_EGRESS_AUDIO_KBPS', 64, 16, 256), + keyFrameInterval: readInt(env, 'MP_GATEWAY_EGRESS_KEYFRAME_SEC', 1, 1, 10), + }), + }) +} diff --git a/script/livekit-poc/mp-gateway/src/jwt.mjs b/script/livekit-poc/mp-gateway/src/jwt.mjs new file mode 100644 index 0000000..af5eff6 --- /dev/null +++ b/script/livekit-poc/mp-gateway/src/jwt.mjs @@ -0,0 +1,128 @@ +// LiveKit 房间 Token 校验与网关管理 Token 签发：仅支持 HS256，与后端 LiveKitClient.signJoinToken 保持一致 +import { createHmac, timingSafeEqual } from 'node:crypto' + +const MAX_TOKEN_LENGTH = 4096 // 拒绝超长 Token，避免无意义的解析开销 +const USER_IDENTITY_PATTERN = /^[1-9]\d{0,18}$/ // 业务 identity：数字 userId，与后端 buildIdentity 一致 +const MAX_ROOM_LENGTH = 128 // 房间名长度上限 + +/** 房间 Token 校验失败；status 为返回给客户端的 HTTP 状态码 */ +export class TokenError extends Error { + /** + * @param {string} code 错误码 + * @param {string} message 错误说明（不包含 Token 内容） + * @param {number} status HTTP 状态码 + */ + constructor(code, message, status = 401) { + super(message) + this.code = code + this.status = status + } +} + +/** + * 解码 JWT 的 JSON 段 + * + * @param {string} segment base64url 段 + * @returns {Record} JSON 对象 + */ +function decodeSegment(segment) { + try { + const value = JSON.parse(Buffer.from(segment, 'base64url').toString('utf8')) + if (value && typeof value === 'object' && !Array.isArray(value)) { + return value + } + } catch { + // 统一按格式错误处理 + } + throw new TokenError('invalid_token', 'Token 格式错误') +} + +/** + * 计算 HS256 签名 + * + * @param {string} signingInput header.payload + * @param {string} secret 签名密钥 + * @returns {Buffer} 签名字节 + */ +function sign(signingInput, secret) { + return createHmac('sha256', secret).update(signingInput).digest() +} + +/** + * 校验后端签发的 LiveKit 房间 Token，并以 Token 内的房间与身份为准 + * + * @param {string | undefined} token 房间 Token + * @param {{ apiKey: string, apiSecret: string, clockSkewSec: number }} config 网关配置 + * @param {number} nowSec 当前时间（秒），便于测试 + * @returns {{ room: string, identity: string, name: string, exp: number }} 授权信息 + */ +export function verifyRoomToken(token, config, nowSec = Math.floor(Date.now() / 1000)) { + if (typeof token !== 'string' || !token || token.length > MAX_TOKEN_LENGTH) { + throw new TokenError('missing_token', '缺少房间 Token') + } + const parts = token.split('.') + if (parts.length !== 3 || parts.some(part => !/^[\w-]+$/.test(part))) { + throw new TokenError('invalid_token', 'Token 格式错误') + } + const header = decodeSegment(parts[0]) + if (header.alg !== 'HS256' || (header.typ !== undefined && header.typ !== 'JWT')) { + throw new TokenError('invalid_token', '仅支持 HS256 Token') + } + const expected = sign(`${parts[0]}.${parts[1]}`, config.apiSecret) + const actual = Buffer.from(parts[2], 'base64url') + if (actual.length !== expected.length || !timingSafeEqual(actual, expected)) { + throw new TokenError('invalid_signature', 'Token 签名无效') + } + const payload = decodeSegment(parts[1]) + if (payload.iss !== config.apiKey) { + throw new TokenError('invalid_issuer', 'Token 签发方不匹配') + } + if (typeof payload.exp !== 'number' || payload.exp + config.clockSkewSec <= nowSec) { + throw new TokenError('token_expired', 'Token 已过期') + } + if (typeof payload.nbf !== 'number' || payload.nbf - config.clockSkewSec > nowSec) { + throw new TokenError('token_not_active', 'Token 尚未生效') + } + if (typeof payload.sub !== 'string' || !USER_IDENTITY_PATTERN.test(payload.sub)) { + throw new TokenError('invalid_identity', 'Token 身份不是业务用户', 403) + } + const video = payload.video + if (!video || typeof video !== 'object' + || video.roomJoin !== true + || typeof video.room !== 'string' || !video.room || video.room.length > MAX_ROOM_LENGTH + || video.canPublish !== true || video.canSubscribe !== true + || video.hidden === true || video.recorder === true) { + throw new TokenError('insufficient_grant', 'Token 缺少进房、发布或订阅授权', 403) + } + if (Array.isArray(video.canPublishSources) + && !(video.canPublishSources.includes('camera') && video.canPublishSources.includes('microphone'))) { + throw new TokenError('insufficient_grant', 'Token 不允许发布摄像头和麦克风', 403) + } + return { + room: video.room, + identity: payload.sub, + name: typeof payload.name === 'string' && payload.name ? payload.name.slice(0, 128) : payload.sub, + exp: payload.exp, + } +} + +/** + * 签发调用 LiveKit Server API 的短期管理 Token + * + * @param {{ apiKey: string, apiSecret: string }} config 网关配置 + * @param {Record} video 授权，如 { roomAdmin: true, room } + * @param {number} ttlSec 有效期（秒） + * @returns {string} JWT + */ +export function signAdminToken(config, video, ttlSec = 300) { + const nowSec = Math.floor(Date.now() / 1000) + const header = Buffer.from(JSON.stringify({ alg: 'HS256', typ: 'JWT' })).toString('base64url') + const payload = Buffer.from(JSON.stringify({ + iss: config.apiKey, + sub: 'yudao-mp-gateway', + nbf: nowSec - 5, + exp: nowSec + ttlSec, + video, + })).toString('base64url') + return `${header}.${payload}.${sign(`${header}.${payload}`, config.apiSecret).toString('base64url')}` +} diff --git a/script/livekit-poc/mp-gateway/src/livekit-api.mjs b/script/livekit-poc/mp-gateway/src/livekit-api.mjs new file mode 100644 index 0000000..4e76acd --- /dev/null +++ b/script/livekit-poc/mp-gateway/src/livekit-api.mjs @@ -0,0 +1,265 @@ +// LiveKit Server API（Twirp JSON）最小客户端：只调用网关需要的 Room / Ingress / Egress 接口 +import { signAdminToken } from './jwt.mjs' + +const PARTICIPANT_KINDS = ['STANDARD', 'INGRESS', 'EGRESS', 'SIP', 'AGENT'] // ParticipantInfo.Kind 枚举 +const PARTICIPANT_STATES = ['JOINING', 'JOINED', 'ACTIVE', 'DISCONNECTED'] // ParticipantInfo.State 枚举 +const TRACK_TYPES = ['AUDIO', 'VIDEO', 'DATA'] // TrackType 枚举 +const TRACK_SOURCES = ['UNKNOWN', 'CAMERA', 'MICROPHONE', 'SCREEN_SHARE', 'SCREEN_SHARE_AUDIO'] // TrackSource 枚举 +const TWIRP_CODE_PATTERN = /^[a-z_]{1,32}$/ // Twirp 错误码格式，如 not_found；不符合时按 unknown 处理 + +/** LiveKit API 调用失败；message 只含方法、HTTP 状态与错误码，不含上游原文、Token 或推流地址 */ +export class LiveKitApiError extends Error { + /** + * @param {number} status HTTP 状态码 + * @param {string} code Twirp 错误码 + * @param {string} message 错误说明 + */ + constructor(status, code, message) { + super(message) + this.status = status + this.code = code + } +} + +/** + * 读取 protobuf JSON 字段；兼容 snake_case 与 lowerCamelCase 输出 + * + * @param {Record} value 对象 + * @param {string} snakeName snake_case 字段名 + * @returns {any} 字段值 + */ +function field(value, snakeName) { + if (value[snakeName] !== undefined) { + return value[snakeName] + } + return value[snakeName.replace(/_([a-z])/g, (_, char) => char.toUpperCase())] +} + +/** + * 读取枚举名；兼容名称与数字输出，缺省值按 protobuf 语义取第 0 项 + * + * @param {string | number | undefined} value 枚举值 + * @param {string[]} names 枚举名称表 + * @returns {string} 枚举名 + */ +function enumName(value, names) { + if (typeof value === 'number') { + return names[value] ?? String(value) + } + return value || names[0] +} + +/** + * 标准化房间参与者 + * + * @param {Record} participant ParticipantInfo + * @returns {{ sid: string, identity: string, kind: string, state: string, joinedAt: number, tracks: Array<{ sid: string, type: string, source: string, muted: boolean }> }} 参与者 + */ +function normalizeParticipant(participant) { + return { + sid: participant.sid, + identity: participant.identity || '', + kind: enumName(participant.kind, PARTICIPANT_KINDS), + state: enumName(participant.state, PARTICIPANT_STATES), + joinedAt: Number(field(participant, 'joined_at') || 0), + tracks: (participant.tracks || []).map(track => ({ + sid: track.sid, + type: enumName(track.type, TRACK_TYPES), + source: enumName(track.source, TRACK_SOURCES), + muted: track.muted === true, + })), + } +} + +/** + * 创建 LiveKit API 客户端；服务地址只来自部署配置，不接受客户端传入 + * + * @param {{ livekitHttpUrl: string, apiKey: string, apiSecret: string, requestTimeoutMs: number }} config 网关配置 + * @returns {Record} API 方法 + */ +export function createLiveKitApi(config) { + /** + * 调用 Twirp 接口 + * + * @param {string} service 服务名，如 RoomService + * @param {string} method 方法名 + * @param {Record} body 请求体 + * @param {Record} video 管理 Token 授权 + * @returns {Promise>} 响应体 + */ + async function call(service, method, body, video) { + let response + let text + try { + response = await fetch(`${config.livekitHttpUrl}/twirp/livekit.${service}/${method}`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Authorization': `Bearer ${signAdminToken(config, video)}`, + }, + body: JSON.stringify(body), + signal: AbortSignal.timeout(config.requestTimeoutMs), + }) + text = await response.text() + } catch (error) { + // 网络错误 / 超时只保留类型，不透传底层信息 + const code = error?.name === 'TimeoutError' ? 'timeout' : 'network_error' + throw new LiveKitApiError(0, code, `${service}.${method} 请求失败：${code}`) + } + let data = {} + try { + data = text ? JSON.parse(text) : {} + } catch { + data = {} + } + if (!data || typeof data !== 'object' || Array.isArray(data)) { // null、数组、数字等非对象 JSON 视为空响应，交由状态码与资源编号校验处理 + data = {} + } + if (!response.ok) { + // 上游 msg 可能回显请求中的 RTMP 地址等凭证，只保留方法、HTTP 状态与规范化的 Twirp 错误码 + const code = typeof data.code === 'string' && TWIRP_CODE_PATTERN.test(data.code) ? data.code : 'unknown' + throw new LiveKitApiError(response.status, code, `${service}.${method} 失败：HTTP ${response.status} ${code}`) + } + return data + } + + /** + * 判断是否为「资源不存在 / 已结束」类错误，释放资源时可忽略 + * + * @param {unknown} error 错误 + * @returns {boolean} 是否可忽略 + */ + function isGone(error) { + return error instanceof LiveKitApiError + && (error.status === 404 || error.code === 'not_found' || error.code === 'failed_precondition') + } + + return { + /** + * 查询房间参与者 + * + * @param {string} room 房间名 + * @returns {Promise[] | null>} 参与者；房间不存在返回 null + */ + async listParticipants(room) { + try { + const data = await call('RoomService', 'ListParticipants', { room }, { roomAdmin: true, room }) + return (data.participants || []).map(normalizeParticipant) + } catch (error) { + if (error instanceof LiveKitApiError && (error.status === 404 || error.code === 'not_found')) { + return null + } + throw error + } + }, + + /** + * 创建 RTMP Ingress；参与者身份直接使用 Token 内的 userId，保留同 identity 互踢语义 + * + * @param {{ room: string, identity: string, name: string, resourceName: string }} options 参数 + * @returns {Promise<{ ingressId: string, url: string, streamKey: string }>} Ingress 信息 + */ + async createIngress(options) { + const data = await call('Ingress', 'CreateIngress', { + input_type: 'RTMP_INPUT', + name: options.resourceName, + room_name: options.room, + participant_identity: options.identity, + participant_name: options.name, + enable_transcoding: true, + }, { ingressAdmin: true }) + const ingressId = field(data, 'ingress_id') + if (typeof ingressId !== 'string' || !ingressId) { // 可能已创建但拿不到编号，交由调用方按结果未知处理 + throw new LiveKitApiError(502, 'invalid_response', 'Ingress.CreateIngress 响应缺少资源编号') + } + return { + ingressId, + url: data.url || '', + streamKey: field(data, 'stream_key') || '', + } + }, + + /** + * 删除 Ingress；不存在时忽略 + * + * @param {string} ingressId Ingress 编号 + * @returns {Promise} + */ + async deleteIngress(ingressId) { + try { + await call('Ingress', 'DeleteIngress', { ingress_id: ingressId }, { ingressAdmin: true }) + } catch (error) { + if (!isGone(error)) { + throw error + } + } + }, + + /** + * 列出全部 Ingress，用于启动时回收本网关残留资源 + * + * @returns {Promise>} Ingress 列表 + */ + async listIngress() { + const data = await call('Ingress', 'ListIngress', {}, { ingressAdmin: true }) + return (data.items || []).map(item => ({ ingressId: field(item, 'ingress_id'), name: item.name || '' })) + }, + + /** + * 启动单个远端参与者的 Track Composite Egress，推 RTMP 到 SRS + * + * @param {{ room: string, audioTrackId?: string, videoTrackId?: string, url: string, encoding: Record }} options 参数 + * @returns {Promise<{ egressId: string }>} Egress 信息 + */ + async startTrackCompositeEgress(options) { + const data = await call('Egress', 'StartTrackCompositeEgress', { + room_name: options.room, + audio_track_id: options.audioTrackId || '', + video_track_id: options.videoTrackId || '', + stream_outputs: [{ protocol: 'RTMP', urls: [options.url] }], + advanced: { + width: options.encoding.width, + height: options.encoding.height, + framerate: options.encoding.framerate, + audio_codec: 'AAC', + audio_bitrate: options.encoding.audioBitrate, + video_codec: 'H264_BASELINE', + video_bitrate: options.encoding.videoBitrate, + key_frame_interval: options.encoding.keyFrameInterval, + }, + }, { roomRecord: true }) + const egressId = field(data, 'egress_id') + if (typeof egressId !== 'string' || !egressId) { // 可能已创建但拿不到编号，交由调用方按结果未知处理 + throw new LiveKitApiError(502, 'invalid_response', 'Egress.StartTrackCompositeEgress 响应缺少资源编号') + } + return { egressId } + }, + + /** + * 停止 Egress；已结束或不存在时忽略 + * + * @param {string} egressId Egress 编号 + * @returns {Promise} + */ + async stopEgress(egressId) { + try { + await call('Egress', 'StopEgress', { egress_id: egressId }, { roomRecord: true }) + } catch (error) { + if (!isGone(error)) { + throw error + } + } + }, + + /** + * 列出活跃 Egress + * + * @param {string | undefined} room 房间名；为空时列出全部 + * @returns {Promise }>>} 活跃 Egress + */ + async listActiveEgress(room) { + const data = await call('Egress', 'ListEgress', room ? { room_name: room, active: true } : { active: true }, { roomRecord: true }) + return (data.items || []).map(item => ({ egressId: field(item, 'egress_id'), raw: item })) + }, + } +} diff --git a/script/livekit-poc/mp-gateway/src/server.mjs b/script/livekit-poc/mp-gateway/src/server.mjs new file mode 100644 index 0000000..5f2d078 --- /dev/null +++ b/script/livekit-poc/mp-gateway/src/server.mjs @@ -0,0 +1,297 @@ +// 小程序 RTMP 网关入口：对小程序提供会话 API，对 SRS 提供推拉流鉴权回调（两个端口分开监听） +import { createServer } from 'node:http' +import { loadConfig } from './config.mjs' +import { TokenError, verifyRoomToken } from './jwt.mjs' +import { createLiveKitApi } from './livekit-api.mjs' +import { GatewayError, createSessionManager } from './session-manager.mjs' + +const MAX_BODY_BYTES = 4096 // 请求体上限；会话 API 与 SRS 回调都只有很小的 JSON +const SHUTDOWN_TIMEOUT_MS = 15000 // 退出时等待资源释放的最长时间 +const SESSION_PATH_PATTERN = /^\/mp-rtc\/sessions\/([^/]+)$/ // 单个会话路径 + +let orphansCleaned = false // 启动回收是否已完成；完成前拒绝创建会话，避免旧残留与新会话并存 + +/** 日志：只输出事件与编号，调用方不得传入 Token、stream key、带凭证的 URL */ +const logger = { + /** + * @param {string} message 事件 + * @param {Record} [fields] 附加字段 + */ + info(message, fields = {}) { + console.log(JSON.stringify({ time: new Date().toISOString(), level: 'info', message, ...fields })) + }, + /** + * @param {string} message 事件 + * @param {Record} [fields] 附加字段 + */ + warn(message, fields = {}) { + console.warn(JSON.stringify({ time: new Date().toISOString(), level: 'warn', message, ...fields })) + }, +} + +/** + * 写 JSON 响应 + * + * @param {import('node:http').ServerResponse} res 响应 + * @param {number} status 状态码 + * @param {unknown} [body] 响应体；为空时不写 body + */ +function sendJson(res, status, body) { + if (res.headersSent) { + return + } + const headers = { 'Cache-Control': 'no-store' } + if (body === undefined) { + res.writeHead(status, headers).end() + return + } + res.writeHead(status, { ...headers, 'Content-Type': 'application/json; charset=utf-8' }).end(JSON.stringify(body)) +} + +/** + * 读取 JSON 请求体 + * + * @param {import('node:http').IncomingMessage} req 请求 + * @returns {Promise>} 请求体；空 body 返回空对象 + */ +async function readJson(req) { + const chunks = [] + let size = 0 + for await (const chunk of req) { + size += chunk.length + if (size > MAX_BODY_BYTES) { + throw new GatewayError(413, 'body_too_large', '请求体过大') + } + chunks.push(chunk) + } + if (size === 0) { + return {} + } + try { + const value = JSON.parse(Buffer.concat(chunks).toString('utf8')) + return value && typeof value === 'object' ? value : {} + } catch { + throw new GatewayError(400, 'invalid_json', '请求体不是合法 JSON') + } +} + +/** + * 读取 Bearer 凭证 + * + * @param {import('node:http').IncomingMessage} req 请求 + * @returns {string | undefined} 凭证 + */ +function readBearer(req) { + const header = req.headers.authorization + return typeof header === 'string' && header.startsWith('Bearer ') ? header.slice(7).trim() : undefined +} + +/** + * 错误转响应；未知错误不回显细节 + * + * @param {import('node:http').ServerResponse} res 响应 + * @param {unknown} error 错误 + */ +function sendError(res, error) { + if (error instanceof GatewayError || error instanceof TokenError) { + sendJson(res, error.status, { code: error.code, message: error.message }) + return + } + logger.warn('请求处理失败', { error: error instanceof Error ? error.message : String(error) }) + sendJson(res, 500, { code: 'internal_error', message: '网关内部错误' }) +} + +/** + * 处理小程序会话 API + * + * @param {Readonly>} config 网关配置 + * @param {ReturnType} manager 会话管理器 + * @param {import('node:http').IncomingMessage} req 请求 + * @param {import('node:http').ServerResponse} res 响应 + * @returns {Promise} + */ +async function handleApi(config, manager, req, res) { + const { pathname } = new URL(req.url || '/', 'http://gateway') + if (req.method === 'GET' && pathname === '/healthz') { + sendJson(res, orphansCleaned ? 200 : 503, { status: orphansCleaned ? 'ok' : 'cleanup_pending' }) + return + } + if (req.method === 'POST' && pathname === '/mp-rtc/sessions') { + if (!orphansCleaned) { + throw new GatewayError(503, 'cleanup_pending', '网关正在回收上次残留的推流资源，请稍后再试') + } + await readJson(req) + const claims = verifyRoomToken(readBearer(req), config) + const { session, sessionToken } = await manager.createSession(claims) + // 小程序在创建过程中断开（页面关闭、超时）：响应无人接收，立即释放刚创建的资源 + if (res.destroyed || !res.socket || res.socket.destroyed) { + await manager.closeSession(session, 'client_aborted') + return + } + sendJson(res, 201, { + sessionId: session.id, + sessionToken, + userId: Number(session.identity), + pushUrl: session.pushUrl, + pollIntervalMs: config.pollIntervalMs, + leaseTtlMs: config.leaseTtlMs, + expiresAt: session.expiresAt, + }) + return + } + const matched = SESSION_PATH_PATTERN.exec(pathname) + if (matched && (req.method === 'GET' || req.method === 'DELETE')) { + const session = manager.authorize(matched[1], readBearer(req)) + if (req.method === 'GET') { + sendJson(res, 200, manager.touch(session)) + return + } + await manager.closeSession(session, 'client_closed') + sendJson(res, 204) + return + } + sendJson(res, 404, { code: 'not_found', message: '接口不存在' }) +} + +/** + * 处理 SRS HTTP 回调；返回 code 0 允许，其它拒绝 + * + * @param {ReturnType} manager 会话管理器 + * @param {import('node:http').IncomingMessage} req 请求 + * @param {import('node:http').ServerResponse} res 响应 + * @returns {Promise} + */ +async function handleHook(manager, req, res) { + const { pathname } = new URL(req.url || '/', 'http://gateway') + if (req.method !== 'POST') { + sendJson(res, 404, { code: 404 }) + return + } + const hook = await readJson(req) + let allowed = false + if (pathname === '/srs/hooks/on_publish') { + allowed = manager.authorizePublish(hook) + } else if (pathname === '/srs/hooks/on_play') { + allowed = manager.authorizePlay(hook) + } else if (pathname === '/srs/hooks/on_unpublish') { + manager.handleUnpublish(hook) + allowed = true + } else { + sendJson(res, 404, { code: 404 }) + return + } + if (!allowed) { + logger.warn('SRS 鉴权拒绝', { action: pathname.slice('/srs/hooks/'.length) }) + } + sendJson(res, allowed ? 200 : 403, { code: allowed ? 0 : 403 }) +} + +/** + * 启动 HTTP 服务 + * + * @param {import('node:http').Server} server 服务 + * @param {string} host 监听地址 + * @param {number} port 端口 + * @returns {Promise} + */ +function listen(server, host, port) { + return new Promise((resolve, reject) => { + server.once('error', reject) + server.listen(port, host, () => { + server.off('error', reject) + resolve() + }) + }) +} + +/** 启动网关 */ +async function main() { + const config = loadConfig() + const liveKit = createLiveKitApi(config) + const manager = createSessionManager(config, liveKit, logger) + let stopping = false // 是否已开始退出 + let cleanupTimer // 启动回收重试定时器 + let cleanupTask // 正在执行的启动回收；退出时等待其结束 + + /** 发起一次启动回收；退出后不再发起 */ + function scheduleCleanup() { + cleanupTimer = undefined + if (stopping) { + return + } + cleanupTask = runCleanup().finally(() => { + cleanupTask = undefined + }) + } + + /** 启动回收：失败后按间隔重试，直到成功或网关退出 */ + async function runCleanup() { + try { + await manager.cleanupOrphans() + orphansCleaned = true + } catch (error) { + logger.warn(stopping ? '残留资源回收失败，网关退出中不再重试' : '残留资源回收失败，稍后重试；完成前拒绝创建会话', + { error: error.message }) + if (!stopping) { + cleanupTimer = setTimeout(scheduleCleanup, config.cleanupRetryMs) + } + } + } + + const apiServer = createServer((req, res) => { + handleApi(config, manager, req, res).catch(error => sendError(res, error)) + }) + const hookServer = createServer((req, res) => { + handleHook(manager, req, res).catch(error => sendError(res, error)) + }) + apiServer.requestTimeout = config.requestTimeoutMs * 3 + hookServer.requestTimeout = config.requestTimeoutMs + await listen(apiServer, config.apiHost, config.apiPort) + await listen(hookServer, config.hookHost, config.hookPort) + scheduleCleanup() + logger.info('小程序网关已启动（试验功能）', { + api: `${config.apiHost}:${config.apiPort}`, + hook: `${config.hookHost}:${config.hookPort}`, + livekit: config.livekitHttpUrl, + maxSessions: config.maxSessions, + maxEgress: config.maxEgress, + }) + + /** + * 退出：停止接收请求并释放全部 Ingress / Egress + * + * @param {string} signal 信号名 + */ + const stop = (signal) => { + if (stopping) { + return + } + stopping = true + clearTimeout(cleanupTimer) + logger.info('网关退出中', { signal }) + apiServer.close() + const timer = setTimeout(() => process.exit(1), SHUTDOWN_TIMEOUT_MS) + // 先等待进行中的启动回收（runCleanup 自身不会拒绝），再释放会话资源；启动回收未完成时残留资源未确认，不能报告正常退出 + Promise.resolve(cleanupTask).then(() => manager.shutdown()).then(() => { + if (!orphansCleaned) { + logger.warn('网关退出时启动残留资源回收未完成') + return 1 + } + return 0 + }, (error) => { + logger.warn('网关退出时仍有资源未确认释放', { error: error.message }) + return 1 + }).then((code) => { + clearTimeout(timer) + hookServer.close() + process.exit(code) + }) + } + process.on('SIGINT', () => stop('SIGINT')) + process.on('SIGTERM', () => stop('SIGTERM')) +} + +main().catch((error) => { + logger.warn('网关启动失败', { error: error.message }) + process.exit(1) +}) diff --git a/script/livekit-poc/mp-gateway/src/session-manager.mjs b/script/livekit-poc/mp-gateway/src/session-manager.mjs new file mode 100644 index 0000000..cb23ea5 --- /dev/null +++ b/script/livekit-poc/mp-gateway/src/session-manager.mjs @@ -0,0 +1,837 @@ +// 小程序网关会话：1 路 Ingress 接收小程序推流 + 每个远端参与者 1 路 Track Composite Egress 推到 SRS +import { createHash, randomBytes, timingSafeEqual } from 'node:crypto' +import { setTimeout as delay } from 'node:timers/promises' +import { GATEWAY_SRS_APP_PATTERN } from './config.mjs' +import { LiveKitApiError } from './livekit-api.mjs' + +const RESOURCE_PREFIX = 'yudao-mp-gw-' // 网关创建的 Ingress 名称前缀；单实例启动时据此回收残留 +const USER_IDENTITY_PATTERN = /^[1-9]\d{0,18}$/ // 业务参与者 identity：数字 userId +const EXCLUDED_KINDS = new Set(['EGRESS', 'AGENT', 'SIP']) // 非业务参与者类型；其它小程序成员的 INGRESS 属于业务参与者 +const EGRESS_LIST_GRACE_MS = 10000 // 新建 Egress 在活跃列表中可见前的宽限期 +const SESSION_ID_PATTERN = /^[0-9a-f]{32}$/ // 会话编号格式 +const STREAM_PREFIX = 'yudaompgw_' // 网关推往 SRS 的流名前缀 +const STREAM_NAME_PATTERN = /^yudaompgw_[0-9a-f]{12}_[1-9]\d{0,18}_[0-9a-f]{16}$/ // 网关流名完整格式 +const RELEASE_CONCURRENCY = 4 // 定期重试时同时进行的释放尝试上限 +const SHUTDOWN_RELEASE_RETRY_MS = 500 // 退出时释放失败后的重试间隔；整体时长由入口的退出超时兜底 +const REDACTED_STREAM_PATTERN = /^\{yud\.\.\.[\w-]{3}\}$/ // LiveKit 脱敏后的网关流名：首 3 位 + ... + 末 3 位（末位来自推流密钥） + +/** 网关业务错误；status 为返回给客户端的 HTTP 状态码 */ +export class GatewayError extends Error { + /** + * @param {number} status HTTP 状态码 + * @param {string} code 错误码 + * @param {string} message 错误说明 + */ + constructor(status, code, message) { + super(message) + this.status = status + this.code = code + } +} + +/** + * 计算 SHA-256 摘要；会话凭证、推拉流密钥只保存摘要用于比对 + * + * @param {string} value 原文 + * @returns {Buffer} 摘要 + */ +function digest(value) { + return createHash('sha256').update(value).digest() +} + +/** + * 常量时间比较凭证 + * + * @param {unknown} value 待校验原文 + * @param {Buffer} expectedDigest 期望摘要 + * @returns {boolean} 是否一致 + */ +function matches(value, expectedDigest) { + return typeof value === 'string' && value.length > 0 && timingSafeEqual(digest(value), expectedDigest) +} + +/** + * 是否为业务参与者：数字 userId、非 Egress 等内部参与者、未断开 + * + * @param {{ identity: string, kind: string, state: string }} participant 参与者 + * @returns {boolean} 是否业务参与者 + */ +function isBusinessParticipant(participant) { + return USER_IDENTITY_PATTERN.test(participant.identity) + && !EXCLUDED_KINDS.has(participant.kind) + && participant.state !== 'DISCONNECTED' +} + +/** + * 选择远端参与者要转出的媒体：视频屏幕共享优先、摄像头回退；音频取麦克风 + *

+ * 摄像头静音不更换 Egress，只回传 videoVisible，避免反复重建推流 + * + * @param {{ tracks: Array<{ sid: string, type: string, source: string, muted: boolean }> }} participant 参与者 + * @returns {{ audioSid?: string, videoSid?: string, videoSource?: string, videoVisible: boolean }} 媒体选择 + */ +function pickMedia(participant) { + const screenShare = participant.tracks.find(track => track.type === 'VIDEO' && track.source === 'SCREEN_SHARE' && !track.muted) + const camera = participant.tracks.find(track => track.type === 'VIDEO' && (track.source === 'CAMERA' || track.source === 'UNKNOWN')) + const video = screenShare || camera + const audio = participant.tracks.find(track => track.type === 'AUDIO' && (track.source === 'MICROPHONE' || track.source === 'UNKNOWN')) + return { + audioSid: audio?.sid, + videoSid: video?.sid, + videoSource: video?.source, + videoVisible: !!video && !video.muted, + } +} + +/** + * 创建调用的结果是否未知：超时、网络错误、HTTP 5xx（含代理 502 / 504 与响应缺少资源编号）时， + * 请求可能已被 LiveKit 受理、资源已建但编号丢失 + *

+ * 4xx 等明确拒绝的失败按未创建处理，沿用冷却重试 + * + * @param {unknown} error 错误 + * @returns {boolean} 是否结果未知 + */ +function isAmbiguousCreateError(error) { + return error instanceof LiveKitApiError && (error.status === 0 || error.status >= 500) +} + +/** + * 解析 SRS 回调中的查询参数 + * + * @param {unknown} param 如 "?token=xxx" + * @returns {URLSearchParams} 参数 + */ +function parseHookParam(param) { + return new URLSearchParams(typeof param === 'string' ? param.replace(/^\?/, '') : '') +} + +/** + * 创建会话管理器（单实例） + * + * @param {Readonly>} config 网关配置 + * @param {ReturnType} liveKit LiveKit API + * @param {{ info: Function, warn: Function }} logger 日志；只记录编号与房间，不记录凭证 + * @returns {Record} 会话管理方法 + */ +export function createSessionManager(config, liveKit, logger) { + const sessions = new Map() // 会话编号 -> 会话 + const sessionsByUser = new Map() // 房间 + identity -> 会话；同一用户同一房间只允许一份会话 + const streams = new Map() // SRS 流名 -> { session, entry }；会话关闭或远端移除时立即删除，撤销推拉流授权 + const pendingReleases = new Map() // 「类型:资源编号」-> 待确认释放的 Ingress / Egress；释放成功前持续占用配额并定期重试 + const inflightCreates = new Set() // 进行中的 Ingress / Egress 创建；结果返回前占用配额，退出时等待其完成 + const unknownCreates = new Set() // 结果未知的创建（超时 / 网络错误）；资源可能已在 LiveKit 侧存在，进程内永久占用配额与用户维度，只能重启后由启动回收清理 + let shuttingDown = false // 是否正在退出 + const reaper = setInterval(reapSessions, 1000) // 租约、Token 过期扫描与释放重试 + + /** + * 会话的用户维度键 + * + * @param {{ room: string, identity: string }} value 会话或授权 + * @returns {string} 键 + */ + function userKey(value) { + return `${value.room}\n${value.identity}` + } + + /** + * 当前占用的 Ingress 数：存活会话 + 待确认释放 + 归属已失效但仍在创建中 + 结果未知的 Ingress + * + * @returns {number} 数量 + */ + function countIngress() { + let count = sessions.size + pendingReleases.forEach((item) => { + if (item.kind === 'ingress') { + count++ + } + }) + inflightCreates.forEach((item) => { + if (item.kind === 'ingress' && !item.isCurrent()) { + count++ + } + }) + unknownCreates.forEach((item) => { + if (item.kind === 'ingress') { + count++ + } + }) + return count + } + + /** + * 当前占用的 Egress 路数：存活会话中的远端流（含启动中）+ 待确认释放 + 归属已失效但仍在创建中 + 结果未知的 Egress + * + * @returns {number} 路数 + */ + function countEgress() { + let count = 0 + sessions.forEach((session) => { + count += session.remotes.size + }) + pendingReleases.forEach((item) => { + if (item.kind === 'egress') { + count++ + } + }) + inflightCreates.forEach((item) => { + if (item.kind === 'egress' && !item.isCurrent()) { + count++ + } + }) + unknownCreates.forEach((item) => { + if (item.kind === 'egress') { + count++ + } + }) + return count + } + + /** + * 同一房间 + identity 是否仍有未释放的会话或推流资源（含释放中、创建中、结果未知） + * + * @param {string} key 用户维度键 + * @returns {boolean} 是否占用 + */ + function isUserBusy(key) { + if (sessionsByUser.has(key)) { + return true + } + for (const item of pendingReleases.values()) { + if (item.ownerKey === key) { + return true + } + } + for (const item of inflightCreates) { + if (item.ownerKey === key) { + return true + } + } + for (const item of unknownCreates) { + if (item.ownerKey === key) { + return true + } + } + return false + } + + /** + * 尝试释放一个待释放资源；同一资源同时只有一次尝试，成功后移出待释放列表，失败则等待下次重试 + * + * @param {{ kind: 'ingress' | 'egress', id: string, attempt?: Promise, nextAt: number }} item 待释放资源 + * @returns {Promise} 是否已确认释放 + */ + function attemptRelease(item) { + if (item.attempt) { + return item.attempt + } + const key = `${item.kind}:${item.id}` + const attempt = (item.kind === 'ingress' ? liveKit.deleteIngress(item.id) : liveKit.stopEgress(item.id)) + .then(() => { + if (pendingReleases.get(key) === item) { + pendingReleases.delete(key) + } + return true + }, (error) => { + item.nextAt = Date.now() + config.releaseRetryMs + logger.warn('资源释放失败，稍后重试', { kind: item.kind, id: item.id, error: error.message }) + return false + }) + .finally(() => { + if (item.attempt === attempt) { + item.attempt = undefined + } + }) + item.attempt = attempt + return attempt + } + + /** + * 登记并立即释放资源；释放确认前资源保留在待释放列表中，继续占用配额 + * + * @param {{ kind: 'ingress' | 'egress', id: string, ownerKey?: string }} resource 资源；Ingress 带用户维度键 + * @returns {Promise} 首次尝试是否已确认释放 + */ + function releaseResource(resource) { + const key = `${resource.kind}:${resource.id}` + let item = pendingReleases.get(key) + if (!item) { + item = { ...resource, attempt: undefined, nextAt: 0 } + pendingReleases.set(key, item) + } + return attemptRelease(item) + } + + /** + * 重试到期的待释放资源，同时进行的释放尝试不超过上限 + * + * @param {boolean} force 是否忽略重试间隔（退出时使用） + */ + function retryReleases(force) { + const now = Date.now() + let active = 0 // 本轮扫描时正在进行的释放尝试 + pendingReleases.forEach((item) => { + if (item.attempt) { + active++ + } + }) + pendingReleases.forEach((item) => { + if (active < RELEASE_CONCURRENCY && !item.attempt && (force || item.nextAt <= now)) { + active++ + void attemptRelease(item) + } + }) + } + + /** + * 跟踪资源创建：结果返回时若归属已失效（会话关闭、远端已更换），在移出创建列表前转入待释放，配额不出现空档； + * 结果未知（超时 / 网络错误）时转入未知列表，继续占用配额与用户维度，不再盲目重建 + * + * @param {{ kind: 'ingress' | 'egress', ownerKey: string, isCurrent: () => boolean }} item 创建项 + * @param {Promise>} promise LiveKit 创建调用 + * @returns {Promise>} 创建结果 + */ + function trackCreate(item, promise) { + inflightCreates.add(item) + const task = promise.then((result) => { + const id = item.kind === 'ingress' ? result.ingressId : result.egressId + if (id && !item.isCurrent()) { + void releaseResource({ kind: item.kind, id, ownerKey: item.ownerKey }) + } + inflightCreates.delete(item) + return result + }, (error) => { + if (isAmbiguousCreateError(error)) { + unknownCreates.add(item) + logger.warn('创建结果未知，保留占位直到网关重启后启动回收', { kind: item.kind, error: error.message }) + } + inflightCreates.delete(item) + throw error + }) + item.settled = task.then(() => undefined, () => undefined) + return task + } + + /** + * 创建会话：同一房间 + identity 已有任何未释放的会话或推流资源时返回 409，不替换正在推流的会话 + *

+ * 替换已进房的 Ingress 会触发 participant_left，后端会据此结束私聊通话，因此由原会话 DELETE 或租约到期后才允许新建 + * + * @param {{ room: string, identity: string, name: string, exp: number }} claims 房间 Token 授权 + * @returns {Promise<{ session: Record, sessionToken: string }>} 会话与会话凭证 + */ + async function createSession(claims) { + if (shuttingDown) { + throw new GatewayError(503, 'shutting_down', '网关正在退出') + } + if (isUserBusy(userKey(claims))) { + throw new GatewayError(409, 'session_busy', '该用户在本房间已有小程序通话会话或资源未释放；请先结束已有会话，持续出现请联系管理员') + } + if (countIngress() >= config.maxSessions) { + throw new GatewayError(429, 'session_limit', '小程序网关会话数已达上限') + } + const now = Date.now() + const sessionToken = randomBytes(32).toString('base64url') + const session = { + id: randomBytes(16).toString('hex'), // 会话编号，同时作为资源名后缀 + tokenDigest: digest(sessionToken), // 会话凭证摘要 + room: claims.room, // 房间名，取自 Token + identity: claims.identity, // 用户 identity，取自 Token + status: 'starting', // starting / active / closed + closed: false, // 是否已关闭；异步步骤完成后据此清理迟到资源 + closeReason: undefined, // 关闭原因 + releaseTask: undefined, // 资源释放任务（首次尝试完成即结束，未确认的资源转入待释放重试） + ingressId: undefined, // 小程序推流 Ingress + pushUrl: undefined, // 小程序推流地址（含 stream key，仅返回给创建者） + publishing: false, // 小程序推流是否已作为参与者进房 + everPublished: false, // 是否曾经进房；之后房间消失即视为通话结束 + remotes: new Map(), // 远端 userId -> 远端流 + waitingUserIds: [], // 已进房但尚无可转出轨道的远端 + overflowUserIds: [], // 超出单会话路数上限的远端 + capacityUserIds: [], // 因全局 Egress 上限暂未转出的远端 + remoteErrors: new Map(), // 远端 userId -> 最近一次转出失败原因 + unknownUserIds: new Set(), // Egress 创建结果未知的远端；本会话内不再为其重建，避免叠加未知资源 + presentUserIds: new Set(), // 最近一次同步时仍在房间内的远端 + retryAfter: new Map(), // 远端 userId -> 允许重试的时间 + syncing: false, // 是否正在同步房间 + syncError: undefined, // 最近一次同步失败原因 + syncTimer: undefined, // 同步定时器 + leaseExpiresAt: now + config.leaseTtlMs, // 租约到期时间 + expiresAt: Math.min(claims.exp * 1000, now + config.maxSessionMs), // 会话最晚结束时间 + } + sessions.set(session.id, session) + sessionsByUser.set(userKey(session), session) + try { + const ingress = await trackCreate({ + kind: 'ingress', + ownerKey: userKey(session), + isCurrent: () => !session.closed, + }, liveKit.createIngress({ + room: session.room, + identity: session.identity, + name: claims.name, + resourceName: `${RESOURCE_PREFIX}${session.id}`, + })) + if (session.closed) { // 创建期间会话已关闭：转入待释放（与 trackCreate 的登记幂等） + if (ingress.ingressId) { + void releaseResource({ kind: 'ingress', id: ingress.ingressId, ownerKey: userKey(session) }) + } + throw new GatewayError(409, 'session_closed', '会话在创建过程中已关闭') + } + session.ingressId = ingress.ingressId + if (!ingress.ingressId || !ingress.url || !ingress.streamKey) { + throw new GatewayError(502, 'ingress_url_missing', 'LiveKit 未返回 RTMP 推流地址，请检查 ingress.rtmp_base_url') + } + session.pushUrl = `${ingress.url}/${ingress.streamKey}` + session.status = 'active' + } catch (error) { + await closeSession(session, 'create_failed') + if (error instanceof GatewayError) { + throw error + } + logger.warn('创建 Ingress 失败', { room: session.room, error: error.message }) + throw new GatewayError(502, 'livekit_unavailable', '创建小程序推流失败') + } + session.syncTimer = setInterval(() => void syncSession(session), config.syncIntervalMs) + void syncSession(session) + logger.info('会话已创建', { session: session.id, room: session.room, userId: session.identity, ingressId: session.ingressId }) + return { session, sessionToken } + } + + /** + * 关闭会话：立即撤销推拉流授权与会话查询，释放 Ingress、Egress；重复调用返回同一个释放任务 + *

+ * 释放任务在首次尝试完成后结束；未确认释放的资源保留在待释放列表中重试，并继续占用配额与用户维度 + * + * @param {Record} session 会话 + * @param {string} reason 关闭原因 + * @returns {Promise} 释放任务 + */ + function closeSession(session, reason) { + if (session.releaseTask) { + return session.releaseTask + } + session.closed = true + session.closeReason = reason + session.status = 'closed' + clearInterval(session.syncTimer) + sessions.delete(session.id) + if (sessionsByUser.get(userKey(session)) === session) { + sessionsByUser.delete(userKey(session)) + } + const entries = Array.from(session.remotes.values()) + session.remotes.clear() + entries.forEach((entry) => { + if (streams.get(entry.streamName)?.entry === entry) { + streams.delete(entry.streamName) + } + }) + const resources = entries.filter(entry => entry.egressId).map(entry => ({ kind: 'egress', id: entry.egressId })) + if (session.ingressId) { + resources.push({ kind: 'ingress', id: session.ingressId, ownerKey: userKey(session) }) + } + session.releaseTask = Promise.all(resources.map(releaseResource)).then((results) => { + logger.info('会话已关闭', { session: session.id, room: session.room, reason, pendingRelease: results.filter(released => !released).length }) + }) + return session.releaseTask + } + + /** + * 移除单路远端流：立即撤销授权并释放 Egress；启动中的 Egress 由 trackCreate 在返回后转入待释放 + * + * @param {Record} session 会话 + * @param {Record} entry 远端流 + */ + function removeRemoteStream(session, entry) { + if (session.remotes.get(entry.userId) === entry) { + session.remotes.delete(entry.userId) + } + if (streams.get(entry.streamName)?.entry === entry) { + streams.delete(entry.streamName) + } + if (entry.egressId) { + void releaseResource({ kind: 'egress', id: entry.egressId }) + } + } + + /** + * 为远端参与者启动 Egress；每次启动使用新的流名与推拉流密钥 + * + * @param {Record} session 会话 + * @param {string} userId 远端 userId + * @param {ReturnType} media 媒体选择 + * @returns {Promise} + */ + async function startRemoteStream(session, userId, media) { + const publishSecret = randomBytes(24).toString('base64url') + const playToken = randomBytes(24).toString('base64url') + const entry = { + userId, // 远端 userId + audioSid: media.audioSid, // 转出的音频轨道 + videoSid: media.videoSid, // 转出的视频轨道 + videoSource: media.videoSource, // 视频来源（CAMERA / SCREEN_SHARE），仅用于日志排查 + videoVisible: media.videoVisible, // 视频是否可见（摄像头静音时为 false） + streamName: `${STREAM_PREFIX}${session.id.slice(0, 12)}_${userId}_${randomBytes(8).toString('hex')}`, // SRS 流名 + publishDigest: digest(publishSecret), // Egress 推流密钥摘要 + playToken, // 小程序拉流凭证明文，仅保存在内存中，用于在会话查询中重新下发 playUrl + playDigest: digest(playToken), // 拉流凭证摘要 + egressId: undefined, // Egress 编号 + publishing: false, // SRS 是否正在接收推流 + everPublished: false, // SRS 是否收到过推流；启动超时只看该标记 + startedAt: Date.now(), // 启动时间 + } + session.remotes.set(userId, entry) + streams.set(entry.streamName, { session, entry }) + const createItem = { // Egress 创建项；结果未知时作为占位保留 + kind: 'egress', + ownerKey: userKey(session), + isCurrent: () => !session.closed && session.remotes.get(userId) === entry, + } + try { + const egress = await trackCreate(createItem, liveKit.startTrackCompositeEgress({ + room: session.room, + audioTrackId: media.audioSid, + videoTrackId: media.videoSid, + url: `${config.srsPublishBaseUrl}/${entry.streamName}?secret=${publishSecret}`, + encoding: config.egressEncoding, + })) + if (session.closed || session.remotes.get(userId) !== entry) { // 启动期间远端已更换：转入待释放（与 trackCreate 的登记幂等） + if (egress.egressId) { + void releaseResource({ kind: 'egress', id: egress.egressId }) + } + return + } + entry.egressId = egress.egressId + session.remoteErrors.delete(userId) + logger.info('远端转出已启动', { session: session.id, userId, egressId: egress.egressId, audio: !!media.audioSid, video: !!media.videoSid }) + } catch (error) { + if (unknownCreates.has(createItem)) { + session.unknownUserIds.add(userId) + } + if (session.closed || session.remotes.get(userId) !== entry) { + return + } + removeRemoteStream(session, entry) + if (!session.unknownUserIds.has(userId)) { // 结果未知的远端不再重试，状态由 unknownUserIds 返回 failed + session.retryAfter.set(userId, Date.now() + config.egressRetryMs) + session.remoteErrors.set(userId, 'egress_failed') + } + logger.warn('远端转出启动失败', { session: session.id, userId, error: error.message }) + } + } + + /** + * 按房间现状同步远端转出：新增、按媒体组合重建、回收已结束的 Egress + * + * @param {Record} session 会话 + * @returns {Promise} + */ + async function syncSession(session) { + if (session.closed || session.syncing) { + return + } + session.syncing = true + try { + const participants = await liveKit.listParticipants(session.room) + if (session.closed) { + return + } + if (participants === null) { + session.publishing = false + if (session.everPublished) { + void closeSession(session, 'room_gone') + } + return + } + session.publishing = participants.some(participant => participant.identity === session.identity + && participant.kind === 'INGRESS' + && participant.state !== 'DISCONNECTED' + && participant.tracks.length > 0) + session.everPublished = session.everPublished || session.publishing + const desired = new Map() // 远端 userId -> 媒体选择 + const waiting = [] + const overflow = [] + participants + .filter(participant => isBusinessParticipant(participant) && participant.identity !== session.identity) + .sort((a, b) => a.joinedAt - b.joinedAt) + .forEach((participant) => { + const media = pickMedia(participant) + if (!media.audioSid && !media.videoSid) { + waiting.push(participant.identity) + } else if (desired.size >= config.maxRemotesPerSession) { + overflow.push(participant.identity) + } else { + desired.set(participant.identity, media) + } + }) + Array.from(session.remotes.values()).forEach((entry) => { + const media = desired.get(entry.userId) + if (!media || media.audioSid !== entry.audioSid || media.videoSid !== entry.videoSid) { + removeRemoteStream(session, entry) + } else { + entry.videoVisible = media.videoVisible + } + }) + const now = Date.now() + // Egress 已启动但始终没有推到 SRS（如编码管线一直拿不到可输出的媒体）：停止、冷却后重试，如实返回 failed + Array.from(session.remotes.values()).forEach((entry) => { + if (entry.egressId && !entry.everPublished && now - entry.startedAt > config.egressPublishTimeoutMs) { + removeRemoteStream(session, entry) + session.retryAfter.set(entry.userId, now + config.egressRetryMs) + session.remoteErrors.set(entry.userId, 'egress_no_media') + logger.warn('远端转出启动超时，未收到 SRS 推流', { session: session.id, userId: entry.userId, egressId: entry.egressId, videoSource: entry.videoSource || 'NONE', audio: !!entry.audioSid }) + } + }) + if (Array.from(session.remotes.values()).some(entry => entry.egressId && now - entry.startedAt > EGRESS_LIST_GRACE_MS)) { + const activeIds = new Set((await liveKit.listActiveEgress(session.room)).map(item => item.egressId)) + if (session.closed) { + return + } + Array.from(session.remotes.values()).forEach((entry) => { + if (entry.egressId && now - entry.startedAt > EGRESS_LIST_GRACE_MS && !activeIds.has(entry.egressId)) { + removeRemoteStream(session, entry) + session.retryAfter.set(entry.userId, now + config.egressRetryMs) + session.remoteErrors.set(entry.userId, 'egress_ended') + } + }) + } + const capacity = [] + desired.forEach((media, userId) => { + if (session.remotes.has(userId) || session.unknownUserIds.has(userId) || (session.retryAfter.get(userId) || 0) > now) { + return + } + if (shuttingDown) { + return + } + if (countEgress() >= config.maxEgress) { + capacity.push(userId) + return + } + void startRemoteStream(session, userId, media) + }) + Array.from(session.retryAfter.keys()).forEach((userId) => { + if (!desired.has(userId)) { + session.retryAfter.delete(userId) + session.remoteErrors.delete(userId) + } + }) + session.presentUserIds = new Set([...desired.keys(), ...waiting, ...overflow]) + session.waitingUserIds = waiting + session.overflowUserIds = overflow + session.capacityUserIds = capacity + session.syncError = undefined + } catch (error) { + session.syncError = 'livekit_unavailable' + logger.warn('同步房间失败', { session: session.id, room: session.room, error: error.message }) + } finally { + session.syncing = false + } + } + + /** 扫描租约与 Token 过期的会话，并重试到期的资源释放 */ + function reapSessions() { + retryReleases(false) + const now = Date.now() + sessions.forEach((session) => { + if (now > session.leaseExpiresAt) { + void closeSession(session, 'lease_expired') + } else if (now > session.expiresAt) { + void closeSession(session, 'token_expired') + } + }) + } + + /** + * 按会话编号与会话凭证查找会话；任何不匹配都按不存在处理 + * + * @param {string} sessionId 会话编号 + * @param {string | undefined} sessionToken 会话凭证 + * @returns {Record} 会话 + */ + function authorize(sessionId, sessionToken) { + const session = SESSION_ID_PATTERN.test(sessionId) ? sessions.get(sessionId) : undefined + if (!session || !matches(sessionToken, session.tokenDigest)) { + throw new GatewayError(404, 'session_not_found', '会话不存在或已结束') + } + return session + } + + /** + * 续租会话并返回小程序可见状态 + * + * @param {Record} session 会话 + * @returns {Record} 会话状态 + */ + function touch(session) { + session.leaseExpiresAt = Date.now() + config.leaseTtlMs + const remotes = Array.from(session.remotes.values()).map(entry => ({ + userId: Number(entry.userId), + state: entry.publishing ? 'ready' : 'starting', + ready: entry.publishing, + playUrl: entry.publishing ? `${config.srsPlayBaseUrl}/${entry.streamName}?token=${entry.playToken}` : null, // SRS 收到推流后才下发，避免播放器提前拉空流 + hasAudio: !!entry.audioSid, + hasVideo: !!entry.videoSid, + videoVisible: entry.videoVisible, + })) + const listed = new Set(session.remotes.keys()) // 已输出的远端，避免同一远端重复出现 + const appendState = (userIds, state) => userIds.forEach((userId) => { + if (!listed.has(userId)) { + listed.add(userId) + remotes.push({ userId: Number(userId), state, ready: false, playUrl: null, hasAudio: false, hasVideo: false, videoVisible: false }) + } + }) + appendState(session.waitingUserIds, 'waiting') + appendState(session.overflowUserIds, 'overflow') + appendState(session.capacityUserIds, 'capacity') + appendState(Array.from(session.remoteErrors.keys()), 'failed') + appendState(Array.from(session.unknownUserIds).filter(userId => session.presentUserIds.has(userId)), 'failed') + return { + sessionId: session.id, + status: session.status, + publishing: session.publishing, + remotes, + error: session.syncError || null, + leaseTtlMs: config.leaseTtlMs, + pollIntervalMs: config.pollIntervalMs, + expiresAt: session.expiresAt, + } + } + + /** + * 查找 SRS 回调对应的远端流 + * + * @param {Record} hook SRS 回调体 + * @returns {{ session: Record, entry: Record } | undefined} 远端流 + */ + function findStream(hook) { + if (!hook || hook.app !== config.srsApp || typeof hook.stream !== 'string') { + return undefined + } + const ref = streams.get(hook.stream) + if (!ref || ref.session.closed || ref.session.remotes.get(ref.entry.userId) !== ref.entry) { + return undefined + } + return ref + } + + /** + * SRS on_publish：只允许网关启动的 Egress 携带对应密钥推流 + * + * @param {Record} hook SRS 回调体 + * @returns {boolean} 是否允许 + */ + function authorizePublish(hook) { + const ref = findStream(hook) + if (!ref || !matches(parseHookParam(hook.param).get('secret'), ref.entry.publishDigest)) { + return false + } + ref.entry.publishing = true + ref.entry.everPublished = true + return true + } + + /** + * SRS on_unpublish：标记远端流未就绪 + * + * @param {Record} hook SRS 回调体 + */ + function handleUnpublish(hook) { + const ref = findStream(hook) + if (ref && matches(parseHookParam(hook.param).get('secret'), ref.entry.publishDigest)) { + ref.entry.publishing = false + } + } + + /** + * SRS on_play：只允许持有本会话拉流凭证的小程序播放对应远端流 + * + * @param {Record} hook SRS 回调体 + * @returns {boolean} 是否允许 + */ + function authorizePlay(hook) { + const ref = findStream(hook) + return !!ref && matches(parseHookParam(hook.param).get('token'), ref.entry.playDigest) + } + + /** + * 判断 Egress 是否推往本网关的 SRS 流 + *

+ * LiveKit 会把 RTMP 地址最后一段脱敏为「{首 3 位...末 3 位}」，只有 scheme / host / port / app 可信； + * 因此归属以网关专用 app 为准：地址需为「SRS 推流根地址 / 单段流名」，且单段为完整网关流名或其脱敏形式 + * + * @param {unknown} value EgressInfo 原始 JSON + * @returns {boolean} 是否本网关创建 + */ + function isGatewayEgress(value) { + if (typeof value === 'string') { + const prefix = `${config.srsPublishBaseUrl}/` + if (!GATEWAY_SRS_APP_PATTERN.test(config.srsApp) || !value.startsWith(prefix)) { + return false + } + const segment = value.slice(prefix.length) + return STREAM_NAME_PATTERN.test(segment.split('?')[0]) || REDACTED_STREAM_PATTERN.test(segment) + } + if (value && typeof value === 'object') { + return Object.values(value).some(isGatewayEgress) + } + return false + } + + /** + * 单实例启动时回收上次异常退出残留的 Ingress / Egress；任一资源释放失败都抛出，由调用方重试 + * + * @returns {Promise} + */ + async function cleanupOrphans() { + const ingresses = (await liveKit.listIngress()).filter(item => item.name.startsWith(RESOURCE_PREFIX)) + const egresses = (await liveKit.listActiveEgress()).filter(item => isGatewayEgress(item.raw)) + const results = await Promise.allSettled([ + ...ingresses.map(item => liveKit.deleteIngress(item.ingressId)), + ...egresses.map(item => liveKit.stopEgress(item.egressId)), + ]) + const failed = results.filter(result => result.status === 'rejected').length + if (failed > 0) { + throw new Error(`残留资源回收失败 ${failed} 项`) + } + logger.info('残留资源回收完成', { ingress: ingresses.length, egress: egresses.length }) + } + + /** + * 退出：拒绝新会话，关闭全部会话，等待进行中的创建返回并确认全部资源释放 + *

+ * 释放持续失败时不会返回，由入口的退出超时以非零状态结束；存在结果未知的创建时抛出错误，同样以非零状态退出，避免误报干净退出 + * + * @returns {Promise} + */ + async function shutdown() { + shuttingDown = true + clearInterval(reaper) + Array.from(sessions.values()).forEach(session => void closeSession(session, 'shutdown')) + while (inflightCreates.size > 0 || pendingReleases.size > 0) { + if (inflightCreates.size > 0) { + await Promise.allSettled(Array.from(inflightCreates, item => item.settled)) + continue + } + retryReleases(true) + await Promise.allSettled(Array.from(pendingReleases.values(), item => item.attempt)) + if (pendingReleases.size > 0) { + await delay(SHUTDOWN_RELEASE_RETRY_MS) + } + } + if (unknownCreates.size > 0) { + throw new Error(`存在 ${unknownCreates.size} 个结果未知的 Ingress / Egress 创建，需重启网关由启动回收清理`) + } + } + + return { + createSession, + closeSession, + authorize, + touch, + authorizePublish, + handleUnpublish, + authorizePlay, + cleanupOrphans, + shutdown, + } +} diff --git a/script/livekit-poc/mp-gateway/srs.conf b/script/livekit-poc/mp-gateway/srs.conf new file mode 100644 index 0000000..d30d2c2 --- /dev/null +++ b/script/livekit-poc/mp-gateway/srs.conf @@ -0,0 +1,28 @@ +# SRS：接收 Egress 推来的远端流，供小程序 live-player 拉取；推流与播放都经网关 HTTP 回调鉴权 +# 本实例由网关独占：回调对整个 vhost 生效，网关只放行专用 app（yudao-mp-gw）下自己创建的流，勿与其它直播业务共用 +# 默认只监听回环；真机联调时改为 listen 1936; 并在防火墙 / 安全组只放通需要的来源 +listen 127.0.0.1:1936; +max_connections 200; +daemon off; +srs_log_tank console; + +vhost __defaultVhost__ { + # 低延迟：关闭合并写、开启 TCP_NODELAY；GOP 缓存让播放器首帧不必等下一个关键帧（Egress 关键帧间隔 1 秒） + min_latency on; + tcp_nodelay on; + play { + gop_cache on; + queue_length 10; + mw_latency 100; + } + publish { + mr off; + } + # 网关回调端口只监听回环；网关不可用时 SRS 拒绝推流与播放 + http_hooks { + enabled on; + on_publish http://127.0.0.1:8091/srs/hooks/on_publish; + on_unpublish http://127.0.0.1:8091/srs/hooks/on_unpublish; + on_play http://127.0.0.1:8091/srs/hooks/on_play; + } +}