Files
AAA 1b9f5bd2a2 feat: add room interactions and media sharing improvements
## Problem and outcome / 问题与结果

This phase adds lightweight room interaction without making media depend on chat,
and adds Windows source-audio exclusion and optional SFU-only hosting at their
existing owners. It also corrects confirmed encoder/capture ownership and native
color-conversion defects. Host microphone commentary remains on the shared media
path; room voice and 4K are outside this phase.

## Release notes

### 中文

这是一次功能更新:Piik 加入了文字聊天与表情互动,并完善了 Windows 声音采集、画面颜色和自部署选项。

#### 新功能

- **边看边聊。** 在可移动、可缩放的聊天窗口中发送消息,或开启弹幕。停止分享后,房间中的朋友仍可继续聊天。
- **表情互动。** 点击沙发上的人物,给自己发表情,或把番茄等表情送给朋友;各端按同一事件显示动画。
- **排除应用声音。** Windows App 分享整个屏幕时,可排除一个应用及其子进程的声音,适合同时使用语音软件。所选应用退出后,来源声音停止,避免恢复为未过滤的系统声音。
- **可选的纯 SFU 模式。** 自部署管理员可让共享画面和声音统一通过媒体服务器转发。默认仍为优先 P2P;App 公网邀请保持纯 P2P。
- **Windows HDR 转 SDR。** 原生采集与缩略图自动将 HDR 来源转换为 SDR,沿用现有观看链路。

#### 修复与改进

- 修复原生颜色转换的范围与矩阵处理,避免已复现的暗部丢失、高光裁切和颜色偏差。
- 修复共享编码器切换时使用过期需求或过期帧信息的问题,保护正在观看的画面;修复换源后旧采集事件可能影响新来源的问题。
- 统一新控件、浮窗与来源选择文案,补充中英文使用说明。

#### 使用与升级

聊天与表情需要新版站点页面和 Server;声音排除需要新版页面与支持此能力的 Windows App。聊天只保留本页最近收到的消息,刷新后清空,不补发历史,且不提供端到端加密。

HDR 转换输出为 SDR,不代表端到端 HDR 或多声道支持。既有房主麦克风与立体声能力保留。

### English

This feature update brings text chat and reactions to Piik, alongside improvements to Windows audio capture, picture colors and self-hosting options.

#### New features

- **Chat while watching.** Send messages in a movable, resizable window, or enable on-screen chat. Friends can keep chatting after the host stops sharing.
- **Reactions.** Select a person on the sofa to react above yourself or send a tomato and other reactions to a friend. Participants animate the same shared event.
- **Exclude application audio.** When sharing a whole screen with Windows App, exclude one application and its child processes, useful when using a separate voice app. Source audio stops if the selected app exits, preventing a return to unfiltered system sound.
- **Optional SFU-only hosting.** Site operators can require shared video and audio to pass through the media server. P2P-first remains the default; App Public invite remains P2P-only.
- **Windows HDR-to-SDR conversion.** Native capture and source thumbnails automatically convert HDR sources to SDR for the existing viewing path.

#### Fixes and improvements

- Correct native color-conversion ranges and matrices, addressing reproduced crushed shadows, clipped highlights and color shifts.
- Correct stale demand and frame evidence during shared-encoder handoffs to protect the current picture. Prevent events from a replaced capture from affecting its successor.
- Align new controls, floating panels and source-picker copy, with updated Chinese and English guides.

#### Using this update

Chat and reactions require updated site pages and Server. Audio exclusion requires updated pages and a Windows App that supports the capability. Chat retains only recent messages received by the current page, clears on refresh and does not replay history. It is not end-to-end encrypted.

HDR conversion produces SDR; it does not add end-to-end HDR or surround sound. Existing host microphone commentary and stereo support remain available.

## Verification / 验证

- Web type checks, 78 test files / 1275 tests and production build: PASS.
- Go core tests, formatting/vet and target builds: PASS; Darwin cgo remains platform-limited on this Windows runner.
- Native build/check/color probes and synthetic HDR-to-SDR H.264/VP8 browser reception: PASS.
- Access/privacy gate, Native H.264 SFU-only media gate and local Auto/H.264 Host gate: PASS.
- Three real production pages: chat/reactions, stop/resume, idle-room join, host departure/return, offline draft recovery and invitation revocation: PASS.
- Synthetic parent/child process-audio exclusion and target lifetime: PASS; no raw audio retained.
- Windows candidate archive extraction and packaged Local startup/shutdown: PASS.
- Documentation and website build: PASS.
- [Candidate CI](https://github.com/TNTcraftHIM/Piik/actions/runs/36873766646) at
  `f26cd208d03bc0dcbe353f8cac217f993216b0a3`: Go core and race checks, Web tests,
  Server/container checks and Windows/macOS/Linux App packaging all PASS.
  Publication was skipped as intended.

The owner accepted the UI/model and bounded local walkthrough for phase closure.
Physical HDR/mixed displays, mobile keyboard,
real voice applications across output devices and macOS/Linux capture remain
unverified. Existing field reports without matched evidence are not claimed
resolved. The owner authorized v1.7.0 publication on 2026-10-02. The final
documentation-only commit records that authorization; product code matches
the verified candidate above.

## Impact / 影响

Authorized version: **v1.7.0**, matching the local release planner. Current wire
contracts remain Browser/server 23, Native control 9 and capture 7. Optional
capabilities gate new functionality; existing room/media authority remains owned
by the existing services. No new ports, message persistence, room voice or
automatic room creation on page visit. Updated README, guides, standards,
research evidence and TODO stay with the candidate phase.
2026-10-02 01:36:45 +08:00

733 lines
25 KiB
Go

package protocol
import (
"errors"
"fmt"
)
// ServerMessage is one member of serverMessageSchema. The interface is sealed:
// only the message structs in this file implement it.
type ServerMessage interface {
isServerMessage()
}
// AuthenticatedHostMessage returns the Host's current room and transport facts.
type AuthenticatedHostMessage struct {
Type string `json:"type"`
Protocol string `json:"protocol"`
PeerID string `json:"peerId"`
MaxViewers Int `json:"maxViewers"`
EndpointMediaCopyCapacity Int `json:"endpointMediaCopyCapacity"`
HostOnline bool `json:"hostOnline"`
HostPaused *bool `json:"hostPaused,omitempty"`
ConnectionID *string `json:"connectionId"`
IceConfig IceConfig `json:"iceConfig"`
RoutePolicy RoutePolicy `json:"routePolicy"`
CodeEntryPolicy CodeEntryPolicy `json:"codeEntryPolicy"`
ViewerAuthorizationGeneration string `json:"viewerAuthorizationGeneration"`
MediaMode string `json:"mediaMode"`
ShareGeneration *string `json:"shareGeneration"`
RouteRevision Int `json:"routeRevision"`
RouteAssignment ParticipantRouteAssignment `json:"routeAssignment"`
QualitySettings QualitySettings `json:"qualitySettings"`
Role Role `json:"role"`
ViewerPasswordEnabled bool `json:"viewerPasswordEnabled"`
}
// AuthenticatedViewerMessage is the viewer member of
// authenticatedMessageSchema.
type AuthenticatedViewerMessage struct {
Type string `json:"type"`
Protocol string `json:"protocol"`
PeerID string `json:"peerId"`
MaxViewers Int `json:"maxViewers"`
EndpointMediaCopyCapacity Int `json:"endpointMediaCopyCapacity"`
HostOnline bool `json:"hostOnline"`
HostPaused *bool `json:"hostPaused,omitempty"`
ConnectionID *string `json:"connectionId"`
IceConfig IceConfig `json:"iceConfig"`
RoutePolicy RoutePolicy `json:"routePolicy"`
CodeEntryPolicy CodeEntryPolicy `json:"codeEntryPolicy"`
ViewerAuthorizationGeneration string `json:"viewerAuthorizationGeneration"`
MediaMode string `json:"mediaMode"`
ShareGeneration *string `json:"shareGeneration"`
RouteRevision Int `json:"routeRevision"`
RouteAssignment ParticipantRouteAssignment `json:"routeAssignment"`
QualitySettings QualitySettings `json:"qualitySettings"`
Role Role `json:"role"`
}
// SharingStartedMessage acknowledges a new media publication on an existing Host session.
type SharingStartedMessage struct {
Type string `json:"type"`
ShareGeneration string `json:"shareGeneration"`
RoutePolicy RoutePolicy `json:"routePolicy"`
RouteRevision Int `json:"routeRevision"`
RouteAssignment ParticipantRouteAssignment `json:"routeAssignment"`
QualitySettings QualitySettings `json:"qualitySettings"`
Paused bool `json:"paused"`
}
type SharingStartFailedMessage struct {
Type string `json:"type"`
ShareGeneration string `json:"shareGeneration"`
Code string `json:"code"`
}
// SignalingChallengeResponseMessage is the challenge echo.
type SignalingChallengeResponseMessage struct {
Type string `json:"type"`
Sequence Int `json:"sequence"`
}
// ServerSignalMessage is the forwarded "signal" message.
type ServerSignalMessage struct {
Type string `json:"type"`
FromPeerID string `json:"fromPeerId"`
Payload SignalPayload `json:"payload"`
}
// ServerRestartRequestMessage is the forwarded "restart-request" message.
type ServerRestartRequestMessage struct {
Type string `json:"type"`
FromPeerID string `json:"fromPeerId"`
ConnectionID string `json:"connectionId"`
Rebuild bool `json:"rebuild"`
}
// RouteUpdatePrepareMessage is the phase "prepare" member of route-update.
type RouteUpdatePrepareMessage struct {
Type string `json:"type"`
Revision Int `json:"revision"`
Phase string `json:"phase"`
Assignment ParticipantRouteAssignment `json:"assignment"`
Candidate PreparedRouteCandidate `json:"candidate"`
}
// RouteUpdateActiveMessage is the phase "active" member of route-update.
type RouteUpdateActiveMessage struct {
Type string `json:"type"`
Revision Int `json:"revision"`
Phase string `json:"phase"`
Assignment ParticipantRouteAssignment `json:"assignment"`
}
// RouteStatusMessage is the crossed (state, reason) union: "waiting" pairs
// with "sfu-admission" and "failed" pairs with "route-exhausted".
type RouteStatusMessage struct {
Type string `json:"type"`
Revision Int `json:"revision"`
State string `json:"state"`
Reason string `json:"reason"`
}
// RouteDiagnosticSnapshotMessage carries the privacy-safe route snapshot.
type RouteDiagnosticSnapshotMessage struct {
Type string `json:"type"`
Snapshot RouteDiagnosticSnapshot `json:"snapshot"`
}
// SfuConfigMessage authorizes one embedded media connection.
type SfuConfigMessage struct {
Type string `json:"type"`
Revision Int `json:"revision"`
PublicationGeneration string `json:"publicationGeneration"`
ConnectionID string `json:"connectionId"`
}
// QualitySettingsMessage broadcasts the host's quality settings.
type QualitySettingsMessage struct {
Type string `json:"type"`
QualitySettings QualitySettings `json:"qualitySettings"`
}
// RoutePolicyMessage broadcasts the host's route policy.
type RoutePolicyMessage struct {
Type string `json:"type"`
ShareGeneration string `json:"shareGeneration"`
RoutePolicy RoutePolicy `json:"routePolicy"`
}
// PauseSharingSourceMessage asks the host to pause its capture source.
type PauseSharingSourceMessage struct {
Type string `json:"type"`
ShareGeneration string `json:"shareGeneration"`
}
// ServerViewerQualityEvidenceMessage is the forwarded evidence envelope.
type ServerViewerQualityEvidenceMessage struct {
Type string `json:"type"`
ViewerPeerID string `json:"viewerPeerId"`
Upstream MediaRouteUpstream `json:"upstream"`
Guard ViewerQualityEvidenceGuard `json:"guard"`
Sequence Int `json:"sequence"`
WindowMs Int `json:"windowMs"`
Metrics ViewerQualityEvidenceMetrics `json:"metrics"`
}
// HostStatusMessage reports host presence and pause state.
type HostStatusMessage struct {
Type string `json:"type"`
Online bool `json:"online"`
Paused bool `json:"paused"`
}
// ViewerPresenceMessage carries the ordered participant roster.
type ViewerPresenceMessage struct {
Type string `json:"type"`
Viewers []ParticipantPresenceEntry `json:"viewers"`
}
// MarshalJSON implements json.Marshaler.
func (v ViewerPresenceMessage) MarshalJSON() ([]byte, error) {
type raw ViewerPresenceMessage
value := raw(v)
if value.Viewers == nil {
value.Viewers = []ParticipantPresenceEntry{}
}
return marshalJSON(value)
}
// ViewerGrantRevokedMessage announces a superseded viewer authorization.
type ViewerGrantRevokedMessage struct {
Type string `json:"type"`
ViewerAuthorizationGeneration string `json:"viewerAuthorizationGeneration"`
}
// SharingStoppedMessage is { type: "sharing-stopped" }.
type SharingStoppedMessage struct {
Type string `json:"type"`
}
// RoomClosedMessage is { type: "room-closed", reason }.
type RoomClosedMessage struct {
Type string `json:"type"`
Reason string `json:"reason"`
}
// ErrorMessage is { type: "error", code, message }.
type ErrorMessage struct {
Type string `json:"type"`
Code string `json:"code"`
Message string `json:"message"`
}
func (AuthenticatedHostMessage) isServerMessage() {}
func (AuthenticatedViewerMessage) isServerMessage() {}
func (SharingStartedMessage) isServerMessage() {}
func (SharingStartFailedMessage) isServerMessage() {}
func (SignalingChallengeResponseMessage) isServerMessage() {}
func (ServerSignalMessage) isServerMessage() {}
func (ServerRestartRequestMessage) isServerMessage() {}
func (RouteUpdatePrepareMessage) isServerMessage() {}
func (RouteUpdateActiveMessage) isServerMessage() {}
func (RouteStatusMessage) isServerMessage() {}
func (RouteDiagnosticSnapshotMessage) isServerMessage() {}
func (SfuConfigMessage) isServerMessage() {}
func (QualitySettingsMessage) isServerMessage() {}
func (RoutePolicyMessage) isServerMessage() {}
func (PauseSharingSourceMessage) isServerMessage() {}
func (ServerViewerQualityEvidenceMessage) isServerMessage() {}
func (HostStatusMessage) isServerMessage() {}
func (ViewerPresenceMessage) isServerMessage() {}
func (ViewerGrantRevokedMessage) isServerMessage() {}
func (SharingStoppedMessage) isServerMessage() {}
func (RoomClosedMessage) isServerMessage() {}
func (ErrorMessage) isServerMessage() {}
// EncodeServerMessage serializes one server message with the key order of the
// TypeScript object literal that produces it.
func EncodeServerMessage(message ServerMessage) ([]byte, error) {
return marshalJSON(message)
}
// DecodeServerMessage supports replaying the shared wire fixture from Go.
// The server never reads its own messages in production.
func DecodeServerMessage(data []byte) (ServerMessage, error) {
messageType, err := typeOf(data, "type")
if err != nil {
return nil, err
}
switch messageType {
case "room-interactions-ready":
return decodeRoomInteractionsReady(data)
case "room-interaction":
return decodeRoomInteraction(data)
case "room-interaction-rejected":
return decodeRoomInteractionRejected(data)
case "authenticated":
return decodeAuthenticated(data)
case "sharing-started":
return decodeSharingStarted(data)
case "sharing-start-failed":
return decodeSharingStartFailed(data)
case "signaling-challenge-response":
return decodeChallengeResponse(data)
case "signal":
return decodeServerSignal(data)
case "restart-request":
return decodeServerRestartRequest(data)
case "route-update":
return decodeRouteUpdate(data)
case "route-status":
return decodeRouteStatus(data)
case "route-diagnostic-snapshot":
return decodeRouteDiagnosticSnapshotMessage(data)
case "sfu-config":
return decodeSfuConfig(data)
case "sfu-signal":
return decodeSfuSignal(data)
case "quality-settings":
return decodeQualitySettingsMessage(data)
case "route-policy":
return decodeRoutePolicyMessage(data)
case "pause-sharing-source":
return decodePauseSharingSource(data)
case "viewer-quality-evidence":
return decodeServerViewerQualityEvidence(data)
case "host-status":
return decodeHostStatus(data)
case "viewer-presence":
return decodeViewerPresence(data)
case "viewer-grant-revoked":
return decodeViewerGrantRevoked(data)
case "sharing-stopped":
return decodeEmptyServerMessage(data, SharingStoppedMessage{Type: messageType})
case "room-closed":
return decodeRoomClosed(data)
case "error":
return decodeErrorMessage(data)
}
return nil, fmt.Errorf("unknown server message type %q", messageType)
}
func decodeEmptyServerMessage(data []byte, message ServerMessage) (ServerMessage, error) {
var body struct {
Type string `json:"type"`
}
present, err := decodeObject(data, &body)
if err != nil {
return nil, err
}
if err := present.require("type"); err != nil {
return nil, err
}
return message, nil
}
func validateAuthenticatedBase(
present fields,
protocol string, peerID string,
maxViewers, endpointCapacity Int, connectionID *string,
codeEntryPolicy, viewerAuthorizationGeneration, mediaMode string,
shareGeneration *string, routeRevision Int,
) error {
if err := present.require("type", "protocol", "peerId", "maxViewers",
"endpointMediaCopyCapacity", "hostOnline", "iceConfig", "codeEntryPolicy",
"viewerAuthorizationGeneration", "mediaMode", "routeRevision",
"routeAssignment", "qualitySettings", "role"); err != nil {
return err
}
if err := present.requireNullable(
"connectionId", "shareGeneration"); err != nil {
return err
}
if err := present.optional("hostPaused", "routePolicy"); err != nil {
return err
}
if protocol != SignalingProtocol {
return errors.New("protocol is not the current signaling generation")
}
if !ValidOpaqueID(peerID) {
return errors.New("peerId is not an opaque id")
}
if !inRangeInt(maxViewers, 1, MaxViewersPerRoomLimit) {
return errors.New("maxViewers is out of range")
}
if !inRangeInt(endpointCapacity, 1, MaxEndpointMediaCopyCapacity) {
return errors.New("endpointMediaCopyCapacity is out of range")
}
if connectionID != nil && !ValidOpaqueID(*connectionID) {
return errors.New("connectionId is not an opaque id")
}
if !enumOf(codeEntryPolicy, CodeEntryOpen, CodeEntryPrivate) {
return fmt.Errorf("unknown codeEntryPolicy %q", codeEntryPolicy)
}
if !ValidOpaqueID(viewerAuthorizationGeneration) {
return errors.New("viewerAuthorizationGeneration is not an opaque id")
}
if mediaMode != "peer-assisted" {
return errors.New("mediaMode must be peer-assisted")
}
if shareGeneration != nil && !ValidOpaqueID(*shareGeneration) {
return errors.New("shareGeneration is not an opaque id")
}
if !validRevision(routeRevision) {
return errors.New("routeRevision is out of range")
}
return nil
}
func decodeAuthenticated(data []byte) (ServerMessage, error) {
role, err := typeOf(data, "role")
if err != nil {
return nil, err
}
switch role {
case RoleHost:
var message AuthenticatedHostMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("viewerPasswordEnabled"); err != nil {
return nil, err
}
if err := validateAuthenticatedBase(present, message.Protocol, message.PeerID,
message.MaxViewers, message.EndpointMediaCopyCapacity,
message.ConnectionID, message.CodeEntryPolicy,
message.ViewerAuthorizationGeneration, message.MediaMode,
message.ShareGeneration, message.RouteRevision); err != nil {
return nil, err
}
if !present.has("routePolicy") {
message.RoutePolicy = DefaultRoutePolicy
}
return message, nil
case RoleViewer:
var message AuthenticatedViewerMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := validateAuthenticatedBase(present, message.Protocol, message.PeerID,
message.MaxViewers, message.EndpointMediaCopyCapacity,
message.ConnectionID, message.CodeEntryPolicy,
message.ViewerAuthorizationGeneration, message.MediaMode,
message.ShareGeneration, message.RouteRevision); err != nil {
return nil, err
}
if !present.has("routePolicy") {
message.RoutePolicy = DefaultRoutePolicy
}
return message, nil
}
return nil, fmt.Errorf("unknown authenticated role %q", role)
}
func decodeSharingStarted(data []byte) (ServerMessage, error) {
var message SharingStartedMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "shareGeneration", "routePolicy", "routeRevision",
"routeAssignment", "qualitySettings", "paused"); err != nil {
return nil, err
}
if !ValidOpaqueID(message.ShareGeneration) {
return nil, errors.New("shareGeneration is not an opaque id")
}
if !validRevision(message.RouteRevision) {
return nil, errors.New("routeRevision is out of range")
}
return message, nil
}
func decodeSharingStartFailed(data []byte) (ServerMessage, error) {
var message SharingStartFailedMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "shareGeneration", "code"); err != nil {
return nil, err
}
if !ValidOpaqueID(message.ShareGeneration) || (message.Code != "FORBIDDEN" && message.Code != "SERVER_ERROR") {
return nil, errors.New("invalid sharing-start-failed response")
}
return message, nil
}
func decodeChallengeResponse(data []byte) (ServerMessage, error) {
var message SignalingChallengeResponseMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "sequence"); err != nil {
return nil, err
}
if !inRangeInt(message.Sequence, 0, MaxSafeInteger) {
return nil, errors.New("sequence is out of range")
}
return message, nil
}
func decodeServerSignal(data []byte) (ServerMessage, error) {
var message ServerSignalMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "fromPeerId", "payload"); err != nil {
return nil, err
}
if !ValidOpaqueID(message.FromPeerID) {
return nil, errors.New("fromPeerId is not an opaque id")
}
return message, nil
}
func decodeServerRestartRequest(data []byte) (ServerMessage, error) {
var message ServerRestartRequestMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "fromPeerId", "connectionId", "rebuild"); err != nil {
return nil, err
}
if !ValidOpaqueID(message.FromPeerID) || !ValidOpaqueID(message.ConnectionID) {
return nil, errors.New("restart-request ids are not opaque ids")
}
return message, nil
}
func decodeRouteUpdate(data []byte) (ServerMessage, error) {
phase, err := typeOf(data, "phase")
if err != nil {
return nil, err
}
switch phase {
case "prepare":
var message RouteUpdatePrepareMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require(
"type", "revision", "phase", "assignment", "candidate"); err != nil {
return nil, err
}
if !validRevision(message.Revision) {
return nil, errors.New("revision is out of range")
}
return message, nil
case "active":
var message RouteUpdateActiveMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "revision", "phase", "assignment"); err != nil {
return nil, err
}
if !validRevision(message.Revision) {
return nil, errors.New("revision is out of range")
}
return message, nil
}
return nil, fmt.Errorf("unknown route-update phase %q", phase)
}
func decodeRouteStatus(data []byte) (ServerMessage, error) {
var message RouteStatusMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "revision", "state", "reason"); err != nil {
return nil, err
}
if !validRevision(message.Revision) {
return nil, errors.New("revision is out of range")
}
switch {
case message.State == "waiting" && message.Reason == "sfu-admission":
case message.State == "failed" && message.Reason == "route-exhausted":
default:
return nil, fmt.Errorf(
"route-status pairs state %q with reason %q", message.State, message.Reason)
}
return message, nil
}
func decodeRouteDiagnosticSnapshotMessage(data []byte) (ServerMessage, error) {
var message RouteDiagnosticSnapshotMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "snapshot"); err != nil {
return nil, err
}
return message, nil
}
func decodeSfuConfig(data []byte) (ServerMessage, error) {
var message SfuConfigMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "revision", "publicationGeneration", "connectionId"); err != nil {
return nil, err
}
if !validRevision(message.Revision) {
return nil, errors.New("revision is out of range")
}
if !ValidOpaqueID(message.PublicationGeneration) || !ValidOpaqueID(message.ConnectionID) {
return nil, errors.New("invalid SFU connection fence")
}
return message, nil
}
func decodeQualitySettingsMessage(data []byte) (ServerMessage, error) {
var message QualitySettingsMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "qualitySettings"); err != nil {
return nil, err
}
return message, nil
}
func decodeRoutePolicyMessage(data []byte) (ServerMessage, error) {
var message RoutePolicyMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "shareGeneration", "routePolicy"); err != nil {
return nil, err
}
if !ValidOpaqueID(message.ShareGeneration) {
return nil, errors.New("shareGeneration is not an opaque id")
}
return message, nil
}
func decodePauseSharingSource(data []byte) (ServerMessage, error) {
var message PauseSharingSourceMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "shareGeneration"); err != nil {
return nil, err
}
if !ValidOpaqueID(message.ShareGeneration) {
return nil, errors.New("shareGeneration is not an opaque id")
}
return message, nil
}
func decodeServerViewerQualityEvidence(data []byte) (ServerMessage, error) {
var message ServerViewerQualityEvidenceMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "viewerPeerId", "upstream", "guard",
"sequence", "windowMs", "metrics"); err != nil {
return nil, err
}
if !ValidOpaqueID(message.ViewerPeerID) {
return nil, errors.New("viewerPeerId is not an opaque id")
}
// The forwarded envelope narrows the upstream to the active variants.
if message.Upstream.Kind != "peer" && message.Upstream.Kind != "sfu" {
return nil, errors.New("forwarded evidence upstream must be peer or sfu")
}
if err := validateEvidenceWindow(message.Sequence, message.WindowMs); err != nil {
return nil, err
}
return message, nil
}
func decodeHostStatus(data []byte) (ServerMessage, error) {
var message HostStatusMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "online", "paused"); err != nil {
return nil, err
}
return message, nil
}
func decodeViewerPresence(data []byte) (ServerMessage, error) {
var message ViewerPresenceMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "viewers"); err != nil {
return nil, err
}
if len(message.Viewers) > MaxParticipantsPerRoomLimit {
return nil, errors.New("viewer presence holds too many entries")
}
// superRefine: viewer presence peer IDs must be unique.
seen := make(map[string]struct{}, len(message.Viewers))
for _, viewer := range message.Viewers {
if _, duplicate := seen[viewer.PeerID]; duplicate {
return nil, errors.New("Viewer presence peer IDs must be unique")
}
seen[viewer.PeerID] = struct{}{}
}
return message, nil
}
func decodeViewerGrantRevoked(data []byte) (ServerMessage, error) {
var message ViewerGrantRevokedMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "viewerAuthorizationGeneration"); err != nil {
return nil, err
}
if !ValidOpaqueID(message.ViewerAuthorizationGeneration) {
return nil, errors.New("viewerAuthorizationGeneration is not an opaque id")
}
return message, nil
}
func decodeRoomClosed(data []byte) (ServerMessage, error) {
var message RoomClosedMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "reason"); err != nil {
return nil, err
}
if message.Reason != "host-ended" {
return nil, fmt.Errorf("unknown room-closed reason %q", message.Reason)
}
return message, nil
}
func decodeErrorMessage(data []byte) (ServerMessage, error) {
var message ErrorMessage
present, err := decodeObject(data, &message)
if err != nil {
return nil, err
}
if err := present.require("type", "code", "message"); err != nil {
return nil, err
}
if !enumOf(message.Code, errorCodes...) {
return nil, fmt.Errorf("unknown error code %q", message.Code)
}
if length := UTF16Length(message.Message); length < 1 || length > 256 {
return nil, errors.New("error message length is out of range")
}
return message, nil
}