fix(web): scope inbound thread updates

Resolve inbound events to their authenticated message before clearing
unread state so activity from another conversation cannot mark the open
thread as read.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: c4b0e509-8bcb-4ace-82f4-3e64f6fcc0a6

Author: AchoArnold

api/pkg/handlers/message_thread_handler_test.go

@@ -93,7 +93,7 @@ func TestMessageThreadHandlerUpdate_ReturnsNotFoundWhenThreadIsMissing(t *testin
 func TestMessageThreadHandlerUpdate_RejectsLegacyIsReadPayload(t *testing.T) {
 	logger := &messageThreadHandlerNoopLogger{}
 	tracer := telemetry.NewOtelLogger("test", logger)
-	service := services.NewMessageThreadService(logger, tracer, &messageThreadHandlerRepositoryStub{}, nil, nil)
+	service := services.NewMessageThreadService(logger, tracer, &messageThreadHandlerRepositoryStub{}, nil, nil, nil)
 	handler := NewMessageThreadHandler(logger, tracer, validators.NewMessageThreadHandlerValidator(logger, tracer), service)
 
 	app := fiber.New()

api/pkg/listeners/websocket_listener.go

@@ -19,6 +19,10 @@ type WebsocketListener struct {
 	client *pusher.Client
 }
 
+type websocketMessagePayload struct {
+	MessageID string `json:"message_id"`
+}
+
 // NewWebsocketListener creates a new instance of WebsocketListener
 func NewWebsocketListener(
 	logger telemetry.Logger,
@@ -49,7 +53,9 @@ func (listener *WebsocketListener) onMessageCallMissed(ctx context.Context, even
 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode [%s] into [%T]", event.Data(), payload))
 	}
 
-	if err := listener.client.Trigger(payload.UserID.String(), event.Type(), event.ID()); err != nil {
+	if err := listener.client.Trigger(payload.UserID.String(), event.Type(), websocketMessagePayload{
+		MessageID: payload.MessageID.String(),
+	}); err != nil {
 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot trigger websocket [%s] event with ID [%s] for user with ID [%s]", event.Type(), event.ID(), payload.UserID))
 	}
 	return nil
@@ -82,7 +88,9 @@ func (listener *WebsocketListener) onMessagePhoneReceived(ctx context.Context, e
 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode [%s] into [%T]", event.Data(), payload))
 	}
 
-	if err := listener.client.Trigger(payload.UserID.String(), event.Type(), event.ID()); err != nil {
+	if err := listener.client.Trigger(payload.UserID.String(), event.Type(), websocketMessagePayload{
+		MessageID: payload.MessageID.String(),
+	}); err != nil {
 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot trigger websocket [%s] event with ID [%s] for user with ID [%s]", event.Type(), event.ID(), payload.UserID))
 	}
 

api/pkg/listeners/websocket_listener_test.go

@@ -1,12 +1,22 @@
 package listeners
 
 import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 
+	"github.com/NdoleStudio/httpsms/pkg/entities"
 	"github.com/NdoleStudio/httpsms/pkg/events"
 	"github.com/NdoleStudio/httpsms/pkg/telemetry"
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/google/uuid"
 	"github.com/pusher/pusher-http-go/v5"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestWebsocketListenerRegistersMissedCalls(t *testing.T) {
@@ -16,3 +26,74 @@ func TestWebsocketListenerRegistersMissedCalls(t *testing.T) {
 
 	assert.Contains(t, routes, events.MessageCallMissed)
 }
+
+func TestWebsocketListenerPublishesReceivedMessageID(t *testing.T) {
+	userID := "user-id"
+	messageID := uuid.New()
+	event := cloudevents.NewEvent()
+	event.SetID(uuid.NewString())
+	event.SetType(events.EventTypeMessagePhoneReceived)
+	require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.MessagePhoneReceivedPayload{
+		MessageID: messageID,
+		UserID:    entities.UserID(userID),
+	}))
+
+	payload := captureWebsocketPayload(t, event, userID)
+
+	assert.Equal(t, messageID.String(), payload.MessageID)
+}
+
+func TestWebsocketListenerPublishesMissedCallMessageID(t *testing.T) {
+	userID := "user-id"
+	messageID := uuid.New()
+	event := cloudevents.NewEvent()
+	event.SetID(uuid.NewString())
+	event.SetType(events.MessageCallMissed)
+	require.NoError(t, event.SetData(cloudevents.ApplicationJSON, events.MessageCallMissedPayload{
+		MessageID: messageID,
+		UserID:    entities.UserID(userID),
+	}))
+
+	payload := captureWebsocketPayload(t, event, userID)
+
+	assert.Equal(t, messageID.String(), payload.MessageID)
+}
+
+func captureWebsocketPayload(t *testing.T, event cloudevents.Event, userID string) websocketMessagePayload {
+	t.Helper()
+
+	requestBody := make(chan []byte, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		body, err := io.ReadAll(request.Body)
+		require.NoError(t, err)
+		requestBody <- body
+		writer.Header().Set("Content-Type", "application/json")
+		_, err = writer.Write([]byte(`{}`))
+		require.NoError(t, err)
+	}))
+	t.Cleanup(server.Close)
+
+	logger := &noopListenerLogger{}
+	tracer := telemetry.NewOtelLogger("test", logger)
+	client := &pusher.Client{
+		AppID:      "app-id",
+		Key:        "key",
+		Secret:     "secret",
+		Host:       strings.TrimPrefix(server.URL, "http://"),
+		HTTPClient: server.Client(),
+	}
+	_, routes := NewWebsocketListener(logger, tracer, client)
+
+	require.NoError(t, routes[event.Type()](context.Background(), event))
+
+	var trigger struct {
+		Channels []string `json:"channels"`
+		Data     string   `json:"data"`
+	}
+	require.NoError(t, json.Unmarshal(<-requestBody, &trigger))
+	assert.Equal(t, []string{userID}, trigger.Channels)
+
+	var payload websocketMessagePayload
+	require.NoError(t, json.Unmarshal([]byte(trigger.Data), &payload))
+	return payload
+}

api/pkg/services/message_thread_service_test.go

@@ -125,7 +125,7 @@ func newMessageThreadServiceForTest(repository repositories.MessageThreadReposit
 func newMessageThreadServiceWithPhoneForTest(repository repositories.MessageThreadRepository, phoneRepository repositories.PhoneRepository) *MessageThreadService {
 	logger := &noopLogger{}
 	tracer := telemetry.NewOtelLogger("test", logger)
-	return NewMessageThreadService(logger, tracer, repository, phoneRepository, nil)
+	return NewMessageThreadService(logger, tracer, repository, phoneRepository, nil, nil)
 }
 
 func TestUpdateThreadPassesUnreadWatermarkForInboundActivity(t *testing.T) {

web/app/pages/threads/[id]/index.vue

@@ -62,6 +62,10 @@ const formMessageRules = [
 
 let webhookChannel: Channel | null = null
 
+interface WebsocketMessageEvent {
+  message_id: string
+}
+
 const contactIsPhoneNumber = computed(() => {
   const thread = currentThread.value
   if (!thread) return false
@@ -137,6 +141,30 @@ async function resetCurrentThreadUnreadCount(force = false) {
   }
 }
 
+async function handleInboundMessage(event: WebsocketMessageEvent) {
+  if (loadingMessages.value) return
+
+  try {
+    const message = await messagesStore.getMessage(event.message_id)
+    await threadsStore.loadThreads()
+
+    const thread = currentThread.value
+    if (
+      !thread ||
+      message.owner !== thread.owner ||
+      message.contact !== thread.contact ||
+      loadingMessages.value
+    ) {
+      return
+    }
+
+    await resetCurrentThreadUnreadCount(true)
+    loadMessages(false, false)
+  } catch (error) {
+    console.error(error)
+  }
+}
+
 function loadMessages(hide = true, resetUnreadCount = true) {
   loadingMessages.value = true
   const threadId = route.params.id as string
@@ -257,17 +285,14 @@ onMounted(async () => {
   webhookChannel.bind('message.send.failed', () => {
     if (!loadingMessages.value) loadMessages(false)
   })
-  webhookChannel.bind('message.phone.received', () => {
-    if (!loadingMessages.value) {
-      void resetCurrentThreadUnreadCount(true)
-      loadMessages(false, false)
-    }
-  })
-  webhookChannel.bind('message.call.missed', () => {
-    if (!loadingMessages.value) {
-      void resetCurrentThreadUnreadCount(true)
-      loadMessages(false, false)
-    }
+  webhookChannel.bind(
+    'message.phone.received',
+    (event: WebsocketMessageEvent) => {
+      void handleInboundMessage(event)
+    },
+  )
+  webhookChannel.bind('message.call.missed', (event: WebsocketMessageEvent) => {
+    void handleInboundMessage(event)
   })
 })
 

web/app/stores/messages.ts

@@ -48,6 +48,13 @@ export const useMessagesStore = defineStore('messages', () => {
     })
   }
 
+  async function getMessage(messageId: string): Promise<EntitiesMessage> {
+    const response = await apiFetch<{ data: EntitiesMessage }>(
+      `/v1/messages/${messageId}`,
+    )
+    return response.data
+  }
+
   async function searchMessages(
     payload: SearchMessagesRequest,
   ): Promise<EntitiesMessage[]> {
@@ -88,6 +95,7 @@ export const useMessagesStore = defineStore('messages', () => {
   return {
     sendMessage,
     deleteMessage,
+    getMessage,
     searchMessages,
     sendBulkMessages,
     fetchBulkMessageOrders,