feat(chat): unread read-receipts with lobby/game dot and history-open ack
CI / changes (pull_request) Successful in 2s
CI / unit (pull_request) Successful in 9s
CI / integration (pull_request) Successful in 15s
CI / ui (pull_request) Successful in 49s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m16s

Persist per-message read state as a chat_messages.unread_seats bitmask
(migration 00008): a text message seeds every recipient seat's bit, a nudge
only the awaited seat's. A seat's bit clears when the player opens the move
history or chat (POST /games/:id/chat/read, sent only when something is
unread), and a nudge additionally clears when its recipient answers by moving
(a wired game NudgeClearer, dependency-inverted so game keeps off social).

UI shows a per-viewer unread dot in the lobby (next to the opponent) and the
game score bar — the unread_chat game-view flag seeds it from authoritative
REST views, live chat/nudge events raise it. Opening the move history counts
as reading (even without entering chat): the 💬 fade-blinks twice and the
client acks. Admin Messages gains an unread-only filter, a read/unread column,
and a per-message card with the per-seat read breakdown. Observability:
chat_read_duration histogram + chat_unread_messages gauge + social tracing.
This commit is contained in:
Ilia Denisov
2026-06-17 11:12:38 +02:00
parent d53ff18a67
commit aaac816dc2
51 changed files with 1000 additions and 111 deletions
+92 -7
View File
@@ -2,6 +2,8 @@ package social
import (
"context"
"database/sql"
"errors"
"fmt"
"time"
@@ -23,16 +25,21 @@ type AdminMessage struct {
Body string
SenderIP string
CreatedAt time.Time
// UnreadSeats is the message's unread bitmask; non-zero means at least one recipient
// has not read it. The console shows a plain read/unread flag derived from it.
UnreadSeats int16
}
// AdminMessageFilter narrows the admin message list. A nil GameID/SenderID leaves that
// field unfiltered; NameMask/ExtMask are glob masks (account.LikePattern) matched
// case-insensitively against the sender's display name / any identity's external id.
// case-insensitively against the sender's display name / any identity's external id;
// UnreadOnly keeps only messages still unread by at least one recipient.
type AdminMessageFilter struct {
GameID uuid.UUID
SenderID uuid.UUID
NameMask string
ExtMask string
GameID uuid.UUID
SenderID uuid.UUID
NameMask string
ExtMask string
UnreadOnly bool
}
// AdminListMessages returns the filtered chat messages — real messages only, nudges
@@ -75,12 +82,15 @@ func adminMessageWhere(f AdminMessageFilter) (string, []any) {
args = append(args, ext)
where += fmt.Sprintf(` AND EXISTS (SELECT 1 FROM backend.identities ie WHERE ie.account_id = a.account_id AND ie.external_id ILIKE $%d ESCAPE '\')`, len(args))
}
if f.UnreadOnly {
where += ` AND m.unread_seats <> 0`
}
return where, args
}
func (s *Store) adminListMessages(ctx context.Context, f AdminMessageFilter, limit, offset int) ([]AdminMessage, error) {
where, args := adminMessageWhere(f)
q := `SELECT m.message_id, m.game_id, m.sender_id, a.display_name, ` + adminMessageSource + ` AS source, m.body, COALESCE(m.sender_ip, ''), m.created_at
q := `SELECT m.message_id, m.game_id, m.sender_id, a.display_name, ` + adminMessageSource + ` AS source, m.body, COALESCE(m.sender_ip, ''), m.created_at, m.unread_seats
FROM backend.chat_messages m
JOIN backend.accounts a ON a.account_id = m.sender_id
WHERE ` + where +
@@ -94,7 +104,7 @@ WHERE ` + where +
var out []AdminMessage
for rows.Next() {
var m AdminMessage
if err := rows.Scan(&m.ID, &m.GameID, &m.SenderID, &m.SenderName, &m.Source, &m.Body, &m.SenderIP, &m.CreatedAt); err != nil {
if err := rows.Scan(&m.ID, &m.GameID, &m.SenderID, &m.SenderName, &m.Source, &m.Body, &m.SenderIP, &m.CreatedAt, &m.UnreadSeats); err != nil {
return nil, fmt.Errorf("social: scan admin message: %w", err)
}
out = append(out, m)
@@ -111,3 +121,78 @@ func (s *Store) adminCountMessages(ctx context.Context, f AdminMessageFilter) (i
}
return n, nil
}
// AdminSeatRead is one game seat's status for a message's per-seat read breakdown: the
// seat index, its occupant's display name, and its role for the message — "sender",
// "read" or "unread". Still-empty (open) seats are omitted.
type AdminSeatRead struct {
Seat int
AccountID uuid.UUID
DisplayName string
Role string
}
// AdminMessageCard is one chat message with the per-seat read breakdown, for the
// console's message detail page.
type AdminMessageCard struct {
AdminMessage
Kind string
Seats []AdminSeatRead
}
// AdminMessageDetail loads one chat message and the read status of every seated player
// in its game, for the moderation message card. The per-seat status is derived from the
// message's unread bitmask (a set bit is an unread recipient seat); the sender's own
// seat is marked "sender".
func (svc *Service) AdminMessageDetail(ctx context.Context, messageID uuid.UUID) (AdminMessageCard, error) {
return svc.store.adminMessageCard(ctx, messageID)
}
func (s *Store) adminMessageCard(ctx context.Context, messageID uuid.UUID) (AdminMessageCard, error) {
var d AdminMessageCard
q := `SELECT m.game_id, m.sender_id, a.display_name, ` + adminMessageSource + ` AS source, m.kind, m.body, COALESCE(m.sender_ip, ''), m.created_at, m.unread_seats
FROM backend.chat_messages m
JOIN backend.accounts a ON a.account_id = m.sender_id
WHERE m.message_id = $1`
if err := s.db.QueryRowContext(ctx, q, messageID).Scan(
&d.GameID, &d.SenderID, &d.SenderName, &d.Source, &d.Kind, &d.Body, &d.SenderIP, &d.CreatedAt, &d.UnreadSeats,
); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return AdminMessageCard{}, ErrMessageNotFound
}
return AdminMessageCard{}, fmt.Errorf("social: admin message card: %w", err)
}
d.ID = messageID
rows, err := s.db.QueryContext(ctx,
`SELECT p.seat, COALESCE(p.account_id::text, ''), p.display_name
FROM backend.game_players p WHERE p.game_id = $1 ORDER BY p.seat`, d.GameID)
if err != nil {
return AdminMessageCard{}, fmt.Errorf("social: admin message seats: %w", err)
}
defer rows.Close()
for rows.Next() {
var (
seat int
accStr string
seatNam string
)
if err := rows.Scan(&seat, &accStr, &seatNam); err != nil {
return AdminMessageCard{}, fmt.Errorf("social: scan message seat: %w", err)
}
if accStr == "" {
continue // a still-empty (open) seat has no reader
}
accID, _ := uuid.Parse(accStr)
sr := AdminSeatRead{Seat: seat, AccountID: accID, DisplayName: seatNam}
switch {
case accID == d.SenderID:
sr.Role = "sender"
case d.UnreadSeats&(int16(1)<<uint(seat)) != 0:
sr.Role = "unread"
default:
sr.Role = "read"
}
d.Seats = append(d.Seats, sr)
}
return d, rows.Err()
}
+20 -5
View File
@@ -100,7 +100,7 @@ func (svc *Service) PostMessage(ctx context.Context, gameID, senderID uuid.UUID,
if err := Clean(body); err != nil {
return Message{}, err
}
msg, err := svc.store.insertChatMessage(ctx, gameID, senderID, kindMessage, body, parseIP(senderIP))
msg, err := svc.store.insertChatMessage(ctx, gameID, senderID, kindMessage, body, parseIP(senderIP), recipientMask(seats, senderID))
if err != nil {
return Message{}, err
}
@@ -148,7 +148,7 @@ func (svc *Service) Nudge(ctx context.Context, gameID, senderID uuid.UUID) (Mess
return Message{}, ErrNudgeTooSoon
}
}
msg, err := svc.store.insertChatMessage(ctx, gameID, senderID, kindNudge, "", nil)
msg, err := svc.store.insertChatMessage(ctx, gameID, senderID, kindNudge, "", nil, int16(1)<<uint(toMove))
if err != nil {
return Message{}, err
}
@@ -258,8 +258,22 @@ func parseIP(raw string) *string {
return &canon
}
// insertChatMessage stores one chat row and returns it.
func (s *Store) insertChatMessage(ctx context.Context, gameID, senderID uuid.UUID, kind, body string, ip *string) (Message, error) {
// recipientMask is the initial unread bitmask for a text message: a set bit for every
// seated recipient (by seat index) other than the sender. Still-empty (open) seats are
// skipped. With at most four seats the result always fits the smallint column.
func recipientMask(seats []uuid.UUID, senderID uuid.UUID) int16 {
var mask int16
for i, id := range seats {
if id == uuid.Nil || id == senderID {
continue
}
mask |= int16(1) << uint(i)
}
return mask
}
// insertChatMessage stores one chat row, seeding its unread bitmask, and returns it.
func (s *Store) insertChatMessage(ctx context.Context, gameID, senderID uuid.UUID, kind, body string, ip *string, unreadMask int16) (Message, error) {
id, err := uuid.NewV7()
if err != nil {
return Message{}, fmt.Errorf("social: new message id: %w", err)
@@ -271,7 +285,8 @@ func (s *Store) insertChatMessage(ctx context.Context, gameID, senderID uuid.UUI
stmt := table.ChatMessages.INSERT(
table.ChatMessages.MessageID, table.ChatMessages.GameID, table.ChatMessages.SenderID,
table.ChatMessages.Kind, table.ChatMessages.Body, table.ChatMessages.SenderIP,
).VALUES(id, gameID, senderID, kind, body, ipVal).
table.ChatMessages.UnreadSeats,
).VALUES(id, gameID, senderID, kind, body, ipVal, unreadMask).
RETURNING(table.ChatMessages.AllColumns)
var row model.ChatMessages
if err := stmt.QueryContext(ctx, s.db, &row); err != nil {
+173
View File
@@ -0,0 +1,173 @@
package social
import (
"context"
"fmt"
"time"
"github.com/google/uuid"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
)
// MarkRead clears viewerID's unread bit on every chat entry of the game they have not
// yet read — the acknowledgement the client sends when the player opens the move
// history (or the chat). For each entry that flips to read it records the
// publish-to-read latency. It returns the number of entries marked, and is a harmless
// no-op (zero) when the viewer holds no seat or has nothing unread, so the caller may
// invoke it without first checking participation.
func (svc *Service) MarkRead(ctx context.Context, gameID, viewerID uuid.UUID) (int, error) {
ctx, span := svc.tracer.Start(ctx, "social.MarkRead",
trace.WithAttributes(attribute.String("game.id", gameID.String())))
defer span.End()
marks, err := svc.store.markRead(ctx, gameID, viewerID)
if err != nil {
span.RecordError(err)
return 0, err
}
now := svc.now()
for _, m := range marks {
svc.metrics.recordRead(ctx, m.kind, now.Sub(m.createdAt))
}
span.SetAttributes(attribute.Int("marked.count", len(marks)))
return len(marks), nil
}
// ClearNudges marks accountID's pending nudges in the game read once they have acted
// (taken their move): a nudge answered by moving stops counting as unread. It records
// the publish-to-read latency for each nudge it clears. It satisfies game.NudgeClearer
// and is wired into the move path; failures are the caller's to log (the move has
// already committed).
func (svc *Service) ClearNudges(ctx context.Context, gameID, accountID uuid.UUID) error {
ctx, span := svc.tracer.Start(ctx, "social.ClearNudges",
trace.WithAttributes(attribute.String("game.id", gameID.String())))
defer span.End()
times, err := svc.store.clearNudges(ctx, gameID, accountID)
if err != nil {
span.RecordError(err)
return err
}
now := svc.now()
for _, t := range times {
svc.metrics.recordRead(ctx, kindNudge, now.Sub(t))
}
span.SetAttributes(attribute.Int("marked.count", len(times)))
return nil
}
// UnreadGames returns the set of games in which viewerID has at least one unread chat
// entry, for seeding the lobby's per-card unread badge in a single query.
func (svc *Service) UnreadGames(ctx context.Context, viewerID uuid.UUID) (map[uuid.UUID]bool, error) {
return svc.store.unreadGames(ctx, viewerID)
}
// HasUnread reports whether viewerID has any unread chat entry in the game, for the
// per-game unread flag of a single game's state and move-result views.
func (svc *Service) HasUnread(ctx context.Context, gameID, viewerID uuid.UUID) (bool, error) {
return svc.store.hasUnread(ctx, gameID, viewerID)
}
// readMark is one chat entry that flipped from unread to read, carrying what the
// publish-to-read latency metric needs.
type readMark struct {
kind string
createdAt time.Time
}
// markRead clears the viewer's seat bit on the game's entries still unread by them,
// resolving the seat through the game_players join, and returns the cleared entries.
// The bitwise terms are cast through int4 so the operators resolve unambiguously.
func (s *Store) markRead(ctx context.Context, gameID, viewerID uuid.UUID) ([]readMark, error) {
const q = `UPDATE backend.chat_messages m
SET unread_seats = (m.unread_seats::int & ~(1 << p.seat::int))::smallint
FROM backend.game_players p
WHERE p.game_id = m.game_id AND p.account_id = $2
AND m.game_id = $1 AND (m.unread_seats::int & (1 << p.seat::int)) <> 0
RETURNING m.kind, m.created_at`
rows, err := s.db.QueryContext(ctx, q, gameID, viewerID)
if err != nil {
return nil, fmt.Errorf("social: mark read: %w", err)
}
defer rows.Close()
var out []readMark
for rows.Next() {
var m readMark
if err := rows.Scan(&m.kind, &m.createdAt); err != nil {
return nil, fmt.Errorf("social: scan mark read: %w", err)
}
out = append(out, m)
}
return out, rows.Err()
}
// clearNudges clears the actor's seat bit on the game's still-unread nudges and returns
// their post times (for the latency metric).
func (s *Store) clearNudges(ctx context.Context, gameID, accountID uuid.UUID) ([]time.Time, error) {
const q = `UPDATE backend.chat_messages m
SET unread_seats = (m.unread_seats::int & ~(1 << p.seat::int))::smallint
FROM backend.game_players p
WHERE p.game_id = m.game_id AND p.account_id = $2
AND m.game_id = $1 AND m.kind = 'nudge' AND (m.unread_seats::int & (1 << p.seat::int)) <> 0
RETURNING m.created_at`
rows, err := s.db.QueryContext(ctx, q, gameID, accountID)
if err != nil {
return nil, fmt.Errorf("social: clear nudges: %w", err)
}
defer rows.Close()
var out []time.Time
for rows.Next() {
var t time.Time
if err := rows.Scan(&t); err != nil {
return nil, fmt.Errorf("social: scan clear nudges: %w", err)
}
out = append(out, t)
}
return out, rows.Err()
}
// unreadGames returns the games where viewerID has an unread entry, resolving their
// seat per game through the game_players join.
func (s *Store) unreadGames(ctx context.Context, viewerID uuid.UUID) (map[uuid.UUID]bool, error) {
const q = `SELECT DISTINCT m.game_id
FROM backend.chat_messages m
JOIN backend.game_players p ON p.game_id = m.game_id AND p.account_id = $1
WHERE m.unread_seats <> 0 AND (m.unread_seats::int & (1 << p.seat::int)) <> 0`
rows, err := s.db.QueryContext(ctx, q, viewerID)
if err != nil {
return nil, fmt.Errorf("social: unread games: %w", err)
}
defer rows.Close()
out := make(map[uuid.UUID]bool)
for rows.Next() {
var id uuid.UUID
if err := rows.Scan(&id); err != nil {
return nil, fmt.Errorf("social: scan unread games: %w", err)
}
out[id] = true
}
return out, rows.Err()
}
// hasUnread reports whether viewerID has an unread entry in the one game.
func (s *Store) hasUnread(ctx context.Context, gameID, viewerID uuid.UUID) (bool, error) {
const q = `SELECT EXISTS(
SELECT 1 FROM backend.chat_messages m
JOIN backend.game_players p ON p.game_id = m.game_id AND p.account_id = $2
WHERE m.game_id = $1 AND (m.unread_seats::int & (1 << p.seat::int)) <> 0)`
var ok bool
if err := s.db.QueryRowContext(ctx, q, gameID, viewerID).Scan(&ok); err != nil {
return false, fmt.Errorf("social: has unread: %w", err)
}
return ok, nil
}
// countUnread counts the chat entries with at least one recipient seat still unread,
// for the chat_unread_messages gauge.
func (s *Store) countUnread(ctx context.Context) (int64, error) {
var n int64
if err := s.db.QueryRowContext(ctx,
`SELECT count(*) FROM backend.chat_messages WHERE unread_seats <> 0`).Scan(&n); err != nil {
return 0, fmt.Errorf("social: count unread: %w", err)
}
return n, nil
}
+32 -2
View File
@@ -2,6 +2,7 @@ package social
import (
"context"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
@@ -15,7 +16,8 @@ const meterName = "scrabble/backend/social"
// no-ops (see defaultSocialMetrics); SetMetrics installs the real meter during
// startup wiring.
type socialMetrics struct {
messages metric.Int64Counter
messages metric.Int64Counter
readDuration metric.Float64Histogram
}
// defaultSocialMetrics returns instruments backed by a no-op meter.
@@ -31,7 +33,13 @@ func newSocialMetrics(meter metric.Meter) *socialMetrics {
if err != nil {
c, _ = noop.NewMeterProvider().Meter(meterName).Int64Counter("chat_messages_total")
}
return &socialMetrics{messages: c}
h, err := meter.Float64Histogram("chat_read_duration",
metric.WithDescription("Time from a chat entry being posted to a recipient reading it, in seconds, labelled by kind (message/nudge)."),
metric.WithUnit("s"))
if err != nil {
h, _ = noop.NewMeterProvider().Meter(meterName).Float64Histogram("chat_read_duration")
}
return &socialMetrics{messages: c, readDuration: h}
}
// SetMetrics installs the meter the social domain records to. It must be called
@@ -41,9 +49,31 @@ func (svc *Service) SetMetrics(meter metric.Meter) {
return
}
svc.metrics = newSocialMetrics(meter)
// chat_unread_messages observes the current backlog of unread chat entries at scrape
// time (the partial index keeps the count cheap). A registration error is ignored —
// the gauge is best-effort observability.
_, _ = meter.Int64ObservableGauge("chat_unread_messages",
metric.WithDescription("Chat entries with at least one recipient seat that has not read them yet."),
metric.WithInt64Callback(func(ctx context.Context, o metric.Int64Observer) error {
n, err := svc.store.countUnread(ctx)
if err != nil {
return err
}
o.Observe(n)
return nil
}))
}
// recordChat counts one posted chat entry of the given kind (message or nudge).
func (m *socialMetrics) recordChat(ctx context.Context, kind string) {
m.messages.Add(ctx, 1, metric.WithAttributes(attribute.String("kind", kind)))
}
// recordRead records one chat entry's publish-to-read latency, labelled by kind. A
// non-positive duration (clock skew) is dropped.
func (m *socialMetrics) recordRead(ctx context.Context, kind string, d time.Duration) {
if d <= 0 {
return
}
m.readDuration.Record(ctx, d.Seconds(), metric.WithAttributes(attribute.String("kind", kind)))
}
+9
View File
@@ -13,11 +13,16 @@ import (
"time"
"github.com/google/uuid"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/trace"
"scrabble/backend/internal/account"
"scrabble/backend/internal/notify"
)
// tracerName scopes the social domain's OpenTelemetry spans.
const tracerName = "scrabble/backend/social"
// GameReader is the slice of the game domain the social package needs: the seated
// accounts in seat order, the seat index whose turn it is, and the game status, plus
// a shared-game test. game.Service satisfies it, so chat, nudge and the
@@ -71,6 +76,8 @@ var (
ErrFriendCodeInvalid = errors.New("social: friend code is invalid or expired")
// ErrNotParticipant is returned when an account is not seated in the game.
ErrNotParticipant = errors.New("social: account is not a player in this game")
// ErrMessageNotFound is returned when an admin message lookup finds no such message.
ErrMessageNotFound = errors.New("social: message not found")
// ErrChatBlocked is returned when the sender has disabled chat for themselves.
ErrChatBlocked = errors.New("social: chat is disabled for this account")
// ErrMessageTooLong is returned when a chat message exceeds the rune limit.
@@ -104,6 +111,7 @@ type Service struct {
games GameReader
pub notify.Publisher
metrics *socialMetrics
tracer trace.Tracer
now func() time.Time
}
@@ -116,6 +124,7 @@ func NewService(store *Store, accounts *account.Store, games GameReader) *Servic
games: games,
pub: notify.Nop{},
metrics: defaultSocialMetrics(),
tracer: otel.Tracer(tracerName),
now: func() time.Time { return time.Now().UTC() },
}
}