wowlab_analytics/analytics/merge/

sections.rs

use wowlab_types::{
    proto,
    sim::{FastMap, IntMap},
};

use super::{
    MergeError,
    histogram::sat_add_u128,
    keys::{ActionKey, AuraKey},
};

pub(super) fn merge_actions(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    let mut index_map: FastMap<ActionKey, usize> = FastMap::default();

    for (i, row) in state.actions.iter().enumerate() {
        index_map.insert(ActionKey::from(row), i);
    }

    for chunk_row in &chunk.actions {
        let key = ActionKey::from(chunk_row);

        if let Some(&idx) = index_map.get(&key) {
            if let Some(existing) = state.actions.get_mut(idx) {
                existing.casts = existing.casts.saturating_add(chunk_row.casts);
                existing.direct_hits = existing.direct_hits.saturating_add(chunk_row.direct_hits);
                existing.ticks = existing.ticks.saturating_add(chunk_row.ticks);
                existing.crits = existing.crits.saturating_add(chunk_row.crits);
                existing.misses = existing.misses.saturating_add(chunk_row.misses);
                existing.dodges = existing.dodges.saturating_add(chunk_row.dodges);
                existing.parries = existing.parries.saturating_add(chunk_row.parries);
                existing.total_damage_x10 =
                    sat_add_u128(existing.total_damage_x10, chunk_row.total_damage_x10);
                existing.execute_time_ms = existing
                    .execute_time_ms
                    .saturating_add(chunk_row.execute_time_ms);
                existing.resource_spent_x100 =
                    sat_add_u128(existing.resource_spent_x100, chunk_row.resource_spent_x100);
                existing.resource_gained_x100 = sat_add_u128(
                    existing.resource_gained_x100,
                    chunk_row.resource_gained_x100,
                );
            }
        } else {
            let new_idx = state.actions.len();

            state.actions.push(*chunk_row);
            index_map.insert(key, new_idx);
        }
    }
}

pub(super) fn merge_auras(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    let mut index_map: FastMap<AuraKey, usize> = FastMap::default();

    for (i, row) in state.auras.iter().enumerate() {
        index_map.insert(AuraKey::from(row), i);
    }

    for chunk_row in &chunk.auras {
        let key = AuraKey::from(chunk_row);

        if let Some(&idx) = index_map.get(&key) {
            if let Some(existing) = state.auras.get_mut(idx) {
                existing.uptime_ms = existing.uptime_ms.saturating_add(chunk_row.uptime_ms);
                existing.applications =
                    existing.applications.saturating_add(chunk_row.applications);
                existing.refreshes = existing.refreshes.saturating_add(chunk_row.refreshes);
                existing.stack_seconds_x100 =
                    sat_add_u128(existing.stack_seconds_x100, chunk_row.stack_seconds_x100);
            }
        } else {
            let new_idx = state.auras.len();

            state.auras.push(*chunk_row);
            index_map.insert(key, new_idx);
        }
    }
}

pub(super) fn merge_resources(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    let mut index_map: IntMap<u32, usize> = IntMap::default();

    for (i, row) in state.resources.iter().enumerate() {
        index_map.insert(row.resource_type, i);
    }

    for chunk_row in &chunk.resources {
        if let Some(&idx) = index_map.get(&chunk_row.resource_type) {
            if let Some(existing) = state.resources.get_mut(idx) {
                existing.gained_x100 = sat_add_u128(existing.gained_x100, chunk_row.gained_x100);
                existing.spent_x100 = sat_add_u128(existing.spent_x100, chunk_row.spent_x100);
                existing.wasted_x100 = sat_add_u128(existing.wasted_x100, chunk_row.wasted_x100);
                existing.time_at_cap_ms = existing
                    .time_at_cap_ms
                    .saturating_add(chunk_row.time_at_cap_ms);
                existing.starved_time_ms = existing
                    .starved_time_ms
                    .saturating_add(chunk_row.starved_time_ms);

                for source_row in &chunk_row.by_source {
                    if let Some(existing_source) = existing
                        .by_source
                        .iter_mut()
                        .find(|row| row.source_spell_id == source_row.source_spell_id)
                    {
                        existing_source.gained_x100 = existing_source
                            .gained_x100
                            .saturating_add(source_row.gained_x100);
                        existing_source.wasted_x100 = existing_source
                            .wasted_x100
                            .saturating_add(source_row.wasted_x100);
                    } else {
                        existing.by_source.push(*source_row);
                    }
                }
            }
        } else {
            let new_idx = state.resources.len();

            // #t(rust_clone_in_loop) cold per-chunk merge; one clone per resource type
            state.resources.push(chunk_row.clone());
            index_map.insert(chunk_row.resource_type, new_idx);
        }
    }
}

pub(super) fn merge_cooldowns(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    let mut index_map: IntMap<u32, usize> = IntMap::default();

    for (i, row) in state.cooldowns.iter().enumerate() {
        index_map.insert(row.spell_id, i);
    }

    for chunk_row in &chunk.cooldowns {
        if let Some(&idx) = index_map.get(&chunk_row.spell_id) {
            if let Some(existing) = state.cooldowns.get_mut(idx) {
                existing.uses = existing.uses.saturating_add(chunk_row.uses);
                existing.possible_uses = existing
                    .possible_uses
                    .saturating_add(chunk_row.possible_uses);
                existing.drift_sum_ms =
                    existing.drift_sum_ms.saturating_add(chunk_row.drift_sum_ms);
                existing.max_drift_ms = existing.max_drift_ms.max(chunk_row.max_drift_ms);
            }
        } else {
            let new_idx = state.cooldowns.len();

            state.cooldowns.push(*chunk_row);
            index_map.insert(chunk_row.spell_id, new_idx);
        }
    }
}

pub(super) fn merge_execution(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    let Some(chunk_exec) = &chunk.execution else {
        return;
    };

    let state_exec = state.execution.get_or_insert_with(|| proto::ExecutionData {
        active_time_ms: 0,
        idle_time_ms: 0,
        gcd_locked_time_ms: 0,
        idle_gcd_count: 0,
        queue_lag_sum_ms: 0,
        queue_lag_count: 0,
        action_count_per_bucket: vec![],
        action_bucket_samples: vec![],
    });

    state_exec.active_time_ms = state_exec
        .active_time_ms
        .saturating_add(chunk_exec.active_time_ms);
    state_exec.idle_time_ms = state_exec
        .idle_time_ms
        .saturating_add(chunk_exec.idle_time_ms);
    state_exec.gcd_locked_time_ms = state_exec
        .gcd_locked_time_ms
        .saturating_add(chunk_exec.gcd_locked_time_ms);
    state_exec.idle_gcd_count = state_exec
        .idle_gcd_count
        .saturating_add(chunk_exec.idle_gcd_count);
    state_exec.queue_lag_sum_ms = state_exec
        .queue_lag_sum_ms
        .saturating_add(chunk_exec.queue_lag_sum_ms);
    state_exec.queue_lag_count = state_exec
        .queue_lag_count
        .saturating_add(chunk_exec.queue_lag_count);

    let max_len = state_exec
        .action_count_per_bucket
        .len()
        .max(chunk_exec.action_count_per_bucket.len());

    state_exec.action_count_per_bucket.resize(max_len, 0);

    for (slot, &val) in state_exec
        .action_count_per_bucket
        .iter_mut()
        .zip(chunk_exec.action_count_per_bucket.iter())
    {
        *slot = slot.saturating_add(val);
    }

    let max_len = state_exec
        .action_bucket_samples
        .len()
        .max(chunk_exec.action_bucket_samples.len());

    state_exec.action_bucket_samples.resize(max_len, 0);

    for (slot, &val) in state_exec
        .action_bucket_samples
        .iter_mut()
        .zip(chunk_exec.action_bucket_samples.iter())
    {
        *slot = slot.saturating_add(val);
    }
}

pub(super) fn merge_damage_profile(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    let Some(chunk_dp) = &chunk.damage_profile else {
        return;
    };

    let state_dp = state
        .damage_profile
        .get_or_insert_with(|| proto::DamageProfileData {
            direct_damage_x10: 0,
            periodic_damage_x10: 0,
            pet_damage_x10: 0,
            by_target: vec![],
        });

    state_dp.direct_damage_x10 =
        sat_add_u128(state_dp.direct_damage_x10, chunk_dp.direct_damage_x10);
    state_dp.periodic_damage_x10 =
        sat_add_u128(state_dp.periodic_damage_x10, chunk_dp.periodic_damage_x10);
    state_dp.pet_damage_x10 = sat_add_u128(state_dp.pet_damage_x10, chunk_dp.pet_damage_x10);

    let mut index_map: IntMap<u32, usize> = IntMap::default();

    for (i, t) in state_dp.by_target.iter().enumerate() {
        index_map.insert(t.target_id, i);
    }

    for chunk_t in &chunk_dp.by_target {
        if let Some(&idx) = index_map.get(&chunk_t.target_id) {
            if let Some(existing) = state_dp.by_target.get_mut(idx) {
                existing.total_damage_x10 =
                    sat_add_u128(existing.total_damage_x10, chunk_t.total_damage_x10);
            }
        } else {
            let new_idx = state_dp.by_target.len();

            state_dp.by_target.push(*chunk_t);
            index_map.insert(chunk_t.target_id, new_idx);
        }
    }
}

pub(super) fn merge_dps_buckets(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    if state.bucket_ms == 0 && chunk.bucket_ms != 0 {
        state.bucket_ms = chunk.bucket_ms;
    }

    let max_len = state
        .dps_bucket_sums_x10
        .len()
        .max(chunk.dps_bucket_sums_x10.len());

    state.dps_bucket_sums_x10.resize(max_len, 0);

    for (slot, &val) in state
        .dps_bucket_sums_x10
        .iter_mut()
        .zip(chunk.dps_bucket_sums_x10.iter())
    {
        *slot = sat_add_u128(*slot, val);
    }

    let max_len = state
        .dps_bucket_samples
        .len()
        .max(chunk.dps_bucket_samples.len());

    state.dps_bucket_samples.resize(max_len, 0);

    for (slot, &val) in state
        .dps_bucket_samples
        .iter_mut()
        .zip(chunk.dps_bucket_samples.iter())
    {
        *slot = slot.saturating_add(val);
    }
}

pub(super) fn merge_representative(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    // Multi-chunk aggregation has no exact representative: a chunk-local winner isn't the multi-chunk mean.
    if state.chunks_total != 1 {
        state.representative = None;
        state.representative_meta = None;

        return;
    }

    let candidate_dps_x10 = chunk.representative_dps_x10;
    let candidate_timeline = &chunk.representative;

    if candidate_timeline.is_none() {
        return;
    }

    let candidate_chunk_index = chunk.chunk_index;
    let candidate_iteration_index: u32 = 0;

    state.representative.clone_from(candidate_timeline);
    state.representative_meta = Some(proto::RepresentativeMeta {
        chunk_index: candidate_chunk_index,
        iteration_index: candidate_iteration_index,
        winner_dps_x10: candidate_dps_x10,
    });
}

pub(super) fn merge_dictionary(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) -> Result<(), MergeError> {
    let Some(chunk_dict) = &chunk.dictionary else {
        return Ok(());
    };

    let state_dict = state
        .dictionary
        .get_or_insert_with(|| proto::DictionaryView {
            spell_ids: vec![],
            aura_ids: vec![],
            targets: vec![],
            units: vec![],
        });

    for &id in &chunk_dict.spell_ids {
        if !state_dict.spell_ids.contains(&id) {
            state_dict.spell_ids.push(id);
        }
    }

    for &id in &chunk_dict.aura_ids {
        if !state_dict.aura_ids.contains(&id) {
            state_dict.aura_ids.push(id);
        }
    }

    let known_ids: IntMap<u32, usize> = state_dict
        .targets
        .iter()
        .enumerate()
        .map(|(i, t)| (t.id, i))
        .collect();

    let mut new_indices = Vec::with_capacity(chunk_dict.targets.len());

    for (i, chunk_t) in chunk_dict.targets.iter().enumerate() {
        if let Some(existing_index) = known_ids.get(&chunk_t.id).copied() {
            if state_dict.targets.get(existing_index) != Some(chunk_t) {
                return Err(MergeError::dictionary_label_conflict(chunk_t.id));
            }
        } else {
            new_indices.push(i);
        }
    }

    let new_targets: Vec<_> = new_indices
        .iter()
        .filter_map(|&i| chunk_dict.targets.get(i).cloned())
        .collect();

    state_dict.targets.extend(new_targets);

    let mut known_units: FastMap<(i32, u32), usize> = state_dict
        .units
        .iter()
        .enumerate()
        .map(|(index, unit)| ((unit.kind, unit.id), index))
        .collect();
    let mut new_units = Vec::with_capacity(chunk_dict.units.len());

    for unit in &chunk_dict.units {
        let key = (unit.kind, unit.id);

        if let Some(index) = known_units.get(&key).copied() {
            if state_dict.units.get(index) != Some(unit) {
                return Err(MergeError::dictionary_label_conflict(
                    unit.target_id.unwrap_or(unit.id),
                ));
            }
        } else {
            known_units.insert(key, state_dict.units.len() + new_units.len());
            new_units.push(unit);
        }
    }

    state_dict.units.extend(new_units.into_iter().cloned());

    Ok(())
}

pub(super) fn merge_instrumentation_coverage(
    state: &mut proto::RunningAggregateStateV1,
    chunk: &proto::ChunkTelemetry,
) {
    let Some(chunk_ic) = &chunk.instrumentation_coverage else {
        return;
    };

    let state_ic = state
        .instrumentation_coverage
        .get_or_insert(proto::InstrumentationCoverage {
            has_resource_cap_time: true,
            has_resource_starvation_time: true,
            has_cooldown_drift: true,
            has_phase_markers: true,
            has_proc_aggregates: true,
        });

    state_ic.has_resource_cap_time =
        state_ic.has_resource_cap_time && chunk_ic.has_resource_cap_time;
    state_ic.has_resource_starvation_time =
        state_ic.has_resource_starvation_time && chunk_ic.has_resource_starvation_time;
    state_ic.has_cooldown_drift = state_ic.has_cooldown_drift && chunk_ic.has_cooldown_drift;
    state_ic.has_phase_markers = state_ic.has_phase_markers && chunk_ic.has_phase_markers;
    state_ic.has_proc_aggregates = state_ic.has_proc_aggregates && chunk_ic.has_proc_aggregates;
}