Files

994 lines
30 KiB
Go

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,
}
}