MM-56071: Thread presence indicator (#25694)

We also track the channelID of the thread opened.

Additionally on every connection create or re-connect
with an existing queue, we reset the active state to empty
to avoid any edge-cases.

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

```release-note
NONE
```

server/channels/api4/websocket_test.go

@@ -440,6 +440,11 @@ func TestWebSocketPresence(t *testing.T) {
 	resp = <-wsClient.ResponseChannel
 	require.Nil(t, resp.Error)
 	require.Equal(t, resp.SeqReply, wsClient.Sequence-1, "bad sequence number")
+
+	wsClient.UpdateActiveThread("threadID")
+	resp = <-wsClient.ResponseChannel
+	require.Nil(t, resp.Error)
+	require.Equal(t, resp.SeqReply, wsClient.Sequence-1, "bad sequence number")
 }
 
 func TestWebSocketUpgrade(t *testing.T) {

server/channels/app/platform/web_conn.go

@@ -105,15 +105,16 @@ 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
+	reuseCount            int
-	sessionToken    atomic.Value
+	sessionToken          atomic.Value
-	session         atomic.Pointer[model.Session]
+	session               atomic.Pointer[model.Session]
-	connectionID    atomic.Value
+	connectionID          atomic.Value
-	activeChannelID atomic.Value
+	activeChannelID       atomic.Value
-	activeTeamID    atomic.Value
+	activeTeamID          atomic.Value
-	endWritePump    chan struct{}
+	activeThreadChannelID atomic.Value
-	pumpFinished    chan struct{}
+	endWritePump          chan struct{}
-	pluginPosted    chan pluginWSPostedHook
+	pumpFinished          chan struct{}
+	pluginPosted          chan pluginWSPostedHook
 
 	// These counters are to suppress spammy websocket.slow
 	// and websocket.full logs which happen continuously, if they
@@ -234,6 +235,9 @@ 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.SetActiveThreadChannelID("")
 
 	ps.Go(func() {
 		runner.RunMultiHook(func(hooks plugin.Hooks) bool {
@@ -312,6 +316,16 @@ func (wc *WebConn) GetActiveTeamID() string {
 	return wc.activeTeamID.Load().(string)
 }
 
+// GetActiveThreadChannelID returns the channel id of the active thread of the connection.
+func (wc *WebConn) GetActiveThreadChannelID() string {
+	return wc.activeThreadChannelID.Load().(string)
+}
+
+// SetActiveThreadChannelID sets the channel id of the active thread of the connection.
+func (wc *WebConn) SetActiveThreadChannelID(id string) {
+	wc.activeThreadChannelID.Store(id)
+}
+
 // areAllInactive returns whether all of the connections
 // are inactive or not.
 func areAllInactive(conns []*WebConn) bool {

server/channels/app/platform/websocket_router.go

@@ -82,6 +82,10 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque
 			// Set active team
 			conn.SetActiveTeamID(teamID)
 		}
+		if thChannelID, ok := r.Data["thread_channel_id"].(string); ok {
+			// Set the channelID of the active thread.
+			conn.SetActiveThreadChannelID(thChannelID)
+		}
 
 		resp := model.NewWebSocketResponse(model.StatusOk, r.Seq, nil)
 		hub := conn.Platform.GetHubForUserId(conn.UserId)

server/public/model/websocket_client.go

@@ -342,6 +342,14 @@ func (wsc *WebSocketClient) UpdateActiveTeam(teamID string) {
 	wsc.SendMessage(string(WebsocketPresenceIndicator), data)
 }
 
+// UpdateActiveThread sets the channel id of the current thread that the user is in.
+func (wsc *WebSocketClient) UpdateActiveThread(channelID string) {
+	data := map[string]any{
+		"thread_channel_id": channelID,
+	}
+	wsc.SendMessage(string(WebsocketPresenceIndicator), data)
+}
+
 func (wsc *WebSocketClient) configurePingHandling() {
 	wsc.Conn.SetPingHandler(wsc.pingHandler)
 	wsc.pingTimeoutTimer = time.NewTimer(time.Second * (60 + PingTimeoutBufferSeconds))

webapp/channels/src/components/threading/thread_viewer/thread_viewer.tsx

@@ -15,6 +15,8 @@ import deferComponentRender from 'components/deferComponentRender';
 import FileUploadOverlay from 'components/file_upload_overlay';
 import LoadingScreen from 'components/loading_screen';
 
+import WebSocketClient from 'client/web_websocket_client';
+
 import type {FakePost} from 'types/store/rhs';
 
 import ThreadViewerVirtualized from '../virtualized_thread_viewer';
@@ -78,6 +80,10 @@ export default class ThreadViewer extends React.PureComponent<Props, State> {
         }
     }
 
+    public componentWillUnmount() {
+        WebSocketClient.updateActiveThread('');
+    }
+
     public componentDidUpdate(prevProps: Props) {
         const reconnected = this.props.socketConnectionStatus && !prevProps.socketConnectionStatus;
 
@@ -177,6 +183,9 @@ export default class ThreadViewer extends React.PureComponent<Props, State> {
             await this.fetchThread();
         }
 
+        if (this.props.channel) {
+            WebSocketClient.updateActiveThread(this.props.channel?.id);
+        }
         this.setState({isLoading: false});
     };
 

webapp/platform/client/src/websocket.ts

@@ -396,6 +396,13 @@ export default class WebSocketClient {
         this.sendMessage('presence', data, callback);
     }
 
+    updateActiveThread(channelId: string, callback?: (msg: any) => void) {
+        const data = {
+            thread_channel_id: channelId,
+        };
+        this.sendMessage('presence', data, callback);
+    }
+
     userUpdateActiveStatus(userIsActive: boolean, manual: boolean, callback?: () => void) {
         const data = {
             user_is_active: userIsActive,