wowlab_engine_sim/telemetry/

mod.rs

//! Cross-iteration telemetry accumulation, capture, encoding, and parallel merge.

mod encode;
mod merge;
mod representative;

use hdrhistogram::Histogram;
use itertools::Itertools;
use representative::RepresentativeTimeline;
use wowlab_engine_telemetry::{
    AuraEvent, AuraEventKind, AuraTelemetryScope, ResourceEventKind, TargetMetadata,
    TelemetryScope, TelemetrySink,
};
use wowlab_types::{
    constants::MS_PER_SECOND,
    numeric::{f64_to_u32_saturating_trunc, f64_to_u64_saturating_round},
    sim::{ActorId, EnemyIdx, FastMap, IntMap, TargetIdx},
};

pub(crate) const TIMELINE_BUCKET_MS: u32 = 1_000;

pub(crate) const HISTOGRAM_SIGNIFICANT_DIGITS: u8 = 3;
pub(crate) const HISTOGRAM_DPS_SCALE: f64 = 10.0;

const HISTOGRAM_MIN_SAMPLE: u64 = 1;
const ENEMY_ACTOR_SORT_ORDER: u8 = 3;
const PET_ACTOR_SORT_ORDER: u8 = 2;

#[expect(
    clippy::cast_possible_truncation,
    reason = "representative previews deliberately trade precision for compact f32 storage"
)]
pub(super) const fn telemetry_preview_f32(value: f64) -> f32 {
    // #t(rust_lossy_cast) representative previews deliberately use compact f32 storage
    value as f32
}

pub(crate) const fn actor_sort_key(actor: ActorId) -> (u8, u16) {
    match actor {
        ActorId::Player => (0, 0),
        ActorId::External => (1, 0),
        ActorId::Pet(pet) => (PET_ACTOR_SORT_ORDER, pet.0),
        ActorId::Enemy(enemy) => (ENEMY_ACTOR_SORT_ORDER, enemy.0),
    }
}

#[inline]
pub(crate) fn dps_histogram_sample(dps: f64) -> u64 {
    let x10 = (dps * HISTOGRAM_DPS_SCALE).round();

    if x10 < 1.0 {
        HISTOGRAM_MIN_SAMPLE
    } else {
        f64_to_u64_saturating_round(x10)
    }
}

pub(crate) fn new_dps_histogram() -> Histogram<u64> {
    Histogram::new(HISTOGRAM_SIGNIFICANT_DIGITS)
        .expect("HISTOGRAM_SIGNIFICANT_DIGITS is within the valid 0..=5 range")
}

pub(crate) fn collect_representative_aura_intervals(
    events: &[AuraEvent],
    mut emit: impl FnMut(&AuraEvent, u32, Option<u32>),
) {
    let mut starts: FastMap<AuraAggregateKey, RepresentativeAuraIntervalStart<'_>> =
        FastMap::default();

    for event in events {
        let key = AuraAggregateKey::from(event);

        match event.kind {
            AuraEventKind::Apply { .. } | AuraEventKind::Refresh { .. } => {
                starts
                    .entry(key)
                    .or_insert(RepresentativeAuraIntervalStart {
                        time_ms: event.time_ms,
                        event,
                    });
            }
            AuraEventKind::Expire => {
                if let Some(start) = starts.remove(&key) {
                    emit(event, start.time_ms, Some(event.time_ms));
                }
            }
        }
    }

    for (_, start) in starts.drain().sorted_by_key(|(key, start)| {
        (
            key.aura_id,
            actor_sort_key(key.scope.affected),
            actor_sort_key(key.scope.source),
            key.scope.pull,
            key.scope.group,
            start.time_ms,
        )
    }) {
        emit(start.event, start.time_ms, None);
    }
}

struct RepresentativeAuraIntervalStart<'a> {
    time_ms: u32,
    event: &'a AuraEvent,
}

#[derive(Debug, Default)]
pub(crate) struct ResourceSourceAggregate {
    pub(crate) gained: f64,
    pub(crate) wasted: f64,
}

#[derive(Debug, Default)]
pub(crate) struct ResourceAggregate {
    pub(crate) total_gained: f64,
    pub(crate) total_spent: f64,
    pub(crate) total_wasted: f64,
    /// Income by gaining spell id, or a reserved `RESOURCE_SOURCE_*` id for regen and auto-attacks.
    pub(crate) gained_by_source: IntMap<u32, ResourceSourceAggregate>,
}

#[derive(Clone, Debug, Default)]
pub(crate) struct SpellAggregate {
    pub(crate) damage: f64,
    pub(crate) casts: u32,
    pub(crate) crits: u32,
    pub(crate) hits: u32,
    pub(crate) ticks: u32,
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub(crate) struct ActionAggregateKey {
    pub(crate) spell_id: u32,
    pub(crate) scope: TelemetryScope,
}

impl ActionAggregateKey {
    const fn from_scope(spell_id: u32, scope: TelemetryScope) -> Self {
        Self { spell_id, scope }
    }
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub(crate) struct AuraAggregateKey {
    pub(crate) aura_id: u32,
    pub(crate) scope: AuraTelemetryScope,
}

impl From<&AuraEvent> for AuraAggregateKey {
    fn from(event: &AuraEvent) -> Self {
        Self {
            aura_id: event.aura_id,
            scope: AuraTelemetryScope {
                source: event.source,
                affected: event.affected,
                pull: event.pull,
                group: event.group,
            },
        }
    }
}

#[derive(Clone, Debug, Default)]
pub(crate) struct AuraAggregate {
    pub(crate) uptime_ms: f64,
    pub(crate) applications: u64,
    pub(crate) refreshes: u64,
}

#[derive(Debug, Default)]
pub(crate) struct CooldownAggregate {
    pub(crate) total_uses: u32,
}

/// Running DPS statistics snapshot.
#[derive(Clone, Debug)]
pub struct RunningDpsStats {
    pub mean_dps: f64,
    /// Population standard deviation of DPS across completed iterations.
    pub std_dps: f64,
    pub min_dps: f64,
    pub max_dps: f64,
}

/// Deterministic identity and DPS for one completed iteration, the only per-iteration state retained for representative selection.
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct RepresentativeIteration {
    pub seed_prefix: u64,
    pub iteration: u32,
    pub dps: f64,
}

/// Accumulates telemetry incrementally across all iterations.
#[derive(Debug)]
// docref:start metrics-telemetry-accumulator
pub struct TelemetryAccumulator {
    pub(crate) iteration_count: u32,
    pub(crate) dps_sum: f64,
    pub(crate) dps_min: f64,
    pub(crate) dps_max: f64,
    pub(crate) spell_totals: FastMap<ActionAggregateKey, SpellAggregate>,
    pub(crate) dps_welford_mean: f64,
    pub(crate) dps_m2: f64,
    pub(crate) dps_histogram: Histogram<u64>,
    pub(crate) representative_dps: Option<f64>,
    pub(crate) representative_candidates: Vec<RepresentativeIteration>,
    pub(crate) aura_totals: FastMap<AuraAggregateKey, AuraAggregate>,
    aura_interval_starts: FastMap<AuraAggregateKey, u32>,
    pub(crate) resource_totals: IntMap<u8, ResourceAggregate>,
    pub(crate) cooldown_totals: IntMap<u32, CooldownAggregate>,
    pub(crate) total_duration_ms: f64,
    pub(crate) gcd_locked_ms_total: f64,
    pub(crate) representative_sink: Option<RepresentativeTimeline>,
    pub(crate) direct_damage_total: f64,
    pub(crate) periodic_damage_total: f64,
    pub(crate) pet_damage_total: f64,
    pub(crate) damage_by_target: FastMap<EnemyIdx, f64>,
    pub(crate) targets: FastMap<TargetIdx, TargetMetadata>,
    pub(crate) bucket_sums: Vec<f64>,
    /// Per-bucket sample count: divide each `bucket_sums` entry by this, not by total iterations.
    pub(crate) bucket_samples: Vec<u64>,
    pub(crate) trace_extras_enabled: bool,
}
// docref:end metrics-telemetry-accumulator

impl TelemetryAccumulator {
    /// Empty accumulator.
    #[must_use]
    pub fn new() -> Self {
        Self {
            iteration_count: 0,
            dps_sum: 0.0,
            dps_min: f64::MAX,
            dps_max: f64::MIN,
            spell_totals: FastMap::default(),
            dps_welford_mean: 0.0,
            dps_m2: 0.0,
            dps_histogram: new_dps_histogram(),
            representative_dps: None,
            representative_candidates: Vec::new(),
            aura_totals: FastMap::default(),
            aura_interval_starts: FastMap::default(),
            resource_totals: IntMap::default(),
            cooldown_totals: IntMap::default(),
            total_duration_ms: 0.0,
            gcd_locked_ms_total: 0.0,
            representative_sink: None,
            direct_damage_total: 0.0,
            periodic_damage_total: 0.0,
            pet_damage_total: 0.0,
            damage_by_target: FastMap::default(),
            targets: FastMap::default(),
            bucket_sums: Vec::new(),
            bucket_samples: Vec::new(),
            trace_extras_enabled: false,
        }
    }

    /// Enable resource-sample + aura-interval collection.
    pub fn enable_trace_extras(&mut self) {
        self.trace_extras_enabled = true;
    }

    /// Running DPS snapshot; `None` before any iterations.
    #[must_use]
    pub fn running_stats(&self) -> Option<RunningDpsStats> {
        if self.iteration_count == 0 {
            return None;
        }

        let n = f64::from(self.iteration_count);
        let mean = self.dps_sum / n;
        let variance = self.dps_m2 / n;
        let std = if variance > 0.0 { variance.sqrt() } else { 0.0 };

        Some(RunningDpsStats {
            mean_dps: mean,
            std_dps: std,
            min_dps: self.dps_min,
            max_dps: self.dps_max,
        })
    }

    /// Record one completed iteration without externally supplied RNG identity.
    #[inline]
    pub fn record_iteration(&mut self, dps: f64) {
        let iteration = self.iteration_count;

        self.record_iteration_for(0, iteration, dps);
    }

    /// Return the completed iteration nearest the final aggregate mean, tie-broken by `(seed_prefix, iteration)` for determinism.
    #[must_use]
    pub fn representative_iteration(&self) -> Option<RepresentativeIteration> {
        let mean = self.running_stats()?.mean_dps;

        // docref:start metrics-representative-pick
        self.representative_candidates
            .iter()
            .copied()
            .min_by(|a, b| {
                let distance_order = (a.dps - mean).abs().total_cmp(&(b.dps - mean).abs());
                distance_order
                    .then_with(|| a.seed_prefix.cmp(&b.seed_prefix))
                    .then_with(|| a.iteration.cmp(&b.iteration))
            })
        // docref:end metrics-representative-pick
    }

    /// DPS of the installed exact representative, or `None` until its capture-only rerun completes.
    #[must_use]
    pub const fn representative_dps(&self) -> Option<f64> {
        self.representative_dps
    }

    /// Aggregate sink events into cross-iteration totals.
    pub(crate) fn record_sink_events(&mut self, sink: &TelemetrySink, duration_secs: f64) {
        let duration_ms = duration_secs * MS_PER_SECOND;

        self.total_duration_ms += duration_ms;
        let duration_ms_u32 = f64_to_u32_saturating_trunc(duration_ms);

        let num_buckets = duration_ms_u32.div_ceil(TIMELINE_BUCKET_MS) as usize;

        if self.bucket_sums.len() < num_buckets {
            self.bucket_sums.resize(num_buckets, 0.0);
        }

        if self.bucket_samples.len() < num_buckets {
            self.bucket_samples.resize(num_buckets, 0);
        }

        for slot in self.bucket_samples.iter_mut().take(num_buckets) {
            *slot += 1;
        }

        self.record_damage_events(sink, duration_ms_u32);
        self.record_aura_events(sink, duration_ms_u32);
        self.record_resource_events(sink);

        for event in &sink.cooldowns {
            let entry = self.cooldown_totals.entry(event.spell_id).or_default();

            entry.total_uses += 1;
        }

        self.record_action_events(sink);

        if self.targets.is_empty() {
            self.targets.extend(
                sink.targets
                    .iter()
                    .cloned()
                    .map(|target| (target.id, target)),
            );
        }
    }

    #[inline]
    pub(crate) fn record_iteration_for(&mut self, seed_prefix: u64, iteration: u32, dps: f64) {
        self.iteration_count += 1;
        self.dps_sum += dps;

        if dps < self.dps_min {
            self.dps_min = dps;
        }

        if dps > self.dps_max {
            self.dps_max = dps;
        }

        let delta = dps - self.dps_welford_mean;

        self.dps_welford_mean += delta / f64::from(self.iteration_count);
        let delta2 = dps - self.dps_welford_mean;

        self.dps_m2 += delta * delta2;

        let _ = self.dps_histogram.record(dps_histogram_sample(dps));

        self.representative_candidates
            .push(RepresentativeIteration {
                seed_prefix,
                iteration,
                dps,
            });
    }

    pub(crate) fn install_representative(
        &mut self,
        dps: f64,
        sink: &TelemetrySink,
        duration_secs: f64,
    ) {
        self.representative_dps = Some(dps);
        let duration_ms = f64_to_u32_saturating_trunc(duration_secs * MS_PER_SECOND);

        self.capture_representative(sink, duration_ms);
    }

    fn record_damage_events(&mut self, sink: &TelemetrySink, duration_ms: u32) {
        for event in &sink.damage {
            let amount = event.amount;

            if event.is_pet {
                self.pet_damage_total += amount;
            } else if event.is_periodic {
                self.periodic_damage_total += amount;
            } else {
                self.direct_damage_total += amount;
            }

            *self
                .damage_by_target
                .entry(event.scope.target)
                .or_insert(0.0) += amount;

            // At an exact bucket boundary, place the terminal damage event in the final elapsed interval, not one past the allocation.
            let timeline_ms = if event.time_ms == duration_ms && duration_ms > 0 {
                event.time_ms - 1
            } else {
                event.time_ms
            };
            let bucket = (timeline_ms / TIMELINE_BUCKET_MS) as usize;

            if let Some(slot) = self.bucket_sums.get_mut(bucket) {
                *slot += amount;
            }
        }
    }

    fn record_aura_events(&mut self, sink: &TelemetrySink, duration_ms: u32) {
        self.aura_interval_starts.clear();

        for event in &sink.auras {
            let key = AuraAggregateKey::from(event);
            let aggregate = self.aura_totals.entry(key).or_default();

            match event.kind {
                AuraEventKind::Apply { .. } => {
                    aggregate.applications += 1;
                    self.aura_interval_starts
                        .entry(key)
                        .or_insert(event.time_ms);
                }
                AuraEventKind::Refresh { .. } => {
                    aggregate.refreshes += 1;
                    self.aura_interval_starts
                        .entry(key)
                        .or_insert(event.time_ms);
                }
                AuraEventKind::Expire => {
                    if let Some(start) = self.aura_interval_starts.remove(&key) {
                        aggregate.uptime_ms += f64::from(event.time_ms.saturating_sub(start));
                    }
                }
            }
        }

        for (key, start) in self.aura_interval_starts.drain() {
            self.aura_totals.entry(key).or_default().uptime_ms +=
                f64::from(duration_ms.saturating_sub(start));
        }
    }

    fn record_resource_events(&mut self, sink: &TelemetrySink) {
        for event in &sink.resources {
            let entry = self.resource_totals.entry(event.resource_type).or_default();

            match event.kind {
                ResourceEventKind::Spend => entry.total_spent += event.amount,
                ResourceEventKind::Gain { wasted } => {
                    entry.total_gained += event.amount;
                    entry.total_wasted += wasted;
                    let source = entry
                        .gained_by_source
                        .entry(event.source_spell_id)
                        .or_default();

                    source.gained += event.amount;
                    source.wasted += wasted;
                }
            }
        }
    }

    fn record_action_events(&mut self, sink: &TelemetrySink) {
        for cast in &sink.casts {
            let entry = self
                .spell_totals
                .entry(ActionAggregateKey::from_scope(cast.spell_id, cast.scope))
                .or_default();

            entry.casts += 1;
            self.gcd_locked_ms_total += f64::from(cast.gcd_ms);
        }

        for damage in &sink.damage {
            let entry = self
                .spell_totals
                .entry(ActionAggregateKey::from_scope(
                    damage.spell_id,
                    damage.scope,
                ))
                .or_default();

            entry.damage += damage.amount;
            // `hits` and `ticks` partition every damage event; `crits` is a subset of the two, so
            // `hits + ticks` is the row's total event count and never double-counts a critical.
            entry.ticks += u32::from(damage.is_periodic);
            entry.crits += u32::from(damage.is_crit);
            entry.hits += u32::from(!damage.is_periodic);
        }
    }
}

impl Default for TelemetryAccumulator {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
#[path = "tests.rs"]
mod tests;