feat: count unread thread messages

api/docs/docs.go

@@ -4129,12 +4129,12 @@ const docTemplate = `{
                 "created_at",
                 "id",
                 "is_archived",
-                "is_read",
                 "last_message_content",
                 "last_message_id",
                 "order_timestamp",
                 "owner",
                 "status",
+                "unread_count",
                 "updated_at",
                 "user_id"
             ],
@@ -4167,10 +4167,6 @@ const docTemplate = `{
                     "type": "boolean",
                     "example": false
                 },
-                "is_read": {
-                    "type": "boolean",
-                    "example": true
-                },
                 "last_message_content": {
                     "type": "string",
                     "example": "This is a sample message content"
@@ -4191,6 +4187,10 @@ const docTemplate = `{
                     "type": "string",
                     "example": "PENDING"
                 },
+                "unread_count": {
+                    "type": "integer",
+                    "example": 2
+                },
                 "updated_at": {
                     "type": "string",
                     "example": "2022-06-05T14:26:09.527976+03:00"
@@ -4920,9 +4920,11 @@ const docTemplate = `{
                     "type": "boolean",
                     "example": true
                 },
-                "is_read": {
-                    "type": "boolean",
-                    "example": true
+                "unread_count": {
+                    "type": "integer",
+                    "maximum": 0,
+                    "minimum": 0,
+                    "example": 0
                 }
             }
         },

api/docs/swagger.json

@@ -4126,12 +4126,12 @@
                 "created_at",
                 "id",
                 "is_archived",
-                "is_read",
                 "last_message_content",
                 "last_message_id",
                 "order_timestamp",
                 "owner",
                 "status",
+                "unread_count",
                 "updated_at",
                 "user_id"
             ],
@@ -4164,10 +4164,6 @@
                     "type": "boolean",
                     "example": false
                 },
-                "is_read": {
-                    "type": "boolean",
-                    "example": true
-                },
                 "last_message_content": {
                     "type": "string",
                     "example": "This is a sample message content"
@@ -4188,6 +4184,10 @@
                     "type": "string",
                     "example": "PENDING"
                 },
+                "unread_count": {
+                    "type": "integer",
+                    "example": 2
+                },
                 "updated_at": {
                     "type": "string",
                     "example": "2022-06-05T14:26:09.527976+03:00"
@@ -4917,9 +4917,11 @@
                     "type": "boolean",
                     "example": true
                 },
-                "is_read": {
-                    "type": "boolean",
-                    "example": true
+                "unread_count": {
+                    "type": "integer",
+                    "maximum": 0,
+                    "minimum": 0,
+                    "example": 0
                 }
             }
         },

api/docs/swagger.yaml

@@ -367,9 +367,6 @@ definitions:
       is_archived:
         example: false
         type: boolean
-      is_read:
-        example: true
-        type: boolean
       last_message_content:
         example: This is a sample message content
         type: string
@@ -385,6 +382,9 @@ definitions:
       status:
         example: PENDING
         type: string
+      unread_count:
+        example: 2
+        type: integer
       updated_at:
         example: "2022-06-05T14:26:09.527976+03:00"
         type: string
@@ -397,12 +397,12 @@ definitions:
     - created_at
     - id
     - is_archived
-    - is_read
     - last_message_content
     - last_message_id
     - order_timestamp
     - owner
     - status
+    - unread_count
     - updated_at
     - user_id
     type: object
@@ -956,9 +956,11 @@ definitions:
       is_archived:
         example: true
         type: boolean
-      is_read:
-        example: true
-        type: boolean
+      unread_count:
+        example: 0
+        maximum: 0
+        minimum: 0
+        type: integer
     type: object
   requests.PhoneAPIKeyStoreRequest:
     properties:

api/pkg/di/container.go

@@ -380,7 +380,39 @@ ALTER TABLE discords ADD CONSTRAINT IF NOT EXISTS uni_discords_server_id CHECK (
 	}
 
 	if err = db.AutoMigrate(&entities.MessageThread{}); err != nil {
-		container.logger.Fatal(stacktrace.Propagatef(err, "cannot migrate %T", &entities.MessageThread{}))
+		container.logger.Fatal(stacktrace.Propagate(err, "cannot migrate message thread schema"))
+	}
+
+	if db.Migrator().HasColumn(&entities.MessageThread{}, "is_read") {
+		if err = db.
+			Model(&entities.MessageThread{}).
+			Where("is_read = ?", false).
+			Where("unread_count = ?", 0).
+			Update("unread_count", 1).
+			Error; err != nil {
+			container.logger.Fatal(stacktrace.Propagate(err, "cannot backfill message thread unread counts"))
+		}
+		if err = db.Migrator().DropColumn(&entities.MessageThread{}, "is_read"); err != nil {
+			container.logger.Fatal(stacktrace.Propagate(err, "cannot drop legacy message thread read state"))
+		}
+	}
+
+	if db.Migrator().HasColumn(&entities.MessageThread{}, "last_read_at") {
+		if err = db.Migrator().DropColumn(&entities.MessageThread{}, "last_read_at"); err != nil {
+			container.logger.Fatal(stacktrace.Propagate(err, "cannot drop message thread read watermark"))
+		}
+	}
+
+	if db.Migrator().HasTable("message_thread_unread_items") {
+		if err = db.Migrator().DropTable("message_thread_unread_items"); err != nil {
+			container.logger.Fatal(stacktrace.Propagate(err, "cannot drop message thread unread item ledger"))
+		}
+	}
+
+	if db.Migrator().HasTable("message_thread_deleted_items") {
+		if err = db.Migrator().DropTable("message_thread_deleted_items"); err != nil {
+			container.logger.Fatal(stacktrace.Propagate(err, "cannot drop deleted message item ledger"))
+		}
 	}
 
 	if err = db.AutoMigrate(&entities.User{}); err != nil {

api/pkg/entities/message_thread.go

@@ -9,12 +9,11 @@ import (
 // MessageThread represents a message thread between 2 phone numbers
 type MessageThread struct {
 	ID                 uuid.UUID     `json:"id" gorm:"primaryKey;type:uuid;" example:"32343a19-da5e-4b1b-a767-3298a73703ca"`
-	Owner              string        `json:"owner" example:"+18005550199"`
-	Contact            string        `json:"contact" example:"+18005550100"`
+	Owner              string        `json:"owner" gorm:"uniqueIndex:idx_message_threads_conversation" example:"+18005550199"`
+	Contact            string        `json:"contact" gorm:"uniqueIndex:idx_message_threads_conversation" example:"+18005550100"`
 	IsArchived         bool          `json:"is_archived" example:"false"`
-	IsRead             bool          `json:"is_read" gorm:"not null;default:true" example:"true"`
-	LastReadAt         time.Time     `json:"-" gorm:"not null;default:CURRENT_TIMESTAMP"`
-	UserID             UserID        `json:"user_id" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
+	UnreadCount        uint          `json:"unread_count" gorm:"not null;default:0" example:"2"`
+	UserID             UserID        `json:"user_id" gorm:"uniqueIndex:idx_message_threads_conversation" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
 	Color              string        `json:"color" example:"indigo"`
 	Status             MessageStatus `json:"status" example:"PENDING"`
 	LastMessageContent *string       `json:"last_message_content" example:"This is a sample message content"`

api/pkg/entities/message_thread_test.go

@@ -9,20 +9,20 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestMessageThreadReadFieldsHaveBackwardCompatibleDefaults(t *testing.T) {
+func TestMessageThreadUnreadFields(t *testing.T) {
 	threadType := reflect.TypeOf(MessageThread{})
 
-	isRead, ok := threadType.FieldByName("IsRead")
-	require.True(t, ok)
-	assert.Contains(t, isRead.Tag.Get("gorm"), "not null")
-	assert.Contains(t, isRead.Tag.Get("gorm"), "default:true")
-	assert.Equal(t, "is_read", isRead.Tag.Get("json"))
+	_, hasIsRead := threadType.FieldByName("IsRead")
+	assert.False(t, hasIsRead)
 
-	lastReadAt, ok := threadType.FieldByName("LastReadAt")
+	unreadCount, ok := threadType.FieldByName("UnreadCount")
 	require.True(t, ok)
-	assert.Contains(t, lastReadAt.Tag.Get("gorm"), "not null")
-	assert.Contains(t, lastReadAt.Tag.Get("gorm"), "default:CURRENT_TIMESTAMP")
-	assert.Equal(t, "-", lastReadAt.Tag.Get("json"))
+	assert.Equal(t, "unread_count", unreadCount.Tag.Get("json"))
+	assert.Contains(t, unreadCount.Tag.Get("gorm"), "not null")
+	assert.Contains(t, unreadCount.Tag.Get("gorm"), "default:0")
+
+	_, hasLastReadAt := threadType.FieldByName("LastReadAt")
+	assert.False(t, hasLastReadAt)
 }
 
 func TestMessageThreadContactDetailsAreTransientAndOmittedWhenNil(t *testing.T) {
@@ -41,3 +41,13 @@ func TestMessageThreadContactDetailsAreTransientAndOmittedWhenNil(t *testing.T)
 	require.NoError(t, json.Unmarshal(data, &payload))
 	assert.NotContains(t, payload, "contact_details")
 }
+
+func TestMessageThreadConversationHasCompositeUniqueIndex(t *testing.T) {
+	threadType := reflect.TypeOf(MessageThread{})
+
+	for _, fieldName := range []string{"UserID", "Owner", "Contact"} {
+		field, ok := threadType.FieldByName(fieldName)
+		require.True(t, ok)
+		assert.Contains(t, field.Tag.Get("gorm"), "uniqueIndex:idx_message_threads_conversation")
+	}
+}

api/pkg/handlers/message_thread_handler_test.go

@@ -25,7 +25,7 @@ import (
 
 type messageThreadHandlerRepositoryStub struct{}
 
-func (stub *messageThreadHandlerRepositoryStub) Store(context.Context, *entities.MessageThread) error {
+func (stub *messageThreadHandlerRepositoryStub) Store(context.Context, repositories.MessageThreadStoreParams) error {
 	return nil
 }
 
@@ -75,7 +75,7 @@ func TestMessageThreadHandlerUpdate_ReturnsNotFoundWhenThreadIsMissing(t *testin
 	handler.RegisterRoutes(app)
 
 	messageThreadID := uuid.New()
-	req := httptest.NewRequest(http.MethodPut, "/v1/message-threads/"+messageThreadID.String(), bytes.NewBufferString(`{"is_read":true}`))
+	req := httptest.NewRequest(http.MethodPut, "/v1/message-threads/"+messageThreadID.String(), bytes.NewBufferString(`{"unread_count":0}`))
 	req.Header.Set("Content-Type", "application/json")
 
 	resp, err := app.Test(req, fiber.TestConfig{Timeout: time.Second})
@@ -90,6 +90,36 @@ func TestMessageThreadHandlerUpdate_ReturnsNotFoundWhenThreadIsMissing(t *testin
 	require.Equal(t, "cannot find message thread with ID ["+messageThreadID.String()+"]", payload.Message)
 }
 
+func TestMessageThreadHandlerUpdate_RejectsLegacyIsReadPayload(t *testing.T) {
+	logger := &messageThreadHandlerNoopLogger{}
+	tracer := telemetry.NewOtelLogger("test", logger)
+	service := services.NewMessageThreadService(logger, tracer, &messageThreadHandlerRepositoryStub{}, nil, nil, nil)
+	handler := NewMessageThreadHandler(logger, tracer, validators.NewMessageThreadHandlerValidator(logger, tracer), service)
+
+	app := fiber.New()
+	app.Use(func(c fiber.Ctx) error {
+		c.Locals(middlewares.ContextKeyAuthUserID, entities.AuthContext{ID: entities.UserID("user-id"), Email: "user@example.com"})
+		return c.Next()
+	})
+	handler.RegisterRoutes(app)
+
+	req := httptest.NewRequest(http.MethodPut, "/v1/message-threads/"+uuid.NewString(), bytes.NewBufferString(`{"is_read":true}`))
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := app.Test(req, fiber.TestConfig{Timeout: time.Second})
+
+	require.NoError(t, err)
+	require.Equal(t, http.StatusUnprocessableEntity, resp.StatusCode)
+
+	var payload struct {
+		Message string              `json:"message"`
+		Data    map[string][]string `json:"data"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&payload))
+	require.Equal(t, "validation errors while updating message thread", payload.Message)
+	require.Equal(t, []string{"at least one of is_archived or unread_count is required"}, payload.Data["payload"])
+}
+
 type messageThreadHandlerNoopLogger struct{}
 
 var _ telemetry.Logger = (*messageThreadHandlerNoopLogger)(nil)

api/pkg/listeners/message_thread_listener.go

@@ -210,15 +210,14 @@ func (listener *MessageThreadListener) OnMessagePhoneReceived(ctx context.Contex
 	}
 
 	updateParams := services.MessageThreadUpdateParams{
-		Owner:          payload.Owner,
-		Contact:        payload.Contact,
-		Timestamp:      payload.Timestamp,
-		UserID:         payload.UserID,
-		Status:         entities.MessageStatusReceived,
-		Content:        payload.Content,
-		MessageID:      payload.MessageID,
-		MarkAsUnread:   true,
-		EventTimestamp: event.Time(),
+		Owner:         payload.Owner,
+		Contact:       payload.Contact,
+		Timestamp:     payload.Timestamp,
+		UserID:        payload.UserID,
+		Status:        entities.MessageStatusReceived,
+		Content:       payload.Content,
+		MessageID:     payload.MessageID,
+		CountAsUnread: true,
 	}
 
 	if err := listener.service.UpdateThread(ctx, updateParams); err != nil {
@@ -239,15 +238,14 @@ func (listener *MessageThreadListener) OnMessageCallMissed(ctx context.Context,
 	}
 
 	params := services.MessageThreadUpdateParams{
-		Owner:          payload.Owner,
-		Contact:        payload.Contact,
-		UserID:         payload.UserID,
-		Status:         entities.MessageStatusReceived,
-		Timestamp:      payload.Timestamp,
-		Content:        "Missed phone call",
-		MessageID:      payload.MessageID,
-		MarkAsUnread:   true,
-		EventTimestamp: event.Time(),
+		Owner:         payload.Owner,
+		Contact:       payload.Contact,
+		UserID:        payload.UserID,
+		Status:        entities.MessageStatusReceived,
+		Timestamp:     payload.Timestamp,
+		Content:       "Missed phone call",
+		MessageID:     payload.MessageID,
+		CountAsUnread: true,
 	}
 	if err := listener.service.UpdateThread(ctx, params); err != nil {
 		return listener.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot update thread for missed call [%s] on event [%s]", payload.MessageID, event.ID()))