endpoint /api/ws/telemtry with auth, broadcast, db

docker-compose.yml

@@ -1,4 +1,3 @@
-version: '3.8'
 
 services:
   db:
@@ -12,9 +11,21 @@ services:
     networks:
       - app_network
 
+  redis:
+    image: redis:latest
+    container_name: redis
+    ports:
+      - "6379:6379"
+    volumes:
+      - redis_data:/data
+    networks:
+      - app_network
+
   web:
     build: .
     command: daphne -b 0.0.0.0 -p 8000 testapi.asgi:application
+    environment:
+      DJANGO_SETTINGS_MODULE: testapi.settings
     ports:
       - "8000:8000"
     volumes:
@@ -40,6 +51,7 @@ services:
       - app_network
 
 volumes:
+  redis_data:
   postgres_data:
   static_volume:
   media_volume:

nginx/nginx.conf

@@ -4,7 +4,10 @@ http {
     upstream django {
         server web:8000;
     }
-
+    map $http_upgrade $connection_upgrade {
+    default upgrade;
+    ''      close;
+    }
     server {
         listen 80;
         server_name localhost;
@@ -36,7 +39,7 @@ http {
 
             proxy_http_version 1.1;
             proxy_set_header Upgrade $http_upgrade;
-            proxy_set_header Connection "upgrade";
+            proxy_set_header Connection $connection_upgrade;
         }
     }
 }

requirements.txt

@@ -8,3 +8,4 @@ django-cors-headers
 Pillow
 python-dotenv
 channels>=4.0
+channels_redis

testapi/asgi.py

@@ -11,18 +11,15 @@ import os
 from channels.routing import ProtocolTypeRouter, URLRouter
 from channels.auth import AuthMiddlewareStack
 from django.core.asgi import get_asgi_application
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'testapi.settings')
+django_asgi_app = get_asgi_application()
 
 from .routing import websocket_urlpatterns
 
-os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'testapi.settings')
-
-django_asgi_app = get_asgi_application()
 
 application = ProtocolTypeRouter({
-    "http": django_asgi_app,
+    'http': django_asgi_app,
-    "websocket": AuthMiddlewareStack(
+    'websocket': AuthMiddlewareStack(
-        URLRouter(
+        URLRouter(websocket_urlpatterns)
-             websocket_urlpatterns
+    )
-        )
-    ),
 })

testapi/consumers.py

@@ -1,14 +1,87 @@
-
-from channels.generic.websocket import AsyncWebsocketConsumer
 import json
+import time
+from channels.generic.websocket import AsyncWebsocketConsumer
+from channels.db import database_sync_to_async
+from django.contrib.auth.models import AnonymousUser
+from rest_framework.authtoken.models import Token
+from api.models import TelemetryPacket, Satellite
+from api.serializers import TelemetryPacketSerializer
+from django.contrib.auth import get_user_model
 
-class EchoConsumer(AsyncWebsocketConsumer):
+User = get_user_model()
+
+class TelemetryConsumer(AsyncWebsocketConsumer):
     async def connect(self):
+        token_key = self.scope["query_string"].decode().split("token=")[-1]
+        self.satellite_id = self.scope["url_route"]["kwargs"]["pk"]
+        self.group_name = f"telemetry_{self.satellite_id}"
+        try:
+            self.token = await database_sync_to_async(Token.objects.select_related("user").get)(key=token_key)
+            self.scope["user"] = self.token.user
+        except Token.DoesNotExist:
+            print("Token.DoesNotExist")
+            await self.close()
+            return
+
+        try:
+            self.satellite = await database_sync_to_async(Satellite.objects.get)(id=self.satellite_id)
+        except Satellite.DoesNotExist:
+            print("Satellite.DoesNotExist")
+            await self.close()
+            return
+
+        
+        await self.channel_layer.group_add(self.group_name, self.channel_name)
+        print("CONNECT success")
         await self.accept()
-        await self.send(text_data=json.dumps({"message": "WebSocket connected!"}))
 
     async def disconnect(self, close_code):
-        pass
+        await self.channel_layer.group_discard(self.group_name, self.channel_name)
+
+    async def telemetry_message(self, event):
+        data = {
+            "id": event["id"],
+            "timestamp": event["timestamp"],
+            "lat": event["lat"],
+            "lon": event["lon"],
+            "alt": event["alt"],
+        }
+        await self.send(text_data=json.dumps(data))
 
     async def receive(self, text_data):
-        await self.send(text_data=json.dumps({"echo": text_data}))
+        data = json.loads(text_data)
+        saved_data = await self.save_telemetry(data)
+
+        message = {
+            "type": "telemetry.message",
+            "id": str(saved_data.id),
+            "timestamp": saved_data.timestamp,
+            "lat": saved_data.lat,
+            "lon": saved_data.lon,
+            "alt": saved_data.alt,
+        }
+
+        await self.channel_layer.group_send(self.group_name, message)
+
+
+    @database_sync_to_async
+    def get_user_from_token(self, token_key):
+        try:
+            token = Token.objects.select_related("user").get(key=token_key)
+            return token.user
+        except Token.DoesNotExist:
+            return None
+
+    @database_sync_to_async
+    def save_telemetry(self, data):
+        packet = TelemetryPacket.objects.create(
+            satellite=self.satellite,
+            user=self.satellite.user,
+            timestamp=int(time.time()),
+            lat=data.get("lat"),
+            lon=data.get("lon"),
+            alt=data.get("alt"),
+        )
+        print(f"Saved telemetry for satellite {self.satellite.id}")
+        return packet
+            

testapi/routing.py

@@ -1,8 +1,6 @@
-# routing.py
+from django.urls import re_path
-
-from django.urls import path
 from . import consumers
 
 websocket_urlpatterns = [
-    path("ws/echo/", consumers.EchoConsumer.as_asgi()),
+    re_path(r'^api/ws/(?P<pk>[0-9a-f-]+)/telemetry/$', consumers.TelemetryConsumer.as_asgi()),
 ]

testapi/settings.py

@@ -104,7 +104,7 @@ ASGI_APPLICATION = 'testapi.asgi.application'
 # Database
 # https://docs.djangoproject.com/en/5.1/ref/settings/#databases
 
-if not PRODUCTION:
+if PRODUCTION:
     DATABASES = {
         'default': {
             'ENGINE': 'django.db.backends.sqlite3',
@@ -195,3 +195,11 @@ SESSION_COOKIE_SECURE = False
 CSRF_TRUSTED_ORIGINS = os.getenv('CSRF_TRUSTED_ORIGINS', 'http://localhost:5173, http://localhost:8000').split(',')
 
 
+CHANNEL_LAYERS = {
+    "default": {
+        "BACKEND": "channels_redis.core.RedisChannelLayer",
+        "CONFIG": {
+            "hosts": [("redis", 6379)],
+        },
+    },
+}