From c9880222ff6b87ccd52993ac6d6d1576f7481dba Mon Sep 17 00:00:00 2001 From: Darshak03 <115038906+Darshak03@users.noreply.github.com> Date: Tue, 8 Sep 2026 00:45:14 +0530 Subject: [PATCH] Support room-level webhooks Webhooks are currently global: every room's events go to the URLs in the `webhook` config, and there is no way to send one room's events to its own endpoint. This adds per-room webhooks, following the shape egress already uses for per-resource webhooks. Egress attaches `repeated WebhookConfig webhooks` to the request that created the resource, keeps that request in the resource state (EgressInfo.Request), derives notify options from it with egress.GetEgressNotifyOptions, and spreads them into QueueNotify at the emit site. Rooms now do the same, with RoomInternal as the state carrier and webhook.GetRoomNotifyOptions as the derivation. Configuration comes from either CreateRoomRequest.webhooks or a named room preset (RoomConfiguration.webhooks), resolved in the allocator next to the existing Egress and Agents handling and persisted on RoomInternal. The preset copy is guarded on a non-empty list rather than mirroring the Agents line verbatim: mirroring it would materialise an empty non-nil slice that wipes previously configured webhooks when an existing room is re-created. The six webhook-emitting telemetry methods take the room's webhook list and derive the options internally, the way NotifyEgressEvent does. A registry keyed by room inside telemetryService would have been a smaller diff, but RoomIDChanged re-keys a room mid-session and a keyed lookup would then silently stop matching. BytesSignalStats passes nil: it calls ParticipantLeft with shouldSendEvent false and emits no webhook, so the signal node never needs the room's config. Room webhooks receive room, participant and track events only. Egress and ingress events are excluded by an allowlist in the protocol helper, since those resources carry their own webhook config. CreateRoom now rejects webhook URLs that are not absolute http(s) and signing keys that no configured API key matches. Without the check a typo means webhooks silently never fire, leaving only a warning log per event. This needed the key provider on RoomService; it was already a provider in the wire injector. Includes a note in config-sample.yaml, unit tests for the allocator resolution and the request validation, and an end-to-end test asserting a room's own endpoint receives its events while the global endpoint still receives everything. --- config-sample.yaml | 22 +++ go.mod | 2 + go.sum | 4 +- pkg/rtc/room.go | 5 +- pkg/rtc/signalanddatastats.go | 2 +- pkg/service/roomallocator.go | 9 ++ pkg/service/roommanager.go | 6 +- pkg/service/roomservice.go | 25 +++ pkg/service/roomservice_test.go | 1 + pkg/service/roomwebhook_test.go | 145 ++++++++++++++++++ pkg/service/wire_gen.go | 10 +- pkg/telemetry/events.go | 20 ++- pkg/telemetry/events_test.go | 8 +- pkg/telemetry/stats_test.go | 6 +- .../telemetryfakes/fake_telemetry_service.go | 138 +++++++++++------ pkg/telemetry/telemetryservice.go | 25 +-- test/webhook_test.go | 76 +++++++++ 17 files changed, 416 insertions(+), 88 deletions(-) create mode 100644 pkg/service/roomwebhook_test.go 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 {