wowlab_sentinel/cron/

rollup.rs

#![expect(
    clippy::cast_precision_loss,
    clippy::items_after_statements,
    reason = "rollup statistics intentionally convert bounded database aggregates to f64 and colocate query row types"
)]

use async_trait::async_trait;
use chrono::{Duration, NaiveDate, Utc};
use wowlab_types::{constants::PROTO_DPS_SCALE, sim::FastMap};

use super::CronJob;
use crate::state::ServerState;

const VARIANCE_EXPONENT: i32 = 2;
const ROLLING_LOOKBACK_DAYS: i64 = 30;
const ROLLING_WINDOWS: [i64; 3] = [7, 14, 30];

#[derive(Debug)]
pub(crate) struct RollupDailyJob {
    schedule: String,
}

impl RollupDailyJob {
    pub(crate) fn new(schedule: &str) -> Self {
        Self {
            schedule: schedule.to_owned(),
        }
    }
}

#[async_trait]
impl CronJob for RollupDailyJob {
    fn name(&self) -> &'static str {
        "rollup_daily"
    }

    fn schedule(&self) -> &str {
        &self.schedule
    }

    async fn run(&self, state: &ServerState) {
        let yesterday = Utc::now().date_naive() - Duration::days(1);

        if let Err(e) = run_daily_rollup(state, yesterday).await {
            tracing::error!(error = %e, day = %yesterday, "Daily rollup failed");
        }
    }
}

async fn run_daily_rollup(
    state: &ServerState,
    day: NaiveDate,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let start = day
        .and_hms_opt(0, 0, 0)
        .ok_or("invalid midnight time for start")?
        .and_utc();
    let end = (day + Duration::days(1))
        .and_hms_opt(0, 0, 0)
        .ok_or("invalid midnight time for end")?
        .and_utc();

    struct JobRow {
        season_id: String,
        spec_id: i32,
        result_pb: Vec<u8>,
    }

    let rows = sqlx::query_file_as!(JobRow, "queries/rollup_daily_fetch_jobs.sql", start, end)
        .fetch_all(state.dbs.get::<crate::cron::CronDb>())
        .await?;

    if rows.is_empty() {
        tracing::debug!(day = %day, "No completed jobs for daily rollup");

        return Ok(());
    }

    // #t(rust_similar_structs) decoded database rows include iteration weight unlike engine running-stat snapshots
    struct Decoded {
        iterations: u64,
        mean_dps: f64,
        std_dps: f64,
        min_dps: f64,
        max_dps: f64,
    }

    let mut decode_failures = 0usize;
    let groups: FastMap<(String, i32), Vec<Decoded>> = rows
        .iter()
        .filter_map(|row| {
            if let Ok(core) = super::decode_result_core(row.result_pb.as_slice()) {
                Some((
                    (row.season_id.clone(), row.spec_id),
                    Decoded {
                        iterations: core.iterations,
                        mean_dps: f64::from(core.mean_dps_x10) / PROTO_DPS_SCALE,
                        std_dps: f64::from(core.std_dps_x10) / PROTO_DPS_SCALE,
                        min_dps: f64::from(core.min_dps_x10) / PROTO_DPS_SCALE,
                        max_dps: f64::from(core.max_dps_x10) / PROTO_DPS_SCALE,
                    },
                ))
            } else {
                decode_failures += 1;

                None
            }
        })
        .fold(FastMap::default(), |mut groups, (key, decoded)| {
            groups.entry(key).or_default().push(decoded);

            groups
        });

    if decode_failures > 0 {
        tracing::warn!(decode_failures, "Failed to decode daily rollup results");
    }

    let mut upserted = 0u32;

    for ((season_id, spec_id), jobs) in &groups {
        let jobs_count = i64::try_from(jobs.len()).unwrap_or(i64::MAX);
        let total_iters: f64 = jobs.iter().map(|j| j.iterations as f64).sum();

        // #t(rust_floating_point_eq) exact zero-check is intentional division-by-zero guard
        if total_iters == 0.0 {
            continue;
        }

        let mean_dps = jobs
            .iter()
            .map(|j| j.mean_dps * j.iterations as f64)
            .sum::<f64>()
            / total_iters;

        let min_dps = jobs.iter().map(|j| j.min_dps).fold(f64::INFINITY, f64::min);
        let max_dps = jobs
            .iter()
            .map(|j| j.max_dps)
            .fold(f64::NEG_INFINITY, f64::max);

        let pooled_var = jobs
            .iter()
            .map(|j| {
                let n = j.iterations as f64;
                let within = j.std_dps * j.std_dps;
                let between = (j.mean_dps - mean_dps).powi(VARIANCE_EXPONENT);

                n * (within + between)
            })
            .sum::<f64>()
            / total_iters;
        let std_dps = pooled_var.sqrt();

        sqlx::query_file!(
            "queries/rollup_daily_upsert.sql",
            season_id.as_str(),
            *spec_id,
            day,
            jobs_count,
            mean_dps,
            std_dps,
            min_dps,
            max_dps
        )
        .execute(state.dbs.get::<crate::cron::CronDb>())
        .await?;

        upserted += 1;
    }

    tracing::info!(
        day = %day,
        specs = upserted,
        jobs = rows.len(),
        "Daily rollup completed"
    );

    Ok(())
}

#[derive(Debug)]
pub(crate) struct RollupRollingJob {
    schedule: String,
}

impl RollupRollingJob {
    pub(crate) fn new(schedule: &str) -> Self {
        Self {
            schedule: schedule.to_owned(),
        }
    }
}

#[async_trait]
impl CronJob for RollupRollingJob {
    fn name(&self) -> &'static str {
        "rollup_rolling"
    }

    fn schedule(&self) -> &str {
        &self.schedule
    }

    async fn run(&self, state: &ServerState) {
        let anchor = Utc::now().date_naive();

        if let Err(e) = run_rolling_rollup(state, anchor).await {
            tracing::error!(error = %e, anchor = %anchor, "Rolling rollup failed");
        }
    }
}

async fn run_rolling_rollup(
    state: &ServerState,
    anchor_day: NaiveDate,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let lower = anchor_day - Duration::days(ROLLING_LOOKBACK_DAYS);

    struct DailyRow {
        season_id: String,
        spec_id: i32,
        day: NaiveDate,
        jobs_count: i64,
        mean_dps: f64,
        std_dps: f64,
        min_dps: f64,
        max_dps: f64,
    }

    let rows = sqlx::query_file_as!(
        DailyRow,
        "queries/rollup_rolling_fetch_daily.sql",
        lower,
        anchor_day
    )
    .fetch_all(state.dbs.get::<crate::cron::CronDb>())
    .await?;

    if rows.is_empty() {
        tracing::debug!(anchor = %anchor_day, "No daily rollups for rolling windows");

        return Ok(());
    }

    let groups: FastMap<(String, i32), Vec<&DailyRow>> =
        rows.iter().fold(FastMap::default(), |mut groups, row| {
            groups
                .entry((row.season_id.clone(), row.spec_id))
                .or_default()
                .push(row);

            groups
        });

    let windows = ROLLING_WINDOWS;
    let mut upserted = 0u32;

    for ((season_id, spec_id), daily_rows) in &groups {
        for &window in &windows {
            let cutoff = anchor_day - Duration::days(window);
            let in_window: Vec<&DailyRow> = daily_rows
                .iter()
                .copied()
                .filter(|r| r.day > cutoff && r.day <= anchor_day)
                .collect();

            if in_window.is_empty() {
                continue;
            }

            let total_jobs: i64 = in_window.iter().map(|r| r.jobs_count).sum();

            if total_jobs == 0 {
                continue;
            }

            let grand_mean = in_window
                .iter()
                .map(|r| r.mean_dps * r.jobs_count as f64)
                .sum::<f64>()
                / total_jobs as f64;

            let min_dps = in_window
                .iter()
                .map(|r| r.min_dps)
                .fold(f64::INFINITY, f64::min);
            let max_dps = in_window
                .iter()
                .map(|r| r.max_dps)
                .fold(f64::NEG_INFINITY, f64::max);

            let pooled_var = in_window
                .iter()
                .map(|r| {
                    let n = r.jobs_count as f64;
                    let within = r.std_dps * r.std_dps;
                    let between = (r.mean_dps - grand_mean).powi(VARIANCE_EXPONENT);

                    n * (within + between)
                })
                .sum::<f64>()
                / total_jobs as f64;
            let std_dps = pooled_var.sqrt();

            sqlx::query_file!(
                "queries/rollup_rolling_upsert.sql",
                season_id.as_str(),
                *spec_id,
                anchor_day,
                i32::try_from(window).unwrap_or(i32::MAX),
                total_jobs,
                grand_mean,
                std_dps,
                min_dps,
                max_dps
            )
            .execute(state.dbs.get::<crate::cron::CronDb>())
            .await?;

            upserted += 1;
        }
    }

    tracing::info!(
        anchor = %anchor_day,
        upserted,
        "Rolling rollup completed"
    );

    Ok(())
}