Skip to content

⭐ Usage Aggregation

Jump to bottom
Terrence Daniels edited this page Sep 17, 2026 · 6 revisions

A Kafka consumer/producer receiving meter readings, deduplicating them by ReadingId, and rolling them up into hourly usage totals per meter — the middle stage of Phase 1's core loop (Meter Simulator → Usage Aggregation → Billing), now event-driven end to end.

Composition

  • MeterReadingRequest — the internal shape ReadingProcessor works with, matching the design doc's meter.readings.raw schema
  • MeterReadingRaw — the Avro wire-format record actually consumed off Kafka; MeterReadingConsumer converts it to a MeterReadingRequest at the boundary, keeping transport concerns out of the domain logic
  • ProcessedReading — one row per reading ever accepted, keyed by ReadingId; its existence is the idempotency check; carries both MeterId and AccountId since one physical reading has both identities
  • HourlyUsage — the rollup record, one row per (AccountId, PeriodStart) hour, enforced unique at the database level — a real account-level rollup now, not per-meter
  • ReadingProcessor — composes the above: checks for a duplicate first, upserts the hour bucket, then publishes the updated total via IUsageAggregatedPublisher
  • MeterReadingConsumer — a BackgroundService subscribed to meter.readings.raw, feeding each message to ReadingProcessor in its own DI scope
  • IUsageAggregatedPublisher / UsageAggregatedPublisher — a small project-owned abstraction wrapping the real Kafka producer, so tests don't need to fake Confluent.Kafka's much larger IProducer<TKey,TValue> interface
  • UsageAnomalyDetector / UsageAnomalyProcessor — a PeriodicTimer-driven BackgroundService plus its testable processor, checking for any meter whose most recent reading is older than a configurable threshold and publishing usage.anomaly.detected for Grid Operations to correlate into outages
  • IUsageAnomalyPublisher / UsageAnomalyPublisher — the same small-abstraction pattern as IUsageAggregatedPublisher, applied to the new anomaly topic
  • Espi/EspiConstants.cs / Espi/EspiFeedBuilder.cs — real NAESB ESPI enum codes and pure, static Atom/XML-building functions backing the Green Button data export (see below)

A Phase 1 deviation, resolved in Phase 3

HourlyUsage used to be keyed by MeterId, not accountId as the design doc's usage.aggregated schema always specified — there was no Account/Customer Service yet to map a meter to an account. Phase 3 built that service and re-keyed HourlyUsage to AccountId for real: two meters on the same account now roll into one hourly total, matching the design doc exactly. Old MeterId-keyed rows had no valid AccountId to migrate to (they predate accounts existing at all), so the migration deletes them rather than carrying forward data that would be wrong under the new column's meaning — an expected one-time clean transition, not silent data loss.

The original REST API is gone — usage.aggregated republishes continuously

POST /readings and GET /usage/{meterId} have both been removed entirely. Every time ReadingProcessor updates an HourlyUsage bucket, it immediately publishes the new running total as a usage.aggregated event — a continuously-updated aggregate stream, similar to a Kafka Streams KTable changelog, rather than a one-time snapshot returned on request. HTTP came back later for a different reason — see the ESPI export below — so this service isn't Kafka-only anymore, just no longer REST-driven for its original read/write purpose.

Green Button / NAESB ESPI data export

Five real HTTP GET endpoints under /espi/1_1/resource/UsagePoint/... implement the real NAESB ESPI data model (UsagePoint, MeterReading, ReadingType, IntervalBlock/IntervalReading) as Atom XML feeds — the actual US DOE-backed Green Button standard for customer energy-usage export, correct against the real XSD's element order and enum codes, not a simplified stand-in. The OAuth2 third-party-authorization half of the real spec is deliberately not implemented (this repo has no auth system anywhere), documented as an explicit scope decision with its real consequence — see docs/architecture.md. A real, live-verification-caught bug along the way: an orphaned, long-forgotten MeterSimulator process had left 19,888 rows with an empty AccountId in the live database, surfaced by the new UsagePoint collection endpoint — traced, fixed, and documented in issue #26.

Detecting meters that stop transmitting, for Grid Operations

Usage Aggregation owns the low-level detection signal for the new Grid Operations feature: a meter whose most recent reading exceeds ExpectedReadingIntervalSeconds × MissedIntervalMultiplier gets a usage.anomaly.detected event, checked every SweepIntervalSeconds by UsageAnomalyDetector. A deliberate scope decision: the signal is "stopped transmitting," not "usage near zero" — the real UCI dataset this repo replays legitimately produces near-zero readings overnight for every meter, which would false-positive constantly on a value threshold. MaxAnomalyLookbackSeconds bounds how old a reading can be and still count as a new anomaly, fixing a real cold-start bug (#30) where leftover readings from a previous dev session got flagged immediately on AppHost startup. Full reasoning in docs/architecture.md.

EF Core migrations against a service AppHost doesn't run standalone

Generating a migration requires EF's design-time tooling to construct the DbContext, but there's no running AppHost at that point to supply the real Aspire-resolved connection string. UsageAggregationDbContextFactory supplies a placeholder one instead — never used to actually connect, only so the tooling knows to generate Postgres-flavored SQL. The real app always uses Aspire's injected connection string at runtime.

Full reasoning in todo.md.

Verified live, not just built

Confirmed against a real running Postgres: idempotency (a resent ReadingId is a no-op, not reprocessed) and hourly rollup (two readings for the same meter/hour sum into one row with the correct total). Receives 100% of its traffic from Meter Simulator through real Kafka — 1,231 real INSERT INTO "ProcessedReadings" statements confirmed in one live run, 0 errors. Re-verified after the AccountId re-keying as part of Phase 3's full pipeline check: real AccountId values flowing correctly from meter.readings.raw through to usage.aggregated. The ESPI endpoints were verified against a real account with real hourly rows — every IntervalReading value cross-checked exactly against TotalKwh × 1000 in Postgres, 404 confirmed for an unknown account, and published-min/published-max filtering confirmed exact.

Tested

17 xUnit tests: 6 cover ReadingProcessor against EF Core's InMemory provider (idempotency and rollup logic unchanged by the move to Kafka, or by the AccountId re-keying) and a hand-rolled fake IUsageAggregatedPublisher — no mocking library needed, including a dedicated test proving two different meters on the same account roll into one account-level hourly total. 7 more cover EspiFeedBuilder's pure XML-building functions directly, including exact-value assertions for every ReadingType field, a hand-computed-and-independently-verified Unix-timestamp case, and a floating-point-noise rounding case. The remaining 4 cover UsageAnomalyProcessor's stale-meter threshold logic — a meter past the threshold gets flagged, one within it doesn't, the most recent reading (not the oldest) is what's checked, and only the actually-stale meter is flagged among several.

Clone this wiki locally