[MM-57737] Improve client side call state consistency

server/client_message.go

@@ -29,6 +29,7 @@ const (
 	clientMessageTypeReact       = "react"
 	clientMessageTypeCaption     = "caption"
 	clientMessageTypeMetric      = "metric"
+	clientMessageTypeCallState   = "call_state"
 )
 
 func (m *clientMessage) ToJSON() ([]byte, error) {

server/websocket.go

@@ -888,6 +888,40 @@ func (p *Plugin) handleReconnect(userID, connID, channelID, originalConnID, prev
 	return nil
 }
 
+func (p *Plugin) handleCallStateRequest(channelID, userID, connID string) error {
+	// We should go through only if the user has permissions to the requested channel
+	// or if the user is the Calls bot.
+	if !(p.isBot(userID) || p.API.HasPermissionToChannel(userID, channelID, model.PermissionReadChannel)) {
+		return fmt.Errorf("forbidden")
+	}
+
+	// Locking is not ideal but it's the only way to guarantee a race free
+	// sequence and a consistent state.
+	// On the client we should make sure to make this request only when strictly
+	// necessary (i.e first load, joining call, reconnecting).
+	state, err := p.lockCallReturnState(channelID)
+	if err != nil {
+		return fmt.Errorf("failed to lock call: %w", err)
+	}
+	defer p.unlockCall(channelID)
+
+	if state == nil {
+		return fmt.Errorf("no call ongoing")
+	}
+
+	clientStateData, err := json.Marshal(state.getClientState(p.getBotID(), userID))
+	if err != nil {
+		return fmt.Errorf("failed to marshal client state: %w", err)
+	}
+
+	p.publishWebSocketEvent(wsEventCallState, map[string]interface{}{
+		"channel_id": channelID,
+		"call":       string(clientStateData),
+	}, &model.WebsocketBroadcast{ConnectionId: connID, ReliableClusterSend: true})
+
+	return nil
+}
+
 func (p *Plugin) WebSocketMessageHasBeenPosted(connID, userID string, req *model.WebSocketRequest) {
 	var msg clientMessage
 	msg.Type = strings.TrimPrefix(req.Action, wsActionPrefix)
@@ -896,10 +930,14 @@ func (p *Plugin) WebSocketMessageHasBeenPosted(connID, userID string, req *model
 	us := p.sessions[connID]
 	p.mut.RUnlock()
 
-	if msg.Type != clientMessageTypeJoin &&
-		msg.Type != clientMessageTypeLeave &&
-		msg.Type != clientMessageTypeReconnect && us == nil {
-		return
+	if us == nil {
+		// Only a few events don't require a user session to exist. For anything else
+		// we should return.
+		switch msg.Type {
+		case clientMessageTypeJoin, clientMessageTypeLeave, clientMessageTypeReconnect, clientMessageTypeCallState:
+		default:
+			return
+		}
 	}
 
 	if us != nil && !us.wsMsgLimiter.Allow() {
@@ -985,6 +1023,19 @@ func (p *Plugin) WebSocketMessageHasBeenPosted(connID, userID string, req *model
 			p.LogError(err.Error())
 		}
 
+		return
+	case clientMessageTypeCallState:
+		p.metrics.IncWebSocketEvent("in", "call_state")
+
+		channelID, _ := req.Data["channelID"].(string)
+		if channelID == "" {
+			p.LogError("missing channelID")
+			return
+		}
+
+		if err := p.handleCallStateRequest(channelID, userID, connID); err != nil {
+			p.LogError("handleCallStateRequest failed", "err", err.Error(), "userID", userID, "connID", connID)
+		}
 		return
 	case clientMessageTypeSDP:
 		msgData, ok := req.Data["data"].([]byte)

server/websocket_test.go

@@ -359,3 +359,68 @@ func TestWSReader(t *testing.T) {
 		})
 	})
 }
+
+func TestHandleCallStateRequest(t *testing.T) {
+	mockAPI := &pluginMocks.MockAPI{}
+	mockMetrics := &serverMocks.MockMetrics{}
+
+	p := Plugin{
+		MattermostPlugin: plugin.MattermostPlugin{
+			API: mockAPI,
+		},
+		callsClusterLocks: map[string]*cluster.Mutex{},
+		metrics:           mockMetrics,
+	}
+
+	store, tearDown := NewTestStore(t)
+	t.Cleanup(tearDown)
+	p.store = store
+
+	mockAPI.On("LogDebug", mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything)
+	mockAPI.On("KVSetWithOptions", mock.Anything, mock.Anything, mock.Anything).Return(true, nil)
+	mockMetrics.On("ObserveClusterMutexGrabTime", "mutex_call", mock.AnythingOfType("float64"))
+	mockMetrics.On("ObserveClusterMutexLockedTime", "mutex_call", mock.AnythingOfType("float64"))
+	mockMetrics.On("IncStoreOp", "KVGet")
+	mockMetrics.On("IncStoreOp", "KVSet")
+
+	channelID := model.NewId()
+	userID := model.NewId()
+	connID := model.NewId()
+
+	t.Run("no permissions", func(t *testing.T) {
+		mockAPI.On("KVDelete", "mutex_call_"+channelID).Return(nil).Once()
+		mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionReadChannel).Return(false).Once()
+		err := p.handleCallStateRequest(channelID, userID, connID)
+		require.EqualError(t, err, "forbidden")
+	})
+
+	t.Run("no call ongoing", func(t *testing.T) {
+		mockAPI.On("KVDelete", "mutex_call_"+channelID).Return(nil).Once()
+		mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionReadChannel).Return(true).Once()
+		err := p.handleCallStateRequest(channelID, userID, connID)
+		require.EqualError(t, err, "no call ongoing")
+	})
+
+	t.Run("active call", func(t *testing.T) {
+		err := p.store.CreateCall(&public.Call{
+			ID:        model.NewId(),
+			CreateAt:  time.Now().UnixMilli(),
+			ChannelID: channelID,
+			StartAt:   time.Now().UnixMilli(),
+			PostID:    model.NewId(),
+			ThreadID:  model.NewId(),
+			OwnerID:   model.NewId(),
+		})
+		require.NoError(t, err)
+
+		mockAPI.On("KVDelete", "mutex_call_"+channelID).Return(nil).Once()
+		mockAPI.On("HasPermissionToChannel", userID, channelID, model.PermissionReadChannel).Return(true).Once()
+		mockMetrics.On("IncWebSocketEvent", "out", "call_state").Once()
+		mockAPI.On("PublishWebSocketEvent", "call_state", mock.Anything, mock.Anything).Once()
+
+		err = p.handleCallStateRequest(channelID, userID, connID)
+		require.NoError(t, err)
+	})
+}

webapp/src/components/expanded_view/component.tsx

@@ -163,6 +163,8 @@ export default class ExpandedView extends React.PureComponent<Props, State> {
     private pushToTalk = false;
     private screenPlayer: HTMLVideoElement | null = null;
 
+    static contextType = window.ProductApi.WebSocketProvider;
+
     #unlockNavigation?: () => void;
 
     private genStyle: () => Record<string, React.CSSProperties> = () => {
@@ -592,9 +594,16 @@ export default class ExpandedView extends React.PureComponent<Props, State> {
     public componentDidMount() {
         const callsClient = getCallsClient();
         if (!callsClient) {
+            logErr('callsClient should be defined');
             return;
         }
 
+        // On mount we request the call state through websocket. This avoids
+        // making a potentially racy HTTP call and should guarantee
+        // a consistent state.
+        logDebug('requesting call state through ws');
+        this.context.sendMessage('custom_com.mattermost.calls_call_state', {channelID: callsClient.channelID});
+
         // keyboard shortcuts
         window.addEventListener('keydown', this.handleKBShortcuts, true);
         window.addEventListener('keyup', this.handleKeyUp, true);