Add twitch-events service

docker-compose.yaml.example

@@ -17,6 +17,7 @@ services:
             - twitch-cache-api
             - twitch-subs-api
             - twitch-webhooks
+            - twitch-events
             - cms
 
     # Cheese Horde members API service with /data/horde-members mounted as database storage
@@ -118,6 +119,16 @@ services:
         expose:
             - 5000
 
+    # Twitch events WSS service
+    # SECRET_KEY is needed for validation
+    twitch-events:
+        build:
+            context: ./twitch-events
+        environment:
+            - SECRET_KEY=__SECRET_KEY__
+        expose:
+            - 8765
+
     # CMS service with /data/grav mounted as user data storage
     cms:
         build:

nginx/nginx.conf

@@ -136,6 +136,16 @@ http {
             proxy_pass          http://twitch-webhooks:5000;
         }
 
+        location ^~ /twitch-events/ {
+            rewrite             ^/twitch-events(/.*)$ $1 break;
+            proxy_set_header    Host        $host;
+            proxy_set_header    X-Real-IP   $remote_addr;
+            proxy_http_version  1.1;
+            proxy_set_header    Upgrade     $http_upgrade;
+            proxy_set_header    Connection  "upgrade";
+            proxy_pass          http://twitch-events:8765;
+        }
+
         location ^~ /schedule {
             return              301     /#livestreams;
         }

twitch-events/.gitignore

@@ -0,0 +1,96 @@
+# ---> Python
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*,cover
+.hypothesis/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# dotenv
+.env
+
+# virtualenv
+.venv
+venv/
+ENV/
+
+# Spyder project settings
+.spyderproject
+
+# Rope project settings
+.ropeproject
+

twitch-events/Dockerfile

@@ -0,0 +1,14 @@
+FROM python:alpine
+
+WORKDIR /app
+COPY . .
+
+RUN pip install --no-cache-dir --requirement requirements.txt
+
+RUN addgroup -g 9999 lilia
+
+EXPOSE 8765 2828/udp
+
+USER nobody:lilia
+
+ENTRYPOINT ["python", "-u", "server.py"]

twitch-events/requirements.txt

@@ -0,0 +1,2 @@
+itsdangerous
+websockets

twitch-events/server.py

@@ -0,0 +1,55 @@
+import asyncio
+import datetime
+import http
+import os
+
+import itsdangerous
+import websockets
+
+
+clients = set()
+
+queue = asyncio.Queue()
+
+
+class Protocol(asyncio.DatagramProtocol):
+    def datagram_received(self, data, addr):
+        queue.put_nowait(data)
+
+
+async def serve(websocket, path):
+    async def send(ws, message):
+        try:
+            await ws.send(message)
+        except websockets.exceptions.ConnectionClosed:
+            pass
+    print(datetime.datetime.utcnow().isoformat(), 'Connection from', websocket.remote_address)
+    clients.add(websocket)
+    try:
+        while True:
+            message = (await queue.get()).strip().decode()
+            print(datetime.datetime.utcnow().isoformat(), 'Sending', message)
+            await asyncio.wait([send(ws, message) for ws in clients])
+    finally:
+        clients.remove(websocket)
+
+
+async def process_request(path, request_headers):
+    error = (http.HTTPStatus.UNAUTHORIZED, {}, bytes())
+    key = request_headers.get('X-Events-Key')
+    if not key:
+        return error
+    s = itsdangerous.TimedJSONWebSignatureSerializer(os.getenv('SECRET_KEY'))
+    try:
+        s.loads(key)
+    except (itsdangerous.SignatureExpired, itsdangerous.BadSignature):
+        return error
+
+
+if __name__ == '__main__':
+    loop = asyncio.get_event_loop()
+    start_consumer = loop.create_datagram_endpoint(Protocol, local_addr=('0.0.0.0', 2828))
+    loop.run_until_complete(start_consumer)
+    start_server = websockets.serve(serve, '0.0.0.0', 8765, process_request=process_request)
+    loop.run_until_complete(start_server)
+    loop.run_forever()