BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 8aed362ea4cb851ac2d25ce331619dd85d6a91d8

Authored by YunaiV
1 parent d608a9a8

feat(im): 增加微信小程序 LiveKit RTMP 网关

Inline Side-by-side
Showing 11 changed files with 2104 additions and 0 deletions
script/livekit-poc/mp-gateway/.env.example 0 → 100644
  View file @8aed362
  1 +# 复制为 .env 后填写；.env 已被 .gitignore 忽略，勿提交真实凭据
  2 +# 必须与后端 yudao.im.rtc.api-key / yudao.im.rtc.api-secret 一致，网关用它校验后端签发的房间 Token
  3 +LIVEKIT_API_KEY=replace-with-api-key
  4 +LIVEKIT_API_SECRET=replace-with-secret-at-least-32-chars
  5 +
  6 +# 网关试验开关：默认 false，网关容器打印说明后退出
  7 +MP_GATEWAY_ENABLED=false
  8 +
  9 +# 后端 webhook（host 网络下容器可直接访问宿主机回环）
  10 +MP_LIVEKIT_WEBHOOK_URL=http://127.0.0.1:48080/admin-api/im/livekit/webhook
  11 +
  12 +# 本机回环联调保持 127.0.0.1；真机联调改为手机可达的主机名 / IP，并同步放开对应监听与防火墙
  13 +MP_LIVEKIT_BIND_ADDRESS=127.0.0.1
  14 +MP_LIVEKIT_NODE_IP=127.0.0.1
  15 +MP_INGRESS_RTMP_PUBLIC_HOST=127.0.0.1
  16 +MP_SRS_PLAY_PUBLIC_HOST=127.0.0.1
  17 +MP_GATEWAY_API_HOST=127.0.0.1
  18 +
  19 +# 资源上限：每个小程序会话 1 路 Ingress 转码，每个远端 1 路 Egress 转码
  20 +MP_GATEWAY_MAX_SESSIONS=4
  21 +MP_GATEWAY_MAX_EGRESS=8
  22 +
  23 +# 网关专用 SRS app（yudao-mp-gw 或 yudao-mp-gw-<后缀>），该 app 只能给网关使用
  24 +# MP_GATEWAY_SRS_APP=yudao-mp-gw
  25 +
  26 +# 其它可调参数，含义与默认值见 README「4. 配置」；不配置即使用默认值
  27 +# MP_GATEWAY_POLL_INTERVAL_MS=2000
  28 +# MP_GATEWAY_LEASE_TTL_MS=20000
  29 +# MP_GATEWAY_MAX_SESSION_MS=7200000
  30 +# MP_GATEWAY_MAX_REMOTES_PER_SESSION=4
  31 +# MP_GATEWAY_EGRESS_RETRY_MS=10000
  32 +# MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS=20000
  33 +# MP_GATEWAY_RELEASE_RETRY_MS=5000
  34 +# MP_GATEWAY_CLEANUP_RETRY_MS=5000
  35 +# MP_GATEWAY_REQUEST_TIMEOUT_MS=10000
  36 +# MP_GATEWAY_CLOCK_SKEW_SEC=30
  37 +# MP_GATEWAY_EGRESS_WIDTH=640
  38 +# MP_GATEWAY_EGRESS_HEIGHT=360
  39 +# MP_GATEWAY_EGRESS_FRAMERATE=15
  40 +# MP_GATEWAY_EGRESS_VIDEO_KBPS=600
  41 +# MP_GATEWAY_EGRESS_AUDIO_KBPS=64
  42 +# MP_GATEWAY_EGRESS_KEYFRAME_SEC=1
script/livekit-poc/mp-gateway/.gitignore 0 → 100644
  View file @8aed362
  1 +.env
script/livekit-poc/mp-gateway/README.md 0 → 100644
  View file @8aed362
  1 +# 小程序 RTMP ↔ LiveKit 网关（试验）
  2 +
  3 +> **试验功能，默认关闭，勿用于生产。** 小程序音视频经 RTMP 转码中转，端到端延迟预计为**秒级**（通常 1~3 秒，弱网更高），明显差于 PC / H5 / App 的 WebRTC 直连；回声消除、资源成本、真机兼容性均未经验收。
  4 +
  5 +## 1. 为什么需要网关
  6 +
  7 +微信小程序不能运行 `livekit-client`（无 WebRTC），原生 `live-pusher` 只能推 RTMP，`live-player` 只能拉 RTMP / FLV。网关在不更换 LiveKit 的前提下把小程序接入同一个房间：
  8 +
  9 +```
  10 +小程序 live-pusher ──RTMP──▶ LiveKit Ingress ──WebRTC──▶ LiveKit 房间 ◀──WebRTC── PC / H5 / App
  11 + │
  12 + 每个远端参与者 1 路 Track Composite Egress
  13 + ▼
  14 +小程序 live-player ◀──RTMP── SRS（推流、播放均回调网关鉴权）◀──RTMP──┘
  15 +```
  16 +
  17 +- 小程序自己的 Ingress 参与者 identity 即后端 Token 内的 userId（不加后缀），与 PC / App 同账号进房时沿用 LiveKit 同 identity 互踢；后端 webhook 的 joined / left 照常生效，**无需修改 Java 代码**。
  18 +- 每个远端单独一路流，**从不混入小程序自己的声音**；视频屏幕共享优先、摄像头回退；Egress、Agent、SIP 等非业务参与者不转出。
  19 +- 群通话一个远端一路 live-player，单会话默认最多 4 路，超出部分如实返回 `overflow`。
  20 +
  21 +## 2. 目录
  22 +
  23 +| 文件 | 说明 |
  24 +| --- | --- |
  25 +| `src/server.mjs` | 入口：小程序会话 API（8090）+ SRS 回调（8091，仅回环） |
  26 +| `src/session-manager.mjs` | 会话、租约、Ingress / Egress 生命周期、SRS 鉴权、残留回收 |
  27 +| `src/jwt.mjs` | 校验后端签发的房间 Token（HS256）、签发管理 Token |
  28 +| `src/livekit-api.mjs` | LiveKit Twirp API 最小客户端（无第三方依赖） |
  29 +| `src/config.mjs` | 环境变量配置，`MP_GATEWAY_ENABLED=true` 才允许启动 |
  30 +| `docker-compose.yml` | 独立试验环境：Redis + LiveKit + Ingress + Egress + SRS + 网关 |
  31 +| `srs.conf` | SRS 配置，HTTP 回调鉴权 |
  32 +| `.env.example` | 凭据与地址占位，复制为 `.env` 使用 |
  33 +
  34 +网关进程只使用 Node.js 22 内置模块（`http`、`crypto`、`fetch`），无需安装 npm 依赖；运行时依赖 LiveKit Server、Ingress、Egress 与 SRS，由同目录 `docker-compose.yml` 一并启动。
  35 +
  36 +## 3. 本机回环联调
  37 +
  38 +与上级目录 `../docker-compose.yml`（LiveKit-only PoC）**二选一**，两者都占用 7880 等端口。
  39 +
  40 +```bash
  41 +cd script/livekit-poc/mp-gateway
  42 +docker compose -f ../docker-compose.yml down # 停掉 LiveKit-only PoC
  43 +cp .env.example .env # 填写与后端 yudao.im.rtc.api-key / api-secret 一致的凭据，MP_GATEWAY_ENABLED=true
  44 +docker compose up -d
  45 +docker compose logs -f mp-gateway # 看到「小程序网关已启动（试验功能）」
  46 +curl -s http://127.0.0.1:8090/healthz # {"status":"ok"}
  47 +```
  48 +
  49 +- 全部服务使用 `network_mode: host`。macOS 需 Docker Desktop 4.34+，并在 Settings → Resources → Network 勾选「Enable host networking」，否则容器能启动但端口不通。
  50 +- 后端 `yudao.im.rtc.livekit-url` 仍为 `ws://127.0.0.1:7880`；webhook 默认回调 `http://127.0.0.1:48080/admin-api/im/livekit/webhook`。
  51 +- 本机端口：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 鉴权），在不可信局域网中请配合防火墙。
  52 +
  53 +### 3.1 用 FFmpeg 模拟小程序
  54 +
  55 +1. 用 H5 或 PC 账号 A 发起通话，账号 B（模拟小程序端）取得同一房间的 Token（例如在 H5 发起 / 接听接口响应中取得 `token`，**只在本机终端使用，勿写日志或提交**）。
  56 +2. 创建会话：
  57 +
  58 + ```bash
  59 + curl -s -X POST -H "Authorization: Bearer $ROOM_TOKEN" http://127.0.0.1:8090/mp-rtc/sessions
  60 + ```
  61 +
  62 +3. 用返回的 `pushUrl` 推测试流（模拟 live-pusher）：
  63 +
  64 + ```bash
  65 + ffmpeg -re -f lavfi -i testsrc=size=640x360:rate=15 -f lavfi -i sine=frequency=440 \
  66 + -c:v libx264 -preset veryfast -tune zerolatency -g 15 -c:a aac -ar 48000 -f flv "$PUSH_URL"
  67 + ```
  68 +
  69 + 账号 A 的页面应出现 B 的测试画面与提示音；会话查询 `publishing` 变为 `true`。
  70 +4. 每 2 秒查询会话（同时续租），待远端 A 的 `state` 为 `ready` 后，用 `playUrl` 拉流（模拟 live-player）：
  71 +
  72 + ```bash
  73 + curl -s -H "Authorization: Bearer $SESSION_TOKEN" http://127.0.0.1:8090/mp-rtc/sessions/$SESSION_ID
  74 + ffplay -fflags nobuffer -flags low_delay "$PLAY_URL"
  75 + ```
  76 +
  77 +5. 反向校验：去掉 `playUrl` 的 `token` 参数或改错一位，SRS 应拒绝播放；停止查询超过 `leaseTtlMs`，或 `DELETE` 会话后，Ingress、Egress 应被释放（`docker compose logs mp-gateway` 有「会话已关闭」）。
  78 +
  79 +## 4. 配置
  80 +
  81 +| 变量 | 默认值 | 说明 |
  82 +| --- | --- | --- |
  83 +| `MP_GATEWAY_ENABLED` | 无 | 必须为 `true` 才启动 |
  84 +| `MP_GATEWAY_API_HOST` / `_PORT` | `127.0.0.1` / `8090` | 小程序 API；公网部署放在 HTTPS 反向代理之后 |
  85 +| `MP_GATEWAY_HOOK_HOST` / `_PORT` | `127.0.0.1` / `8091` | SRS 回调，只允许 SRS 访问 |
  86 +| `MP_GATEWAY_LIVEKIT_HTTP_URL` | `http://127.0.0.1:7880` | LiveKit API 地址 |
  87 +| `MP_GATEWAY_LIVEKIT_API_KEY` / `_SECRET` | 必填 | 与后端一致；secret 至少 32 位 |
  88 +| `MP_GATEWAY_SRS_PUBLISH_BASE_URL` | `rtmp://127.0.0.1:1936/yudao-mp-gw` | Egress 推到 SRS 的内部地址；app 必须为网关专用的 `yudao-mp-gw` 或 `yudao-mp-gw-<后缀>`，否则拒绝启动 |
  89 +| `MP_GATEWAY_SRS_PLAY_BASE_URL` | 同上 | 小程序拉流的外部地址，app 需与推流一致 |
  90 +| `MP_GATEWAY_POLL_INTERVAL_MS` | `2000` | 建议轮询间隔，也是网关同步房间的间隔 |
  91 +| `MP_GATEWAY_LEASE_TTL_MS` | `20000` | 租约，至少 3 倍轮询间隔；超时未查询即释放 |
  92 +| `MP_GATEWAY_MAX_SESSION_MS` | 2 小时 | 单会话最长存活，另受房间 Token `exp` 限制 |
  93 +| `MP_GATEWAY_MAX_SESSIONS` | `4` | 全局会话数（每个 1 路 Ingress 转码） |
  94 +| `MP_GATEWAY_MAX_REMOTES_PER_SESSION` | `4` | 单会话最多拉取的远端 |
  95 +| `MP_GATEWAY_MAX_EGRESS` | `8` | 全局 Egress 路数 |
  96 +| `MP_GATEWAY_EGRESS_RETRY_MS` | `10000` | 单路 Egress 失败后的冷却 |
  97 +| `MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS` | `20000` | Egress 启动后等待 SRS `on_publish` 的最长时间，超时停止并按 `failed` 冷却重试 |
  98 +| `MP_GATEWAY_CLEANUP_RETRY_MS` | `5000` | 启动回收失败后的重试间隔 |
  99 +| `MP_GATEWAY_RELEASE_RETRY_MS` | `5000` | Ingress / Egress 释放失败后的重试间隔；确认释放前继续占用配额 |
  100 +| `MP_GATEWAY_REQUEST_TIMEOUT_MS` | `10000` | 调用 LiveKit API 的超时 |
  101 +| `MP_GATEWAY_CLOCK_SKEW_SEC` | `30` | 校验房间 Token `nbf` / `exp` 允许的时钟偏差 |
  102 +| `MP_GATEWAY_EGRESS_WIDTH` / `HEIGHT` / `FRAMERATE` / `VIDEO_KBPS` / `AUDIO_KBPS` / `KEYFRAME_SEC` | 640 / 360 / 15 / 600 / 64 / 1 | Egress 编码（H264 Baseline + AAC） |
  103 +
  104 +上表是网关进程直接读取的环境变量，自行运行 `node src/server.mjs` 时全部按此配置。使用本目录 `docker-compose.yml` 时，网关容器的环境由 compose 组装，`.env` 中的变量并非全部原样转发：
  105 +
  106 +- **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 内各服务端口一一对应。
  107 +- **改名转发**：LiveKit 凭据读取 `.env` 的 `LIVEKIT_API_KEY` / `LIVEKIT_API_SECRET`（同时供 LiveKit、Ingress、Egress 使用），写 `MP_GATEWAY_LIVEKIT_API_KEY` / `_SECRET` 不生效。
  108 +- **由 compose 变量拼出**：`MP_GATEWAY_SRS_PUBLISH_BASE_URL` 固定为 `rtmp://127.0.0.1:1936/<app>`，`MP_GATEWAY_SRS_PLAY_BASE_URL` 为 `rtmp://<MP_SRS_PLAY_PUBLIC_HOST>:1936/<app>`；`<app>` 取 `MP_GATEWAY_SRS_APP`（默认 `yudao-mp-gw`），`MP_SRS_PLAY_PUBLIC_HOST` 默认 `127.0.0.1`。
  109 +- **原样转发**：`MP_GATEWAY_ENABLED`（默认 `false`）、`MP_GATEWAY_API_HOST`（默认 `127.0.0.1`），以及从 `MP_GATEWAY_POLL_INTERVAL_MS` 到 Egress 编码的其余可调参数，未配置即用上表默认值（见 `.env.example`）。
  110 +
  111 +## 5. 会话 API
  112 +
  113 +所有响应 `Cache-Control: no-store`，错误体为 `{ "code": "...", "message": "..." }`。不支持 CORS（小程序 `uni.request` 不受 CORS 限制）。
  114 +
  115 +### 5.1 创建会话 `POST /mp-rtc/sessions`
  116 +
  117 +- 请求头 `Authorization: Bearer <后端签发的房间 Token>`；请求体可为空（上限 4KB）。**不得**携带后端登录 accessToken。
  118 +- 网关严格校验：HS256 签名、`iss` = apiKey、`exp` / `nbf`（允许 30 秒时钟偏差）、`sub` 为数字 userId、`video.roomJoin` / `canPublish` / `canSubscribe` 为 `true`、非 hidden / recorder。房间与身份**只取自 Token**。
  119 +- 同一用户在同一房间已有未释放的会话或推流资源（含释放中、创建中）时返回 `409 session_busy`，不会替换正在推流的会话：替换已进房的 Ingress 会触发 `participant_left`，后端会据此结束私聊通话。原会话 `DELETE` 或租约到期、资源确认释放后才能再次创建；若该房间 + 用户存在创建结果未知的占位，则需重启网关才能再次创建（见第 7 节）。网关不会返回已有会话的凭证。
  120 +
  121 +`201` 响应：
  122 +
  123 +```json
  124 +{
  125 + "sessionId": "32 位十六进制",
  126 + "sessionToken": "会话凭证，仅本会话后续请求使用",
  127 + "userId": 1,
  128 + "pushUrl": "rtmp://host:1935/x/<stream key>",
  129 + "pollIntervalMs": 2000,
  130 + "leaseTtlMs": 20000,
  131 + "expiresAt": 1760000000000
  132 +}
  133 +```
  134 +
  135 +错误：`401` Token 无效 / 过期，`403` 授权不足，`429` 会话数已满，`502` LiveKit / Ingress 不可用或未配置 `ingress.rtmp_base_url`，`409` 同一用户已有会话（`session_busy`）或会话在创建过程中被关闭（`session_closed`），`503` 网关正在退出（`shutting_down`）或启动回收尚未完成（`cleanup_pending`）。创建过程中客户端断开，网关会立即释放刚创建的资源。
  136 +
  137 +### 5.2 查询并续租 `GET /mp-rtc/sessions/{sessionId}`
  138 +
  139 +请求头 `Authorization: Bearer <sessionToken>`；会话不存在、已结束或凭证错误统一返回 `404`。
  140 +
  141 +```json
  142 +{
  143 + "sessionId": "...",
  144 + "status": "active",
  145 + "publishing": true,
  146 + "remotes": [
  147 + { "userId": 2, "state": "ready", "ready": true, "playUrl": "rtmp://host:1936/yudao-mp-gw/<stream>?token=<拉流凭证>", "hasAudio": true, "hasVideo": true, "videoVisible": true }
  148 + ],
  149 + "error": null,
  150 + "leaseTtlMs": 20000,
  151 + "pollIntervalMs": 2000,
  152 + "expiresAt": 1760000000000
  153 +}
  154 +```
  155 +
  156 +- `publishing`：小程序推流已作为 LiveKit 参与者进房（Ingress 参与者已有轨道）。
  157 +- `remotes[].state`：
  158 + - `ready`：SRS 已收到该远端的流，`playUrl` 可播放；只有此状态下 `playUrl` 非空。
  159 + - `starting`：Egress 启动中。
  160 + - `waiting`：远端已进房，暂无可转出的音视频轨道。
  161 + - `overflow`：超出单会话路数上限，不会转出。
  162 + - `capacity`：全局 Egress 已满，空出后自动重试。
  163 + - `failed`：Egress 启动失败、超过 `MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS` 仍未推到 SRS，或中途结束时冷却后自动重试；创建结果未知（见「7. 生命周期与资源」）时不再为该远端新建转流，直到网关重启。具体原因（`egress_failed` / `egress_no_media` / `egress_ended` / 创建结果未知）只记录在网关日志。
  164 +- 远端媒体组合变化（如开始 / 结束屏幕共享）时会换新的流名与 `playUrl`，客户端需按最新值重新播放。
  165 +- `error`：最近一次同步房间失败为 `livekit_unavailable`，否则为 `null`。
  166 +
  167 +### 5.3 结束会话 `DELETE /mp-rtc/sessions/{sessionId}`
  168 +
  169 +请求头同上，成功 `204`，释放 Ingress 与全部 Egress。
  170 +
  171 +### 5.4 健康检查 `GET /healthz`
  172 +
  173 +启动回收完成后 `200 {"status":"ok"}`；完成前 `503 {"status":"cleanup_pending"}`。
  174 +
  175 +## 6. SRS 鉴权与凭证
  176 +
  177 +- 每路 Egress 使用新的随机流名、推流密钥（`secret`）、拉流凭证（`token`）。推流密钥与会话凭证只保存摘要；拉流凭证需要在会话查询中重新下发 `playUrl`，因此明文保存在网关进程内存中，会话关闭或远端流移除时随之删除。所有凭证都不落盘、不写日志。
  178 +- `on_publish`：只允许当前存活会话中对应流名 + 正确 `secret` 推流；`on_play`：只允许正确 `token` 播放；网关不可用时 SRS 拒绝。
  179 +- 日志只记录会话编号、房间、userId、资源编号，**不记录**房间 Token、会话凭证、stream key、推拉流 URL；LiveKit API 出错时只记录方法、HTTP 状态与 Twirp 错误码，不记录上游返回的错误原文。
  180 +
  181 +## 7. 生命周期与资源
  182 +
  183 +- 会话超过 `leaseTtlMs` 未查询、到达 Token `exp` 或 `MP_GATEWAY_MAX_SESSION_MS`、房间被销毁（曾进房之后）时自动释放。
  184 +- 会话关闭或远端流移除时立即撤销推拉流授权与会话查询；对应的 Ingress / Egress 删除失败时保留在内存待释放列表中，按 `MP_GATEWAY_RELEASE_RETRY_MS` 重试（同时最多 4 个重试请求），**确认释放前继续计入会话数与 Egress 路数上限**，同一用户也不能新建会话。
  185 +- 异步创建 Ingress / Egress 期间会话被关闭或远端已变化时，结果返回后立即转入待释放；创建中的资源同样计入上限。
  186 +- `SIGINT` / `SIGTERM` 时拒绝新会话，关闭全部会话，等待进行中的创建返回并确认全部资源释放后以 0 退出；15 秒内未能全部释放则以非零状态退出，剩余资源由下次启动回收或按 7.2 手动处理。
  187 +- **创建结果未知**：创建 Ingress / Egress 时超时、网络错误、HTTP 5xx（含反向代理 502 / 504），或成功响应缺少资源编号，LiveKit 侧都可能已建好资源而网关拿不到编号。网关按 fail-closed 处理：该次创建作为占位**永久计入**会话数 / Egress 路数上限，同一房间 + 用户不能再新建会话（`409 session_busy`），发现结果未知后不再为对应远端新建转流，避免不断叠加未知资源：若该远端换轨后的新转流在旧请求结果未知前已经启动，则继续按其真实状态显示和运行，否则显示 `failed`；旧的未知占位仍计入配额直到重启。占位只能**重启网关**后由启动回收（Ingress 按名称前缀、Egress 按专用 app）清除；存在占位时网关退出以非零状态结束。出现日志「创建结果未知」后请择机重启网关。注意 LiveKit 在 Egress 容量不足时也可能返回 5xx，同样会被保守占位、需要重启恢复，这是试验阶段的取舍；4xx 等明确拒绝的失败按未创建处理，沿用冷却重试。
  188 +- **单实例限制**：会话保存在内存，多实例共用同一 LiveKit 会互相清理，暂不支持。
  189 +
  190 +### 7.1 启动回收与专用 SRS app
  191 +
  192 +- 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-<后缀>`）只能给网关使用。
  193 +- 网关启动后先回收残留：删除名称前缀为 `yudao-mp-gw-` 的 Ingress；停止输出地址为「`MP_GATEWAY_SRS_PUBLISH_BASE_URL`/单段流名」且流名为完整网关流名或 `{yud...xxx}` 脱敏形式的活跃 Egress。其它 app（如 `live`）下的 Egress 不受影响。
  194 +- 回收任一步失败都会按 `MP_GATEWAY_CLEANUP_RETRY_MS` 后台重试；完成前 `/healthz` 与创建会话返回 `503 cleanup_pending`，避免旧残留与新会话并存。
  195 +- `srs.conf` 的 HTTP 回调对整个 vhost 生效，网关会拒绝非专用 app 与未知流名的推流 / 播放，所以 SRS 需**独占一个实例或 vhost**，不要与其它直播业务共用。
  196 +- 已经建立的 RTMP 连接不会因网关重启而重新鉴权；残留 Egress 停止后其推流断开，播放端随之断流。
  197 +
  198 +### 7.2 无法自动回收时的手动恢复
  199 +
  200 +LiveKit 长时间不可用、回收一直失败时，网关会持续返回 `cleanup_pending`，不会自行放行。恢复步骤：
  201 +
  202 +1. `docker compose stop mp-gateway`；
  203 +2. 通过 LiveKit API（`ListIngress`）删除名称前缀为 `yudao-mp-gw-` 的 Ingress；
  204 +3. 通过 `ListEgress`（active）停止输出地址以 `MP_GATEWAY_SRS_PUBLISH_BASE_URL/` 开头的 Egress；
  205 +4. `docker compose restart srs`，断开 SRS 上残留的推流与播放连接；
  206 +5. `docker compose start mp-gateway`，确认 `/healthz` 返回 `ok`。
  207 +- 成本估算（LiveKit 官方 CPU 估值）：每个小程序会话 1 路 RTMP Ingress 转码约 2 核，每个远端 1 路 Track Composite Egress 约 2 核。1 对 1 小程序通话约 4 核，Egress 超出 CPU 余量会被 LiveKit 拒绝并如实返回 `failed`。
  208 +
  209 +## 8. 已知限制与待验证点
  210 +
  211 +- **延迟**：RTMP 推流 + Ingress 转码 + Egress 转码 + RTMP 拉流，秒级延迟，不适合对实时性要求高的场景。
  212 +- **回声**：小程序 live-pusher 与 live-player 是两个独立组件，系统回声消除效果未验证，试验阶段建议戴耳机。
  213 +- **静音状态**：小程序关闭摄像头或麦克风时，Ingress 推流仍在，PC / App 看到的是黑屏或静音画面，而不是 LiveKit 的 muted 状态。
  214 +- **后台 / 断网**：小程序进入后台或 RTMP 网络断开（live-pusher 内部重连耗尽，状态码 -1307）时推流中断，Ingress 参与者离房，1 对 1 通话会由后端 webhook 结束；小程序端也会在推流丢失超过时限后结束本次通话。不支持无缝恢复，需要重新发起。
  215 +- **屏幕共享转出**：本地真实媒体验证中出现过屏幕共享的 Track Composite Egress 处于 `EGRESS_ACTIVE` 但始终未推到 SRS 的情况（摄像头转出正常），根因待排查；网关只做启动超时与 `failed` 如实返回，不静默回退摄像头。
  216 +- 本地真实媒体验证（FFmpeg 模拟推流 + H5，非微信原生组件）已观察到 Egress 地址中的 `?secret=` 原样传给 SRS 并通过回调校验。
  217 +- **待真实运行验证**：纯音频 Track Composite 推 RTMP；live-player 拉带 `?token=` 的 RTMP 地址；Ingress 对 live-pusher RTC 模式码流的兼容性。
  218 +
  219 +## 9. 真机要求
  220 +
  221 +- 小程序主体为企业，类目在 live-pusher / live-player 支持范围内，并在小程序后台「开发 → 开发管理 → 接口设置」开通「实时播放音视频流」「实时录制音视频流」。
  222 +- 网关 API 部署在 HTTPS 域名后，并加入小程序「request 合法域名」。
  223 +- 手机可达的地址：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`）。
  224 +- 开发者工具不支持 live-pusher 推流，必须真机调试。
script/livekit-poc/mp-gateway/docker-compose.yml 0 → 100644
  View file @8aed362
  1 +# 小程序 RTMP ↔ LiveKit 网关试验环境（默认关闭，勿用于生产）
  2 +# 与上级目录的 LiveKit-only PoC 二选一：两者都占用 7880 等端口，启动本环境前先 docker compose -f ../docker-compose.yml down
  3 +#
  4 +# 全部服务使用 host 网络：Ingress / Egress 作为 WebRTC 客户端连接 LiveKit 广播的 node_ip（127.0.0.1），
  5 +# 端口映射模式下容器内的 127.0.0.1 不是 LiveKit；macOS 需 Docker Desktop 4.34+ 并开启 Settings → Resources → Network → Enable host networking
  6 +#
  7 +# 凭据通过同目录 .env 提供（参考 .env.example），必须与后端 yudao.im.rtc.api-key / api-secret 一致
  8 +name: yudao-livekit-mp-gateway
  9 +
  10 +services:
  11 + redis:
  12 + image: redis:7.4-alpine
  13 + restart: unless-stopped
  14 + network_mode: host
  15 + # LiveKit / Ingress / Egress 共享同一 Redis 交换任务；只监听回环、不持久化
  16 + command: ["redis-server", "--bind", "127.0.0.1", "--port", "16379", "--save", "", "--appendonly", "no"]
  17 +
  18 + livekit:
  19 + image: livekit/livekit-server:v1.13.10
  20 + restart: unless-stopped
  21 + network_mode: host
  22 + depends_on:
  23 + - redis
  24 + environment:
  25 + LIVEKIT_CONFIG: |
  26 + port: 7880
  27 + bind_addresses:
  28 + - "${MP_LIVEKIT_BIND_ADDRESS:-127.0.0.1}"
  29 + rtc:
  30 + tcp_port: 7881
  31 + udp_port: 7882
  32 + use_external_ip: false
  33 + node_ip: ${MP_LIVEKIT_NODE_IP:-127.0.0.1}
  34 + redis:
  35 + address: 127.0.0.1:16379
  36 + keys:
  37 + ${LIVEKIT_API_KEY:?需在 .env 配置 LIVEKIT_API_KEY}: ${LIVEKIT_API_SECRET:?需在 .env 配置 LIVEKIT_API_SECRET}
  38 + webhook:
  39 + api_key: ${LIVEKIT_API_KEY}
  40 + urls:
  41 + - ${MP_LIVEKIT_WEBHOOK_URL:-http://127.0.0.1:48080/admin-api/im/livekit/webhook}
  42 + ingress:
  43 + rtmp_base_url: rtmp://${MP_INGRESS_RTMP_PUBLIC_HOST:-127.0.0.1}:1935/x
  44 + room:
  45 + empty_timeout: 300
  46 + departure_timeout: 20
  47 + logging:
  48 + level: info
  49 +
  50 + ingress:
  51 + image: livekit/ingress:v1.5.0
  52 + restart: unless-stopped
  53 + network_mode: host
  54 + depends_on:
  55 + - redis
  56 + - livekit
  57 + environment:
  58 + INGRESS_CONFIG_BODY: |
  59 + api_key: ${LIVEKIT_API_KEY}
  60 + api_secret: ${LIVEKIT_API_SECRET}
  61 + ws_url: ws://127.0.0.1:7880
  62 + redis:
  63 + address: 127.0.0.1:16379
  64 + rtmp_port: 1935
  65 + whip_port: 18080
  66 + http_relay_port: 19090
  67 + health_port: 18081
  68 + rtc_config:
  69 + udp_port: 7885
  70 + use_external_ip: false
  71 + logging:
  72 + level: info
  73 +
  74 + egress:
  75 + image: livekit/egress:v1.15.0
  76 + restart: unless-stopped
  77 + network_mode: host
  78 + # 官方运行方式要求；Track Composite 不启动 Chrome，但保持与官方镜像文档一致
  79 + cap_add:
  80 + - SYS_ADMIN
  81 + depends_on:
  82 + - redis
  83 + - livekit
  84 + environment:
  85 + EGRESS_CONFIG_BODY: |
  86 + api_key: ${LIVEKIT_API_KEY}
  87 + api_secret: ${LIVEKIT_API_SECRET}
  88 + ws_url: ws://127.0.0.1:7880
  89 + redis:
  90 + address: 127.0.0.1:16379
  91 + health_port: 18082
  92 + template_port: 17980
  93 + log_level: info
  94 +
  95 + srs:
  96 + image: ossrs/srs:6
  97 + restart: unless-stopped
  98 + network_mode: host
  99 + volumes:
  100 + - ./srs.conf:/usr/local/srs/conf/mp-gateway.conf:ro
  101 + command: ["./objs/srs", "-c", "conf/mp-gateway.conf"]
  102 +
  103 + mp-gateway:
  104 + image: node:22-alpine
  105 + # 未显式开启时网关打印说明后退出，不自动重启
  106 + restart: "no"
  107 + network_mode: host
  108 + depends_on:
  109 + - livekit
  110 + - ingress
  111 + - egress
  112 + - srs
  113 + working_dir: /app
  114 + volumes:
  115 + - ./src:/app/src:ro
  116 + command: ["node", "src/server.mjs"]
  117 + environment:
  118 + MP_GATEWAY_ENABLED: ${MP_GATEWAY_ENABLED:-false}
  119 + MP_GATEWAY_API_HOST: ${MP_GATEWAY_API_HOST:-127.0.0.1}
  120 + MP_GATEWAY_API_PORT: 8090
  121 + MP_GATEWAY_HOOK_HOST: 127.0.0.1
  122 + MP_GATEWAY_HOOK_PORT: 8091
  123 + MP_GATEWAY_LIVEKIT_HTTP_URL: http://127.0.0.1:7880
  124 + MP_GATEWAY_LIVEKIT_API_KEY: ${LIVEKIT_API_KEY}
  125 + MP_GATEWAY_LIVEKIT_API_SECRET: ${LIVEKIT_API_SECRET}
  126 + # SRS 主机与端口固定为本 compose 内的实例，只允许通过 MP_GATEWAY_SRS_APP 调整专用 app 后缀
  127 + MP_GATEWAY_SRS_PUBLISH_BASE_URL: rtmp://127.0.0.1:1936/${MP_GATEWAY_SRS_APP:-yudao-mp-gw}
  128 + MP_GATEWAY_SRS_PLAY_BASE_URL: rtmp://${MP_SRS_PLAY_PUBLIC_HOST:-127.0.0.1}:1936/${MP_GATEWAY_SRS_APP:-yudao-mp-gw}
  129 + # 以下可调参数未配置时为空串，由网关按 README 默认值处理
  130 + MP_GATEWAY_POLL_INTERVAL_MS: ${MP_GATEWAY_POLL_INTERVAL_MS:-}
  131 + MP_GATEWAY_LEASE_TTL_MS: ${MP_GATEWAY_LEASE_TTL_MS:-}
  132 + MP_GATEWAY_MAX_SESSION_MS: ${MP_GATEWAY_MAX_SESSION_MS:-}
  133 + MP_GATEWAY_MAX_SESSIONS: ${MP_GATEWAY_MAX_SESSIONS:-}
  134 + MP_GATEWAY_MAX_REMOTES_PER_SESSION: ${MP_GATEWAY_MAX_REMOTES_PER_SESSION:-}
  135 + MP_GATEWAY_MAX_EGRESS: ${MP_GATEWAY_MAX_EGRESS:-}
  136 + MP_GATEWAY_EGRESS_RETRY_MS: ${MP_GATEWAY_EGRESS_RETRY_MS:-}
  137 + MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS: ${MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS:-}
  138 + MP_GATEWAY_RELEASE_RETRY_MS: ${MP_GATEWAY_RELEASE_RETRY_MS:-}
  139 + MP_GATEWAY_CLEANUP_RETRY_MS: ${MP_GATEWAY_CLEANUP_RETRY_MS:-}
  140 + MP_GATEWAY_REQUEST_TIMEOUT_MS: ${MP_GATEWAY_REQUEST_TIMEOUT_MS:-}
  141 + MP_GATEWAY_CLOCK_SKEW_SEC: ${MP_GATEWAY_CLOCK_SKEW_SEC:-}
  142 + MP_GATEWAY_EGRESS_WIDTH: ${MP_GATEWAY_EGRESS_WIDTH:-}
  143 + MP_GATEWAY_EGRESS_HEIGHT: ${MP_GATEWAY_EGRESS_HEIGHT:-}
  144 + MP_GATEWAY_EGRESS_FRAMERATE: ${MP_GATEWAY_EGRESS_FRAMERATE:-}
  145 + MP_GATEWAY_EGRESS_VIDEO_KBPS: ${MP_GATEWAY_EGRESS_VIDEO_KBPS:-}
  146 + MP_GATEWAY_EGRESS_AUDIO_KBPS: ${MP_GATEWAY_EGRESS_AUDIO_KBPS:-}
  147 + MP_GATEWAY_EGRESS_KEYFRAME_SEC: ${MP_GATEWAY_EGRESS_KEYFRAME_SEC:-}
script/livekit-poc/mp-gateway/package.json 0 → 100644
  View file @8aed362
  1 +{
  2 + "name": "yudao-im-mp-rtc-gateway",
  3 + "version": "0.1.0",
  4 + "private": true,
  5 + "description": "IM 微信小程序 RTMP ↔ LiveKit 网关试验：默认关闭，秒级延迟，仅供验证",
  6 + "type": "module",
  7 + "engines": {
  8 + "node": ">=22"
  9 + },
  10 + "scripts": {
  11 + "start": "node src/server.mjs"
  12 + }
  13 +}
script/livekit-poc/mp-gateway/src/config.mjs 0 → 100644
  View file @8aed362
  1 +// 小程序 RTMP 网关配置：全部来自环境变量；密钥只保存在内存中，不打印、不回显
  2 +
  3 +/**
  4 + * 网关专用 SRS app 命名空间
  5 + * <p>
  6 + * LiveKit 返回的 EgressInfo 会把 RTMP 地址最后一段（流名与查询参数）脱敏，只剩 scheme / host / port / app 可信，
  7 + * 启动回收据此识别本网关的残留 Egress，因此该 app 只能给网关使用
  8 + */
  9 +export const GATEWAY_SRS_APP_PATTERN = /^yudao-mp-gw(?:-[\w-]+)?$/
  10 +
  11 +/**
  12 + * 读取整数配置
  13 + *
  14 + * @param {Record<string, string | undefined>} env 环境变量
  15 + * @param {string} name 变量名
  16 + * @param {number} fallback 默认值
  17 + * @param {number} min 最小值
  18 + * @param {number} max 最大值
  19 + * @returns {number} 配置值
  20 + */
  21 +function readInt(env, name, fallback, min, max) {
  22 + const raw = env[name]
  23 + if (raw === undefined || raw === '') {
  24 + return fallback
  25 + }
  26 + const value = Number(raw)
  27 + if (!Number.isInteger(value) || value < min || value > max) {
  28 + throw new Error(`${name} 需为 ${min}~${max} 的整数`)
  29 + }
  30 + return value
  31 +}
  32 +
  33 +/**
  34 + * 读取 URL 配置并校验协议
  35 + *
  36 + * @param {Record<string, string | undefined>} env 环境变量
  37 + * @param {string} name 变量名
  38 + * @param {string} fallback 默认值
  39 + * @param {string[]} protocols 允许的协议，如 ['http:']
  40 + * @returns {string} 去掉末尾斜杠的 URL
  41 + */
  42 +function readUrl(env, name, fallback, protocols) {
  43 + const raw = env[name] || fallback
  44 + let url
  45 + try {
  46 + url = new URL(raw)
  47 + } catch {
  48 + throw new Error(`${name} 不是合法 URL`)
  49 + }
  50 + if (!protocols.includes(url.protocol) || url.search || url.hash || url.username || url.password) {
  51 + throw new Error(`${name} 仅支持 ${protocols.join('/')} 且不能带查询参数或账号`)
  52 + }
  53 + return raw.replace(/\/+$/, '')
  54 +}
  55 +
  56 +/**
  57 + * 读取必填字符串配置
  58 + *
  59 + * @param {Record<string, string | undefined>} env 环境变量
  60 + * @param {string} name 变量名
  61 + * @param {number} minLength 最小长度
  62 + * @returns {string} 配置值
  63 + */
  64 +function readRequired(env, name, minLength) {
  65 + const value = env[name]
  66 + if (!value || value.length < minLength) {
  67 + throw new Error(`${name} 未配置或长度不足 ${minLength}`)
  68 + }
  69 + return value
  70 +}
  71 +
  72 +/**
  73 + * 加载网关配置；未显式开启时拒绝启动
  74 + *
  75 + * @param {Record<string, string | undefined>} env 环境变量
  76 + * @returns {Readonly<Record<string, any>>} 只读配置
  77 + */
  78 +export function loadConfig(env = process.env) {
  79 + if (env.MP_GATEWAY_ENABLED !== 'true') {
  80 + throw new Error('MP_GATEWAY_ENABLED 未设置为 true：小程序网关为默认关闭的试验功能')
  81 + }
  82 + const srsPublishBaseUrl = readUrl(env, 'MP_GATEWAY_SRS_PUBLISH_BASE_URL', 'rtmp://127.0.0.1:1936/yudao-mp-gw', ['rtmp:'])
  83 + const srsPlayBaseUrl = readUrl(env, 'MP_GATEWAY_SRS_PLAY_BASE_URL', 'rtmp://127.0.0.1:1936/yudao-mp-gw', ['rtmp:'])
  84 + const srsApp = new URL(srsPublishBaseUrl).pathname.replace(/^\/+/, '')
  85 + if (!GATEWAY_SRS_APP_PATTERN.test(srsApp) || new URL(srsPlayBaseUrl).pathname.replace(/^\/+/, '') !== srsApp) {
  86 + throw new Error('MP_GATEWAY_SRS_PUBLISH_BASE_URL 与 MP_GATEWAY_SRS_PLAY_BASE_URL 需使用同一个网关专用 app（yudao-mp-gw 或 yudao-mp-gw-<后缀>）')
  87 + }
  88 + const pollIntervalMs = readInt(env, 'MP_GATEWAY_POLL_INTERVAL_MS', 2000, 500, 10000)
  89 + return Object.freeze({
  90 + apiHost: env.MP_GATEWAY_API_HOST || '127.0.0.1', // 小程序 API 监听地址；公网部署放在 HTTPS 反向代理之后
  91 + apiPort: readInt(env, 'MP_GATEWAY_API_PORT', 8090, 1, 65535), // 小程序 API 端口
  92 + hookHost: env.MP_GATEWAY_HOOK_HOST || '127.0.0.1', // SRS 回调监听地址，只允许 SRS 访问
  93 + hookPort: readInt(env, 'MP_GATEWAY_HOOK_PORT', 8091, 1, 65535), // SRS 回调端口
  94 + livekitHttpUrl: readUrl(env, 'MP_GATEWAY_LIVEKIT_HTTP_URL', 'http://127.0.0.1:7880', ['http:', 'https:']), // LiveKit Server API 地址，只取自部署配置
  95 + apiKey: readRequired(env, 'MP_GATEWAY_LIVEKIT_API_KEY', 1), // 与后端签发房间 Token 的 apiKey 一致
  96 + apiSecret: readRequired(env, 'MP_GATEWAY_LIVEKIT_API_SECRET', 32), // 与后端签发房间 Token 的 apiSecret 一致
  97 + srsPublishBaseUrl, // Egress 推流到 SRS 的内部地址
  98 + srsPlayBaseUrl, // 小程序 live-player 拉流的外部地址
  99 + srsApp, // SRS app 名，回调时校验
  100 + pollIntervalMs, // 建议小程序轮询间隔
  101 + syncIntervalMs: pollIntervalMs, // 网关同步房间成员的间隔
  102 + leaseTtlMs: readInt(env, 'MP_GATEWAY_LEASE_TTL_MS', 20000, pollIntervalMs * 3, 120000), // 会话租约：超过该时间未轮询即释放
  103 + maxSessionMs: readInt(env, 'MP_GATEWAY_MAX_SESSION_MS', 2 * 60 * 60 * 1000, 60000, 24 * 60 * 60 * 1000), // 单会话最长存活时间
  104 + maxSessions: readInt(env, 'MP_GATEWAY_MAX_SESSIONS', 4, 1, 100), // 全局最大小程序会话数（每个会话 1 路 Ingress 转码）
  105 + maxRemotesPerSession: readInt(env, 'MP_GATEWAY_MAX_REMOTES_PER_SESSION', 4, 1, 16), // 单会话最多拉取的远端路数
  106 + maxEgress: readInt(env, 'MP_GATEWAY_MAX_EGRESS', 8, 1, 200), // 全局最大 Egress 路数
  107 + egressRetryMs: readInt(env, 'MP_GATEWAY_EGRESS_RETRY_MS', 10000, 1000, 120000), // 单路 Egress 失败后的冷却时间
  108 + egressPublishTimeoutMs: readInt(env, 'MP_GATEWAY_EGRESS_PUBLISH_TIMEOUT_MS', 20000, 5000, 120000), // Egress 启动后等待 SRS on_publish 的最长时间
  109 + cleanupRetryMs: readInt(env, 'MP_GATEWAY_CLEANUP_RETRY_MS', 5000, 1000, 60000), // 启动回收失败后的重试间隔
  110 + releaseRetryMs: readInt(env, 'MP_GATEWAY_RELEASE_RETRY_MS', 5000, 1000, 60000), // Ingress / Egress 释放失败后的重试间隔；确认释放前继续占用配额
  111 + requestTimeoutMs: readInt(env, 'MP_GATEWAY_REQUEST_TIMEOUT_MS', 10000, 1000, 60000), // 调用 LiveKit API 超时
  112 + clockSkewSec: readInt(env, 'MP_GATEWAY_CLOCK_SKEW_SEC', 30, 0, 300), // 校验 nbf / exp 允许的时钟偏差
  113 + egressEncoding: Object.freeze({ // live-player 兼容的 H264 Baseline + AAC 编码参数
  114 + width: readInt(env, 'MP_GATEWAY_EGRESS_WIDTH', 640, 160, 1920),
  115 + height: readInt(env, 'MP_GATEWAY_EGRESS_HEIGHT', 360, 120, 1080),
  116 + framerate: readInt(env, 'MP_GATEWAY_EGRESS_FRAMERATE', 15, 5, 30),
  117 + videoBitrate: readInt(env, 'MP_GATEWAY_EGRESS_VIDEO_KBPS', 600, 100, 4000),
  118 + audioBitrate: readInt(env, 'MP_GATEWAY_EGRESS_AUDIO_KBPS', 64, 16, 256),
  119 + keyFrameInterval: readInt(env, 'MP_GATEWAY_EGRESS_KEYFRAME_SEC', 1, 1, 10),
  120 + }),
  121 + })
  122 +}
script/livekit-poc/mp-gateway/src/jwt.mjs 0 → 100644
  View file @8aed362
  1 +// LiveKit 房间 Token 校验与网关管理 Token 签发：仅支持 HS256，与后端 LiveKitClient.signJoinToken 保持一致
  2 +import { createHmac, timingSafeEqual } from 'node:crypto'
  3 +
  4 +const MAX_TOKEN_LENGTH = 4096 // 拒绝超长 Token，避免无意义的解析开销
  5 +const USER_IDENTITY_PATTERN = /^[1-9]\d{0,18}$/ // 业务 identity：数字 userId，与后端 buildIdentity 一致
  6 +const MAX_ROOM_LENGTH = 128 // 房间名长度上限
  7 +
  8 +/** 房间 Token 校验失败；status 为返回给客户端的 HTTP 状态码 */
  9 +export class TokenError extends Error {
  10 + /**
  11 + * @param {string} code 错误码
  12 + * @param {string} message 错误说明（不包含 Token 内容）
  13 + * @param {number} status HTTP 状态码
  14 + */
  15 + constructor(code, message, status = 401) {
  16 + super(message)
  17 + this.code = code
  18 + this.status = status
  19 + }
  20 +}
  21 +
  22 +/**
  23 + * 解码 JWT 的 JSON 段
  24 + *
  25 + * @param {string} segment base64url 段
  26 + * @returns {Record<string, any>} JSON 对象
  27 + */
  28 +function decodeSegment(segment) {
  29 + try {
  30 + const value = JSON.parse(Buffer.from(segment, 'base64url').toString('utf8'))
  31 + if (value && typeof value === 'object' && !Array.isArray(value)) {
  32 + return value
  33 + }
  34 + } catch {
  35 + // 统一按格式错误处理
  36 + }
  37 + throw new TokenError('invalid_token', 'Token 格式错误')
  38 +}
  39 +
  40 +/**
  41 + * 计算 HS256 签名
  42 + *
  43 + * @param {string} signingInput header.payload
  44 + * @param {string} secret 签名密钥
  45 + * @returns {Buffer} 签名字节
  46 + */
  47 +function sign(signingInput, secret) {
  48 + return createHmac('sha256', secret).update(signingInput).digest()
  49 +}
  50 +
  51 +/**
  52 + * 校验后端签发的 LiveKit 房间 Token，并以 Token 内的房间与身份为准
  53 + *
  54 + * @param {string | undefined} token 房间 Token
  55 + * @param {{ apiKey: string, apiSecret: string, clockSkewSec: number }} config 网关配置
  56 + * @param {number} nowSec 当前时间（秒），便于测试
  57 + * @returns {{ room: string, identity: string, name: string, exp: number }} 授权信息
  58 + */
  59 +export function verifyRoomToken(token, config, nowSec = Math.floor(Date.now() / 1000)) {
  60 + if (typeof token !== 'string' || !token || token.length > MAX_TOKEN_LENGTH) {
  61 + throw new TokenError('missing_token', '缺少房间 Token')
  62 + }
  63 + const parts = token.split('.')
  64 + if (parts.length !== 3 || parts.some(part => !/^[\w-]+$/.test(part))) {
  65 + throw new TokenError('invalid_token', 'Token 格式错误')
  66 + }
  67 + const header = decodeSegment(parts[0])
  68 + if (header.alg !== 'HS256' || (header.typ !== undefined && header.typ !== 'JWT')) {
  69 + throw new TokenError('invalid_token', '仅支持 HS256 Token')
  70 + }
  71 + const expected = sign(`${parts[0]}.${parts[1]}`, config.apiSecret)
  72 + const actual = Buffer.from(parts[2], 'base64url')
  73 + if (actual.length !== expected.length || !timingSafeEqual(actual, expected)) {
  74 + throw new TokenError('invalid_signature', 'Token 签名无效')
  75 + }
  76 + const payload = decodeSegment(parts[1])
  77 + if (payload.iss !== config.apiKey) {
  78 + throw new TokenError('invalid_issuer', 'Token 签发方不匹配')
  79 + }
  80 + if (typeof payload.exp !== 'number' || payload.exp + config.clockSkewSec <= nowSec) {
  81 + throw new TokenError('token_expired', 'Token 已过期')
  82 + }
  83 + if (typeof payload.nbf !== 'number' || payload.nbf - config.clockSkewSec > nowSec) {
  84 + throw new TokenError('token_not_active', 'Token 尚未生效')
  85 + }
  86 + if (typeof payload.sub !== 'string' || !USER_IDENTITY_PATTERN.test(payload.sub)) {
  87 + throw new TokenError('invalid_identity', 'Token 身份不是业务用户', 403)
  88 + }
  89 + const video = payload.video
  90 + if (!video || typeof video !== 'object'
  91 + || video.roomJoin !== true
  92 + || typeof video.room !== 'string' || !video.room || video.room.length > MAX_ROOM_LENGTH
  93 + || video.canPublish !== true || video.canSubscribe !== true
  94 + || video.hidden === true || video.recorder === true) {
  95 + throw new TokenError('insufficient_grant', 'Token 缺少进房、发布或订阅授权', 403)
  96 + }
  97 + if (Array.isArray(video.canPublishSources)
  98 + && !(video.canPublishSources.includes('camera') && video.canPublishSources.includes('microphone'))) {
  99 + throw new TokenError('insufficient_grant', 'Token 不允许发布摄像头和麦克风', 403)
  100 + }
  101 + return {
  102 + room: video.room,
  103 + identity: payload.sub,
  104 + name: typeof payload.name === 'string' && payload.name ? payload.name.slice(0, 128) : payload.sub,
  105 + exp: payload.exp,
  106 + }
  107 +}
  108 +
  109 +/**
  110 + * 签发调用 LiveKit Server API 的短期管理 Token
  111 + *
  112 + * @param {{ apiKey: string, apiSecret: string }} config 网关配置
  113 + * @param {Record<string, any>} video 授权，如 { roomAdmin: true, room }
  114 + * @param {number} ttlSec 有效期（秒）
  115 + * @returns {string} JWT
  116 + */
  117 +export function signAdminToken(config, video, ttlSec = 300) {
  118 + const nowSec = Math.floor(Date.now() / 1000)
  119 + const header = Buffer.from(JSON.stringify({ alg: 'HS256', typ: 'JWT' })).toString('base64url')
  120 + const payload = Buffer.from(JSON.stringify({
  121 + iss: config.apiKey,
  122 + sub: 'yudao-mp-gateway',
  123 + nbf: nowSec - 5,
  124 + exp: nowSec + ttlSec,
  125 + video,
  126 + })).toString('base64url')
  127 + return `${header}.${payload}.${sign(`${header}.${payload}`, config.apiSecret).toString('base64url')}`
  128 +}
script/livekit-poc/mp-gateway/src/livekit-api.mjs 0 → 100644
  View file @8aed362
  1 +// LiveKit Server API（Twirp JSON）最小客户端：只调用网关需要的 Room / Ingress / Egress 接口
  2 +import { signAdminToken } from './jwt.mjs'
  3 +
  4 +const PARTICIPANT_KINDS = ['STANDARD', 'INGRESS', 'EGRESS', 'SIP', 'AGENT'] // ParticipantInfo.Kind 枚举
  5 +const PARTICIPANT_STATES = ['JOINING', 'JOINED', 'ACTIVE', 'DISCONNECTED'] // ParticipantInfo.State 枚举
  6 +const TRACK_TYPES = ['AUDIO', 'VIDEO', 'DATA'] // TrackType 枚举
  7 +const TRACK_SOURCES = ['UNKNOWN', 'CAMERA', 'MICROPHONE', 'SCREEN_SHARE', 'SCREEN_SHARE_AUDIO'] // TrackSource 枚举
  8 +const TWIRP_CODE_PATTERN = /^[a-z_]{1,32}$/ // Twirp 错误码格式，如 not_found；不符合时按 unknown 处理
  9 +
  10 +/** LiveKit API 调用失败；message 只含方法、HTTP 状态与错误码，不含上游原文、Token 或推流地址 */
  11 +export class LiveKitApiError extends Error {
  12 + /**
  13 + * @param {number} status HTTP 状态码
  14 + * @param {string} code Twirp 错误码
  15 + * @param {string} message 错误说明
  16 + */
  17 + constructor(status, code, message) {
  18 + super(message)
  19 + this.status = status
  20 + this.code = code
  21 + }
  22 +}
  23 +
  24 +/**
  25 + * 读取 protobuf JSON 字段；兼容 snake_case 与 lowerCamelCase 输出
  26 + *
  27 + * @param {Record<string, any>} value 对象
  28 + * @param {string} snakeName snake_case 字段名
  29 + * @returns {any} 字段值
  30 + */
  31 +function field(value, snakeName) {
  32 + if (value[snakeName] !== undefined) {
  33 + return value[snakeName]
  34 + }
  35 + return value[snakeName.replace(/_([a-z])/g, (_, char) => char.toUpperCase())]
  36 +}
  37 +
  38 +/**
  39 + * 读取枚举名；兼容名称与数字输出，缺省值按 protobuf 语义取第 0 项
  40 + *
  41 + * @param {string | number | undefined} value 枚举值
  42 + * @param {string[]} names 枚举名称表
  43 + * @returns {string} 枚举名
  44 + */
  45 +function enumName(value, names) {
  46 + if (typeof value === 'number') {
  47 + return names[value] ?? String(value)
  48 + }
  49 + return value || names[0]
  50 +}
  51 +
  52 +/**
  53 + * 标准化房间参与者
  54 + *
  55 + * @param {Record<string, any>} participant ParticipantInfo
  56 + * @returns {{ sid: string, identity: string, kind: string, state: string, joinedAt: number, tracks: Array<{ sid: string, type: string, source: string, muted: boolean }> }} 参与者
  57 + */
  58 +function normalizeParticipant(participant) {
  59 + return {
  60 + sid: participant.sid,
  61 + identity: participant.identity || '',
  62 + kind: enumName(participant.kind, PARTICIPANT_KINDS),
  63 + state: enumName(participant.state, PARTICIPANT_STATES),
  64 + joinedAt: Number(field(participant, 'joined_at') || 0),
  65 + tracks: (participant.tracks || []).map(track => ({
  66 + sid: track.sid,
  67 + type: enumName(track.type, TRACK_TYPES),
  68 + source: enumName(track.source, TRACK_SOURCES),
  69 + muted: track.muted === true,
  70 + })),
  71 + }
  72 +}
  73 +
  74 +/**
  75 + * 创建 LiveKit API 客户端；服务地址只来自部署配置，不接受客户端传入
  76 + *
  77 + * @param {{ livekitHttpUrl: string, apiKey: string, apiSecret: string, requestTimeoutMs: number }} config 网关配置
  78 + * @returns {Record<string, Function>} API 方法
  79 + */
  80 +export function createLiveKitApi(config) {
  81 + /**
  82 + * 调用 Twirp 接口
  83 + *
  84 + * @param {string} service 服务名，如 RoomService
  85 + * @param {string} method 方法名
  86 + * @param {Record<string, any>} body 请求体
  87 + * @param {Record<string, any>} video 管理 Token 授权
  88 + * @returns {Promise<Record<string, any>>} 响应体
  89 + */
  90 + async function call(service, method, body, video) {
  91 + let response
  92 + let text
  93 + try {
  94 + response = await fetch(`${config.livekitHttpUrl}/twirp/livekit.${service}/${method}`, {
  95 + method: 'POST',
  96 + headers: {
  97 + 'Content-Type': 'application/json',
  98 + 'Authorization': `Bearer ${signAdminToken(config, video)}`,
  99 + },
  100 + body: JSON.stringify(body),
  101 + signal: AbortSignal.timeout(config.requestTimeoutMs),
  102 + })
  103 + text = await response.text()
  104 + } catch (error) {
  105 + // 网络错误 / 超时只保留类型，不透传底层信息
  106 + const code = error?.name === 'TimeoutError' ? 'timeout' : 'network_error'
  107 + throw new LiveKitApiError(0, code, `${service}.${method} 请求失败：${code}`)
  108 + }
  109 + let data = {}
  110 + try {
  111 + data = text ? JSON.parse(text) : {}
  112 + } catch {
  113 + data = {}
  114 + }
  115 + if (!data || typeof data !== 'object' || Array.isArray(data)) { // null、数组、数字等非对象 JSON 视为空响应，交由状态码与资源编号校验处理
  116 + data = {}
  117 + }
  118 + if (!response.ok) {
  119 + // 上游 msg 可能回显请求中的 RTMP 地址等凭证，只保留方法、HTTP 状态与规范化的 Twirp 错误码
  120 + const code = typeof data.code === 'string' && TWIRP_CODE_PATTERN.test(data.code) ? data.code : 'unknown'
  121 + throw new LiveKitApiError(response.status, code, `${service}.${method} 失败：HTTP ${response.status} ${code}`)
  122 + }
  123 + return data
  124 + }
  125 +
  126 + /**
  127 + * 判断是否为「资源不存在 / 已结束」类错误，释放资源时可忽略
  128 + *
  129 + * @param {unknown} error 错误
  130 + * @returns {boolean} 是否可忽略
  131 + */
  132 + function isGone(error) {
  133 + return error instanceof LiveKitApiError
  134 + && (error.status === 404 || error.code === 'not_found' || error.code === 'failed_precondition')
  135 + }
  136 +
  137 + return {
  138 + /**
  139 + * 查询房间参与者
  140 + *
  141 + * @param {string} room 房间名
  142 + * @returns {Promise<ReturnType<typeof normalizeParticipant>[] | null>} 参与者；房间不存在返回 null
  143 + */
  144 + async listParticipants(room) {
  145 + try {
  146 + const data = await call('RoomService', 'ListParticipants', { room }, { roomAdmin: true, room })
  147 + return (data.participants || []).map(normalizeParticipant)
  148 + } catch (error) {
  149 + if (error instanceof LiveKitApiError && (error.status === 404 || error.code === 'not_found')) {
  150 + return null
  151 + }
  152 + throw error
  153 + }
  154 + },
  155 +
  156 + /**
  157 + * 创建 RTMP Ingress；参与者身份直接使用 Token 内的 userId，保留同 identity 互踢语义
  158 + *
  159 + * @param {{ room: string, identity: string, name: string, resourceName: string }} options 参数
  160 + * @returns {Promise<{ ingressId: string, url: string, streamKey: string }>} Ingress 信息
  161 + */
  162 + async createIngress(options) {
  163 + const data = await call('Ingress', 'CreateIngress', {
  164 + input_type: 'RTMP_INPUT',
  165 + name: options.resourceName,
  166 + room_name: options.room,
  167 + participant_identity: options.identity,
  168 + participant_name: options.name,
  169 + enable_transcoding: true,
  170 + }, { ingressAdmin: true })
  171 + const ingressId = field(data, 'ingress_id')
  172 + if (typeof ingressId !== 'string' || !ingressId) { // 可能已创建但拿不到编号，交由调用方按结果未知处理
  173 + throw new LiveKitApiError(502, 'invalid_response', 'Ingress.CreateIngress 响应缺少资源编号')
  174 + }
  175 + return {
  176 + ingressId,
  177 + url: data.url || '',
  178 + streamKey: field(data, 'stream_key') || '',
  179 + }
  180 + },
  181 +
  182 + /**
  183 + * 删除 Ingress；不存在时忽略
  184 + *
  185 + * @param {string} ingressId Ingress 编号
  186 + * @returns {Promise<void>}
  187 + */
  188 + async deleteIngress(ingressId) {
  189 + try {
  190 + await call('Ingress', 'DeleteIngress', { ingress_id: ingressId }, { ingressAdmin: true })
  191 + } catch (error) {
  192 + if (!isGone(error)) {
  193 + throw error
  194 + }
  195 + }
  196 + },
  197 +
  198 + /**
  199 + * 列出全部 Ingress，用于启动时回收本网关残留资源
  200 + *
  201 + * @returns {Promise<Array<{ ingressId: string, name: string }>>} Ingress 列表
  202 + */
  203 + async listIngress() {
  204 + const data = await call('Ingress', 'ListIngress', {}, { ingressAdmin: true })
  205 + return (data.items || []).map(item => ({ ingressId: field(item, 'ingress_id'), name: item.name || '' }))
  206 + },
  207 +
  208 + /**
  209 + * 启动单个远端参与者的 Track Composite Egress，推 RTMP 到 SRS
  210 + *
  211 + * @param {{ room: string, audioTrackId?: string, videoTrackId?: string, url: string, encoding: Record<string, number> }} options 参数
  212 + * @returns {Promise<{ egressId: string }>} Egress 信息
  213 + */
  214 + async startTrackCompositeEgress(options) {
  215 + const data = await call('Egress', 'StartTrackCompositeEgress', {
  216 + room_name: options.room,
  217 + audio_track_id: options.audioTrackId || '',
  218 + video_track_id: options.videoTrackId || '',
  219 + stream_outputs: [{ protocol: 'RTMP', urls: [options.url] }],
  220 + advanced: {
  221 + width: options.encoding.width,
  222 + height: options.encoding.height,
  223 + framerate: options.encoding.framerate,
  224 + audio_codec: 'AAC',
  225 + audio_bitrate: options.encoding.audioBitrate,
  226 + video_codec: 'H264_BASELINE',
  227 + video_bitrate: options.encoding.videoBitrate,
  228 + key_frame_interval: options.encoding.keyFrameInterval,
  229 + },
  230 + }, { roomRecord: true })
  231 + const egressId = field(data, 'egress_id')
  232 + if (typeof egressId !== 'string' || !egressId) { // 可能已创建但拿不到编号，交由调用方按结果未知处理
  233 + throw new LiveKitApiError(502, 'invalid_response', 'Egress.StartTrackCompositeEgress 响应缺少资源编号')
  234 + }
  235 + return { egressId }
  236 + },
  237 +
  238 + /**
  239 + * 停止 Egress；已结束或不存在时忽略
  240 + *
  241 + * @param {string} egressId Egress 编号
  242 + * @returns {Promise<void>}
  243 + */
  244 + async stopEgress(egressId) {
  245 + try {
  246 + await call('Egress', 'StopEgress', { egress_id: egressId }, { roomRecord: true })
  247 + } catch (error) {
  248 + if (!isGone(error)) {
  249 + throw error
  250 + }
  251 + }
  252 + },
  253 +
  254 + /**
  255 + * 列出活跃 Egress
  256 + *
  257 + * @param {string | undefined} room 房间名；为空时列出全部
  258 + * @returns {Promise<Array<{ egressId: string, raw: Record<string, any> }>>} 活跃 Egress
  259 + */
  260 + async listActiveEgress(room) {
  261 + const data = await call('Egress', 'ListEgress', room ? { room_name: room, active: true } : { active: true }, { roomRecord: true })
  262 + return (data.items || []).map(item => ({ egressId: field(item, 'egress_id'), raw: item }))
  263 + },
  264 + }
  265 +}
script/livekit-poc/mp-gateway/src/server.mjs 0 → 100644
  View file @8aed362
  1 +// 小程序 RTMP 网关入口：对小程序提供会话 API，对 SRS 提供推拉流鉴权回调（两个端口分开监听）
  2 +import { createServer } from 'node:http'
  3 +import { loadConfig } from './config.mjs'
  4 +import { TokenError, verifyRoomToken } from './jwt.mjs'
  5 +import { createLiveKitApi } from './livekit-api.mjs'
  6 +import { GatewayError, createSessionManager } from './session-manager.mjs'
  7 +
  8 +const MAX_BODY_BYTES = 4096 // 请求体上限；会话 API 与 SRS 回调都只有很小的 JSON
  9 +const SHUTDOWN_TIMEOUT_MS = 15000 // 退出时等待资源释放的最长时间
  10 +const SESSION_PATH_PATTERN = /^\/mp-rtc\/sessions\/([^/]+)$/ // 单个会话路径
  11 +
  12 +let orphansCleaned = false // 启动回收是否已完成；完成前拒绝创建会话，避免旧残留与新会话并存
  13 +
  14 +/** 日志：只输出事件与编号，调用方不得传入 Token、stream key、带凭证的 URL */
  15 +const logger = {
  16 + /**
  17 + * @param {string} message 事件
  18 + * @param {Record<string, unknown>} [fields] 附加字段
  19 + */
  20 + info(message, fields = {}) {
  21 + console.log(JSON.stringify({ time: new Date().toISOString(), level: 'info', message, ...fields }))
  22 + },
  23 + /**
  24 + * @param {string} message 事件
  25 + * @param {Record<string, unknown>} [fields] 附加字段
  26 + */
  27 + warn(message, fields = {}) {
  28 + console.warn(JSON.stringify({ time: new Date().toISOString(), level: 'warn', message, ...fields }))
  29 + },
  30 +}
  31 +
  32 +/**
  33 + * 写 JSON 响应
  34 + *
  35 + * @param {import('node:http').ServerResponse} res 响应
  36 + * @param {number} status 状态码
  37 + * @param {unknown} [body] 响应体；为空时不写 body
  38 + */
  39 +function sendJson(res, status, body) {
  40 + if (res.headersSent) {
  41 + return
  42 + }
  43 + const headers = { 'Cache-Control': 'no-store' }
  44 + if (body === undefined) {
  45 + res.writeHead(status, headers).end()
  46 + return
  47 + }
  48 + res.writeHead(status, { ...headers, 'Content-Type': 'application/json; charset=utf-8' }).end(JSON.stringify(body))
  49 +}
  50 +
  51 +/**
  52 + * 读取 JSON 请求体
  53 + *
  54 + * @param {import('node:http').IncomingMessage} req 请求
  55 + * @returns {Promise<Record<string, any>>} 请求体；空 body 返回空对象
  56 + */
  57 +async function readJson(req) {
  58 + const chunks = []
  59 + let size = 0
  60 + for await (const chunk of req) {
  61 + size += chunk.length
  62 + if (size > MAX_BODY_BYTES) {
  63 + throw new GatewayError(413, 'body_too_large', '请求体过大')
  64 + }
  65 + chunks.push(chunk)
  66 + }
  67 + if (size === 0) {
  68 + return {}
  69 + }
  70 + try {
  71 + const value = JSON.parse(Buffer.concat(chunks).toString('utf8'))
  72 + return value && typeof value === 'object' ? value : {}
  73 + } catch {
  74 + throw new GatewayError(400, 'invalid_json', '请求体不是合法 JSON')
  75 + }
  76 +}
  77 +
  78 +/**
  79 + * 读取 Bearer 凭证
  80 + *
  81 + * @param {import('node:http').IncomingMessage} req 请求
  82 + * @returns {string | undefined} 凭证
  83 + */
  84 +function readBearer(req) {
  85 + const header = req.headers.authorization
  86 + return typeof header === 'string' && header.startsWith('Bearer ') ? header.slice(7).trim() : undefined
  87 +}
  88 +
  89 +/**
  90 + * 错误转响应；未知错误不回显细节
  91 + *
  92 + * @param {import('node:http').ServerResponse} res 响应
  93 + * @param {unknown} error 错误
  94 + */
  95 +function sendError(res, error) {
  96 + if (error instanceof GatewayError || error instanceof TokenError) {
  97 + sendJson(res, error.status, { code: error.code, message: error.message })
  98 + return
  99 + }
  100 + logger.warn('请求处理失败', { error: error instanceof Error ? error.message : String(error) })
  101 + sendJson(res, 500, { code: 'internal_error', message: '网关内部错误' })
  102 +}
  103 +
  104 +/**
  105 + * 处理小程序会话 API
  106 + *
  107 + * @param {Readonly<Record<string, any>>} config 网关配置
  108 + * @param {ReturnType<typeof createSessionManager>} manager 会话管理器
  109 + * @param {import('node:http').IncomingMessage} req 请求
  110 + * @param {import('node:http').ServerResponse} res 响应
  111 + * @returns {Promise<void>}
  112 + */
  113 +async function handleApi(config, manager, req, res) {
  114 + const { pathname } = new URL(req.url || '/', 'http://gateway')
  115 + if (req.method === 'GET' && pathname === '/healthz') {
  116 + sendJson(res, orphansCleaned ? 200 : 503, { status: orphansCleaned ? 'ok' : 'cleanup_pending' })
  117 + return
  118 + }
  119 + if (req.method === 'POST' && pathname === '/mp-rtc/sessions') {
  120 + if (!orphansCleaned) {
  121 + throw new GatewayError(503, 'cleanup_pending', '网关正在回收上次残留的推流资源，请稍后再试')
  122 + }
  123 + await readJson(req)
  124 + const claims = verifyRoomToken(readBearer(req), config)
  125 + const { session, sessionToken } = await manager.createSession(claims)
  126 + // 小程序在创建过程中断开（页面关闭、超时）：响应无人接收，立即释放刚创建的资源
  127 + if (res.destroyed || !res.socket || res.socket.destroyed) {
  128 + await manager.closeSession(session, 'client_aborted')
  129 + return
  130 + }
  131 + sendJson(res, 201, {
  132 + sessionId: session.id,
  133 + sessionToken,
  134 + userId: Number(session.identity),
  135 + pushUrl: session.pushUrl,
  136 + pollIntervalMs: config.pollIntervalMs,
  137 + leaseTtlMs: config.leaseTtlMs,
  138 + expiresAt: session.expiresAt,
  139 + })
  140 + return
  141 + }
  142 + const matched = SESSION_PATH_PATTERN.exec(pathname)
  143 + if (matched && (req.method === 'GET' || req.method === 'DELETE')) {
  144 + const session = manager.authorize(matched[1], readBearer(req))
  145 + if (req.method === 'GET') {
  146 + sendJson(res, 200, manager.touch(session))
  147 + return
  148 + }
  149 + await manager.closeSession(session, 'client_closed')
  150 + sendJson(res, 204)
  151 + return
  152 + }
  153 + sendJson(res, 404, { code: 'not_found', message: '接口不存在' })
  154 +}
  155 +
  156 +/**
  157 + * 处理 SRS HTTP 回调；返回 code 0 允许，其它拒绝
  158 + *
  159 + * @param {ReturnType<typeof createSessionManager>} manager 会话管理器
  160 + * @param {import('node:http').IncomingMessage} req 请求
  161 + * @param {import('node:http').ServerResponse} res 响应
  162 + * @returns {Promise<void>}
  163 + */
  164 +async function handleHook(manager, req, res) {
  165 + const { pathname } = new URL(req.url || '/', 'http://gateway')
  166 + if (req.method !== 'POST') {
  167 + sendJson(res, 404, { code: 404 })
  168 + return
  169 + }
  170 + const hook = await readJson(req)
  171 + let allowed = false
  172 + if (pathname === '/srs/hooks/on_publish') {
  173 + allowed = manager.authorizePublish(hook)
  174 + } else if (pathname === '/srs/hooks/on_play') {
  175 + allowed = manager.authorizePlay(hook)
  176 + } else if (pathname === '/srs/hooks/on_unpublish') {
  177 + manager.handleUnpublish(hook)
  178 + allowed = true
  179 + } else {
  180 + sendJson(res, 404, { code: 404 })
  181 + return
  182 + }
  183 + if (!allowed) {
  184 + logger.warn('SRS 鉴权拒绝', { action: pathname.slice('/srs/hooks/'.length) })
  185 + }
  186 + sendJson(res, allowed ? 200 : 403, { code: allowed ? 0 : 403 })
  187 +}
  188 +
  189 +/**
  190 + * 启动 HTTP 服务
  191 + *
  192 + * @param {import('node:http').Server} server 服务
  193 + * @param {string} host 监听地址
  194 + * @param {number} port 端口
  195 + * @returns {Promise<void>}
  196 + */
  197 +function listen(server, host, port) {
  198 + return new Promise((resolve, reject) => {
  199 + server.once('error', reject)
  200 + server.listen(port, host, () => {
  201 + server.off('error', reject)
  202 + resolve()
  203 + })
  204 + })
  205 +}
  206 +
  207 +/** 启动网关 */
  208 +async function main() {
  209 + const config = loadConfig()
  210 + const liveKit = createLiveKitApi(config)
  211 + const manager = createSessionManager(config, liveKit, logger)
  212 + let stopping = false // 是否已开始退出
  213 + let cleanupTimer // 启动回收重试定时器
  214 + let cleanupTask // 正在执行的启动回收；退出时等待其结束
  215 +
  216 + /** 发起一次启动回收；退出后不再发起 */
  217 + function scheduleCleanup() {
  218 + cleanupTimer = undefined
  219 + if (stopping) {
  220 + return
  221 + }
  222 + cleanupTask = runCleanup().finally(() => {
  223 + cleanupTask = undefined
  224 + })
  225 + }
  226 +
  227 + /** 启动回收：失败后按间隔重试，直到成功或网关退出 */
  228 + async function runCleanup() {
  229 + try {
  230 + await manager.cleanupOrphans()
  231 + orphansCleaned = true
  232 + } catch (error) {
  233 + logger.warn(stopping ? '残留资源回收失败，网关退出中不再重试' : '残留资源回收失败，稍后重试；完成前拒绝创建会话',
  234 + { error: error.message })
  235 + if (!stopping) {
  236 + cleanupTimer = setTimeout(scheduleCleanup, config.cleanupRetryMs)
  237 + }
  238 + }
  239 + }
  240 +
  241 + const apiServer = createServer((req, res) => {
  242 + handleApi(config, manager, req, res).catch(error => sendError(res, error))
  243 + })
  244 + const hookServer = createServer((req, res) => {
  245 + handleHook(manager, req, res).catch(error => sendError(res, error))
  246 + })
  247 + apiServer.requestTimeout = config.requestTimeoutMs * 3
  248 + hookServer.requestTimeout = config.requestTimeoutMs
  249 + await listen(apiServer, config.apiHost, config.apiPort)
  250 + await listen(hookServer, config.hookHost, config.hookPort)
  251 + scheduleCleanup()
  252 + logger.info('小程序网关已启动（试验功能）', {
  253 + api: `${config.apiHost}:${config.apiPort}`,
  254 + hook: `${config.hookHost}:${config.hookPort}`,
  255 + livekit: config.livekitHttpUrl,
  256 + maxSessions: config.maxSessions,
  257 + maxEgress: config.maxEgress,
  258 + })
  259 +
  260 + /**
  261 + * 退出：停止接收请求并释放全部 Ingress / Egress
  262 + *
  263 + * @param {string} signal 信号名
  264 + */
  265 + const stop = (signal) => {
  266 + if (stopping) {
  267 + return
  268 + }
  269 + stopping = true
  270 + clearTimeout(cleanupTimer)
  271 + logger.info('网关退出中', { signal })
  272 + apiServer.close()
  273 + const timer = setTimeout(() => process.exit(1), SHUTDOWN_TIMEOUT_MS)
  274 + // 先等待进行中的启动回收（runCleanup 自身不会拒绝），再释放会话资源；启动回收未完成时残留资源未确认，不能报告正常退出
  275 + Promise.resolve(cleanupTask).then(() => manager.shutdown()).then(() => {
  276 + if (!orphansCleaned) {
  277 + logger.warn('网关退出时启动残留资源回收未完成')
  278 + return 1
  279 + }
  280 + return 0
  281 + }, (error) => {
  282 + logger.warn('网关退出时仍有资源未确认释放', { error: error.message })
  283 + return 1
  284 + }).then((code) => {
  285 + clearTimeout(timer)
  286 + hookServer.close()
  287 + process.exit(code)
  288 + })
  289 + }
  290 + process.on('SIGINT', () => stop('SIGINT'))
  291 + process.on('SIGTERM', () => stop('SIGTERM'))
  292 +}
  293 +
  294 +main().catch((error) => {
  295 + logger.warn('网关启动失败', { error: error.message })
  296 + process.exit(1)
  297 +})
script/livekit-poc/mp-gateway/src/session-manager.mjs 0 → 100644
  View file @8aed362
  1 +// 小程序网关会话：1 路 Ingress 接收小程序推流 + 每个远端参与者 1 路 Track Composite Egress 推到 SRS
  2 +import { createHash, randomBytes, timingSafeEqual } from 'node:crypto'
  3 +import { setTimeout as delay } from 'node:timers/promises'
  4 +import { GATEWAY_SRS_APP_PATTERN } from './config.mjs'
  5 +import { LiveKitApiError } from './livekit-api.mjs'
  6 +
  7 +const RESOURCE_PREFIX = 'yudao-mp-gw-' // 网关创建的 Ingress 名称前缀；单实例启动时据此回收残留
  8 +const USER_IDENTITY_PATTERN = /^[1-9]\d{0,18}$/ // 业务参与者 identity：数字 userId
  9 +const EXCLUDED_KINDS = new Set(['EGRESS', 'AGENT', 'SIP']) // 非业务参与者类型；其它小程序成员的 INGRESS 属于业务参与者
  10 +const EGRESS_LIST_GRACE_MS = 10000 // 新建 Egress 在活跃列表中可见前的宽限期
  11 +const SESSION_ID_PATTERN = /^[0-9a-f]{32}$/ // 会话编号格式
  12 +const STREAM_PREFIX = 'yudaompgw_' // 网关推往 SRS 的流名前缀
  13 +const STREAM_NAME_PATTERN = /^yudaompgw_[0-9a-f]{12}_[1-9]\d{0,18}_[0-9a-f]{16}$/ // 网关流名完整格式
  14 +const RELEASE_CONCURRENCY = 4 // 定期重试时同时进行的释放尝试上限
  15 +const SHUTDOWN_RELEASE_RETRY_MS = 500 // 退出时释放失败后的重试间隔；整体时长由入口的退出超时兜底
  16 +const REDACTED_STREAM_PATTERN = /^\{yud\.\.\.[\w-]{3}\}$/ // LiveKit 脱敏后的网关流名：首 3 位 + ... + 末 3 位（末位来自推流密钥）
  17 +
  18 +/** 网关业务错误；status 为返回给客户端的 HTTP 状态码 */
  19 +export class GatewayError extends Error {
  20 + /**
  21 + * @param {number} status HTTP 状态码
  22 + * @param {string} code 错误码
  23 + * @param {string} message 错误说明
  24 + */
  25 + constructor(status, code, message) {
  26 + super(message)
  27 + this.status = status
  28 + this.code = code
  29 + }
  30 +}
  31 +
  32 +/**
  33 + * 计算 SHA-256 摘要；会话凭证、推拉流密钥只保存摘要用于比对
  34 + *
  35 + * @param {string} value 原文
  36 + * @returns {Buffer} 摘要
  37 + */
  38 +function digest(value) {
  39 + return createHash('sha256').update(value).digest()
  40 +}
  41 +
  42 +/**
  43 + * 常量时间比较凭证
  44 + *
  45 + * @param {unknown} value 待校验原文
  46 + * @param {Buffer} expectedDigest 期望摘要
  47 + * @returns {boolean} 是否一致
  48 + */
  49 +function matches(value, expectedDigest) {
  50 + return typeof value === 'string' && value.length > 0 && timingSafeEqual(digest(value), expectedDigest)
  51 +}
  52 +
  53 +/**
  54 + * 是否为业务参与者：数字 userId、非 Egress 等内部参与者、未断开
  55 + *
  56 + * @param {{ identity: string, kind: string, state: string }} participant 参与者
  57 + * @returns {boolean} 是否业务参与者
  58 + */
  59 +function isBusinessParticipant(participant) {
  60 + return USER_IDENTITY_PATTERN.test(participant.identity)
  61 + && !EXCLUDED_KINDS.has(participant.kind)
  62 + && participant.state !== 'DISCONNECTED'
  63 +}
  64 +
  65 +/**
  66 + * 选择远端参与者要转出的媒体：视频屏幕共享优先、摄像头回退；音频取麦克风
  67 + * <p>
  68 + * 摄像头静音不更换 Egress，只回传 videoVisible，避免反复重建推流
  69 + *
  70 + * @param {{ tracks: Array<{ sid: string, type: string, source: string, muted: boolean }> }} participant 参与者
  71 + * @returns {{ audioSid?: string, videoSid?: string, videoSource?: string, videoVisible: boolean }} 媒体选择
  72 + */
  73 +function pickMedia(participant) {
  74 + const screenShare = participant.tracks.find(track => track.type === 'VIDEO' && track.source === 'SCREEN_SHARE' && !track.muted)
  75 + const camera = participant.tracks.find(track => track.type === 'VIDEO' && (track.source === 'CAMERA' || track.source === 'UNKNOWN'))
  76 + const video = screenShare || camera
  77 + const audio = participant.tracks.find(track => track.type === 'AUDIO' && (track.source === 'MICROPHONE' || track.source === 'UNKNOWN'))
  78 + return {
  79 + audioSid: audio?.sid,
  80 + videoSid: video?.sid,
  81 + videoSource: video?.source,
  82 + videoVisible: !!video && !video.muted,
  83 + }
  84 +}
  85 +
  86 +/**
  87 + * 创建调用的结果是否未知：超时、网络错误、HTTP 5xx（含代理 502 / 504 与响应缺少资源编号）时，
  88 + * 请求可能已被 LiveKit 受理、资源已建但编号丢失
  89 + * <p>
  90 + * 4xx 等明确拒绝的失败按未创建处理，沿用冷却重试
  91 + *
  92 + * @param {unknown} error 错误
  93 + * @returns {boolean} 是否结果未知
  94 + */
  95 +function isAmbiguousCreateError(error) {
  96 + return error instanceof LiveKitApiError && (error.status === 0 || error.status >= 500)
  97 +}
  98 +
  99 +/**
  100 + * 解析 SRS 回调中的查询参数
  101 + *
  102 + * @param {unknown} param 如 "?token=xxx"
  103 + * @returns {URLSearchParams} 参数
  104 + */
  105 +function parseHookParam(param) {
  106 + return new URLSearchParams(typeof param === 'string' ? param.replace(/^\?/, '') : '')
  107 +}
  108 +
  109 +/**
  110 + * 创建会话管理器（单实例）
  111 + *
  112 + * @param {Readonly<Record<string, any>>} config 网关配置
  113 + * @param {ReturnType<import('./livekit-api.mjs').createLiveKitApi>} liveKit LiveKit API
  114 + * @param {{ info: Function, warn: Function }} logger 日志；只记录编号与房间，不记录凭证
  115 + * @returns {Record<string, Function>} 会话管理方法
  116 + */
  117 +export function createSessionManager(config, liveKit, logger) {
  118 + const sessions = new Map() // 会话编号 -> 会话
  119 + const sessionsByUser = new Map() // 房间 + identity -> 会话；同一用户同一房间只允许一份会话
  120 + const streams = new Map() // SRS 流名 -> { session, entry }；会话关闭或远端移除时立即删除，撤销推拉流授权
  121 + const pendingReleases = new Map() // 「类型:资源编号」-> 待确认释放的 Ingress / Egress；释放成功前持续占用配额并定期重试
  122 + const inflightCreates = new Set() // 进行中的 Ingress / Egress 创建；结果返回前占用配额，退出时等待其完成
  123 + const unknownCreates = new Set() // 结果未知的创建（超时 / 网络错误）；资源可能已在 LiveKit 侧存在，进程内永久占用配额与用户维度，只能重启后由启动回收清理
  124 + let shuttingDown = false // 是否正在退出
  125 + const reaper = setInterval(reapSessions, 1000) // 租约、Token 过期扫描与释放重试
  126 +
  127 + /**
  128 + * 会话的用户维度键
  129 + *
  130 + * @param {{ room: string, identity: string }} value 会话或授权
  131 + * @returns {string} 键
  132 + */
  133 + function userKey(value) {
  134 + return `${value.room}\n${value.identity}`
  135 + }
  136 +
  137 + /**
  138 + * 当前占用的 Ingress 数：存活会话 + 待确认释放 + 归属已失效但仍在创建中 + 结果未知的 Ingress
  139 + *
  140 + * @returns {number} 数量
  141 + */
  142 + function countIngress() {
  143 + let count = sessions.size
  144 + pendingReleases.forEach((item) => {
  145 + if (item.kind === 'ingress') {
  146 + count++
  147 + }
  148 + })
  149 + inflightCreates.forEach((item) => {
  150 + if (item.kind === 'ingress' && !item.isCurrent()) {
  151 + count++
  152 + }
  153 + })
  154 + unknownCreates.forEach((item) => {
  155 + if (item.kind === 'ingress') {
  156 + count++
  157 + }
  158 + })
  159 + return count
  160 + }
  161 +
  162 + /**
  163 + * 当前占用的 Egress 路数：存活会话中的远端流（含启动中）+ 待确认释放 + 归属已失效但仍在创建中 + 结果未知的 Egress
  164 + *
  165 + * @returns {number} 路数
  166 + */
  167 + function countEgress() {
  168 + let count = 0
  169 + sessions.forEach((session) => {
  170 + count += session.remotes.size
  171 + })
  172 + pendingReleases.forEach((item) => {
  173 + if (item.kind === 'egress') {
  174 + count++
  175 + }
  176 + })
  177 + inflightCreates.forEach((item) => {
  178 + if (item.kind === 'egress' && !item.isCurrent()) {
  179 + count++
  180 + }
  181 + })
  182 + unknownCreates.forEach((item) => {
  183 + if (item.kind === 'egress') {
  184 + count++
  185 + }
  186 + })
  187 + return count
  188 + }
  189 +
  190 + /**
  191 + * 同一房间 + identity 是否仍有未释放的会话或推流资源（含释放中、创建中、结果未知）
  192 + *
  193 + * @param {string} key 用户维度键
  194 + * @returns {boolean} 是否占用
  195 + */
  196 + function isUserBusy(key) {
  197 + if (sessionsByUser.has(key)) {
  198 + return true
  199 + }
  200 + for (const item of pendingReleases.values()) {
  201 + if (item.ownerKey === key) {
  202 + return true
  203 + }
  204 + }
  205 + for (const item of inflightCreates) {
  206 + if (item.ownerKey === key) {
  207 + return true
  208 + }
  209 + }
  210 + for (const item of unknownCreates) {
  211 + if (item.ownerKey === key) {
  212 + return true
  213 + }
  214 + }
  215 + return false
  216 + }
  217 +
  218 + /**
  219 + * 尝试释放一个待释放资源；同一资源同时只有一次尝试，成功后移出待释放列表，失败则等待下次重试
  220 + *
  221 + * @param {{ kind: 'ingress' | 'egress', id: string, attempt?: Promise<boolean>, nextAt: number }} item 待释放资源
  222 + * @returns {Promise<boolean>} 是否已确认释放
  223 + */
  224 + function attemptRelease(item) {
  225 + if (item.attempt) {
  226 + return item.attempt
  227 + }
  228 + const key = `${item.kind}:${item.id}`
  229 + const attempt = (item.kind === 'ingress' ? liveKit.deleteIngress(item.id) : liveKit.stopEgress(item.id))
  230 + .then(() => {
  231 + if (pendingReleases.get(key) === item) {
  232 + pendingReleases.delete(key)
  233 + }
  234 + return true
  235 + }, (error) => {
  236 + item.nextAt = Date.now() + config.releaseRetryMs
  237 + logger.warn('资源释放失败，稍后重试', { kind: item.kind, id: item.id, error: error.message })
  238 + return false
  239 + })
  240 + .finally(() => {
  241 + if (item.attempt === attempt) {
  242 + item.attempt = undefined
  243 + }
  244 + })
  245 + item.attempt = attempt
  246 + return attempt
  247 + }
  248 +
  249 + /**
  250 + * 登记并立即释放资源；释放确认前资源保留在待释放列表中，继续占用配额
  251 + *
  252 + * @param {{ kind: 'ingress' | 'egress', id: string, ownerKey?: string }} resource 资源；Ingress 带用户维度键
  253 + * @returns {Promise<boolean>} 首次尝试是否已确认释放
  254 + */
  255 + function releaseResource(resource) {
  256 + const key = `${resource.kind}:${resource.id}`
  257 + let item = pendingReleases.get(key)
  258 + if (!item) {
  259 + item = { ...resource, attempt: undefined, nextAt: 0 }
  260 + pendingReleases.set(key, item)
  261 + }
  262 + return attemptRelease(item)
  263 + }
  264 +
  265 + /**
  266 + * 重试到期的待释放资源，同时进行的释放尝试不超过上限
  267 + *
  268 + * @param {boolean} force 是否忽略重试间隔（退出时使用）
  269 + */
  270 + function retryReleases(force) {
  271 + const now = Date.now()
  272 + let active = 0 // 本轮扫描时正在进行的释放尝试
  273 + pendingReleases.forEach((item) => {
  274 + if (item.attempt) {
  275 + active++
  276 + }
  277 + })
  278 + pendingReleases.forEach((item) => {
  279 + if (active < RELEASE_CONCURRENCY && !item.attempt && (force || item.nextAt <= now)) {
  280 + active++
  281 + void attemptRelease(item)
  282 + }
  283 + })
  284 + }
  285 +
  286 + /**
  287 + * 跟踪资源创建：结果返回时若归属已失效（会话关闭、远端已更换），在移出创建列表前转入待释放，配额不出现空档；
  288 + * 结果未知（超时 / 网络错误）时转入未知列表，继续占用配额与用户维度，不再盲目重建
  289 + *
  290 + * @param {{ kind: 'ingress' | 'egress', ownerKey: string, isCurrent: () => boolean }} item 创建项
  291 + * @param {Promise<Record<string, any>>} promise LiveKit 创建调用
  292 + * @returns {Promise<Record<string, any>>} 创建结果
  293 + */
  294 + function trackCreate(item, promise) {
  295 + inflightCreates.add(item)
  296 + const task = promise.then((result) => {
  297 + const id = item.kind === 'ingress' ? result.ingressId : result.egressId
  298 + if (id && !item.isCurrent()) {
  299 + void releaseResource({ kind: item.kind, id, ownerKey: item.ownerKey })
  300 + }
  301 + inflightCreates.delete(item)
  302 + return result
  303 + }, (error) => {
  304 + if (isAmbiguousCreateError(error)) {
  305 + unknownCreates.add(item)
  306 + logger.warn('创建结果未知，保留占位直到网关重启后启动回收', { kind: item.kind, error: error.message })
  307 + }
  308 + inflightCreates.delete(item)
  309 + throw error
  310 + })
  311 + item.settled = task.then(() => undefined, () => undefined)
  312 + return task
  313 + }
  314 +
  315 + /**
  316 + * 创建会话：同一房间 + identity 已有任何未释放的会话或推流资源时返回 409，不替换正在推流的会话
  317 + * <p>
  318 + * 替换已进房的 Ingress 会触发 participant_left，后端会据此结束私聊通话，因此由原会话 DELETE 或租约到期后才允许新建
  319 + *
  320 + * @param {{ room: string, identity: string, name: string, exp: number }} claims 房间 Token 授权
  321 + * @returns {Promise<{ session: Record<string, any>, sessionToken: string }>} 会话与会话凭证
  322 + */
  323 + async function createSession(claims) {
  324 + if (shuttingDown) {
  325 + throw new GatewayError(503, 'shutting_down', '网关正在退出')
  326 + }
  327 + if (isUserBusy(userKey(claims))) {
  328 + throw new GatewayError(409, 'session_busy', '该用户在本房间已有小程序通话会话或资源未释放；请先结束已有会话，持续出现请联系管理员')
  329 + }
  330 + if (countIngress() >= config.maxSessions) {
  331 + throw new GatewayError(429, 'session_limit', '小程序网关会话数已达上限')
  332 + }
  333 + const now = Date.now()
  334 + const sessionToken = randomBytes(32).toString('base64url')
  335 + const session = {
  336 + id: randomBytes(16).toString('hex'), // 会话编号，同时作为资源名后缀
  337 + tokenDigest: digest(sessionToken), // 会话凭证摘要
  338 + room: claims.room, // 房间名，取自 Token
  339 + identity: claims.identity, // 用户 identity，取自 Token
  340 + status: 'starting', // starting / active / closed
  341 + closed: false, // 是否已关闭；异步步骤完成后据此清理迟到资源
  342 + closeReason: undefined, // 关闭原因
  343 + releaseTask: undefined, // 资源释放任务（首次尝试完成即结束，未确认的资源转入待释放重试）
  344 + ingressId: undefined, // 小程序推流 Ingress
  345 + pushUrl: undefined, // 小程序推流地址（含 stream key，仅返回给创建者）
  346 + publishing: false, // 小程序推流是否已作为参与者进房
  347 + everPublished: false, // 是否曾经进房；之后房间消失即视为通话结束
  348 + remotes: new Map(), // 远端 userId -> 远端流
  349 + waitingUserIds: [], // 已进房但尚无可转出轨道的远端
  350 + overflowUserIds: [], // 超出单会话路数上限的远端
  351 + capacityUserIds: [], // 因全局 Egress 上限暂未转出的远端
  352 + remoteErrors: new Map(), // 远端 userId -> 最近一次转出失败原因
  353 + unknownUserIds: new Set(), // Egress 创建结果未知的远端；本会话内不再为其重建，避免叠加未知资源
  354 + presentUserIds: new Set(), // 最近一次同步时仍在房间内的远端
  355 + retryAfter: new Map(), // 远端 userId -> 允许重试的时间
  356 + syncing: false, // 是否正在同步房间
  357 + syncError: undefined, // 最近一次同步失败原因
  358 + syncTimer: undefined, // 同步定时器
  359 + leaseExpiresAt: now + config.leaseTtlMs, // 租约到期时间
  360 + expiresAt: Math.min(claims.exp * 1000, now + config.maxSessionMs), // 会话最晚结束时间
  361 + }
  362 + sessions.set(session.id, session)
  363 + sessionsByUser.set(userKey(session), session)
  364 + try {
  365 + const ingress = await trackCreate({
  366 + kind: 'ingress',
  367 + ownerKey: userKey(session),
  368 + isCurrent: () => !session.closed,
  369 + }, liveKit.createIngress({
  370 + room: session.room,
  371 + identity: session.identity,
  372 + name: claims.name,
  373 + resourceName: `${RESOURCE_PREFIX}${session.id}`,
  374 + }))
  375 + if (session.closed) { // 创建期间会话已关闭：转入待释放（与 trackCreate 的登记幂等）
  376 + if (ingress.ingressId) {
  377 + void releaseResource({ kind: 'ingress', id: ingress.ingressId, ownerKey: userKey(session) })
  378 + }
  379 + throw new GatewayError(409, 'session_closed', '会话在创建过程中已关闭')
  380 + }
  381 + session.ingressId = ingress.ingressId
  382 + if (!ingress.ingressId || !ingress.url || !ingress.streamKey) {
  383 + throw new GatewayError(502, 'ingress_url_missing', 'LiveKit 未返回 RTMP 推流地址，请检查 ingress.rtmp_base_url')
  384 + }
  385 + session.pushUrl = `${ingress.url}/${ingress.streamKey}`
  386 + session.status = 'active'
  387 + } catch (error) {
  388 + await closeSession(session, 'create_failed')
  389 + if (error instanceof GatewayError) {
  390 + throw error
  391 + }
  392 + logger.warn('创建 Ingress 失败', { room: session.room, error: error.message })
  393 + throw new GatewayError(502, 'livekit_unavailable', '创建小程序推流失败')
  394 + }
  395 + session.syncTimer = setInterval(() => void syncSession(session), config.syncIntervalMs)
  396 + void syncSession(session)
  397 + logger.info('会话已创建', { session: session.id, room: session.room, userId: session.identity, ingressId: session.ingressId })
  398 + return { session, sessionToken }
  399 + }
  400 +
  401 + /**
  402 + * 关闭会话：立即撤销推拉流授权与会话查询，释放 Ingress、Egress；重复调用返回同一个释放任务
  403 + * <p>
  404 + * 释放任务在首次尝试完成后结束；未确认释放的资源保留在待释放列表中重试，并继续占用配额与用户维度
  405 + *
  406 + * @param {Record<string, any>} session 会话
  407 + * @param {string} reason 关闭原因
  408 + * @returns {Promise<void>} 释放任务
  409 + */
  410 + function closeSession(session, reason) {
  411 + if (session.releaseTask) {
  412 + return session.releaseTask
  413 + }
  414 + session.closed = true
  415 + session.closeReason = reason
  416 + session.status = 'closed'
  417 + clearInterval(session.syncTimer)
  418 + sessions.delete(session.id)
  419 + if (sessionsByUser.get(userKey(session)) === session) {
  420 + sessionsByUser.delete(userKey(session))
  421 + }
  422 + const entries = Array.from(session.remotes.values())
  423 + session.remotes.clear()
  424 + entries.forEach((entry) => {
  425 + if (streams.get(entry.streamName)?.entry === entry) {
  426 + streams.delete(entry.streamName)
  427 + }
  428 + })
  429 + const resources = entries.filter(entry => entry.egressId).map(entry => ({ kind: 'egress', id: entry.egressId }))
  430 + if (session.ingressId) {
  431 + resources.push({ kind: 'ingress', id: session.ingressId, ownerKey: userKey(session) })
  432 + }
  433 + session.releaseTask = Promise.all(resources.map(releaseResource)).then((results) => {
  434 + logger.info('会话已关闭', { session: session.id, room: session.room, reason, pendingRelease: results.filter(released => !released).length })
  435 + })
  436 + return session.releaseTask
  437 + }
  438 +
  439 + /**
  440 + * 移除单路远端流：立即撤销授权并释放 Egress；启动中的 Egress 由 trackCreate 在返回后转入待释放
  441 + *
  442 + * @param {Record<string, any>} session 会话
  443 + * @param {Record<string, any>} entry 远端流
  444 + */
  445 + function removeRemoteStream(session, entry) {
  446 + if (session.remotes.get(entry.userId) === entry) {
  447 + session.remotes.delete(entry.userId)
  448 + }
  449 + if (streams.get(entry.streamName)?.entry === entry) {
  450 + streams.delete(entry.streamName)
  451 + }
  452 + if (entry.egressId) {
  453 + void releaseResource({ kind: 'egress', id: entry.egressId })
  454 + }
  455 + }
  456 +
  457 + /**
  458 + * 为远端参与者启动 Egress；每次启动使用新的流名与推拉流密钥
  459 + *
  460 + * @param {Record<string, any>} session 会话
  461 + * @param {string} userId 远端 userId
  462 + * @param {ReturnType<typeof pickMedia>} media 媒体选择
  463 + * @returns {Promise<void>}
  464 + */
  465 + async function startRemoteStream(session, userId, media) {
  466 + const publishSecret = randomBytes(24).toString('base64url')
  467 + const playToken = randomBytes(24).toString('base64url')
  468 + const entry = {
  469 + userId, // 远端 userId
  470 + audioSid: media.audioSid, // 转出的音频轨道
  471 + videoSid: media.videoSid, // 转出的视频轨道
  472 + videoSource: media.videoSource, // 视频来源（CAMERA / SCREEN_SHARE），仅用于日志排查
  473 + videoVisible: media.videoVisible, // 视频是否可见（摄像头静音时为 false）
  474 + streamName: `${STREAM_PREFIX}${session.id.slice(0, 12)}_${userId}_${randomBytes(8).toString('hex')}`, // SRS 流名
  475 + publishDigest: digest(publishSecret), // Egress 推流密钥摘要
  476 + playToken, // 小程序拉流凭证明文，仅保存在内存中，用于在会话查询中重新下发 playUrl
  477 + playDigest: digest(playToken), // 拉流凭证摘要
  478 + egressId: undefined, // Egress 编号
  479 + publishing: false, // SRS 是否正在接收推流
  480 + everPublished: false, // SRS 是否收到过推流；启动超时只看该标记
  481 + startedAt: Date.now(), // 启动时间
  482 + }
  483 + session.remotes.set(userId, entry)
  484 + streams.set(entry.streamName, { session, entry })
  485 + const createItem = { // Egress 创建项；结果未知时作为占位保留
  486 + kind: 'egress',
  487 + ownerKey: userKey(session),
  488 + isCurrent: () => !session.closed && session.remotes.get(userId) === entry,
  489 + }
  490 + try {
  491 + const egress = await trackCreate(createItem, liveKit.startTrackCompositeEgress({
  492 + room: session.room,
  493 + audioTrackId: media.audioSid,
  494 + videoTrackId: media.videoSid,
  495 + url: `${config.srsPublishBaseUrl}/${entry.streamName}?secret=${publishSecret}`,
  496 + encoding: config.egressEncoding,
  497 + }))
  498 + if (session.closed || session.remotes.get(userId) !== entry) { // 启动期间远端已更换：转入待释放（与 trackCreate 的登记幂等）
  499 + if (egress.egressId) {
  500 + void releaseResource({ kind: 'egress', id: egress.egressId })
  501 + }
  502 + return
  503 + }
  504 + entry.egressId = egress.egressId
  505 + session.remoteErrors.delete(userId)
  506 + logger.info('远端转出已启动', { session: session.id, userId, egressId: egress.egressId, audio: !!media.audioSid, video: !!media.videoSid })
  507 + } catch (error) {
  508 + if (unknownCreates.has(createItem)) {
  509 + session.unknownUserIds.add(userId)
  510 + }
  511 + if (session.closed || session.remotes.get(userId) !== entry) {
  512 + return
  513 + }
  514 + removeRemoteStream(session, entry)
  515 + if (!session.unknownUserIds.has(userId)) { // 结果未知的远端不再重试，状态由 unknownUserIds 返回 failed
  516 + session.retryAfter.set(userId, Date.now() + config.egressRetryMs)
  517 + session.remoteErrors.set(userId, 'egress_failed')
  518 + }
  519 + logger.warn('远端转出启动失败', { session: session.id, userId, error: error.message })
  520 + }
  521 + }
  522 +
  523 + /**
  524 + * 按房间现状同步远端转出：新增、按媒体组合重建、回收已结束的 Egress
  525 + *
  526 + * @param {Record<string, any>} session 会话
  527 + * @returns {Promise<void>}
  528 + */
  529 + async function syncSession(session) {
  530 + if (session.closed || session.syncing) {
  531 + return
  532 + }
  533 + session.syncing = true
  534 + try {
  535 + const participants = await liveKit.listParticipants(session.room)
  536 + if (session.closed) {
  537 + return
  538 + }
  539 + if (participants === null) {
  540 + session.publishing = false
  541 + if (session.everPublished) {
  542 + void closeSession(session, 'room_gone')
  543 + }
  544 + return
  545 + }
  546 + session.publishing = participants.some(participant => participant.identity === session.identity
  547 + && participant.kind === 'INGRESS'
  548 + && participant.state !== 'DISCONNECTED'
  549 + && participant.tracks.length > 0)
  550 + session.everPublished = session.everPublished || session.publishing
  551 + const desired = new Map() // 远端 userId -> 媒体选择
  552 + const waiting = []
  553 + const overflow = []
  554 + participants
  555 + .filter(participant => isBusinessParticipant(participant) && participant.identity !== session.identity)
  556 + .sort((a, b) => a.joinedAt - b.joinedAt)
  557 + .forEach((participant) => {
  558 + const media = pickMedia(participant)
  559 + if (!media.audioSid && !media.videoSid) {
  560 + waiting.push(participant.identity)
  561 + } else if (desired.size >= config.maxRemotesPerSession) {
  562 + overflow.push(participant.identity)
  563 + } else {
  564 + desired.set(participant.identity, media)
  565 + }
  566 + })
  567 + Array.from(session.remotes.values()).forEach((entry) => {
  568 + const media = desired.get(entry.userId)
  569 + if (!media || media.audioSid !== entry.audioSid || media.videoSid !== entry.videoSid) {
  570 + removeRemoteStream(session, entry)
  571 + } else {
  572 + entry.videoVisible = media.videoVisible
  573 + }
  574 + })
  575 + const now = Date.now()
  576 + // Egress 已启动但始终没有推到 SRS（如编码管线一直拿不到可输出的媒体）：停止、冷却后重试，如实返回 failed
  577 + Array.from(session.remotes.values()).forEach((entry) => {
  578 + if (entry.egressId && !entry.everPublished && now - entry.startedAt > config.egressPublishTimeoutMs) {
  579 + removeRemoteStream(session, entry)
  580 + session.retryAfter.set(entry.userId, now + config.egressRetryMs)
  581 + session.remoteErrors.set(entry.userId, 'egress_no_media')
  582 + logger.warn('远端转出启动超时，未收到 SRS 推流', { session: session.id, userId: entry.userId, egressId: entry.egressId, videoSource: entry.videoSource || 'NONE', audio: !!entry.audioSid })
  583 + }
  584 + })
  585 + if (Array.from(session.remotes.values()).some(entry => entry.egressId && now - entry.startedAt > EGRESS_LIST_GRACE_MS)) {
  586 + const activeIds = new Set((await liveKit.listActiveEgress(session.room)).map(item => item.egressId))
  587 + if (session.closed) {
  588 + return
  589 + }
  590 + Array.from(session.remotes.values()).forEach((entry) => {
  591 + if (entry.egressId && now - entry.startedAt > EGRESS_LIST_GRACE_MS && !activeIds.has(entry.egressId)) {
  592 + removeRemoteStream(session, entry)
  593 + session.retryAfter.set(entry.userId, now + config.egressRetryMs)
  594 + session.remoteErrors.set(entry.userId, 'egress_ended')
  595 + }
  596 + })
  597 + }
  598 + const capacity = []
  599 + desired.forEach((media, userId) => {
  600 + if (session.remotes.has(userId) || session.unknownUserIds.has(userId) || (session.retryAfter.get(userId) || 0) > now) {
  601 + return
  602 + }
  603 + if (shuttingDown) {
  604 + return
  605 + }
  606 + if (countEgress() >= config.maxEgress) {
  607 + capacity.push(userId)
  608 + return
  609 + }
  610 + void startRemoteStream(session, userId, media)
  611 + })
  612 + Array.from(session.retryAfter.keys()).forEach((userId) => {
  613 + if (!desired.has(userId)) {
  614 + session.retryAfter.delete(userId)
  615 + session.remoteErrors.delete(userId)
  616 + }
  617 + })
  618 + session.presentUserIds = new Set([...desired.keys(), ...waiting, ...overflow])
  619 + session.waitingUserIds = waiting
  620 + session.overflowUserIds = overflow
  621 + session.capacityUserIds = capacity
  622 + session.syncError = undefined
  623 + } catch (error) {
  624 + session.syncError = 'livekit_unavailable'
  625 + logger.warn('同步房间失败', { session: session.id, room: session.room, error: error.message })
  626 + } finally {
  627 + session.syncing = false
  628 + }
  629 + }
  630 +
  631 + /** 扫描租约与 Token 过期的会话，并重试到期的资源释放 */
  632 + function reapSessions() {
  633 + retryReleases(false)
  634 + const now = Date.now()
  635 + sessions.forEach((session) => {
  636 + if (now > session.leaseExpiresAt) {
  637 + void closeSession(session, 'lease_expired')
  638 + } else if (now > session.expiresAt) {
  639 + void closeSession(session, 'token_expired')
  640 + }
  641 + })
  642 + }
  643 +
  644 + /**
  645 + * 按会话编号与会话凭证查找会话；任何不匹配都按不存在处理
  646 + *
  647 + * @param {string} sessionId 会话编号
  648 + * @param {string | undefined} sessionToken 会话凭证
  649 + * @returns {Record<string, any>} 会话
  650 + */
  651 + function authorize(sessionId, sessionToken) {
  652 + const session = SESSION_ID_PATTERN.test(sessionId) ? sessions.get(sessionId) : undefined
  653 + if (!session || !matches(sessionToken, session.tokenDigest)) {
  654 + throw new GatewayError(404, 'session_not_found', '会话不存在或已结束')
  655 + }
  656 + return session
  657 + }
  658 +
  659 + /**
  660 + * 续租会话并返回小程序可见状态
  661 + *
  662 + * @param {Record<string, any>} session 会话
  663 + * @returns {Record<string, any>} 会话状态
  664 + */
  665 + function touch(session) {
  666 + session.leaseExpiresAt = Date.now() + config.leaseTtlMs
  667 + const remotes = Array.from(session.remotes.values()).map(entry => ({
  668 + userId: Number(entry.userId),
  669 + state: entry.publishing ? 'ready' : 'starting',
  670 + ready: entry.publishing,
  671 + playUrl: entry.publishing ? `${config.srsPlayBaseUrl}/${entry.streamName}?token=${entry.playToken}` : null, // SRS 收到推流后才下发，避免播放器提前拉空流
  672 + hasAudio: !!entry.audioSid,
  673 + hasVideo: !!entry.videoSid,
  674 + videoVisible: entry.videoVisible,
  675 + }))
  676 + const listed = new Set(session.remotes.keys()) // 已输出的远端，避免同一远端重复出现
  677 + const appendState = (userIds, state) => userIds.forEach((userId) => {
  678 + if (!listed.has(userId)) {
  679 + listed.add(userId)
  680 + remotes.push({ userId: Number(userId), state, ready: false, playUrl: null, hasAudio: false, hasVideo: false, videoVisible: false })
  681 + }
  682 + })
  683 + appendState(session.waitingUserIds, 'waiting')
  684 + appendState(session.overflowUserIds, 'overflow')
  685 + appendState(session.capacityUserIds, 'capacity')
  686 + appendState(Array.from(session.remoteErrors.keys()), 'failed')
  687 + appendState(Array.from(session.unknownUserIds).filter(userId => session.presentUserIds.has(userId)), 'failed')
  688 + return {
  689 + sessionId: session.id,
  690 + status: session.status,
  691 + publishing: session.publishing,
  692 + remotes,
  693 + error: session.syncError || null,
  694 + leaseTtlMs: config.leaseTtlMs,
  695 + pollIntervalMs: config.pollIntervalMs,
  696 + expiresAt: session.expiresAt,
  697 + }
  698 + }
  699 +
  700 + /**
  701 + * 查找 SRS 回调对应的远端流
  702 + *
  703 + * @param {Record<string, any>} hook SRS 回调体
  704 + * @returns {{ session: Record<string, any>, entry: Record<string, any> } | undefined} 远端流
  705 + */
  706 + function findStream(hook) {
  707 + if (!hook || hook.app !== config.srsApp || typeof hook.stream !== 'string') {
  708 + return undefined
  709 + }
  710 + const ref = streams.get(hook.stream)
  711 + if (!ref || ref.session.closed || ref.session.remotes.get(ref.entry.userId) !== ref.entry) {
  712 + return undefined
  713 + }
  714 + return ref
  715 + }
  716 +
  717 + /**
  718 + * SRS on_publish：只允许网关启动的 Egress 携带对应密钥推流
  719 + *
  720 + * @param {Record<string, any>} hook SRS 回调体
  721 + * @returns {boolean} 是否允许
  722 + */
  723 + function authorizePublish(hook) {
  724 + const ref = findStream(hook)
  725 + if (!ref || !matches(parseHookParam(hook.param).get('secret'), ref.entry.publishDigest)) {
  726 + return false
  727 + }
  728 + ref.entry.publishing = true
  729 + ref.entry.everPublished = true
  730 + return true
  731 + }
  732 +
  733 + /**
  734 + * SRS on_unpublish：标记远端流未就绪
  735 + *
  736 + * @param {Record<string, any>} hook SRS 回调体
  737 + */
  738 + function handleUnpublish(hook) {
  739 + const ref = findStream(hook)
  740 + if (ref && matches(parseHookParam(hook.param).get('secret'), ref.entry.publishDigest)) {
  741 + ref.entry.publishing = false
  742 + }
  743 + }
  744 +
  745 + /**
  746 + * SRS on_play：只允许持有本会话拉流凭证的小程序播放对应远端流
  747 + *
  748 + * @param {Record<string, any>} hook SRS 回调体
  749 + * @returns {boolean} 是否允许
  750 + */
  751 + function authorizePlay(hook) {
  752 + const ref = findStream(hook)
  753 + return !!ref && matches(parseHookParam(hook.param).get('token'), ref.entry.playDigest)
  754 + }
  755 +
  756 + /**
  757 + * 判断 Egress 是否推往本网关的 SRS 流
  758 + * <p>
  759 + * LiveKit 会把 RTMP 地址最后一段脱敏为「{首 3 位...末 3 位}」，只有 scheme / host / port / app 可信；
  760 + * 因此归属以网关专用 app 为准：地址需为「SRS 推流根地址 / 单段流名」，且单段为完整网关流名或其脱敏形式
  761 + *
  762 + * @param {unknown} value EgressInfo 原始 JSON
  763 + * @returns {boolean} 是否本网关创建
  764 + */
  765 + function isGatewayEgress(value) {
  766 + if (typeof value === 'string') {
  767 + const prefix = `${config.srsPublishBaseUrl}/`
  768 + if (!GATEWAY_SRS_APP_PATTERN.test(config.srsApp) || !value.startsWith(prefix)) {
  769 + return false
  770 + }
  771 + const segment = value.slice(prefix.length)
  772 + return STREAM_NAME_PATTERN.test(segment.split('?')[0]) || REDACTED_STREAM_PATTERN.test(segment)
  773 + }
  774 + if (value && typeof value === 'object') {
  775 + return Object.values(value).some(isGatewayEgress)
  776 + }
  777 + return false
  778 + }
  779 +
  780 + /**
  781 + * 单实例启动时回收上次异常退出残留的 Ingress / Egress；任一资源释放失败都抛出，由调用方重试
  782 + *
  783 + * @returns {Promise<void>}
  784 + */
  785 + async function cleanupOrphans() {
  786 + const ingresses = (await liveKit.listIngress()).filter(item => item.name.startsWith(RESOURCE_PREFIX))
  787 + const egresses = (await liveKit.listActiveEgress()).filter(item => isGatewayEgress(item.raw))
  788 + const results = await Promise.allSettled([
  789 + ...ingresses.map(item => liveKit.deleteIngress(item.ingressId)),
  790 + ...egresses.map(item => liveKit.stopEgress(item.egressId)),
  791 + ])
  792 + const failed = results.filter(result => result.status === 'rejected').length
  793 + if (failed > 0) {
  794 + throw new Error(`残留资源回收失败 ${failed} 项`)
  795 + }
  796 + logger.info('残留资源回收完成', { ingress: ingresses.length, egress: egresses.length })
  797 + }
  798 +
  799 + /**
  800 + * 退出：拒绝新会话，关闭全部会话，等待进行中的创建返回并确认全部资源释放
  801 + * <p>
  802 + * 释放持续失败时不会返回，由入口的退出超时以非零状态结束；存在结果未知的创建时抛出错误，同样以非零状态退出，避免误报干净退出
  803 + *
  804 + * @returns {Promise<void>}
  805 + */
  806 + async function shutdown() {
  807 + shuttingDown = true
  808 + clearInterval(reaper)
  809 + Array.from(sessions.values()).forEach(session => void closeSession(session, 'shutdown'))
  810 + while (inflightCreates.size > 0 || pendingReleases.size > 0) {
  811 + if (inflightCreates.size > 0) {
  812 + await Promise.allSettled(Array.from(inflightCreates, item => item.settled))
  813 + continue
  814 + }
  815 + retryReleases(true)
  816 + await Promise.allSettled(Array.from(pendingReleases.values(), item => item.attempt))
  817 + if (pendingReleases.size > 0) {
  818 + await delay(SHUTDOWN_RELEASE_RETRY_MS)
  819 + }
  820 + }
  821 + if (unknownCreates.size > 0) {
  822 + throw new Error(`存在 ${unknownCreates.size} 个结果未知的 Ingress / Egress 创建，需重启网关由启动回收清理`)
  823 + }
  824 + }
  825 +
  826 + return {
  827 + createSession,
  828 + closeSession,
  829 + authorize,
  830 + touch,
  831 + authorizePublish,
  832 + handleUnpublish,
  833 + authorizePlay,
  834 + cleanupOrphans,
  835 + shutdown,
  836 + }
  837 +}
script/livekit-poc/mp-gateway/srs.conf 0 → 100644
  View file @8aed362
  1 +# SRS：接收 Egress 推来的远端流，供小程序 live-player 拉取；推流与播放都经网关 HTTP 回调鉴权
  2 +# 本实例由网关独占：回调对整个 vhost 生效，网关只放行专用 app（yudao-mp-gw）下自己创建的流，勿与其它直播业务共用
  3 +# 默认只监听回环；真机联调时改为 listen 1936; 并在防火墙 / 安全组只放通需要的来源
  4 +listen 127.0.0.1:1936;
  5 +max_connections 200;
  6 +daemon off;
  7 +srs_log_tank console;
  8 +
  9 +vhost __defaultVhost__ {
  10 + # 低延迟：关闭合并写、开启 TCP_NODELAY；GOP 缓存让播放器首帧不必等下一个关键帧（Egress 关键帧间隔 1 秒）
  11 + min_latency on;
  12 + tcp_nodelay on;
  13 + play {
  14 + gop_cache on;
  15 + queue_length 10;
  16 + mw_latency 100;
  17 + }
  18 + publish {
  19 + mr off;
  20 + }
  21 + # 网关回调端口只监听回环；网关不可用时 SRS 拒绝推流与播放
  22 + http_hooks {
  23 + enabled on;
  24 + on_publish http://127.0.0.1:8091/srs/hooks/on_publish;
  25 + on_unpublish http://127.0.0.1:8091/srs/hooks/on_unpublish;
  26 + on_play http://127.0.0.1:8091/srs/hooks/on_play;
  27 + }
  28 +}