feat(comms): per-user chat inbox category (priority / general)

api/comms/chat.go

@@ -250,6 +250,30 @@ func isInPermitList(permit ChatPermission, permitList []ChatPermission) bool {
 	return false
 }
 
+// chatSetCategory sets (or, when category is nil, clears) the calling user's
+// inbox category for a chat. Like updatePermissions/chatUnblock it is guarded
+// by the RPC timestamp so a late-arriving RPC can't clobber newer state.
+func chatSetCategory(db dbv1.DBTX, ctx context.Context, userId int32, chatId string, category *string, messageTimestamp time.Time) error {
+	var err error
+	if category != nil {
+		_, err = db.Exec(ctx, `
+		insert into user_conversation_preferences
+			(user_id, chat_id, category, updated_at)
+		values
+			($1, $2, $3, $4)
+		on conflict (user_id, chat_id)
+		do update set category = excluded.category, updated_at = excluded.updated_at
+		where user_conversation_preferences.updated_at < excluded.updated_at`,
+			userId, chatId, *category, messageTimestamp.UTC())
+	} else {
+		_, err = db.Exec(ctx, `
+		delete from user_conversation_preferences
+		where user_id = $1 and chat_id = $2 and updated_at < $3`,
+			userId, chatId, messageTimestamp.UTC())
+	}
+	return err
+}
+
 func updatePermissions(db dbv1.DBTX, ctx context.Context, userId int32, permit ChatPermission, permitAllowed bool, messageTimestamp time.Time) error {
 	_, err := db.Exec(ctx, `
     insert into chat_permissions (user_id, permits, allowed, updated_at)

api/comms/chat_set_category_test.go

@@ -0,0 +1,127 @@
+package comms
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"strconv"
+	"testing"
+	"time"
+
+	"api.audius.co/database"
+	"api.audius.co/trashid"
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestChatSetCategory(t *testing.T) {
+	pool := database.CreateTestDatabase(t, "test_comms")
+	defer pool.Close()
+
+	ctx := context.Background()
+
+	chatId := trashid.ChatID(1, 2)
+
+	seededRand := rand.New(rand.NewSource(time.Now().UnixNano()))
+	inviteCode1 := strconv.Itoa(seededRand.Int())
+	inviteCode2 := strconv.Itoa(seededRand.Int())
+
+	SetupChatWithMembers(t, pool, ctx, chatId, 1, 2, inviteCode1, inviteCode2)
+
+	// getCategory returns the stored category for (userId, chatId), or nil when
+	// there is no row (uncategorized).
+	getCategory := func(userId int32) *string {
+		var category string
+		err := pool.QueryRow(ctx,
+			"select category from user_conversation_preferences where user_id = $1 and chat_id = $2",
+			userId, chatId).Scan(&category)
+		if err == pgx.ErrNoRows {
+			return nil
+		}
+		require.NoError(t, err)
+		return &category
+	}
+
+	validator := CreateTestValidator(t, pool, DefaultRateLimitConfig, DefaultTestValidatorConfig)
+
+	// Validation
+	{
+		priorityRpc := RawRPC{
+			Params: []byte(fmt.Sprintf(`{"chat_id": "%s", "category": "priority"}`, chatId)),
+		}
+		generalRpc := RawRPC{
+			Params: []byte(fmt.Sprintf(`{"chat_id": "%s", "category": "general"}`, chatId)),
+		}
+		nullRpc := RawRPC{
+			Params: []byte(fmt.Sprintf(`{"chat_id": "%s", "category": null}`, chatId)),
+		}
+		bogusRpc := RawRPC{
+			Params: []byte(fmt.Sprintf(`{"chat_id": "%s", "category": "spam"}`, chatId)),
+		}
+
+		// members may set any valid category or clear it
+		assert.NoError(t, validator.validateChatSetCategory(1, priorityRpc))
+		assert.NoError(t, validator.validateChatSetCategory(2, generalRpc))
+		assert.NoError(t, validator.validateChatSetCategory(1, nullRpc))
+
+		// non-members are rejected
+		err := validator.validateChatSetCategory(3, priorityRpc)
+		assert.Error(t, err, "User 3 is not a member and should not be able to set a category")
+		assert.Contains(t, err.Error(), "user is not a member of this chat")
+
+		// unknown category values are rejected, even for members
+		err = validator.validateChatSetCategory(1, bogusRpc)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "invalid chat category")
+
+		// the full Validate entrypoint routes chat.set_category to the validator
+		err = validator.Validate(ctx, 3, RawRPC{Method: string(RPCMethodChatSetCategory), Params: priorityRpc.Params})
+		assert.Error(t, err)
+	}
+
+	general := string(ChatCategoryGeneral)
+	priority := string(ChatCategoryPriority)
+
+	// no row to start with
+	assert.Nil(t, getCategory(1))
+
+	// set "general"
+	t1 := time.Now().UTC().Add(-time.Minute)
+	require.NoError(t, chatSetCategory(pool, ctx, 1, chatId, &general, t1))
+	if got := getCategory(1); assert.NotNil(t, got) {
+		assert.Equal(t, general, *got)
+	}
+
+	// update to "priority" with a newer timestamp
+	t2 := t1.Add(10 * time.Second)
+	require.NoError(t, chatSetCategory(pool, ctx, 1, chatId, &priority, t2))
+	if got := getCategory(1); assert.NotNil(t, got) {
+		assert.Equal(t, priority, *got)
+	}
+
+	// a late-arriving older RPC must not clobber newer state
+	tOld := t1.Add(5 * time.Second)
+	require.NoError(t, chatSetCategory(pool, ctx, 1, chatId, &general, tOld))
+	if got := getCategory(1); assert.NotNil(t, got) {
+		assert.Equal(t, priority, *got, "older set_category should be ignored")
+	}
+
+	// the other member's preference is independent
+	assert.Nil(t, getCategory(2))
+
+	// clearing with an older timestamp must not delete the newer row
+	require.NoError(t, chatSetCategory(pool, ctx, 1, chatId, nil, tOld))
+	if got := getCategory(1); assert.NotNil(t, got) {
+		assert.Equal(t, priority, *got, "older clear should be ignored")
+	}
+
+	// clearing with a newer timestamp deletes the row
+	t3 := t2.Add(10 * time.Second)
+	require.NoError(t, chatSetCategory(pool, ctx, 1, chatId, nil, t3))
+	assert.Nil(t, getCategory(1))
+
+	// clearing when there is no row is a no-op, not an error
+	require.NoError(t, chatSetCategory(pool, ctx, 1, chatId, nil, t3.Add(time.Second)))
+	assert.Nil(t, getCategory(1))
+}

api/comms/rpc_processor.go

@@ -277,6 +277,16 @@ select last_active_at from chat_member where chat_id = $1 and user_id = $2`
 			if err != nil {
 				return err
 			}
+		case RPCMethodChatSetCategory:
+			var params ChatSetCategoryRPCParams
+			err = json.Unmarshal(rawRpc.Params, &params)
+			if err != nil {
+				return err
+			}
+			err = chatSetCategory(tx, ctx, userId, params.ChatID, params.Category, messageTs)
+			if err != nil {
+				return err
+			}
 		case RPCMethodChatPermit:
 			var params ChatPermitRPCParams
 			err = json.Unmarshal(rawRpc.Params, &params)

api/comms/schema.go

@@ -102,6 +102,18 @@ type ChatReadAllRPC struct {
 
 type ChatReadAllRPCParams struct{}
 
+type ChatSetCategoryRPC struct {
+	Method ChatSetCategoryRPCMethod `json:"method"`
+	Params ChatSetCategoryRPCParams `json:"params"`
+}
+
+// ChatSetCategoryRPCParams sets the calling user's inbox category for a chat.
+// Category is "priority" or "general"; null clears it (chat becomes uncategorized).
+type ChatSetCategoryRPCParams struct {
+	ChatID   string  `json:"chat_id"`
+	Category *string `json:"category"`
+}
+
 type ChatBlockRPC struct {
 	Method ChatBlockRPCMethod `json:"method"`
 	Params ChatBlockRPCParams `json:"params"`
@@ -149,6 +161,7 @@ type RPCPayloadRequestParams struct {
 	MessageID           *string              `json:"message_id,omitempty"`
 	ParentMessageID     *string              `json:"parent_message_id,omitempty"`
 	Reaction            *string              `json:"reaction"`
+	Category            *string              `json:"category"`
 	UserID              *string              `json:"user_id,omitempty"`
 	Allow               *bool                `json:"allow,omitempty"`
 	Permit              *ChatPermission      `json:"permit,omitempty"`
@@ -164,6 +177,7 @@ type UserChat struct {
 	Audience               ChatBlastAudience `json:"audience"`
 	AudienceContentID      *string           `json:"audience_content_id,omitempty"`
 	AudienceContentType    *string           `json:"audience_content_type,omitempty"`
+	Category               *string           `json:"category"`
 	ChatID                 string            `json:"chat_id"`
 	ChatMembers            []ChatMember      `json:"chat_members"`
 	ClearedHistoryAt       string            `json:"cleared_history_at"`
@@ -298,6 +312,7 @@ type RPCPayloadParams struct {
 	MessageID           *string              `json:"message_id,omitempty"`
 	ParentMessageID     *string              `json:"parent_message_id,omitempty"`
 	Reaction            *string              `json:"reaction"`
+	Category            *string              `json:"category"`
 	UserID              *string              `json:"user_id,omitempty"`
 	Allow               *bool                `json:"allow,omitempty"`
 	Permit              *ChatPermission      `json:"permit,omitempty"`
@@ -380,6 +395,12 @@ const (
 	MethodChatReadAll ChatReadAllRPCMethod = "chat.read_all"
 )
 
+type ChatSetCategoryRPCMethod string
+
+const (
+	MethodChatSetCategory ChatSetCategoryRPCMethod = "chat.set_category"
+)
+
 type ChatBlockRPCMethod string
 
 const (
@@ -411,6 +432,14 @@ const (
 	ChatPermissionVerified  ChatPermission = "verified"
 )
 
+// Per-user inbox category for a chat. A chat with no category is "uncategorized".
+type ChatCategory string
+
+const (
+	ChatCategoryPriority ChatCategory = "priority"
+	ChatCategoryGeneral  ChatCategory = "general"
+)
+
 type RPCMethod string
 
 const (
@@ -424,6 +453,7 @@ const (
 	RPCMethodChatReact           RPCMethod = "chat.react"
 	RPCMethodChatRead            RPCMethod = "chat.read"
 	RPCMethodChatReadAll         RPCMethod = "chat.read_all"
+	RPCMethodChatSetCategory     RPCMethod = "chat.set_category"
 	RPCMethodChatUnblock         RPCMethod = "chat.unblock"
 	RPCMethodUserValidateCanChat RPCMethod = "user.validate_can_chat"
 )

api/comms/validator.go

@@ -70,6 +70,8 @@ func (vtor *Validator) Validate(ctx context.Context, userId int32, rawRpc RawRPC
 	case RPCMethodChatReadAll:
 		// No params to validate; ban check above already gates this call.
 		return nil
+	case RPCMethodChatSetCategory:
+		return vtor.validateChatSetCategory(userId, rawRpc)
 	case RPCMethodChatPermit:
 		return vtor.validateChatPermit(userId, rawRpc)
 	case RPCMethodChatBlock:
@@ -250,6 +252,32 @@ func (vtor *Validator) validateChatRead(userId int32, rpc RawRPC) error {
 	return nil
 }
 
+func (vtor *Validator) validateChatSetCategory(userId int32, rpc RawRPC) error {
+	// validate rpc.params valid
+	var params ChatSetCategoryRPCParams
+	err := json.Unmarshal(rpc.Params, &params)
+	if err != nil {
+		return err
+	}
+
+	// validate category is one of the known values (nil clears the category)
+	if params.Category != nil {
+		switch ChatCategory(*params.Category) {
+		case ChatCategoryPriority, ChatCategoryGeneral:
+		default:
+			return fmt.Errorf("invalid chat category %q: must be %q, %q, or null", *params.Category, ChatCategoryPriority, ChatCategoryGeneral)
+		}
+	}
+
+	// validate userId is a member of chatId in good standing
+	err = validateChatMembership(vtor.pool, context.Background(), userId, params.ChatID)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (vtor *Validator) validateChatPermit(userId int32, rpc RawRPC) error {
 	// validate rpc.params valid
 	var params ChatPermitRPCParams

api/comms_chat.go

@@ -28,14 +28,18 @@ func (app *ApiServer) getChat(c *fiber.Ctx) error {
 	NULL AS audience_content_id,
 	(
 		SELECT json_agg(json_build_object(
-			'user_id', chat_member.user_id,
-			'cleared_history_at', chat_member.cleared_history_at
+			'user_id', m.user_id,
+			'cleared_history_at', m.cleared_history_at
 		))
-		FROM chat_member
-		WHERE chat_member.chat_id = chat.chat_id
-	)::jsonb AS members
+		FROM chat_member m
+		WHERE m.chat_id = chat.chat_id
+	)::jsonb AS members,
+	ucp.category::text AS category
 	FROM chat_member
 	JOIN chat ON chat.chat_id = chat_member.chat_id
+	LEFT JOIN user_conversation_preferences ucp
+		ON ucp.user_id = chat_member.user_id
+		AND ucp.chat_id = chat_member.chat_id
 	WHERE chat_member.user_id = @user_id AND chat_member.chat_id = @chat_id
 
 	-- Query blasts as well
@@ -58,7 +62,8 @@ func (app *ApiServer) getChat(c *fiber.Ctx) error {
 			audience,
 			audience_content_type,
 			audience_content_id,
-			json_build_array()::jsonb AS members
+			json_build_array()::jsonb AS members,
+			NULL::text AS category
 		FROM chat_blast b
 		WHERE from_user_id = @user_id
 			AND concat_ws(':', audience, audience_content_type,

api/comms_chats.go

@@ -35,14 +35,18 @@ func (app *ApiServer) getChats(c *fiber.Ctx) error {
 		NULL AS audience_content_id,
 		(
 			SELECT json_agg(json_build_object(
-				'user_id', chat_member.user_id,
-				'cleared_history_at', chat_member.cleared_history_at
+				'user_id', m.user_id,
+				'cleared_history_at', m.cleared_history_at
 			))
-			FROM chat_member
-			WHERE chat_member.chat_id = chat.chat_id
-		)::jsonb AS members
+			FROM chat_member m
+			WHERE m.chat_id = chat.chat_id
+		)::jsonb AS members,
+		ucp.category::text AS category
 	FROM chat_member
 	JOIN chat ON chat.chat_id = chat_member.chat_id
+	LEFT JOIN user_conversation_preferences ucp
+		ON ucp.user_id = chat_member.user_id
+		AND ucp.chat_id = chat_member.chat_id
 	WHERE chat_member.user_id = @user_id
 	AND chat_member.is_hidden = false
 	AND chat.last_message IS NOT NULL
@@ -71,7 +75,8 @@ func (app *ApiServer) getChats(c *fiber.Ctx) error {
 			audience,
 			audience_content_type,
 			audience_content_id,
-			json_build_array()::jsonb AS members
+			json_build_array()::jsonb AS members,
+			NULL::text AS category
 		FROM chat_blast
 		WHERE from_user_id = @user_id
 			AND created_at < @before