@@ -13,94 +13,112 @@ import (
1313 "github.com/sirupsen/logrus"
1414)
1515
16- func (m * NatsModel ) OnAfterUserJoined (roomId , userId string ) {
16+ func (m * NatsModel ) OnAfterUserJoined (roomId , userId , calledFrom string ) {
1717 log := m .logger .WithFields (logrus.Fields {
18- "roomId" : roomId ,
19- "userId" : userId ,
20- "method" : "OnAfterUserJoined" ,
18+ "roomId" : roomId ,
19+ "userId" : userId ,
20+ "called_from" : calledFrom ,
21+ "method" : "OnAfterUserJoined" ,
2122 })
2223
23- status , err := m .natsService .GetRoomUserStatus (roomId , userId )
24- // If there's an error or the user is already online, we don't need to proceed.
25- if err != nil {
26- log .WithError (err ).Error ("Failed to get room user status" )
27- return
28- }
29- if status == natsservice .UserStatusOnline {
30- // This is a frequent case due to pings, so we don't log it to avoid noise.
31- return
32- }
33-
34- log .Info ("Handling user joined event" )
35- if err = m .natsService .UpdateUserStatus (roomId , userId , natsservice .UserStatusOnline ); err != nil {
36- log .WithError (err ).WithField ("status" , natsservice .UserStatusOnline ).Warn ("failed to update user status" )
37- }
38-
39- userInfo , err := m .natsService .GetUserInfo (roomId , userId )
40- if err == nil && userInfo != nil {
41- // broadcast this user to everyone
42- if err = m .natsService .BroadcastSystemEventToEveryoneExceptUserId (plugnmeet .NatsMsgServerToClientEvents_USER_JOINED , roomId , userInfo , userId ); err != nil {
43- log .WithError (err ).Error ("Failed to broadcast USER_JOINED event" )
24+ key := fmt .Sprintf ("user-joined-%s-%s" , roomId , userId )
25+ _ , _ , shared := m .sfGroup .Do (key , func () (interface {}, error ) {
26+ status , err := m .natsService .GetRoomUserStatus (roomId , userId )
27+ // If there's an error or the user is already online, we don't need to proceed.
28+ if err != nil {
29+ log .WithError (err ).Error ("Failed to get room user status" )
30+ return nil , err
31+ }
32+ if status == natsservice .UserStatusOnline {
33+ // This is a frequent case due to pings, so we don't log it to avoid noise.
34+ return nil , nil
4435 }
4536
46- now := fmt .Sprintf ("%d" , time .Now ().UnixMilli ())
47- m .analyticsModel .HandleEvent (& plugnmeet.AnalyticsDataMsg {
48- EventType : plugnmeet .AnalyticsEventType_ANALYTICS_EVENT_TYPE_ROOM ,
49- EventName : plugnmeet .AnalyticsEvents_ANALYTICS_EVENT_USER_JOINED ,
50- RoomId : roomId ,
51- UserId : & userId ,
52- UserName : & userInfo .Name ,
53- ExtraData : & userInfo .Metadata ,
54- HsetValue : & now ,
55- })
56- log .Info ("Successfully processed user joined event" )
57-
58- roomInfo , err := m .natsService .GetRoomInfo (roomId )
59- if err != nil {
60- log .WithError (err ).Error ("Failed to get room info" )
37+ log .Info ("Handling user joined event" )
38+ if err = m .natsService .UpdateUserStatus (roomId , userId , natsservice .UserStatusOnline ); err != nil {
39+ log .WithError (err ).WithField ("status" , natsservice .UserStatusOnline ).Warn ("failed to update user status" )
6140 }
62- if roomInfo != nil {
63- if _ , err := m .ds .IncrementOrDecrementNumParticipants (roomInfo .GetRoomSid (), "+" ); err != nil {
64- log .WithError (err ).Error ("Failed to increment num participants" )
41+
42+ userInfo , err := m .natsService .GetUserInfo (roomId , userId )
43+ if err == nil && userInfo != nil {
44+ // broadcast this user to everyone
45+ if err = m .natsService .BroadcastSystemEventToEveryoneExceptUserId (plugnmeet .NatsMsgServerToClientEvents_USER_JOINED , roomId , userInfo , userId ); err != nil {
46+ log .WithError (err ).Error ("Failed to broadcast USER_JOINED event" )
6547 }
48+
49+ now := fmt .Sprintf ("%d" , time .Now ().UnixMilli ())
50+ m .analyticsModel .HandleEvent (& plugnmeet.AnalyticsDataMsg {
51+ EventType : plugnmeet .AnalyticsEventType_ANALYTICS_EVENT_TYPE_ROOM ,
52+ EventName : plugnmeet .AnalyticsEvents_ANALYTICS_EVENT_USER_JOINED ,
53+ RoomId : roomId ,
54+ UserId : & userId ,
55+ UserName : & userInfo .Name ,
56+ ExtraData : & userInfo .Metadata ,
57+ HsetValue : & now ,
58+ })
59+ log .Info ("Successfully processed user joined event" )
60+
61+ roomInfo , err := m .natsService .GetRoomInfo (roomId )
62+ if err != nil {
63+ log .WithError (err ).Error ("Failed to get room info" )
64+ }
65+ if roomInfo != nil {
66+ if _ , err := m .ds .IncrementOrDecrementNumParticipants (roomInfo .GetRoomSid (), "+" ); err != nil {
67+ log .WithError (err ).Error ("Failed to increment num participants" )
68+ }
69+ }
70+ } else if err != nil {
71+ log .WithError (err ).Warn ("Could not get user info after join" )
6672 }
67- } else if err != nil {
68- log .WithError (err ).Warn ("Could not get user info after join" )
73+ return nil , nil
74+ })
75+
76+ if shared {
77+ log .Debug ("OnAfterUserJoined call already in progress for this user, skipping." )
6978 }
7079}
7180
7281// OnAfterUserDisconnected should be run in separate goroutine
7382// we'll wait for 5 seconds before declare user as offline
7483// but will broadcast as disconnected
75- func (m * NatsModel ) OnAfterUserDisconnected (roomId , userId string ) {
84+ func (m * NatsModel ) OnAfterUserDisconnected (roomId , userId , calledFrom string ) {
7685 log := m .logger .WithFields (logrus.Fields {
77- "roomId" : roomId ,
78- "userId" : userId ,
79- "method" : "OnAfterUserDisconnected" ,
86+ "roomId" : roomId ,
87+ "userId" : userId ,
88+ "called_from" : calledFrom ,
89+ "method" : "OnAfterUserDisconnected" ,
8090 })
8191 log .Info ("Handling user disconnected event" )
8292
83- // Immediately set status to disconnected and notify clients.
84- if err := m .natsService .UpdateUserStatus (roomId , userId , natsservice .UserStatusDisconnected ); err != nil {
85- log .WithError (err ).WithField ("status" , natsservice .UserStatusDisconnected ).Warn ("Failed to update user status" )
86- }
93+ key := fmt .Sprintf ("user-disconnected-%s-%s" , roomId , userId )
94+ _ , _ , shared := m .sfGroup .Do (key , func () (interface {}, error ) {
95+ // Immediately set status to disconnected and notify clients.
96+ if err := m .natsService .UpdateUserStatus (roomId , userId , natsservice .UserStatusDisconnected ); err != nil {
97+ log .WithError (err ).WithField ("status" , natsservice .UserStatusDisconnected ).Warn ("Failed to update user status" )
98+ }
8799
88- // update analytics & db for the user leaving.
89- m .updateUserLeftAnalytics (roomId , userId , log )
100+ // update analytics & db for the user leaving.
101+ m .updateUserLeftAnalytics (roomId , userId , log )
90102
91- // Try to get user info for a richer disconnect message.
92- userInfo , err := m .natsService .GetUserInfo (roomId , userId )
93- if err != nil || userInfo == nil {
94- // If we can't get user info, send a basic event and update analytics.
95- if err = m .natsService .BroadcastSystemEventToEveryoneExceptUserId (plugnmeet .NatsMsgServerToClientEvents_USER_DISCONNECTED , roomId , & plugnmeet.NatsKvUserInfo {UserId : userId , RoomId : roomId }, userId ); err != nil {
96- log .WithError (err ).Error ("Failed to broadcast basic USER_DISCONNECTED event" )
103+ // Try to get user info for a richer disconnect message.
104+ userInfo , err := m .natsService .GetUserInfo (roomId , userId )
105+ if err != nil || userInfo == nil {
106+ // If we can't get user info, send a basic event and update analytics.
107+ if err = m .natsService .BroadcastSystemEventToEveryoneExceptUserId (plugnmeet .NatsMsgServerToClientEvents_USER_DISCONNECTED , roomId , & plugnmeet.NatsKvUserInfo {UserId : userId , RoomId : roomId }, userId ); err != nil {
108+ log .WithError (err ).Error ("Failed to broadcast basic USER_DISCONNECTED event" )
109+ }
110+ } else {
111+ _ = m .natsService .BroadcastSystemEventToEveryoneExceptUserId (plugnmeet .NatsMsgServerToClientEvents_USER_DISCONNECTED , roomId , userInfo , userId )
97112 }
98- } else {
99- _ = m .natsService .BroadcastSystemEventToEveryoneExceptUserId (plugnmeet .NatsMsgServerToClientEvents_USER_DISCONNECTED , roomId , userInfo , userId )
100- }
101113
102- // Start a non-blocking background task to handle the full offline/cleanup lifecycle.
103- go m .handleDelayedOfflineTasks (roomId , userId , userInfo , log )
114+ // Start a non-blocking background task to handle the full offline/cleanup lifecycle.
115+ go m .handleDelayedOfflineTasks (roomId , userId , userInfo , log )
116+ return nil , nil
117+ })
118+
119+ if shared {
120+ log .Debug ("OnAfterUserDisconnected call already in progress for this user, skipping." )
121+ }
104122}
105123
106124// handleDelayedOfflineTasks manages the grace period for user reconnection and subsequent cleanup using periodic checks.
0 commit comments