fix(api): preserve deleted message markers

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 332cfd7b-8a60-4e28-9ea8-61854aa712c3

Author: AchoArnold

api/pkg/entities/message_thread_deleted_item.go

@@ -0,0 +1,8 @@
+package entities
+
+import "github.com/google/uuid"
+
+// MessageThreadDeletedItem records a permanently deleted message activity.
+type MessageThreadDeletedItem struct {
+	MessageID uuid.UUID `gorm:"primaryKey;type:uuid"`
+}

api/pkg/entities/message_thread_test.go

@@ -59,3 +59,15 @@ func TestMessageThreadUnreadItemRetainsCountedState(t *testing.T) {
 	assert.Contains(t, counted.Tag.Get("gorm"), "not null")
 	assert.Contains(t, counted.Tag.Get("gorm"), "default:true")
 }
+
+func TestMessageThreadDeletedItemIsIndependentFromThreadLifecycle(t *testing.T) {
+	itemType := reflect.TypeOf(MessageThreadDeletedItem{})
+
+	require.Equal(t, 1, itemType.NumField())
+	messageID, ok := itemType.FieldByName("MessageID")
+	require.True(t, ok)
+	assert.Contains(t, messageID.Tag.Get("gorm"), "primaryKey")
+	assert.Contains(t, messageID.Tag.Get("gorm"), "type:uuid")
+	_, hasMessageThreadID := itemType.FieldByName("MessageThreadID")
+	assert.False(t, hasMessageThreadID)
+}

api/pkg/listeners/message_thread_listener_test.go

@@ -64,15 +64,6 @@ func TestMessageThreadListenerMarksMissedCallUnread(t *testing.T) {
 
 func TestMessageThreadListenerDeletesNonLastUnreadMessage(t *testing.T) {
 	repository, routes := newMessageThreadListenerForTest()
-	currentLastMessageID := uuid.New()
-	repository.thread = &entities.MessageThread{
-		ID:            uuid.New(),
-		UserID:        entities.UserID("user-id"),
-		Owner:         "+18005550199",
-		Contact:       "+18005550100",
-		LastMessageID: &currentLastMessageID,
-	}
-
 	deletedMessageID := uuid.New()
 	previousMessageID := uuid.New()
 	previousStatus := entities.MessageStatus(entities.MessageStatusDelivered)
@@ -95,6 +86,8 @@ func TestMessageThreadListenerDeletesNonLastUnreadMessage(t *testing.T) {
 
 	require.NoError(t, err)
 	assert.Equal(t, deletedMessageID, repository.deletedUpdate.DeletedMessageID)
+	assert.Equal(t, "+18005550199", repository.deletedUpdate.Owner)
+	assert.Equal(t, "+18005550100", repository.deletedUpdate.Contact)
 	require.NotNil(t, repository.deletedUpdate.LastMessageID)
 	assert.Equal(t, previousMessageID, *repository.deletedUpdate.LastMessageID)
 }

api/pkg/migrations/message_thread_unread_count.go

@@ -20,7 +20,11 @@ func (messageThreadConversationIndex) TableName() string {
 
 // MigrateMessageThreadUnreadCount migrates message thread unread count schema.
 func MigrateMessageThreadUnreadCount(db *gorm.DB) error {
-	if err := db.AutoMigrate(&entities.MessageThread{}, &entities.MessageThreadUnreadItem{}); err != nil {
+	if err := db.AutoMigrate(
+		&entities.MessageThread{},
+		&entities.MessageThreadUnreadItem{},
+		&entities.MessageThreadDeletedItem{},
+	); err != nil {
 		return stacktrace.Propagate(err, "cannot migrate message thread unread count schema")
 	}
 

api/pkg/migrations/message_thread_unread_count_test.go

@@ -28,6 +28,19 @@ func TestMigrateMessageThreadUnreadCountSkipsLegacyBackfillWhenIsReadColumnMissi
 	assert.Contains(t, strings.Join(recorder.execs, "\n"), `"counted" boolean NOT NULL DEFAULT true`)
 }
 
+func TestMigrateMessageThreadUnreadCountCreatesIndependentDeletedItemSchema(t *testing.T) {
+	db, recorder := newMigrationTestDB(t, migrationTestDBOptions{})
+
+	require.NoError(t, MigrateMessageThreadUnreadCount(db))
+
+	deletedItems := migrationExecIndex(recorder, `CREATE TABLE "message_thread_deleted_items"`)
+	require.NotEqual(t, -1, deletedItems)
+	assert.Contains(t, recorder.execs[deletedItems], `"message_id" uuid`)
+	assert.Contains(t, recorder.execs[deletedItems], `PRIMARY KEY`)
+	assert.NotContains(t, recorder.execs[deletedItems], `"message_thread_id"`)
+	assert.NotContains(t, recorder.execs[deletedItems], `FOREIGN KEY`)
+}
+
 func TestMigrateMessageThreadUnreadCountBackfillsBeforeDropAndSkipsOnSecondRun(t *testing.T) {
 	db, recorder := newMigrationTestDB(t, migrationTestDBOptions{hasLegacyIsRead: true})
 

api/pkg/repositories/gorm_message_thread_repository.go

@@ -148,6 +148,28 @@ func insertUnreadItem(tx *gorm.DB, item entities.MessageThreadUnreadItem) (bool,
 	return result.RowsAffected == 1, nil
 }
 
+func insertDeletedItem(tx *gorm.DB, messageID uuid.UUID) error {
+	if err := tx.
+		Clauses(clause.OnConflict{DoNothing: true}).
+		Create(&entities.MessageThreadDeletedItem{MessageID: messageID}).
+		Error; err != nil {
+		return stacktrace.Propagatef(err, "cannot insert deleted message marker for message [%s]", messageID)
+	}
+	return nil
+}
+
+func isDeletedItem(tx *gorm.DB, messageID uuid.UUID) (bool, error) {
+	item := new(entities.MessageThreadDeletedItem)
+	result := tx.
+		Where("message_id = ?", messageID).
+		Limit(1).
+		Find(item)
+	if result.Error != nil {
+		return false, stacktrace.Propagatef(result.Error, "cannot load deleted message marker for message [%s]", messageID)
+	}
+	return result.RowsAffected != 0, nil
+}
+
 func markUnreadItemDeleted(tx *gorm.DB, messageID uuid.UUID, threadID uuid.UUID) (bool, error) {
 	result := tx.
 		Model(&entities.MessageThreadUnreadItem{}).
@@ -251,26 +273,33 @@ func (repository *gormMessageThreadRepository) UpdateAfterDeletedMessage(ctx con
 
 	err := crdbgorm.ExecuteTx(ctx, repository.db, nil, func(tx *gorm.DB) error {
 		tx = tx.WithContext(ctx)
-		thread, err := lockMessageThread(tx, params.UserID, params.MessageThreadID)
+		if err := insertDeletedItem(tx, params.DeletedMessageID); err != nil {
+			return err
+		}
+
+		thread, err := lockMessageThreadByConversation(tx, params.UserID, params.Owner, params.Contact)
+		if stacktrace.GetCode(err) == ErrCodeNotFound {
+			return nil
+		}
 		if err != nil {
 			return err
 		}
 
-		deleted, err := markUnreadItemDeleted(tx, params.DeletedMessageID, params.MessageThreadID)
+		deleted, err := markUnreadItemDeleted(tx, params.DeletedMessageID, thread.ID)
 		if err != nil {
 			return err
 		}
 		if deleted {
 			if err := tx.
 				Model(thread).
 				Where("user_id = ?", params.UserID).
-				Where("id = ?", params.MessageThreadID).
+				Where("id = ?", thread.ID).
 				UpdateColumn("unread_count", gorm.Expr("GREATEST(unread_count - 1, 0)")).
 				Error; err != nil {
 				return stacktrace.Propagatef(
 					err,
 					"cannot decrement unread count for thread [%s] and user [%s]",
-					params.MessageThreadID,
+					thread.ID,
 					params.UserID,
 				)
 			}
@@ -285,13 +314,13 @@ func (repository *gormMessageThreadRepository) UpdateAfterDeletedMessage(ctx con
 		if params.LastMessageID == nil {
 			if err := tx.
 				Where("user_id = ?", params.UserID).
-				Where("id = ?", params.MessageThreadID).
+				Where("id = ?", thread.ID).
 				Delete(&entities.MessageThread{}).
 				Error; err != nil {
 				return stacktrace.Propagatef(
 					err,
 					"cannot delete message thread [%s] for user [%s] after deleting final message [%s]",
-					params.MessageThreadID,
+					thread.ID,
 					params.UserID,
 					params.DeletedMessageID,
 				)
@@ -306,13 +335,13 @@ func (repository *gormMessageThreadRepository) UpdateAfterDeletedMessage(ctx con
 		if err := tx.
 			Model(thread).
 			Where("user_id = ?", params.UserID).
-			Where("id = ?", params.MessageThreadID).
+			Where("id = ?", thread.ID).
 			Updates(updates).
 			Error; err != nil {
 			return stacktrace.Propagatef(
 				err,
 				"cannot update deleted-message metadata for thread [%s] and user [%s]",
-				params.MessageThreadID,
+				thread.ID,
 				params.UserID,
 			)
 		}
@@ -323,9 +352,10 @@ func (repository *gormMessageThreadRepository) UpdateAfterDeletedMessage(ctx con
 			span,
 			stacktrace.Propagatef(
 				err,
-				"cannot apply deleted message [%s] to thread [%s] for user [%s]",
+				"cannot apply deleted message [%s] to conversation [%s/%s] for user [%s]",
 				params.DeletedMessageID,
-				params.MessageThreadID,
+				params.Owner,
+				params.Contact,
 				params.UserID,
 			),
 		)
@@ -341,6 +371,13 @@ func (repository *gormMessageThreadRepository) Store(ctx context.Context, params
 
 	err := crdbgorm.ExecuteTx(ctx, repository.db, nil, func(tx *gorm.DB) error {
 		tx = tx.WithContext(ctx)
+		if params.Thread.LastMessageID != nil {
+			deleted, err := isDeletedItem(tx, *params.Thread.LastMessageID)
+			if err != nil || deleted {
+				return err
+			}
+		}
+
 		candidate := *params.Thread
 		result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&candidate)
 		if result.Error != nil {
@@ -417,6 +454,11 @@ func (repository *gormMessageThreadRepository) UpdateActivity(ctx context.Contex
 
 	err := crdbgorm.ExecuteTx(ctx, repository.db, nil, func(tx *gorm.DB) error {
 		tx = tx.WithContext(ctx)
+		deleted, err := isDeletedItem(tx, params.MessageID)
+		if err != nil || deleted {
+			return err
+		}
+
 		thread, err := lockMessageThread(tx, params.UserID, params.MessageThreadID)
 		if err != nil {
 			return err