Registered via `brain.defineAggregate(def)`. Stored in a `Map<string, AggregateDefinition>` keyed by aggregate name. Persisted to storage under `__aggregation_definitions__` on flush.
### Group State
Each aggregate maintains a `Map<string, AggregateGroupState>` where keys are serialized group key values (e.g., `category=food|date=2024-01`). Each group holds per-metric `MetricState`:
```typescript
interface MetricState {
sum: number // Running total
count: number // Entity count
min: number // Minimum (Infinity if empty)
max: number // Maximum (-Infinity if empty)
m2?: number // Welford's M2 for stddev/variance
}
```
### Change Detection
On restart, definition hashes (FNV-1a 32-bit) are compared with the persisted hash. If a definition changed (different groupBy, metrics, or source), the aggregate state is reset and must be rebuilt.
## Write-Time Update Flow
When `brain.add(entity)` is called:
```
1. For each registered aggregate:
├─ Source filter check (type, service, where)
│ └─ Skip if entity doesn't match
├─ Aggregate entity check
│ └─ Skip if entity.service === 'brainy:aggregation'
│ or entity.metadata.__aggregate is set
├─ Group key computation
│ └─ Extract groupBy fields from metadata
│ Apply time bucketing for windowed dimensions
└─ Metric update
└─ For each metric in the definition:
├─ count: increment count
├─ sum/avg: add value to sum, increment count
├─ min/max: compare and update
└─ stddev/variance: Welford's online update
```
### Update Handling
On `brain.update(entity)`, the engine reverses the old entity's contribution and applies the new entity's contribution. This correctly handles:
- **Value changes**: old amount=10, new amount=20 — sum adjusts by +10
- **Group key changes**: entity moves from category "food" to "drink" — both groups update
- **Source filter changes**: entity type changes from Event to Document — removed from matching aggregates
-`min`/`max` may become stale (marked in `staleMinMax` for lazy recompute)
- Welford's M2 is updated with the inverse formula
- Empty groups (all metric counts at zero) are removed
## Algorithms
### Welford's Online Algorithm
Standard deviation and variance use Welford's numerically stable online algorithm with M2 tracking. This computes incrementally without storing individual values:
```
On add(x):
count += 1
oldMean = (sum - x) / (count - 1) // mean before this value
sum += x
mean = sum / count // mean after this value
M2 += (x - oldMean) * (x - mean)
On remove(x):
oldMean = sum / count
sum -= x
count -= 1
newMean = sum / count
M2 = max(0, M2 - (x - oldMean) * (x - newMean))
Sample variance = M2 / (count - 1)
Sample stddev = sqrt(variance)
```
M2 is clamped to zero on remove to prevent floating-point drift from producing negative values.
### MIN/MAX Handling
The TypeScript engine uses simple comparison for add operations and marks MIN/MAX as potentially stale on delete (since removing the current min/max value requires a rescan). Stale values are lazily recomputed on the next query.
The Cortex native engine uses a `BTreeMap<OrderedFloat<f64>, u64>` that tracks the exact frequency of every value, providing precise MIN/MAX after any sequence of operations without rescanning.
### Time Window Bucketing
Timestamps (Unix milliseconds) are bucketed using UTC-based formatting:
| Granularity | Bucket Key | Algorithm |
|------------|-----------|-----------|
| `hour` | `2024-01-15T14` | UTC year-month-day-hour |
| `day` | `2024-01-15` | UTC year-month-day |
| `week` | `2024-W03` | ISO 8601 week (Monday start, week 1 contains first Thursday) |
| `month` | `2024-01` | UTC year-month |
| `quarter` | `2024-Q1` | `ceil((month) / 3)` |
| `year` | `2024` | UTC year |
| `{ seconds: N }` | ISO timestamp | `floor(timestamp / interval) * interval` |
Bucket keys can be parsed back into `{ start, end }` timestamp ranges via `parseBucketRange()`.
## Provider Interface
The `AggregationProvider` interface defines the contract between Brainy's `AggregationIndex` and plugin-provided native implementations:
```typescript
interface AggregationProvider {
defineAggregate?(def: AggregateDefinition): void
removeAggregate?(name: string): void
incrementalUpdate(
name: string,
def: AggregateDefinition,
entity: Record<string,unknown>,
op: 'add' | 'update' | 'delete',
prev?: Record<string,unknown>
): AggregateGroupState[]
computeGroupKey(
entity: Record<string,unknown>,
groupBy: GroupByDimension[]
): Record<string,string|number>
rebuildAggregate(
def: AggregateDefinition,
entities: Array<Record<string,unknown>>
): Map<string,AggregateGroupState>
queryAggregate(
state: Map<string,AggregateGroupState>,
params: AggregateQueryParams
): AggregateResult[]
restoreState?(data: string): void
serializeState?(): string
}
```
When a native provider is registered:
1.`AggregationIndex` delegates `incrementalUpdate()` to the provider instead of running TypeScript logic
2. Provider returns updated `AggregateGroupState[]` which are applied back into the state maps
3. Query execution is delegated via `queryAggregate()`
4. State serialization is delegated via `serializeState()`/`restoreState()`
Brainy retains ownership of the state maps and persistence. The provider handles computation.
## Materialization
The `AggregateMaterializer` converts aggregate group states into `NounType.Measurement` entities:
1. When an aggregate group is updated and `materialize` is enabled, `scheduleMaterialize()` is called
2. Materialization is debounced (default: 1000ms) to batch rapid updates during ingestion
3. On trigger, the materializer either creates or updates a `NounType.Measurement` entity
4. Materialized entities include `service: 'brainy:aggregation'` and `metadata.__aggregate` to prevent infinite loops
Materialized entities are automatically visible through:
- OData endpoints
- Google Sheets integration
- Server-Sent Events (SSE)
- Webhook notifications
## Persistence
### Storage Keys
| Key | Content |
|-----|---------|
| `__aggregation_definitions__` | Array of all definitions with FNV-1a hashes |
| `__aggregation_state_{name}__` | Per-aggregate group states (array of `AggregateGroupState`) |
| `__aggregation_native_state__` | Serialized native provider state (JSON string) |
