mirror of
https://github.com/IS1DI/x0gp.git
synced 2026-07-17 04:57:55 +00:00
- 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.
799 lines
22 KiB
Go
799 lines
22 KiB
Go
package races
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/x0gp/server/internal/lobby"
|
|
)
|
|
|
|
// Service composes lobby.Service (live) with PgStore (finished/plans/queue)
|
|
// and exposes the read/join methods used by the HTTP handlers and the
|
|
// scheduler.
|
|
type Service struct {
|
|
pg *PgStore
|
|
lobby *lobby.Service
|
|
live *LiveStore // optional: when set, live reads go through DB
|
|
now func() time.Time
|
|
maxIDs int
|
|
}
|
|
|
|
// NewService wires a Service. now is overridable for tests.
|
|
func NewService(pg *PgStore, lb *lobby.Service) *Service {
|
|
return &Service{
|
|
pg: pg,
|
|
lobby: lb,
|
|
now: time.Now,
|
|
maxIDs: 3,
|
|
}
|
|
}
|
|
|
|
// SetLiveStore installs the LiveStore (DB-backed live races). When set,
|
|
// the read side of the races list pulls from Postgres instead of the
|
|
// in-memory lobby. Writes still go through lobby.Service.
|
|
func (s *Service) SetLiveStore(live *LiveStore) {
|
|
s.live = live
|
|
}
|
|
|
|
// RestoreFromDB rehydrates the in-memory lobby from the live_races and
|
|
// lobby_drivers tables. Called at startup so that active races and
|
|
// driver presence survive a server restart. If LiveStore is not
|
|
// configured this is a no-op.
|
|
func (s *Service) RestoreFromDB(ctx context.Context) (int, int, error) {
|
|
if s.live == nil {
|
|
return 0, 0, nil
|
|
}
|
|
races, err := s.live.ListAllRaces(ctx)
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("restore races: %w", err)
|
|
}
|
|
drivers, err := s.live.ListAllDrivers(ctx)
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("restore drivers: %w", err)
|
|
}
|
|
// Restore drivers first so the race restore can attach them.
|
|
for _, d := range drivers {
|
|
// We bypass the normal AddDriver because we don't want to
|
|
// re-trigger a persistence write for already-persisted rows.
|
|
// Use the internal helper if available; otherwise go through
|
|
// 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
|
|
// for races that are already in the DB. AddRace (a new helper
|
|
// on lobby.Service) inserts the race without mirroring.
|
|
s.lobby.AddRace(r)
|
|
for _, did := range r.DriverIDs {
|
|
_ = s.lobby.AddDriverToRace(did, r.ID)
|
|
}
|
|
}
|
|
return len(races), len(drivers), nil
|
|
}
|
|
|
|
// SetMaxUpcoming overrides the default "next N" queue size (default 3).
|
|
func (s *Service) SetMaxUpcoming(n int) {
|
|
if n > 0 && n <= 16 {
|
|
s.maxIDs = n
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// List (keyset, mixed live + finished)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// ListFilter is the parsed query for ListRaces.
|
|
type ListFilter struct {
|
|
Statuses []StatusFilter // resolved filter; empty = all
|
|
TrackID string
|
|
Cursor Cursor
|
|
Limit int
|
|
}
|
|
|
|
// RaceItem is one row in the paginated result.
|
|
type RaceItem struct {
|
|
Source string // "live" | "finished"
|
|
Meta lobby.RaceMeta // always populated
|
|
// Podium is populated for finished races only.
|
|
Podium []PodiumEntry
|
|
}
|
|
|
|
// ListResult is a keyset page.
|
|
type ListResult struct {
|
|
Items []RaceItem
|
|
NextCursor string // empty when no more pages
|
|
TotalCount int
|
|
}
|
|
|
|
// ListRaces returns one keyset page of races matching the filter.
|
|
//
|
|
// Pagination strategy: live races (in-memory) are stable per snapshot but
|
|
// don't have a DB-side index. To honour the "keyset" contract we page
|
|
// live and finished sources independently and merge by sort key:
|
|
//
|
|
// * Live races use started_ms if set, otherwise created_ms.
|
|
// * Finished races use finished_ms.
|
|
//
|
|
// Pages are pulled live-first when any "active" status is in the filter,
|
|
// or finished-first when status=finished. The cursor encodes
|
|
// ("live"|"finished", sort_ms, race_id) so successive pages stay in order.
|
|
func (s *Service) ListRaces(ctx context.Context, f ListFilter) (ListResult, error) {
|
|
limit := f.Limit
|
|
if limit <= 0 {
|
|
limit = 50
|
|
}
|
|
if limit > 200 {
|
|
limit = 200
|
|
}
|
|
|
|
wantFinished := false
|
|
wantLive := false
|
|
for _, st := range f.Statuses {
|
|
if st == StatusAll {
|
|
wantLive = true
|
|
wantFinished = true
|
|
break
|
|
}
|
|
if st.matchesLive("finished") || st == StatusFinished {
|
|
wantFinished = true
|
|
}
|
|
if st.matchesLive("lobby") || st == StatusLobby ||
|
|
st == StatusRacing {
|
|
wantLive = true
|
|
}
|
|
}
|
|
|
|
// Parse cursor split (source/ms/id) if present.
|
|
curSource, curMs, curID := "live", int64(0), ""
|
|
if !f.Cursor.IsZero() {
|
|
curSource, curMs, curID = splitCursor(f.Cursor)
|
|
}
|
|
|
|
// Collect candidates from each source up to (limit+1) entries.
|
|
out := make([]RaceItem, 0, limit+1)
|
|
|
|
if wantLive {
|
|
live, err := s.collectLive(ctx, f.Statuses, f.TrackID, curSource, curMs, curID, limit+1)
|
|
if err != nil {
|
|
return ListResult{}, err
|
|
}
|
|
out = append(out, live...)
|
|
}
|
|
if wantFinished {
|
|
finished, err := s.collectFinished(ctx, f.Statuses, f.TrackID, curSource, curMs, curID, limit+1)
|
|
if err != nil {
|
|
return ListResult{}, err
|
|
}
|
|
out = append(out, finished...)
|
|
}
|
|
|
|
// Sort by (sort_ms DESC, id ASC), stable.
|
|
sort.SliceStable(out, func(i, j int) bool {
|
|
mi := sortMs(out[i].Meta)
|
|
mj := sortMs(out[j].Meta)
|
|
if mi != mj {
|
|
return mi > mj
|
|
}
|
|
return out[i].Meta.ID < out[j].Meta.ID
|
|
})
|
|
|
|
hasMore := false
|
|
if len(out) > limit {
|
|
hasMore = true
|
|
out = out[:limit]
|
|
}
|
|
|
|
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{
|
|
Ms: sortMs(last.Meta),
|
|
ID: last.Meta.ID,
|
|
}.Encode()
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func sortMs(m lobby.RaceMeta) int64 {
|
|
if m.Status == lobby.RaceStatusFinished {
|
|
// Finished meta from PgStore has no FinishedMs field; StartedMs is the
|
|
// best proxy for the original finished time within the lobby shape.
|
|
if m.StartedMs > 0 {
|
|
return m.StartedMs
|
|
}
|
|
return m.CreatedMs
|
|
}
|
|
if m.StartedMs > 0 {
|
|
return m.StartedMs
|
|
}
|
|
return m.CreatedMs
|
|
}
|
|
|
|
// collectLive fetches one page of live races. If s.live (LiveStore) is
|
|
// set, the data comes from Postgres; otherwise it falls back to the
|
|
// in-memory lobby.
|
|
func (s *Service) collectLive(ctx context.Context, filters []StatusFilter, trackID, curSource string, curMs int64, curID string, limit int) ([]RaceItem, error) {
|
|
if s.live != nil && curSource == "live" {
|
|
return s.collectLiveDB(ctx, filters, trackID, curMs, curID, limit)
|
|
}
|
|
// Fallback: in-memory lobby. The in-memory branch still applies the
|
|
// cursor manually because the legacy splitCursor format encodes the
|
|
// source in the id suffix.
|
|
all := s.lobby.ListRaces()
|
|
out := make([]RaceItem, 0, limit)
|
|
for _, r := range all {
|
|
if !statusMatches(filters, string(r.Status)) {
|
|
continue
|
|
}
|
|
if trackID != "" && r.TrackID != trackID {
|
|
continue
|
|
}
|
|
sortKey := sortMs(r)
|
|
if curSource == "live" {
|
|
if sortKey < curMs || (sortKey == curMs && r.ID >= curID) {
|
|
continue
|
|
}
|
|
}
|
|
out = append(out, RaceItem{Source: "live", Meta: r})
|
|
if len(out) >= limit {
|
|
break
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *Service) collectLiveDB(ctx context.Context, filters []StatusFilter, trackID string, curMs int64, curID string, limit int) ([]RaceItem, error) {
|
|
statuses := filterToStatusStrings(filters)
|
|
cur := Cursor{Ms: curMs, ID: curID}
|
|
rows, err := s.live.ListLivePaged(ctx, statuses, trackID, cur, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list live: %w", err)
|
|
}
|
|
out := make([]RaceItem, 0, len(rows))
|
|
for _, r := range rows {
|
|
// Skip anything that doesn't match the in-memory status filter
|
|
// (DB already filtered; this is a safety belt).
|
|
if !statusMatches(filters, string(r.Status)) {
|
|
continue
|
|
}
|
|
out = append(out, RaceItem{Source: "live", Meta: r})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// filterToStatusStrings returns the list of status values that the DB
|
|
// query should match. StatusAll expands to the three live statuses
|
|
// (lobby, countdown, racing). The caller adds the finished source
|
|
// separately.
|
|
func filterToStatusStrings(filters []StatusFilter) []string {
|
|
if len(filters) == 0 {
|
|
return []string{string(lobby.RaceStatusLobby), string(lobby.RaceStatusCountdown), string(lobby.RaceStatusRacing)}
|
|
}
|
|
seen := map[string]bool{}
|
|
for _, f := range filters {
|
|
switch f {
|
|
case StatusAll:
|
|
return []string{string(lobby.RaceStatusLobby), string(lobby.RaceStatusCountdown), string(lobby.RaceStatusRacing)}
|
|
case StatusRacing:
|
|
seen["racing"] = true
|
|
case StatusLobby:
|
|
seen["lobby"] = true
|
|
case StatusFinished:
|
|
// finished is served by collectFinished; skip here.
|
|
}
|
|
}
|
|
out := make([]string, 0, len(seen))
|
|
for k := range seen {
|
|
out = append(out, k)
|
|
}
|
|
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
|
|
}
|
|
cur := Cursor{}
|
|
if curSource == "finished" {
|
|
cur = Cursor{Ms: curMs, ID: curID}
|
|
}
|
|
rows, err := s.pg.ListFinished(ctx, trackID, cur, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([]RaceItem, 0, len(rows))
|
|
for _, f := range rows {
|
|
out = append(out, RaceItem{Source: "finished", Meta: f.ToLobbyMeta(), Podium: f.Podium})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func statusMatches(filters []StatusFilter, liveStatus string) bool {
|
|
if len(filters) == 0 {
|
|
return true
|
|
}
|
|
for _, f := range filters {
|
|
if f == StatusAll {
|
|
return true
|
|
}
|
|
if f.matchesLive(liveStatus) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// statusWantsFinished is true if the filter list allows finished races
|
|
// through. Used to gate the Postgres query in collectFinished.
|
|
func statusWantsFinished(filters []StatusFilter) bool {
|
|
if len(filters) == 0 {
|
|
return true
|
|
}
|
|
for _, f := range filters {
|
|
if f == StatusAll || f == StatusFinished {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// splitCursor is a no-op shim; we encode source in the id field for v1
|
|
// simplicity (suffix ":src"). Returns ("live"|"finished", ms, id).
|
|
func splitCursor(c Cursor) (string, int64, string) {
|
|
id := c.ID
|
|
src := "live"
|
|
// Convention: id may end with ":finished" or ":live". If absent, treat as live.
|
|
for _, suf := range []string{":finished", ":live"} {
|
|
if len(id) > len(suf) && id[len(id)-len(suf):] == suf {
|
|
id = id[:len(id)-len(suf)]
|
|
src = suf[1:]
|
|
break
|
|
}
|
|
}
|
|
return src, c.Ms, id
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Upcoming (next N) and queue join
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// UpcomingItem is one row in the upcoming list.
|
|
type UpcomingItem struct {
|
|
Meta lobby.RaceMeta
|
|
QueueLen int
|
|
PlanID string
|
|
StartAtMs int64 // best-effort: StartedMs for live races, otherwise CreatedMs+offset
|
|
}
|
|
|
|
// Upcoming returns the next N open (lobby|countdown) races, soonest first.
|
|
func (s *Service) Upcoming(ctx context.Context, limit int) ([]UpcomingItem, error) {
|
|
if limit <= 0 {
|
|
limit = s.maxIDs
|
|
}
|
|
if limit > 16 {
|
|
limit = 16
|
|
}
|
|
// When LiveStore is configured, read upcoming from Postgres.
|
|
if s.live != nil {
|
|
return s.upcomingDB(ctx, limit)
|
|
}
|
|
all := s.lobby.ListRaces()
|
|
out := make([]UpcomingItem, 0, limit)
|
|
for _, r := range all {
|
|
if r.Status != lobby.RaceStatusLobby && r.Status != lobby.RaceStatusCountdown {
|
|
continue
|
|
}
|
|
n, err := s.pg.CountQueueByRace(ctx, r.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
startAt := r.StartedMs
|
|
if startAt == 0 {
|
|
startAt = r.CreatedMs
|
|
}
|
|
out = append(out, UpcomingItem{
|
|
Meta: r,
|
|
QueueLen: n,
|
|
StartAtMs: startAt,
|
|
})
|
|
}
|
|
sort.SliceStable(out, func(i, j int) bool {
|
|
if out[i].StartAtMs != out[j].StartAtMs {
|
|
return out[i].StartAtMs < out[j].StartAtMs
|
|
}
|
|
return out[i].Meta.ID < out[j].Meta.ID
|
|
})
|
|
if len(out) > limit {
|
|
out = out[:limit]
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *Service) upcomingDB(ctx context.Context, limit int) ([]UpcomingItem, error) {
|
|
rows, err := s.live.ListUpcoming(ctx, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list upcoming: %w", err)
|
|
}
|
|
out := make([]UpcomingItem, 0, len(rows))
|
|
for _, r := range rows {
|
|
n, err := s.pg.CountQueueByRace(ctx, r.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
startAt := r.StartedMs
|
|
if startAt == 0 {
|
|
startAt = r.CreatedMs
|
|
}
|
|
out = append(out, UpcomingItem{
|
|
Meta: r,
|
|
QueueLen: n,
|
|
StartAtMs: startAt,
|
|
})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// JoinUpcomingResult is what JoinUpcoming returns.
|
|
type JoinUpcomingResult struct {
|
|
Joined []QueueEntry
|
|
Skipped []string // race ids that were already at capacity
|
|
}
|
|
|
|
// JoinUpcoming puts the driver into the queue for each of the next N
|
|
// upcoming races. Idempotent per (driver, race).
|
|
func (s *Service) JoinUpcoming(ctx context.Context, driverID string, limit int) (JoinUpcomingResult, error) {
|
|
if driverID == "" {
|
|
return JoinUpcomingResult{}, fmt.Errorf("%w: driver_id required", ErrInvalidInput)
|
|
}
|
|
up, err := s.Upcoming(ctx, limit)
|
|
if err != nil {
|
|
return JoinUpcomingResult{}, err
|
|
}
|
|
res := JoinUpcomingResult{Joined: make([]QueueEntry, 0, len(up))}
|
|
for _, u := range up {
|
|
// Skip races that are full; surfaced in Skipped for the caller.
|
|
if len(u.Meta.DriverIDs) >= u.Meta.MaxCars {
|
|
res.Skipped = append(res.Skipped, u.Meta.ID)
|
|
continue
|
|
}
|
|
// Best-effort auto-attach: if the race is still lobby and the driver
|
|
// can fit, AddDriverToRace moves them in. The queue entry remains
|
|
// for diagnostics.
|
|
if u.Meta.Status == lobby.RaceStatusLobby && len(u.Meta.DriverIDs) < u.Meta.MaxCars {
|
|
_ = s.lobby.AddDriverToRace(driverID, u.Meta.ID)
|
|
}
|
|
q, err := s.pg.Enqueue(ctx, driverID, u.Meta.ID, u.PlanID)
|
|
if err != nil {
|
|
return res, err
|
|
}
|
|
res.Joined = append(res.Joined, q)
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
// LeaveQueue removes a single (driver, race) entry.
|
|
func (s *Service) LeaveQueue(ctx context.Context, driverID, raceID string) error {
|
|
return s.pg.Dequeue(ctx, driverID, raceID)
|
|
}
|
|
|
|
// ListQueue returns the driver's queue entries.
|
|
func (s *Service) ListQueue(ctx context.Context, driverID string) ([]QueueEntry, error) {
|
|
return s.pg.ListQueueByDriver(ctx, driverID)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Finished-race snapshot hook
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// 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
|
|
TotalLaps int
|
|
WinnerDriverID string
|
|
WinnerName string
|
|
BestLapMs int64
|
|
Results []DriverResult
|
|
}
|
|
|
|
// PersistFinished upserts the finished race into Postgres.
|
|
func (s *Service) PersistFinished(ctx context.Context, sf SnapshotFinished) error {
|
|
startedMs := sf.Meta.StartedMs
|
|
finishedMs := sf.FinishedMs
|
|
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),
|
|
Results: results,
|
|
}
|
|
return s.pg.InsertFinished(ctx, r)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Race plans
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// CreatePlanInput is what the HTTP handler hands to the service.
|
|
type CreatePlanInput struct {
|
|
ID string
|
|
Name string
|
|
TrackID string
|
|
MaxCars int
|
|
Laps int
|
|
TimeLimitS int
|
|
StartAtMs int64
|
|
IntervalS int
|
|
Count int
|
|
Enabled bool
|
|
}
|
|
|
|
// CreatePlan validates and persists a plan.
|
|
func (s *Service) CreatePlan(ctx context.Context, in CreatePlanInput) (RacePlan, error) {
|
|
if in.Name == "" {
|
|
return RacePlan{}, fmt.Errorf("%w: name required", ErrInvalidInput)
|
|
}
|
|
if in.MaxCars < 1 || in.MaxCars > 8 {
|
|
return RacePlan{}, fmt.Errorf("%w: max_cars must be 1..8", ErrInvalidInput)
|
|
}
|
|
if in.StartAtMs <= 0 {
|
|
return RacePlan{}, fmt.Errorf("%w: start_at_ms required", ErrInvalidInput)
|
|
}
|
|
if in.IntervalS < 0 {
|
|
return RacePlan{}, fmt.Errorf("%w: interval_s must be >= 0", ErrInvalidInput)
|
|
}
|
|
if in.Count < 0 {
|
|
return RacePlan{}, fmt.Errorf("%w: count must be >= 0", ErrInvalidInput)
|
|
}
|
|
if in.ID == "" {
|
|
in.ID = fmt.Sprintf("plan-%d-%d", s.now().UnixMilli(), time.Now().UnixNano()%1000)
|
|
}
|
|
if in.TrackID == "" {
|
|
in.TrackID = "default"
|
|
}
|
|
p := RacePlan{
|
|
ID: in.ID,
|
|
Name: in.Name,
|
|
TrackID: in.TrackID,
|
|
MaxCars: in.MaxCars,
|
|
Laps: in.Laps,
|
|
TimeLimitS: in.TimeLimitS,
|
|
StartAtMs: in.StartAtMs,
|
|
IntervalS: in.IntervalS,
|
|
Count: in.Count,
|
|
Enabled: in.Enabled,
|
|
NextFireMs: in.StartAtMs,
|
|
}
|
|
if err := s.pg.CreatePlan(ctx, p); err != nil {
|
|
return RacePlan{}, err
|
|
}
|
|
return s.pg.GetPlan(ctx, p.ID)
|
|
}
|
|
|
|
// ListPlans returns plans in start_at ascending order.
|
|
func (s *Service) ListPlans(ctx context.Context, cur Cursor, limit int) ([]RacePlan, error) {
|
|
if limit <= 0 {
|
|
limit = 50
|
|
}
|
|
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)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Scheduler
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// Scheduler materialises due race plans into lobby races every tick.
|
|
type Scheduler struct {
|
|
pg *PgStore
|
|
lobby *lobby.Service
|
|
tick time.Duration
|
|
stopCh chan struct{}
|
|
doneCh chan struct{}
|
|
once sync.Once
|
|
}
|
|
|
|
// NewScheduler creates a scheduler. tick default is 5s.
|
|
func NewScheduler(pg *PgStore, lb *lobby.Service, tick time.Duration) *Scheduler {
|
|
if tick <= 0 {
|
|
tick = 5 * time.Second
|
|
}
|
|
return &Scheduler{
|
|
pg: pg,
|
|
lobby: lb,
|
|
tick: tick,
|
|
stopCh: make(chan struct{}),
|
|
doneCh: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// Run blocks until Stop is called. Safe to launch in a goroutine.
|
|
func (s *Scheduler) Run(ctx context.Context) {
|
|
defer close(s.doneCh)
|
|
t := time.NewTicker(s.tick)
|
|
defer t.Stop()
|
|
// Fire once on start so newly-enabled plans materialise promptly.
|
|
s.tickOnce(ctx)
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-s.stopCh:
|
|
return
|
|
case <-t.C:
|
|
s.tickOnce(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Stop signals the scheduler to exit.
|
|
func (s *Scheduler) Stop() {
|
|
s.once.Do(func() { close(s.stopCh) })
|
|
}
|
|
|
|
// Done blocks until the scheduler goroutine has returned.
|
|
func (s *Scheduler) Done() <-chan struct{} { return s.doneCh }
|
|
|
|
func (s *Scheduler) tickOnce(ctx context.Context) {
|
|
now := s.now().UnixMilli()
|
|
plans, err := s.pg.DuePlans(ctx, now, 32)
|
|
if err != nil {
|
|
return
|
|
}
|
|
for _, p := range plans {
|
|
meta, err := s.lobby.CreateRace(lobby.CreateRaceOptions{
|
|
Name: p.Name,
|
|
TrackID: p.TrackID,
|
|
MaxCars: p.MaxCars,
|
|
Laps: p.Laps,
|
|
TimeLimitS: p.TimeLimitS,
|
|
})
|
|
if err != nil {
|
|
// Could not create; advance anyway to avoid hot-loop on the
|
|
// same plan.
|
|
next := p.NextFireMs
|
|
if p.IntervalS > 0 {
|
|
next = next + int64(p.IntervalS)*1000
|
|
}
|
|
_ = s.pg.AdvancePlan(ctx, p.ID, next)
|
|
continue
|
|
}
|
|
// Compute next fire time.
|
|
next := p.NextFireMs
|
|
if p.IntervalS > 0 {
|
|
next = p.NextFireMs + int64(p.IntervalS)*1000
|
|
}
|
|
_ = s.advance(ctx, p, next)
|
|
// Auto-attach queued drivers to the new race (best-effort).
|
|
s.attachQueued(ctx, meta.ID)
|
|
}
|
|
}
|
|
|
|
func (s *Scheduler) advance(ctx context.Context, p RacePlan, nextFireMs int64) error {
|
|
if nextFireMs == 0 && p.IntervalS == 0 {
|
|
// Single-shot: disable.
|
|
return s.pg.SetPlanEnabled(ctx, p.ID, false)
|
|
}
|
|
return s.pg.AdvancePlan(ctx, p.ID, nextFireMs)
|
|
}
|
|
|
|
func (s *Scheduler) attachQueued(ctx context.Context, raceID string) {
|
|
entries, err := s.pg.ListQueueByRace(ctx, raceID)
|
|
if err != nil {
|
|
return
|
|
}
|
|
for _, q := range entries {
|
|
_, _ = s.lobby.AddDriver(q.DriverID, "", "") //nolint:errcheck // best-effort
|
|
_ = s.lobby.AddDriverToRace(q.DriverID, raceID)
|
|
_ = s.pg.Dequeue(ctx, q.DriverID, raceID)
|
|
}
|
|
}
|
|
|
|
func (s *Scheduler) now() time.Time { return time.Now() }
|