---
# System prepended metadata

title: Requirements

---

# Requirements

* Analytics should be incremental
* Analytics should work via the in-memory representation
* Analytics have to consider protocol parameters

# Scalability

* Memory
    * How much memory do we need for incremental analytics?
        * What are the expensive analytics
            * Addresses with a balance (`HashSet`s of addresses)
            * Daily Active Addresses (`HashSet`s of addresses)
            * NFT / Alias Activity
* Time
    * Should be done within a milestone


# Ledger Update based

TODO:
 * Handling protocol parameters change
 * Reading unspent outputs

```rust=
pub trait LedgerUpdateAnalytics {
    type Measurement;
    fn begin(&mut self, _marker: LedgerUpdateMarker);
    fn handle_created(&mut self, output: &LedgerOutput);
    fn handle_consumed(&mut self, spent: &LedgerSpent);
    fn flush(&mut self, _marker: LedgerUpdateMarker) -> Option<Self::Measurement>;
}
```

# Cone Stream based

Should be better for temporal coherence of transactions because the blocks within a cone are published in white-flag order.

```rust=
pub trait TransactionAnalytics {
    type Measurement;
    fn begin(&mut self);
    fn handle_payload(&mut self, payload: &TransactionPayload);
    fn finish(&mut self) -> Option<Self::Measurement>;
}
```

# Storage

```rust=
struct AnalyticsStorage {
    addresses: HashMap<Address, AddressStorage>,
}
s
struct AddressStorage {
    balance: u64,
    sent_outputs: u32?,
    received_outputs: u32?,
}

let shared_state = Arc::new(AnalyticsStorage::default());
let address_balance_stat = AddressBalanceStatistic::new(shared_state.clone());
let active_addresses_stat = DailyActiveAddressesStatistic::new(shared_state);

// Per miletone...

let cone_stream = ...

shared_state.begin();
for payload in cone_stream {
    shared_state.handle_payload(&payload);
}
shared_state.finish();

spawn_task(async {
    address_balance_stat.begin(); 
    for payload in cone_stream {
        address_balance_stat.handle_payload(&payload);
    }
    address_balance_stat.finish();
});

spawn_task(async {
    active_addresses_stat.begin(); 
    for payload in cone_stream {
        active_addresses_stat.handle_payload(&payload);
    }
    active_addresses_stat.finish();
});

await tasks
```

# Stream Mappers

```rust=
trait Map {
    type Input;
    type Output;
    
    fn process(input: Self::Input) -> (Self::Input, Self::Output);
    
    fn post_process(input: Self::Input, output: Self::Output);
    
    fn connect<M: Map>(&self, other: M)
        where M::Input = Self::Output;
}
```