matrix-receiver: Map Hive message IDs to Matrix event IDs

services/matrix-connector/docker-compose.yml

@@ -4,7 +4,8 @@ services:
     init: true
     restart: unless-stopped
     networks:
-      - matrix-sender
+      - matrix-connector
+      - matrix-valkey
       - message-bus
     volumes:
       - matrix_sender:/var/lib/matrix
@@ -19,7 +20,8 @@ services:
     init: true
     restart: unless-stopped
     networks:
-      - matrix-receiver
+      - matrix-connector
+      - matrix-valkey
       - message-bus
     volumes:
       - matrix_receiver:/var/lib/matrix
@@ -29,11 +31,26 @@ services:
     secrets:
       - rabbitmq.env
 
+  matrix-valkey:
+    image: valkey/valkey:8
+    restart: unless-stopped
+    networks:
+      - vane-valkey
+    volumes:
+      - matrix_valkey:/data
+    command:
+      - valkey-server
+      - --save
+      - 60 1
+    hostname: vane-valkey
+
 networks:
-  matrix-sender:
-  matrix-receiver:
+  matrix-connector:
+  matrix-valkey:
+    internal: true
 
 volumes:
   matrix_sender:
   matrix_receiver:
   matrix_received_media:
+  matrix_valkey:

services/matrix-connector/hive/matrix_connector/receiver.py

@@ -7,7 +7,10 @@
 from functools import cached_property
 from typing import Literal, Sequence
 
+from valkey import Valkey
+
 from hive.chat import ChatMessage, tell_user
+from hive.common.units import HOUR
 from hive.service import HiveService
 
 logger = logging.getLogger(__name__)
@@ -22,6 +25,12 @@ class Receiver(HiveService):
         "--download-media-name", "eventid",
         "--output", "json-max",
     )
+    valkey_url: str = "valkey://matrix-valkey"
+    id_correlation_lifetime: float = 1 * HOUR
+
+    @cached_property
+    def _valkey(self) -> Valkey:
+        return Valkey.from_url(self.valkey_url)
 
     def __post_init__(self):
         super().__post_init__()
@@ -106,5 +115,18 @@ def on_matrix_event(self, event: dict):
             raise ValueError(event) from e
         tell_user(message, channel=self._channel)
 
+        message_id = str(message.uuid)
+        event_id = message.matrix.event_id
+        self._valkey.set(
+            f"message:{message_id}:event_id",
+            event_id,
+            ex=self.id_correlation_lifetime,
+        )
+        self._valkey.set(
+            f"event:{event_id}:message_id",
+            message_id,
+            ex=self.id_correlation_lifetime,
+        )
+
 
 main = Receiver.main

services/matrix-connector/pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "hive-messaging",
     "hive-service",
     "matrix-commander",
+    "valkey",
 ]
 
 [project.scripts]

services/matrix-connector/tests/conftest.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-class MockChannel:
+class CallLogger:
     def __init__(self):
         self.call_log = []
 
@@ -20,4 +20,9 @@ def __call__(self, *args, **kwargs):
 
 @pytest.fixture
 def mock_channel():
-    return MockChannel()
+    return CallLogger()
+
+
+@pytest.fixture
+def mock_valkey():
+    return CallLogger()

services/matrix-connector/tests/test_receiver.py

@@ -8,16 +8,17 @@
 
 
 @pytest.fixture
-def mock_receiver(mock_channel):
+def mock_receiver(mock_channel, mock_valkey):
     receiver = Receiver()
     receiver._channel = mock_channel
+    receiver._valkey = mock_valkey
     try:
         yield receiver
     finally:
         sys.modules.pop("matrix_commander")
 
 
-def test_basic(mock_channel, mock_receiver):
+def test_basic(mock_receiver, mock_channel, mock_valkey):
     event = json.loads(read_resource("resources/text.json"))
     mock_receiver.on_matrix_event(event)
     assert len(mock_channel.call_log) == 2
@@ -43,9 +44,23 @@ def test_basic(mock_channel, mock_receiver):
             "routing_key": "chat.messages",
          }),
     ]
+    assert mock_valkey.call_log == [
+        ("set", (
+            f"message:{uuid}:event_id",
+            "$NWixTiloQs5UwmlcdJfSfFVtw5SX3awbu3NXvDdOZwo",
+        ), {
+            "ex": 3600,
+        }),
+        ("set", (
+            "event:$NWixTiloQs5UwmlcdJfSfFVtw5SX3awbu3NXvDdOZwo:message_id",
+            uuid,
+        ), {
+            "ex": 3600,
+        }),
+    ]
 
 
-def test_html(mock_channel, mock_receiver):
+def test_html(mock_receiver, mock_channel, mock_valkey):
     event = json.loads(read_resource("resources/html.json"))
     mock_receiver.on_matrix_event(event)
     assert len(mock_channel.call_log) == 2
@@ -72,9 +87,23 @@ def test_html(mock_channel, mock_receiver):
             "routing_key": "chat.messages",
          }),
     ]
+    assert mock_valkey.call_log == [
+        ("set", (
+            f"message:{uuid}:event_id",
+            "$r9Ul_OMug-vwLOY0yQY2kLtQtIFlxNff6nROekWc4Co",
+        ), {
+            "ex": 3600,
+        }),
+        ("set", (
+            "event:$r9Ul_OMug-vwLOY0yQY2kLtQtIFlxNff6nROekWc4Co:message_id",
+            uuid,
+        ), {
+            "ex": 3600,
+        }),
+    ]
 
 
-def test_image(mock_channel, mock_receiver):
+def test_image(mock_receiver, mock_channel, mock_valkey):
     event = json.loads(read_resource("resources/image.json"))
     mock_receiver.on_matrix_event(event)
     assert len(mock_channel.call_log) == 2
@@ -100,9 +129,23 @@ def test_image(mock_channel, mock_receiver):
             "routing_key": "chat.messages",
          }),
     ]
+    assert mock_valkey.call_log == [
+        ("set", (
+            f"message:{uuid}:event_id",
+            "$pOp5KHHL3ECE3ZWtRw_PmnrH-mRqDFlDHcKgzMBSEVY",
+        ), {
+            "ex": 3600,
+        }),
+        ("set", (
+            "event:$pOp5KHHL3ECE3ZWtRw_PmnrH-mRqDFlDHcKgzMBSEVY:message_id",
+            uuid,
+        ), {
+            "ex": 3600,
+        }),
+    ]
 
 
-def test_redaction(mock_channel, mock_receiver):
+def test_redaction(mock_receiver, mock_channel, mock_valkey):
     event = json.loads(read_resource("resources/redaction.json"))
     with pytest.raises(ValueError) as excinfo:
         mock_receiver.on_matrix_event(event)
@@ -117,4 +160,5 @@ def test_redaction(mock_channel, mock_receiver):
             "mandatory": True,
         }),
     ]
+    assert mock_valkey.call_log == []
     assert str(excinfo.value) == repr(event)