// Copyright 2023 LiveKit, Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package livekit import ( "context" "encoding/base64" "encoding/json" "errors" "fmt" "io" "slices" "time" "buf.build/go/protoyaml" "github.com/dennwc/iters" "go.opentelemetry.io/otel/attribute" proto "google.golang.org/protobuf/proto" "gopkg.in/yaml.v3" ) const ( TraceKeyPref = "lk." TraceKeyRoomPrefix = TraceKeyPref + "room." TraceKeyRoomID = attribute.Key(TraceKeyRoomPrefix + "id") TraceKeyRoomName = attribute.Key(TraceKeyRoomPrefix + "name") TraceKeyParticipantPrefix = TraceKeyPref + "participant." TraceKeyParticipantID = attribute.Key(TraceKeyParticipantPrefix + "id") TraceKeyParticipantIdentity = attribute.Key(TraceKeyParticipantPrefix + "identity") TraceKeyTrackPrefix = TraceKeyPref + "track." TraceKeyTrackID = attribute.Key(TraceKeyTrackPrefix + "id") TraceKeySIPPrefix = TraceKeyPref + "sip." TraceKeySIPHeaderPrefix = TraceKeySIPPrefix + "h." TraceKeySIPCallID = attribute.Key(TraceKeySIPPrefix + "call.id") TraceKeySIPCallIDHeader = attribute.Key(TraceKeySIPHeaderPrefix + "CallID") ) type TrackID string type ParticipantID string type ParticipantIdentity string type ParticipantName string type RoomID string type RoomName string type ConnectionID string type NodeID string type RoomKey struct { ProjectID string RoomName RoomName } type ParticipantKey struct { RoomKey Identity ParticipantIdentity } type JobID string type DispatchID string type AgentName string type SIPCallID string type SIPCallIDHeader string func (s TrackID) String() string { return string(s) } func (s ParticipantID) String() string { return string(s) } func (s ParticipantIdentity) String() string { return string(s) } func (s ParticipantName) String() string { return string(s) } func (s RoomID) String() string { return string(s) } func (s RoomName) String() string { return string(s) } func (s ConnectionID) String() string { return string(s) } func (s NodeID) String() string { return string(s) } func (s JobID) String() string { return string(s) } func (s DispatchID) String() string { return string(s) } func (s AgentName) String() string { return string(s) } func (s SIPCallID) String() string { return string(s) } func (s SIPCallIDHeader) String() string { return string(s) } func (s ParticipantKey) String() string { return fmt.Sprintf("%s_%s_%s", s.ProjectID, s.RoomName, s.Identity) } func (s ParticipantID) Trace() attribute.KeyValue { return TraceKeyParticipantID.String(string(s)) } func (s ParticipantIdentity) Trace() attribute.KeyValue { return TraceKeyParticipantIdentity.String(string(s)) } func (s RoomID) Trace() attribute.KeyValue { return TraceKeyRoomID.String(string(s)) } func (s RoomName) Trace() attribute.KeyValue { return TraceKeyRoomName.String(string(s)) } func (s TrackID) Trace() attribute.KeyValue { return TraceKeyTrackID.String(string(s)) } func (s SIPCallID) Trace() attribute.KeyValue { return TraceKeySIPCallID.String(string(s)) } func (s SIPCallIDHeader) Trace() attribute.KeyValue { return TraceKeySIPCallIDHeader.String(string(s)) } type stringTypes interface { ParticipantID | RoomID | TrackID | ParticipantIdentity | ParticipantName | RoomName | ConnectionID | NodeID } func IDsAsStrings[T stringTypes](ids []T) []string { strs := make([]string, 0, len(ids)) for _, id := range ids { strs = append(strs, string(id)) } return strs } func StringsAsIDs[T stringTypes](ids []string) []T { asID := make([]T, 0, len(ids)) for _, id := range ids { asID = append(asID, T(id)) } return asID } type Guid interface { TrackID | ParticipantID | RoomID } type GuidBlock [9]byte func (r *RoomConfiguration) UnmarshalYAML(value *yaml.Node) error { // Marshall the Node back to yaml to pass it to the protobuf specific unmarshaller str, err := yaml.Marshal(value) if err != nil { return err } return protoyaml.Unmarshal(str, r) } func (r *RoomConfiguration) MarshalYAML() (interface{}, error) { return marshalProto(r) } func (r *RoomEgress) UnmarshalYAML(value *yaml.Node) error { // Marshall the Node back to yaml to pass it to the protobuf specific unmarshaller str, err := yaml.Marshal(value) if err != nil { return err } return protoyaml.Unmarshal(str, r) } func (r *RoomEgress) MarshalYAML() (interface{}, error) { return marshalProto(r) } func (r *RoomAgent) UnmarshalYAML(value *yaml.Node) error { // Marshall the Node back to yaml to pass it to the protobuf specific unmarshaller str, err := yaml.Marshal(value) if err != nil { return err } return protoyaml.Unmarshal(str, r) } func (r *RoomAgent) MarshalYAML() (interface{}, error) { return marshalProto(r) } func marshalProto(o proto.Message) (map[string]interface{}, error) { // Marshall the Node to yaml using the protobuf specific marshaller to ensure the proper field names are used str, err := protoyaml.MarshalOptions{UseProtoNames: true}.Marshal(o) if err != nil { return nil, err } m := make(map[string]interface{}) err = yaml.Unmarshal(str, &m) if err != nil { return nil, err } return m, nil } func cloneProto[T proto.Message](m T) T { return proto.Clone(m).(T) } func IsJobType(jobType JobType) bool { _, ok := JobType_name[int32(jobType)] return ok } func (p *Pagination) Filter(v PageItem) bool { if p == nil { return true } if p.AfterId != "" { if id := v.ID(); id != "" && id <= p.AfterId { return false } } return true } // TokenPaginationData represents the data encoded in a TokenPagination token type TokenPaginationData struct { Offset int32 `json:"offset"` Limit int32 `json:"limit"` } // EncodeTokenPagination encodes offset and limit into a TokenPagination. // The token is a base64-encoded JSON object containing the offset and limit values. func EncodeTokenPagination(offset, limit int32) (*TokenPagination, error) { data := TokenPaginationData{ Offset: offset, Limit: limit, } jsonData, err := json.Marshal(data) if err != nil { return nil, fmt.Errorf("failed to marshal token pagination data: %w", err) } token := base64.URLEncoding.EncodeToString(jsonData) return &TokenPagination{Token: token}, nil } // DecodeTokenPagination decodes a TokenPagination into offset and limit. // Returns an error if the token is invalid or cannot be decoded. // If the TokenPagination is nil or has an empty token, returns zero values without error. func DecodeTokenPagination(tp *TokenPagination) (offset, limit int32, err error) { if tp == nil || tp.Token == "" { return 0, 0, nil } decoded, err := base64.URLEncoding.DecodeString(tp.Token) if err != nil { return 0, 0, fmt.Errorf("failed to decode token: %w", err) } var data TokenPaginationData if err := json.Unmarshal(decoded, &data); err != nil { return 0, 0, fmt.Errorf("failed to unmarshal token pagination data: %w", err) } return data.Offset, data.Limit, nil } // CursorTokenData holds the encoded pieces used for cursor pagination. // // Instead of paginating with an offset, cursor pagination represents // a position in an ordered list. The server returns a token for the last item in // a page, and the client sends it back to request the next page. // // A cursor token contains: // - sort_key: the value of the primary ordering column (e.g. created_at) // - tie_breaker: a secondary stable key (usually a unique ID) to make ordering // deterministic when multiple rows share the same sort_key value // // This matches common SQL ordering like: // // ORDER BY created_at DESC, id DESC // // and next-page predicate like: // // WHERE (created_at, id) < (:created_at, :id) // // (for descending order). type CursorTokenData struct { SortKey string `json:"sort_key"` TieBreaker string `json:"tie_breaker"` } // ErrNoCursor indicates that no cursor was provided (e.g. first page request). var ErrNoCursor = errors.New("no cursor") // EncodeTokenPaginationWithCursor encodes cursor token data into a TokenPagination. func EncodeTokenPaginationWithCursor(data CursorTokenData) (*TokenPagination, error) { token, err := EncodeCursorToken(data) if err != nil { return nil, err } return &TokenPagination{Token: token}, nil } // DecodeTokenPaginationWithCursor decodes a TokenPagination into CursorTokenData. // Returns ErrNoCursor if the TokenPagination is nil or has an empty token. func DecodeTokenPaginationWithCursor(tp *TokenPagination) (data CursorTokenData, err error) { if tp == nil || tp.Token == "" { return CursorTokenData{}, ErrNoCursor } data, err = DecodeCursorToken(tp.Token) if err != nil { return CursorTokenData{}, err } return data, nil } func EncodeCursorToken(data CursorTokenData) (string, error) { jsonData, err := json.Marshal(data) if err != nil { return "", fmt.Errorf("failed to marshal cursor token data: %w", err) } return base64.URLEncoding.EncodeToString(jsonData), nil } func DecodeCursorToken(token string) (CursorTokenData, error) { decoded, err := base64.URLEncoding.DecodeString(token) if err != nil { return CursorTokenData{}, fmt.Errorf("failed to decode cursor token: %w", err) } var data CursorTokenData if err := json.Unmarshal(decoded, &data); err != nil { return CursorTokenData{}, fmt.Errorf("failed to unmarshal cursor token data: %w", err) } return data, nil } type ScalarCursorCodec[T any] struct { Encode func(T) (string, error) Decode func(string) (T, error) } type CursorCodec[S, T any] struct { SortKey ScalarCursorCodec[S] TieBreaker ScalarCursorCodec[T] } func (c CursorCodec[P, T]) Decode(tp *TokenPagination) (primary P, tie T, err error) { data, err := DecodeTokenPaginationWithCursor(tp) if err != nil { return *new(P), *new(T), err } if data.SortKey == "" && data.TieBreaker == "" { return *new(P), *new(T), ErrNoCursor } primary, err = c.SortKey.Decode(data.SortKey) if err != nil { return *new(P), *new(T), fmt.Errorf("decode primary: %w", err) } tie, err = c.TieBreaker.Decode(data.TieBreaker) if err != nil { return *new(P), *new(T), fmt.Errorf("decode tie: %w", err) } return primary, tie, nil } func (c CursorCodec[P, T]) Encode(primary P, tie T) (*TokenPagination, error) { sortKey, err := c.SortKey.Encode(primary) if err != nil { return nil, fmt.Errorf("encode primary: %w", err) } tieKey, err := c.TieBreaker.Encode(tie) if err != nil { return nil, fmt.Errorf("encode tie: %w", err) } if sortKey == "" && tieKey == "" { return nil, nil } return EncodeTokenPaginationWithCursor(CursorTokenData{ SortKey: sortKey, TieBreaker: tieKey, }) } var StringCursorCodec = ScalarCursorCodec[string]{ Encode: func(s string) (string, error) { return s, nil }, Decode: func(s string) (string, error) { return s, nil }, } var TimeRFC3339NanoCursorCodec = ScalarCursorCodec[time.Time]{ Encode: func(t time.Time) (string, error) { return t.UTC().Format(time.RFC3339Nano), nil }, Decode: func(s string) (time.Time, error) { return time.Parse(time.RFC3339Nano, s) }, } type pageIterReq[T any] interface { GetPage() *Pagination Filter(v T) bool } type pageIterResp[T any] interface { GetItems() []T } type PageItem interface { ID() string } func ListPageIter[T PageItem, Req pageIterReq[T], Resp pageIterResp[T]](fnc func(ctx context.Context, req Req) (Resp, error), req Req) iters.PageIter[T] { it := &listPageIter[T, Req, Resp]{fnc: fnc, req: req} return iters.FilterPage(it, func(v T) bool { return req.Filter(v) }) } type listPageIter[T PageItem, Req pageIterReq[T], Resp pageIterResp[T]] struct { fnc func(ctx context.Context, opts Req) (Resp, error) req Req done bool } func (it *listPageIter[T, Req, Resp]) NextPage(ctx context.Context) ([]T, error) { if it.done { return nil, io.EOF } opts := it.req.GetPage() resp, err := it.fnc(ctx, it.req) page := resp.GetItems() if opts == nil { // No pagination set - returns all items. // We have to do this to support legacy implementations. it.done = true return page, err } // Advance pagination cursor. hasID := false for i := len(page) - 1; i >= 0; i-- { if id := page[i].ID(); id > opts.AfterId { opts.AfterId = id hasID = true } } if err == nil && (len(page) == 0 || !hasID) { err = io.EOF it.done = true } return page, err } func (it *listPageIter[_, _, _]) Close() { it.done = true } func (p *ListUpdate) Validate() error { if p == nil { return nil } change := len(p.Set)+len(p.Add)+len(p.Remove) > 0 if !p.Clear && !change { return fmt.Errorf("unsupported list update operation") } if p.Clear && change { return fmt.Errorf("cannot clear and change the list at the same time") } if len(p.Set) > 0 && len(p.Add)+len(p.Remove) > 0 { return fmt.Errorf("cannot set and change the list at the same time") } for _, v := range p.Set { if v == "" { return fmt.Errorf("empty element in the list") } } for _, v := range p.Add { if v == "" { return fmt.Errorf("empty element in the list") } } for _, v := range p.Remove { if v == "" { return fmt.Errorf("empty element in the list") } } return nil } func (p *ListUpdate) Apply(arr []string) ([]string, error) { if err := p.Validate(); err != nil { return arr, err } applyListUpdate(&arr, p) return arr, nil } func applyUpdate[T any](dst *T, set *T) { if set != nil { *dst = *set } } func applyUpdatePtr[T any](dst **T, set *T) { if set != nil { *dst = set } } func applyListUpdate[T ~string](dst *[]T, u *ListUpdate) { if u == nil { return } if u.Clear { *dst = nil return } if len(u.Set) != 0 { arr := make([]T, 0, len(u.Set)) for _, v := range u.Set { arr = append(arr, T(v)) } *dst = arr return } arr := slices.Clone(*dst) for _, v := range u.Remove { if i := slices.Index(arr, T(v)); i >= 0 { arr = slices.Delete(arr, i, i+1) } } for _, v := range u.Add { if i := slices.Index(arr, T(v)); i < 0 { arr = append(arr, T(v)) } } *dst = arr } func applyMapDiff(dst *map[string]string, diff map[string]string) { m := *dst for k, v := range diff { if v != "" { if m == nil { m = make(map[string]string) } m[k] = v } else { delete(m, k) } } *dst = m } // ToProto implements DataPacket in Go SDK. func (p *ChatMessage) ToProto() *DataPacket { return &DataPacket{ Value: &DataPacket_ChatMessage{ ChatMessage: p, }, } } // ToProto implements DataPacket in Go SDK. func (p *DataStream_Header) ToProto() *DataPacket { return &DataPacket{ Value: &DataPacket_StreamHeader{ StreamHeader: p, }, } } // ToProto implements DataPacket in Go SDK. func (p *DataStream_Chunk) ToProto() *DataPacket { return &DataPacket{ Value: &DataPacket_StreamChunk{ StreamChunk: p, }, } } // ToProto implements DataPacket in Go SDK. func (p *DataStream_Trailer) ToProto() *DataPacket { return &DataPacket{ Value: &DataPacket_StreamTrailer{ StreamTrailer: p, }, } }