mirror of
https://github.com/IS1DI/x0gp.git
synced 2026-07-17 04:57:55 +00:00
x0gp
49351b4928
- Implement UDPVideoService to stream MJPEG video frames from ESP32s via UDP and send control commands. - Normalize race database schema: - Remove redundant driver_ids array from races table (migration 011). - Move per-driver outcomes (total_time_ms, best_lap_ms, position) to race_drivers table (migration 012). - Add device_id column to lobby_drivers to link active physical devices (migration 013). - Update WebSocket handler to pass lobbySvc, driversSvc, and videoSvc for driver identity tracking, syncing in-memory driver metadata, and forwarding WS controls to physical cars. - Add CORS middleware to HTTP routes. - Regenerate Swagger documentation.
774 lines
24 KiB
Go
774 lines
24 KiB
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"
|
|
)
|
|
|
|
const 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")
|
|
)
|
|
flag.Parse()
|
|
|
|
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("starting x0gp poc-server",
|
|
"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, 5*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))
|
|
|
|
// 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)
|
|
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("/health", healthHandler(hub, engine))
|
|
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/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/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("/ws", wsHandler(cfg, engine, hub, cat, lobbySvc, driversSvc, videoSvc, logger))
|
|
|
|
// 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 ------------------------------------------------------------
|
|
|
|
// 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, 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, 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
|
|
}
|
|
}
|
|
car, err := e.AddCar(c.ID, name, slot)
|
|
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 driver is assigned one
|
|
driverID := c.DriverID
|
|
if driverID == "" {
|
|
driverID = c.ID
|
|
}
|
|
if d, err := lobbySvc.GetDriver(driverID); err == nil && d.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(*d.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
|
|
}
|