fix(api): return updated thread atomically

Use PostgreSQL RETURNING so concurrent activity cannot make status update responses stale.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: bf00a0ac-e11f-4015-b295-3cdd9b491229

Author: AchoArnold

api/pkg/handlers/message_thread_handler_test.go

@@ -33,8 +33,8 @@ func (stub *messageThreadHandlerRepositoryStub) UpdateActivity(context.Context,
 	return nil
 }
 
-func (stub *messageThreadHandlerRepositoryStub) UpdateStatus(context.Context, entities.UserID, uuid.UUID, repositories.MessageThreadStatusUpdate) error {
-	return stacktrace.PropagateWithCode(gorm.ErrRecordNotFound, repositories.ErrCodeNotFound, "not found")
+func (stub *messageThreadHandlerRepositoryStub) UpdateStatus(context.Context, entities.UserID, uuid.UUID, repositories.MessageThreadStatusUpdate) (*entities.MessageThread, error) {
+	return nil, stacktrace.PropagateWithCode(gorm.ErrRecordNotFound, repositories.ErrCodeNotFound, "not found")
 }
 
 func (stub *messageThreadHandlerRepositoryStub) LoadByOwnerContact(context.Context, entities.UserID, string, string) (*entities.MessageThread, error) {

api/pkg/listeners/read_receipts_test_helpers_test.go

@@ -36,8 +36,8 @@ func (repository *listenerMessageThreadRepository) UpdateActivity(_ context.Cont
 	return nil
 }
 
-func (repository *listenerMessageThreadRepository) UpdateStatus(context.Context, entities.UserID, uuid.UUID, repositories.MessageThreadStatusUpdate) error {
-	return nil
+func (repository *listenerMessageThreadRepository) UpdateStatus(_ context.Context, _ entities.UserID, threadID uuid.UUID, _ repositories.MessageThreadStatusUpdate) (*entities.MessageThread, error) {
+	return &entities.MessageThread{ID: threadID}, nil
 }
 
 func (repository *listenerMessageThreadRepository) UpdateAfterDeletedMessage(context.Context, repositories.MessageThreadDeletedUpdate) error {

api/pkg/repositories/gorm_message_thread_repository.go

@@ -40,14 +40,22 @@ func messageThreadActivityUpdates(params MessageThreadActivityUpdate) map[string
 		"order_timestamp":      params.Timestamp,
 		"last_message_id":      params.MessageID,
 		"last_message_content": params.Content,
-		"status":               string(params.Status),
+		"status":               params.Status,
 	}
 	if params.Unarchive {
 		updates["is_archived"] = false
 	}
 	return updates
 }
 
+func messageThreadDeletedUpdates(params MessageThreadDeletedUpdate) map[string]any {
+	return map[string]any{
+		"last_message_id":      params.LastMessageID,
+		"last_message_content": params.LastMessageContent,
+		"status":               params.LastMessageStatus,
+	}
+}
+
 func messageThreadStatusUpdates(params MessageThreadStatusUpdate) map[string]any {
 	updates := make(map[string]any)
 	if params.IsArchived != nil {
@@ -95,11 +103,7 @@ func (repository *gormMessageThreadRepository) UpdateAfterDeletedMessage(ctx con
 		Model(&entities.MessageThread{}).
 		Where("user_id = ?", params.UserID).
 		Where("id = ?", params.MessageThreadID).
-		Updates(map[string]any{
-			"last_message_id":      params.LastMessageID,
-			"last_message_content": params.LastMessageContent,
-			"status":               string(params.LastMessageStatus),
-		})
+		Updates(messageThreadDeletedUpdates(params))
 	if result.Error != nil {
 		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagate(result.Error, "cannot update deleted-message metadata for thread [%s]", params.MessageThreadID))
 	}
@@ -183,23 +187,25 @@ func (repository *gormMessageThreadRepository) UpdateStatus(
 	userID entities.UserID,
 	messageThreadID uuid.UUID,
 	params MessageThreadStatusUpdate,
-) error {
+) (*entities.MessageThread, error) {
 	ctx, span := repository.tracer.Start(ctx)
 	defer span.End()
 
+	thread := new(entities.MessageThread)
 	result := repository.db.WithContext(ctx).
-		Model(&entities.MessageThread{}).
+		Model(thread).
+		Clauses(clause.Returning{}).
 		Where("user_id = ?", userID).
 		Where("id = ?", messageThreadID).
 		Updates(messageThreadStatusUpdates(params))
 	if result.Error != nil {
-		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagate(result.Error, "cannot update status for thread [%s]", messageThreadID))
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagate(result.Error, "cannot update status for thread [%s]", messageThreadID))
 	}
 	if result.RowsAffected == 0 {
-		return repository.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCode(gorm.ErrRecordNotFound, ErrCodeNotFound, "thread with id [%s] not found", messageThreadID))
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCode(gorm.ErrRecordNotFound, ErrCodeNotFound, "thread with id [%s] not found", messageThreadID))
 	}
 
-	return nil
+	return thread, nil
 }
 
 // LoadByOwnerContact a thread between 2 users

api/pkg/repositories/gorm_message_thread_repository_test.go

@@ -116,13 +116,29 @@ func TestMessageThreadActivityUpdatesOwnOnlyMessageColumns(t *testing.T) {
 		"order_timestamp":      time.Date(2026, 7, 18, 7, 0, 0, 0, time.UTC),
 		"last_message_id":      messageID,
 		"last_message_content": "hello",
-		"status":               entities.MessageStatusReceived,
+		"status":               entities.MessageStatus(entities.MessageStatusReceived),
 	}, updates)
 	assert.NotContains(t, updates, "is_read")
 	assert.NotContains(t, updates, "is_archived")
 	assert.NotContains(t, updates, "last_read_at")
 }
 
+func TestMessageThreadDeletedUpdatesPreserveStatusType(t *testing.T) {
+	messageID := uuid.New()
+	content := "previous message"
+	updates := messageThreadDeletedUpdates(MessageThreadDeletedUpdate{
+		LastMessageID:      &messageID,
+		LastMessageContent: &content,
+		LastMessageStatus:  entities.MessageStatusDelivered,
+	})
+
+	assert.Equal(t, map[string]any{
+		"last_message_id":      &messageID,
+		"last_message_content": &content,
+		"status":               entities.MessageStatus(entities.MessageStatusDelivered),
+	}, updates)
+}
+
 func TestMessageThreadStatusUpdatesReadOnly(t *testing.T) {
 	isRead := true
 	readAt := time.Date(2026, 7, 18, 7, 1, 0, 0, time.UTC)

api/pkg/repositories/message_thread_repository.go

@@ -44,7 +44,7 @@ type MessageThreadRepository interface {
 	UpdateActivity(ctx context.Context, params MessageThreadActivityUpdate) error
 
 	// UpdateStatus persists archive/read status fields for a thread
-	UpdateStatus(ctx context.Context, userID entities.UserID, messageThreadID uuid.UUID, params MessageThreadStatusUpdate) error
+	UpdateStatus(ctx context.Context, userID entities.UserID, messageThreadID uuid.UUID, params MessageThreadStatusUpdate) (*entities.MessageThread, error)
 
 	// LoadByOwnerContact fetches a thread between owner and contact
 	LoadByOwnerContact(ctx context.Context, userID entities.UserID, owner string, contact string) (*entities.MessageThread, error)

api/pkg/services/message_thread_service.go

@@ -128,7 +128,7 @@ func (service *MessageThreadService) UpdateThread(ctx context.Context, params Me
 		return service.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCode(err, stacktrace.GetCode(err), "cannot update message thread with id [%s] after adding message [%s]", thread.ID, params.MessageID))
 	}
 
-	ctxLogger.Info(fmt.Sprintf("thread with id [%s] updated with last message [%s] and status [%s]", thread.ID, thread.LastMessageID, thread.Status))
+	ctxLogger.Info(fmt.Sprintf("thread with id [%s] updated with last message [%s] and status [%s]", thread.ID, params.MessageID, params.Status))
 	return nil
 }
 
@@ -145,30 +145,16 @@ func (service *MessageThreadService) UpdateStatus(ctx context.Context, params Me
 	ctx, span := service.tracer.Start(ctx)
 	defer span.End()
 
-	thread, err := service.repository.Load(ctx, params.UserID, params.MessageThreadID)
-	if err != nil {
-		return nil, service.tracer.WrapErrorSpan(span, stacktrace.Propagate(err, "cannot find thread with id [%s]", params.MessageThreadID))
-	}
-
 	update := repositories.MessageThreadStatusUpdate{
 		IsArchived: params.IsArchived,
 		IsRead:     params.IsRead,
 		ReadAt:     time.Now().UTC(),
 	}
-	if err = service.repository.UpdateStatus(ctx, params.UserID, params.MessageThreadID, update); err != nil {
+	thread, err := service.repository.UpdateStatus(ctx, params.UserID, params.MessageThreadID, update)
+	if err != nil {
 		return nil, service.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCode(err, stacktrace.GetCode(err), "cannot update message thread with id [%s]", params.MessageThreadID))
 	}
 
-	if params.IsArchived != nil {
-		thread.IsArchived = *params.IsArchived
-	}
-	if params.IsRead != nil {
-		thread.IsRead = *params.IsRead
-		if *params.IsRead {
-			thread.LastReadAt = update.ReadAt
-		}
-	}
-
 	return thread, nil
 }
 

api/pkg/services/message_thread_service_test.go

@@ -20,7 +20,7 @@ type messageThreadRepositoryStub struct {
 	load               func(context.Context, entities.UserID, uuid.UUID) (*entities.MessageThread, error)
 	store              func(context.Context, *entities.MessageThread) error
 	updateActivity     func(context.Context, repositories.MessageThreadActivityUpdate) error
-	updateStatus       func(context.Context, entities.UserID, uuid.UUID, repositories.MessageThreadStatusUpdate) error
+	updateStatus       func(context.Context, entities.UserID, uuid.UUID, repositories.MessageThreadStatusUpdate) (*entities.MessageThread, error)
 }
 
 func (stub *messageThreadRepositoryStub) Store(ctx context.Context, thread *entities.MessageThread) error {
@@ -37,11 +37,11 @@ func (stub *messageThreadRepositoryStub) UpdateActivity(ctx context.Context, par
 	return nil
 }
 
-func (stub *messageThreadRepositoryStub) UpdateStatus(ctx context.Context, userID entities.UserID, threadID uuid.UUID, params repositories.MessageThreadStatusUpdate) error {
+func (stub *messageThreadRepositoryStub) UpdateStatus(ctx context.Context, userID entities.UserID, threadID uuid.UUID, params repositories.MessageThreadStatusUpdate) (*entities.MessageThread, error) {
 	if stub.updateStatus != nil {
 		return stub.updateStatus(ctx, userID, threadID, params)
 	}
-	return nil
+	return &entities.MessageThread{ID: threadID}, nil
 }
 
 func (stub *messageThreadRepositoryStub) UpdateAfterDeletedMessage(context.Context, repositories.MessageThreadDeletedUpdate) error {
@@ -181,16 +181,10 @@ func TestUpdateStatusChangesOnlyRequestedState(t *testing.T) {
 	threadID := uuid.New()
 	isRead := false
 	var captured repositories.MessageThreadStatusUpdate
-	var calls []string
 	repository := &messageThreadRepositoryStub{
-		updateStatus: func(_ context.Context, _ entities.UserID, _ uuid.UUID, params repositories.MessageThreadStatusUpdate) error {
-			calls = append(calls, "update")
+		updateStatus: func(_ context.Context, _ entities.UserID, _ uuid.UUID, params repositories.MessageThreadStatusUpdate) (*entities.MessageThread, error) {
 			captured = params
-			return nil
-		},
-		load: func(context.Context, entities.UserID, uuid.UUID) (*entities.MessageThread, error) {
-			calls = append(calls, "load")
-			return &entities.MessageThread{ID: threadID, IsArchived: true, IsRead: true}, nil
+			return &entities.MessageThread{ID: threadID, IsArchived: true, IsRead: false}, nil
 		},
 	}
 
@@ -207,16 +201,12 @@ func TestUpdateStatusChangesOnlyRequestedState(t *testing.T) {
 	assert.False(t, captured.ReadAt.IsZero())
 	assert.True(t, thread.IsArchived)
 	assert.False(t, thread.IsRead)
-	assert.Equal(t, []string{"load", "update"}, calls)
 }
 
 func TestUpdateStatusPreservesNotFoundCode(t *testing.T) {
 	repository := &messageThreadRepositoryStub{
-		load: func(context.Context, entities.UserID, uuid.UUID) (*entities.MessageThread, error) {
-			return &entities.MessageThread{ID: uuid.New()}, nil
-		},
-		updateStatus: func(context.Context, entities.UserID, uuid.UUID, repositories.MessageThreadStatusUpdate) error {
-			return stacktrace.PropagateWithCode(gorm.ErrRecordNotFound, repositories.ErrCodeNotFound, "not found")
+		updateStatus: func(context.Context, entities.UserID, uuid.UUID, repositories.MessageThreadStatusUpdate) (*entities.MessageThread, error) {
+			return nil, stacktrace.PropagateWithCode(gorm.ErrRecordNotFound, repositories.ErrCodeNotFound, "not found")
 		},
 	}
 

tests/read_receipts_test.go

@@ -158,6 +158,9 @@ func TestMessageThreadReadReceipts(t *testing.T) {
 
 	updated := markMessageThreadRead(ctx, t, thread.ID)
 	assert.True(t, updated.IsRead)
+	assert.Equal(t, contact, updated.Contact)
+	require.NotNil(t, updated.LastMessageContent)
+	assert.Equal(t, "Unread inbound message", *updated.LastMessageContent)
 	waitForMessageThread(ctx, t, phone.PhoneNumber, contact, 10*time.Second, func(thread integrationMessageThread) bool {
 		return thread.IsRead
 	})