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) } // GetLeaderboard retrieves the leaderboard page and current driver position. func (s *Service) GetLeaderboard(ctx context.Context, trackID string, year int, limit, offset int, currentDriverID string) (*LeaderboardResult, error) { if year == 0 { year = s.now().Year() } var items []LeaderboardEntry var total int var err error if trackID != "" { items, total, err = s.pg.GetTrackLeaderboard(ctx, trackID, year, limit, offset) } else { items, total, err = s.pg.GetOverallLeaderboard(ctx, year, limit, offset) } if err != nil { return nil, err } var currentDriver *LeaderboardEntry if currentDriverID != "" { if trackID != "" { currentDriver, err = s.pg.GetTrackLeaderboardDriver(ctx, trackID, year, currentDriverID) } else { currentDriver, err = s.pg.GetOverallLeaderboardDriver(ctx, year, currentDriverID) } if err != nil { return nil, err } if currentDriver == nil { // fallback: get profile from drivers table currentDriver, err = s.pg.GetDriverLeaderboardProfile(ctx, currentDriverID) if err != nil { return nil, err } // if driver exists but has no points, set points/rank to 0 if currentDriver != nil { currentDriver.Points = 0 currentDriver.Rank = 0 } } } if items == nil { items = []LeaderboardEntry{} } return &LeaderboardResult{ Items: items, Total: total, CurrentDriver: currentDriver, }, nil } // --------------------------------------------------------------------------- // 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() }