x0gp/server/cmd/poc-server/main.go

// Command poc-server is the minimal x0gp server for end-to-end PoC.
//
// Topology (PoC):
//
//	browser -- WS --> poc-server -- in-process --> RaceEngine (60 Hz tick)
//	                                       \-> Snapshot fan-out via Hub
//
// What's NOT here (intentionally, for PoC simplicity):
//   - auth / JWT (DevMode = true)
//   - Postgres / Redis (in-memory state)
//   - agent-linker (no real cars yet)
//   - overhead-CV
//   - billing
//   - rate limiting beyond per-send-buffer drop
//
// @title           x0gp PoC server API
// @version         0.1.0-poc
// @description     Minimal HTTP+WebSocket server for the x0gp RC racing PoC. REST endpoints under /api/* manage the track/car catalog; /health and /stats expose server liveness and realtime metrics; /ws carries the realtime race protocol.
// @description     Auth is intentionally disabled in PoC mode (X0GP_DEV_MODE=true). All catalog mutations are allowed without a JWT.
// @BasePath        /
// @schemes         http ws
// @host            localhost:8080
// @contact.name    x0gp
// @license.name    MIT
package main

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
	httpSwagger "github.com/swaggo/http-swagger"

	// Side-effect import: pulls in the generated OpenAPI spec at init().
	// `swag init` (see Makefile / docs/) regenerates this file from
	// the // @... annotations on the handlers.
	_ "github.com/x0gp/server/docs"

	"github.com/x0gp/server/internal/catalog"
	"github.com/x0gp/server/internal/clans"
	"github.com/x0gp/server/internal/config"
	"github.com/x0gp/server/internal/control"
	"github.com/x0gp/server/internal/drivers"
	"github.com/x0gp/server/internal/lobby"
	"github.com/x0gp/server/internal/races"
	"github.com/x0gp/server/internal/races/seed"
	"github.com/x0gp/server/internal/realtime"
	"github.com/x0gp/server/internal/storage/postgres"
	"github.com/x0gp/server/internal/transport"
)

var serverVersion = "0.1.0-poc"

var (
	upgrader = websocket.Upgrader{
		ReadBufferSize:  4096,
		WriteBufferSize: 4096,
		CheckOrigin:     func(r *http.Request) bool { return true }, // PoC: dev only
	}
)

func withCORS(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Разрешаем доступ с любых доменов. Для безопасности можно указать конкретный домен вместо *
		w.Header().Set("Access-Control-Allow-Origin", "*")
		w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
		w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization")

		// Важно: обрабатываем предварительный запрос (Preflight)
		if r.Method == http.MethodOptions {
			w.WriteHeader(http.StatusOK)
			return
		}

		next.ServeHTTP(w, r)
	})
}

func main() {
	var (
		addr      = flag.String("addr", "", "override HTTP listen address")
		devMode   = flag.Bool("dev", true, "enable development mode")
		seedRaces = flag.Bool("seed-races", false, "seed mock race data (finished/live/plans) on startup")
		seedReset = flag.Bool("reset", false, "with --seed-races, wipe seed-managed data before inserting")
		version   = flag.Bool("version", false, "print version and exit")
	)
	flag.Parse()

	if *version {
		fmt.Println(serverVersion)
		return
	}

	if *addr != "" {
		_ = os.Setenv("X0GP_HTTP_ADDR", *addr)
	}
	if !*devMode {
		_ = os.Setenv("X0GP_DEV_MODE", "false")
	}
	if *seedRaces {
		_ = os.Setenv("X0GP_SEED_RACES", "1")
	}
	if *seedReset {
		_ = os.Setenv("X0GP_SEED_RESET", "1")
	}

	cfg, err := config.Load()
	if err != nil {
		fmt.Fprintf(os.Stderr, "config error: %v\n", err)
		os.Exit(1)
	}

	logger := newLogger(cfg.LogLevel)
	logger.Info(fmt.Sprintf("starting x0gp poc-server version %s", serverVersion),
		"version", serverVersion,
		"addr", cfg.HTTPAddr,
		"tick_hz", cfg.TickRate,
		"snapshot_hz", cfg.SnapshotRate,
		"dev_mode", cfg.DevMode,
	)

	ctx, cancel := signal.NotifyContext(context.Background(),
		syscall.SIGINT, syscall.SIGTERM)
	defer cancel()

	// --- Database: fail-fast. ---
	if cfg.DatabaseURL == "" {
		logger.Error("DATABASE_URL is required")
		os.Exit(1)
	}
	bootCtx, bootCancel := context.WithTimeout(ctx, 15*time.Second)
	pool, err := postgres.Open(bootCtx, postgres.Config{URL: cfg.DatabaseURL})
	bootCancel()
	if err != nil {
		logger.Error("postgres connect failed", "err", err)
		os.Exit(1)
	}
	defer pool.Close()

	migCtx, migCancel := context.WithTimeout(ctx, 30*time.Second)
	if err := postgres.Migrate(migCtx, pool); err != nil {
		migCancel()
		logger.Error("postgres migrate failed", "err", err)
		os.Exit(1)
	}
	migCancel()
	logger.Info("postgres ready")

	engine := control.NewEngine(cfg.TickRate)
	hub := NewHub(logger)
	pgStore := postgres.NewPgStore(pool)
	cat, err := catalog.NewService(ctx, pgStore)
	if err != nil {
		logger.Error("catalog load failed", "err", err)
		os.Exit(1)
	}
	defer cat.Stop()
	cb := newCatalogBroadcaster(cat, hub, logger)

	// Initialize UDP video and command service.
	videoSvc := NewUDPVideoService(logger, cfg.UDPVideoAddr, cfg.ESPCmdPort)

	// Lobby + races: live races live in lobby.Service, finished races get
	// snapshotted into Postgres via the lifecycle hook.
	lobbySvc := lobby.NewService()
	racesPgStore := races.NewPgStore(pool)
	racesSvc := races.NewService(racesPgStore, lobbySvc)
	liveStore := races.NewLiveStore(racesPgStore)
	lobbySvc.SetPersistence(liveStore)
	racesSvc.SetLiveStore(liveStore)

	// Restore the in-memory lobby from Postgres so active races and
	// driver presence survive a restart.
	restoreCtx, restoreCancel := context.WithTimeout(ctx, 15*time.Second)
	if racesRestored, driversRestored, err := racesSvc.RestoreFromDB(restoreCtx); err != nil {
		logger.Warn("restore from db failed", "err", err)
	} else {
		logger.Info("restored lobby from db", "races", racesRestored, "drivers", driversRestored)
	}
	restoreCancel()

	// Drivers and clans.
	clansSvc := clans.NewService(clans.NewPgStore(pool))
	driversSvc := drivers.NewService(drivers.NewPgStore(pool))

	// WebRTC 1-to-1 driver-car proxy service
	webrtcSvc := NewWebRTCService(logger, engine, lobbySvc, videoSvc)

	// Persist finished races so the /api/races list and keyset pagination
	// can serve historical data across restarts. The snapshot is best-
	// effort: errors are logged but never block the lobby.
	lobbySvc.SetRaceLifecycleHook(
		func(meta lobby.RaceMeta) {
			logger.Debug("race created", "race_id", meta.ID)
		},
		func(raceID string) {
			meta, err := lobbySvc.GetRace(raceID)
			if err != nil {
				return
			}
			if meta.Status != lobby.RaceStatusFinished {
				return
			}
			hookCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
			defer cancel()
			if err := racesSvc.PersistFinished(hookCtx, races.SnapshotFinished{
				Meta:       meta,
				FinishedMs: time.Now().UnixMilli(),
			}); err != nil {
				logger.Warn("persist finished race failed", "race_id", raceID, "err", err)
			}
		},
	)

	// Scheduler: ticks every 5s, materialises due race_plans into the lobby
	// and auto-attaches queued drivers.
	scheduler := races.NewScheduler(racesPgStore, lobbySvc, 5*time.Second)

	// Optional: seed mock race data on startup.
	if os.Getenv("X0GP_SEED_RACES") == "1" {
		seedCtx, seedCancel := context.WithTimeout(ctx, 30*time.Second)
		runner := seed.NewRunner(racesPgStore, lobbySvc, liveStore, clans.NewPgStore(pool), drivers.NewPgStore(pool))
		reset := os.Getenv("X0GP_SEED_RESET") == "1"
		summary, err := runner.Run(seedCtx, seed.Options{
			Reset: reset,
			Now:   time.Now(),
		})
		seedCancel()
		if err != nil {
			logger.Error("seed races failed", "err", err)
			os.Exit(1)
		}
		logger.Info("seed races done",
			"finished", summary.FinishedInserted,
			"live", summary.LiveCreated,
			"plans", summary.PlansInserted,
			"queue", summary.QueueInserted,
			"reset", reset,
			"dur_ms", summary.Duration.Milliseconds(),
		)
	}

	var wg sync.WaitGroup
	wg.Add(4)

	go func() {
		defer wg.Done()
		hub.Run(ctx)
	}()
	go func() {
		defer wg.Done()
		engine.Run(ctx)
	}()
	go func() {
		defer wg.Done()
		cb.Run()
	}()
	go func() {
		defer wg.Done()
		scheduler.Run(ctx)
	}()

	// Snapshot publisher: engine -> hub.
	wg.Add(1)
	go func() {
		defer wg.Done()
		snapshotLoop(ctx, engine, hub, logger)
	}()

	// UDP video and command receiver.
	videoSvc.Start(ctx, &wg)

	// Start WebRTC service
	webrtcSvc.Start(ctx, &wg)

	mux := http.NewServeMux()
	mux.HandleFunc("/health", healthHandler(hub, engine))
	mux.HandleFunc("/api/version", versionHandler())
	mux.HandleFunc("/stats", statsHandler(hub, engine))
	mux.HandleFunc("/api/catalog", catalogHandler(cat))
	mux.HandleFunc("/api/tracks", tracksRouter(cat))
	mux.HandleFunc("/api/tracks/", trackByIDHandler(cat))
	mux.HandleFunc("/api/tracks/calendar", trackCalendarRouter(cat))
	mux.HandleFunc("/api/tracks/calendar/", deleteTrackCalendarHandler(cat))
	mux.HandleFunc("/api/cars", carsRouter(cat))
	mux.HandleFunc("/api/cars/", carByIDHandler(cat))
	mux.HandleFunc("/api/races", racesListHandler(racesSvc))
	mux.HandleFunc("/api/races/upcoming", racesUpcomingHandler(racesSvc))
	mux.HandleFunc("/api/races/queue", racesQueueRouter(racesSvc))
	mux.HandleFunc("/api/races/queue/join", racesQueueJoinHandler(racesSvc))
	mux.HandleFunc("/api/races/plans", racePlansRouter(racesSvc))
	mux.HandleFunc("/api/races/plans/", racePlansDeleteHandler(racesSvc))
	mux.HandleFunc("/api/leaderboard", leaderboardHandler(racesSvc))
	mux.HandleFunc("/api/leaderboard/tracks", calendarLeaderboardHandler(racesSvc))
	mux.HandleFunc("/api/clans", clansRouter(clansSvc))
	mux.HandleFunc("/api/clans/", clansByIDHandler(clansSvc))
	mux.HandleFunc("/api/drivers", driversRouter(driversSvc))
	mux.HandleFunc("/api/drivers/", driversByIDHandler(driversSvc, lobbySvc))
	mux.HandleFunc("/api/video/stream", videoSvc.StreamHandler())
	mux.HandleFunc("/api/video/control", videoSvc.ControlHandler())
	mux.HandleFunc("/api/webrtc/connect", webrtcSvc.ConnectHandler())
	mux.HandleFunc("/ws", wsHandler(cfg, engine, hub, cat, lobbySvc, driversSvc, videoSvc, logger))

	// Static client files (resolving TODO)
	mux.Handle("/", http.FileServer(http.Dir("./web")))

	// Swagger UI + OpenAPI spec.
	//   UI:    GET /swagger/index.html
	//   Spec:  GET /swagger/doc.json
	mux.Handle("/swagger/", httpSwagger.WrapHandler)

	srv := &http.Server{
		Addr:              cfg.HTTPAddr,
		Handler:           withCORS(withLogging(mux, logger)),
		ReadHeaderTimeout: 5 * time.Second,
	}

	// Start HTTP server.
	go func() {
		logger.Info("http listening", "addr", cfg.HTTPAddr)
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			logger.Error("http server failed", "err", err)
			cancel()
		}
	}()

	<-ctx.Done()
	logger.Info("shutdown signal received")

	shutCtx, shutCancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer shutCancel()
	_ = srv.Shutdown(shutCtx)

	engine.Stop()
	hub.Stop()
	cb.Stop()
	wg.Wait()
	logger.Info("shutdown complete")
}

// snapshotLoop forwards RaceEngine snapshots to all connected clients.
func snapshotLoop(ctx context.Context, e *control.Engine, h *Hub, logger *slog.Logger) {
	for {
		select {
		case <-ctx.Done():
			return
		case snap, ok := <-e.Snapshots():
			if !ok {
				return
			}
			h.BroadcastSnapshot(snap)
		}
	}
}

// HubWrapper extends realtime.Hub with helpers tailored to this binary.
// Using a thin wrapper keeps realtime package free of transport-specific deps.
type Hub struct {
	*realtime.Hub
}

func NewHub(logger *slog.Logger) *Hub {
	return &Hub{Hub: realtime.NewHub()}
}

// BroadcastSnapshot wraps a RaceSnapshot in an envelope and broadcasts.
func (h *Hub) BroadcastSnapshot(snap transport.RaceSnapshot) {
	h.Publish(&transport.Envelope{
		Type:    transport.TypeRaceSnapshot,
		TSMs:    time.Now().UnixMilli(),
		Payload: snap,
	})
}

// HTTP handlers ------------------------------------------------------------

// versionHandler godoc
// @Summary      Get server version
// @Description  Returns the current server version.
// @Tags         system
// @Produce      json
// @Success      200 {object} transport.VersionResponse "Success"
// @Router       /api/version [get]
func versionHandler() http.HandlerFunc {
	return func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		_ = json.NewEncoder(w).Encode(transport.VersionResponse{
			Version: serverVersion,
		})
	}
}

// healthHandler godoc
// @Summary      Liveness probe
// @Description  Returns 200 OK with current engine phase, WS connection count and total snapshot drops. Intended for load-balancer health checks.
// @Tags         system
// @Produce      json
// @Success      200 {object} map[string]interface{} "Server is up"
// @Router       /health [get]
func healthHandler(h *Hub, e *control.Engine) http.HandlerFunc {
	return func(w http.ResponseWriter, _ *http.Request) {
		stats := h.Stats()
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusOK)
		_ = json.NewEncoder(w).Encode(map[string]any{
			"status":         "ok",
			"version":        serverVersion,
			"phase":          e.Phase().String(),
			"tick":           0, // Engine has no public getter; would add later
			"connections":    stats.Connections,
			"snapshot_drops": stats.DropsTotal,
			"time":           time.Now().UTC().Format(time.RFC3339Nano),
		})
	}
}

// statsHandler godoc
// @Summary      Realtime runtime stats
// @Description  Returns the current race phase plus realtime counters: active WS connections and total snapshots dropped due to slow consumers.
// @Tags         system
// @Produce      json
// @Success      200 {object} map[string]interface{} "Runtime counters"
// @Router       /stats [get]
func statsHandler(h *Hub, e *control.Engine) http.HandlerFunc {
	return func(w http.ResponseWriter, _ *http.Request) {
		stats := h.Stats()
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{
			"connections":    stats.Connections,
			"snapshot_drops": stats.DropsTotal,
			"race_phase":     e.Phase().String(),
		})
	}
}

// wsHandler godoc
// @Summary      WebSocket endpoint (binary JSON envelope protocol)
// @Description  Opens a bidirectional WebSocket for the realtime race protocol. The server replies with `server_hello`, then broadcasts `race_snapshot` frames at `X0GP_SNAPSHOT_HZ`.
//
// @Description  ## Frame format
// @Description  Every frame is a JSON envelope of the form:
// @Description  ```json
// @Description  { "type": "<message_type>", "seq": 42, "ts_ms": 1700000000000, "payload": { ... } }
// @Description  ```
// @Description  `type` discriminates the payload; `payload` is documented per message type in the schemas listed in `components/schemas` (see `Envelope`, `ClientHello`, `ServerHello`, `InputState`, `JoinRace`, `LeaveRace`, `RaceSnapshot`, `InputAck`, `RaceEvent`, `Ping`, `Pong`, `ErrorMsg`, plus the catalog/lobby/stats envelopes).
// @Description  ## Client -> Server
// @Description  - `client_hello` — first frame after connect. Sent once.
// @Description  - `join_race` / `leave_race` — join or leave the race session.
// @Description  - `input_state` — high-rate control (steering/throttle/brake/gear). Send at 60-120 Hz.
// @Description  - `ping` — round-trip time probe.
// @Description  - `track_*` / `car_*` — catalog CRUD (mirror of the REST /api/* endpoints).
// @Description  - `lobby_*` — lobby management.
// @Description  - `stats_request` — ask for a stats snapshot.
// @Description  ## Server -> Client
// @Description  - `server_hello` — handshake reply; includes session id and engine config.
// @Description  - `race_snapshot` — periodic authoritative state broadcast.
// @Description  - `input_ack` — per-input acknowledgement (optional; drop-tolerant).
// @Description  - `race_event` — start/lap/sector/dnf/finish.
// @Description  - `track_snapshot` / `car_snapshot` / `catalog_summary` — catalog feeds on change.
// @Description  - `lobby_snapshot` — full lobby state on change.
// @Description  - `stats_snapshot` / `race_standings` / `lap_recorded` / `race_result` — stats feeds.
// @Description  - `error` — protocol or validation error.
// @Description  ## Notes
// @Description  - Origin checks are disabled in PoC mode (`CheckOrigin: always-allow`).
// @Description  - Ping interval: 20 s. Pong wait: 60 s. Max frame size: 8 KiB.
// @Description  - Input acks are best-effort and silently dropped under load.
// @Tags         websocket
// @Success      101 {object} transport.Envelope "Switching protocols. Subsequent frames follow the envelope schema."
// @Router       /ws [get]
func wsHandler(cfg *config.Config, e *control.Engine, h *Hub, cat *catalog.Service, lobbySvc *lobby.Service, driversSvc *drivers.Service, videoSvc *UDPVideoService, logger *slog.Logger) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			logger.Warn("ws upgrade failed", "err", err)
			return
		}
		client := &realtime.Client{
			ID:   uuid.NewString(),
			Send: make(chan []byte, 64),
			Done: make(chan struct{}),
		}
		h.Register(client)
		logger.Info("client connected",
			"client_id", client.ID,
			"remote", r.RemoteAddr,
		)
		go writePump(client, conn, logger)
		go readPump(cfg, client, conn, e, h, cat, lobbySvc, driversSvc, videoSvc, logger)
	}
}

// Per-connection goroutines ----------------------------------------------

const (
	pongWait     = 60 * time.Second
	pingPeriod   = 20 * time.Second
	writeTimeout = 5 * time.Second
)

func writePump(c *realtime.Client, conn *websocket.Conn, logger *slog.Logger) {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		_ = conn.Close()
		c.Close()
	}()
	for {
		select {
		case <-c.Done:
			return
		case msg, ok := <-c.Send:
			if !ok {
				_ = conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}
			_ = conn.SetWriteDeadline(time.Now().Add(writeTimeout))
			if err := conn.WriteMessage(websocket.TextMessage, msg); err != nil {
				logger.Debug("ws write failed", "client_id", c.ID, "err", err)
				return
			}
		case <-ticker.C:
			_ = conn.SetWriteDeadline(time.Now().Add(writeTimeout))
			if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

func readPump(cfg *config.Config, c *realtime.Client, conn *websocket.Conn, e *control.Engine, h *Hub, cat *catalog.Service, lobbySvc *lobby.Service, driversSvc *drivers.Service, videoSvc *UDPVideoService, logger *slog.Logger) {
	defer func() {
		h.Unregister(c)
		logger.Info("client disconnected", "client_id", c.ID)
		_ = conn.Close()
	}()

	conn.SetReadLimit(8192)
	_ = conn.SetReadDeadline(time.Now().Add(pongWait))
	conn.SetPongHandler(func(string) error {
		_ = conn.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})

	var seq uint64

	for {
		_, raw, err := conn.ReadMessage()
		if err != nil {
			return
		}

		env, err := transport.Decode(raw)
		if err != nil {
			sendError(c, "decode_error", err.Error())
			continue
		}
		seq++
		env.Seq = seq

		switch env.Type {
		case transport.TypeClientHello:
			handleClientHello(c, e, cfg, lobbySvc, driversSvc, env)
		case transport.TypeJoinRace:
			handleJoinRace(c, e, lobbySvc, env)
		case transport.TypeLeaveRace:
			handleLeaveRace(c, e, env)
		case transport.TypeInputState:
			handleInput(c, e, lobbySvc, videoSvc, env)
		case transport.TypePing:
			handlePing(c, env)
		case transport.TypeChatMessage:
			// Not implemented in PoC; acknowledge silently.
		case transport.TypeTrackList,
			transport.TypeTrackGet,
			transport.TypeTrackCreate,
			transport.TypeTrackUpdate,
			transport.TypeTrackDelete:
			handleTrackWSMessage(c, cat, env)
		case transport.TypeCarList,
			transport.TypeCarGet,
			transport.TypeCarCreate,
			transport.TypeCarUpdate,
			transport.TypeCarDelete:
			handleCarWSMessage(c, cat, env)
		default:
			sendError(c, "unknown_type", string(env.Type))
		}
	}
}

// Message handlers --------------------------------------------------------

func handleClientHello(c *realtime.Client, e *control.Engine, cfg *config.Config, lobbySvc *lobby.Service, driversSvc *drivers.Service, env *transport.Envelope) {
	hello := transport.ServerHello{
		SessionID: c.ID,
		RaceTick:  0,
	}
	hello.ServerVersion = serverVersion
	hello.Config.TickRate = cfg.TickRate
	hello.Config.SnapshotRate = cfg.SnapshotRate
	hello.Config.MaxCars = 4
	c.SessionID = c.ID

	// Extract client_id (DriverID) if present in payload
	payload, _ := env.Payload.(map[string]any)
	if payload != nil {
		if clientID, ok := payload["client_id"].(string); ok && clientID != "" {
			c.DriverID = clientID
			// Pre-populate driver in the lobby
			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			if drv, err := driversSvc.Get(ctx, clientID); err == nil {
				_, _ = lobbySvc.AddDriver(clientID, drv.Name, "")
				lobbySvc.SetDriverProfile(clientID, drv.Nickname, drv.AvatarURL, drv.ClanID, "")
			}
			cancel()
		}
	}

	c.Send <- mustEncode(transport.TypeServerHello, hello)
}

func handleJoinRace(c *realtime.Client, e *control.Engine, lobbySvc *lobby.Service, env *transport.Envelope) {
	payload, _ := env.Payload.(map[string]any)
	slot := 0
	name := "anon"
	if payload != nil {
		if v, ok := payload["car_slot"].(float64); ok {
			slot = int(v)
		}
		if v, ok := payload["car_name"].(string); ok && v != "" {
			name = v
		}
	}

	// Resolve DeviceID for this driver internally
	driverID := c.DriverID
	if driverID == "" {
		driverID = c.ID
	}
	var deviceID *int
	if d, err := lobbySvc.GetDriver(driverID); err == nil {
		deviceID = d.DeviceID
	}

	car, err := e.AddCar(c.ID, name, slot, deviceID)
	if err != nil {
		sendError(c, "join_failed", err.Error())
		return
	}
	c.Send <- mustEncode(transport.TypeRaceEvent, transport.RaceEvent{
		Event: "joined",
		CarID: car.ID,
		TSMs:  time.Now().UnixMilli(),
	})
}

func handleLeaveRace(c *realtime.Client, e *control.Engine, _ *transport.Envelope) {
	e.RemoveCar(c.ID)
}

func handleInput(c *realtime.Client, e *control.Engine, lobbySvc *lobby.Service, videoSvc *UDPVideoService, env *transport.Envelope) {
	payload, ok := env.Payload.(map[string]any)
	if !ok {
		return
	}
	in := transport.InputState{
		Steering: getFloat(payload, "steering"),
		Throttle: getFloat(payload, "throttle"),
		Brake:    getFloat(payload, "brake"),
	}
	if v, ok := payload["gear"].(float64); ok {
		in.Gear = int(v)
	}
	car := e.GetCarByDriver(c.ID)
	if car == nil {
		return
	}
	car.ApplyInput(in, 1.0/60.0)

	// Forward command transitions to ESP32 physical car if car has a device_id assigned
	if car.DeviceID != nil {
		var cmd string
		if in.Throttle > 0.1 {
			cmd = "start"
		} else {
			cmd = "stop"
		}
		if in.Steering < -0.2 {
			cmd = "left"
		} else if in.Steering > 0.2 {
			cmd = "right"
		}

		if cmd != "" && cmd != c.LastUDPCommand {
			c.LastUDPCommand = cmd
			_ = videoSvc.SendCommand(uint8(*car.DeviceID), cmd)
		}
	}

	// Optional lightweight ack (clients can disable for max throughput).
	ack := transport.InputAck{
		Seq:          env.Seq,
		AppliedAtMs:  time.Now().UnixMilli(),
		ServerTickMs: time.Now().UnixMilli(),
	}
	select {
	case c.Send <- mustEncode(transport.TypeInputAck, ack):
	default:
		// Drop ack silently.
	}
}

func handlePing(c *realtime.Client, env *transport.Envelope) {
	payload, _ := env.Payload.(map[string]any)
	clientTS := int64(0)
	if payload != nil {
		if v, ok := payload["client_ts_ms"].(float64); ok {
			clientTS = int64(v)
		}
	}
	c.Send <- mustEncode(transport.TypePong, transport.Pong{
		ClientTSMs: clientTS,
		ServerTSMs: time.Now().UnixMilli(),
	})
}

// Helpers ------------------------------------------------------------------

func sendError(c *realtime.Client, code, msg string) {
	select {
	case c.Send <- mustEncode(transport.TypeError, transport.ErrorMsg{
		Code: code, Message: msg,
	}):
	default:
	}
}

func mustEncode(t transport.MessageType, payload any) []byte {
	data, err := transport.Encode(&transport.Envelope{
		Type:    t,
		TSMs:    time.Now().UnixMilli(),
		Payload: payload,
	})
	if err != nil {
		return nil
	}
	return data
}

func getFloat(m map[string]any, key string) float64 {
	if v, ok := m[key].(float64); ok {
		return v
	}
	return 0
}

func newLogger(level string) *slog.Logger {
	var lvl slog.Level
	switch level {
	case "debug":
		lvl = slog.LevelDebug
	case "warn":
		lvl = slog.LevelWarn
	case "error":
		lvl = slog.LevelError
	default:
		lvl = slog.LevelInfo
	}
	h := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: lvl})
	return slog.New(h)
}

func withLogging(next http.Handler, logger *slog.Logger) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()
		rw := &statusRecorder{ResponseWriter: w, status: 200}
		next.ServeHTTP(rw, r)
		logger.Debug("http",
			"method", r.Method,
			"path", r.URL.Path,
			"status", rw.status,
			"dur_ms", time.Since(start).Milliseconds(),
		)
	})
}

type statusRecorder struct {
	http.ResponseWriter
	status int
}

func (r *statusRecorder) WriteHeader(code int) {
	r.status = code
	r.ResponseWriter.WriteHeader(code)
}

// Hijack passes through to the underlying ResponseWriter so that the
// gorilla/websocket upgrader can take over the connection.
func (r *statusRecorder) Hijack() (net.Conn, *bufio.ReadWriter, error) {
	if h, ok := r.ResponseWriter.(http.Hijacker); ok {
		return h.Hijack()
	}
	return nil, nil, http.ErrNotSupported
}