brainy/docs/architecture/aggregation.md

Aggregation Architecture

Write-time incremental aggregation with O(1) reads

Design Principles

1. Write-time computation — aggregates update on every add(), update(), and delete(), not as batch jobs
2. Incremental state — running totals maintained per group, never rescanning the dataset
3. Provider interface — TypeScript engine is the default; plugins can replace it with native implementations
4. Zero-allocation reads — query results are computed from pre-aggregated state

Component Overview

┌──────────────────────────────────────────────────────────┐
│                         Brainy                           │
│                                                          │
│  add() / update() / delete()                             │
│          │                                               │
│          ▼                                               │
│  ┌──────────────────────┐    ┌────────────────────────┐  │
│  │  AggregationIndex    │    │  AggregateMaterializer  │  │
│  │                      │───▶│  (debounced writes)     │  │
│  │  ├─ definitions Map  │    └────────────────────────┘  │
│  │  ├─ states Map       │                                │
│  │  └─ staleMinMax Set  │    ┌────────────────────────┐  │
│  │                      │    │  timeWindows.ts         │  │
│  │  Source filter ──────│───▶│  bucketTimestamp()      │  │
│  │  Group key ──────────│───▶│  parseBucketRange()     │  │
│  └──────────┬───────────┘    └────────────────────────┘  │
│             │                                            │
│             │ provider interface                          │
│             ▼                                            │
│  ┌──────────────────────┐                                │
│  │  AggregationProvider │  (optional, registered by      │
│  │  ├─ incrementalUpdate│   plugin like @soulcraft/cor)   │
│  │  ├─ rebuildAggregate │                                │
│  │  ├─ queryAggregate   │                                │
│  │  └─ serialize/restore│                                │
│  └──────────────────────┘                                │
└──────────────────────────────────────────────────────────┘

State Management

Definitions

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:

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

Delete Handling

On brain.remove(id), the engine reverses the entity's contribution:

• count and sum are decremented
• 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 Cor 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:

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)

Lifecycle

1. init() — Load definitions, compare hashes, load matching state, restore native provider state
2. Write operations — Mark modified aggregates as dirty
3. flush() — Persist all dirty aggregate states and native provider state
4. close() — Flush and release resources

Source Files

File	Purpose
src/aggregation/AggregationIndex.ts	Core engine: definitions, state, write hooks, query
src/aggregation/materializer.ts	Debounced materialization of results as entities
src/aggregation/timeWindows.ts	Time bucketing and bucket range parsing
src/aggregation/index.ts	Module exports
src/types/brainy.types.ts	Type definitions for all aggregation interfaces