galaxy-game/backend/internal/lobby/runtime_hooks.go

package lobby

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"go.uber.org/zap"
)

// OnRuntimeSnapshot updates the denormalised runtime view on the game
// row from a snapshot reported by the runtime module. The lobby
// transitions the game's lifecycle status when the snapshot reports a
// state change relevant to the lobby state machine:
//
// - `running` → `running` (after `starting`).
// - `engine_unreachable` / `start_failed` → `start_failed` while
// `starting`.
// - `finished` → triggers `OnGameFinished`.
//
// Per-player MaxPlanets / MaxPopulation are accumulated across the
// game lifetime so the capable-finish evaluation in `OnGameFinished`
// has the data it needs.
//
// The current implementation ships the entry point + state-machine logic; The implementation // (runtime) wires the actual call site.
func (s *Service) OnRuntimeSnapshot(ctx context.Context, gameID uuid.UUID, snapshot RuntimeSnapshot) error {
	game, err := s.GetGame(ctx, gameID)
	if err != nil {
		return err
	}
	prevTurn := game.RuntimeSnapshot.CurrentTurn
	merged := mergeRuntimeSnapshot(game.RuntimeSnapshot, snapshot)
	now := s.deps.Now().UTC()
	updated, err := s.deps.Store.UpdateGameRuntimeSnapshot(ctx, gameID, merged, now)
	if err != nil {
		return err
	}
	transitionedToPaused := false
	if next, transition := nextStatusFromSnapshot(updated.Status, snapshot); transition {
		switch next {
		case GameStatusFinished:
			s.deps.Cache.PutGame(updated)
			return s.OnGameFinished(ctx, gameID)
		default:
			rec, err := s.deps.Store.UpdateGameStatus(ctx, gameID, statusUpdate{
				NewStatus: next,
				UpdatedAt: now,
				SetStarted: next == GameStatusRunning && updated.StartedAt == nil,
				StartedAt: now,
			})
			if err != nil {
				return err
			}
			updated = rec
			if next == GameStatusPaused {
				transitionedToPaused = true
			}
		}
	}
	s.deps.Cache.PutGame(updated)
	if merged.CurrentTurn > prevTurn {
		s.publishTurnReady(ctx, gameID, merged.CurrentTurn)
	}
	if transitionedToPaused {
		s.publishGamePaused(ctx, gameID, merged.CurrentTurn, snapshot.RuntimeStatus)
	}
	return nil
}

// publishTurnReady fans out a `game.turn.ready` notification to every
// active member of the game once the engine reports a new
// `current_turn`. The intent is best-effort: a publisher failure is
// logged at warn level (matching the rest of OnRuntimeSnapshot's
// notification calls) and does not abort the snapshot bookkeeping.
// Idempotency is anchored on (game_id, turn), so a duplicate snapshot
// for the same turn collapses into a single notification at the
// notification.Submit boundary.
func (s *Service) publishTurnReady(ctx context.Context, gameID uuid.UUID, turn int32) {
	memberships, err := s.deps.Store.ListMembershipsForGame(ctx, gameID)
	if err != nil {
		s.deps.Logger.Warn("turn-ready notification: list memberships failed",
			zap.String("game_id", gameID.String()),
			zap.Int32("turn", turn),
			zap.Error(err))
		return
	}
	recipients := make([]uuid.UUID, 0, len(memberships))
	for _, m := range memberships {
		if m.Status != MembershipStatusActive {
			continue
		}
		recipients = append(recipients, m.UserID)
	}
	if len(recipients) == 0 {
		return
	}
	intent := LobbyNotification{
		Kind:           NotificationGameTurnReady,
		IdempotencyKey: fmt.Sprintf("turn-ready:%s:%d", gameID, turn),
		Recipients:     recipients,
		Payload: map[string]any{
			"game_id": gameID.String(),
			"turn":    turn,
		},
	}
	if pubErr := s.deps.Notification.PublishLobbyEvent(ctx, intent); pubErr != nil {
		s.deps.Logger.Warn("turn-ready notification failed",
			zap.String("game_id", gameID.String()),
			zap.Int32("turn", turn),
			zap.Error(pubErr))
	}
}

// publishGamePaused fans out a `game.paused` notification to every
// active member of the game when the lobby flips the game to
// `paused` in reaction to a runtime snapshot (typically a failed
// turn generation). The intent is best-effort: a publisher failure
// is logged at warn level and does not abort the snapshot
// bookkeeping. Idempotency is anchored on (game_id, turn) so a
// repeated `generation_failed` snapshot for the same turn collapses
// into a single notification at the notification.Submit boundary.
//
// reason carries the raw runtime status that triggered the pause
// (`engine_unreachable` / `generation_failed`); the UI displays a
// status-agnostic banner today but the payload is preserved so a
// future revision of the order tab can differentiate.
func (s *Service) publishGamePaused(ctx context.Context, gameID uuid.UUID, turn int32, reason string) {
	memberships, err := s.deps.Store.ListMembershipsForGame(ctx, gameID)
	if err != nil {
		s.deps.Logger.Warn("game-paused notification: list memberships failed",
			zap.String("game_id", gameID.String()),
			zap.Int32("turn", turn),
			zap.Error(err))
		return
	}
	recipients := make([]uuid.UUID, 0, len(memberships))
	for _, m := range memberships {
		if m.Status != MembershipStatusActive {
			continue
		}
		recipients = append(recipients, m.UserID)
	}
	if len(recipients) == 0 {
		return
	}
	intent := LobbyNotification{
		Kind:           NotificationGamePaused,
		IdempotencyKey: fmt.Sprintf("paused:%s:%d", gameID, turn),
		Recipients:     recipients,
		Payload: map[string]any{
			"game_id": gameID.String(),
			"turn":    turn,
			"reason":  reason,
		},
	}
	if pubErr := s.deps.Notification.PublishLobbyEvent(ctx, intent); pubErr != nil {
		s.deps.Logger.Warn("game-paused notification failed",
			zap.String("game_id", gameID.String()),
			zap.Int32("turn", turn),
			zap.Error(pubErr))
	}
}

// OnGameFinished completes the game lifecycle: marks the game as
// `finished`, evaluates capable-finish per active member, and
// transitions reservation rows to either `pending_registration`
// (capable) or deletes them (non-capable).
func (s *Service) OnGameFinished(ctx context.Context, gameID uuid.UUID) error {
	game, err := s.GetGame(ctx, gameID)
	if err != nil {
		return err
	}
	now := s.deps.Now().UTC()
	if game.Status != GameStatusFinished {
		updated, err := s.deps.Store.UpdateGameStatus(ctx, gameID, statusUpdate{
			NewStatus: GameStatusFinished,
			UpdatedAt: now,
			SetFinished: true,
			FinishedAt: now,
		})
		if err != nil {
			return err
		}
		game = updated
	}
	memberships, err := s.deps.Store.ListMembershipsForGame(ctx, gameID)
	if err != nil {
		return err
	}
	statsByUser := make(map[uuid.UUID]PlayerTurnStats, len(game.RuntimeSnapshot.PlayerStats))
	for _, st := range game.RuntimeSnapshot.PlayerStats {
		statsByUser[st.UserID] = st
	}
	expiry := now.Add(s.deps.Config.PendingRegistrationTTL)
	var promoteErrs []error
	for _, m := range memberships {
		if m.Status != MembershipStatusActive {
			continue
		}
		stats, hasStats := statsByUser[m.UserID]
		canonical := CanonicalKey(m.CanonicalKey)
		if hasStats && capableFinish(stats) {
			// Best-effort: drop the existing reservation row before
			// inserting the pending_registration so the per-game PK
			// does not block the transition.
			if err := s.deps.Store.DeleteRaceName(ctx, canonical, gameID); err != nil {
				promoteErrs = append(promoteErrs, fmt.Errorf("delete reservation %s: %w", canonical, err))
				continue
			}
			s.deps.Cache.RemoveRaceName(canonical)
			entry, err := s.deps.Store.InsertRaceName(ctx, raceNameInsert{
				Name: m.RaceName,
				Canonical: canonical,
				Status: RaceNameStatusPendingRegistration,
				OwnerUserID: m.UserID,
				GameID: gameID,
				SourceGameID: ptrUUID(gameID),
				ExpiresAt: &expiry,
			})
			if err != nil {
				promoteErrs = append(promoteErrs, fmt.Errorf("promote pending %s: %w", canonical, err))
				continue
			}
			s.deps.Cache.PutRaceName(entry)
			intent := LobbyNotification{
				Kind: NotificationLobbyRaceNamePending,
				IdempotencyKey: "racename-pending:" + string(canonical) + ":" + gameID.String(),
				Recipients: []uuid.UUID{m.UserID},
				Payload: map[string]any{
					"race_name": m.RaceName,
					"expires_at": expiry.Format(time.RFC3339),
				},
			}
			if pubErr := s.deps.Notification.PublishLobbyEvent(ctx, intent); pubErr != nil {
				s.deps.Logger.Warn("race-name pending notification failed",
					zap.String("canonical", string(canonical)),
					zap.Error(pubErr))
			}
			continue
		}
		if err := s.deps.Store.DeleteRaceName(ctx, canonical, gameID); err != nil {
			promoteErrs = append(promoteErrs, fmt.Errorf("delete non-capable reservation %s: %w", canonical, err))
			continue
		}
		s.deps.Cache.RemoveRaceName(canonical)
	}
	s.deps.Cache.PutGame(game)
	return errors.Join(promoteErrs...)
}

// OnRuntimeJobResult consumes adoption / removal events emitted by the
// runtime reconciler. The wiring connects the runtime → lobby callback
// through this entry point; the canonical mapping is:
//
// - reconciler reports `removed` → lobby cancels the game (the
// engine container is gone). Games already in `cancelled` or
// `finished` are ignored.
//
// Future job paths (start, stop, restart) may reuse the same shape.
func (s *Service) OnRuntimeJobResult(ctx context.Context, gameID uuid.UUID, result RuntimeJobResult) error {
	if s == nil {
		return nil
	}
	game, err := s.GetGame(ctx, gameID)
	if err != nil {
		if errors.Is(err, ErrNotFound) {
			return nil
		}
		return err
	}
	if game.Status == GameStatusCancelled || game.Status == GameStatusFinished {
		return nil
	}
	if result.Status != "removed" && result.Status != "stopped" {
		// Unknown status — ignore for forward compatibility.
		return nil
	}
	now := s.deps.Now().UTC()
	updated, err := s.deps.Store.UpdateGameStatus(ctx, gameID, statusUpdate{
		NewStatus: GameStatusCancelled,
		UpdatedAt: now,
	})
	if err != nil {
		return err
	}
	s.deps.Cache.PutGame(updated)
	s.deps.Logger.Info("game cancelled by runtime reconciler",
		zap.String("game_id", gameID.String()),
		zap.String("op", result.Op),
		zap.String("status", result.Status),
		zap.String("message", result.Message),
	)
	return nil
}

// mergeRuntimeSnapshot merges the incoming snapshot into the previous
// one, preserving running maxima of per-player planets and population
// across the game lifetime.
func mergeRuntimeSnapshot(prev, next RuntimeSnapshot) RuntimeSnapshot {
	out := RuntimeSnapshot{
		CurrentTurn: next.CurrentTurn,
		RuntimeStatus: next.RuntimeStatus,
		EngineHealth: next.EngineHealth,
		ObservedAt: next.ObservedAt,
	}
	statsByUser := make(map[uuid.UUID]PlayerTurnStats, len(prev.PlayerStats)+len(next.PlayerStats))
	for _, st := range prev.PlayerStats {
		statsByUser[st.UserID] = st
	}
	for _, st := range next.PlayerStats {
		existing, ok := statsByUser[st.UserID]
		if !ok {
			st.MaxPlanets = max32(st.MaxPlanets, st.CurrentPlanets)
			st.MaxPopulation = max32(st.MaxPopulation, st.CurrentPopulation)
			statsByUser[st.UserID] = st
			continue
		}
		st.InitialPlanets = existing.InitialPlanets
		st.InitialPopulation = existing.InitialPopulation
		st.MaxPlanets = max32(existing.MaxPlanets, max32(st.MaxPlanets, st.CurrentPlanets))
		st.MaxPopulation = max32(existing.MaxPopulation, max32(st.MaxPopulation, st.CurrentPopulation))
		statsByUser[st.UserID] = st
	}
	if len(statsByUser) > 0 {
		out.PlayerStats = make([]PlayerTurnStats, 0, len(statsByUser))
		for _, st := range statsByUser {
			out.PlayerStats = append(out.PlayerStats, st)
		}
	}
	return out
}

// nextStatusFromSnapshot maps the runtime-reported runtime status into
// a lobby status transition. Returns (next, true) when the lobby
// status must change; (current, false) otherwise.
//
// The map intentionally distinguishes the pre-running boot path
// (`starting → start_failed`) from the in-flight failure path
// (`running → paused`). Paused games can be resumed by the admin via
// the explicit `/resume` transition; the runtime keeps the engine
// container alive, the scheduler short-circuits ticks while paused,
// and any user-games command/order is rejected by the order handler
// with `turn_already_closed` until the game resumes.
func nextStatusFromSnapshot(currentStatus string, snapshot RuntimeSnapshot) (string, bool) {
	switch snapshot.RuntimeStatus {
	case "running":
		if currentStatus == GameStatusStarting {
			return GameStatusRunning, true
		}
	case "engine_unreachable", "generation_failed":
		if currentStatus == GameStatusStarting {
			return GameStatusStartFailed, true
		}
		if currentStatus == GameStatusRunning {
			return GameStatusPaused, true
		}
	case "start_failed":
		if currentStatus == GameStatusStarting {
			return GameStatusStartFailed, true
		}
	case "finished":
		if currentStatus != GameStatusFinished && currentStatus != GameStatusCancelled {
			return GameStatusFinished, true
		}
	case "stopped":
		if currentStatus == GameStatusRunning || currentStatus == GameStatusPaused {
			return GameStatusFinished, true
		}
	}
	return currentStatus, false
}

// capableFinish reports whether a per-player observation satisfies the
// "capable finish" criterion documented in
// `backend/PLAN.md` §5.4: max_planets > initial AND max_population >
// initial. Either of the inputs being zero (no observation) defaults
// to non-capable.
func capableFinish(stats PlayerTurnStats) bool {
	if stats.InitialPlanets == 0 || stats.InitialPopulation == 0 {
		return false
	}
	return stats.MaxPlanets > stats.InitialPlanets &&
		stats.MaxPopulation > stats.InitialPopulation
}

func max32(a, b int32) int32 {
	if a > b {
		return a
	}
	return b
}

func ptrUUID(u uuid.UUID) *uuid.UUID { v := u; return &v }