feat: chat reactions

2026-08-30 01:51:32 +05:30
parent 9c458ba19f
commit cee2d3c874
8 changed files with 758 additions and 43 deletions
@@ -336,27 +336,58 @@ func (r *ChatRepo) Inbox(ctx context.Context, userID string, kinds []string, lim
 // Messages returns one page of history, newest first. beforeSeq == 0 starts at
 // the newest message; otherwise it is an exclusive upper bound, which makes
 // paging stable even as new messages arrive at the top.
+// messageSelectFrom is the canonical read shape for a message, including the
+// quoted-reply preview. Shared by the paged thread read and the single-message
+// read so the two can never drift into returning different shapes.
+const messageSelectFrom = `
+	SELECT m.id, m.conversation_id, m.seq, m.sender_id, m.kind, m.body, m.entities,
+	       m.client_msg_id, m.reply_to_id, m.forwarded_from_id, m.edited_at, m.deleted_at, m.created_at,
+	       rep.id, rep.seq, rep.sender_id, rep.body, rep.deleted_at, COALESCE(rep.sender_name, '')
+	FROM messages m
+	LEFT JOIN LATERAL (
+	    SELECT r.id, r.seq, r.sender_id, r.body, r.deleted_at, ru.name AS sender_name
+	    FROM messages r
+	    LEFT JOIN users ru ON ru.id = r.sender_id
+	    -- The conversation guard is repeated here on purpose: even a row
+	    -- written before the write-side check existed cannot leak another
+	    -- thread's text through a quote.
+	    WHERE r.id = m.reply_to_id AND r.conversation_id = m.conversation_id
+	) rep ON TRUE
+`
+
+// Message reads one message of a conversation, in the same shape as the paged
+// read. Hidden (deleted-for-me) messages are excluded, so a refresh can't
+// resurrect something the caller removed from their own view.
+func (r *ChatRepo) Message(ctx context.Context, conversationID, messageID, userID string) (*model.Message, error) {
+	msgs, err := r.scanMessages(ctx, messageSelectFrom+`
+		WHERE m.conversation_id = $1
+		  AND m.id = $2
+		  AND NOT EXISTS (SELECT 1 FROM message_hides h
+		                  WHERE h.message_id = m.id AND h.user_id = $3)`,
+		conversationID, messageID, userID)
+	if err != nil {
+		return nil, err
+	}
+	if len(msgs) == 0 {
+		return nil, ErrMessageNotHere
+	}
+	return &msgs[0], nil
+}
+
 func (r *ChatRepo) Messages(ctx context.Context, conversationID, userID string, beforeSeq int64, limit int) ([]model.Message, error) {
-	rows, err := r.pg.Pool.Query(ctx, `
-		SELECT m.id, m.conversation_id, m.seq, m.sender_id, m.kind, m.body, m.entities,
-		       m.client_msg_id, m.reply_to_id, m.forwarded_from_id, m.edited_at, m.deleted_at, m.created_at,
-		       rep.id, rep.seq, rep.sender_id, rep.body, rep.deleted_at, COALESCE(rep.sender_name, '')
-		FROM messages m
-		LEFT JOIN LATERAL (
-		    SELECT r.id, r.seq, r.sender_id, r.body, r.deleted_at, ru.name AS sender_name
-		    FROM messages r
-		    LEFT JOIN users ru ON ru.id = r.sender_id
-		    -- The conversation guard is repeated here on purpose: even a row
-		    -- written before the write-side check existed cannot leak another
-		    -- thread's text through a quote.
-		    WHERE r.id = m.reply_to_id AND r.conversation_id = m.conversation_id
-		) rep ON TRUE
+	return r.scanMessages(ctx, messageSelectFrom+`
 		WHERE m.conversation_id = $1
 		  AND ($2 = 0 OR m.seq < $2)
 		  AND NOT EXISTS (SELECT 1 FROM message_hides h
 		                  WHERE h.message_id = m.id AND h.user_id = $3)
 		ORDER BY m.seq DESC
 		LIMIT $4`, conversationID, beforeSeq, userID, limit)
+}
+
+// scanMessages runs a query built on messageSelectFrom and decodes its rows.
+// Shared so the paged read and the single-message read decode identically.
+func (r *ChatRepo) scanMessages(ctx context.Context, query string, args ...any) ([]model.Message, error) {
+	rows, err := r.pg.Pool.Query(ctx, query, args...)
 	if err != nil {
 		return nil, fmt.Errorf("query messages: %w", err)
 	}
@@ -1157,6 +1188,108 @@ func (r *ChatRepo) ReportMessage(ctx context.Context, reporterID, messageID, con
 	return nil
 }

+// SetReaction records one person's emoji on a message, replacing whatever
+// they chose before. Returns the message's conversation so the caller can
+// authorize and publish without a second query.
+func (r *ChatRepo) SetReaction(ctx context.Context, messageID, userID, emoji string) (conversationID string, err error) {
+	err = r.pg.Pool.QueryRow(ctx, `
+		INSERT INTO message_reactions (message_id, user_id, conversation_id, emoji)
+		SELECT $1, $2, m.conversation_id, $3
+		FROM messages m
+		WHERE m.id = $1 AND m.deleted_at IS NULL
+		ON CONFLICT (message_id, user_id)
+		DO UPDATE SET emoji = EXCLUDED.emoji, updated_at = now()
+		RETURNING conversation_id`, messageID, userID, emoji).Scan(&conversationID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		// No row inserted means no such message, or it was deleted — you
+		// cannot react to something that isn't there.
+		return "", ErrMessageNotHere
+	}
+	if err != nil {
+		return "", fmt.Errorf("set reaction: %w", err)
+	}
+	return conversationID, nil
+}
+
+// ClearReaction removes this user's reaction. Returns the conversation so the
+// caller can publish the change; ErrMessageNotHere when there was none.
+func (r *ChatRepo) ClearReaction(ctx context.Context, messageID, userID string) (conversationID string, err error) {
+	err = r.pg.Pool.QueryRow(ctx, `
+		DELETE FROM message_reactions
+		WHERE message_id = $1 AND user_id = $2
+		RETURNING conversation_id`, messageID, userID).Scan(&conversationID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return "", ErrMessageNotHere
+	}
+	if err != nil {
+		return "", fmt.Errorf("clear reaction: %w", err)
+	}
+	return conversationID, nil
+}
+
+// ReactionDetails lists everyone who reacted to one message, newest first.
+//
+// Separate from ReactionsFor: the grouped counts are what a thread renders,
+// and sending the full membership of every reaction on every page would grow
+// with the room. This is only read when someone opens the list.
+func (r *ChatRepo) ReactionDetails(ctx context.Context, messageID string) ([]model.ReactionDetail, error) {
+	rows, err := r.pg.Pool.Query(ctx, `
+		SELECT u.id, u.handle, u.name, u.avatar_url, mr.emoji, mr.updated_at
+		FROM message_reactions mr
+		JOIN users u ON u.id = mr.user_id
+		WHERE mr.message_id = $1
+		ORDER BY mr.updated_at DESC`, messageID)
+	if err != nil {
+		return nil, fmt.Errorf("reaction details: %w", err)
+	}
+	defer rows.Close()
+
+	out := []model.ReactionDetail{}
+	for rows.Next() {
+		var d model.ReactionDetail
+		if err := rows.Scan(&d.User.ID, &d.User.Handle, &d.User.Name,
+			&d.User.AvatarURL, &d.Emoji, &d.CreatedAt); err != nil {
+			return nil, fmt.Errorf("scan reaction detail: %w", err)
+		}
+		out = append(out, d)
+	}
+	return out, rows.Err()
+}
+
+// ReactionsFor groups reactions for a page of messages in one query.
+//
+// Grouped in SQL rather than in Go: the counts are what the UI draws, and
+// sending one row per person would grow with the room.
+func (r *ChatRepo) ReactionsFor(ctx context.Context, messageIDs []string, viewerID string) (map[string][]model.Reaction, error) {
+	out := map[string][]model.Reaction{}
+	if len(messageIDs) == 0 {
+		return out, nil
+	}
+	rows, err := r.pg.Pool.Query(ctx, `
+		SELECT message_id, emoji, count(*),
+		       bool_or(user_id = $2) AS mine
+		FROM message_reactions
+		WHERE message_id = ANY($1)
+		GROUP BY message_id, emoji
+		-- Most-used first, then stable by emoji so the chip order doesn't
+		-- shuffle between reloads.
+		ORDER BY message_id, count(*) DESC, emoji`, messageIDs, viewerID)
+	if err != nil {
+		return nil, fmt.Errorf("reactions for: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var messageID string
+		var re model.Reaction
+		if err := rows.Scan(&messageID, &re.Emoji, &re.Count, &re.Mine); err != nil {
+			return nil, fmt.Errorf("scan reaction: %w", err)
+		}
+		out[messageID] = append(out[messageID], re)
+	}
+	return out, rows.Err()
+}
+
 // PushCandidate is a participant plus the state that decides whether a new
 // message should reach them out-of-band.
 type PushCandidate struct {