From 49351b4928f109cc110cb7439a80a6cfe87b2ef0 Mon Sep 17 00:00:00 2001 From: x0gp Date: Mon, 6 Jul 2026 12:38:34 +0400 Subject: [PATCH] feat(server): integrate UDP video service and normalize database schema - 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. --- .../cmd/poc-server/drivers_clans_handlers.go | 74 +++- server/cmd/poc-server/main.go | 90 ++++- server/cmd/poc-server/races_handlers.go | 9 +- server/cmd/poc-server/video_udp.go | 381 ++++++++++++++++++ server/docs/docs.go | 140 +++++++ server/docs/swagger.json | 140 +++++++ server/docs/swagger.yaml | 96 +++++ server/internal/config/config.go | 5 + server/internal/lobby/lobby.go | 49 ++- server/internal/lobby/lobby_test.go | 3 + server/internal/lobby/types.go | 20 + server/internal/races/seed/seed.go | 58 +-- server/internal/races/service.go | 124 +++++- server/internal/races/store.go | 274 ++++++++++--- server/internal/realtime/hub.go | 12 +- .../migrations/011_drop_driver_ids.sql | 11 + .../migrations/012_normalize_race_results.sql | 45 +++ .../postgres/migrations/013_driver_device.sql | 6 + server/internal/transport/codec.go | 1 + 19 files changed, 1406 insertions(+), 132 deletions(-) create mode 100644 server/cmd/poc-server/video_udp.go create mode 100644 server/internal/storage/postgres/migrations/011_drop_driver_ids.sql create mode 100644 server/internal/storage/postgres/migrations/012_normalize_race_results.sql create mode 100644 server/internal/storage/postgres/migrations/013_driver_device.sql diff --git a/server/cmd/poc-server/drivers_clans_handlers.go b/server/cmd/poc-server/drivers_clans_handlers.go index d338cff..d849af8 100644 --- a/server/cmd/poc-server/drivers_clans_handlers.go +++ b/server/cmd/poc-server/drivers_clans_handlers.go @@ -26,6 +26,7 @@ import ( "github.com/x0gp/server/internal/clans" "github.com/x0gp/server/internal/drivers" + "github.com/x0gp/server/internal/lobby" "github.com/x0gp/server/internal/transport" ) @@ -282,13 +283,84 @@ func driversListHandler(svc *drivers.Service) http.HandlerFunc { // @Router /api/drivers/{id} [get] // @Router /api/drivers/{id} [put] // @Router /api/drivers/{id} [delete] -func driversByIDHandler(svc *drivers.Service) http.HandlerFunc { +// @Router /api/drivers/{id}/device [put] +func driversByIDHandler(svc *drivers.Service, lobbySvc *lobby.Service) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { id := strings.TrimPrefix(r.URL.Path, "/api/drivers/") if id == "" { writeError(w, http.StatusBadRequest, "bad_request", "missing driver id") return } + if strings.HasSuffix(id, "/device") { + driverID := strings.TrimSuffix(id, "/device") + if driverID == "" { + writeError(w, http.StatusBadRequest, "bad_request", "missing driver id") + return + } + if r.Method != http.MethodPut { + w.Header().Set("Allow", "PUT") + writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", r.Method) + return + } + + // Read device_id from JSON body or query param + var deviceID *int + type deviceRequest struct { + DeviceID *int `json:"device_id"` + } + var req deviceRequest + if r.ContentLength > 0 { + if err := json.NewDecoder(r.Body).Decode(&req); err == nil { + deviceID = req.DeviceID + } + } + + // Fallback to query parameter if body didn't provide it + if deviceID == nil { + qVal := r.URL.Query().Get("device_id") + if qVal != "" { + if qVal == "null" { + deviceID = nil + } else if idInt, err := strconv.Atoi(qVal); err == nil { + deviceID = &idInt + } else { + writeError(w, http.StatusBadRequest, "bad_request", "invalid device_id query parameter") + return + } + } + } + + // Validate driver exists in DB + drv, err := svc.Get(r.Context(), driverID) + if err != nil { + if errors.Is(err, drivers.ErrNotFound) { + writeError(w, http.StatusNotFound, "not_found", "driver not found") + return + } + writeError(w, http.StatusInternalServerError, "internal", err.Error()) + return + } + + // Make sure driver is registered in lobby + _, _ = lobbySvc.AddDriver(driverID, drv.Name, "") + lobbySvc.SetDriverProfile(driverID, drv.Nickname, drv.AvatarURL, drv.ClanID, "") + + // Select device in lobby (handles constraints and write-through) + if err := lobbySvc.SelectDevice(driverID, deviceID); err != nil { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + + // Return updated lobby driver status + lobbyDriver, err := lobbySvc.GetDriver(driverID) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal", err.Error()) + return + } + + writeJSON(w, http.StatusOK, lobbyDriver) + return + } if strings.HasPrefix(id, "by-nick/") { nick := strings.TrimPrefix(id, "by-nick/") if nick == "" { diff --git a/server/cmd/poc-server/main.go b/server/cmd/poc-server/main.go index b39b820..fc92185 100644 --- a/server/cmd/poc-server/main.go +++ b/server/cmd/poc-server/main.go @@ -58,8 +58,8 @@ import ( "github.com/x0gp/server/internal/races" "github.com/x0gp/server/internal/races/seed" "github.com/x0gp/server/internal/realtime" - "github.com/x0gp/server/internal/transport" "github.com/x0gp/server/internal/storage/postgres" + "github.com/x0gp/server/internal/transport" ) const serverVersion = "0.1.0-poc" @@ -72,10 +72,27 @@ var ( } ) +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") + 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") ) @@ -147,6 +164,9 @@ func main() { 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() @@ -251,6 +271,9 @@ func main() { 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)) @@ -268,8 +291,10 @@ func main() { mux.HandleFunc("/api/clans", clansRouter(clansSvc)) mux.HandleFunc("/api/clans/", clansByIDHandler(clansSvc)) mux.HandleFunc("/api/drivers", driversRouter(driversSvc)) - mux.HandleFunc("/api/drivers/", driversByIDHandler(driversSvc)) - mux.HandleFunc("/ws", wsHandler(cfg, engine, hub, cat, logger)) + 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 @@ -278,7 +303,7 @@ func main() { srv := &http.Server{ Addr: cfg.HTTPAddr, - Handler: withLogging(mux, logger), + Handler: withCORS(withLogging(mux, logger)), ReadHeaderTimeout: 5 * time.Second, } @@ -418,7 +443,7 @@ func statsHandler(h *Hub, e *control.Engine) http.HandlerFunc { // @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, logger *slog.Logger) http.HandlerFunc { +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 { @@ -436,7 +461,7 @@ func wsHandler(cfg *config.Config, e *control.Engine, h *Hub, cat *catalog.Servi "remote", r.RemoteAddr, ) go writePump(client, conn, logger) - go readPump(cfg, client, conn, e, h, cat, logger) + go readPump(cfg, client, conn, e, h, cat, lobbySvc, driversSvc, videoSvc, logger) } } @@ -478,7 +503,7 @@ func writePump(c *realtime.Client, conn *websocket.Conn, logger *slog.Logger) { } } -func readPump(cfg *config.Config, c *realtime.Client, conn *websocket.Conn, e *control.Engine, h *Hub, cat *catalog.Service, logger *slog.Logger) { +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) @@ -510,13 +535,13 @@ func readPump(cfg *config.Config, c *realtime.Client, conn *websocket.Conn, e *c switch env.Type { case transport.TypeClientHello: - handleClientHello(c, e, cfg, env) + 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, env) + handleInput(c, e, lobbySvc, videoSvc, env) case transport.TypePing: handlePing(c, env) case transport.TypeChatMessage: @@ -541,7 +566,7 @@ func readPump(cfg *config.Config, c *realtime.Client, conn *websocket.Conn, e *c // Message handlers -------------------------------------------------------- -func handleClientHello(c *realtime.Client, e *control.Engine, cfg *config.Config, env *transport.Envelope) { +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, @@ -552,6 +577,21 @@ func handleClientHello(c *realtime.Client, e *control.Engine, cfg *config.Config 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) } @@ -583,7 +623,7 @@ func handleLeaveRace(c *realtime.Client, e *control.Engine, _ *transport.Envelop e.RemoveCar(c.ID) } -func handleInput(c *realtime.Client, e *control.Engine, env *transport.Envelope) { +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 @@ -602,6 +642,30 @@ func handleInput(c *realtime.Client, e *control.Engine, env *transport.Envelope) } 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, diff --git a/server/cmd/poc-server/races_handlers.go b/server/cmd/poc-server/races_handlers.go index 911010f..6b7d31c 100644 --- a/server/cmd/poc-server/races_handlers.go +++ b/server/cmd/poc-server/races_handlers.go @@ -85,7 +85,7 @@ func racesListHandler(svc *races.Service) http.HandlerFunc { writeJSON(w, http.StatusOK, transport.RaceListResponse{ Items: items, NextCursor: res.NextCursor, - Count: len(items), + Count: res.TotalCount, }) } } @@ -382,10 +382,15 @@ func racePlansListHandler(svc *races.Service) http.HandlerFunc { last := out[len(out)-1] nextCur = races.Cursor{Ms: last.StartAtMs, ID: last.ID}.Encode() } + totalCount, err := svc.CountPlans(r.Context()) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal", err.Error()) + return + } writeJSON(w, http.StatusOK, transport.RacePlanListResponse{ Items: out, NextCursor: nextCur, - Count: len(out), + Count: totalCount, }) } } diff --git a/server/cmd/poc-server/video_udp.go b/server/cmd/poc-server/video_udp.go new file mode 100644 index 0000000..d75de90 --- /dev/null +++ b/server/cmd/poc-server/video_udp.go @@ -0,0 +1,381 @@ +package main + +import ( + "bufio" + "context" + "fmt" + "log/slog" + "net" + "net/http" + "os" + "strconv" + "strings" + "sync" + "time" +) + +type deviceState struct { + addr *net.UDPAddr + latestFrame []byte + currentFrame []byte + lastSeen time.Time +} + +// UDPVideoService handles UDP video frame collection and command forwarding. +type UDPVideoService struct { + logger *slog.Logger + listenAddr string + espCmdPort int + + mu sync.RWMutex + devices map[uint8]*deviceState + + clientsMu sync.RWMutex + clients map[uint8]map[chan []byte]struct{} + + conn *net.UDPConn +} + +// NewUDPVideoService initializes a new UDPVideoService. +func NewUDPVideoService(logger *slog.Logger, listenAddr string, espCmdPort int) *UDPVideoService { + return &UDPVideoService{ + logger: logger, + listenAddr: listenAddr, + espCmdPort: espCmdPort, + devices: make(map[uint8]*deviceState), + clients: make(map[uint8]map[chan []byte]struct{}), + } +} + +// Start runs the UDP receiver loop (gracefully shut down by ctx). +// Stdin command reader runs in a separate un-tracked goroutine so it doesn't block shutdown. +func (s *UDPVideoService) Start(ctx context.Context, wg *sync.WaitGroup) { + addr, err := net.ResolveUDPAddr("udp", s.listenAddr) + if err != nil { + s.logger.Error("failed to resolve UDP address", "listenAddr", s.listenAddr, "err", err) + return + } + + conn, err := net.ListenUDP("udp", addr) + if err != nil { + s.logger.Error("failed to listen UDP", "listenAddr", s.listenAddr, "err", err) + return + } + s.conn = conn + + s.logger.Info("UDP video server started. Listening for video frames...", "listenAddr", s.listenAddr) + + // Goroutine for handling UDP receive loop + wg.Add(1) + go func() { + defer wg.Done() + defer conn.Close() + + buf := make([]byte, 65535) + + for { + select { + case <-ctx.Done(): + s.logger.Info("stopping UDP video receiver loop") + return + default: + } + + // Use a read deadline to check for context cancellation periodically + _ = conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + n, remoteAddr, err := conn.ReadFromUDP(buf) + if err != nil { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + continue + } + s.logger.Debug("UDP read error", "err", err) + continue + } + + if n < 1 { + continue + } + deviceID := buf[0] // first byte is DEVICE_ID (0..127) + payload := buf[1:n] + + s.mu.Lock() + dev, ok := s.devices[deviceID] + if !ok { + dev = &deviceState{ + addr: remoteAddr, + lastSeen: time.Now(), + } + s.devices[deviceID] = dev + s.logger.Info("ESP32 video source discovered", "device_id", deviceID, "ip", remoteAddr.IP.String()) + } else { + if dev.addr == nil || dev.addr.IP.String() != remoteAddr.IP.String() { + dev.addr = remoteAddr + s.logger.Info("ESP32 video source IP updated", "device_id", deviceID, "ip", remoteAddr.IP.String()) + } + dev.lastSeen = time.Now() + } + + // Reassemble JPEG frame for this device + if len(payload) >= 2 && payload[0] == 0xFF && payload[1] == 0xD8 { + dev.currentFrame = make([]byte, 0, 65535) + } + dev.currentFrame = append(dev.currentFrame, payload...) + + if len(payload) >= 2 && payload[len(payload)-2] == 0xFF && payload[len(payload)-1] == 0xD9 { + // Frame fully assembled + s.broadcastFrame(deviceID, dev.currentFrame) + } + s.mu.Unlock() + } + }() + + // Stdin command reader (run in background, not tracked by wg to avoid blocking shutdown) + go func() { + reader := bufio.NewReader(os.Stdin) + s.logger.Info("Console command reader active. Enter commands (e.g., '1 left', '1 start', or 'left') in console:") + + for { + text, err := reader.ReadString('\n') + if err != nil { + // Stdin closed or EOF (e.g. non-interactive/background mode) + s.logger.Debug("stdin closed, stopping console command reader", "err", err) + return + } + + text = strings.TrimSpace(text) + if text == "" { + continue + } + + // Format: "ID command" (e.g. "1 left") or "command" (defaults to the first discovered/available device) + parts := strings.SplitN(text, " ", 2) + var deviceID uint8 + var cmd string + hasDevice := false + + if len(parts) == 2 { + if idVal, err := strconv.Atoi(parts[0]); err == nil && idVal >= 0 && idVal <= 127 { + deviceID = uint8(idVal) + cmd = parts[1] + hasDevice = true + } + } + + if !hasDevice { + cmd = text + // Find the first discovered/active device + s.mu.RLock() + for id := range s.devices { + deviceID = id + hasDevice = true + break + } + s.mu.RUnlock() + } + + if !hasDevice { + s.logger.Warn("cannot send command: no ESP32 devices discovered yet") + continue + } + + err = s.SendCommand(deviceID, cmd) + if err != nil { + s.logger.Error("failed to send command to ESP32", "device_id", deviceID, "cmd", cmd, "err", err) + } else { + s.logger.Info("command sent to ESP32", "device_id", deviceID, "cmd", cmd) + } + } + }() +} + +// SendCommand sends a command string to the ESP32 representing deviceID. +func (s *UDPVideoService) SendCommand(deviceID uint8, cmd string) error { + s.mu.RLock() + dev, ok := s.devices[deviceID] + conn := s.conn + s.mu.RUnlock() + + if !ok || dev.addr == nil { + return fmt.Errorf("ESP32 device %d address unknown (no video packets received yet)", deviceID) + } + if conn == nil { + return fmt.Errorf("UDP service not started") + } + + cmdAddr := &net.UDPAddr{ + IP: dev.addr.IP, + Port: s.espCmdPort, + } + + _, err := conn.WriteToUDP([]byte(cmd), cmdAddr) + return err +} + +func (s *UDPVideoService) broadcastFrame(deviceID uint8, frame []byte) { + // dev.latestFrame is updated inside the caller's lock (Start loop) to avoid races + if dev, ok := s.devices[deviceID]; ok { + dev.latestFrame = make([]byte, len(frame)) + copy(dev.latestFrame, frame) + } + + s.clientsMu.RLock() + defer s.clientsMu.RUnlock() + deviceClients, ok := s.clients[deviceID] + if ok { + for ch := range deviceClients { + select { + case ch <- frame: + default: + // Drop frame if client is slow to read + } + } + } +} + +func (s *UDPVideoService) registerClient(deviceID uint8, ch chan []byte) { + s.clientsMu.Lock() + defer s.clientsMu.Unlock() + if s.clients[deviceID] == nil { + s.clients[deviceID] = make(map[chan []byte]struct{}) + } + s.clients[deviceID][ch] = struct{}{} +} + +func (s *UDPVideoService) unregisterClient(deviceID uint8, ch chan []byte) { + s.clientsMu.Lock() + defer s.clientsMu.Unlock() + if s.clients[deviceID] != nil { + delete(s.clients[deviceID], ch) + if len(s.clients[deviceID]) == 0 { + delete(s.clients, deviceID) + } + } +} + +// StreamHandler godoc +// @Summary Get MJPEG video stream from ESP32 +// @Description Streams video frames received via UDP as multipart/x-mixed-replace. Suitable for raw img tags in browser. +// @Tags video +// @Produce multipart/x-mixed-replace +// @Param id query int false "Device ID (0..127)" +// @Success 200 {string} string "MJPEG stream" +// @Router /api/video/stream [get] +func (s *UDPVideoService) StreamHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + idVal := r.URL.Query().Get("id") + deviceID := uint8(1) // default device ID is 1 as per client.ino + if idVal != "" { + if idInt, err := strconv.Atoi(idVal); err == nil && idInt >= 0 && idInt <= 127 { + deviceID = uint8(idInt) + } else { + http.Error(w, "invalid device id", http.StatusBadRequest) + return + } + } + + w.Header().Set("Content-Type", "multipart/x-mixed-replace; boundary=frame") + w.Header().Set("Cache-Control", "no-cache, private, max-age=0, no-store, must-revalidate") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("Pragma", "no-cache") + + frameCh := make(chan []byte, 10) + s.registerClient(deviceID, frameCh) + defer s.unregisterClient(deviceID, frameCh) + + // Send the latest frame immediately if we have one + s.mu.RLock() + var initFrame []byte + if dev, ok := s.devices[deviceID]; ok { + initFrame = dev.latestFrame + } + s.mu.RUnlock() + if len(initFrame) > 0 { + _, _ = fmt.Fprintf(w, "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: %d\r\n\r\n", len(initFrame)) + _, _ = w.Write(initFrame) + _, _ = w.Write([]byte("\r\n")) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + } + + for { + select { + case <-r.Context().Done(): + return + case frame, ok := <-frameCh: + if !ok { + return + } + _, err := fmt.Fprintf(w, "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: %d\r\n\r\n", len(frame)) + if err != nil { + return + } + _, err = w.Write(frame) + if err != nil { + return + } + _, err = w.Write([]byte("\r\n")) + if err != nil { + return + } + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + } + } + } +} + +// ControlHandler godoc +// @Summary Send command to ESP32 +// @Description Sends a text command (e.g. start, stop, left, right) to the ESP32 command port (UDP 8888). +// @Tags video +// @Param id query int false "Device ID (0..127)" +// @Param cmd query string true "Command text" +// @Produce json +// @Success 200 {object} map[string]interface{} "Command sent" +// @Failure 400 {string} string "Missing cmd parameter" +// @Failure 500 {string} string "Internal error" +// @Router /api/video/control [post] +func (s *UDPVideoService) ControlHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only POST is allowed") + return + } + cmd := r.URL.Query().Get("cmd") + if cmd == "" { + cmd = r.FormValue("cmd") + } + if cmd == "" { + writeError(w, http.StatusBadRequest, "bad_request", "Missing 'cmd' parameter") + return + } + + idVal := r.URL.Query().Get("id") + if idVal == "" { + idVal = r.FormValue("id") + } + deviceID := uint8(1) + if idVal != "" { + if idInt, err := strconv.Atoi(idVal); err == nil && idInt >= 0 && idInt <= 127 { + deviceID = uint8(idInt) + } else { + writeError(w, http.StatusBadRequest, "bad_request", "invalid device id") + return + } + } + + err := s.SendCommand(deviceID, cmd) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal_error", err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "status": "success", + "device_id": deviceID, + "command": cmd, + }) + } +} diff --git a/server/docs/docs.go b/server/docs/docs.go index 6b86914..4e08b5a 100644 --- a/server/docs/docs.go +++ b/server/docs/docs.go @@ -948,6 +948,70 @@ const docTemplate = `{ } } }, + "/api/drivers/{id}/device": { + "put": { + "description": "Dispatches on HTTP method. Path ` + "`" + `/api/drivers/by-nick/{nick}` + "`" + ` is a convenience that returns the driver by 3-letter nickname.", + "produces": [ + "application/json" + ], + "tags": [ + "drivers" + ], + "summary": "Get / update / delete a driver by id", + "parameters": [ + { + "type": "string", + "description": "Driver id", + "name": "id", + "in": "path", + "required": true + }, + { + "description": "Patch payload (only for PUT)", + "name": "driver", + "in": "body", + "schema": { + "$ref": "#/definitions/transport.DriverUpdateRequest" + } + } + ], + "responses": { + "200": { + "description": "Driver payload or update result", + "schema": { + "$ref": "#/definitions/transport.Driver" + } + }, + "204": { + "description": "Delete ack", + "schema": { + "$ref": "#/definitions/transport.Driver" + } + }, + "400": { + "description": "Bad request", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "404": { + "description": "Driver not found", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "500": { + "description": "Internal server error", + "schema": { + "type": "object", + "additionalProperties": true + } + } + } + } + }, "/api/races": { "get": { "description": "Returns one keyset page of races. Live races (status lobby|racing) come from in-memory lobby.Service; finished races come from Postgres. The two sources are merged and ordered by (started_ms DESC, id ASC). The cursor is opaque: pass back the next_cursor verbatim from a previous page.\n## Filters\n- ` + "`" + `status` + "`" + ` (comma-separated, e.g. \"racing,finished\"). Recognised values: all, lobby, racing, finished.\n- ` + "`" + `track_id` + "`" + ` (exact match against a physical track id; custom lobbies are not supported in this build).\n- ` + "`" + `limit` + "`" + ` (1..200, default 50)", @@ -1686,6 +1750,82 @@ const docTemplate = `{ } } }, + "/api/video/control": { + "post": { + "description": "Sends a text command (e.g. start, stop, left, right) to the ESP32 command port (UDP 8888).", + "produces": [ + "application/json" + ], + "tags": [ + "video" + ], + "summary": "Send command to ESP32", + "parameters": [ + { + "type": "integer", + "description": "Device ID (0..127)", + "name": "id", + "in": "query" + }, + { + "type": "string", + "description": "Command text", + "name": "cmd", + "in": "query", + "required": true + } + ], + "responses": { + "200": { + "description": "Command sent", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Missing cmd parameter", + "schema": { + "type": "string" + } + }, + "500": { + "description": "Internal error", + "schema": { + "type": "string" + } + } + } + } + }, + "/api/video/stream": { + "get": { + "description": "Streams video frames received via UDP as multipart/x-mixed-replace. Suitable for raw img tags in browser.", + "produces": [ + "multipart/x-mixed-replace" + ], + "tags": [ + "video" + ], + "summary": "Get MJPEG video stream from ESP32", + "parameters": [ + { + "type": "integer", + "description": "Device ID (0..127)", + "name": "id", + "in": "query" + } + ], + "responses": { + "200": { + "description": "MJPEG stream", + "schema": { + "type": "string" + } + } + } + } + }, "/health": { "get": { "description": "Returns 200 OK with current engine phase, WS connection count and total snapshot drops. Intended for load-balancer health checks.", diff --git a/server/docs/swagger.json b/server/docs/swagger.json index dddc91b..ea0b208 100644 --- a/server/docs/swagger.json +++ b/server/docs/swagger.json @@ -946,6 +946,70 @@ } } }, + "/api/drivers/{id}/device": { + "put": { + "description": "Dispatches on HTTP method. Path `/api/drivers/by-nick/{nick}` is a convenience that returns the driver by 3-letter nickname.", + "produces": [ + "application/json" + ], + "tags": [ + "drivers" + ], + "summary": "Get / update / delete a driver by id", + "parameters": [ + { + "type": "string", + "description": "Driver id", + "name": "id", + "in": "path", + "required": true + }, + { + "description": "Patch payload (only for PUT)", + "name": "driver", + "in": "body", + "schema": { + "$ref": "#/definitions/transport.DriverUpdateRequest" + } + } + ], + "responses": { + "200": { + "description": "Driver payload or update result", + "schema": { + "$ref": "#/definitions/transport.Driver" + } + }, + "204": { + "description": "Delete ack", + "schema": { + "$ref": "#/definitions/transport.Driver" + } + }, + "400": { + "description": "Bad request", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "404": { + "description": "Driver not found", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "500": { + "description": "Internal server error", + "schema": { + "type": "object", + "additionalProperties": true + } + } + } + } + }, "/api/races": { "get": { "description": "Returns one keyset page of races. Live races (status lobby|racing) come from in-memory lobby.Service; finished races come from Postgres. The two sources are merged and ordered by (started_ms DESC, id ASC). The cursor is opaque: pass back the next_cursor verbatim from a previous page.\n## Filters\n- `status` (comma-separated, e.g. \"racing,finished\"). Recognised values: all, lobby, racing, finished.\n- `track_id` (exact match against a physical track id; custom lobbies are not supported in this build).\n- `limit` (1..200, default 50)", @@ -1684,6 +1748,82 @@ } } }, + "/api/video/control": { + "post": { + "description": "Sends a text command (e.g. start, stop, left, right) to the ESP32 command port (UDP 8888).", + "produces": [ + "application/json" + ], + "tags": [ + "video" + ], + "summary": "Send command to ESP32", + "parameters": [ + { + "type": "integer", + "description": "Device ID (0..127)", + "name": "id", + "in": "query" + }, + { + "type": "string", + "description": "Command text", + "name": "cmd", + "in": "query", + "required": true + } + ], + "responses": { + "200": { + "description": "Command sent", + "schema": { + "type": "object", + "additionalProperties": true + } + }, + "400": { + "description": "Missing cmd parameter", + "schema": { + "type": "string" + } + }, + "500": { + "description": "Internal error", + "schema": { + "type": "string" + } + } + } + } + }, + "/api/video/stream": { + "get": { + "description": "Streams video frames received via UDP as multipart/x-mixed-replace. Suitable for raw img tags in browser.", + "produces": [ + "multipart/x-mixed-replace" + ], + "tags": [ + "video" + ], + "summary": "Get MJPEG video stream from ESP32", + "parameters": [ + { + "type": "integer", + "description": "Device ID (0..127)", + "name": "id", + "in": "query" + } + ], + "responses": { + "200": { + "description": "MJPEG stream", + "schema": { + "type": "string" + } + } + } + } + }, "/health": { "get": { "description": "Returns 200 OK with current engine phase, WS connection count and total snapshot drops. Intended for load-balancer health checks.", diff --git a/server/docs/swagger.yaml b/server/docs/swagger.yaml index f7e2764..c3848c2 100644 --- a/server/docs/swagger.yaml +++ b/server/docs/swagger.yaml @@ -1331,6 +1331,50 @@ paths: summary: Get / update / delete a driver by id tags: - drivers + /api/drivers/{id}/device: + put: + description: Dispatches on HTTP method. Path `/api/drivers/by-nick/{nick}` is + a convenience that returns the driver by 3-letter nickname. + parameters: + - description: Driver id + in: path + name: id + required: true + type: string + - description: Patch payload (only for PUT) + in: body + name: driver + schema: + $ref: '#/definitions/transport.DriverUpdateRequest' + produces: + - application/json + responses: + "200": + description: Driver payload or update result + schema: + $ref: '#/definitions/transport.Driver' + "204": + description: Delete ack + schema: + $ref: '#/definitions/transport.Driver' + "400": + description: Bad request + schema: + additionalProperties: true + type: object + "404": + description: Driver not found + schema: + additionalProperties: true + type: object + "500": + description: Internal server error + schema: + additionalProperties: true + type: object + summary: Get / update / delete a driver by id + tags: + - drivers /api/races: get: description: |- @@ -1850,6 +1894,58 @@ paths: summary: Get / update / delete a track by id tags: - tracks + /api/video/control: + post: + description: Sends a text command (e.g. start, stop, left, right) to the ESP32 + command port (UDP 8888). + parameters: + - description: Device ID (0..127) + in: query + name: id + type: integer + - description: Command text + in: query + name: cmd + required: true + type: string + produces: + - application/json + responses: + "200": + description: Command sent + schema: + additionalProperties: true + type: object + "400": + description: Missing cmd parameter + schema: + type: string + "500": + description: Internal error + schema: + type: string + summary: Send command to ESP32 + tags: + - video + /api/video/stream: + get: + description: Streams video frames received via UDP as multipart/x-mixed-replace. + Suitable for raw img tags in browser. + parameters: + - description: Device ID (0..127) + in: query + name: id + type: integer + produces: + - multipart/x-mixed-replace + responses: + "200": + description: MJPEG stream + schema: + type: string + summary: Get MJPEG video stream from ESP32 + tags: + - video /health: get: description: Returns 200 OK with current engine phase, WS connection count and diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 6183452..076b1c9 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -26,6 +26,9 @@ type Config struct { DevMode bool // If true, allow multiple clients without auth DatabaseURL string // Postgres connection string; required at runtime + + UDPVideoAddr string // UDP address to listen on for video, e.g. ":9999" + ESPCmdPort int // Port on ESP32 that listens for commands, e.g. 8888 } // Load reads configuration from environment variables, applying defaults. @@ -44,6 +47,8 @@ func Load() (*Config, error) { SnapshotRate: getEnvInt("X0GP_SNAPSHOT_RATE", 30), DevMode: getEnvBool("X0GP_DEV_MODE", true), DatabaseURL: getEnv("DATABASE_URL", ""), + UDPVideoAddr: getEnv("X0GP_UDP_VIDEO_ADDR", ":9999"), + ESPCmdPort: getEnvInt("X0GP_ESP_CMD_PORT", 8888), } if c.TickRate <= 0 || c.TickRate > 240 { diff --git a/server/internal/lobby/lobby.go b/server/internal/lobby/lobby.go index 3f70015..c7c3706 100644 --- a/server/internal/lobby/lobby.go +++ b/server/internal/lobby/lobby.go @@ -94,7 +94,9 @@ func (s *Service) CreateRace(opts CreateRaceOptions) (RaceMeta, error) { hook(meta) } s.bump("race_created", id, "") - s.mirror(func(ctx context.Context, p Persistence) error { return p.UpsertRace(ctx, meta) }) + // Synchronous mirror so the race row is in Postgres before any + // subsequent AddDriver mirrors try to attach drivers (FK). + s.mirrorSync(func(ctx context.Context, p Persistence) error { return p.UpsertRace(ctx, meta) }) return meta, nil } @@ -185,6 +187,7 @@ func (s *Service) RemoveDriver(id string) { d.Status = DriverStatusOffline d.LastSeenMs = time.Now().UnixMilli() d.RaceID = "" + d.DeviceID = nil hook := s.onRaceRemoved s.mu.Unlock() @@ -241,6 +244,10 @@ func (s *Service) AddDriverToRace(driverID, raceID string) error { s.mu.Unlock() return ErrDriverNotFound } + if d.DeviceID == nil { + s.mu.Unlock() + return fmt.Errorf("driver %s has no device selected", driverID) + } r, ok := s.races[raceID] if !ok { s.mu.Unlock() @@ -299,6 +306,7 @@ func (s *Service) RemoveDriverFromRace(driverID string) { } d.Status = DriverStatusIdle d.RaceID = "" + d.DeviceID = nil prevRaceID := d.RaceID hook := s.onRaceRemoved s.mu.Unlock() @@ -341,6 +349,7 @@ func (s *Service) SetRaceStatus(raceID string, status RaceStatus) error { if d, ok := s.drivers[did]; ok { d.Status = DriverStatusIdle d.RaceID = "" + d.DeviceID = nil affected = append(affected, *d) } } @@ -445,6 +454,44 @@ func (s *Service) GetDriver(driverID string) (DriverMeta, error) { return *d, nil } +// SelectDevice assigns a physical device ID (0..127) to a driver. +// Pass deviceID = nil to release the device. +// Returns an error if the device is already taken by another online driver. +func (s *Service) SelectDevice(driverID string, deviceID *int) error { + if deviceID != nil { + if *deviceID < 0 || *deviceID > 127 { + return fmt.Errorf("invalid device ID: must be 0..127") + } + } + + s.mu.Lock() + d, ok := s.drivers[driverID] + if !ok { + s.mu.Unlock() + return ErrDriverNotFound + } + + // Enforce that device cannot be selected if already assigned to another driver. + if deviceID != nil { + for otherID, otherD := range s.drivers { + if otherID != driverID && otherD.Status != DriverStatusOffline && otherD.DeviceID != nil && *otherD.DeviceID == *deviceID { + s.mu.Unlock() + return fmt.Errorf("device %d is already taken by driver %s", *deviceID, otherID) + } + } + } + + d.DeviceID = deviceID + out := *d + s.mu.Unlock() + + s.bump("driver_updated", "", driverID) + s.mirror(func(ctx context.Context, p Persistence) error { + return p.UpsertDriver(ctx, out) + }) + return nil +} + // ListRaces returns a snapshot copy of all races. func (s *Service) ListRaces() []RaceMeta { s.mu.RLock() diff --git a/server/internal/lobby/lobby_test.go b/server/internal/lobby/lobby_test.go index 07e4b3a..c2f0968 100644 --- a/server/internal/lobby/lobby_test.go +++ b/server/internal/lobby/lobby_test.go @@ -75,6 +75,9 @@ func TestQuickPlayJoinsFirstRace(t *testing.T) { s := NewService() _, _ = s.AddDriver("d1", "Alice", "") _, _ = s.AddDriver("d2", "Bob", "") + dev1, dev2 := 1, 2 + _ = s.SelectDevice("d1", &dev1) + _ = s.SelectDevice("d2", &dev2) r1, _ := s.QuickPlay("d1", 4) r2, err := s.QuickPlay("d2", 4) if err != nil { diff --git a/server/internal/lobby/types.go b/server/internal/lobby/types.go index 71accf9..bfbd25b 100644 --- a/server/internal/lobby/types.go +++ b/server/internal/lobby/types.go @@ -65,6 +65,7 @@ type DriverMeta struct { LastSeenMs int64 `json:"last_seen_ms"` Country string `json:"country,omitempty"` // optional, for UI AvatarHue int `json:"avatar_hue,omitempty"` // 0..360, deterministic per id + DeviceID *int `json:"device_id,omitempty"` } // Snapshot is the immutable view broadcast to clients. @@ -290,4 +291,23 @@ func (s *Service) mirror(fn func(ctx context.Context, p Persistence) error) { slog.Warn("lobby persist failed", "err", err) } }() +} + +// mirrorSync is mirror but synchronous. Used for writes whose +// successors depend on the row already being in the database (e.g. +// AddDriver must come after UpsertRace so the FK is satisfied). +// Failures are still logged but never block the caller: the +// in-memory state stays consistent even when the DB write fails. +func (s *Service) mirrorSync(fn func(ctx context.Context, p Persistence) error) { + s.mu.RLock() + p := s.persist + s.mu.RUnlock() + if p == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := fn(ctx, p); err != nil { + slog.Warn("lobby persist failed", "err", err) + } } \ No newline at end of file diff --git a/server/internal/races/seed/seed.go b/server/internal/races/seed/seed.go index d173a46..0678494 100644 --- a/server/internal/races/seed/seed.go +++ b/server/internal/races/seed/seed.go @@ -394,9 +394,9 @@ func (r *Runner) makeFinished(idx int) races.FinishedRace { } bestLapMs := int64(20_000 + rng.Intn(8_000)) - // Podium: top-3 by total time. The winner always finishes; the - // remaining drivers are sorted by a per-driver total time derived - // from the winner's best lap. + // Per-driver total time. The winner is fastest; the rest are + // winner's time + 2..16 seconds. Build the Results slice with + // computed positions. perDriver := make([]int64, len(picked)) for i, d := range picked { if d.ID == winner.ID { @@ -419,35 +419,35 @@ func (r *Runner) makeFinished(idx int) races.FinishedRace { all[j-1], all[j] = all[j], all[j-1] } } - podium := make([]races.PodiumEntry, 0, 3) - for i := 0; i < len(all) && i < 3; i++ { - podium = append(podium, races.PodiumEntry{ - Position: i + 1, - DriverID: all[i].d.ID, - Name: all[i].d.Nickname, - TotalTimeMs: all[i].time, + results := make([]races.DriverResult, 0, len(all)) + for i, row := range all { + pos := i + 1 + // Podium only for the top-3; the rest are just DNF-ish finishers + // (they have a time and a position). + results = append(results, races.DriverResult{ + DriverID: row.d.ID, + TotalTimeMs: row.time, + BestLapMs: bestLapMs, + Position: &pos, }) } return races.FinishedRace{ - ID: fmt.Sprintf("seed-finished-%03d", idx+1), - Name: fmt.Sprintf("Seed Race #%d (%s)", idx+1, track), - TrackID: track, - MaxCars: 4, - Laps: laps, - TimeLimitS: 0, - DriverIDs: driverIDs, - Status: "finished", - CreatedMs: finishedMs - randDuration(rng, 5, 120).Milliseconds(), - StartedMs: startedMs, - FinishedMs: finishedMs, - DurationMs: finishedMs - startedMs, - TotalLaps: totalLaps, - TotalDrivers: len(driverIDs), - WinnerDriverID: winner.ID, - WinnerName: winner.Nickname, - BestLapMs: bestLapMs, - Podium: podium, + ID: fmt.Sprintf("seed-finished-%03d", idx+1), + Name: fmt.Sprintf("Seed Race #%d (%s)", idx+1, track), + TrackID: track, + MaxCars: 4, + Laps: laps, + TimeLimitS: 0, + DriverIDs: driverIDs, + Status: "finished", + CreatedMs: finishedMs - randDuration(rng, 5, 120).Milliseconds(), + StartedMs: startedMs, + FinishedMs: finishedMs, + DurationMs: finishedMs - startedMs, + TotalLaps: totalLaps, + TotalDrivers: len(driverIDs), + Results: results, } } @@ -481,6 +481,8 @@ func (r *Runner) createLive(status lobby.RaceStatus, idx int) (lobby.RaceMeta, e continue } r.lb.SetDriverProfile(d.ID, d.Nickname, "", d.ClanID, d.ClanTag) + devID := i + _ = r.lb.SelectDevice(d.ID, &devID) if err := r.lb.AddDriverToRace(d.ID, meta.ID); err != nil { // ignore: race may be at capacity } diff --git a/server/internal/races/service.go b/server/internal/races/service.go index 2db58d0..a058755 100644 --- a/server/internal/races/service.go +++ b/server/internal/races/service.go @@ -62,6 +62,9 @@ func (s *Service) RestoreFromDB(ctx context.Context) (int, int, error) { // AddDriver+SetDriverProfile and accept a no-op mirror. _, _ = s.lobby.AddDriver(d.ID, d.Name, "") s.lobby.SetDriverProfile(d.ID, d.Nickname, d.AvatarURL, d.ClanID, d.ClanTag) + if d.DeviceID != nil { + _ = s.lobby.SelectDevice(d.ID, d.DeviceID) + } } for _, r := range races { // We bypass CreateRace to avoid duplicating persistence writes @@ -106,6 +109,7 @@ type RaceItem struct { type ListResult struct { Items []RaceItem NextCursor string // empty when no more pages + TotalCount int } // ListRaces returns one keyset page of races matching the filter. @@ -186,7 +190,34 @@ func (s *Service) ListRaces(ctx context.Context, f ListFilter) (ListResult, erro out = out[:limit] } - res := ListResult{Items: out} + var totalCount int + sqlStatuses := getSQLStatuses(f.Statuses) + if s.live != nil { + var err error + totalCount, err = s.pg.CountRaces(ctx, sqlStatuses, f.TrackID) + if err != nil { + return ListResult{}, err + } + } else { + // Fallback: in-memory live + pg finished + liveCount := 0 + for _, r := range s.lobby.ListRaces() { + if statusMatches(f.Statuses, string(r.Status)) && (f.TrackID == "" || r.TrackID == f.TrackID) { + liveCount++ + } + } + var finishedCount int + if statusWantsFinished(f.Statuses) { + var err error + finishedCount, err = s.pg.CountRaces(ctx, []string{"finished", "cancelled"}, f.TrackID) + if err != nil { + return ListResult{}, err + } + } + totalCount = liveCount + finishedCount + } + + res := ListResult{Items: out, TotalCount: totalCount} if hasMore && len(out) > 0 { last := out[len(out)-1] res.NextCursor = Cursor{ @@ -292,6 +323,37 @@ func filterToStatusStrings(filters []StatusFilter) []string { return out } +func getSQLStatuses(filters []StatusFilter) []string { + var sqlStatuses []string + wantLive := false + wantFinished := false + if len(filters) == 0 { + wantLive = true + wantFinished = true + } else { + for _, f := range filters { + if f == StatusAll { + wantLive = true + wantFinished = true + break + } + if f.matchesLive("finished") || f == StatusFinished { + wantFinished = true + } + if f.matchesLive("lobby") || f == StatusLobby || f == StatusRacing { + wantLive = true + } + } + } + if wantLive { + sqlStatuses = append(sqlStatuses, filterToStatusStrings(filters)...) + } + if wantFinished { + sqlStatuses = append(sqlStatuses, "finished", "cancelled") + } + return sqlStatuses +} + func (s *Service) collectFinished(ctx context.Context, filters []StatusFilter, trackID, curSource string, curMs int64, curID string, limit int) ([]RaceItem, error) { if !statusWantsFinished(filters) { return nil, nil @@ -488,8 +550,11 @@ func (s *Service) ListQueue(ctx context.Context, driverID string) ([]QueueEntry, // Finished-race snapshot hook // --------------------------------------------------------------------------- -// SnapshotFinished is the input to the "persist on finish" hook. Race is -// a copy of the in-memory meta. Result is filled by the engine. +// SnapshotFinished is the input to the "persist on finish" hook. +// The engine fills Meta, FinishedMs, TotalLaps and either: +// * Results (full per-driver totals / positions), or +// * the legacy WinnerDriverID + BestLapMs fallback (used by the +// seeder and any caller that doesn't track per-driver times). type SnapshotFinished struct { Meta lobby.RaceMeta FinishedMs int64 @@ -497,6 +562,7 @@ type SnapshotFinished struct { WinnerDriverID string WinnerName string BestLapMs int64 + Results []DriverResult } // PersistFinished upserts the finished race into Postgres. @@ -506,24 +572,37 @@ func (s *Service) PersistFinished(ctx context.Context, sf SnapshotFinished) erro if finishedMs == 0 { finishedMs = s.now().UnixMilli() } + // Build the per-driver results from the in-memory race meta and + // the snapshot. In PoC we don't have per-driver totals from the + // engine yet, so we use the snapshot's WinnerDriverID as a + // single-row fallback when Results is empty. The full per-driver + // totals will be supplied by the engine in a future iteration. + results := sf.Results + if len(results) == 0 && sf.WinnerDriverID != "" { + pos := 1 + results = []DriverResult{{ + DriverID: sf.WinnerDriverID, + TotalTimeMs: finishedMs - startedMs, + BestLapMs: sf.BestLapMs, + Position: &pos, + }} + } r := FinishedRace{ - ID: sf.Meta.ID, - Name: sf.Meta.Name, - TrackID: sf.Meta.TrackID, - MaxCars: sf.Meta.MaxCars, - Laps: sf.Meta.Laps, - TimeLimitS: sf.Meta.TimeLimitS, - DriverIDs: append([]string(nil), sf.Meta.DriverIDs...), - Status: "finished", - CreatedMs: sf.Meta.CreatedMs, - StartedMs: startedMs, - FinishedMs: finishedMs, - DurationMs: finishedMs - startedMs, - TotalLaps: sf.TotalLaps, - TotalDrivers: len(sf.Meta.DriverIDs), - WinnerDriverID: sf.WinnerDriverID, - WinnerName: sf.WinnerName, - BestLapMs: sf.BestLapMs, + ID: sf.Meta.ID, + Name: sf.Meta.Name, + TrackID: sf.Meta.TrackID, + MaxCars: sf.Meta.MaxCars, + Laps: sf.Meta.Laps, + TimeLimitS: sf.Meta.TimeLimitS, + DriverIDs: append([]string(nil), sf.Meta.DriverIDs...), + Status: "finished", + CreatedMs: sf.Meta.CreatedMs, + StartedMs: startedMs, + FinishedMs: finishedMs, + DurationMs: finishedMs - startedMs, + TotalLaps: sf.TotalLaps, + TotalDrivers: len(sf.Meta.DriverIDs), + Results: results, } return s.pg.InsertFinished(ctx, r) } @@ -596,6 +675,11 @@ func (s *Service) ListPlans(ctx context.Context, cur Cursor, limit int) ([]RaceP return s.pg.ListPlans(ctx, cur, limit) } +// CountPlans returns the total number of race plans. +func (s *Service) CountPlans(ctx context.Context) (int, error) { + return s.pg.CountPlans(ctx) +} + // DeletePlan removes a plan by id. func (s *Service) DeletePlan(ctx context.Context, id string) error { return s.pg.DeletePlan(ctx, id) diff --git a/server/internal/races/store.go b/server/internal/races/store.go index a0cbd63..8149471 100644 --- a/server/internal/races/store.go +++ b/server/internal/races/store.go @@ -2,9 +2,9 @@ package races import ( "context" - "encoding/json" "errors" "fmt" + "strings" "time" "github.com/jackc/pgx/v5" @@ -24,23 +24,44 @@ type PodiumEntry struct { TotalTimeMs int64 `json:"total_time_ms" example:"123456"` } +// DriverResult is the per-driver outcome of a race. Stored in +// race_drivers. DriverID + TotalTimeMs are required; BestLapMs is +// the driver's fastest lap; Position is 1..N when the driver +// finished, or NULL when DNF/DSQ. The aggregate podium and winner +// fields are derived from this row set on read. +type DriverResult struct { + DriverID string + TotalTimeMs int64 + BestLapMs int64 + Position *int // nil for DNF +} + // FinishedRace is the on-disk shape of a completed race. Mirrors the -// lobby.RaceMeta plus outcome fields populated by the engine at finish. +// lobby.RaceMeta plus aggregate outcome fields. Per-driver results +// (including the podium) live in race_drivers and are read into +// Results on demand. type FinishedRace struct { - ID string - Name string - TrackID string - MaxCars int - Laps int - TimeLimitS int - DriverIDs []string - Status string // "finished" | "cancelled" - CreatedMs int64 - StartedMs int64 - FinishedMs int64 - DurationMs int64 - TotalLaps int - TotalDrivers int + ID string + Name string + TrackID string + MaxCars int + Laps int + TimeLimitS int + DriverIDs []string + Status string // "finished" | "cancelled" + CreatedMs int64 + StartedMs int64 + FinishedMs int64 + DurationMs int64 + TotalLaps int + TotalDrivers int + + // Results is the per-driver race_drivers data, hydrated on read. + // Includes the winner and the podium. + Results []DriverResult + + // Derived fields (computed in scanFinished from Results). The + // public API exposes them so existing clients keep working. WinnerDriverID string WinnerName string BestLapMs int64 @@ -126,8 +147,8 @@ func (s *PgStore) UpsertLive(ctx context.Context, r lobby.RaceMeta) error { _, err := s.pool.Exec(ctx, ` INSERT INTO races (id, name, track_id, max_cars, laps, time_limit_s, status, - created_ms, started_ms, finished_ms, updated_ms, podium) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 0, $10, '[]'::jsonb) + created_ms, started_ms, finished_ms, updated_ms) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 0, $10) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, track_id = EXCLUDED.track_id, @@ -363,8 +384,8 @@ func (s *PgStore) UpsertLobbyDriver(ctx context.Context, d lobby.DriverMeta) err _, err := s.pool.Exec(ctx, ` INSERT INTO lobby_drivers (driver_id, name, nickname, avatar_url, clan_id, clan_tag, - status, current_race_id, connected_ms, last_seen_ms) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + status, current_race_id, connected_ms, last_seen_ms, device_id) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT (driver_id) DO UPDATE SET name = EXCLUDED.name, nickname = EXCLUDED.nickname, @@ -373,9 +394,10 @@ func (s *PgStore) UpsertLobbyDriver(ctx context.Context, d lobby.DriverMeta) err clan_tag = EXCLUDED.clan_tag, status = EXCLUDED.status, current_race_id = EXCLUDED.current_race_id, - last_seen_ms = EXCLUDED.last_seen_ms`, + last_seen_ms = EXCLUDED.last_seen_ms, + device_id = EXCLUDED.device_id`, d.ID, d.Name, d.Nickname, d.AvatarURL, d.ClanID, d.ClanTag, - string(d.Status), d.RaceID, d.ConnectedMs, d.LastSeenMs) + string(d.Status), d.RaceID, d.ConnectedMs, d.LastSeenMs, d.DeviceID) if err != nil { return fmt.Errorf("upsert lobby driver %s: %w", d.ID, err) } @@ -392,7 +414,7 @@ func (s *PgStore) DeleteLobbyDriver(ctx context.Context, driverID string) error func (s *PgStore) ListLobbyDrivers(ctx context.Context) ([]lobby.DriverMeta, error) { rows, err := s.pool.Query(ctx, ` SELECT driver_id, name, nickname, avatar_url, clan_id, clan_tag, - status, current_race_id, connected_ms, last_seen_ms + status, current_race_id, connected_ms, last_seen_ms, device_id FROM lobby_drivers ORDER BY last_seen_ms DESC`) if err != nil { return nil, err @@ -403,7 +425,7 @@ func (s *PgStore) ListLobbyDrivers(ctx context.Context) ([]lobby.DriverMeta, err var d lobby.DriverMeta var status string if err := rows.Scan(&d.ID, &d.Name, &d.Nickname, &d.AvatarURL, &d.ClanID, &d.ClanTag, - &status, &d.RaceID, &d.ConnectedMs, &d.LastSeenMs); err != nil { + &status, &d.RaceID, &d.ConnectedMs, &d.LastSeenMs, &d.DeviceID); err != nil { return nil, err } d.Status = lobby.DriverStatus(status) @@ -416,41 +438,66 @@ func (s *PgStore) ListLobbyDrivers(ctx context.Context) ([]lobby.DriverMeta, err // finished_races // --------------------------------------------------------------------------- -// InsertFinished upserts a finished race row. +// InsertFinished upserts a finished race row and replaces its +// race_drivers entries in a single transaction. Per-driver outcome +// (total_time_ms, best_lap_ms, position) lives on race_drivers; +// the race row itself carries only metadata. func (s *PgStore) InsertFinished(ctx context.Context, r FinishedRace) error { - podium := r.Podium - if podium == nil { - podium = []PodiumEntry{} - } - podiumJSON, err := json.Marshal(podium) - if err != nil { - return fmt.Errorf("marshal podium: %w", err) - } - _, err = s.pool.Exec(ctx, ` + _, err := s.pool.Exec(ctx, ` INSERT INTO races (id, name, track_id, max_cars, laps, time_limit_s, - driver_ids, status, created_ms, started_ms, finished_ms, duration_ms, - total_laps, total_drivers, winner_driver_id, winner_name, best_lap_ms, - podium) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18) + status, created_ms, started_ms, finished_ms, duration_ms, + total_laps, total_drivers) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, status = EXCLUDED.status, finished_ms = EXCLUDED.finished_ms, duration_ms = EXCLUDED.duration_ms, total_laps = EXCLUDED.total_laps, - total_drivers = EXCLUDED.total_drivers, - winner_driver_id = EXCLUDED.winner_driver_id, - winner_name = EXCLUDED.winner_name, - best_lap_ms = EXCLUDED.best_lap_ms, - podium = EXCLUDED.podium`, + total_drivers = EXCLUDED.total_drivers`, r.ID, r.Name, r.TrackID, r.MaxCars, r.Laps, r.TimeLimitS, - r.DriverIDs, r.Status, r.CreatedMs, r.StartedMs, r.FinishedMs, r.DurationMs, - r.TotalLaps, r.TotalDrivers, r.WinnerDriverID, r.WinnerName, r.BestLapMs, - podiumJSON) + r.Status, r.CreatedMs, r.StartedMs, r.FinishedMs, r.DurationMs, + r.TotalLaps, r.TotalDrivers) if err != nil { return fmt.Errorf("insert finished race %s: %w", r.ID, err) } + // Replace race_drivers with the provided list atomically. + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin tx: %w", err) + } + defer tx.Rollback(ctx) + if _, err := tx.Exec(ctx, `DELETE FROM race_drivers WHERE race_id = $1`, r.ID); err != nil { + return fmt.Errorf("clear race_drivers for %s: %w", r.ID, err) + } + // Build a map from driver_id -> DriverResult for lookup. + resByDriver := make(map[string]DriverResult, len(r.Results)) + for _, dr := range r.Results { + resByDriver[dr.DriverID] = dr + } + for i, did := range r.DriverIDs { + dr, ok := resByDriver[did] + _ = ok // missing result: total_time_ms=0, best_lap_ms=0, position=NULL + var totalTime, bestLap int64 + var pos *int + if ok { + totalTime = dr.TotalTimeMs + bestLap = dr.BestLapMs + pos = dr.Position + } + if _, err := tx.Exec(ctx, ` + INSERT INTO race_drivers + (race_id, driver_id, slot, joined_ms, + total_time_ms, best_lap_ms, position) + VALUES ($1, $2, $3, $4, $5, $6, $7)`, + r.ID, did, i, r.CreatedMs, totalTime, bestLap, pos); err != nil { + return fmt.Errorf("insert race_drivers %s: %w", r.ID, err) + } + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit race_drivers: %w", err) + } return nil } @@ -474,9 +521,8 @@ func (s *PgStore) ListFinished(ctx context.Context, trackID string, cur Cursor, } args = append(args, limit+1) q := `SELECT id, name, track_id, max_cars, laps, time_limit_s, - driver_ids, status, created_ms, started_ms, finished_ms, duration_ms, - total_laps, total_drivers, winner_driver_id, winner_name, best_lap_ms, - podium + status, created_ms, started_ms, finished_ms, duration_ms, + total_laps, total_drivers FROM races WHERE ` + joinAnd(conds) + ` ORDER BY finished_ms DESC, id DESC @@ -486,36 +532,130 @@ func (s *PgStore) ListFinished(ctx context.Context, trackID string, cur Cursor, return nil, fmt.Errorf("list finished: %w", err) } defer rows.Close() - return scanFinished(rows) + return s.scanFinished(ctx, rows) } -func scanFinished(rows pgx.Rows) ([]FinishedRace, error) { +// CountRaces returns the total number of races matching the status filter and trackID. +func (s *PgStore) CountRaces(ctx context.Context, statuses []string, trackID string) (int, error) { + args := []any{} + conds := []string{"1=1"} + if len(statuses) > 0 { + placeholders := make([]string, len(statuses)) + for i, st := range statuses { + args = append(args, st) + placeholders[i] = fmt.Sprintf("$%d", len(args)) + } + conds = append(conds, fmt.Sprintf("status IN (%s)", strings.Join(placeholders, ","))) + } + if trackID != "" { + args = append(args, trackID) + conds = append(conds, fmt.Sprintf("track_id = $%d", len(args))) + } + q := `SELECT COUNT(*) FROM races WHERE ` + joinAnd(conds) + var count int + err := s.pool.QueryRow(ctx, q, args...).Scan(&count) + if err != nil { + return 0, fmt.Errorf("count races: %w", err) + } + return count, nil +} + +// scanFinished reads the rows from the SELECT above, then hydrates +// DriverIDs and the per-driver Results from race_drivers. The +// Winner* and Podium fields are derived from Results. +func (s *PgStore) scanFinished(ctx context.Context, rows pgx.Rows) ([]FinishedRace, error) { out := make([]FinishedRace, 0) for rows.Next() { var r FinishedRace - var podiumJSON []byte if err := rows.Scan( &r.ID, &r.Name, &r.TrackID, - &r.MaxCars, &r.Laps, &r.TimeLimitS, &r.DriverIDs, &r.Status, + &r.MaxCars, &r.Laps, &r.TimeLimitS, &r.Status, &r.CreatedMs, &r.StartedMs, &r.FinishedMs, &r.DurationMs, - &r.TotalLaps, &r.TotalDrivers, &r.WinnerDriverID, &r.WinnerName, &r.BestLapMs, - &podiumJSON, + &r.TotalLaps, &r.TotalDrivers, ); err != nil { return nil, err } - if len(podiumJSON) > 0 { - if err := json.Unmarshal(podiumJSON, &r.Podium); err != nil { - return nil, fmt.Errorf("unmarshal podium: %w", err) - } - } - if r.Podium == nil { - r.Podium = []PodiumEntry{} + if err := s.hydrateFinished(ctx, &r); err != nil { + return nil, err } out = append(out, r) } return out, rows.Err() } +// hydrateFinished pulls race_drivers for the given race into Results +// and derives the Winner* and Podium fields. +func (s *PgStore) hydrateFinished(ctx context.Context, r *FinishedRace) error { + rows, err := s.pool.Query(ctx, ` + SELECT driver_id, total_time_ms, best_lap_ms, position + FROM race_drivers + WHERE race_id = $1 + ORDER BY slot ASC, joined_ms ASC`, r.ID) + if err != nil { + return fmt.Errorf("hydrate race_drivers %s: %w", r.ID, err) + } + defer rows.Close() + + r.Results = make([]DriverResult, 0) + r.DriverIDs = make([]string, 0) + r.Podium = make([]PodiumEntry, 0) + for rows.Next() { + var dr DriverResult + var pos *int + if err := rows.Scan(&dr.DriverID, &dr.TotalTimeMs, &dr.BestLapMs, &pos); err != nil { + return err + } + dr.Position = pos + r.Results = append(r.Results, dr) + r.DriverIDs = append(r.DriverIDs, dr.DriverID) + } + if err := rows.Err(); err != nil { + return err + } + // Derive winner (lowest total_time_ms among rows with time > 0). + var bestTotal int64 = -1 + for _, dr := range r.Results { + if dr.TotalTimeMs <= 0 { + continue + } + if bestTotal == -1 || dr.TotalTimeMs < bestTotal { + bestTotal = dr.TotalTimeMs + r.WinnerDriverID = dr.DriverID + r.WinnerName = dr.DriverID + } + if dr.BestLapMs > 0 && (r.BestLapMs == 0 || dr.BestLapMs < r.BestLapMs) { + r.BestLapMs = dr.BestLapMs + } + } + // Derive podium: top-3 drivers by total_time_ms (> 0) ASC. + type row struct { + id string + t int64 + best int64 + } + var sorted []row + for _, dr := range r.Results { + if dr.TotalTimeMs <= 0 { + continue + } + sorted = append(sorted, row{dr.DriverID, dr.TotalTimeMs, dr.BestLapMs}) + } + for i := 1; i < len(sorted); i++ { + for j := i; j > 0 && sorted[j-1].t > sorted[j].t; j-- { + sorted[j-1], sorted[j] = sorted[j], sorted[j-1] + } + } + for i := 0; i < len(sorted) && i < 3; i++ { + r.Podium = append(r.Podium, PodiumEntry{ + Position: i + 1, + DriverID: sorted[i].id, + Name: sorted[i].id, + TotalTimeMs: sorted[i].t, + }) + } + return nil +} + func joinAnd(parts []string) string { out := "" for i, p := range parts { @@ -624,6 +764,16 @@ func scanPlans(rows pgx.Rows) ([]RacePlan, error) { return out, rows.Err() } +// CountPlans returns the total number of race plans. +func (s *PgStore) CountPlans(ctx context.Context) (int, error) { + var count int + err := s.pool.QueryRow(ctx, "SELECT COUNT(*) FROM race_plans").Scan(&count) + if err != nil { + return 0, fmt.Errorf("count plans: %w", err) + } + return count, nil +} + // DeletePlan removes a plan by id. func (s *PgStore) DeletePlan(ctx context.Context, id string) error { tag, err := s.pool.Exec(ctx, `DELETE FROM race_plans WHERE id = $1`, id) diff --git a/server/internal/realtime/hub.go b/server/internal/realtime/hub.go index 0db9b03..f717af4 100644 --- a/server/internal/realtime/hub.go +++ b/server/internal/realtime/hub.go @@ -13,11 +13,13 @@ import ( // Client is a single connected WebSocket client. type Client struct { - ID string - Send chan []byte - SessionID string - Done chan struct{} - once sync.Once + ID string + Send chan []byte + SessionID string + Done chan struct{} + once sync.Once + DriverID string + LastUDPCommand string } // Close signals the read/write pumps to exit and frees the send channel. diff --git a/server/internal/storage/postgres/migrations/011_drop_driver_ids.sql b/server/internal/storage/postgres/migrations/011_drop_driver_ids.sql new file mode 100644 index 0000000..cd3e3af --- /dev/null +++ b/server/internal/storage/postgres/migrations/011_drop_driver_ids.sql @@ -0,0 +1,11 @@ +-- 011_drop_driver_ids.sql — remove the driver_ids array column from +-- the unified races table. +-- +-- The authoritative source for "which drivers are in this race" is +-- the race_drivers join table. The denormalised driver_ids TEXT[] was +-- redundant (and racy — the mirror goroutines could write one side +-- without the other). After this migration the only place where the +-- driver list lives is race_drivers, and the PgStore methods hydrate +-- it on read. + +ALTER TABLE races DROP COLUMN IF EXISTS driver_ids; diff --git a/server/internal/storage/postgres/migrations/012_normalize_race_results.sql b/server/internal/storage/postgres/migrations/012_normalize_race_results.sql new file mode 100644 index 0000000..8f47b87 --- /dev/null +++ b/server/internal/storage/postgres/migrations/012_normalize_race_results.sql @@ -0,0 +1,45 @@ +-- 012_normalize_race_results.sql — move per-driver race outcomes +-- (total time, best lap, finishing position) from races into +-- race_drivers. The races table is left with only the metadata that +-- applies to the race as a whole; everything that depends on a +-- specific driver now lives in the join table. +-- +-- Drop from races: +-- * podium (JSONB) — derived from race_drivers ordered by +-- total_time_ms ASC, top-3. +-- * winner_driver_id (TEXT) — first row by total_time_ms. +-- * winner_name (TEXT) — derived. +-- * best_lap_ms (BIGINT) — moved to race_drivers as the +-- MIN(best_lap_ms) over all rows in the race +-- (or stored on each row from the engine; +-- the aggregate is the same). +-- +-- Add to race_drivers: +-- * total_time_ms BIGINT NOT NULL DEFAULT 0 (the driver's total race time) +-- * best_lap_ms BIGINT NOT NULL DEFAULT 0 (the driver's fastest lap) +-- * position INT (1..N, NULL when DNF/DSQ) +-- +-- Note: race_drivers is shared by the live surface too. For live +-- races total_time_ms and best_lap_ms stay 0 (default) and position +-- stays NULL. The semantics of `position` is "final position when +-- the race finished"; the engine writes it on the finishing tick. + +ALTER TABLE races DROP COLUMN IF EXISTS podium; +ALTER TABLE races DROP COLUMN IF EXISTS winner_driver_id; +ALTER TABLE races DROP COLUMN IF EXISTS winner_name; +ALTER TABLE races DROP COLUMN IF EXISTS best_lap_ms; + +DROP INDEX IF EXISTS finished_races_podium_gin; + +ALTER TABLE race_drivers + ADD COLUMN IF NOT EXISTS total_time_ms BIGINT NOT NULL DEFAULT 0; +ALTER TABLE race_drivers + ADD COLUMN IF NOT EXISTS best_lap_ms BIGINT NOT NULL DEFAULT 0; +ALTER TABLE race_drivers + ADD COLUMN IF NOT EXISTS position INT; + +-- Keyset index over the join table for podium queries: "top-3 by +-- total_time_ms within this race" is the canonical podium lookup. +CREATE INDEX IF NOT EXISTS race_drivers_podium_idx + ON race_drivers (race_id, total_time_ms ASC, position ASC) + WHERE total_time_ms > 0; diff --git a/server/internal/storage/postgres/migrations/013_driver_device.sql b/server/internal/storage/postgres/migrations/013_driver_device.sql new file mode 100644 index 0000000..29afa4e --- /dev/null +++ b/server/internal/storage/postgres/migrations/013_driver_device.sql @@ -0,0 +1,6 @@ +-- 013_driver_device.sql — add device_id to lobby_drivers. +-- +-- Represents the currently selected physical ESP32 DEVICE_ID (0..127) for a driver. +-- This device is pinned to the driver during races and released when the race ends or they leave. + +ALTER TABLE lobby_drivers ADD COLUMN IF NOT EXISTS device_id INT; diff --git a/server/internal/transport/codec.go b/server/internal/transport/codec.go index 92bfe60..820a47f 100644 --- a/server/internal/transport/codec.go +++ b/server/internal/transport/codec.go @@ -241,6 +241,7 @@ type LobbyDriver struct { LastSeenMs int64 `json:"last_seen_ms"` Country string `json:"country,omitempty"` AvatarHue int `json:"avatar_hue,omitempty"` + DeviceID *int `json:"device_id,omitempty"` } // LobbySnapshot is the full lobby state, broadcast on change.