scrabble-game/backend/internal/notify/notify.go

// Package notify is the backend's in-process live-event seam. Domain services
// publish Intents after a successful commit; the gRPC push server (internal
// /pushgrpc) subscribes to the hub and streams them to the gateway, which fans
// them out to clients (docs/ARCHITECTURE.md §10). Event payloads are
// FlatBuffers-encoded by the typed constructors in events.go, so the domain
// services stay free of the wire schema and only depend on this package.
//
// Publishing is best-effort and non-blocking: a live event is a convenience, not
// a correctness requirement, so a slow or absent subscriber never blocks a game
// transition. The default Publisher is Nop, which keeps every domain service (and
// its tests) runnable without a live channel.
package notify

import (
	"sync"

	"github.com/google/uuid"
)

// Notification kinds — the catalog in docs/ARCHITECTURE.md §10.
const (
	KindYourTurn      = "your_turn"
	KindOpponentMoved = "opponent_moved"
	KindChatMessage   = "chat_message"
	KindNudge         = "nudge"
	KindMatchFound    = "match_found"
	// KindOpponentJoined tells the starter of an auto-match game still "searching for an
	// opponent" that the empty seat has been taken (by a human or a substituted robot),
	// carrying the refreshed StateView so the client fills the opponent card and
	// re-enables resign and chat in place. In-app only (never an out-of-app push).
	KindOpponentJoined = "opponent_joined"
	// KindNotification is a lightweight "re-poll your lobby counters" signal
	// (incoming friend requests, invitations) that drives the lobby badge.
	KindNotification = "notify"
	// KindGameOver announces a finished game to each seated player, driving the
	// out-of-app "game over" push.
	KindGameOver = "game_over"
	// KindChatAccessChanged signals that a player's eligibility to write in the
	// moderated Telegram discussion chat may have changed (an admin block or unblock,
	// a chat_muted grant or revoke, or a temporary block lapsing). It carries no
	// payload and is never fanned out to in-app clients: the gateway consumes it to
	// resolve the player's Telegram identity and current eligibility and push the
	// resulting chat-gate command to the bot.
	KindChatAccessChanged = "chat_access_changed"
)

// Notification sub-kinds carried in a KindNotification event payload; the client
// re-fetches its lobby counters on any of them.
const (
	NotifyFriendRequest = "friend_request"
	NotifyFriendAdded   = "friend_added"
	// NotifyFriendDeclined tells the original requester their request was declined, so a
	// game screen watching that opponent re-derives its "add to friends" state.
	NotifyFriendDeclined = "friend_declined"
	NotifyInvitation     = "invitation"
	// NotifyInvitationUpdate carries a changed invitation — an updated invitee response, or a
	// terminal status (started, declined, cancelled, expired) — so the client patches its lobby
	// invitations list without a refetch. Unlike NotifyInvitation (a brand-new invitation), it is
	// in-app only: the Telegram connector renders no message for it, so a withdrawal or decline never
	// becomes an out-of-app push.
	NotifyInvitationUpdate = "invitation_update"
	NotifyGameStarted      = "game_started"
	// NotifyAdminReply tells the player an operator has answered their feedback, so
	// the client raises the Settings/Info badge and re-fetches the reply. It carries
	// no payload (the reply is fetched on the feedback screen). In-app only.
	NotifyAdminReply = "admin_reply"
	// NotifyBanner tells the client that the viewer's advertising-banner eligibility
	// may have changed (an operator granted hints or toggled the no_banner role), so
	// it re-fetches profile.get to show or hide the banner. It carries no payload (the
	// banner set rides the profile response). In-app only.
	NotifyBanner = "banner"
	// NotifyProfile tells the client that the viewer's own profile changed out of band
	// (e.g. an email was confirmed via the one-tap deeplink opened in another browser),
	// so it re-fetches profile.get. It carries no payload. In-app only.
	NotifyProfile = "profile"
	// NotifyUserBlocked confirms to the blocker that a per-user block took effect,
	// carrying the blocked account, so every one of the blocker's sessions updates the
	// in-game block/add-friend controls and the struck name in place. It is delivered
	// only to the blocker — never to the blocked user, who must not learn of the block —
	// and is in-app only (no out-of-app push: blocking is the viewer's own action).
	NotifyUserBlocked = "user_blocked"
	// NotifyUserUnblocked confirms an unblock to the (former) blocker, carrying the
	// unblocked account, so their open game screens restore the controls and un-strike
	// the name in place. Like NotifyUserBlocked it reaches only the actor and is in-app only.
	NotifyUserUnblocked = "user_unblocked"
)

// Intent is one live event destined for a single user. Payload is the
// FlatBuffers-encoded body (a scrabblefb.* table) that the gateway forwards
// verbatim to the client; EventID is a correlation id carried through unchanged.
type Intent struct {
	UserID  uuid.UUID
	Kind    string
	Payload []byte
	EventID string
}

// Publisher accepts live-event intents. Implementations must be safe for
// concurrent use and must not block the caller.
type Publisher interface {
	Publish(intents ...Intent)
}

// Nop is the default Publisher: it discards every intent.
type Nop struct{}

// Publish discards the intents.
func (Nop) Publish(...Intent) {}

// Hub is the in-process fan-in/fan-out between the domain publishers and the
// push subscribers (the gRPC stream). It is safe for concurrent use.
type Hub struct {
	mu      sync.Mutex
	subs    map[int]chan Intent
	nextID  int
	bufSize int
}

// defaultBuffer is the per-subscriber queue depth used when NewHub is given a
// non-positive size.
const defaultBuffer = 256

// NewHub returns a Hub whose per-subscriber buffer holds bufSize intents before
// dropping (a slow subscriber never blocks a publisher).
func NewHub(bufSize int) *Hub {
	if bufSize <= 0 {
		bufSize = defaultBuffer
	}
	return &Hub{subs: make(map[int]chan Intent), bufSize: bufSize}
}

// Publish delivers each intent to every current subscriber, dropping it for any
// subscriber whose buffer is full (best-effort live delivery).
func (h *Hub) Publish(intents ...Intent) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, in := range intents {
		for _, ch := range h.subs {
			select {
			case ch <- in:
			default:
			}
		}
	}
}

// Subscribe registers a new subscriber and returns its intent channel and an
// unsubscribe func that closes the channel. The caller reads the channel until
// it is closed or its own context ends, then calls unsubscribe.
func (h *Hub) Subscribe() (<-chan Intent, func()) {
	h.mu.Lock()
	defer h.mu.Unlock()
	id := h.nextID
	h.nextID++
	ch := make(chan Intent, h.bufSize)
	h.subs[id] = ch
	return ch, func() { h.unsubscribe(id) }
}

// unsubscribe removes and closes the subscriber's channel. It holds the same
// lock as Publish, so it never closes a channel mid-send.
func (h *Hub) unsubscribe(id int) {
	h.mu.Lock()
	defer h.mu.Unlock()
	if ch, ok := h.subs[id]; ok {
		delete(h.subs, id)
		close(ch)
	}
}