[MM-42574] Single handler mode

server/activate.go

@@ -2,7 +2,9 @@ package main
 
 import (
 	"net"
+	"os"
 	"syscall"
+	"time"
 
 	"github.com/mattermost/mattermost-server/v6/model"
 
@@ -26,6 +28,29 @@ func (p *Plugin) OnActivate() error {
 		return appErr
 	}
 
+	if os.Getenv("CALLS_IS_HANDLER") != "" {
+		go func() {
+			p.LogInfo("calls handler, setting state", "clusterID", status.ClusterId)
+			if err := p.setHandlerID(status.ClusterId); err != nil {
+				p.LogError(err.Error())
+				return
+			}
+			ticker := time.NewTicker(handlerKeyCheckInterval)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-ticker.C:
+					if err := p.setHandlerID(status.ClusterId); err != nil {
+						p.LogError(err.Error())
+						return
+					}
+				case <-p.stopCh:
+					return
+				}
+			}
+		}()
+	}
+
 	cfg := p.getConfiguration()
 	if err := cfg.IsValid(); err != nil {
 		p.LogError(err.Error())

server/api.go

@@ -140,7 +140,7 @@ func (p *Plugin) handleGetAllChannels(w http.ResponseWriter, r *http.Request) {
 		}
 
 		for _, channelID := range channelIDs {
-			if !p.hasPermissionToChannel(channelMembers[channelID], model.PermissionReadChannel) {
+			if len(channelID) != 26 || !p.hasPermissionToChannel(channelMembers[channelID], model.PermissionReadChannel) {
 				continue
 			}
 

server/call.go

@@ -23,6 +23,15 @@ func (c *call) getScreenSession() *session {
 	return c.screenSession
 }
 
+func (c *call) getScreenSessionID() string {
+	c.mut.RLock()
+	defer c.mut.RUnlock()
+	if c.screenSession == nil {
+		return ""
+	}
+	return c.screenSession.userID
+}
+
 func (c *call) setScreenSession(s *session) {
 	c.mut.Lock()
 	defer c.mut.Unlock()

server/channel_state.go

@@ -93,6 +93,21 @@ func (p *Plugin) cleanUpState() error {
 			break
 		}
 		for _, k := range keys {
+			if k == handlerKey {
+				handlerID, err := p.getHandlerID()
+				if err != nil {
+					p.LogError(err.Error())
+					continue
+				}
+
+				if p.nodeID == handlerID {
+					if appErr = p.API.KVDelete(k); appErr != nil {
+						p.LogError(err.Error())
+					}
+				}
+				continue
+			}
+
 			if err := p.kvSetAtomicChannelState(k, func(state *channelState) (*channelState, error) {
 				if state == nil {
 					return nil, nil

server/cluster_message.go

@@ -23,6 +23,7 @@ const (
 	clusterMessageTypeDisconnect clusterMessageType = "disconnect"
 	clusterMessageTypeSignaling  clusterMessageType = "signaling"
 	clusterMessageTypeUserState  clusterMessageType = "user_state"
+	clusterMessageTypeCallEnded  clusterMessageType = "call_ended"
 )
 
 func (m *clusterMessage) ToJSON() ([]byte, error) {

server/configuration.go

@@ -50,6 +50,9 @@ func (is *ICEServers) UnmarshalJSON(data []byte) error {
 	if err != nil {
 		return err
 	}
+	if unquoted == "" {
+		return nil
+	}
 	*is = strings.Split(strings.TrimSpace(unquoted), ",")
 	return nil
 }

server/plugin.go

@@ -109,6 +109,13 @@ func (p *Plugin) handleEvent(ev model.PluginClusterEvent) error {
 		}
 		us := newUserSession(msg.UserID, msg.ChannelID, "")
 		p.mut.Lock()
+		if _, ok := p.calls[msg.ChannelID]; !ok {
+			p.LogDebug("new call, setting state")
+			p.calls[msg.ChannelID] = &call{
+				channelID: msg.ChannelID,
+				sessions:  map[string]*session{},
+			}
+		}
 		p.sessions[msg.UserID] = us
 		p.mut.Unlock()
 		go p.startSession(us, msg.SenderID)
@@ -117,6 +124,11 @@ func (p *Plugin) handleEvent(ev model.PluginClusterEvent) error {
 			return fmt.Errorf("session doesn't exist, ev=%s, userID=%q, channelID=%q", ev.Id, msg.UserID, msg.ChannelID)
 		}
 		p.LogDebug("disconnect event", "ChannelID", msg.ChannelID, "UserID", msg.UserID)
+		if call := p.getCall(us.channelID); call != nil {
+			if userID := call.getScreenSessionID(); userID == msg.UserID {
+				call.setScreenSession(nil)
+			}
+		}
 		p.mut.Lock()
 		delete(p.sessions, us.userID)
 		p.mut.Unlock()
@@ -150,7 +162,19 @@ func (p *Plugin) handleEvent(ev model.PluginClusterEvent) error {
 		if us == nil {
 			return fmt.Errorf("session doesn't exist, ev=%s, userID=%q, channelID=%q", ev.Id, msg.UserID, msg.ChannelID)
 		}
-		us.trackEnableCh <- (msg.ClientMessage.Type == clientMessageTypeMute)
+		if msg.ClientMessage.Type == clientMessageTypeScreenOff {
+			if call := p.getCall(us.channelID); call != nil {
+				if userID := call.getScreenSessionID(); userID == msg.UserID {
+					call.setScreenSession(nil)
+				}
+			}
+		} else {
+			us.trackEnableCh <- (msg.ClientMessage.Type == clientMessageTypeMute)
+		}
+	case clusterMessageTypeCallEnded:
+		p.mut.Lock()
+		delete(p.calls, msg.ChannelID)
+		p.mut.Unlock()
 	default:
 		return fmt.Errorf("unexpected event type %q", ev.Id)
 	}

server/utils.go

@@ -14,6 +14,26 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+const (
+	handlerKey              = "handler"
+	handlerKeyCheckInterval = 5 * time.Second
+)
+
+func (p *Plugin) getHandlerID() (string, error) {
+	data, appErr := p.API.KVGet(handlerKey)
+	if appErr != nil {
+		return "", fmt.Errorf("failed to get handler id: %w", appErr)
+	}
+	return string(data), nil
+}
+
+func (p *Plugin) setHandlerID(nodeID string) error {
+	if appErr := p.API.KVSetWithExpiry(handlerKey, []byte(nodeID), int64(handlerKeyCheckInterval.Seconds()*2)); appErr != nil {
+		return fmt.Errorf("failed to set handler id: %w", appErr)
+	}
+	return nil
+}
+
 func (p *Plugin) kvSetAtomic(key string, cb func(data []byte) ([]byte, error)) error {
 	for {
 		p.metrics.StoreOpCounters.With(prometheus.Labels{"type": "KVGet"}).Inc()

server/websocket.go

@@ -30,7 +30,7 @@ const (
 	wsEventError            = "error"
 )
 
-func (p *Plugin) handleClientMessageTypeScreen(msg clientMessage, channelID, userID string) error {
+func (p *Plugin) handleClientMessageTypeScreen(msg clientMessage, channelID, userID, handlerID string) error {
 	data := map[string]string{}
 	if msg.Type == clientMessageTypeScreenOn {
 		if err := json.Unmarshal(msg.Data, &data); err != nil {
@@ -69,6 +69,16 @@ func (p *Plugin) handleClientMessageTypeScreen(msg clientMessage, channelID, use
 	}
 
 	if msg.Type == clientMessageTypeScreenOff {
+		if handlerID != p.nodeID {
+			if err := p.sendClusterMessage(clusterMessage{
+				UserID:        userID,
+				ChannelID:     channelID,
+				SenderID:      p.nodeID,
+				ClientMessage: msg,
+			}, clusterMessageTypeUserState, handlerID); err != nil {
+				p.LogError(err.Error())
+			}
+		}
 		p.API.PublishWebSocketEvent(wsEventUserScreenOff, map[string]interface{}{}, &model.WebsocketBroadcast{ChannelId: channelID})
 	}
 
@@ -167,7 +177,7 @@ func (p *Plugin) handleClientMsg(us *session, msg clientMessage, handlerID strin
 			"userID": us.userID,
 		}, &model.WebsocketBroadcast{ChannelId: us.channelID})
 	case clientMessageTypeScreenOn, clientMessageTypeScreenOff:
-		if err := p.handleClientMessageTypeScreen(msg, us.channelID, us.userID); err != nil {
+		if err := p.handleClientMessageTypeScreen(msg, us.channelID, us.userID, handlerID); err != nil {
 			p.LogError(err.Error())
 		}
 	case clientMessageTypeRaiseHand, clientMessageTypeUnraiseHand:
@@ -336,7 +346,14 @@ func (p *Plugin) handleJoin(userID, connID, channelID string) error {
 		"CallID":        state.Call.ID,
 	})
 
-	handlerID := state.NodeID
+	handlerID, err := p.getHandlerID()
+	p.LogDebug("got handlerID", "handlerID", handlerID)
+	if err != nil {
+		p.LogError(err.Error())
+	}
+	if handlerID == "" {
+		handlerID = state.NodeID
+	}
 
 	var wg sync.WaitGroup
 	if handlerID == p.nodeID {
@@ -426,6 +443,18 @@ func (p *Plugin) handleJoin(userID, connID, channelID string) error {
 				"Duration":     dur,
 				"Participants": prevState.Call.Stats.Participants,
 			})
+
+			p.mut.Lock()
+			delete(p.calls, channelID)
+			p.mut.Unlock()
+			if handlerID != p.nodeID {
+				if err := p.sendClusterMessage(clusterMessage{
+					ChannelID: channelID,
+					SenderID:  p.nodeID,
+				}, clusterMessageTypeCallEnded, handlerID); err != nil {
+					p.LogError(err.Error())
+				}
+			}
 		}
 	}
 

webapp/src/client.ts

@@ -158,10 +158,11 @@ export default class CallsClient extends EventEmitter {
                 return;
             }
 
+            const iceServers = config.ICEServers?.length > 0 ? [{urls: config.ICEServers}] : [];
             const peer = new SimplePeer({
                 initiator: true,
                 trickle: true,
-                config: {iceServers: [{urls: config.ICEServers}]},
+                config: {iceServers},
             }) as SimplePeer.Instance;
 
             this.peer = peer;