add basic websocket server for notifications

Author: neckhair

.github/workflows/docker-images.yaml

@@ -18,6 +18,7 @@ jobs:
           registry: ghcr.io
           username: ${{ github.actor }}
           password: ${{ secrets.GITHUB_TOKEN }}
+
       - name: "Tasks API: Build and push Docker image"
         uses: docker/build-push-action@v6
         with:
@@ -26,6 +27,7 @@ jobs:
           tags: |
             ghcr.io/${{ github.repository }}/tasks:latest
             ghcr.io/${{ github.repository }}/tasks:${{ github.sha }}
+
       - name: "Frontend: Build and push Docker image"
         uses: docker/build-push-action@v6
         with:
@@ -34,3 +36,12 @@ jobs:
           tags: |
             ghcr.io/${{ github.repository }}/frontend:latest
             ghcr.io/${{ github.repository }}/frontend:${{ github.sha }}
+
+      - name: "Websocket Server: Build and push Docker image"
+        uses: docker/build-push-action@v6
+        with:
+          push: ${{ github.event_name != 'pull_request' }}
+          context: backend/websocket-server
+          tags: |
+            ghcr.io/${{ github.repository }}/websocket-server:latest
+            ghcr.io/${{ github.repository }}/websocket-server:${{ github.sha }}

README.md

@@ -8,7 +8,8 @@ and deploy a modern web application with several different services in the backg
 This is an overview. You will find more detailed explanations for the components in their own Readme. The components are intentionally written in different languages and frameworks. First, we wanted to showcase the interactions between different kinds of applications, and second, we wanted to learn and have fun.
 
 - **[Frontend](./frontend/)** A simple Svelte client application.
-- **[Tasks API](./backend/tasks/)** The API for interacting with tasks, written in Go, using the Fiber framework.
+- **[Tasks API](./backend/tasks/)** The API for interacting with tasks, written in Go, using the Fiber framework. Changes to tasks are published as events to Redis.
+- **[WebsocketServer](./backend/websocket-server/)** Listens for events in Redis and broadcasts them to all connected Websocket clients. The frontend connects to it by default.
 
 ## Services
 
@@ -49,11 +50,13 @@ graph LR;
     subgraph Backend
         TasksAPI-- :27017 --> MongoDB
         TasksAPI-- :6379 --> Redis
+        WebsocketServer-- :6379 -->Redis
     end
 
     subgraph lb[Load Balancer]
         Traeffik-- :5173 --> Frontend
         Traeffik-- :3000 --> TasksAPI
+        Traeffik-- :8765 --> WebsocketServer
     end
 
 

apps.compose.yaml

@@ -41,3 +41,19 @@ services:
       - "traefik.http.services.tasks.loadbalancer.server.port=3000"
     networks:
       - internal
+
+  websocket-server:
+    build:
+      context: backend/websocket-server
+    environment:
+      - REDIS_URL=redis://redis:6379/0
+      - WEBSOCKET_SERVER_HOST=0.0.0.0
+    labels:
+      - "traefik.enable=true"
+      - "traefik.http.routers.websocket-server.rule=PathPrefix(`/ws`)"
+      - "traefik.http.routers.websocket-server.entrypoints=web"
+      - "traefik.http.services.websocket-server.loadbalancer.server.port=8765"
+    volumes:
+      - ./backend/websocket-server:/app
+    networks:
+      - internal

backend/websocket-server/.gitignore

@@ -0,0 +1 @@
+__pycache__

backend/websocket-server/Dockerfile

@@ -0,0 +1,10 @@
+FROM python:3.13-alpine
+
+WORKDIR /app
+
+RUN pip install poetry
+
+COPY . /app
+RUN poetry install
+
+CMD [ "poetry", "run", "python", "main.py" ]

backend/websocket-server/README.md

@@ -0,0 +1,12 @@
+# Websocket Server
+
+This app is a websocket server, which broadcasts messages it receives from the message queue to all connected clients.
+
+It is a Python app with Poetry. You can run it like this:
+
+```sh
+poetry install
+poetry run python main.py
+```
+
+Or just run it via `docker-compose` from the main directory.

backend/websocket-server/app/__init__.py

@@ -0,0 +1,31 @@
+import asyncio
+import logging
+from typing import Callable
+
+import redis.asyncio
+from redis.asyncio import from_url
+from tenacity import retry, retry_if_exception_type, wait_fixed
+
+RETRY_INTERVALL = 2
+
+
+async def redis_reader(channel: redis.asyncio.client.PubSub, callback: Callable[[str], None]):
+    while True:
+        message = await channel.get_message(ignore_subscribe_messages=True)
+
+        if message and message["data"]:
+            data = message["data"].decode("utf-8")
+            logging.info(f"Received data from Redis: {data}")
+            callback(data)
+
+
+@retry(wait=wait_fixed(RETRY_INTERVALL), retry=retry_if_exception_type(redis.exceptions.ConnectionError))
+async def start_redis_subscription(url: str, callback: Callable[[str], None]):
+    client = from_url(url)
+    async with client.pubsub() as pubsub:
+        try:
+            await pubsub.psubscribe("events.*")
+            await asyncio.create_task(redis_reader(pubsub, callback))
+        except redis.exceptions.ConnectionError as e:
+            logging.error(f"Connection to Redis failed, trying again in {RETRY_INTERVALL} seconds.")
+            raise e

backend/websocket-server/app/pubsub.py

@@ -0,0 +1,27 @@
+import asyncio
+import logging
+from typing import Set
+
+import websockets.asyncio.connection
+from websockets.asyncio.server import ServerConnection, serve
+
+connected_clients: Set[ServerConnection] = set()
+
+
+def broadcast(message: str):
+    websockets.asyncio.connection.broadcast(connected_clients.copy(), message)
+
+
+async def start_websocket_server(host: str, port: int):
+    async with serve(websocket_handler, host, port):
+        await asyncio.get_running_loop().create_future()
+
+
+async def websocket_handler(websocket: ServerConnection):
+    logging.info(f"Websocket client connected from {websocket.remote_address[0]}")
+    connected_clients.add(websocket)
+
+    async for message in websocket:
+        logging.info(f"Client sent a message: {str(message)}")
+
+    connected_clients.remove(websocket)

backend/websocket-server/app/websockets.py

@@ -0,0 +1,43 @@
+import asyncio
+import logging
+import os
+from signal import SIGINT, SIGTERM
+
+from app.pubsub import start_redis_subscription
+from app.websockets import broadcast, start_websocket_server
+
+logging.basicConfig(
+    format="%(asctime)s %(message)s",
+    level=logging.INFO,
+)
+
+host = os.getenv("WEBSOCKET_SERVER_HOST", "localhost")
+port = int(os.getenv("WEBSOCKET_SERVER_PORT", 8765))
+redis_url = os.getenv("WEBSOCKET_SERVER_REDIS_URL", "redis://redis")
+
+
+def send_message_to_all(message: str):
+    broadcast(message)
+
+
+async def main():
+    async with asyncio.TaskGroup() as tg:
+        tg.create_task(start_websocket_server(host, port))
+        tg.create_task(start_redis_subscription(redis_url, send_message_to_all))
+
+
+if __name__ == "__main__":
+    logging.info("🎸 Websocket server started.")
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    main_task = asyncio.ensure_future(main())
+    for signal in [SIGINT, SIGTERM]:
+        loop.add_signal_handler(signal, main_task.cancel)
+
+    try:
+        loop.run_until_complete(main_task)
+    except asyncio.exceptions.CancelledError:
+        pass
+    finally:
+        loop.close()