Make Aggregator process incoming prices into “latest aggregated” snapshots

[grantfox-oss]

📝 Description

Context: apps/aggregator already has: normalization, aggregation strategies, metrics, health checks, and an in-memory debug store exposed at GET /debug/prices. It also has DataReceptionService which connects to the ingestor WebSocket and emits price.received events — but there’s currently no “glue” that listens to those events, normalizes the payload, runs aggregation, and keeps a continuously-updated latest snapshot per symbol.

Goal: Add a small “stream processor” inside the aggregator that:

• listens for incoming price events
• converts/validates them
• normalizes them into the canonical format
• maintains a short rolling buffer of recent normalized prices per symbol
• calls AggregationService.aggregate(...) when enough sources are present
• updates the in-memory latest aggregated snapshot (already handled via DebugService.setLastAggregated inside AggregationService)

This makes apps/aggregator actually produce real-time aggregated results that downstream services (API/Frontend) can consume.

Tech stack / conventions: NestJS (existing), TypeScript, @nestjs/event-emitter for @OnEvent(...) handler, reuse existing NormalizationService, AggregationService, and DebugService. Keep it in-memory (no Redis persistence in this issue).

Complexity: ~6/10 — event-driven wiring + buffering + tests; nothing on-chain.

---

✅ Requirements

• Add an event listener for price.received (emitted by DataReceptionService) using @OnEvent('price.received').
• Convert the event payload into a RawPrice-compatible object (notably: parse ISO timestamp into unix ms) and run it through NormalizationService.
• Maintain an in-memory buffer per symbol (and optionally per source) for recent NormalizedPrice entries:
  • configurable time window (e.g. AGG_TIME_WINDOW_MS, default 30s)
  • configurable minSources (default 3, matching AggregationService)
• When a symbol has enough recent sources, call AggregationService.aggregate(symbol, prices, options) and let it update DebugService’s lastAggregated and lastNormalized.
• Fix/complete any missing Nest module wiring needed for this flow (e.g. ensure EventEmitterModule, HttpModule and providers referenced in AppModule are properly imported/registered).
• Add tests proving: an incoming event updates the rolling buffer; aggregation runs when thresholds are met; debug store updates.

---

🎯 Acceptance Criteria

• Aggregator reacts to live price.received events and continuously updates the latest aggregated snapshot for each symbol.
• GET /debug/prices shows non-empty aggregated output after receiving valid events (in tests and/or local run).
• Buffering behavior works: old entries are dropped outside the time window; aggregation requires minSources.
• Tests cover the event handler and aggregation trigger logic (mocks acceptable for event emitter and services).
• apps/aggregator builds and tests pass.

---

📁 Expected files to change/structure

• apps/aggregator/src/services/price-stream-processor.service.ts (new) — @OnEvent handler + buffering + trigger aggregation.
• apps/aggregator/src/app.module.ts — ensure all imports/providers are correctly wired for event emitter + reception + processor.
• apps/aggregator/src/services/*.spec.ts (new or update) — unit tests for processor behavior.
• apps/aggregator/.env.example (optional) — document AGG_TIME_WINDOW_MS, AGG_MIN_SOURCES, and ingestor connection vars if needed.

---

Note: don't worry about CI workflow for now, it's an issue on our end we will fix after we merge all issues!

Thank you for taking this issue! You are helping us make RWAs consumer friendly on Stellar.

[Joaco2603]

Hello! I’m a backend engineer specializing in NestJS and TypeScript, with strong experience in event-driven architectures. I’d like to take ownership of this challenge, as it aligns directly with my technical strengths.

I will implement an @OnEvent listener for price.received, handle timestamp normalization from ISO to Unix, and design an efficient in-memory sliding window that triggers AggregationService only when the minSources and time-window criteria are satisfied.

My approach emphasizes clean, well-tested code and predictable performance, helping strengthen the reliability of RWA oracles on Stellar.

[gelluisaac]

I’m interested in tackling this! I have a strong background in smart contract development and full-stack systems, so I'm well-equipped to handle the requirements here. Could you please assign this to me?

[gelluisaac]

I’m interested in tackling this! I have a strong background in smart contract development and full-stack systems, so I'm well-equipped to handle the requirements here. Could you please assign this to me?

[Naomi-Gift]

Hi,

I’d like to be assigned to this "glue" service implementation. I have extensive experience with NestJS event-driven microservices and building high-throughput in-memory stream processors for fintech data.

[Awointa]

Hi maintainer 👋 I’d love to work on this issue.

I’ll implement the stream processor to handle price.received events, normalize and validate incoming data, maintain the rolling buffer per symbol, and trigger aggregation when thresholds are met. I’ll also ensure proper module wiring and add tests to verify buffering, aggregation, and debug snapshot updates.

Happy to share progress updates as I work — please assign it to me if available.

[akintewe]

I'm a full-stack developer with 4+ years of experience building complex forms and e-commerce systems. I've previously worked on this project in the past as well as also developed inventory management systems for agricultural businesses and am well-versed in handling image processing and state management at scale.
I would love to work on this issue, my ETA is 1 hour.

[walterthesmart]

Hi, I would like to work on this issues.

[AFuentes1]

Hi team 👋 I’m Anthony and I’d like to take on Issue #33 because I have experience running data systems "live" from receipt until they’re ready for consumption, with a focus on clarity, stability, and good testing. My plan is to complete the remaining flow so that the aggregator, by receiving prices, consistently processes them, maintains a short history per symbol, and continuously updates the most recent aggregate result when conditions are met (time window and minimum of sources), all in memory and without overcomplicating the solution. I also make sure to keep it well connected within the project and covered with tests so that it is reliable and easy to evolve. Can I be assigned to it?

[salazarsebas]

Hello Neko team,
I have experience in oss, especially with Stellar ecosystem, can I work on this issue?