package races import ( "context" "errors" "fmt" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" "github.com/x0gp/server/internal/lobby" ) // PodiumEntry is one row in the top-3 of a finished race. Ordered by // Position (1..3). DriverID/Name are denormalised from the future // drivers table; TotalTimeMs is the driver's total race time. type PodiumEntry struct { Position int `json:"position" example:"1"` DriverID string `json:"driver_id" example:"driver-alice"` Name string `json:"name" example:"driver-alice"` TotalTimeMs int64 `json:"total_time_ms" example:"123456"` Nickname string `json:"nickname" example:"driver-alice"` AvatarUrl string `json:"avatar_url"` ClanID string `json:"clan_id" example:"clan-bob"` } // 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 Name string TotalTimeMs int64 BestLapMs int64 Position *int // nil for DNF Nickname string `json:"nickname" example:"driver-alice"` AvatarUrl string `json:"avatar_url"` ClanID string `json:"clan_id" example:"clan-bob"` } // FinishedRace is the on-disk shape of a completed race. Mirrors the // 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 // 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 Podium []PodiumEntry } // RacePlan is a recurring or one-off scheduled race description. type RacePlan struct { ID string Name string TrackID string MaxCars int Laps int TimeLimitS int StartAtMs int64 IntervalS int Count int Enabled bool CreatedMs int64 UpdatedMs int64 NextFireMs int64 FiresDone int } // QueueEntry is a single (driver, race) subscription. type QueueEntry struct { DriverID string RaceID string PlanID string EnqueuedMs int64 } // PgStore is the Postgres-backed half of the races package. type PgStore struct { pool *pgxpool.Pool } // NewPgStore wraps an open pgxpool. func NewPgStore(pool *pgxpool.Pool) *PgStore { return &PgStore{pool: pool} } // Exec exposes a raw Exec for the seeder / maintenance scripts. func (s *PgStore) Exec(ctx context.Context, sql string, args ...any) (int64, error) { tag, err := s.pool.Exec(ctx, sql, args...) if err != nil { return 0, err } return tag.RowsAffected(), nil } // ListTrackIDs returns all track ids from the tracks table. Used by // the seeder to bind race_plans/finished_races to real catalog entries. func (s *PgStore) ListTrackIDs(ctx context.Context) ([]string, error) { rows, err := s.pool.Query(ctx, `SELECT id FROM tracks ORDER BY id`) if err != nil { return nil, fmt.Errorf("list track ids: %w", err) } defer rows.Close() out := make([]string, 0) for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return nil, err } out = append(out, id) } return out, rows.Err() } // --------------------------------------------------------------------------- // Live race surface (status in lobby | countdown | racing). // // The unified `races` table holds both live and finished rows. Methods // below are the write-side mirror for lobby.Service; reads flow through // races.Service via ListRacesPaged (status filter) and ListUpcoming. // --------------------------------------------------------------------------- // UpsertLive mirrors a lobby.RaceMeta row. Only used for status in // (lobby, countdown, racing). Status is enforced at the call site. func (s *PgStore) UpsertLive(ctx context.Context, r lobby.RaceMeta) error { now := time.Now().UnixMilli() _, 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) 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, max_cars = EXCLUDED.max_cars, laps = EXCLUDED.laps, time_limit_s = EXCLUDED.time_limit_s, status = EXCLUDED.status, started_ms = EXCLUDED.started_ms, updated_ms = EXCLUDED.updated_ms`, r.ID, r.Name, r.TrackID, r.MaxCars, r.Laps, r.TimeLimitS, string(r.Status), r.CreatedMs, r.StartedMs, now) if err != nil { return fmt.Errorf("upsert live race %s: %w", r.ID, err) } return nil } // DeleteLive removes a live race row. The FK from race_drivers has // CASCADE so the related drivers are also removed. func (s *PgStore) DeleteLive(ctx context.Context, raceID string) error { _, err := s.pool.Exec(ctx, `DELETE FROM races WHERE id = $1 AND status IN ('lobby','countdown','racing')`, raceID) return err } // AddLiveDriver adds a driver to a live race (slot=position). Idempotent. func (s *PgStore) AddLiveDriver(ctx context.Context, raceID, driverID string, slot int) error { now := time.Now().UnixMilli() _, err := s.pool.Exec(ctx, ` INSERT INTO race_drivers (race_id, driver_id, slot, joined_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (race_id, driver_id) DO UPDATE SET slot = EXCLUDED.slot`, raceID, driverID, slot, now) if err != nil { return fmt.Errorf("add live driver %s to %s: %w", driverID, raceID, err) } return nil } // RemoveLiveDriver removes a driver from a live race. func (s *PgStore) RemoveLiveDriver(ctx context.Context, raceID, driverID string) error { _, err := s.pool.Exec(ctx, `DELETE FROM race_drivers WHERE race_id = $1 AND driver_id = $2`, raceID, driverID) return err } // SetLiveStatus updates the status (and started_ms) of a live race. func (s *PgStore) SetLiveStatus(ctx context.Context, raceID string, status lobby.RaceStatus, startedMs int64) error { now := time.Now().UnixMilli() _, err := s.pool.Exec(ctx, ` UPDATE races SET status = $2, started_ms = $3, updated_ms = $4 WHERE id = $1 AND status IN ('lobby','countdown','racing')`, raceID, string(status), startedMs, now) return err } // ListLivePaged returns one keyset page of live races. func (s *PgStore) ListLivePaged(ctx context.Context, statuses []string, trackID string, cur Cursor, limit int) ([]lobby.RaceMeta, error) { if limit <= 0 { limit = 50 } args := []any{} conds := []string{"status IN ('lobby','countdown','racing')"} if len(statuses) > 0 { args = append(args, statuses) conds = append(conds, fmt.Sprintf("status = ANY($%d)", len(args))) } if trackID != "" { args = append(args, trackID) conds = append(conds, fmt.Sprintf("track_id = $%d", len(args))) } if !cur.IsZero() { args = append(args, cur.Ms, cur.ID) conds = append(conds, fmt.Sprintf("(COALESCE(started_ms, created_ms), id) < ($%d, $%d)", len(args)-1, len(args))) } args = append(args, limit+1) q := `SELECT id, name, track_id, max_cars, laps, time_limit_s, status, created_ms, started_ms FROM races WHERE ` + joinAnd(conds) + ` ORDER BY COALESCE(started_ms, created_ms) DESC, id ASC LIMIT $` + fmt.Sprintf("%d", len(args)) rows, err := s.pool.Query(ctx, q, args...) if err != nil { return nil, err } defer rows.Close() out := make([]lobby.RaceMeta, 0) for rows.Next() { var r lobby.RaceMeta var status string if err := rows.Scan(&r.ID, &r.Name, &r.TrackID, &r.MaxCars, &r.Laps, &r.TimeLimitS, &status, &r.CreatedMs, &r.StartedMs); err != nil { return nil, err } r.Status = lobby.RaceStatus(status) out = append(out, r) } if err := rows.Err(); err != nil { return nil, err } for i := range out { ids, err := s.listDriverIDs(ctx, out[i].ID) if err != nil { return nil, err } out[i].DriverIDs = ids } return out, nil } // ListLiveUpcoming returns up to limit live races in (lobby, countdown) // ordered soonest first. func (s *PgStore) ListLiveUpcoming(ctx context.Context, limit int) ([]lobby.RaceMeta, error) { if limit <= 0 { limit = 3 } rows, err := s.pool.Query(ctx, ` SELECT id, name, track_id, max_cars, laps, time_limit_s, status, created_ms, started_ms FROM races WHERE status IN ('lobby', 'countdown') ORDER BY COALESCE(started_ms, created_ms) ASC, id ASC LIMIT $1`, limit+1) if err != nil { return nil, err } defer rows.Close() out := make([]lobby.RaceMeta, 0) for rows.Next() { var r lobby.RaceMeta var status string if err := rows.Scan(&r.ID, &r.Name, &r.TrackID, &r.MaxCars, &r.Laps, &r.TimeLimitS, &status, &r.CreatedMs, &r.StartedMs); err != nil { return nil, err } r.Status = lobby.RaceStatus(status) out = append(out, r) } if err := rows.Err(); err != nil { return nil, err } for i := range out { ids, err := s.listDriverIDs(ctx, out[i].ID) if err != nil { return nil, err } out[i].DriverIDs = ids } return out, nil } // GetLive fetches a single live race. func (s *PgStore) GetLive(ctx context.Context, id string) (lobby.RaceMeta, error) { row := s.pool.QueryRow(ctx, ` SELECT id, name, track_id, max_cars, laps, time_limit_s, status, created_ms, started_ms FROM races WHERE id = $1 AND status IN ('lobby','countdown','racing')`, id) var r lobby.RaceMeta var status string if err := row.Scan(&r.ID, &r.Name, &r.TrackID, &r.MaxCars, &r.Laps, &r.TimeLimitS, &status, &r.CreatedMs, &r.StartedMs); err != nil { return lobby.RaceMeta{}, err } r.Status = lobby.RaceStatus(status) ids, err := s.listDriverIDs(ctx, id) if err != nil { return lobby.RaceMeta{}, err } r.DriverIDs = ids return r, nil } // ListAllRaces returns every row in the races table (live+finished). // Used by RestoreFromDB to rehydrate the lobby. func (s *PgStore) ListAllRaces(ctx context.Context) ([]lobby.RaceMeta, error) { rows, err := s.pool.Query(ctx, ` SELECT id, name, track_id, max_cars, laps, time_limit_s, status, created_ms, started_ms FROM races WHERE status IN ('lobby','countdown','racing') ORDER BY created_ms ASC`) if err != nil { return nil, err } defer rows.Close() out := make([]lobby.RaceMeta, 0) for rows.Next() { var r lobby.RaceMeta var status string if err := rows.Scan(&r.ID, &r.Name, &r.TrackID, &r.MaxCars, &r.Laps, &r.TimeLimitS, &status, &r.CreatedMs, &r.StartedMs); err != nil { return nil, err } r.Status = lobby.RaceStatus(status) out = append(out, r) } if err := rows.Err(); err != nil { return nil, err } for i := range out { ids, err := s.listDriverIDs(ctx, out[i].ID) if err != nil { return nil, err } out[i].DriverIDs = ids } return out, nil } func (s *PgStore) listDriverIDs(ctx context.Context, raceID string) ([]string, error) { rows, err := s.pool.Query(ctx, `SELECT driver_id FROM race_drivers WHERE race_id = $1 ORDER BY slot ASC, joined_ms ASC`, raceID) if err != nil { return nil, err } defer rows.Close() out := make([]string, 0) for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return nil, err } out = append(out, id) } return out, rows.Err() } // UpsertLobbyDriver mirrors a lobby.DriverMeta row. func (s *PgStore) UpsertLobbyDriver(ctx context.Context, d lobby.DriverMeta) error { _, 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, car_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, avatar_url = EXCLUDED.avatar_url, clan_id = EXCLUDED.clan_id, clan_tag = EXCLUDED.clan_tag, status = EXCLUDED.status, current_race_id = EXCLUDED.current_race_id, last_seen_ms = EXCLUDED.last_seen_ms, car_id = EXCLUDED.car_id`, d.ID, d.Name, d.Nickname, d.AvatarURL, d.ClanID, d.ClanTag, string(d.Status), d.RaceID, d.ConnectedMs, d.LastSeenMs, d.CarID) if err != nil { return fmt.Errorf("upsert lobby driver %s: %w", d.ID, err) } return nil } // DeleteLobbyDriver removes a lobby driver row. func (s *PgStore) DeleteLobbyDriver(ctx context.Context, driverID string) error { _, err := s.pool.Exec(ctx, `DELETE FROM lobby_drivers WHERE driver_id = $1`, driverID) return err } // ListLobbyDrivers returns all lobby drivers. 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, car_id FROM lobby_drivers ORDER BY last_seen_ms DESC`) if err != nil { return nil, err } defer rows.Close() out := make([]lobby.DriverMeta, 0) for rows.Next() { 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, &d.CarID); err != nil { return nil, err } d.Status = lobby.DriverStatus(status) out = append(out, d) } return out, rows.Err() } // --------------------------------------------------------------------------- // finished_races // --------------------------------------------------------------------------- // 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 { _, 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, 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`, r.ID, r.Name, r.TrackID, r.MaxCars, r.Laps, r.TimeLimitS, 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 } // ListFinished returns finished races matching the filter using keyset // pagination on (finished_ms DESC, id DESC). The cursor's Ms is the // last finished_ms from the previous page; rows strictly older are // returned. Tie-breaker on id. func (s *PgStore) ListFinished(ctx context.Context, trackID string, cur Cursor, limit int) ([]FinishedRace, error) { if limit <= 0 { limit = 50 } args := []any{} conds := []string{"status IN ('finished','cancelled')"} if !cur.IsZero() { args = append(args, cur.Ms, cur.ID) conds = append(conds, fmt.Sprintf("(finished_ms, id) < ($%d, $%d)", len(args)-1, len(args))) } if trackID != "" { args = append(args, trackID) conds = append(conds, fmt.Sprintf("track_id = $%d", len(args))) } args = append(args, limit+1) q := `SELECT id, name, track_id, max_cars, laps, time_limit_s, 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 LIMIT $` + fmt.Sprintf("%d", len(args)) rows, err := s.pool.Query(ctx, q, args...) if err != nil { return nil, fmt.Errorf("list finished: %w", err) } defer rows.Close() return s.scanFinished(ctx, rows) } // 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 if err := rows.Scan( &r.ID, &r.Name, &r.TrackID, &r.MaxCars, &r.Laps, &r.TimeLimitS, &r.Status, &r.CreatedMs, &r.StartedMs, &r.FinishedMs, &r.DurationMs, &r.TotalLaps, &r.TotalDrivers, ); err != nil { return nil, err } 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 rd.driver_id, rd.total_time_ms, rd.best_lap_ms, rd.position, d.name, d.nickname, d.avatar_url, d.clan_id FROM race_drivers rd left join drivers d on rd.driver_id = d.id 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 var name, nick, avatar, clanID *string if err := rows.Scan( &dr.DriverID, &dr.TotalTimeMs, &dr.BestLapMs, &pos, &name, &nick, &avatar, &clanID, ); err != nil { return err } dr.Position = pos if name != nil { dr.Name = *name } else { dr.Name = dr.DriverID } if nick != nil { dr.Nickname = *nick } if avatar != nil { dr.AvatarUrl = *avatar } if clanID != nil { dr.ClanID = *clanID } 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.Name } 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 name string nick string avatar string clanID string } var sorted []row for _, dr := range r.Results { if dr.TotalTimeMs <= 0 { continue } sorted = append(sorted, row{ id: dr.DriverID, t: dr.TotalTimeMs, best: dr.BestLapMs, name: dr.Name, nick: dr.Nickname, avatar: dr.AvatarUrl, clanID: dr.ClanID, }) } 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].name, TotalTimeMs: sorted[i].t, Nickname: sorted[i].nick, AvatarUrl: sorted[i].avatar, ClanID: sorted[i].clanID, }) } return nil } func joinAnd(parts []string) string { out := "" for i, p := range parts { if i > 0 { out += " AND " } out += p } return out } // --------------------------------------------------------------------------- // race_plans // --------------------------------------------------------------------------- // CreatePlan inserts a new race plan. func (s *PgStore) CreatePlan(ctx context.Context, p RacePlan) error { now := time.Now().UnixMilli() if p.CreatedMs == 0 { p.CreatedMs = now } p.UpdatedMs = now if p.NextFireMs == 0 { p.NextFireMs = p.StartAtMs } _, err := s.pool.Exec(ctx, ` INSERT INTO race_plans (id, name, track_id, max_cars, laps, time_limit_s, start_at_ms, interval_s, count, enabled, created_ms, updated_ms, next_fire_ms, fires_done) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)`, p.ID, p.Name, p.TrackID, p.MaxCars, p.Laps, p.TimeLimitS, p.StartAtMs, p.IntervalS, p.Count, p.Enabled, p.CreatedMs, p.UpdatedMs, p.NextFireMs, p.FiresDone) if err != nil { var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == "23505" { return fmt.Errorf("%w: plan %s", ErrAlreadyExists, p.ID) } return fmt.Errorf("insert plan: %w", err) } return nil } // GetPlan loads a plan by id. func (s *PgStore) GetPlan(ctx context.Context, id string) (RacePlan, error) { row := s.pool.QueryRow(ctx, ` SELECT id, name, track_id, max_cars, laps, time_limit_s, start_at_ms, interval_s, count, enabled, created_ms, updated_ms, next_fire_ms, fires_done FROM race_plans WHERE id = $1`, id) var p RacePlan if err := row.Scan( &p.ID, &p.Name, &p.TrackID, &p.MaxCars, &p.Laps, &p.TimeLimitS, &p.StartAtMs, &p.IntervalS, &p.Count, &p.Enabled, &p.CreatedMs, &p.UpdatedMs, &p.NextFireMs, &p.FiresDone, ); err != nil { if errors.Is(err, pgx.ErrNoRows) { return RacePlan{}, ErrNotFound } return RacePlan{}, err } return p, nil } // ListPlans returns plans in ascending start order, keyset-paginated. func (s *PgStore) ListPlans(ctx context.Context, cur Cursor, limit int) ([]RacePlan, error) { if limit <= 0 { limit = 50 } args := []any{} conds := []string{"1=1"} if !cur.IsZero() { args = append(args, cur.Ms, cur.ID) conds = append(conds, fmt.Sprintf("(start_at_ms, id) > ($%d, $%d)", len(args)-1, len(args))) } args = append(args, limit+1) q := `SELECT id, name, track_id, max_cars, laps, time_limit_s, start_at_ms, interval_s, count, enabled, created_ms, updated_ms, next_fire_ms, fires_done FROM race_plans WHERE ` + joinAnd(conds) + ` ORDER BY start_at_ms ASC, id ASC LIMIT $` + fmt.Sprintf("%d", len(args)) rows, err := s.pool.Query(ctx, q, args...) if err != nil { return nil, fmt.Errorf("list plans: %w", err) } defer rows.Close() return scanPlans(rows) } func scanPlans(rows pgx.Rows) ([]RacePlan, error) { out := make([]RacePlan, 0) for rows.Next() { var p RacePlan if err := rows.Scan( &p.ID, &p.Name, &p.TrackID, &p.MaxCars, &p.Laps, &p.TimeLimitS, &p.StartAtMs, &p.IntervalS, &p.Count, &p.Enabled, &p.CreatedMs, &p.UpdatedMs, &p.NextFireMs, &p.FiresDone, ); err != nil { return nil, err } out = append(out, p) } 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) if err != nil { return fmt.Errorf("delete plan: %w", err) } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } // SetPlanEnabled toggles the enabled flag. func (s *PgStore) SetPlanEnabled(ctx context.Context, id string, enabled bool) error { tag, err := s.pool.Exec(ctx, `UPDATE race_plans SET enabled = $2, updated_ms = $3 WHERE id = $1`, id, enabled, time.Now().UnixMilli()) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } // AdvancePlan moves the next_fire_ms cursor after a successful materialise // and increments fires_done. If count is reached, disables the plan. func (s *PgStore) AdvancePlan(ctx context.Context, id string, nextFireMs int64) error { _, err := s.pool.Exec(ctx, ` UPDATE race_plans SET next_fire_ms = $2, fires_done = fires_done + 1, updated_ms = $3, enabled = CASE WHEN count > 0 AND fires_done + 1 >= count THEN FALSE ELSE enabled END WHERE id = $1`, id, nextFireMs, time.Now().UnixMilli()) return err } // DuePlans returns plans whose next_fire_ms is <= now. func (s *PgStore) DuePlans(ctx context.Context, nowMs int64, limit int) ([]RacePlan, error) { if limit <= 0 { limit = 16 } rows, err := s.pool.Query(ctx, ` SELECT id, name, track_id, max_cars, laps, time_limit_s, start_at_ms, interval_s, count, enabled, created_ms, updated_ms, next_fire_ms, fires_done FROM race_plans WHERE enabled AND next_fire_ms <= $1 ORDER BY next_fire_ms ASC LIMIT $2`, nowMs, limit) if err != nil { return nil, err } defer rows.Close() return scanPlans(rows) } // --------------------------------------------------------------------------- // queue // --------------------------------------------------------------------------- // Enqueue adds (driver, race) to the queue. Idempotent: returns the existing // entry if present. func (s *PgStore) Enqueue(ctx context.Context, driverID, raceID, planID string) (QueueEntry, error) { if driverID == "" || raceID == "" { return QueueEntry{}, fmt.Errorf("%w: driver_id and race_id required", ErrInvalidInput) } now := time.Now().UnixMilli() _, err := s.pool.Exec(ctx, ` INSERT INTO race_queue (driver_id, race_id, plan_id, enqueued_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (driver_id, race_id) DO NOTHING`, driverID, raceID, planID, now) if err != nil { return QueueEntry{}, fmt.Errorf("enqueue: %w", err) } row := s.pool.QueryRow(ctx, ` SELECT driver_id, race_id, plan_id, enqueued_ms FROM race_queue WHERE driver_id = $1 AND race_id = $2`, driverID, raceID) var q QueueEntry if err := row.Scan(&q.DriverID, &q.RaceID, &q.PlanID, &q.EnqueuedMs); err != nil { return QueueEntry{}, err } return q, nil } // Dequeue removes (driver, race) entry. func (s *PgStore) Dequeue(ctx context.Context, driverID, raceID string) error { tag, err := s.pool.Exec(ctx, `DELETE FROM race_queue WHERE driver_id = $1 AND race_id = $2`, driverID, raceID) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } // ListQueueByDriver returns queue entries for a driver. func (s *PgStore) ListQueueByDriver(ctx context.Context, driverID string) ([]QueueEntry, error) { rows, err := s.pool.Query(ctx, ` SELECT driver_id, race_id, plan_id, enqueued_ms FROM race_queue WHERE driver_id = $1 ORDER BY enqueued_ms ASC`, driverID) if err != nil { return nil, err } defer rows.Close() out := make([]QueueEntry, 0) for rows.Next() { var q QueueEntry if err := rows.Scan(&q.DriverID, &q.RaceID, &q.PlanID, &q.EnqueuedMs); err != nil { return nil, err } out = append(out, q) } return out, rows.Err() } // ListQueueByRace returns all queue entries for a given race, oldest first. func (s *PgStore) ListQueueByRace(ctx context.Context, raceID string) ([]QueueEntry, error) { rows, err := s.pool.Query(ctx, ` SELECT driver_id, race_id, plan_id, enqueued_ms FROM race_queue WHERE race_id = $1 ORDER BY enqueued_ms ASC`, raceID) if err != nil { return nil, err } defer rows.Close() out := make([]QueueEntry, 0) for rows.Next() { var q QueueEntry if err := rows.Scan(&q.DriverID, &q.RaceID, &q.PlanID, &q.EnqueuedMs); err != nil { return nil, err } out = append(out, q) } return out, rows.Err() } // CountQueueByRace returns the queue length for a race. func (s *PgStore) CountQueueByRace(ctx context.Context, raceID string) (int, error) { var n int row := s.pool.QueryRow(ctx, `SELECT COUNT(*) FROM race_queue WHERE race_id = $1`, raceID) if err := row.Scan(&n); err != nil { return 0, err } return n, nil } // ToLobbyMeta converts a finished race to a wire-friendly RaceMeta for // uniform presentation alongside live races. Podium is dropped here; // the wire type carries the full PodiumEntry list separately. func (f FinishedRace) ToLobbyMeta() lobby.RaceMeta { return lobby.RaceMeta{ ID: f.ID, Name: f.Name, TrackID: f.TrackID, MaxCars: f.MaxCars, Laps: f.Laps, TimeLimitS: f.TimeLimitS, DriverIDs: append([]string(nil), f.DriverIDs...), Status: lobby.RaceStatus(f.Status), CreatedMs: f.CreatedMs, StartedMs: f.StartedMs, } }