[MM-19914] Fix data races in WebSocketEvent (#13039)

* Make WebSocketEvent type immutable

* Update code to use updated immutable WebSocketEvent type

* Export WebSocketEvent fields and mark them as deprecated

app/plugin_api.go

@@ -734,11 +734,9 @@ func (api *PluginAPI) KVList(page, perPage int) ([]string, *model.AppError) {
 }
 
 func (api *PluginAPI) PublishWebSocketEvent(event string, payload map[string]interface{}, broadcast *model.WebsocketBroadcast) {
-	api.app.Publish(&model.WebSocketEvent{
-		Event:     fmt.Sprintf("custom_%v_%v", api.id, event),
-		Data:      payload,
-		Broadcast: broadcast,
-	})
+	ev := model.NewWebSocketEvent(fmt.Sprintf("custom_%v_%v", api.id, event), "", "", "", nil)
+	ev = ev.SetBroadcast(broadcast).SetData(payload)
+	api.app.Publish(ev)
 }
 
 func (api *PluginAPI) HasPermissionTo(userId string, permission *model.Permission) bool {

app/plugin_statuses.go

@@ -79,7 +79,7 @@ func (a *App) notifyPluginStatusesChanged() error {
 	// Notify any system admins.
 	message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_PLUGIN_STATUSES_CHANGED, "", "", "", nil)
 	message.Add("plugin_statuses", pluginStatuses)
-	message.Broadcast.ContainsSensitiveData = true
+	message.GetBroadcast().ContainsSensitiveData = true
 	a.Publish(message)
 
 	return nil

app/user.go

@@ -1112,13 +1112,13 @@ func (a *App) sendUpdatedUserEvent(user model.User) {
 	a.SanitizeProfile(adminCopyOfUser, true)
 	adminMessage := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_USER_UPDATED, "", "", "", nil)
 	adminMessage.Add("user", &adminCopyOfUser)
-	adminMessage.Broadcast.ContainsSensitiveData = true
+	adminMessage.GetBroadcast().ContainsSensitiveData = true
 	a.Publish(adminMessage)
 
 	a.SanitizeProfile(&user, false)
 	message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_USER_UPDATED, "", "", "", nil)
 	message.Add("user", &user)
-	message.Broadcast.ContainsSanitizedData = true
+	message.GetBroadcast().ContainsSanitizedData = true
 	a.Publish(message)
 }
 

app/web_conn.go

@@ -185,7 +185,7 @@ func (c *WebConn) writePump() {
 						"websocket.slow: dropping message",
 						mlog.String("user_id", c.UserId),
 						mlog.String("type", msg.EventType()),
-						mlog.String("channel_id", evt.Broadcast.ChannelId),
+						mlog.String("channel_id", evt.GetBroadcast().ChannelId),
 					)
 					skipSend = true
 				}
@@ -194,9 +194,7 @@ func (c *WebConn) writePump() {
 			if !skipSend {
 				var msgBytes []byte
 				if evtOk {
-					cpyEvt := &model.WebSocketEvent{}
-					*cpyEvt = *evt
-					cpyEvt.Sequence = c.Sequence
+					cpyEvt := evt.SetSequence(c.Sequence)
 					msgBytes = []byte(cpyEvt.ToJson())
 					c.Sequence++
 				} else {
@@ -209,7 +207,7 @@ func (c *WebConn) writePump() {
 							"websocket.full",
 							mlog.String("user_id", c.UserId),
 							mlog.String("type", msg.EventType()),
-							mlog.String("channel_id", evt.Broadcast.ChannelId),
+							mlog.String("channel_id", evt.GetBroadcast().ChannelId),
 							mlog.Int("size", len(msg.ToJson())),
 						)
 					} else {
@@ -305,11 +303,11 @@ func (webCon *WebConn) shouldSendEventToGuest(msg *model.WebSocketEvent) bool {
 	var userId string
 	var canSee bool
 
-	switch msg.Event {
+	switch msg.EventType() {
 	case model.WEBSOCKET_EVENT_USER_UPDATED:
-		userId = msg.Data["user"].(*model.User).Id
+		userId = msg.GetData()["user"].(*model.User).Id
 	case model.WEBSOCKET_EVENT_NEW_USER:
-		userId = msg.Data["user_id"].(string)
+		userId = msg.GetData()["user_id"].(string)
 	default:
 		return true
 	}
@@ -332,7 +330,7 @@ func (webCon *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	// If the event contains sanitized data, only send to users that don't have permission to
 	// see sensitive data. Prevents admin clients from receiving events with bad data
 	var hasReadPrivateDataPermission *bool
-	if msg.Broadcast.ContainsSanitizedData {
+	if msg.GetBroadcast().ContainsSanitizedData {
 		hasReadPrivateDataPermission = model.NewBool(webCon.App.RolesGrantPermission(webCon.GetSession().GetUserRoles(), model.PERMISSION_MANAGE_SYSTEM.Id))
 
 		if *hasReadPrivateDataPermission {
@@ -341,7 +339,7 @@ func (webCon *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	}
 
 	// If the event contains sensitive data, only send to users with permission to see it
-	if msg.Broadcast.ContainsSensitiveData {
+	if msg.GetBroadcast().ContainsSensitiveData {
 		if hasReadPrivateDataPermission == nil {
 			hasReadPrivateDataPermission = model.NewBool(webCon.App.RolesGrantPermission(webCon.GetSession().GetUserRoles(), model.PERMISSION_MANAGE_SYSTEM.Id))
 		}
@@ -352,19 +350,19 @@ func (webCon *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	}
 
 	// If the event is destined to a specific user
-	if len(msg.Broadcast.UserId) > 0 {
-		return webCon.UserId == msg.Broadcast.UserId
+	if len(msg.GetBroadcast().UserId) > 0 {
+		return webCon.UserId == msg.GetBroadcast().UserId
 	}
 
 	// if the user is omitted don't send the message
-	if len(msg.Broadcast.OmitUsers) > 0 {
-		if _, ok := msg.Broadcast.OmitUsers[webCon.UserId]; ok {
+	if len(msg.GetBroadcast().OmitUsers) > 0 {
+		if _, ok := msg.GetBroadcast().OmitUsers[webCon.UserId]; ok {
 			return false
 		}
 	}
 
 	// Only report events to users who are in the channel for the event
-	if len(msg.Broadcast.ChannelId) > 0 {
+	if len(msg.GetBroadcast().ChannelId) > 0 {
 		if model.GetMillis()-webCon.LastAllChannelMembersTime > WEBCONN_MEMBER_CACHE_TIME {
 			webCon.AllChannelMembers = nil
 			webCon.LastAllChannelMembersTime = 0
@@ -380,15 +378,15 @@ func (webCon *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 			webCon.LastAllChannelMembersTime = model.GetMillis()
 		}
 
-		if _, ok := webCon.AllChannelMembers[msg.Broadcast.ChannelId]; ok {
+		if _, ok := webCon.AllChannelMembers[msg.GetBroadcast().ChannelId]; ok {
 			return true
 		}
 		return false
 	}
 
 	// Only report events to users who are in the team for the event
-	if len(msg.Broadcast.TeamId) > 0 {
-		return webCon.IsMemberOfTeam(msg.Broadcast.TeamId)
+	if len(msg.GetBroadcast().TeamId) > 0 {
+		return webCon.IsMemberOfTeam(msg.GetBroadcast().TeamId)
 	}
 
 	if webCon.GetSession().Props[model.SESSION_PROP_IS_GUEST] == "true" {

app/web_conn_test.go

@@ -72,9 +72,9 @@ func TestWebConnShouldSendEvent(t *testing.T) {
 		// needs more cases to get full coverage
 	}
 
-	event := &model.WebSocketEvent{Event: "some_event"}
+	event := model.NewWebSocketEvent("some_event", "", "", "", nil)
 	for _, c := range cases {
-		event.Broadcast = c.Broadcast
+		event = event.SetBroadcast(c.Broadcast)
 		assert.Equal(t, c.User1Expected, basicUserWc.ShouldSendEvent(event), c.Description)
 		assert.Equal(t, c.User2Expected, basicUser2Wc.ShouldSendEvent(event), c.Description)
 		assert.Equal(t, c.AdminExpected, adminUserWc.ShouldSendEvent(event), c.Description)

app/web_hub.go

@@ -164,7 +164,7 @@ func (a *App) HubUnregister(webConn *WebConn) {
 
 func (a *App) Publish(message *model.WebSocketEvent) {
 	if metrics := a.Metrics; metrics != nil {
-		metrics.IncrementWebsocketEvent(message.Event)
+		metrics.IncrementWebsocketEvent(message.EventType())
 	}
 
 	a.PublishSkipClusterSend(message)
@@ -176,11 +176,11 @@ func (a *App) Publish(message *model.WebSocketEvent) {
 			Data:     message.ToJson(),
 		}
 
-		if message.Event == model.WEBSOCKET_EVENT_POSTED ||
-			message.Event == model.WEBSOCKET_EVENT_POST_EDITED ||
-			message.Event == model.WEBSOCKET_EVENT_DIRECT_ADDED ||
-			message.Event == model.WEBSOCKET_EVENT_GROUP_ADDED ||
-			message.Event == model.WEBSOCKET_EVENT_ADDED_TO_TEAM {
+		if message.EventType() == model.WEBSOCKET_EVENT_POSTED ||
+			message.EventType() == model.WEBSOCKET_EVENT_POST_EDITED ||
+			message.EventType() == model.WEBSOCKET_EVENT_DIRECT_ADDED ||
+			message.EventType() == model.WEBSOCKET_EVENT_GROUP_ADDED ||
+			message.EventType() == model.WEBSOCKET_EVENT_ADDED_TO_TEAM {
 			cm.SendType = model.CLUSTER_SEND_RELIABLE
 		}
 
@@ -189,8 +189,8 @@ func (a *App) Publish(message *model.WebSocketEvent) {
 }
 
 func (a *App) PublishSkipClusterSend(message *model.WebSocketEvent) {
-	if message.Broadcast.UserId != "" {
-		hub := a.GetHubForUserId(message.Broadcast.UserId)
+	if message.GetBroadcast().UserId != "" {
+		hub := a.GetHubForUserId(message.GetBroadcast().UserId)
 		if hub != nil {
 			hub.Broadcast(message)
 		}
@@ -485,10 +485,10 @@ func (h *Hub) Start() {
 				}
 			case msg := <-h.broadcast:
 				candidates := connections.All()
-				if msg.Broadcast.UserId != "" {
-					candidates = connections.ForUser(msg.Broadcast.UserId)
+				if msg.GetBroadcast().UserId != "" {
+					candidates = connections.ForUser(msg.GetBroadcast().UserId)
 				}
-				msg.PrecomputeJSON()
+				msg = msg.PrecomputeJSON()
 				for _, webCon := range candidates {
 					if webCon.ShouldSendEvent(msg) {
 						select {

app/web_hub_test.go

@@ -91,7 +91,7 @@ func TestHubStopRaceCondition(t *testing.T) {
 		hub.UpdateActivity("userId", "sessionToken", 0)
 
 		for i := 0; i <= BROADCAST_QUEUE_SIZE; i++ {
-			hub.Broadcast(&model.WebSocketEvent{})
+			hub.Broadcast(model.NewWebSocketEvent("", "", "", "", nil))
 		}
 
 		hub.InvalidateUser("userId")