diff --git a/config-sample.yaml b/config-sample.yaml index 459b693f65b..07c5953618a 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -242,6 +242,28 @@ keys: # urls: # - https://your-host.com/handler +# Room-level webhooks +# in addition to the global `webhook` config above, individual rooms can carry their own +# webhook endpoints. Set them per room on CreateRoom (`webhooks` field), or attach them to +# a named room preset here and create the room with `room_preset: `. +# Room webhooks receive room_started, room_finished, participant_joined, participant_left, +# participant_connection_aborted, track_published and track_unpublished. Egress and ingress +# events are never delivered to them; those carry their own webhooks on the egress request. +# room: +# room_configurations: +# support-call: +# empty_timeout: 300 +# webhooks: +# - url: https://your-host.com/support-hook +# # api key used to sign the message; must be one of the keys below. +# # leave unset to sign with the key from the global `webhook` config +# signing_key: +# # optional; include_events takes precedence over exclude_events +# filter_params: +# include_events: +# - participant_joined +# - participant_left + # Signal Relay # since v1.4.0, a more reliable, psrpc based signal relay is available # this gives us the ability to reliably proxy messages between a signal server and RTC node diff --git a/go.mod b/go.mod index 4f13448d4a5..1cb075f0ba0 100644 --- a/go.mod +++ b/go.mod @@ -168,3 +168,5 @@ replace github.com/pion/webrtc/v4 => github.com/livekit/webrtc-pion/v4 v4.2.18-w replace github.com/pion/dtls/v3 => github.com/livekit/dtls/v3 v3.1.5-warp.1 replace github.com/pion/ice/v4 => github.com/livekit/ice/v4 v4.4.0-warp.2 + +replace github.com/livekit/protocol => github.com/Darshak03/protocol v1.51.1-0.20260907191956-dcc0838d4b61 diff --git a/go.sum b/go.sum index 083396f1338..9114ee80ada 100644 --- a/go.sum +++ b/go.sum @@ -8,6 +8,8 @@ cel.dev/cel-go v0.32.0 h1:irvpFKr5EuGPyxeME03ERh0rii1TX+BDAnB9eL3IvNk= cel.dev/cel-go v0.32.0/go.mod h1:DnVip7tpJSsgZymwfT+m1tnEVy3ivAjSMXPx12YrMkU= cel.dev/expr v0.25.3 h1:A2jO8jwOugrrovveCWfj0KEZOfqiLgAcwjpHPhzIGw0= cel.dev/expr v0.25.3/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= +github.com/Darshak03/protocol v1.51.1-0.20260907191956-dcc0838d4b61 h1:6WEMZWB+K62S9xS58eyI949dAZNH7Jxgf8Wet76YE2o= +github.com/Darshak03/protocol v1.51.1-0.20260907191956-dcc0838d4b61/go.mod h1:zxowkRnQlJ2VMn6ZyinXMDi985wcKXuWNeXmEERqFAs= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= @@ -164,8 +166,6 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095 h1:BcliKAXoMhl/nWmzQweQ5kmh4Qqagxl4s3Z5pvM/7AY= github.com/livekit/mediatransportutil v0.0.0-20260821083140-f234b534b095/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.51.1-0.20260905133529-a4f4b5c0c23f h1:+48IWNrsoTgbB0JGv+xJl4umx6e/vh1BX1Tcokuacwc= -github.com/livekit/protocol v1.51.1-0.20260905133529-a4f4b5c0c23f/go.mod h1:zxowkRnQlJ2VMn6ZyinXMDi985wcKXuWNeXmEERqFAs= github.com/livekit/psrpc v0.7.6 h1:YG07lUMTtf+eaYI2goT9zcVZ0kGJNWN1K6ETNFtv1HQ= github.com/livekit/psrpc v0.7.6/go.mod h1:DMw15RO7x5XmcgfwzWJYk2In605kx+wu1QRVbPfzf8M= github.com/livekit/webrtc-pion/v4 v4.2.18-warp.1 h1:fH+v4W+NFp9FfPzON6FaUFNmazGcctaAhb2P+Ksf+1s= diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 76224e73173..54b0372f2cd 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1268,6 +1268,7 @@ func (r *Room) onStateChange(p types.LocalParticipant) { false, p.IsWarpEnabled(), p.TelemetryGuard(), + r.Internal().GetWebhooks(), ) p.GetReporter().Tx(func(tx roomobs.ParticipantSessionTx) { @@ -2057,11 +2058,11 @@ func (l participantTelemetryListener) OnTrackPublishRequested(pID livekit.Partic } func (l participantTelemetryListener) OnTrackPublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) { - l.room.telemetry.TrackPublished(context.Background(), l.eventRoom(), pID, identity, ti, shouldSendEvent) + l.room.telemetry.TrackPublished(context.Background(), l.eventRoom(), pID, identity, ti, shouldSendEvent, l.room.Internal().GetWebhooks()) } func (l participantTelemetryListener) OnTrackUnpublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool) { - l.room.telemetry.TrackUnpublished(context.Background(), l.eventRoom(), pID, identity, ti, wasPublishedLocally, shouldSendEvent) + l.room.telemetry.TrackUnpublished(context.Background(), l.eventRoom(), pID, identity, ti, wasPublishedLocally, shouldSendEvent, l.room.Internal().GetWebhooks()) } func (l participantTelemetryListener) OnTrackSubscribeRequested(pID livekit.ParticipantID, ti *livekit.TrackInfo) { diff --git a/pkg/rtc/signalanddatastats.go b/pkg/rtc/signalanddatastats.go index fd7227d6e6f..19e564a7e22 100644 --- a/pkg/rtc/signalanddatastats.go +++ b/pkg/rtc/signalanddatastats.go @@ -291,7 +291,7 @@ func (s *BytesSignalStats) maybeStart() { func (s *BytesSignalStats) worker() { s.BytesTrackStats.worker() - s.telemetry.ParticipantLeft(s.ctx, s.ri, s.pi, false, s.guard) + s.telemetry.ParticipantLeft(s.ctx, s.ri, s.pi, false, s.guard, nil) close(s.stopped) } diff --git a/pkg/service/roomallocator.go b/pkg/service/roomallocator.go index 436a433c056..3316dfbc422 100644 --- a/pkg/service/roomallocator.go +++ b/pkg/service/roomallocator.go @@ -113,6 +113,9 @@ func (r *StandardRoomAllocator) CreateRoom(ctx context.Context, req *livekit.Cre if req.Agents != nil { internal.AgentDispatches = req.Agents } + if req.Webhooks != nil { + internal.Webhooks = req.Webhooks + } if req.MinPlayoutDelay > 0 || req.MaxPlayoutDelay > 0 { internal.PlayoutDelay = &livekit.PlayoutDelay{ Enabled: true, @@ -243,6 +246,12 @@ func (r *StandardRoomAllocator) applyNamedRoomConfiguration(req *livekit.CreateR if clone.Metadata == "" { clone.Metadata = conf.Metadata } + if clone.Webhooks == nil && len(conf.Webhooks) > 0 { + clone.Webhooks = make([]*livekit.WebhookConfig, 0, len(conf.Webhooks)) + for _, wh := range conf.Webhooks { + clone.Webhooks = append(clone.Webhooks, utils.CloneProto(wh)) + } + } return clone, nil } diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index e25fae88321..4686d30472b 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -617,7 +617,7 @@ func (r *RoomManager) StartSession( // update room store with new numParticipants proto := room.ToProto() persistRoomForParticipantCount(proto) - r.telemetry.ParticipantLeft(ctx, proto, p.ToProto(), true, participant.TelemetryGuard()) + r.telemetry.ParticipantLeft(ctx, proto, p.ToProto(), true, participant.TelemetryGuard(), room.Internal().GetWebhooks()) }) participant.OnClaimsChanged(func(participant types.LocalParticipant) { pLogger.Debugw("refreshing client token after claims change") @@ -697,7 +697,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, createRoom *livekit.C killDispServer() roomInfo := newRoom.ToProto() - r.telemetry.RoomEnded(ctx, roomInfo, reason.ToProto()) + r.telemetry.RoomEnded(ctx, roomInfo, reason.ToProto(), newRoom.Internal().GetWebhooks()) prometheus.RoomEnded(time.Unix(roomInfo.CreationTime, 0)) if err := r.deleteRoom(ctx, roomName); err != nil { newRoom.Logger().Errorw("could not delete room", err) @@ -726,7 +726,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, createRoom *livekit.C newRoom.Hold() - r.telemetry.RoomStarted(ctx, newRoom.ToProto()) + r.telemetry.RoomStarted(ctx, newRoom.ToProto(), newRoom.Internal().GetWebhooks()) prometheus.RoomStarted() if created && createRoom.GetEgress().GetRoom() != nil { diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 2cdf373a2ef..a55432b12d7 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -17,6 +17,7 @@ package service import ( "context" "fmt" + "net/url" "strconv" "github.com/twitchtv/twirp" @@ -24,6 +25,7 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/rtc" + "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" @@ -33,6 +35,7 @@ import ( type RoomService struct { limitConf config.LimitConfig apiConf config.APIConfig + keyProvider auth.KeyProvider router routing.MessageRouter roomAllocator RoomAllocator roomStore ServiceStore @@ -48,6 +51,7 @@ type RoomService struct { func NewRoomService( limitConf config.LimitConfig, apiConf config.APIConfig, + keyProvider auth.KeyProvider, router routing.MessageRouter, roomAllocator RoomAllocator, serviceStore ServiceStore, @@ -59,6 +63,7 @@ func NewRoomService( svc = &RoomService{ limitConf: limitConf, apiConf: apiConf, + keyProvider: keyProvider, router: router, roomAllocator: roomAllocator, roomStore: serviceStore, @@ -97,6 +102,10 @@ func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq } } + if err := s.validateWebhooks(req.Webhooks); err != nil { + return nil, err + } + err := s.roomAllocator.SelectRoomNode(ctx, livekit.RoomName(req.Name), livekit.NodeID(req.NodeId)) if err != nil { return nil, err @@ -107,6 +116,22 @@ func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq return room, err } +// validateWebhooks rejects room-scoped webhook configs that could never deliver, so +// the caller learns at CreateRoom time rather than from a per-event warning log. +func (s *RoomService) validateWebhooks(webhooks []*livekit.WebhookConfig) error { + for _, wh := range webhooks { + u, err := url.Parse(wh.Url) + if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" { + return twirp.InvalidArgumentError("webhooks.url", "must be an absolute http(s) URL") + } + // an empty signing key means "use the default key from the webhook config" + if wh.SigningKey != "" && s.keyProvider.GetSecret(wh.SigningKey) == "" { + return twirp.InvalidArgumentError("webhooks.signing_key", "unknown api key") + } + } + return nil +} + func (s *RoomService) ListRooms(ctx context.Context, req *livekit.ListRoomsRequest) (*livekit.ListRoomsResponse, error) { RecordRequest(ctx, req) diff --git a/pkg/service/roomservice_test.go b/pkg/service/roomservice_test.go index 5b27f2fdf6f..c9958c446c1 100644 --- a/pkg/service/roomservice_test.go +++ b/pkg/service/roomservice_test.go @@ -182,6 +182,7 @@ func newTestRoomService(limitConf config.LimitConfig) *TestRoomService { svc, err := service.NewRoomService( limitConf, config.APIConfig{ExecutionTimeout: 2}, + auth.NewFileBasedKeyProviderFromMap(map[string]string{"APIkey": "secret-that-is-long-enough"}), router, allocator, store, diff --git a/pkg/service/roomwebhook_test.go b/pkg/service/roomwebhook_test.go new file mode 100644 index 00000000000..b70328ab623 --- /dev/null +++ b/pkg/service/roomwebhook_test.go @@ -0,0 +1,145 @@ +// Copyright 2026 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 service_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "github.com/twitchtv/twirp" + + "github.com/livekit/protocol/auth" + "github.com/livekit/protocol/livekit" + + "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/livekit-server/pkg/service" +) + +func TestCreateRoomWebhooks(t *testing.T) { + newAllocator := func(t *testing.T, mut func(*config.Config)) (service.RoomAllocator, *config.Config) { + conf, err := config.NewConfig("", true, nil, nil) + require.NoError(t, err) + if mut != nil { + mut(conf) + } + node, err := routing.NewLocalNode(conf) + require.NoError(t, err) + ra, conf := newTestRoomAllocator(t, conf, node.Clone()) + return ra, conf + } + + reqWebhook := &livekit.WebhookConfig{Url: "https://example.com/req"} + presetWebhook := &livekit.WebhookConfig{Url: "https://example.com/preset"} + + t.Run("webhooks on the request land on RoomInternal", func(t *testing.T) { + ra, _ := newAllocator(t, nil) + + _, internal, _, err := ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{ + Name: "myroom", + Webhooks: []*livekit.WebhookConfig{reqWebhook}, + }, true) + require.NoError(t, err) + require.Len(t, internal.GetWebhooks(), 1) + require.Equal(t, reqWebhook.Url, internal.GetWebhooks()[0].Url) + }) + + t.Run("a room with no webhooks configured gets none", func(t *testing.T) { + ra, _ := newAllocator(t, nil) + + _, internal, _, err := ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{Name: "myroom"}, true) + require.NoError(t, err) + require.Empty(t, internal.GetWebhooks()) + }) + + t.Run("preset webhooks apply when the request has none", func(t *testing.T) { + ra, _ := newAllocator(t, func(conf *config.Config) { + conf.Room.RoomConfigurations = map[string]*livekit.RoomConfiguration{ + "support": {Webhooks: []*livekit.WebhookConfig{presetWebhook}}, + } + }) + + _, internal, _, err := ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{ + Name: "myroom", + RoomPreset: "support", + }, true) + require.NoError(t, err) + require.Len(t, internal.GetWebhooks(), 1) + require.Equal(t, presetWebhook.Url, internal.GetWebhooks()[0].Url) + }) + + t.Run("request webhooks win over the preset", func(t *testing.T) { + ra, _ := newAllocator(t, func(conf *config.Config) { + conf.Room.RoomConfigurations = map[string]*livekit.RoomConfiguration{ + "support": {Webhooks: []*livekit.WebhookConfig{presetWebhook}}, + } + }) + + _, internal, _, err := ra.CreateRoom(context.Background(), &livekit.CreateRoomRequest{ + Name: "myroom", + RoomPreset: "support", + Webhooks: []*livekit.WebhookConfig{reqWebhook}, + }, true) + require.NoError(t, err) + require.Len(t, internal.GetWebhooks(), 1) + require.Equal(t, reqWebhook.Url, internal.GetWebhooks()[0].Url) + }) +} + +func TestCreateRoomWebhookValidation(t *testing.T) { + createCtx := func() context.Context { + return service.WithGrants(context.Background(), &auth.ClaimGrants{Video: &auth.VideoGrant{RoomCreate: true}}, "") + } + requireInvalidArg := func(t *testing.T, err error) { + t.Helper() + terr, ok := err.(twirp.Error) + require.True(t, ok, "expected twirp error, got %T (%v)", err, err) + require.Equal(t, twirp.InvalidArgument, terr.Code()) + } + + // newTestRoomService registers "APIkey" with the key provider + for _, tc := range []struct { + name string + wh *livekit.WebhookConfig + }{ + {"empty url", &livekit.WebhookConfig{}}, + {"relative url", &livekit.WebhookConfig{Url: "/hook"}}, + {"non-http scheme", &livekit.WebhookConfig{Url: "ftp://example.com/hook"}}, + {"no host", &livekit.WebhookConfig{Url: "https:///hook"}}, + {"unknown signing key", &livekit.WebhookConfig{Url: "https://example.com/hook", SigningKey: "nope"}}, + } { + t.Run(tc.name, func(t *testing.T) { + svc := newTestRoomService(config.LimitConfig{}) + _, err := svc.CreateRoom(createCtx(), &livekit.CreateRoomRequest{ + Name: "myroom", + Webhooks: []*livekit.WebhookConfig{tc.wh}, + }) + requireInvalidArg(t, err) + }) + } + + t.Run("valid webhook is accepted", func(t *testing.T) { + svc := newTestRoomService(config.LimitConfig{}) + _, err := svc.CreateRoom(createCtx(), &livekit.CreateRoomRequest{ + Name: "myroom", + Webhooks: []*livekit.WebhookConfig{ + {Url: "https://example.com/hook", SigningKey: "APIkey"}, + {Url: "http://example.com/hook2"}, // empty signing key means the default + }, + }) + require.NoError(t, err) + }) +} diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index f7c5a626775..cacb28e49f1 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -34,6 +34,10 @@ import ( func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*LivekitServer, error) { limitConfig := getLimitConf(conf) apiConfig := getAPIConf(conf) + keyProvider, err := createKeyProvider(conf) + if err != nil { + return nil, err + } universalClient, err := createRedisClient(conf) if err != nil { return nil, err @@ -70,10 +74,6 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live egressStore := getEgressStore(objectStore) ingressStore := getIngressStore(objectStore) sipStore := getSIPStore(objectStore) - keyProvider, err := createKeyProvider(conf) - if err != nil { - return nil, err - } queuedNotifier, err := createWebhookNotifier(conf, keyProvider) if err != nil { return nil, err @@ -94,7 +94,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live if err != nil { return nil, err } - roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v2, v3) + roomService, err := NewRoomService(limitConfig, apiConfig, keyProvider, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v2, v3) if err != nil { return nil, err } diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index 657ea69e813..092e43dcaf7 100644 --- a/pkg/telemetry/events.go +++ b/pkg/telemetry/events.go @@ -42,12 +42,12 @@ func (t *telemetryService) NotifyEvent(ctx context.Context, event *livekit.Webho } } -func (t *telemetryService) RoomStarted(ctx context.Context, room *livekit.Room) { +func (t *telemetryService) RoomStarted(ctx context.Context, room *livekit.Room, roomWebhooks []*livekit.WebhookConfig) { t.enqueue(func() { t.NotifyEvent(ctx, &livekit.WebhookEvent{ Event: webhook.EventRoomStarted, Room: room, - }) + }, webhook.GetRoomNotifyOptions(webhook.EventRoomStarted, roomWebhooks)...) t.SendEvent(ctx, &livekit.AnalyticsEvent{ Type: livekit.AnalyticsEventType_ROOM_CREATED, @@ -57,13 +57,13 @@ func (t *telemetryService) RoomStarted(ctx context.Context, room *livekit.Room) }) } -func (t *telemetryService) RoomEnded(ctx context.Context, room *livekit.Room, reason livekit.RoomEndReason) { +func (t *telemetryService) RoomEnded(ctx context.Context, room *livekit.Room, reason livekit.RoomEndReason, roomWebhooks []*livekit.WebhookConfig) { t.enqueue(func() { t.NotifyEvent(ctx, &livekit.WebhookEvent{ Event: webhook.EventRoomFinished, Room: room, RoomEndReason: reason, - }) + }, webhook.GetRoomNotifyOptions(webhook.EventRoomFinished, roomWebhooks)...) t.SendEvent(ctx, &livekit.AnalyticsEvent{ Type: livekit.AnalyticsEventType_ROOM_ENDED, @@ -115,6 +115,7 @@ func (t *telemetryService) ParticipantActive( isMigration bool, isWarp bool, guard *ReferenceGuard, + roomWebhooks []*livekit.WebhookConfig, ) { t.enqueue(func() { if !isMigration { @@ -123,7 +124,7 @@ func (t *telemetryService) ParticipantActive( Event: webhook.EventParticipantJoined, Room: room, Participant: participant, - }) + }, webhook.GetRoomNotifyOptions(webhook.EventParticipantJoined, roomWebhooks)...) } worker, found := t.getOrCreateWorker( @@ -201,6 +202,7 @@ func (t *telemetryService) ParticipantLeft(ctx context.Context, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard, + roomWebhooks []*livekit.WebhookConfig, ) { t.enqueue(func() { isConnected := false @@ -232,7 +234,7 @@ func (t *telemetryService) ParticipantLeft(ctx context.Context, Event: webhookEvent, Room: room, Participant: participant, - }) + }, webhook.GetRoomNotifyOptions(webhookEvent, roomWebhooks)...) t.SendEvent(ctx, newParticipantEvent(analyticsEvent, room, participant)) } @@ -268,6 +270,7 @@ func (t *telemetryService) TrackPublished( identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool, + roomWebhooks []*livekit.WebhookConfig, ) { t.enqueue(func() { prometheus.AddPublishedTrack(track.Type.String()) @@ -285,7 +288,7 @@ func (t *telemetryService) TrackPublished( Room: room, Participant: participant, Track: track, - }) + }, webhook.GetRoomNotifyOptions(webhook.EventTrackPublished, roomWebhooks)...) ev := newTrackEvent(livekit.AnalyticsEventType_TRACK_PUBLISHED, room, participantID, track) ev.Participant = participant @@ -398,6 +401,7 @@ func (t *telemetryService) TrackUnpublished( track *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool, + roomWebhooks []*livekit.WebhookConfig, ) { t.enqueue(func() { if wasPublishedLocally { @@ -416,7 +420,7 @@ func (t *telemetryService) TrackUnpublished( Room: room, Participant: participant, Track: track, - }) + }, webhook.GetRoomNotifyOptions(webhook.EventTrackUnpublished, roomWebhooks)...) t.SendEvent(ctx, newTrackEvent(livekit.AnalyticsEventType_TRACK_UNPUBLISHED, room, participantID, track)) }) diff --git a/pkg/telemetry/events_test.go b/pkg/telemetry/events_test.go index ae275c89fff..ea6973c2dca 100644 --- a/pkg/telemetry/events_test.go +++ b/pkg/telemetry/events_test.go @@ -86,8 +86,8 @@ func Test_OnParticipantLeft_EventIsSent(t *testing.T) { guard := &telemetry.ReferenceGuard{} // do - fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false, false, guard) - fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard) + fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false, false, guard, nil) + fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard, nil) time.Sleep(time.Millisecond * 500) // test @@ -179,7 +179,7 @@ func Test_OnParticipantActive_EventIsSent(t *testing.T) { ClientConnectTime: 420, } - fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false, false, guard) + fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false, false, guard, nil) time.Sleep(time.Millisecond * 500) require.Equal(t, 2, fixture.analytics.SendEventCallCount()) @@ -248,7 +248,7 @@ func Test_OnRoomEnded_ReasonIsSent(t *testing.T) { fixture := createFixture() room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"} - fixture.sut.RoomEnded(context.Background(), room, livekit.RoomEndReason_ROOM_END_API_DELETE) + fixture.sut.RoomEnded(context.Background(), room, livekit.RoomEndReason_ROOM_END_API_DELETE, nil) fixture.flush() diff --git a/pkg/telemetry/stats_test.go b/pkg/telemetry/stats_test.go index f4f9d3dbefe..4bd86b1eb3c 100644 --- a/pkg/telemetry/stats_test.go +++ b/pkg/telemetry/stats_test.go @@ -479,7 +479,7 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) { require.True(t, found2) // remove 1 track - track stats were flushed above, so no more calls to SendStats - fixture.sut.TrackUnpublished(context.Background(), room, partSID, identity, &livekit.TrackInfo{Sid: string(trackID2)}, true, true) + fixture.sut.TrackUnpublished(context.Background(), room, partSID, identity, &livekit.TrackInfo{Sid: string(trackID2)}, true, true, nil) // flush fixture.flush() @@ -498,7 +498,7 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) { fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard) // do - fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard) + fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard, nil) // should not be called if there are no track stats time.Sleep(time.Millisecond * 500) @@ -665,7 +665,7 @@ func Test_RoomIDChangeReKeysStatsWorkers(t *testing.T) { require.Equal(t, uint64(44), byRoom[restartedRoom.Sid].Streams[0].PrimaryBytes) // the worker moved rather than being duplicated, so closing it out drains everything - fixture.sut.ParticipantLeft(context.Background(), restartedRoom, participantInfo, true, guard) + fixture.sut.ParticipantLeft(context.Background(), restartedRoom, participantInfo, true, guard, nil) fixture.flush() require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) } diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index 12d5421279d..ce734612fa5 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -83,7 +83,7 @@ type FakeTelemetryService struct { arg2 string arg3 *livekit.EgressInfo } - ParticipantActiveStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard) + ParticipantActiveStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard, []*livekit.WebhookConfig) participantActiveMutex sync.RWMutex participantActiveArgsForCall []struct { arg1 context.Context @@ -93,6 +93,7 @@ type FakeTelemetryService struct { arg5 bool arg6 bool arg7 *telemetry.ReferenceGuard + arg8 []*livekit.WebhookConfig } ParticipantJoinedStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) participantJoinedMutex sync.RWMutex @@ -105,7 +106,7 @@ type FakeTelemetryService struct { arg6 bool arg7 *telemetry.ReferenceGuard } - ParticipantLeftStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard) + ParticipantLeftStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard, []*livekit.WebhookConfig) participantLeftMutex sync.RWMutex participantLeftArgsForCall []struct { arg1 context.Context @@ -113,6 +114,7 @@ type FakeTelemetryService struct { arg3 *livekit.ParticipantInfo arg4 bool arg5 *telemetry.ReferenceGuard + arg6 []*livekit.WebhookConfig } ParticipantResumedStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, livekit.NodeID, livekit.ReconnectReason) participantResumedMutex sync.RWMutex @@ -129,12 +131,13 @@ type FakeTelemetryService struct { arg1 context.Context arg2 *livekit.ReportInfo } - RoomEndedStub func(context.Context, *livekit.Room, livekit.RoomEndReason) + RoomEndedStub func(context.Context, *livekit.Room, livekit.RoomEndReason, []*livekit.WebhookConfig) roomEndedMutex sync.RWMutex roomEndedArgsForCall []struct { arg1 context.Context arg2 *livekit.Room arg3 livekit.RoomEndReason + arg4 []*livekit.WebhookConfig } RoomIDChangedStub func(context.Context, livekit.RoomID, *livekit.Room) roomIDChangedMutex sync.RWMutex @@ -154,11 +157,12 @@ type FakeTelemetryService struct { roomProjectReporterReturnsOnCall map[int]struct { result1 roomobs.ProjectReporter } - RoomStartedStub func(context.Context, *livekit.Room) + RoomStartedStub func(context.Context, *livekit.Room, []*livekit.WebhookConfig) roomStartedMutex sync.RWMutex roomStartedArgsForCall []struct { arg1 context.Context arg2 *livekit.Room + arg3 []*livekit.WebhookConfig } SendEventStub func(context.Context, *livekit.AnalyticsEvent) sendEventMutex sync.RWMutex @@ -217,7 +221,7 @@ type FakeTelemetryService struct { arg5 *livekit.TrackInfo arg6 bool } - TrackPublishedStub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) + TrackPublishedStub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, []*livekit.WebhookConfig) trackPublishedMutex sync.RWMutex trackPublishedArgsForCall []struct { arg1 context.Context @@ -226,6 +230,7 @@ type FakeTelemetryService struct { arg4 livekit.ParticipantIdentity arg5 *livekit.TrackInfo arg6 bool + arg7 []*livekit.WebhookConfig } TrackPublishedUpdateStub func(context.Context, *livekit.Room, livekit.ParticipantID, *livekit.TrackInfo) trackPublishedUpdateMutex sync.RWMutex @@ -289,7 +294,7 @@ type FakeTelemetryService struct { arg3 livekit.ParticipantID arg4 *livekit.TrackInfo } - TrackUnpublishedStub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool) + TrackUnpublishedStub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool, []*livekit.WebhookConfig) trackUnpublishedMutex sync.RWMutex trackUnpublishedArgsForCall []struct { arg1 context.Context @@ -299,6 +304,7 @@ type FakeTelemetryService struct { arg5 *livekit.TrackInfo arg6 bool arg7 bool + arg8 []*livekit.WebhookConfig } TrackUnsubscribedStub func(context.Context, *livekit.Room, livekit.ParticipantID, *livekit.TrackInfo, bool) trackUnsubscribedMutex sync.RWMutex @@ -707,7 +713,12 @@ func (fake *FakeTelemetryService) NotifyEgressEventArgsForCall(i int) (context.C return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } -func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.AnalyticsClientMeta, arg5 bool, arg6 bool, arg7 *telemetry.ReferenceGuard) { +func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.AnalyticsClientMeta, arg5 bool, arg6 bool, arg7 *telemetry.ReferenceGuard, arg8 []*livekit.WebhookConfig) { + var arg8Copy []*livekit.WebhookConfig + if arg8 != nil { + arg8Copy = make([]*livekit.WebhookConfig, len(arg8)) + copy(arg8Copy, arg8) + } fake.participantActiveMutex.Lock() fake.participantActiveArgsForCall = append(fake.participantActiveArgsForCall, struct { arg1 context.Context @@ -717,12 +728,13 @@ func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 * arg5 bool arg6 bool arg7 *telemetry.ReferenceGuard - }{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) + arg8 []*livekit.WebhookConfig + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8Copy}) stub := fake.ParticipantActiveStub - fake.recordInvocation("ParticipantActive", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) + fake.recordInvocation("ParticipantActive", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8Copy}) fake.participantActiveMutex.Unlock() if stub != nil { - fake.ParticipantActiveStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7) + fake.ParticipantActiveStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8) } } @@ -732,17 +744,17 @@ func (fake *FakeTelemetryService) ParticipantActiveCallCount() int { return len(fake.participantActiveArgsForCall) } -func (fake *FakeTelemetryService) ParticipantActiveCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard)) { +func (fake *FakeTelemetryService) ParticipantActiveCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard, []*livekit.WebhookConfig)) { fake.participantActiveMutex.Lock() defer fake.participantActiveMutex.Unlock() fake.ParticipantActiveStub = stub } -func (fake *FakeTelemetryService) ParticipantActiveArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard) { +func (fake *FakeTelemetryService) ParticipantActiveArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, bool, *telemetry.ReferenceGuard, []*livekit.WebhookConfig) { fake.participantActiveMutex.RLock() defer fake.participantActiveMutex.RUnlock() argsForCall := fake.participantActiveArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7, argsForCall.arg8 } func (fake *FakeTelemetryService) ParticipantJoined(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.ClientInfo, arg5 *livekit.AnalyticsClientMeta, arg6 bool, arg7 *telemetry.ReferenceGuard) { @@ -783,7 +795,12 @@ func (fake *FakeTelemetryService) ParticipantJoinedArgsForCall(i int) (context.C return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7 } -func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 bool, arg5 *telemetry.ReferenceGuard) { +func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 bool, arg5 *telemetry.ReferenceGuard, arg6 []*livekit.WebhookConfig) { + var arg6Copy []*livekit.WebhookConfig + if arg6 != nil { + arg6Copy = make([]*livekit.WebhookConfig, len(arg6)) + copy(arg6Copy, arg6) + } fake.participantLeftMutex.Lock() fake.participantLeftArgsForCall = append(fake.participantLeftArgsForCall, struct { arg1 context.Context @@ -791,12 +808,13 @@ func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *li arg3 *livekit.ParticipantInfo arg4 bool arg5 *telemetry.ReferenceGuard - }{arg1, arg2, arg3, arg4, arg5}) + arg6 []*livekit.WebhookConfig + }{arg1, arg2, arg3, arg4, arg5, arg6Copy}) stub := fake.ParticipantLeftStub - fake.recordInvocation("ParticipantLeft", []interface{}{arg1, arg2, arg3, arg4, arg5}) + fake.recordInvocation("ParticipantLeft", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6Copy}) fake.participantLeftMutex.Unlock() if stub != nil { - fake.ParticipantLeftStub(arg1, arg2, arg3, arg4, arg5) + fake.ParticipantLeftStub(arg1, arg2, arg3, arg4, arg5, arg6) } } @@ -806,17 +824,17 @@ func (fake *FakeTelemetryService) ParticipantLeftCallCount() int { return len(fake.participantLeftArgsForCall) } -func (fake *FakeTelemetryService) ParticipantLeftCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard)) { +func (fake *FakeTelemetryService) ParticipantLeftCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard, []*livekit.WebhookConfig)) { fake.participantLeftMutex.Lock() defer fake.participantLeftMutex.Unlock() fake.ParticipantLeftStub = stub } -func (fake *FakeTelemetryService) ParticipantLeftArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard) { +func (fake *FakeTelemetryService) ParticipantLeftArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard, []*livekit.WebhookConfig) { fake.participantLeftMutex.RLock() defer fake.participantLeftMutex.RUnlock() argsForCall := fake.participantLeftArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6 } func (fake *FakeTelemetryService) ParticipantResumed(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 livekit.NodeID, arg5 livekit.ReconnectReason) { @@ -888,18 +906,24 @@ func (fake *FakeTelemetryService) ReportArgsForCall(i int) (context.Context, *li return argsForCall.arg1, argsForCall.arg2 } -func (fake *FakeTelemetryService) RoomEnded(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.RoomEndReason) { +func (fake *FakeTelemetryService) RoomEnded(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.RoomEndReason, arg4 []*livekit.WebhookConfig) { + var arg4Copy []*livekit.WebhookConfig + if arg4 != nil { + arg4Copy = make([]*livekit.WebhookConfig, len(arg4)) + copy(arg4Copy, arg4) + } fake.roomEndedMutex.Lock() fake.roomEndedArgsForCall = append(fake.roomEndedArgsForCall, struct { arg1 context.Context arg2 *livekit.Room arg3 livekit.RoomEndReason - }{arg1, arg2, arg3}) + arg4 []*livekit.WebhookConfig + }{arg1, arg2, arg3, arg4Copy}) stub := fake.RoomEndedStub - fake.recordInvocation("RoomEnded", []interface{}{arg1, arg2, arg3}) + fake.recordInvocation("RoomEnded", []interface{}{arg1, arg2, arg3, arg4Copy}) fake.roomEndedMutex.Unlock() if stub != nil { - fake.RoomEndedStub(arg1, arg2, arg3) + fake.RoomEndedStub(arg1, arg2, arg3, arg4) } } @@ -909,17 +933,17 @@ func (fake *FakeTelemetryService) RoomEndedCallCount() int { return len(fake.roomEndedArgsForCall) } -func (fake *FakeTelemetryService) RoomEndedCalls(stub func(context.Context, *livekit.Room, livekit.RoomEndReason)) { +func (fake *FakeTelemetryService) RoomEndedCalls(stub func(context.Context, *livekit.Room, livekit.RoomEndReason, []*livekit.WebhookConfig)) { fake.roomEndedMutex.Lock() defer fake.roomEndedMutex.Unlock() fake.RoomEndedStub = stub } -func (fake *FakeTelemetryService) RoomEndedArgsForCall(i int) (context.Context, *livekit.Room, livekit.RoomEndReason) { +func (fake *FakeTelemetryService) RoomEndedArgsForCall(i int) (context.Context, *livekit.Room, livekit.RoomEndReason, []*livekit.WebhookConfig) { fake.roomEndedMutex.RLock() defer fake.roomEndedMutex.RUnlock() argsForCall := fake.roomEndedArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } func (fake *FakeTelemetryService) RoomIDChanged(arg1 context.Context, arg2 livekit.RoomID, arg3 *livekit.Room) { @@ -1017,17 +1041,23 @@ func (fake *FakeTelemetryService) RoomProjectReporterReturnsOnCall(i int, result }{result1} } -func (fake *FakeTelemetryService) RoomStarted(arg1 context.Context, arg2 *livekit.Room) { +func (fake *FakeTelemetryService) RoomStarted(arg1 context.Context, arg2 *livekit.Room, arg3 []*livekit.WebhookConfig) { + var arg3Copy []*livekit.WebhookConfig + if arg3 != nil { + arg3Copy = make([]*livekit.WebhookConfig, len(arg3)) + copy(arg3Copy, arg3) + } fake.roomStartedMutex.Lock() fake.roomStartedArgsForCall = append(fake.roomStartedArgsForCall, struct { arg1 context.Context arg2 *livekit.Room - }{arg1, arg2}) + arg3 []*livekit.WebhookConfig + }{arg1, arg2, arg3Copy}) stub := fake.RoomStartedStub - fake.recordInvocation("RoomStarted", []interface{}{arg1, arg2}) + fake.recordInvocation("RoomStarted", []interface{}{arg1, arg2, arg3Copy}) fake.roomStartedMutex.Unlock() if stub != nil { - fake.RoomStartedStub(arg1, arg2) + fake.RoomStartedStub(arg1, arg2, arg3) } } @@ -1037,17 +1067,17 @@ func (fake *FakeTelemetryService) RoomStartedCallCount() int { return len(fake.roomStartedArgsForCall) } -func (fake *FakeTelemetryService) RoomStartedCalls(stub func(context.Context, *livekit.Room)) { +func (fake *FakeTelemetryService) RoomStartedCalls(stub func(context.Context, *livekit.Room, []*livekit.WebhookConfig)) { fake.roomStartedMutex.Lock() defer fake.roomStartedMutex.Unlock() fake.RoomStartedStub = stub } -func (fake *FakeTelemetryService) RoomStartedArgsForCall(i int) (context.Context, *livekit.Room) { +func (fake *FakeTelemetryService) RoomStartedArgsForCall(i int) (context.Context, *livekit.Room, []*livekit.WebhookConfig) { fake.roomStartedMutex.RLock() defer fake.roomStartedMutex.RUnlock() argsForCall := fake.roomStartedArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } func (fake *FakeTelemetryService) SendEvent(arg1 context.Context, arg2 *livekit.AnalyticsEvent) { @@ -1301,7 +1331,12 @@ func (fake *FakeTelemetryService) TrackPublishRequestedArgsForCall(i int) (conte return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6 } -func (fake *FakeTelemetryService) TrackPublished(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 livekit.ParticipantIdentity, arg5 *livekit.TrackInfo, arg6 bool) { +func (fake *FakeTelemetryService) TrackPublished(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 livekit.ParticipantIdentity, arg5 *livekit.TrackInfo, arg6 bool, arg7 []*livekit.WebhookConfig) { + var arg7Copy []*livekit.WebhookConfig + if arg7 != nil { + arg7Copy = make([]*livekit.WebhookConfig, len(arg7)) + copy(arg7Copy, arg7) + } fake.trackPublishedMutex.Lock() fake.trackPublishedArgsForCall = append(fake.trackPublishedArgsForCall, struct { arg1 context.Context @@ -1310,12 +1345,13 @@ func (fake *FakeTelemetryService) TrackPublished(arg1 context.Context, arg2 *liv arg4 livekit.ParticipantIdentity arg5 *livekit.TrackInfo arg6 bool - }{arg1, arg2, arg3, arg4, arg5, arg6}) + arg7 []*livekit.WebhookConfig + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7Copy}) stub := fake.TrackPublishedStub - fake.recordInvocation("TrackPublished", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6}) + fake.recordInvocation("TrackPublished", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7Copy}) fake.trackPublishedMutex.Unlock() if stub != nil { - fake.TrackPublishedStub(arg1, arg2, arg3, arg4, arg5, arg6) + fake.TrackPublishedStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7) } } @@ -1325,17 +1361,17 @@ func (fake *FakeTelemetryService) TrackPublishedCallCount() int { return len(fake.trackPublishedArgsForCall) } -func (fake *FakeTelemetryService) TrackPublishedCalls(stub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool)) { +func (fake *FakeTelemetryService) TrackPublishedCalls(stub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, []*livekit.WebhookConfig)) { fake.trackPublishedMutex.Lock() defer fake.trackPublishedMutex.Unlock() fake.TrackPublishedStub = stub } -func (fake *FakeTelemetryService) TrackPublishedArgsForCall(i int) (context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) { +func (fake *FakeTelemetryService) TrackPublishedArgsForCall(i int) (context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, []*livekit.WebhookConfig) { fake.trackPublishedMutex.RLock() defer fake.trackPublishedMutex.RUnlock() argsForCall := fake.trackPublishedArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7 } func (fake *FakeTelemetryService) TrackPublishedUpdate(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 *livekit.TrackInfo) { @@ -1589,7 +1625,12 @@ func (fake *FakeTelemetryService) TrackUnmutedArgsForCall(i int) (context.Contex return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } -func (fake *FakeTelemetryService) TrackUnpublished(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 livekit.ParticipantIdentity, arg5 *livekit.TrackInfo, arg6 bool, arg7 bool) { +func (fake *FakeTelemetryService) TrackUnpublished(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 livekit.ParticipantIdentity, arg5 *livekit.TrackInfo, arg6 bool, arg7 bool, arg8 []*livekit.WebhookConfig) { + var arg8Copy []*livekit.WebhookConfig + if arg8 != nil { + arg8Copy = make([]*livekit.WebhookConfig, len(arg8)) + copy(arg8Copy, arg8) + } fake.trackUnpublishedMutex.Lock() fake.trackUnpublishedArgsForCall = append(fake.trackUnpublishedArgsForCall, struct { arg1 context.Context @@ -1599,12 +1640,13 @@ func (fake *FakeTelemetryService) TrackUnpublished(arg1 context.Context, arg2 *l arg5 *livekit.TrackInfo arg6 bool arg7 bool - }{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) + arg8 []*livekit.WebhookConfig + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8Copy}) stub := fake.TrackUnpublishedStub - fake.recordInvocation("TrackUnpublished", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) + fake.recordInvocation("TrackUnpublished", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8Copy}) fake.trackUnpublishedMutex.Unlock() if stub != nil { - fake.TrackUnpublishedStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7) + fake.TrackUnpublishedStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8) } } @@ -1614,17 +1656,17 @@ func (fake *FakeTelemetryService) TrackUnpublishedCallCount() int { return len(fake.trackUnpublishedArgsForCall) } -func (fake *FakeTelemetryService) TrackUnpublishedCalls(stub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool)) { +func (fake *FakeTelemetryService) TrackUnpublishedCalls(stub func(context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool, []*livekit.WebhookConfig)) { fake.trackUnpublishedMutex.Lock() defer fake.trackUnpublishedMutex.Unlock() fake.TrackUnpublishedStub = stub } -func (fake *FakeTelemetryService) TrackUnpublishedArgsForCall(i int) (context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool) { +func (fake *FakeTelemetryService) TrackUnpublishedArgsForCall(i int) (context.Context, *livekit.Room, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool, bool, []*livekit.WebhookConfig) { fake.trackUnpublishedMutex.RLock() defer fake.trackUnpublishedMutex.RUnlock() argsForCall := fake.trackUnpublishedArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7, argsForCall.arg8 } func (fake *FakeTelemetryService) TrackUnsubscribed(arg1 context.Context, arg2 *livekit.Room, arg3 livekit.ParticipantID, arg4 *livekit.TrackInfo, arg5 bool) { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index c8e31efd048..d71f29d3dca 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -35,26 +35,26 @@ type TelemetryService interface { TrackStats(roomID livekit.RoomID, roomName livekit.RoomName, key StatsKey, stat *livekit.AnalyticsStat) // events - RoomStarted(ctx context.Context, room *livekit.Room) - RoomEnded(ctx context.Context, room *livekit.Room, reason livekit.RoomEndReason) + RoomStarted(ctx context.Context, room *livekit.Room, roomWebhooks []*livekit.WebhookConfig) + RoomEnded(ctx context.Context, room *livekit.Room, reason livekit.RoomEndReason, roomWebhooks []*livekit.WebhookConfig) // ParticipantJoined - a participant establishes signal connection to a room ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool, guard *ReferenceGuard) // ParticipantActive - a participant establishes media connection - ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, isWarp bool, guard *ReferenceGuard) + ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, isWarp bool, guard *ReferenceGuard, roomWebhooks []*livekit.WebhookConfig) // ParticipantResumed - there has been an ICE restart or connection resume attempt, and we've received their signal connection ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason) // ParticipantLeft - the participant leaves the room, only sent if ParticipantActive has been called before - ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard) + ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard, roomWebhooks []*livekit.WebhookConfig) // RoomIDChanged - the room kept its session, but got a different id (a provisional room id // replaced by the resolved one), re-keys the stats workers of every participant in the room RoomIDChanged(ctx context.Context, prevRoomID livekit.RoomID, room *livekit.Room) // TrackPublishRequested - a publication attempt has been received TrackPublishRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) // TrackPublished - a publication attempt has been successful - TrackPublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) + TrackPublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool, roomWebhooks []*livekit.WebhookConfig) // TrackUnpublished - a participant unpublished a track - TrackUnpublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool) + TrackUnpublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool, roomWebhooks []*livekit.WebhookConfig) // TrackSubscribeRequested - a participant requested to subscribe to a track TrackSubscribeRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, track *livekit.TrackInfo) // TrackSubscribed - a participant subscribed to a track successfully @@ -108,24 +108,25 @@ type NullTelemetryService struct { func (n NullTelemetryService) TrackStats(roomID livekit.RoomID, roomName livekit.RoomName, key StatsKey, stat *livekit.AnalyticsStat) { } -func (n NullTelemetryService) RoomStarted(ctx context.Context, room *livekit.Room) {} -func (n NullTelemetryService) RoomEnded(ctx context.Context, room *livekit.Room, reason livekit.RoomEndReason) { +func (n NullTelemetryService) RoomStarted(ctx context.Context, room *livekit.Room, roomWebhooks []*livekit.WebhookConfig) { +} +func (n NullTelemetryService) RoomEnded(ctx context.Context, room *livekit.Room, reason livekit.RoomEndReason, roomWebhooks []*livekit.WebhookConfig) { } func (n NullTelemetryService) ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool, guard *ReferenceGuard) { } -func (n NullTelemetryService) ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, isWarp bool, guard *ReferenceGuard) { +func (n NullTelemetryService) ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, isWarp bool, guard *ReferenceGuard, roomWebhooks []*livekit.WebhookConfig) { } func (n NullTelemetryService) ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason) { } -func (n NullTelemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard) { +func (n NullTelemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard, roomWebhooks []*livekit.WebhookConfig) { } func (n NullTelemetryService) RoomIDChanged(ctx context.Context, prevRoomID livekit.RoomID, room *livekit.Room) { } func (n NullTelemetryService) TrackPublishRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) { } -func (n NullTelemetryService) TrackPublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) { +func (n NullTelemetryService) TrackPublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool, roomWebhooks []*livekit.WebhookConfig) { } -func (n NullTelemetryService) TrackUnpublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool) { +func (n NullTelemetryService) TrackUnpublished(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, wasPublishedLocally bool, shouldSendEvent bool, roomWebhooks []*livekit.WebhookConfig) { } func (n NullTelemetryService) TrackSubscribeRequested(ctx context.Context, room *livekit.Room, participantID livekit.ParticipantID, track *livekit.TrackInfo) { } diff --git a/test/webhook_test.go b/test/webhook_test.go index 7fff2d22601..0ca45d1c466 100644 --- a/test/webhook_test.go +++ b/test/webhook_test.go @@ -124,6 +124,82 @@ func TestWebhooks(t *testing.T) { } } +// TestRoomWebhooks verifies that webhooks configured on a single room receive that +// room's events in addition to the globally configured webhook URLs. +func TestRoomWebhooks(t *testing.T) { + server, globalTS, finish, err := setupServerWithWebhook() + require.NoError(t, err) + defer finish() + + // the room's own webhook endpoint, separate from the global one + roomTS := newTestServer(":7891") + require.NoError(t, roomTS.Start()) + defer roomTS.Stop() + + // setupServerWithWebhook does not populate the shared roomClient + rc := livekit.NewRoomServiceJSONClient(fmt.Sprintf("http://localhost:%d", defaultServerPort), &http.Client{}) + + // signing_key left empty: signed with the key from the global webhook config + _, err = rc.CreateRoom(contextWithToken(createRoomToken()), &livekit.CreateRoomRequest{ + Name: testRoom, + Webhooks: []*livekit.WebhookConfig{{Url: "http://localhost:7891"}}, + }) + require.NoError(t, err) + + c1 := createRTCClient("c1", defaultServerPort, testRTCServicePaths[0], nil) + waitUntilConnected(t, c1) + + testutils.WithTimeout(t, func() string { + if roomTS.GetEvent(webhook.EventRoomStarted) == nil { + return "room webhook did not receive RoomStarted" + } + if roomTS.GetEvent(webhook.EventParticipantJoined) == nil { + return "room webhook did not receive ParticipantJoined" + } + return "" + }) + require.Equal(t, testRoom, roomTS.GetEvent(webhook.EventRoomStarted).Room.Name) + require.Equal(t, "c1", roomTS.GetEvent(webhook.EventParticipantJoined).Participant.Identity) + + // the global webhook still receives everything + require.NotNil(t, globalTS.GetEvent(webhook.EventRoomStarted)) + require.NotNil(t, globalTS.GetEvent(webhook.EventParticipantJoined)) + roomTS.ClearEvents() + globalTS.ClearEvents() + + writers := publishTracksForClients(t, c1) + defer stopWriters(writers...) + testutils.WithTimeout(t, func() string { + ev := roomTS.GetEvent(webhook.EventTrackPublished) + if ev == nil { + return "room webhook did not receive TrackPublished" + } + require.Equal(t, string(c1.ID()), ev.Participant.Sid) + return "" + }) + roomTS.ClearEvents() + + c1.Stop() + testutils.WithTimeout(t, func() string { + if roomTS.GetEvent(webhook.EventParticipantLeft) == nil { + return "room webhook did not receive ParticipantLeft" + } + return "" + }) + require.Equal(t, "c1", roomTS.GetEvent(webhook.EventParticipantLeft).Participant.Identity) + roomTS.ClearEvents() + + rm := server.RoomManager().GetRoom(context.Background(), testRoom) + rm.Close(types.RoomCloseReasonAPIDelete) + testutils.WithTimeout(t, func() string { + if roomTS.GetEvent(webhook.EventRoomFinished) == nil { + return "room webhook did not receive RoomFinished" + } + return "" + }) + require.Equal(t, testRoom, roomTS.GetEvent(webhook.EventRoomFinished).Room.Name) +} + func setupServerWithWebhook() (server *service.LivekitServer, testServer *webhookTestServer, finishFunc func(), err error) { conf, err := config.NewConfig("", true, nil, nil) if err != nil {