feat(broker): add Redis consumer groups for reliable message processing

1 files changed, 35 insertions, 6 deletions

diff --git a/services/portfolio-manager/src/portfolio_manager/main.py b/services/portfolio-manager/src/portfolio_manager/main.py
index a7f1a14..87e4c64 100644
--- a/services/portfolio-manager/src/portfolio_manager/main.py
+++ b/services/portfolio-manager/src/portfolio_manager/main.py
@@ -84,13 +84,43 @@ async def run() -> None:
         snapshot_loop(db, tracker, notifier, config.snapshot_interval_hours, log)
     )
 
-    last_id = "$"
+    GROUP = "portfolio-manager"
+    CONSUMER = "portfolio-1"
     log.info("service_started", stream=ORDERS_STREAM)
 
+    await broker.ensure_group(ORDERS_STREAM, GROUP)
+
+    # Process pending messages first (from previous crash)
+    pending = await broker.read_pending(ORDERS_STREAM, GROUP, CONSUMER)
+    for msg_id, msg in pending:
+        try:
+            event = Event.from_dict(msg)
+            if isinstance(event, OrderEvent):
+                order = event.data
+                tracker.apply_order(order)
+                log.info(
+                    "pending_order_applied",
+                    symbol=order.symbol,
+                    side=str(order.side),
+                    quantity=str(order.quantity),
+                    price=str(order.price),
+                )
+                metrics.events_processed.labels(
+                    service="portfolio-manager", event_type="order"
+                ).inc()
+            await broker.ack(ORDERS_STREAM, GROUP, msg_id)
+        except Exception as exc:
+            log.error("pending_process_failed", error=str(exc), msg_id=msg_id)
+            metrics.errors_total.labels(
+                service="portfolio-manager", error_type="processing"
+            ).inc()
+
     try:
         while True:
-            messages = await broker.read(ORDERS_STREAM, last_id=last_id, block=1000)
-            for msg in messages:
+            messages = await broker.read_group(
+                ORDERS_STREAM, GROUP, CONSUMER, count=10, block=1000
+            )
+            for msg_id, msg in messages:
                 try:
                     event = Event.from_dict(msg)
                     if isinstance(event, OrderEvent):
@@ -108,13 +138,12 @@ async def run() -> None:
                         metrics.events_processed.labels(
                             service="portfolio-manager", event_type="order"
                         ).inc()
+                    await broker.ack(ORDERS_STREAM, GROUP, msg_id)
                 except Exception as exc:
-                    log.exception("message_processing_failed", error=str(exc))
+                    log.exception("message_processing_failed", error=str(exc), msg_id=msg_id)
                     metrics.errors_total.labels(
                         service="portfolio-manager", error_type="processing"
                     ).inc()
-            # Update last_id to the latest processed message id if broker returns ids
-            # Since broker.read returns parsed payloads (not ids), we use "$" to get new msgs
     except Exception as exc:
         log.error("fatal_error", error=str(exc))
         await notifier.send_error(str(exc), "portfolio-manager")