diff --git a/docs/battlelogs.md b/docs/battlelogs.md index 41b4e43..1a47202 100644 --- a/docs/battlelogs.md +++ b/docs/battlelogs.md @@ -65,6 +65,8 @@ Reads target tables and the requested player's current Town Hall from `basic_pla Ranked and Legend rows retain the normalized share code directly. The hot writer inserts each distinct code into `army_compositions`, keyed only by `share_code`; there is no army hash or parser-version identity. The daily Legend closeout reuses those compositions for family and item aggregation. +Hero mode segments such as `h2m1p16e5_41` are accepted and preserved in canonical share codes. Mode is not a composition column and does not participate in army-family similarity; `m0` and `m1` can remain distinct exact codes in the same family. The earlier rejection came from Tracking's share-code validator, not clashy.go. + The battle identity is `(player_tag, battle_time)`. Deploy this writer only with the final DevKit migration 017 contract. Existing raw rows remain intact. The scheduled closeout rebuilds completed Legend-day and Ranked-season aggregates from raw attack perspectives. It groups newly observed Legend armies into immutable direct-anchor families and writes daily family outcomes. Farming and Ranked/Legend raw rows retain one year. diff --git a/docs/leaderboards.md b/docs/leaderboards.md index e728b08..4d74914 100644 --- a/docs/leaderboards.md +++ b/docs/leaderboards.md @@ -2,11 +2,11 @@ ## What this workload is for -Leaderboards fetch current global/location rankings and maintain the typed current/history tables and lightweight API cache. It is documented separately because it makes many paging and replacement decisions even though it runs inside `scheduled`. +Scheduled ranking jobs maintain official current/history tables. The adjacent leaderboard loop refreshes Legend I players and publishes SQL materialized snapshots for player Town Hall and league boards. No player or clan leaderboard is published to Valkey. ## When it runs -The current leaderboard loop runs every `leaderboards.interval_seconds`. Ordinary boards share `scheduled.requests_per_second`; the live Legend I player refresh has its own fixed 20-request-per-second limiter so its contract is independent of the general scheduled budget. +The live Legend I loop runs every `leaderboards.interval_seconds` with its own fixed 20-request-per-second limiter. A separate maintenance loop checks every minute and refreshes player Town Hall/league materialized views once six hours have elapsed since the last successful refresh. It populates uninitialized views on its first run. Ordinary official ranking fetches still share `scheduled.requests_per_second`. ## Targets and endpoints @@ -25,7 +25,11 @@ Load locations -> append/replace completed historical snapshot -> fetch every stored Legend I player at 20 requests per second -> atomically replace legend_rankings_current - -> refresh API cache metadata + -> publish SQL-backed live Legend leaderboard + +basic_player + -> every six hours: refresh player_townhall_leaderboards / player_league_leaderboards + -> API reads up to 500 rows per board, without rebuilding ranks per request ``` ## Data written @@ -36,6 +40,11 @@ Load locations - `legend_rankings_current`, atomically replaced only after the full Legend I refresh. - `ranked_league_group_members`. - Changed `basic_player` facts learned from ranked players. +- `player_townhall_leaderboards`: ranked tiers only, ordered by league descending, trophies descending, then tag. +- `player_league_leaderboards`: exact ranked tier, ordered by trophies descending, then tag. +- `tracking_scheduled_jobs`: durable refresh completion and weekly reset progress. + +The SQL snapshots include ranked players with zero trophies. Unranked and unknown leagues are excluded. Snapshots use existing `basic_player` observations; there is no separate Town Hall/league candidate-fetch pass. Clan display metadata is still joined from `basic_clan` by the API. Leaderboard writes to `basic_player` are accepted because volume is bounded and the shared SQL upsert writes only changed columns. @@ -52,6 +61,20 @@ Leaderboard writes to `basic_player` are accepted because volume is bounded and A board is replaced only with a complete accepted result. API errors are recorded per request. Availability pauses the entire fetch path, and a later scheduled pass retries naturally. +Recognized "rankings not found for location" errors on non-global boards are counted and skipped without replacing existing data. Global 404s and unrelated errors remain failures. Maintenance errors have their own `leaderboards.maintenance` readiness entry and do not terminate the Legend refresh loop. + +## Weekly Ranked trophy reset + +On Monday between 16:30 and 17:00 UTC only, maintenance resets nonzero `basic_player.trophies` for league IDs 105000000 through 105000035, including Unranked. Legend I (105000036) and unknown leagues are untouched. Monday's Ranked closeout must have recorded successful completion first. + +Each transaction changes at most 5,000 players and saves its last tag. A process restart resumes that cursor inside the same safe window. Completed jobs never repeat for that Monday; a missed or partial job is not caught up outside the window. Completion forces both player snapshots to refresh, regardless of the six-hour timer. No historical trophy records are changed, no Legend season reset is performed, and normal profile writers remain free to record later observations. + +The maintenance session holds a PostgreSQL advisory lock to prevent overlapping scheduled processes from running it simultaneously. There is deliberately no cross-writer stale-response coordination. + +## Rollout order + +Apply DevKit migration 020, deploy this Tracking change, verify both player views are populated, then deploy the API snapshot reader. The views are created WITH NO DATA so migration itself does not perform the population query. Deploying the API before the first refresh will fail those board reads. No production migration or deployment is performed by this PR. + ## What it deliberately does not do - It does not produce live events. diff --git a/docs/scheduled.md b/docs/scheduled.md index 80ad574..9d32c5a 100644 --- a/docs/scheduled.md +++ b/docs/scheduled.md @@ -6,12 +6,12 @@ ## When it runs -The main scheduled cycle uses `scheduled.interval_seconds`. Current player/clan leaderboards run alongside it at the leaderboard interval. League closeout processes the latest eligible shifted Legend day on startup, then runs daily at 05:12 UTC; Monday's closeout also discovers and finalizes completed Ranked seasons from the IDs returned by player profiles and matching league-history entries. +The main scheduled cycle uses `scheduled.interval_seconds`. The Legend player refresh runs alongside it. League closeout runs daily at 05:12 UTC, not on startup; Monday's closeout also discovers and finalizes completed Ranked seasons from player profiles and matching league-history entries, then records completion for the guarded weekly trophy reset. ## Work owned here - Current and historical player/clan leaderboard snapshots. -- Legend history completion/backfill for completed seasons. +- Forward-only Legend history completion for recent completed seasons. Already stored season IDs are skipped before parsing; routine discovery never backfills beyond a 35-day lookback or the newest stored completion, whichever is later. Older repairs require an explicit separate operation. - Ranked league member snapshots and season tier aggregates. - Legend daily hit-rate, usage, and immutable army-family aggregates. - Scheduled broad statistics and date-bound maintenance already implemented in `scripts/scheduled.go`. @@ -44,7 +44,9 @@ Army-family matching compares each exact army directly with immutable anchors. T ## Events and Valkey -Scheduled statistics normally write SQL/cache snapshots and do not emit live Discord events. Leaderboard cache output is consumed by API reads. This process does not use the event stream as a job queue. +Scheduled statistics write SQL snapshots and do not emit live Discord events. Player and clan leaderboards are not published to Valkey. This process does not use the event stream as a job queue. + +Official player history retains valid ranking facts when optional clan metadata is incomplete, leaving all three clan snapshot fields absent rather than borrowing today's clan details. Upstream `previousRank=-1` means no previous rank and is stored as SQL NULL; other negative ranks remain invalid. Current clan ranking writes use only the existing five schema columns, without `updated_at`. ## Configuration diff --git a/scripts/battlelogs.go b/scripts/battlelogs.go index 77935ac..62745d2 100644 --- a/scripts/battlelogs.go +++ b/scripts/battlelogs.go @@ -1110,19 +1110,26 @@ func validateArmyHeroSection(payload string) error { } petSeen := false equipmentSeen := false + modeSeen := false for rest != "" { marker := rest[0] if marker == '_' { if !equipmentSeen { return fmt.Errorf("equipment continuation without equipment in %q", part) } - } else if marker != 'p' && marker != 'e' { + } else if marker != 'p' && marker != 'e' && marker != 'm' { return fmt.Errorf("unknown hero item marker %q in %q", marker, part) } value, next := leadingInt(rest[1:]) if value < 0 { return fmt.Errorf("missing hero item id in %q", part) } + if marker == 'm' { + if modeSeen { + return fmt.Errorf("multiple hero modes in %q", part) + } + modeSeen = true + } if marker == 'p' { if petSeen { return fmt.Errorf("multiple pets in %q", part) @@ -1298,6 +1305,7 @@ func normalizeArmyShareCode(link string) string { type armyHeroLoadout struct { HeroID int + Mode *int // Preserved in share codes, intentionally absent from composition/family fields. PetID int Equipment []int } @@ -1315,6 +1323,14 @@ func parseArmyHeroLoadouts(payload string) []armyHeroLoadout { loadout := armyHeroLoadout{HeroID: heroID, PetID: -1} for rest != "" { marker := rest[0] + if marker == 'm' { + value, next := leadingInt(rest[1:]) + if value >= 0 { + loadout.Mode = &value + } + rest = next + continue + } if marker != 'p' && marker != 'e' { if marker == '_' { rest = rest[1:] @@ -1364,6 +1380,9 @@ func encodeArmyHeroSection(heroes []armyHeroLoadout) string { parts := make([]string, 0, len(heroes)) for _, hero := range heroes { part := strconv.Itoa(hero.HeroID) + if hero.Mode != nil { + part += "m" + strconv.Itoa(*hero.Mode) + } if hero.PetID >= 0 { part += "p" + strconv.Itoa(hero.PetID) } diff --git a/scripts/leaderboards.go b/scripts/leaderboards.go index 7ecc498..3477924 100644 --- a/scripts/leaderboards.go +++ b/scripts/leaderboards.go @@ -3,7 +3,6 @@ package scripts import ( "context" "database/sql" - "encoding/json" "errors" "fmt" "sort" @@ -18,7 +17,6 @@ import ( clashy "github.com/clashkinginc/clashy.go" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" - valkey "github.com/valkey-io/valkey-go" ) const ( @@ -33,13 +31,6 @@ const ( leaderboardMaterializedViewCount = 5 ) -var leaderboardCacheScript = valkey.NewLuaScript(` - for i = 1, #KEYS do - redis.call('SET', KEYS[i], ARGV[i]) - end - return #KEYS -`) - var leaderboardMaterializedViewRefreshQueries = [...]string{ `REFRESH MATERIALIZED VIEW CONCURRENTLY clan_leaderboards`, `REFRESH MATERIALIZED VIEW war_league_counts`, @@ -137,8 +128,7 @@ type leaderboardCandidate struct { } type timescaleLeaderboardStore struct { - pool *pgxpool.Pool - valkey valkey.Client + pool *pgxpool.Pool } type leaderboardCacheSet struct { @@ -208,6 +198,13 @@ func (d *leaderboardsDomain) Run(ctx context.Context, app *platform.App) error { d.store = store d.refreshMaterializedViews = store.RefreshMaterializedViews defer store.Close() + maintenanceCtx, stopMaintenance := context.WithCancel(ctx) + maintenanceDone := make(chan struct{}) + go func() { + defer close(maintenanceDone) + store.runPlayerLeaderboardMaintenance(maintenanceCtx, app) + }() + defer func() { stopMaintenance(); <-maintenanceDone }() interval := time.Duration(app.Config.LeaderboardIntervalSeconds) * time.Second for { @@ -247,9 +244,6 @@ func validateLeaderboardsConfig(cfg platform.Config) error { if cfg.TimescaleURL == "" { return errors.New("TIMESCALE_* connection variables are required for leaderboards") } - if cfg.ValkeyAddr == "" { - return errors.New("valkey_addr is required for leaderboards") - } return nil } @@ -258,7 +252,7 @@ func (d *leaderboardsDomain) openStore(ctx context.Context, app *platform.App) ( if err != nil { return nil, err } - return ×caleLeaderboardStore{pool: pool, valkey: app.Valkey}, nil + return ×caleLeaderboardStore{pool: pool}, nil } func (d *leaderboardsDomain) runCycle(ctx context.Context, app *platform.App) error { @@ -268,59 +262,7 @@ func (d *leaderboardsDomain) runCycle(ctx context.Context, app *platform.App) er if d.legendLimiter == nil { return errors.New("20 RPS Legend request limiter is required for leaderboards") } - legendPlayers, err := d.refreshLegendRankings(ctx, app) - if err != nil { - return err - } - candidates, err := d.store.LoadCandidates(ctx, d.limit) - if err != nil { - return err - } - tags := leaderboardCandidateTags(candidates) - if len(tags) == 0 { - cache := buildLeaderboardCache(candidates, nil, nil, time.Now().UTC(), d.limit, d.nullAssetURL) - return d.store.CacheBoards(ctx, cache) - } - leagues, err := d.fetchLeagues(ctx, app, d.limiter) - if err != nil { - return err - } - legendByTag := make(map[string]leaderboardPlayerRow, len(legendPlayers)) - for _, player := range legendPlayers { - legendByTag[player.Tag] = player - } - standardTags := make([]string, 0, len(tags)) - for _, tag := range tags { - if _, refreshedAsLegend := legendByTag[tag]; !refreshedAsLegend { - standardTags = append(standardTags, tag) - } - } - players, deletedTags, err := d.fetchPlayers(ctx, app, d.limiter, standardTags, leagues) - if err != nil { - return err - } - for _, tag := range tags { - if player, ok := legendByTag[tag]; ok { - players = append(players, player) - } - } - basicPlayers := leaderboardBasicPlayerRows(players) - deleted, err := d.store.DeletePlayers(ctx, deletedTags) - if err != nil { - return err - } - if err := d.store.UpdatePlayers(ctx, basicPlayers); err != nil { - return err - } - clans, err := d.store.LoadClanMetadata(ctx, leaderboardPlayerClanTags(players)) - if err != nil { - return err - } - cache := buildLeaderboardCache(candidates, players, clans, time.Now().UTC(), d.limit, d.nullAssetURL) - err = d.store.CacheBoards(ctx, cache) - if err == nil { - app.Stats.RecordWrite(leaderboardsDomainName, len(basicPlayers)+deleted+len(cache.Boards)+1) - } + _, err := d.refreshLegendRankings(ctx, app) return err } @@ -789,26 +731,6 @@ func (s *timescaleLeaderboardStore) LoadClanMetadata(ctx context.Context, clanTa return out, rows.Err() } -func (s *timescaleLeaderboardStore) CacheBoards(ctx context.Context, cache leaderboardCacheSet) error { - rawIndex, err := json.Marshal(cache) - if err != nil { - return err - } - keys := make([]string, 0, len(cache.Boards)+1) - values := make([]string, 0, len(cache.Boards)+1) - for _, board := range cache.Boards { - raw, err := json.Marshal(board) - if err != nil { - return err - } - keys = append(keys, leaderboardCacheKey(board.Kind, board.Key)) - values = append(values, string(raw)) - } - keys = append(keys, leaderboardIndexKey) - values = append(values, string(rawIndex)) - return leaderboardCacheScript.Exec(ctx, s.valkey, keys, values).Error() -} - func (s *timescaleLeaderboardStore) RefreshMaterializedViews(ctx context.Context) error { for i, query := range leaderboardMaterializedViewRefreshQueries { if _, err := s.pool.Exec(ctx, query); err != nil { diff --git a/scripts/player_leaderboard_maintenance.go b/scripts/player_leaderboard_maintenance.go new file mode 100644 index 0000000..ee46045 --- /dev/null +++ b/scripts/player_leaderboard_maintenance.go @@ -0,0 +1,153 @@ +package scripts + +import ( + "context" + "errors" + "fmt" + "time" + + "clashking_tracking/internal/platform" + "github.com/jackc/pgx/v5" +) + +// UTC, after the 05:12 closeout and before the 17:00 tournament opening. +// Never catch up a missed destructive reset outside this half-hour window. +func rankedResetWindow(now time.Time) (time.Time, bool) { + now = now.UTC() + day := dayStart(now) + return day, now.Weekday() == time.Monday && !now.Before(day.Add(16*time.Hour+30*time.Minute)) && now.Before(day.Add(17*time.Hour)) +} + +func (s *timescaleLeaderboardStore) runPlayerLeaderboardMaintenance(ctx context.Context, app *platform.App) { + for { + if err := s.playerLeaderboardMaintenance(ctx); err != nil && ctx.Err() == nil { + app.Logger.Error("player leaderboard maintenance failed", "err", err) + app.Stats.SetReady("leaderboards.maintenance", false, err.Error()) + } else if ctx.Err() == nil { + app.Stats.SetReady("leaderboards.maintenance", true, "") + } + if err := sleepOrDone(ctx, time.Minute); err != nil { + return + } + } +} + +func (s *timescaleLeaderboardStore) playerLeaderboardMaintenance(ctx context.Context) error { + conn, err := s.pool.Acquire(ctx) + if err != nil { + return err + } + defer conn.Release() + // Session lock serializes maintenance across overlapping scheduled deployments. + var locked bool + if err = conn.QueryRow(ctx, "SELECT pg_try_advisory_lock(73194201)").Scan(&locked); err != nil || !locked { + return err + } + defer func() { + unlockCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, e := conn.Exec(unlockCtx, "SELECT pg_advisory_unlock(73194201)"); e != nil { + _ = conn.Conn().Close(unlockCtx) + } + }() + var now time.Time + if err = conn.QueryRow(ctx, "SELECT clock_timestamp()").Scan(&now); err != nil { + return err + } + if period, due := rankedResetWindow(now); due { + for { + done, e := resetRankedTrophyBatch(ctx, conn.Conn(), period) + if e != nil { + return e + } + if done { + break + } + } + } + var due bool + err = conn.QueryRow(ctx, `SELECT NOT EXISTS ( + SELECT 1 FROM tracking_scheduled_jobs refresh WHERE job='player_board_refresh' + AND completed_at > clock_timestamp()-interval '6 hours' + AND completed_at >= COALESCE((SELECT max(completed_at) FROM tracking_scheduled_jobs WHERE job='ranked_trophy_reset'),'-infinity'::timestamptz) + )`).Scan(&due) + if err != nil || !due { + return err + } + for _, view := range []string{"player_townhall_leaderboards", "player_league_leaderboards"} { + var populated bool + if err = conn.QueryRow(ctx, "SELECT ispopulated FROM pg_matviews WHERE schemaname='public' AND matviewname=$1", view).Scan(&populated); err != nil { + return err + } + query := "REFRESH MATERIALIZED VIEW " + if populated { + query += "CONCURRENTLY " + } + if _, err = conn.Exec(ctx, query+view); err != nil { + return err + } + } + _, err = conn.Exec(ctx, `INSERT INTO tracking_scheduled_jobs(job,period,completed_at) + VALUES('player_board_refresh',(clock_timestamp() AT TIME ZONE 'UTC')::date,clock_timestamp()) + ON CONFLICT(job,period) DO UPDATE SET completed_at=EXCLUDED.completed_at`) + return err +} + +type rankedResetConnection interface { + Begin(context.Context) (pgx.Tx, error) +} + +func resetRankedTrophyBatch(ctx context.Context, conn rankedResetConnection, period time.Time) (bool, error) { + tx, err := conn.Begin(ctx) + if err != nil { + return false, err + } + defer tx.Rollback(ctx) + var now time.Time + if err = tx.QueryRow(ctx, "SELECT clock_timestamp()").Scan(&now); err != nil { + return false, err + } + return resetRankedTrophyBatchTx(ctx, tx, period, now) +} + +// Separated from the database clock for deterministic safe-window tests. +func resetRankedTrophyBatchTx(ctx context.Context, tx pgx.Tx, period, now time.Time) (bool, error) { + current, allowed := rankedResetWindow(now) + if !allowed || !current.Equal(period) { + return false, errors.New("weekly trophy reset safe window closed; remaining rows left unchanged") + } + var closeoutDone bool + var err error + if err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM tracking_scheduled_jobs WHERE job='ranked_closeout' AND period=$1 AND completed_at IS NOT NULL)`, period).Scan(&closeoutDone); err != nil { + return false, err + } + if !closeoutDone { + return false, fmt.Errorf("weekly trophy reset waiting for ranked closeout %s", period.Format("2006-01-02")) + } + if _, err = tx.Exec(ctx, `INSERT INTO tracking_scheduled_jobs(job,period) VALUES('ranked_trophy_reset',$1) ON CONFLICT DO NOTHING`, period); err != nil { + return false, err + } + var lastTag string + var completed *time.Time + if err = tx.QueryRow(ctx, `SELECT last_tag,completed_at FROM tracking_scheduled_jobs WHERE job='ranked_trophy_reset' AND period=$1 FOR UPDATE`, period).Scan(&lastTag, &completed); err != nil { + return false, err + } + if completed != nil { + return true, nil + } + var next string + var count int + err = tx.QueryRow(ctx, `WITH batch AS MATERIALIZED ( + SELECT tag FROM basic_player WHERE tag>$1 AND league_id BETWEEN 105000000 AND 105000035 AND trophies<>0 ORDER BY tag LIMIT 5000 + ), changed AS (UPDATE basic_player p SET trophies=0 FROM batch b WHERE p.tag=b.tag + AND p.league_id BETWEEN 105000000 AND 105000035 AND p.trophies<>0 RETURNING p.tag) + SELECT COALESCE((SELECT max(tag) FROM batch),$1),(SELECT count(*) FROM changed)`, lastTag).Scan(&next, &count) + if err != nil { + return false, err + } + done := next == lastTag + if _, err = tx.Exec(ctx, `UPDATE tracking_scheduled_jobs SET last_tag=$2,completed_at=CASE WHEN $3 THEN clock_timestamp() ELSE NULL END WHERE job='ranked_trophy_reset' AND period=$1`, period, next, done); err != nil { + return false, err + } + return done, tx.Commit(ctx) +} diff --git a/scripts/player_leaderboard_maintenance_integration_test.go b/scripts/player_leaderboard_maintenance_integration_test.go new file mode 100644 index 0000000..b545327 --- /dev/null +++ b/scripts/player_leaderboard_maintenance_integration_test.go @@ -0,0 +1,111 @@ +//go:build league_analytics_integration + +package scripts + +import ( + "context" + "github.com/jackc/pgx/v5/pgxpool" + "os" + "testing" + "time" +) + +func TestPlayerBoardRefreshAndWeeklyReset(t *testing.T) { + if os.Getenv("CLASHKING_DISPOSABLE_TIMESCALE") != "1" { + t.Skip("schema-owned disposable fixture required") + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, os.Getenv("TEST_DATABASE_URL")) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + exec := func(sql string, args ...any) { + t.Helper() + if _, e := pool.Exec(ctx, sql, args...); e != nil { + t.Fatal(e) + } + } + exec(`INSERT INTO basic_player(tag,name,townhall_level,league_id,trophies) + SELECT '#RESET'||lpad(n::text,5,'0'),'Reset',16,105000033,3000 FROM generate_series(1,5001) n; + INSERT INTO basic_player(tag,name,townhall_level,league_id,trophies) VALUES + ('#RESETLEGEND','Legend',16,105000036,6000),('#RESETUNKNOWN','Unknown',16,NULL,3000),('#RESETUNRANKED','Unranked',16,105000000,99);`) + now := time.Date(2026, 9, 21, 16, 35, 0, 0, time.UTC) + period := dayStart(now) + batch := func(at time.Time) (bool, error) { + tx, e := pool.Begin(ctx) + if e != nil { + return false, e + } + defer tx.Rollback(ctx) + return resetRankedTrophyBatchTx(ctx, tx, period, at) + } + if _, e := batch(now); e == nil { + t.Fatal("reset without closeout") + } + exec(`INSERT INTO tracking_scheduled_jobs(job,period,completed_at) VALUES('ranked_closeout',$1,now())`, period) + if _, e := batch(now.Add(time.Hour)); e == nil { + t.Fatal("reset outside window") + } + done, e := batch(now) + if e != nil || done { + t.Fatalf("first batch: %v %v", done, e) + } + var remaining int + if e = pool.QueryRow(ctx, `SELECT count(*) FROM basic_player WHERE tag LIKE '#RESET%' AND league_id BETWEEN 105000000 AND 105000035 AND trophies<>0`).Scan(&remaining); e != nil || remaining != 2 { + t.Fatalf("batch remaining=%d: %v", remaining, e) + } + // Each call opens a fresh transaction, matching a restart resuming the stored cursor. + for i := 0; i < 3 && !done; i++ { + done, e = batch(now) + if e != nil { + t.Fatal(e) + } + } + if !done { + t.Fatal("reset did not finish") + } + var preserved int + if e = pool.QueryRow(ctx, `SELECT sum(trophies) FROM basic_player WHERE tag IN ('#RESETLEGEND','#RESETUNKNOWN')`).Scan(&preserved); e != nil || preserved != 9000 { + t.Fatalf("excluded players changed: %d %v", preserved, e) + } + exec(`UPDATE basic_player SET trophies=123 WHERE tag='#RESET00001'`) + if done, e = batch(now); e != nil || !done { + t.Fatalf("repeat: %v %v", done, e) + } + var trophies int + if e = pool.QueryRow(ctx, `SELECT trophies FROM basic_player WHERE tag='#RESET00001'`).Scan(&trophies); e != nil || trophies != 123 { + t.Fatalf("repeated reset: %d %v", trophies, e) + } + store := ×caleLeaderboardStore{pool: pool} + clans := ×caleScheduledStore{pool: pool} + group := currentClanRankingGroup{RankingType: "home", LocationID: "global", Rows: []currentClanRankingRow{{ClanTag: "#CLANRESET", Rank: 1, Points: 100}}} + if _, e = clans.ReplaceCurrentClanRankingGroup(ctx, group); e != nil { + t.Fatal(e) + } + group.Rows[0].Points = 200 + if _, e = clans.ReplaceCurrentClanRankingGroup(ctx, group); e != nil { + t.Fatal(e) + } + var points int + if e = pool.QueryRow(ctx, `SELECT points FROM clan_rankings_current WHERE clan_tag='#CLANRESET' AND ranking_type='home' AND location_id='global'`).Scan(&points); e != nil || points != 200 { + t.Fatalf("clan ranking writer: %d %v", points, e) + } + if e = store.playerLeaderboardMaintenance(ctx); e != nil { + t.Fatal(e) + } + exec(`UPDATE basic_player SET trophies=456 WHERE tag='#RESET00001'`) + if e = store.playerLeaderboardMaintenance(ctx); e != nil { + t.Fatal(e) + } + if e = pool.QueryRow(ctx, `SELECT trophies FROM player_league_leaderboards WHERE tag='#RESET00001'`).Scan(&trophies); e != nil || trophies != 123 { + t.Fatalf("refreshed before six hours: %d %v", trophies, e) + } + exec(`UPDATE tracking_scheduled_jobs SET completed_at=now()-interval '7 hours' WHERE job='player_board_refresh'`) + if e = store.playerLeaderboardMaintenance(ctx); e != nil { + t.Fatal(e) + } + if e = pool.QueryRow(ctx, `SELECT trophies FROM player_league_leaderboards WHERE tag='#RESET00001'`).Scan(&trophies); e != nil || trophies != 456 { + t.Fatalf("six-hour refresh: %d %v", trophies, e) + } +} diff --git a/scripts/ranking_ingestion_rollover_test.go b/scripts/ranking_ingestion_rollover_test.go new file mode 100644 index 0000000..17c9a54 --- /dev/null +++ b/scripts/ranking_ingestion_rollover_test.go @@ -0,0 +1,94 @@ +package scripts + +import ( + clashy "github.com/clashkinginc/clashy.go" + "reflect" + "strings" + "testing" + "time" +) + +func TestWardenModePreservedButCompositionUnchanged(t *testing.T) { + a, e := normalizeArmyShareCodeChecked("h2m1p16e5_41u2x0") + if e != nil || a != "h2m1p16e5_41u2x0" { + t.Fatalf("mode lost: %s %v", a, e) + } + b, e := normalizeArmyShareCodeChecked("h2m0p16e5_41u2x0") + if e != nil || a == b { + t.Fatalf("distinct stored modes: %s %v", b, e) + } + if !reflect.DeepEqual(parseArmyColumns(a), parseArmyColumns(b)) { + t.Fatal("mode changed composition columns") + } + for _, bad := range []string{"h2m", "h2m1m0p16", "h2z1p16"} { + if _, e := normalizeArmyShareCodeChecked(bad); e == nil { + t.Fatalf("accepted %s", bad) + } + } +} + +func TestPreviousRankSentinelIsNotGenericNegativeAcceptance(t *testing.T) { + rank, e := optionalPreviousRank(-1) + if rank != nil || e != nil { + t.Fatalf("sentinel: %v %v", rank, e) + } + if _, e = optionalPreviousRank(-2); e == nil { + t.Fatal("accepted invalid rank") + } + if _, e = optionalHistoryPositiveInt(-1); e == nil { + t.Fatal("generic validator relaxed") + } +} + +func TestExpectedMissingLocalRankingsOnly(t *testing.T) { + e := &clashy.NotFound{HTTPException: &clashy.HTTPException{Status: 404, Message: "Rankings not found for location"}} + if !unavailableLocalRanking("32000000", e) || unavailableLocalRanking("global", e) { + t.Fatal("incorrect missing scope") + } + other := &clashy.NotFound{HTTPException: &clashy.HTTPException{Status: 404, Message: "invalid endpoint"}} + if unavailableLocalRanking("32000000", other) { + t.Fatal("hid unrelated failure") + } +} + +func TestIncompleteClanDoesNotRejectPlayerHistory(t *testing.T) { + a, b, c, e := leaderboardPlayerClan(&clashy.PlayerClan{Tag: "#P0Y", Name: "Clan"}) + if e != nil || a != nil || b != nil || c != nil { + t.Fatalf("optional metadata: %v %v %v %v", a, b, c, e) + } +} + +func TestAlreadyArchivedAndOldSeasonIdentifiersAreSkipped(t *testing.T) { + now := time.Date(2026, 9, 17, 0, 0, 0, 0, time.UTC) + got, e := missingCompletedLegendSeasons([]string{"2025-10-06", "2025-09"}, map[string]struct{}{"2025-10-06": {}}, now) + if e != nil || len(got) != 0 { + t.Fatalf("old history retried: %v %v", got, e) + } + got, e = missingCompletedLegendSeasons([]string{"2025-10-06"}, nil, now) + if e != nil || len(got) != 0 { + t.Fatalf("old unlisted history retried: %v %v", got, e) + } +} + +func TestWeeklyResetWindow(t *testing.T) { + for _, test := range []struct { + at string + want bool + }{ + {"2026-09-21T16:29:59Z", false}, {"2026-09-21T16:30:00Z", true}, + {"2026-09-21T16:59:59Z", true}, {"2026-09-21T17:00:00Z", false}, + {"2026-09-22T16:30:00Z", false}, {"2026-09-21T11:30:00-05:00", true}, + } { + at, _ := time.Parse(time.RFC3339, test.at) + _, got := rankedResetWindow(at) + if got != test.want { + t.Errorf("%s: %v", test.at, got) + } + } +} + +func TestCurrentClanRankingsUseExistingSchema(t *testing.T) { + if strings.Contains(replaceCurrentClanRankingGroupSQL, "updated_at") { + t.Fatal("writer references removed column") + } +} diff --git a/scripts/scheduled.go b/scripts/scheduled.go index cc529df..cdd87de 100644 --- a/scripts/scheduled.go +++ b/scripts/scheduled.go @@ -334,6 +334,12 @@ func (d *scheduledDomain) runLeagueCloseoutLoop(ctx context.Context, app *platfo return err } app.Stats.RecordWrite(trackingProgressName(scheduledDomainName, "ranked-closeout"), rankedWrites) + if store, ok := d.store.(*timescaleScheduledStore); ok { + if _, err := store.pool.Exec(ctx, `INSERT INTO tracking_scheduled_jobs(job,period,completed_at) + VALUES('ranked_closeout',$1,clock_timestamp()) ON CONFLICT(job,period) DO UPDATE SET completed_at=EXCLUDED.completed_at`, dayStart(now)); err != nil { + return err + } + } } } } @@ -636,6 +642,7 @@ func (d *scheduledDomain) doLeaderboardHistory( ) ([]leaderboardHistoryGroup, error) { date := dayStart(now) groups := make([]leaderboardHistoryGroup, 0, len(leaderboardHistoryPaths)*len(locationIDs)) + unavailable := 0 var groupErrors []error for _, locationID := range locationIDs { for _, path := range leaderboardHistoryPaths { @@ -649,6 +656,10 @@ func (d *scheduledDomain) doLeaderboardHistory( return payload, err }) if err != nil { + if unavailableLocalRanking(locationID, err) { + unavailable++ + continue + } groupErrors = append( groupErrors, fmt.Errorf("fetch %s leaderboard history for %s: %w", path.Kind, locationID, err), @@ -663,6 +674,9 @@ func (d *scheduledDomain) doLeaderboardHistory( groups = append(groups, group) } } + if unavailable > 0 { + app.Logger.Info("leaderboard history skipped unavailable local rankings", "groups", unavailable) + } return groups, errors.Join(groupErrors...) } @@ -693,12 +707,12 @@ func (d *scheduledDomain) doLegendHistory( return 0, err } missing, err := missingCompletedLegendSeasons(officialSeasons, completed, now) - if err != nil { - return 0, err - } loadRankings := d.loadLegendRankings writes := 0 var seasonErrors []error + if err != nil { + seasonErrors = append(seasonErrors, err) + } for _, season := range missing { var rankings []legendRankingItem if loadRankings == nil { @@ -842,10 +856,34 @@ func missingCompletedLegendSeasons( now = now.UTC() missing := make([]string, 0, len(official)) seen := make(map[string]struct{}, len(official)) + // Routine scheduling is forward-only, not a historical backfill. A 35-day + // startup horizon covers the latest four-week season if no checkpoint exists. + floor := now.AddDate(0, 0, -35) + for season := range completed { + if window, err := officialLegendSeasonWindow(season); err == nil && window.EndTime.After(floor) { + floor = window.EndTime + } + } + var seasonErrors []error for _, season := range official { + // Already archived seasons do not need their legacy identifiers parsed again. + if _, exists := completed[season]; exists { + continue + } + // Old date identifiers are only used as a cutoff here, never interpreted as + // a season boundary. They belong to explicit historical repair tooling. + if len(season) == 10 { + if date, e := time.Parse("2006-01-02", season); e == nil && date.Before(floor) { + continue + } + } window, err := officialLegendSeasonWindow(season) if err != nil { - return nil, err + seasonErrors = append(seasonErrors, err) + continue + } + if !window.EndTime.After(floor) { + continue } if _, duplicate := seen[season]; duplicate { continue @@ -858,7 +896,7 @@ func missingCompletedLegendSeasons( missing = append(missing, season) } } - return missing, nil + return missing, errors.Join(seasonErrors...) } func officialLegendSeasonWindow(season string) (clashy.SeasonWindow, error) { @@ -981,6 +1019,7 @@ func (d *scheduledDomain) doCurrentClanRankings( updatedAt time.Time, ) (int, error) { writes := 0 + unavailable := 0 var groupErrors []error for _, locationID := range locationIDs { for _, path := range currentClanRankingPaths { @@ -991,6 +1030,10 @@ func (d *scheduledDomain) doCurrentClanRankings( return rankings, err }) if err != nil { + if unavailableLocalRanking(locationID, err) { + unavailable++ + continue + } groupErrors = append(groupErrors, fmt.Errorf("fetch %s clan rankings for %s: %w", path.RankingType, locationID, err)) continue } @@ -1013,6 +1056,9 @@ func (d *scheduledDomain) doCurrentClanRankings( writes += count } } + if unavailable > 0 { + app.Logger.Info("current clan rankings skipped unavailable local rankings", "groups", unavailable) + } return writes, errors.Join(groupErrors...) } @@ -1211,7 +1257,7 @@ func playerTrophyHistoryRow( if err != nil { return models.PlayerTrophyHistoryRow{}, fmt.Errorf("player trophy history %s: %w", item.Tag, err) } - previousRank, err := optionalHistoryPositiveInt(item.PreviousRank) + previousRank, err := optionalPreviousRank(item.PreviousRank) if err != nil { return models.PlayerTrophyHistoryRow{}, fmt.Errorf("player trophy history %s previous rank: %w", item.Tag, err) } @@ -1254,7 +1300,7 @@ func playerBuilderBaseTrophyHistoryRow( if err != nil { return models.PlayerBuilderBaseTrophyHistoryRow{}, fmt.Errorf("builder base trophy history %s: %w", item.Tag, err) } - previousRank, err := optionalHistoryPositiveInt(item.PreviousRank) + previousRank, err := optionalPreviousRank(item.PreviousRank) if err != nil { return models.PlayerBuilderBaseTrophyHistoryRow{}, fmt.Errorf("builder base trophy history %s previous rank: %w", item.Tag, err) } @@ -1397,7 +1443,7 @@ func leaderboardClanHistoryValues( item clashy.RankedClan, ) (leaderboardClanHistoryCommon, error) { token := badgeToken(item.Badge) - previousRank, err := optionalHistoryPositiveInt(item.PreviousRank) + previousRank, err := optionalPreviousRank(item.PreviousRank) if err != nil { return leaderboardClanHistoryCommon{}, fmt.Errorf("clan history %s previous rank: %w", item.Tag, err) } @@ -1442,7 +1488,8 @@ func leaderboardPlayerClan( if strings.TrimSpace(clan.Tag) == "" || strings.TrimSpace(clan.Name) == "" || token == "" { - return nil, nil, nil, fmt.Errorf("invalid clan snapshot") + // Clan metadata is optional; it must not discard the player's ranking. + return nil, nil, nil, nil } tag := clan.Tag name := clan.Name @@ -1460,6 +1507,19 @@ func optionalHistoryPositiveInt(value int) (*int, error) { return &result, nil } +func optionalPreviousRank(value int) (*int, error) { + if value == -1 { + return nil, nil + } + return optionalHistoryPositiveInt(value) +} + +func unavailableLocalRanking(location string, err error) bool { + var missing *clashy.NotFound + return location != "global" && errors.As(err, &missing) && + strings.Contains(strings.ToLower(err.Error()), "rankings not found for location") +} + func validatePlayerTrophyHistoryRow(row models.PlayerTrophyHistoryRow) error { if !validLeaderboardHistoryLocation(row.LocationID) || row.Date.IsZero() || @@ -2195,14 +2255,13 @@ func validLegendHistoryClan(row models.LegendHistoryRow) bool { const replaceCurrentClanRankingGroupSQL = ` INSERT INTO clan_rankings_current ( - clan_tag, ranking_type, location_id, rank, points, updated_at + clan_tag, ranking_type, location_id, rank, points ) - SELECT clan_tag, $1, $2, rank, points, updated_at + SELECT clan_tag, $1, $2, rank, points FROM clan_rankings_current_stage ON CONFLICT (clan_tag, ranking_type, location_id) DO UPDATE SET rank = EXCLUDED.rank, - points = EXCLUDED.points, - updated_at = EXCLUDED.updated_at + points = EXCLUDED.points ` const deleteStaleCurrentClanRankingGroupSQL = ` @@ -2238,8 +2297,7 @@ func (s *timescaleScheduledStore) ReplaceCurrentClanRankingGroup( CREATE TEMP TABLE clan_rankings_current_stage ( clan_tag text NOT NULL, rank integer NOT NULL, - points integer NOT NULL, - updated_at timestamp with time zone NOT NULL + points integer NOT NULL ) ON COMMIT DROP `); err != nil { return 0, err @@ -2248,10 +2306,10 @@ func (s *timescaleScheduledStore) ReplaceCurrentClanRankingGroup( if _, err := tx.CopyFrom( ctx, pgx.Identifier{"clan_rankings_current_stage"}, - []string{"clan_tag", "rank", "points", "updated_at"}, + []string{"clan_tag", "rank", "points"}, pgx.CopyFromSlice(len(group.Rows), func(index int) ([]any, error) { row := group.Rows[index] - return []any{row.ClanTag, row.Rank, row.Points, row.UpdatedAt}, nil + return []any{row.ClanTag, row.Rank, row.Points}, nil }), ); err != nil { return 0, err diff --git a/scripts/scheduled_test.go b/scripts/scheduled_test.go index 9ea09ba..36f758f 100644 --- a/scripts/scheduled_test.go +++ b/scripts/scheduled_test.go @@ -707,7 +707,9 @@ func TestMissingCompletedLegendSeasonsUsesExactOfficialIDs(t *testing.T) { if err != nil { t.Fatal(err) } - if want := []string{"2026-04", v2Season}; !reflect.DeepEqual(missing, want) { + // Old seasons are deliberately excluded from routine forward-only discovery. + // Keep the old ID in the input to guard against accidental backfilling. + if want := []string{v2Season}; !reflect.DeepEqual(missing, want) { t.Fatalf("missing completed legend seasons = %#v, want %#v", missing, want) } atBoundary, err := missingCompletedLegendSeasons(