Fix SSE architecture for reliable connections (#13)
All checks were successful
CI / Deploy / test (push) Successful in 14s
CI / Deploy / lint (push) Successful in 25s
CI / Deploy / deploy (push) Successful in 1m32s

This commit was merged in pull request #13.
This commit is contained in:
2026-03-03 23:33:13 +00:00
parent 331c4c8759
commit 67a768ea22
20 changed files with 2950 additions and 231 deletions

View File

@@ -1,41 +1,24 @@
package snakegame
import (
"fmt"
"errors"
"net/http"
"strconv"
"time"
"github.com/alexedwards/scs/v2"
"github.com/go-chi/chi/v5"
"github.com/nats-io/nats.go"
"github.com/starfederation/datastar-go/datastar"
"github.com/ryanhamamura/games/chat"
chatcomponents "github.com/ryanhamamura/games/chat/components"
sharedcomponents "github.com/ryanhamamura/games/features/common/components"
"github.com/ryanhamamura/games/features/snakegame/pages"
"github.com/ryanhamamura/games/features/snakegame/services"
"github.com/ryanhamamura/games/sessions"
"github.com/ryanhamamura/games/snake"
)
func snakeChatColor(slot int) string {
if slot >= 0 && slot < len(snake.SnakeColors) {
return snake.SnakeColors[slot]
}
return "#666"
}
func snakeChatConfig(gameID string) chatcomponents.Config {
return chatcomponents.Config{
CSSPrefix: "snake",
PostURL: fmt.Sprintf("/snake/%s/chat", gameID),
Color: snakeChatColor,
StopKeyPropagation: true,
}
}
func HandleSnakePage(snakeStore *snake.SnakeStore, sm *scs.SessionManager) http.HandlerFunc {
func HandleSnakePage(snakeStore *snake.SnakeStore, svc *services.GameService, sm *scs.SessionManager) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
gameID := chi.URLParam(r, "id")
si, ok := snakeStore.Get(gameID)
@@ -77,13 +60,14 @@ func HandleSnakePage(snakeStore *snake.SnakeStore, sm *scs.SessionManager) http.
}
sg := si.GetGame()
if err := pages.GamePage(sg, mySlot, nil, snakeChatConfig(gameID), gameID).Render(r.Context(), w); err != nil {
chatCfg := svc.ChatConfig(gameID)
if err := pages.GamePage(sg, mySlot, nil, chatCfg, gameID).Render(r.Context(), w); err != nil {
http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
}
}
}
func HandleSnakeEvents(snakeStore *snake.SnakeStore, nc *nats.Conn, sm *scs.SessionManager) http.HandlerFunc {
func HandleSnakeEvents(snakeStore *snake.SnakeStore, svc *services.GameService, sm *scs.SessionManager) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
gameID := chi.URLParam(r, "id")
si, ok := snakeStore.Get(gameID)
@@ -95,17 +79,25 @@ func HandleSnakeEvents(snakeStore *snake.SnakeStore, nc *nats.Conn, sm *scs.Sess
playerID := sessions.GetPlayerID(sm, r)
mySlot := si.GetPlayerSlot(playerID)
// Subscribe to game updates BEFORE creating SSE (following portigo pattern)
gameSub, gameCh, err := svc.SubscribeGameUpdates(gameID)
if err != nil {
http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
return
}
defer gameSub.Unsubscribe() //nolint:errcheck
sse := datastar.NewSSE(w, r, datastar.WithCompression(
datastar.WithBrotli(datastar.WithBrotliLevel(5)),
))
chatCfg := snakeChatConfig(gameID)
chatCfg := svc.ChatConfig(gameID)
// Chat room (multiplayer only)
var room *chat.Room
sg := si.GetGame()
if sg.Mode == snake.ModeMultiplayer {
room = chat.NewRoom(nc, snake.ChatSubject(gameID))
room = svc.ChatRoom(gameID)
}
chatMessages := func() []chat.Message {
@@ -118,36 +110,21 @@ func HandleSnakeEvents(snakeStore *snake.SnakeStore, nc *nats.Conn, sm *scs.Sess
patchAll := func() error {
si, ok = snakeStore.Get(gameID)
if !ok {
return fmt.Errorf("game not found")
return errors.New("game not found")
}
mySlot = si.GetPlayerSlot(playerID)
sg = si.GetGame()
return sse.PatchElementTempl(pages.GameContent(sg, mySlot, chatMessages(), chatCfg, gameID))
}
sendPing := func() error {
return sse.PatchElementTempl(sharedcomponents.ConnectionIndicator(time.Now().UnixMilli()))
}
// Send initial render and ping
if err := sendPing(); err != nil {
return
}
// Send initial render
if err := patchAll(); err != nil {
return
}
heartbeat := time.NewTicker(15 * time.Second)
heartbeat := time.NewTicker(1 * time.Second)
defer heartbeat.Stop()
// Subscribe to game updates via NATS
gameCh := make(chan *nats.Msg, 64)
gameSub, err := nc.ChanSubscribe(snake.GameSubject(gameID), gameCh)
if err != nil {
return
}
defer gameSub.Unsubscribe() //nolint:errcheck
// Chat subscription (multiplayer only)
var chatCh <-chan chat.Message
var cleanupChat func()
@@ -164,7 +141,8 @@ func HandleSnakeEvents(snakeStore *snake.SnakeStore, nc *nats.Conn, sm *scs.Sess
return
case <-heartbeat.C:
if err := sendPing(); err != nil {
// Heartbeat refreshes game state to keep connection alive
if err := patchAll(); err != nil {
return
}
@@ -231,7 +209,7 @@ type chatSignals struct {
ChatMsg string `json:"chatMsg"`
}
func HandleSendChat(snakeStore *snake.SnakeStore, nc *nats.Conn, sm *scs.SessionManager) http.HandlerFunc {
func HandleSendChat(snakeStore *snake.SnakeStore, svc *services.GameService, sm *scs.SessionManager) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
gameID := chi.URLParam(r, "id")
si, ok := snakeStore.Get(gameID)
@@ -264,7 +242,7 @@ func HandleSendChat(snakeStore *snake.SnakeStore, nc *nats.Conn, sm *scs.Session
Message: signals.ChatMsg,
}
room := chat.NewRoom(nc, snake.ChatSubject(gameID))
room := svc.ChatRoom(gameID)
room.Send(msg)
sse := datastar.NewSSE(w, r)