brk_computer 0.11.0

use brk_cohort::{ByAddrType, zip2_by_addr_type};
use brk_error::Result;
use brk_indexer::Lengths;
use brk_traversable::Traversable;
use brk_types::{PartsPerMillion32, StoredF32, StoredU32, StoredU64, Version};
use rayon::prelude::*;
use vecdb::{AnyStoredVec, AnyVec, Database, Exit, Rw, StorageMode, WritableVec};

use crate::{
    indexes, inputs,
    internal::{
        CachedBlockCountReader, CachedWindowStartVec, CountPerBlockRollingAverage,
        LazyPercentCumulativeRolling, PerBlockCumulativeRolling, PerBlockRollingAverage, RatioU64,
        Windows, WithAddrTypes,
    },
    outputs,
};

use super::state::AddrTypeToAddrEventCount;

/// Per-block reused-address event metrics. Holds three families of
/// signals: output-level (use), input-level (spend), and address-level
/// (active in block).
///
/// `output_to_reused_addr_count`: every output landing on an address that had
/// already received at least one prior output anywhere in its lifetime,
/// i.e. an output-level reuse event. Outputs are not deduplicated per
/// address within a block: an address receiving N outputs in one block
/// that had `before` lifetime outputs contributes
/// `max(0, N - max(0, 1 - before))` events. Only the very first output
/// an address ever sees is excluded. Every subsequent output counts,
/// matching the standard "% of outputs to previously-used addresses"
/// reuse ratio reported by external sources. `output_to_reused_addr_share`
/// uses `outputs::ByTypeVecs::output_count` (all 12 output types) as
/// denominator. `spendable_output_to_reused_addr_share` uses the
/// op_return-excluded 11-type aggregate (`spendable_output_count`).
///
/// `input_from_reused_addr_count`: every input spending from an address
/// whose lifetime `funded_txo_count > 1` at the time of the spend (i.e.
/// the address is in the same reused set tracked by
/// `reused_addr_count`). Every input is checked independently. If a
/// single address has multiple inputs in one block each one counts.
/// This is a *stable-predicate* signal about the sending address, not
/// an output-level repeat event: the first spend from a reused address
/// counts just as much as the tenth. Denominator
/// (`input_from_reused_addr_share`): `inputs::ByTypeVecs::input_count` (11
/// spendable types, where `p2ms`, `unknown`, `empty` count as true
/// negatives).
///
/// `active_reused_addr_count` / `active_reused_addr_share`: block-level
/// *address* signals (single aggregate, not per-type).
/// `active_reused_addr_count` is the count of distinct addresses
/// involved in this block (sent ∪ received) that satisfy `is_reused()`
/// after the block's events, populated inline in `process_received`
/// (each receiver, post-receive) and in `process_sent` (each
/// first-encounter sender, deduped against `received_addrs` so
/// addresses that did both aren't double-counted).
/// `active_reused_addr_share` is the per-block ratio
/// `reused / active * 100` as a percentage in `[0, 100]` (or `0.0` for
/// empty blocks). The denominator (distinct active addrs per block)
/// lives on `ActivityCountVecs::active` (`addrs.activity.all.active`),
/// derived from `sending + receiving - bidirectional`. Both fields
/// expose lazy 24h/1w/1m/1y rolling *averages* of the per-block values.
/// Sums and cumulatives of distinct-address counts would be misleading
/// because the same address can appear in multiple blocks, so the
/// cumulative count remains an internal source for the lazy views.
#[derive(Clone, Traversable)]
pub struct AddrEventShares {
    pub all: LazyPercentCumulativeRolling<PartsPerMillion32>,
    #[traversable(flatten)]
    pub by_addr_type: ByAddrType<LazyPercentCumulativeRolling<PartsPerMillion32>>,
}

impl AddrEventShares {
    fn new(
        name: &str,
        version: Version,
        indexes: &indexes::Vecs,
        cached_starts: &Windows<&CachedWindowStartVec>,
        all: LazyPercentCumulativeRolling<PartsPerMillion32>,
        numerators: &ByAddrType<PerBlockCumulativeRolling<StoredU64>>,
        denominators: &ByAddrType<CachedBlockCountReader>,
    ) -> Result<Self> {
        let by_addr_type = zip2_by_addr_type(
            numerators,
            denominators,
            |type_name, numerator, denominator| {
                Ok(LazyPercentCumulativeRolling::from_cached_block_count(
                    &format!("{type_name}_{name}"),
                    version,
                    &numerator.cumulative.height,
                    denominator.clone(),
                    cached_starts,
                    indexes,
                ))
            },
        )?;
        Ok(Self { all, by_addr_type })
    }
}

#[derive(Traversable)]
pub struct AddrEventsVecs<M: StorageMode = Rw> {
    pub output_to_reused_addr_count: WithAddrTypes<PerBlockCumulativeRolling<StoredU64, M>>,
    pub output_to_reused_addr_share: AddrEventShares,
    pub spendable_output_to_reused_addr_share: LazyPercentCumulativeRolling<PartsPerMillion32>,
    pub input_from_reused_addr_count: WithAddrTypes<PerBlockCumulativeRolling<StoredU64, M>>,
    pub input_from_reused_addr_share: AddrEventShares,
    pub active_reused_addr_count: CountPerBlockRollingAverage<M>,
    pub active_reused_addr_share: PerBlockRollingAverage<StoredF32, StoredF32, M>,
}

impl AddrEventsVecs {
    pub(crate) fn forced_import(
        db: &Database,
        name: &str,
        version: Version,
        indexes: &indexes::Vecs,
        cached_starts: &Windows<&CachedWindowStartVec>,
        outputs_by_type: &outputs::ByTypeVecs,
        inputs_by_type: &inputs::ByTypeVecs,
    ) -> Result<Self> {
        let import_count = |name: &str| {
            WithAddrTypes::<PerBlockCumulativeRolling<StoredU64>>::forced_import(
                db,
                name,
                version,
                indexes,
                cached_starts,
            )
        };

        let output_to_reused_addr_count = import_count(&format!("output_to_{name}_addr_count"))?;
        let output_share_name = format!("output_to_{name}_addr_share");
        let output_denominators = outputs_by_type.output_count.cached_addr_type_counts();
        let output_to_reused_addr_share = AddrEventShares::new(
            &output_share_name,
            version,
            indexes,
            cached_starts,
            outputs_by_type.output_count.lazy_share(
                &output_share_name,
                version,
                &output_to_reused_addr_count.all.cumulative.height,
                cached_starts,
                indexes,
            ),
            &output_to_reused_addr_count.by_addr_type,
            &output_denominators,
        )?;
        let spendable_share_name = format!("spendable_output_to_{name}_addr_share");
        let spendable_output_to_reused_addr_share =
            LazyPercentCumulativeRolling::from_cumulative_ratio::<
                StoredU64,
                StoredU64,
                RatioU64<PartsPerMillion32>,
            >(
                &spendable_share_name,
                version,
                &output_to_reused_addr_count.all.cumulative.height,
                outputs_by_type.spendable_output_count.cached_cumulative(),
                cached_starts,
                indexes,
            );
        let input_from_reused_addr_count = import_count(&format!("input_from_{name}_addr_count"))?;
        let input_share_name = format!("input_from_{name}_addr_share");
        let input_denominators = inputs_by_type.input_count.cached_addr_type_counts();
        let input_from_reused_addr_share = AddrEventShares::new(
            &input_share_name,
            version,
            indexes,
            cached_starts,
            inputs_by_type.input_count.lazy_share(
                &input_share_name,
                version,
                &input_from_reused_addr_count.all.cumulative.height,
                cached_starts,
                indexes,
            ),
            &input_from_reused_addr_count.by_addr_type,
            &input_denominators,
        )?;

        let active_reused_addr_count = CountPerBlockRollingAverage::forced_import(
            db,
            &format!("active_{name}_addr_count"),
            version,
            indexes,
            cached_starts,
        )?;
        let active_reused_addr_share = PerBlockRollingAverage::forced_import(
            db,
            &format!("active_{name}_addr_share"),
            version,
            indexes,
            cached_starts,
        )?;

        Ok(Self {
            output_to_reused_addr_count,
            output_to_reused_addr_share,
            spendable_output_to_reused_addr_share,
            input_from_reused_addr_count,
            input_from_reused_addr_share,
            active_reused_addr_count,
            active_reused_addr_share,
        })
    }

    pub(crate) fn min_stateful_len(&self) -> usize {
        self.output_to_reused_addr_count
            .min_stateful_len()
            .min(self.input_from_reused_addr_count.min_stateful_len())
            .min(self.active_reused_addr_count.min_stateful_len())
            .min(self.active_reused_addr_share.block.len())
    }

    pub(crate) fn par_iter_height_mut(
        &mut self,
    ) -> impl ParallelIterator<Item = &mut dyn AnyStoredVec> {
        self.output_to_reused_addr_count
            .par_iter_height_mut()
            .chain(self.input_from_reused_addr_count.par_iter_height_mut())
            .chain([
                self.active_reused_addr_count.stored_mut(),
                &mut self.active_reused_addr_share.block as &mut dyn AnyStoredVec,
            ])
    }

    pub(crate) fn reset_height(&mut self) -> Result<()> {
        self.output_to_reused_addr_count.reset_height()?;
        self.input_from_reused_addr_count.reset_height()?;
        self.active_reused_addr_count.reset()?;
        self.active_reused_addr_share.block.reset()?;
        Ok(())
    }

    #[inline(always)]
    pub(crate) fn push_height(
        &mut self,
        uses: &AddrTypeToAddrEventCount,
        spends: &AddrTypeToAddrEventCount,
        active_addr_count: u32,
        active_reused_addr_count: u32,
    ) {
        self.output_to_reused_addr_count
            .push_height(uses.sum(), uses.values().copied());
        self.input_from_reused_addr_count
            .push_height(spends.sum(), spends.values().copied());
        self.active_reused_addr_count
            .push_block(StoredU32::from(active_reused_addr_count));
        // Stored as a percentage in [0, 100] to match the rest of the
        // codebase (Unit.percentage on the website expects 0..100). The
        // `active_addr_count` denominator lives on `ActivityCountVecs`
        // (`addrs.activity.all.active`), passed in here so we can
        // compute the per-block ratio inline.
        let share = if active_addr_count > 0 {
            100.0 * (active_reused_addr_count as f32 / active_addr_count as f32)
        } else {
            0.0
        };
        self.active_reused_addr_share
            .block
            .push(StoredF32::from(share));
    }

    pub(crate) fn compute_rest(&mut self, starting_lengths: &Lengths, exit: &Exit) -> Result<()> {
        self.active_reused_addr_share
            .compute_rest(starting_lengths.height, exit)?;
        Ok(())
    }
}