Do not send ParticipantLeft webhook event unless connected successfully. (#1234)

Fixes #1130

pkg/rtc/types/typesfakes/fake_local_participant.go

@@ -612,11 +612,6 @@ type FakeLocalParticipant struct {
 	setICEConfigArgsForCall []struct {
 		arg1 types.IceConfig
 	}
-	SetNameStub        func(string)
-	setNameMutex       sync.RWMutex
-	setNameArgsForCall []struct {
-		arg1 string
-	}
 	SetMetadataStub        func(string)
 	setMetadataMutex       sync.RWMutex
 	setMetadataArgsForCall []struct {
@@ -635,6 +630,11 @@ type FakeLocalParticipant struct {
 	setMigrateStateArgsForCall []struct {
 		arg1 types.MigrateState
 	}
+	SetNameStub        func(string)
+	setNameMutex       sync.RWMutex
+	setNameArgsForCall []struct {
+		arg1 string
+	}
 	SetPermissionStub        func(*livekit.ParticipantPermission) bool
 	setPermissionMutex       sync.RWMutex
 	setPermissionArgsForCall []struct {
@@ -4011,38 +4011,6 @@ func (fake *FakeLocalParticipant) SetICEConfigArgsForCall(i int) types.IceConfig
 	return argsForCall.arg1
 }
 
-func (fake *FakeLocalParticipant) SetName(arg1 string) {
-	fake.setNameMutex.Lock()
-	fake.setNameArgsForCall = append(fake.setNameArgsForCall, struct {
-		arg1 string
-	}{arg1})
-	stub := fake.SetNameStub
-	fake.recordInvocation("SetName", []interface{}{arg1})
-	fake.setNameMutex.Unlock()
-	if stub != nil {
-		fake.SetNameStub(arg1)
-	}
-}
-
-func (fake *FakeLocalParticipant) SetNameCallCount() int {
-	fake.setNameMutex.RLock()
-	defer fake.setNameMutex.RUnlock()
-	return len(fake.setNameArgsForCall)
-}
-
-func (fake *FakeLocalParticipant) SetNameCalls(stub func(string)) {
-	fake.setNameMutex.Lock()
-	defer fake.setNameMutex.Unlock()
-	fake.SetNameStub = stub
-}
-
-func (fake *FakeLocalParticipant) SetNameArgsForCall(i int) string {
-	fake.setNameMutex.RLock()
-	defer fake.setNameMutex.RUnlock()
-	argsForCall := fake.setNameArgsForCall[i]
-	return argsForCall.arg1
-}
-
 func (fake *FakeLocalParticipant) SetMetadata(arg1 string) {
 	fake.setMetadataMutex.Lock()
 	fake.setMetadataArgsForCall = append(fake.setMetadataArgsForCall, struct {
@@ -4152,6 +4120,38 @@ func (fake *FakeLocalParticipant) SetMigrateStateArgsForCall(i int) types.Migrat
 	return argsForCall.arg1
 }
 
+func (fake *FakeLocalParticipant) SetName(arg1 string) {
+	fake.setNameMutex.Lock()
+	fake.setNameArgsForCall = append(fake.setNameArgsForCall, struct {
+		arg1 string
+	}{arg1})
+	stub := fake.SetNameStub
+	fake.recordInvocation("SetName", []interface{}{arg1})
+	fake.setNameMutex.Unlock()
+	if stub != nil {
+		fake.SetNameStub(arg1)
+	}
+}
+
+func (fake *FakeLocalParticipant) SetNameCallCount() int {
+	fake.setNameMutex.RLock()
+	defer fake.setNameMutex.RUnlock()
+	return len(fake.setNameArgsForCall)
+}
+
+func (fake *FakeLocalParticipant) SetNameCalls(stub func(string)) {
+	fake.setNameMutex.Lock()
+	defer fake.setNameMutex.Unlock()
+	fake.SetNameStub = stub
+}
+
+func (fake *FakeLocalParticipant) SetNameArgsForCall(i int) string {
+	fake.setNameMutex.RLock()
+	defer fake.setNameMutex.RUnlock()
+	argsForCall := fake.setNameArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakeLocalParticipant) SetPermission(arg1 *livekit.ParticipantPermission) bool {
 	fake.setPermissionMutex.Lock()
 	ret, specificReturn := fake.setPermissionReturnsOnCall[len(fake.setPermissionArgsForCall)]
@@ -5110,6 +5110,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} {
 	defer fake.setMigrateInfoMutex.RUnlock()
 	fake.setMigrateStateMutex.RLock()
 	defer fake.setMigrateStateMutex.RUnlock()
+	fake.setNameMutex.RLock()
+	defer fake.setNameMutex.RUnlock()
 	fake.setPermissionMutex.RLock()
 	defer fake.setPermissionMutex.RUnlock()
 	fake.setResponseSinkMutex.RLock()

pkg/rtc/types/typesfakes/fake_participant.go

@@ -120,16 +120,16 @@ type FakeParticipant struct {
 		arg2 livekit.TrackID
 		arg3 bool
 	}
-	SetNameStub        func(string)
-	setNameMutex       sync.RWMutex
-	setNameArgsForCall []struct {
-		arg1 string
-	}
 	SetMetadataStub        func(string)
 	setMetadataMutex       sync.RWMutex
 	setMetadataArgsForCall []struct {
 		arg1 string
 	}
+	SetNameStub        func(string)
+	setNameMutex       sync.RWMutex
+	setNameArgsForCall []struct {
+		arg1 string
+	}
 	StartStub        func()
 	startMutex       sync.RWMutex
 	startArgsForCall []struct {
@@ -759,38 +759,6 @@ func (fake *FakeParticipant) RemoveSubscriberArgsForCall(i int) (types.LocalPart
 	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
 }
 
-func (fake *FakeParticipant) SetName(arg1 string) {
-	fake.setNameMutex.Lock()
-	fake.setNameArgsForCall = append(fake.setNameArgsForCall, struct {
-		arg1 string
-	}{arg1})
-	stub := fake.SetNameStub
-	fake.recordInvocation("SetName", []interface{}{arg1})
-	fake.setNameMutex.Unlock()
-	if stub != nil {
-		fake.SetNameStub(arg1)
-	}
-}
-
-func (fake *FakeParticipant) SetNameCallCount() int {
-	fake.setNameMutex.RLock()
-	defer fake.setNameMutex.RUnlock()
-	return len(fake.setNameArgsForCall)
-}
-
-func (fake *FakeParticipant) SetNameCalls(stub func(string)) {
-	fake.setNameMutex.Lock()
-	defer fake.setNameMutex.Unlock()
-	fake.SetNameStub = stub
-}
-
-func (fake *FakeParticipant) SetNameArgsForCall(i int) string {
-	fake.setNameMutex.RLock()
-	defer fake.setNameMutex.RUnlock()
-	argsForCall := fake.setNameArgsForCall[i]
-	return argsForCall.arg1
-}
-
 func (fake *FakeParticipant) SetMetadata(arg1 string) {
 	fake.setMetadataMutex.Lock()
 	fake.setMetadataArgsForCall = append(fake.setMetadataArgsForCall, struct {
@@ -823,6 +791,38 @@ func (fake *FakeParticipant) SetMetadataArgsForCall(i int) string {
 	return argsForCall.arg1
 }
 
+func (fake *FakeParticipant) SetName(arg1 string) {
+	fake.setNameMutex.Lock()
+	fake.setNameArgsForCall = append(fake.setNameArgsForCall, struct {
+		arg1 string
+	}{arg1})
+	stub := fake.SetNameStub
+	fake.recordInvocation("SetName", []interface{}{arg1})
+	fake.setNameMutex.Unlock()
+	if stub != nil {
+		fake.SetNameStub(arg1)
+	}
+}
+
+func (fake *FakeParticipant) SetNameCallCount() int {
+	fake.setNameMutex.RLock()
+	defer fake.setNameMutex.RUnlock()
+	return len(fake.setNameArgsForCall)
+}
+
+func (fake *FakeParticipant) SetNameCalls(stub func(string)) {
+	fake.setNameMutex.Lock()
+	defer fake.setNameMutex.Unlock()
+	fake.SetNameStub = stub
+}
+
+func (fake *FakeParticipant) SetNameArgsForCall(i int) string {
+	fake.setNameMutex.RLock()
+	defer fake.setNameMutex.RUnlock()
+	argsForCall := fake.setNameArgsForCall[i]
+	return argsForCall.arg1
+}
+
 func (fake *FakeParticipant) Start() {
 	fake.startMutex.Lock()
 	fake.startArgsForCall = append(fake.startArgsForCall, struct {
@@ -1108,6 +1108,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} {
 	defer fake.removeSubscriberMutex.RUnlock()
 	fake.setMetadataMutex.RLock()
 	defer fake.setMetadataMutex.RUnlock()
+	fake.setNameMutex.RLock()
+	defer fake.setNameMutex.RUnlock()
 	fake.startMutex.RLock()
 	defer fake.startMutex.RUnlock()
 	fake.subscriptionPermissionMutex.RLock()

pkg/service/roommanager.go

@@ -332,7 +332,7 @@ func (r *RoomManager) StartSession(
 		pLogger.Errorw("could not store participant", err)
 	}
 
-	updateParticipantCount := func(proto *livekit.Room) {
+	persistRoomForParticipantCount := func(proto *livekit.Room) {
 		if !participant.Hidden() {
 			err = r.roomStore.StoreRoom(ctx, proto, room.Internal())
 			if err != nil {
@@ -342,7 +342,7 @@ func (r *RoomManager) StartSession(
 	}
 
 	// update room store with new numParticipants
-	updateParticipantCount(room.ToProto())
+	persistRoomForParticipantCount(room.ToProto())
 
 	clientMeta := &livekit.AnalyticsClientMeta{Region: r.currentNode.Region, Node: r.currentNode.Id}
 	r.telemetry.ParticipantJoined(ctx, protoRoom, participant.ToProto(), pi.Client, clientMeta)
@@ -353,8 +353,8 @@ func (r *RoomManager) StartSession(
 
 		// update room store with new numParticipants
 		proto := room.ToProto()
-		updateParticipantCount(proto)
-		r.telemetry.ParticipantLeft(ctx, proto, p.ToProto())
+		persistRoomForParticipantCount(proto)
+		r.telemetry.ParticipantLeft(ctx, proto, p.ToProto(), true)
 
 		room.RemoveDisallowedSubscriptions(p, disallowedSubscriptions)
 	})

pkg/telemetry/events.go

@@ -105,8 +105,9 @@ func (t *telemetryService) ParticipantActive(
 			Participant: participant,
 		})
 
-		if _, ok := t.getWorker(livekit.ParticipantID(participant.Sid)); !ok {
-			t.createWorker(
+		worker, ok := t.getWorker(livekit.ParticipantID(participant.Sid))
+		if !ok {
+			worker = t.createWorker(
 				ctx,
 				livekit.RoomID(room.Sid),
 				livekit.RoomName(room.Name),
@@ -114,6 +115,7 @@ func (t *telemetryService) ParticipantActive(
 				livekit.ParticipantIdentity(participant.Identity),
 			)
 		}
+		worker.SetConnected()
 
 		t.SendEvent(ctx, &livekit.AnalyticsEvent{
 			Type:          livekit.AnalyticsEventType_PARTICIPANT_ACTIVE,
@@ -126,28 +128,32 @@ func (t *telemetryService) ParticipantActive(
 	})
 }
 
-func (t *telemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo) {
+func (t *telemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool) {
 	t.enqueue(func() {
+		isConnected := false
 		if worker, ok := t.getWorker(livekit.ParticipantID(participant.Sid)); ok {
+			isConnected = worker.IsConnected()
 			worker.Close()
 		}
 
 		prometheus.SubParticipant()
 
-		t.NotifyEvent(ctx, &livekit.WebhookEvent{
-			Event:       webhook.EventParticipantLeft,
-			Room:        room,
-			Participant: participant,
-		})
+		if isConnected && shouldSendEvent {
+			t.NotifyEvent(ctx, &livekit.WebhookEvent{
+				Event:       webhook.EventParticipantLeft,
+				Room:        room,
+				Participant: participant,
+			})
 
-		t.SendEvent(ctx, &livekit.AnalyticsEvent{
-			Type:          livekit.AnalyticsEventType_PARTICIPANT_LEFT,
-			Timestamp:     timestamppb.Now(),
-			RoomId:        room.Sid,
-			ParticipantId: participant.Sid,
-			Participant:   participant,
-			Room:          room,
-		})
+			t.SendEvent(ctx, &livekit.AnalyticsEvent{
+				Type:          livekit.AnalyticsEventType_PARTICIPANT_LEFT,
+				Timestamp:     timestamppb.Now(),
+				RoomId:        room.Sid,
+				ParticipantId: participant.Sid,
+				Participant:   participant,
+				Room:          room,
+			})
+		}
 	})
 }
 

pkg/telemetry/events_test.go

@@ -69,12 +69,13 @@ func Test_OnParticipantLeft_EventIsSent(t *testing.T) {
 	participantInfo := &livekit.ParticipantInfo{Sid: partSID}
 
 	// do
-	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo)
+	fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{})
+	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true)
 	time.Sleep(time.Millisecond * 500)
 
 	// test
-	require.Equal(t, 1, fixture.analytics.SendEventCallCount())
-	_, event := fixture.analytics.SendEventArgsForCall(0)
+	require.Equal(t, 2, fixture.analytics.SendEventCallCount())
+	_, event := fixture.analytics.SendEventArgsForCall(1)
 	require.Equal(t, livekit.AnalyticsEventType_PARTICIPANT_LEFT, event.Type)
 	require.Equal(t, partSID, event.ParticipantId)
 	require.Equal(t, room.Sid, event.RoomId)

pkg/telemetry/stats_test.go

@@ -475,7 +475,7 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) {
 	fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil)
 
 	// do
-	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo)
+	fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true)
 
 	// should not be called if there are no track stats
 	time.Sleep(time.Millisecond * 500)

pkg/telemetry/statsworker.go

@@ -20,6 +20,7 @@ type StatsWorker struct {
 	roomName            livekit.RoomName
 	participantID       livekit.ParticipantID
 	participantIdentity livekit.ParticipantIdentity
+	isConnected         bool
 
 	lock             sync.RWMutex
 	outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
@@ -62,6 +63,19 @@ func (s *StatsWorker) ParticipantID() livekit.ParticipantID {
 	return s.participantID
 }
 
+func (s *StatsWorker) SetConnected() {
+	s.lock.Lock()
+	s.isConnected = true
+	s.lock.Unlock()
+}
+
+func (s *StatsWorker) IsConnected() bool {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return s.isConnected
+}
+
 func (s *StatsWorker) Flush() {
 	ts := timestamppb.Now()
 

pkg/telemetry/telemetryfakes/fake_telemetry_service.go

@@ -49,12 +49,13 @@ type FakeTelemetryService struct {
 		arg4 *livekit.ClientInfo
 		arg5 *livekit.AnalyticsClientMeta
 	}
-	ParticipantLeftStub        func(context.Context, *livekit.Room, *livekit.ParticipantInfo)
+	ParticipantLeftStub        func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool)
 	participantLeftMutex       sync.RWMutex
 	participantLeftArgsForCall []struct {
 		arg1 context.Context
 		arg2 *livekit.Room
 		arg3 *livekit.ParticipantInfo
+		arg4 bool
 	}
 	RoomEndedStub        func(context.Context, *livekit.Room)
 	roomEndedMutex       sync.RWMutex
@@ -334,18 +335,19 @@ func (fake *FakeTelemetryService) ParticipantJoinedArgsForCall(i int) (context.C
 	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5
 }
 
-func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo) {
+func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 bool) {
 	fake.participantLeftMutex.Lock()
 	fake.participantLeftArgsForCall = append(fake.participantLeftArgsForCall, struct {
 		arg1 context.Context
 		arg2 *livekit.Room
 		arg3 *livekit.ParticipantInfo
-	}{arg1, arg2, arg3})
+		arg4 bool
+	}{arg1, arg2, arg3, arg4})
 	stub := fake.ParticipantLeftStub
-	fake.recordInvocation("ParticipantLeft", []interface{}{arg1, arg2, arg3})
+	fake.recordInvocation("ParticipantLeft", []interface{}{arg1, arg2, arg3, arg4})
 	fake.participantLeftMutex.Unlock()
 	if stub != nil {
-		fake.ParticipantLeftStub(arg1, arg2, arg3)
+		fake.ParticipantLeftStub(arg1, arg2, arg3, arg4)
 	}
 }
 
@@ -355,17 +357,17 @@ func (fake *FakeTelemetryService) ParticipantLeftCallCount() int {
 	return len(fake.participantLeftArgsForCall)
 }
 
-func (fake *FakeTelemetryService) ParticipantLeftCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo)) {
+func (fake *FakeTelemetryService) ParticipantLeftCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool)) {
 	fake.participantLeftMutex.Lock()
 	defer fake.participantLeftMutex.Unlock()
 	fake.ParticipantLeftStub = stub
 }
 
-func (fake *FakeTelemetryService) ParticipantLeftArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo) {
+func (fake *FakeTelemetryService) ParticipantLeftArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, bool) {
 	fake.participantLeftMutex.RLock()
 	defer fake.participantLeftMutex.RUnlock()
 	argsForCall := fake.participantLeftArgsForCall[i]
-	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
 }
 
 func (fake *FakeTelemetryService) RoomEnded(arg1 context.Context, arg2 *livekit.Room) {

pkg/telemetry/telemetryservice.go

@@ -23,7 +23,7 @@ type TelemetryService interface {
 	RoomEnded(ctx context.Context, room *livekit.Room)
 	ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta)
 	ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta)
-	ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo)
+	ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool)
 	TrackPublished(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo)
 	TrackUnpublished(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, ssrc uint32)
 	TrackSubscribed(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, publisher *livekit.ParticipantInfo)
@@ -116,8 +116,12 @@ func (t *telemetryService) getWorker(participantID livekit.ParticipantID) (worke
 	return
 }
 
-func (t *telemetryService) createWorker(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName,
-	participantID livekit.ParticipantID, participantIdentity livekit.ParticipantIdentity) {
+func (t *telemetryService) createWorker(ctx context.Context,
+	roomID livekit.RoomID,
+	roomName livekit.RoomName,
+	participantID livekit.ParticipantID,
+	participantIdentity livekit.ParticipantIdentity,
+) *StatsWorker {
 	worker := newStatsWorker(
 		ctx,
 		t,
@@ -130,6 +134,7 @@ func (t *telemetryService) createWorker(ctx context.Context, roomID livekit.Room
 	t.lock.Lock()
 	t.workers[participantID] = worker
 	t.lock.Unlock()
+	return worker
 }
 
 func (t *telemetryService) cleanupWorkers() {