wowlab_sentinel/cron/
ranking.rs

1use async_trait::async_trait;
2use chrono::{DateTime, Utc};
3use uuid::Uuid;
4use wowlab_types::{constants::PROTO_DPS_SCALE, sim::FastMap};
5
6use super::CronJob;
7use crate::state::ServerState;
8
9#[derive(Debug)]
10pub(super) struct RankingRefreshJob {
11    pub schedule: String,
12    pub top_n: usize,
13}
14
15#[async_trait]
16impl CronJob for RankingRefreshJob {
17    fn name(&self) -> &'static str {
18        "ranking_refresh"
19    }
20
21    fn schedule(&self) -> &str {
22        &self.schedule
23    }
24
25    async fn run(&self, state: &ServerState) {
26        if let Err(e) = refresh_rankings(state, self.top_n).await {
27            tracing::error!(error = %e, "Ranking refresh failed");
28        }
29    }
30}
31
32struct ScoredJob {
33    id: Uuid,
34    mean_dps: f64,
35    std_dps: f64,
36    completed_at: DateTime<Utc>,
37}
38
39struct JobRow {
40    id: Uuid,
41    season_id: String,
42    spec_id: i32,
43    result_pb: Vec<u8>,
44    completed_at: DateTime<Utc>,
45}
46
47async fn refresh_rankings(
48    state: &ServerState,
49    top_n: usize,
50) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
51    let rows = sqlx::query_file_as!(JobRow, "queries/ranking_fetch_completed_jobs.sql")
52        .fetch_all(state.dbs.get::<crate::cron::CronDb>())
53        .await?;
54
55    if rows.is_empty() {
56        tracing::debug!("No completed jobs for ranking refresh");
57
58        return Ok(());
59    }
60
61    let (groups, decode_failures) = group_jobs(&rows);
62
63    if decode_failures > 0 {
64        tracing::warn!(decode_failures, "Failed to decode ranking results");
65    }
66
67    let snapshot_at = Utc::now();
68    let mut total_ranked = 0u32;
69
70    for ((season_id, spec_id), mut jobs) in groups {
71        jobs.sort_by(|a, b| {
72            b.mean_dps
73                .partial_cmp(&a.mean_dps)
74                .unwrap_or(std::cmp::Ordering::Equal)
75                .then(
76                    a.std_dps
77                        .partial_cmp(&b.std_dps)
78                        .unwrap_or(std::cmp::Ordering::Equal),
79                )
80                .then(b.completed_at.cmp(&a.completed_at))
81                .then(a.id.cmp(&b.id))
82        });
83
84        jobs.truncate(top_n);
85
86        let mut tx = state.dbs.get::<crate::cron::CronDb>().begin().await?;
87
88        sqlx::query_file!("queries/ranking_delete.sql", &season_id, spec_id)
89            .execute(&mut *tx)
90            .await?;
91
92        for (i, job) in jobs.iter().enumerate() {
93            let rank = i32::try_from(i + 1).unwrap_or(i32::MAX);
94
95            sqlx::query_file!(
96                "queries/ranking_insert.sql",
97                &season_id,
98                spec_id,
99                rank,
100                job.id,
101                job.mean_dps
102            )
103            .execute(&mut *tx)
104            .await?;
105
106            sqlx::query_file!(
107                "queries/ranking_snapshot_insert.sql",
108                &season_id,
109                spec_id,
110                snapshot_at,
111                rank,
112                job.id,
113                job.mean_dps
114            )
115            .execute(&mut *tx)
116            .await?;
117        }
118
119        tx.commit().await?;
120        total_ranked = total_ranked.saturating_add(u32::try_from(jobs.len()).unwrap_or(u32::MAX));
121    }
122
123    tracing::info!(ranked_jobs = total_ranked, "Ranking refresh completed");
124
125    Ok(())
126}
127
128fn group_jobs(rows: &[JobRow]) -> (FastMap<(String, i32), Vec<ScoredJob>>, usize) {
129    let mut groups: FastMap<(String, i32), Vec<ScoredJob>> = FastMap::default();
130    let mut decode_failures = 0;
131
132    for row in rows {
133        match super::decode_result_core(row.result_pb.as_slice()) {
134            Ok(core) => {
135                groups
136                    // #t(rust_clone_in_loop) season_id must be owned for HashMap key
137                    .entry((row.season_id.clone(), row.spec_id))
138                    .or_default()
139                    .push(ScoredJob {
140                        id: row.id,
141                        mean_dps: f64::from(core.mean_dps_x10) / PROTO_DPS_SCALE,
142                        std_dps: f64::from(core.std_dps_x10) / PROTO_DPS_SCALE,
143                        completed_at: row.completed_at,
144                    });
145            }
146            Err(_) => decode_failures += 1,
147        }
148    }
149
150    (groups, decode_failures)
151}
wowlab_sentinel/cron/ranking.rs

wowlab_sentinel/cron/
ranking.rs
