chat-router: New service

.github/workflows/hive-chat-router.yml

@@ -0,0 +1,56 @@
+name: hive-chat-router
+
+on:
+  push:
+    tags:
+      - "hive-chat-router-[0-9]+.[0-9]+.[0-9]+"
+
+jobs:
+  build:
+    name: Build and push hive-chat-router
+    runs-on: ubuntu-latest
+
+    defaults:
+      run:
+        working-directory: services/chat-router
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Login to Docker Hub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ vars.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+      - name: Write __version__.py
+        if: startsWith(github.ref, 'refs/tags/hive-chat-router-')
+        run: ../../ci/write-version-py hive/chat_router
+
+      - name: Collect Docker metadata
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: gbenson/hive-chat-router
+          tags: |
+            type=match,pattern=hive-chat-router-(\d+\.\d+\.\d+),group=1
+
+      - name: Build and push Docker images
+        uses: docker/build-push-action@v5
+        with:
+          context: services/chat-router
+          platforms: linux/amd64,linux/arm64
+          push: ${{ startsWith(github.ref, 'refs/tags/hive-chat-router-') }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          build-args: |
+            GITHUB_SERVER_URL=${{ github.server_url }}
+            GITHUB_REPOSITORY=${{ github.repository }}
+            GITHUB_SHA=${{ github.sha }}

docker-compose.yml

@@ -3,6 +3,7 @@ include:
     env_file:
       - $HOME/.config/hive/mapped-ports.env
 
+  - path: services/chat-router/docker-compose.yml
   - path: services/email-receiver/docker-compose.yml
   - path: services/matrix-connector/docker-compose.yml
   - path: services/matrix-router/docker-compose.yml

services/chat-router/.dockerignore

@@ -0,0 +1,7 @@
+**/.[^.]*
+**/*.egg-info
+**/__pycache__
+**/*.pyc
+**/*~
+**/venv
+**/.venv

services/chat-router/Dockerfile

@@ -0,0 +1,77 @@
+ARG BASE_IMAGE=debian:12-slim
+
+FROM ${BASE_IMAGE} AS base
+ENV DEBIAN_FRONTEND=noninteractive
+
+RUN --mount=type=tmpfs,target=/var/cache \
+    --mount=type=tmpfs,target=/var/lib/apt/lists \
+    --mount=type=tmpfs,target=/var/log \
+  set -eux \
+  \
+  && apt-get -y update \
+  && apt-get -y upgrade --no-install-recommends \
+  && apt-get -y install --no-install-recommends \
+       python3 \
+       python3-venv \
+  && rm -f /var/lib/dpkg/*-old
+
+RUN --mount=type=tmpfs,target=/root/.cache \
+  set -eux \
+  \
+  && python3 -m venv /venv \
+  && . /venv/bin/activate \
+  && pip install --upgrade pip \
+  && pip install wheel
+
+FROM base AS wheel
+RUN mkdir -p /src
+COPY hive /src/hive
+COPY pyproject.toml README.md /src
+RUN /venv/bin/pip wheel --no-deps /src
+
+FROM base AS install
+RUN --mount=type=bind,from=wheel,target=/wheel \
+    --mount=type=tmpfs,target=/root/.cache \
+  set -eux \
+  \
+  && python3 -m venv /venv \
+  && . /venv/bin/activate \
+  && pip install /wheel/*.whl \
+  && pip check
+
+RUN ln -s ../../venv/bin/hive-chat-router /usr/bin
+
+RUN set -eux \
+  \
+  && addgroup --system --gid 3951 vane \
+  && adduser --system --uid 3951 --gid 3951 \
+       --home /var/lib/vane --disabled-password vane \
+  && rm -f /run/adduser \
+  && install -d -oroot -gvane -m710 /run/secrets
+
+FROM install AS test
+RUN set -eux \
+  \
+  && install -d -ovane -gvane /src \
+  && /venv/bin/pip install \
+       pep440-version-utils \
+       pytest-cov
+COPY tests /src/tests
+
+USER vane:vane
+WORKDIR /src
+
+RUN set -eux \
+  \
+  && . /venv/bin/activate \
+  && pytest --cov hive.chat_router \
+  && coverage report > coverage.report
+
+FROM install AS dist
+RUN --mount=type=bind,from=test,target=/test \
+  grep -q '[^ ][^0]%$' /test/src/coverage.report
+
+USER vane:vane
+WORKDIR /var/lib/vane
+
+ENTRYPOINT ["hive-chat-router"]

services/chat-router/README.md

@@ -0,0 +1,22 @@
+[![version badge]](https://hub.docker.com/r/gbenson/hive-chat-router)
+
+[version badge]: https://img.shields.io/docker/v/gbenson/hive-chat-router?color=limegreen
+
+# hive-chat-router
+
+Chat message router for Hive
+
+## Installation
+
+### For development
+
+```sh
+git clone https://github.com/gbenson/hive.git
+cd hive/services/chat-router
+python3 -m venv .venv
+. .venv/bin/activate
+pip install --upgrade pip
+pip install -r requirements.txt
+pip install -e .
+flake8 && pytest
+```

services/chat-router/docker-compose.yml

@@ -0,0 +1,13 @@
+services:
+  chat-router:
+    image: gbenson/hive-chat-router
+    init: true
+    restart: unless-stopped
+    networks:
+      - message-bus
+    volumes:
+      - ./hive/chat_router:/venv/lib/python3.11/site-packages/hive/chat_router:ro
+    command:
+      - hive-chat-router
+    secrets:
+      - rabbitmq.env

services/chat-router/hive/chat_router/__init__.py

@@ -0,0 +1,3 @@
+from .service import Service
+
+main = Service.main

services/chat-router/hive/chat_router/__version__.py

@@ -0,0 +1 @@
+__version__ = "0.0.0"

services/chat-router/hive/chat_router/handler.py

@@ -0,0 +1,18 @@
+from abc import ABC, abstractmethod
+
+from hive.chat import ChatMessage
+from hive.messaging import Channel
+
+
+class Handler(ABC):
+    @property
+    def priority(self) -> int:
+        return 50
+
+    @abstractmethod
+    def handle(self, channel: Channel, message: ChatMessage) -> bool:
+        """Handle `message`.
+
+        :return: True if `message` was handled, otherwise False.
+        """
+        raise NotImplementedError  # pragma: no cover

services/chat-router/hive/chat_router/loader.py

@@ -0,0 +1,109 @@
+import logging
+import os
+
+from collections.abc import Iterable
+from importlib import import_module, reload
+from pkgutil import iter_modules
+
+from . import handlers
+from .handler import Handler
+
+logger = logging.getLogger(__name__)
+
+
+class HandlerLoader:
+    def __init__(self, toplevel_module=handlers, handler_type=Handler):
+        self.toplevel_module = toplevel_module
+        self.handler_type = handler_type
+        self._handlers = {}
+
+    @property
+    def package_name(self) -> str:
+        return self.toplevel_module.__package__
+
+    @property
+    def search_path(self) -> list[str]:
+        return self.toplevel_module.__path__
+
+    @property
+    def module_names(self) -> Iterable[str]:
+        package_name = self.package_name
+        for module_info in iter_modules(self.search_path):
+            if not module_info.ispkg:
+                yield f"{package_name}.{module_info.name}"
+
+    @property
+    def modules(self):
+        for name in self.module_names:
+            try:
+                module = import_module(name)
+                m_mtime = getattr(module, "__mtime__", None)
+                f_mtime = os.path.getmtime(module.__file__)
+                if f_mtime != m_mtime:
+                    stale_handlers = [
+                        key
+                        for key, handler in self._handlers.items()
+                        if handler.__class__.__module__ == name
+                    ]
+                    for handler in sorted(stale_handlers):
+                        logger.info("Unlinking %s", handler)
+                        self._handlers.pop(handler)
+
+                    if m_mtime:
+                        logger.info("Reloading %s", name)
+                        reload(module)
+                    else:
+                        logger.info("Loaded %s", name)
+                    module.__mtime__ = f_mtime
+                yield module
+            except Exception:
+                logger.exception("EXCEPTION")
+
+    @property
+    def handler_classes(self):
+        for module in self.modules:
+            for attr in dir(module):
+                if attr.startswith("_"):
+                    continue
+                item = getattr(module, attr)
+                if item is self.handler_type:
+                    continue
+                try:
+                    if issubclass(item, self.handler_type):
+                        yield item
+                except TypeError:
+                    pass
+
+    @property
+    def handlers(self) -> Iterable[Handler]:
+        for cls in self.handler_classes:
+            fullname = f"{cls.__module__}.{cls.__name__}"
+            handler = self._handlers.get(fullname)
+            try:
+                if isinstance(handler, cls):
+                    yield handler
+                    continue
+
+                logger.info(
+                    "%s %s",
+                    ("Recreating" if handler else "Creating"),
+                    fullname,
+                )
+                self._handlers[fullname] = handler = cls()
+                yield handler
+
+            except Exception:
+                logger.exception("EXCEPTION")
+
+    def __iter__(self):
+        return (
+            handler
+            for _, handler in sorted(
+                    ((handler.priority,
+                      handler.__class__.__module__,
+                      handler.__class__.__name__,
+                      id(handler)),
+                     handler)
+                    for handler in self.handlers
+            )
+        )

services/chat-router/hive/chat_router/service.py

@@ -0,0 +1,56 @@
+import logging
+
+from collections.abc import Iterable
+from dataclasses import dataclass, field
+
+from hive.chat import ChatMessage
+from hive.common import SmallCircularBuffer
+from hive.messaging import Channel, Message
+from hive.service import HiveService
+
+from .handler import Handler
+from .loader import HandlerLoader
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class Service(HiveService):
+    handlers: Iterable[Handler] = field(default_factory=HandlerLoader)
+    our_recently_sent_messages: SmallCircularBuffer = field(
+        default_factory=lambda: SmallCircularBuffer(8, coerce=str),
+    )
+
+    def _on_channel_open(self, channel: Channel):
+        channel.add_pre_publish_hook(self.on_publish_event)
+
+    def on_publish_event(self, channel: Channel, **kwargs):
+        if not (message := kwargs.get("message")):
+            return
+        if not (uuid := message.get("uuid")):
+            return
+        self.our_recently_sent_messages.add(uuid)
+
+    def on_chat_message(self, channel: Channel, message: Message):
+        message = ChatMessage.from_json(message.json())
+        if message.uuid in self.our_recently_sent_messages:
+            return
+
+        for handler in self.handlers:
+            try:
+                if handler.handle(channel, message):
+                    return
+            except Exception:
+                logger.exception("EXCEPTION processing %s", message)
+        logger.warning("Unhandled %s", message)
+
+    def run(self):
+        with self.blocking_connection(
+                on_channel_open=self._on_channel_open,
+        ) as conn:
+            channel = conn.channel()
+            channel.consume_events(
+                queue="chat.messages",
+                on_message_callback=self.on_chat_message,
+            )
+            channel.start_consuming()