MM-56201, MM-56280: Suppress typing and emoji events (#25794)

We do not send the typing event when the originating
channel is not the active channel or active channel thread.

https://mattermost.atlassian.net/browse/MM-56201

```release-note
NONE
```

Co-authored-by: harshil Sharma <harshilsharma63@gmail.com>
Co-authored-by: Harrison Healey <harrisonmhealey@gmail.com>

server/channels/app/platform/web_conn.go

@@ -25,6 +25,7 @@ import (
 	"github.com/mattermost/mattermost/server/public/shared/i18n"
 	"github.com/mattermost/mattermost/server/public/shared/mlog"
 	"github.com/mattermost/mattermost/server/public/shared/request"
+	"github.com/mattermost/mattermost/server/public/utils"
 )
 
 const (
@@ -48,6 +49,10 @@ const (
 
 const websocketMessagePluginPrefix = "custom_"
 
+// UnsetPresenceIndicator is the value that gets set initially for active channel/
+// thread/team. This is done to differentiate it from an explicitly set empty value.
+const UnsetPresenceIndicator = "<>"
+
 type pluginWSPostedHook struct {
 	connectionID string
 	userID       string
@@ -105,17 +110,19 @@ type WebConn struct {
 	// a reused connection.
 	// It's theoretically possible for this number to wrap around. But we
 	// leave that as an edge-case.
-	reuseCount                      int
-	sessionToken                    atomic.Value
-	session                         atomic.Pointer[model.Session]
-	connectionID                    atomic.Value
+	reuseCount   int
+	sessionToken atomic.Value
+	session      atomic.Pointer[model.Session]
+	connectionID atomic.Value
+
 	activeChannelID                 atomic.Value
 	activeTeamID                    atomic.Value
 	activeRHSThreadChannelID        atomic.Value
 	activeThreadViewThreadChannelID atomic.Value
-	endWritePump                    chan struct{}
-	pumpFinished                    chan struct{}
-	pluginPosted                    chan pluginWSPostedHook
+
+	endWritePump chan struct{}
+	pumpFinished chan struct{}
+	pluginPosted chan pluginWSPostedHook
 
 	// These counters are to suppress spammy websocket.slow
 	// and websocket.full logs which happen continuously, if they
@@ -236,10 +243,12 @@ func (ps *PlatformService) NewWebConn(cfg *WebConnConfig, suite SuiteIFace, runn
 	wc.SetSessionToken(cfg.Session.Token)
 	wc.SetSessionExpiresAt(cfg.Session.ExpiresAt)
 	wc.SetConnectionID(cfg.ConnectionID)
-	wc.SetActiveChannelID("")
-	wc.SetActiveTeamID("")
-	wc.SetActiveRHSThreadChannelID("")
-	wc.SetActiveThreadViewThreadChannelID("")
+	// <> means unset. This is to differentiate from empty value.
+	// Because we need to support mobile clients where the value might be unset.
+	wc.SetActiveChannelID(UnsetPresenceIndicator)
+	wc.SetActiveTeamID(UnsetPresenceIndicator)
+	wc.SetActiveRHSThreadChannelID(UnsetPresenceIndicator)
+	wc.SetActiveThreadViewThreadChannelID(UnsetPresenceIndicator)
 
 	ps.Go(func() {
 		runner.RunMultiHook(func(hooks plugin.Hooks) bool {
@@ -305,6 +314,9 @@ func (wc *WebConn) SetActiveChannelID(id string) {
 
 // GetActiveChannelID returns the active channel id of the connection.
 func (wc *WebConn) GetActiveChannelID() string {
+	if wc.activeChannelID.Load() == nil {
+		return UnsetPresenceIndicator
+	}
 	return wc.activeChannelID.Load().(string)
 }
 
@@ -315,11 +327,17 @@ func (wc *WebConn) SetActiveTeamID(id string) {
 
 // GetActiveTeamID returns the active team id of the connection.
 func (wc *WebConn) GetActiveTeamID() string {
+	if wc.activeTeamID.Load() == nil {
+		return UnsetPresenceIndicator
+	}
 	return wc.activeTeamID.Load().(string)
 }
 
 // GetActiveRHSThreadChannelID returns the channel id of the active thread of the connection.
 func (wc *WebConn) GetActiveRHSThreadChannelID() string {
+	if wc.activeRHSThreadChannelID.Load() == nil {
+		return UnsetPresenceIndicator
+	}
 	return wc.activeRHSThreadChannelID.Load().(string)
 }
 
@@ -330,6 +348,9 @@ func (wc *WebConn) SetActiveRHSThreadChannelID(id string) {
 
 // GetActiveThreadViewThreadChannelID returns the channel id of the active thread of the connection.
 func (wc *WebConn) GetActiveThreadViewThreadChannelID() string {
+	if wc.activeThreadViewThreadChannelID.Load() == nil {
+		return UnsetPresenceIndicator
+	}
 	return wc.activeThreadViewThreadChannelID.Load().(string)
 }
 
@@ -338,6 +359,11 @@ func (wc *WebConn) SetActiveThreadViewThreadChannelID(id string) {
 	wc.activeThreadViewThreadChannelID.Store(id)
 }
 
+// isSet is a helper to check if a value is unset or not.
+func (wc *WebConn) isSet(val string) bool {
+	return val != UnsetPresenceIndicator
+}
+
 // areAllInactive returns whether all of the connections
 // are inactive or not.
 func areAllInactive(conns []*WebConn) bool {
@@ -847,7 +873,18 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	}
 
 	// Only report events to users who are in the channel for the event
-	if msg.GetBroadcast().ChannelId != "" {
+	if chID := msg.GetBroadcast().ChannelId; chID != "" {
+		// For typing events, we don't send them to users who don't have
+		// that channel or thread opened.
+		if wc.Platform.Config().FeatureFlags.WebSocketEventScope &&
+			utils.Contains([]model.WebsocketEventType{
+				model.WebsocketEventTyping,
+				model.WebsocketEventReactionAdded,
+				model.WebsocketEventReactionRemoved,
+			}, msg.EventType()) && wc.notInChannel(chID) && wc.notInThread(chID) {
+			return false
+		}
+
 		if model.GetMillis()-wc.lastAllChannelMembersTime > webConnMemberCacheTime {
 			wc.allChannelMembers = nil
 			wc.lastAllChannelMembersTime = 0
@@ -863,7 +900,7 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 			wc.lastAllChannelMembersTime = model.GetMillis()
 		}
 
-		if _, ok := wc.allChannelMembers[msg.GetBroadcast().ChannelId]; ok {
+		if _, ok := wc.allChannelMembers[chID]; ok {
 			return true
 		}
 		return false
@@ -881,6 +918,15 @@ func (wc *WebConn) ShouldSendEvent(msg *model.WebSocketEvent) bool {
 	return true
 }
 
+func (wc *WebConn) notInChannel(val string) bool {
+	return (wc.isSet(wc.GetActiveChannelID()) && val != wc.GetActiveChannelID())
+}
+
+func (wc *WebConn) notInThread(val string) bool {
+	return (wc.isSet(wc.GetActiveRHSThreadChannelID()) && val != wc.GetActiveRHSThreadChannelID()) &&
+		(wc.isSet(wc.GetActiveThreadViewThreadChannelID()) && val != wc.GetActiveThreadViewThreadChannelID())
+}
+
 // IsMemberOfTeam returns whether the user of the WebConn
 // is a member of the given teamID or not.
 func (wc *WebConn) isMemberOfTeam(teamID string) bool {

server/channels/app/web_conn_test.go

@@ -4,6 +4,7 @@
 package app
 
 import (
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -15,6 +16,8 @@ import (
 )
 
 func TestWebConnShouldSendEvent(t *testing.T) {
+	os.Setenv("MM_FEATUREFLAGS_WEBSOCKETEVENTSCOPE", "true")
+	defer os.Unsetenv("MM_FEATUREFLAGS_WEBSOCKETEVENTSCOPE")
 	th := Setup(t).InitBasic()
 	defer th.TearDown()
 	session, err := th.App.CreateSession(th.Context, &model.Session{UserId: th.BasicUser.Id, Roles: th.BasicUser.GetRawRoles(), TeamMembers: []*model.TeamMember{
@@ -162,6 +165,63 @@ func TestWebConnShouldSendEvent(t *testing.T) {
 		assert.False(t, adminUserWc.ShouldSendEvent(event), "did not expect admin")
 	})
 
+	t.Run("should not send typing event unless in scope", func(t *testing.T) {
+		event2 := model.NewWebSocketEvent(model.WebsocketEventTyping, "", th.BasicChannel.Id, "", nil, "")
+		// Basic, unset case
+		basicUserWc.SetActiveChannelID(platform.UnsetPresenceIndicator)
+		basicUserWc.SetActiveRHSThreadChannelID(platform.UnsetPresenceIndicator)
+		basicUserWc.SetActiveThreadViewThreadChannelID(platform.UnsetPresenceIndicator)
+		assert.True(t, basicUserWc.ShouldSendEvent(event2))
+
+		// Active channel is set to something else, thread unset
+		basicUserWc.SetActiveChannelID("ch1")
+		basicUserWc.SetActiveRHSThreadChannelID(platform.UnsetPresenceIndicator)
+		basicUserWc.SetActiveThreadViewThreadChannelID(platform.UnsetPresenceIndicator)
+		assert.True(t, basicUserWc.ShouldSendEvent(event2))
+
+		// Active channel is unset, thread set
+		basicUserWc.SetActiveChannelID(platform.UnsetPresenceIndicator)
+		basicUserWc.SetActiveRHSThreadChannelID("ch1")
+		basicUserWc.SetActiveThreadViewThreadChannelID("ch2")
+		assert.True(t, basicUserWc.ShouldSendEvent(event2))
+
+		// both are set to correct channel
+		basicUserWc.SetActiveChannelID(th.BasicChannel.Id)
+		basicUserWc.SetActiveRHSThreadChannelID(th.BasicChannel.Id)
+		basicUserWc.SetActiveThreadViewThreadChannelID(th.BasicChannel.Id)
+		assert.True(t, basicUserWc.ShouldSendEvent(event2))
+
+		// channel is correct, thread is something else.
+		basicUserWc.SetActiveChannelID(th.BasicChannel.Id)
+		basicUserWc.SetActiveRHSThreadChannelID("ch1")
+		basicUserWc.SetActiveThreadViewThreadChannelID("ch2")
+		assert.True(t, basicUserWc.ShouldSendEvent(event2))
+
+		// channel is wrong, thread is correct.
+		basicUserWc.SetActiveChannelID("ch1")
+		basicUserWc.SetActiveRHSThreadChannelID(th.BasicChannel.Id)
+		basicUserWc.SetActiveThreadViewThreadChannelID(th.BasicChannel.Id)
+		assert.True(t, basicUserWc.ShouldSendEvent(event2))
+
+		// FINALLY, both are set to something else.
+		basicUserWc.SetActiveChannelID("ch1")
+		basicUserWc.SetActiveRHSThreadChannelID("ch1")
+		basicUserWc.SetActiveThreadViewThreadChannelID("ch2")
+		assert.False(t, basicUserWc.ShouldSendEvent(event2))
+
+		// Different threads and channel
+		basicUserWc.SetActiveChannelID("ch1")
+		basicUserWc.SetActiveRHSThreadChannelID("ch2")
+		basicUserWc.SetActiveThreadViewThreadChannelID("ch3")
+		assert.False(t, basicUserWc.ShouldSendEvent(event2))
+
+		// Other channel. Thread unset explicitly.
+		basicUserWc.SetActiveChannelID("ch1")
+		basicUserWc.SetActiveRHSThreadChannelID("")
+		basicUserWc.SetActiveThreadViewThreadChannelID("")
+		assert.False(t, basicUserWc.ShouldSendEvent(event2))
+	})
+
 	t.Run("should send to basic user and admin in channel2", func(t *testing.T) {
 		event = event.SetBroadcast(&model.WebsocketBroadcast{ChannelId: channel2.Id})
 

server/public/model/feature_flags.go

@@ -49,7 +49,9 @@ type FeatureFlags struct {
 	CloudIPFiltering bool
 	ConsumePostHook  bool
 
-	CloudAnnualRenewals bool
+	CloudAnnualRenewals      bool
+	OutgoingOAuthConnections bool
+	WebSocketEventScope      bool
 }
 
 func (f *FeatureFlags) SetDefaults() {
@@ -69,6 +71,8 @@ func (f *FeatureFlags) SetDefaults() {
 	f.CloudIPFiltering = false
 	f.ConsumePostHook = false
 	f.CloudAnnualRenewals = false
+	f.OutgoingOAuthConnections = false
+	f.WebSocketEventScope = false
 }
 
 // ToMap returns the feature flags as a map[string]string