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

2 files changed, 185 insertions, 1 deletions

diff --git a/shared/src/shared/broker.py b/shared/src/shared/broker.py
index 9c6c4c6..c060c24 100644
--- a/shared/src/shared/broker.py
+++ b/shared/src/shared/broker.py
@@ -17,6 +17,70 @@ class RedisBroker:
         payload = json.dumps(data)
         await self._redis.xadd(stream, {"payload": payload})
 
+    async def ensure_group(self, stream: str, group: str) -> None:
+        """Create a consumer group if it doesn't exist."""
+        try:
+            await self._redis.xgroup_create(stream, group, id="0", mkstream=True)
+        except redis.ResponseError as e:
+            if "BUSYGROUP" not in str(e):
+                raise
+
+    async def read_group(
+        self,
+        stream: str,
+        group: str,
+        consumer: str,
+        count: int = 10,
+        block: int = 0,
+    ) -> list[tuple[str, dict[str, Any]]]:
+        """Read messages from a consumer group. Returns list of (message_id, data)."""
+        results = await self._redis.xreadgroup(
+            group, consumer, {stream: ">"}, count=count, block=block
+        )
+        messages = []
+        if results:
+            for _stream, entries in results:
+                for msg_id, fields in entries:
+                    payload = fields.get(b"payload") or fields.get("payload")
+                    if payload:
+                        if isinstance(payload, bytes):
+                            payload = payload.decode()
+                        if isinstance(msg_id, bytes):
+                            msg_id = msg_id.decode()
+                        messages.append((msg_id, json.loads(payload)))
+        return messages
+
+    async def ack(self, stream: str, group: str, *msg_ids: str) -> None:
+        """Acknowledge messages in a consumer group."""
+        if msg_ids:
+            await self._redis.xack(stream, group, *msg_ids)
+
+    async def read_pending(
+        self,
+        stream: str,
+        group: str,
+        consumer: str,
+        count: int = 10,
+    ) -> list[tuple[str, dict[str, Any]]]:
+        """Read pending (unacknowledged) messages for this consumer."""
+        results = await self._redis.xreadgroup(
+            group, consumer, {stream: "0"}, count=count
+        )
+        messages = []
+        if results:
+            for _stream, entries in results:
+                for msg_id, fields in entries:
+                    if not fields:  # Empty fields means already acknowledged
+                        continue
+                    payload = fields.get(b"payload") or fields.get("payload")
+                    if payload:
+                        if isinstance(payload, bytes):
+                            payload = payload.decode()
+                        if isinstance(msg_id, bytes):
+                            msg_id = msg_id.decode()
+                        messages.append((msg_id, json.loads(payload)))
+        return messages
+
     async def read(
         self,
         stream: str,
@@ -24,7 +88,7 @@ class RedisBroker:
         count: int = 10,
         block: int = 0,
     ) -> list[dict[str, Any]]:
-        """Read messages from a Redis stream."""
+        """Read messages (original method, kept for backward compatibility)."""
         results = await self._redis.xread({stream: last_id}, count=count, block=block)
         messages = []
         if results:
diff --git a/shared/tests/test_broker.py b/shared/tests/test_broker.py
index ea8b148..c33f6ec 100644
--- a/shared/tests/test_broker.py
+++ b/shared/tests/test_broker.py
@@ -2,6 +2,7 @@
 
 import pytest
 import json
+import redis
 from unittest.mock import AsyncMock, patch
 
 
@@ -68,3 +69,122 @@ async def test_broker_close():
         await broker.close()
 
         mock_redis.aclose.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_broker_ensure_group():
+    """Test that ensure_group creates a consumer group."""
+    from shared.broker import RedisBroker
+
+    mock_redis = AsyncMock()
+    broker = RedisBroker.__new__(RedisBroker)
+    broker._redis = mock_redis
+
+    await broker.ensure_group("test-stream", "test-group")
+    mock_redis.xgroup_create.assert_called_once_with(
+        "test-stream", "test-group", id="0", mkstream=True
+    )
+
+
+@pytest.mark.asyncio
+async def test_broker_ensure_group_already_exists():
+    """Test that ensure_group ignores BUSYGROUP error."""
+    from shared.broker import RedisBroker
+
+    mock_redis = AsyncMock()
+    mock_redis.xgroup_create = AsyncMock(
+        side_effect=redis.ResponseError("BUSYGROUP Consumer Group name already exists")
+    )
+    broker = RedisBroker.__new__(RedisBroker)
+    broker._redis = mock_redis
+
+    # Should not raise
+    await broker.ensure_group("test-stream", "test-group")
+
+
+@pytest.mark.asyncio
+async def test_broker_read_group():
+    """Test that read_group parses xreadgroup response correctly."""
+    from shared.broker import RedisBroker
+
+    mock_redis = AsyncMock()
+    mock_redis.xreadgroup = AsyncMock(
+        return_value=[
+            (b"stream", [(b"1-0", {b"payload": b'{"type": "test"}'})])
+        ]
+    )
+    broker = RedisBroker.__new__(RedisBroker)
+    broker._redis = mock_redis
+
+    messages = await broker.read_group("stream", "group", "consumer")
+    assert len(messages) == 1
+    assert messages[0][0] == "1-0"
+    assert messages[0][1] == {"type": "test"}
+
+
+@pytest.mark.asyncio
+async def test_broker_ack():
+    """Test that ack calls xack on the redis connection."""
+    from shared.broker import RedisBroker
+
+    mock_redis = AsyncMock()
+    broker = RedisBroker.__new__(RedisBroker)
+    broker._redis = mock_redis
+
+    await broker.ack("stream", "group", "1-0", "2-0")
+    mock_redis.xack.assert_called_once_with("stream", "group", "1-0", "2-0")
+
+
+@pytest.mark.asyncio
+async def test_broker_read_pending():
+    """Test that read_pending reads unacknowledged messages."""
+    from shared.broker import RedisBroker
+
+    mock_redis = AsyncMock()
+    mock_redis.xreadgroup = AsyncMock(
+        return_value=[
+            (b"stream", [(b"1-0", {b"payload": b'{"type": "pending"}'})])
+        ]
+    )
+    broker = RedisBroker.__new__(RedisBroker)
+    broker._redis = mock_redis
+
+    messages = await broker.read_pending("stream", "group", "consumer")
+    assert len(messages) == 1
+    assert messages[0][0] == "1-0"
+    assert messages[0][1] == {"type": "pending"}
+    # Verify it uses "0" (not ">") to read pending
+    mock_redis.xreadgroup.assert_called_once_with(
+        "group", "consumer", {"stream": "0"}, count=10
+    )
+
+
+@pytest.mark.asyncio
+async def test_broker_read_pending_skips_empty_fields():
+    """Test that read_pending skips already-acknowledged entries with empty fields."""
+    from shared.broker import RedisBroker
+
+    mock_redis = AsyncMock()
+    mock_redis.xreadgroup = AsyncMock(
+        return_value=[
+            (b"stream", [(b"1-0", {})])
+        ]
+    )
+    broker = RedisBroker.__new__(RedisBroker)
+    broker._redis = mock_redis
+
+    messages = await broker.read_pending("stream", "group", "consumer")
+    assert len(messages) == 0
+
+
+@pytest.mark.asyncio
+async def test_broker_ack_no_ids():
+    """Test that ack does nothing when no message IDs are provided."""
+    from shared.broker import RedisBroker
+
+    mock_redis = AsyncMock()
+    broker = RedisBroker.__new__(RedisBroker)
+    broker._redis = mock_redis
+
+    await broker.ack("stream", "group")
+    mock_redis.xack.assert_not_called()