better with files...

deployment/channels.py

@@ -0,0 +1,51 @@
+from django.contrib.auth.models import User
+from django_eventstream import send_event
+from django_eventstream.channelmanager import DefaultChannelManager
+
+from deployment.models import Deployment
+
+
+def parse_channel(channel: str) -> tuple[str] | None:
+    parts = channel.lstrip("deployment_").split("_")
+    if len_part := len(parts):
+        if len_part == 1:
+            return (int(parts[0]), "")
+        if len_part == 2:
+            return (int(parts[0]), parts[1])
+    return
+
+
+class DeploymentChannelManager(DefaultChannelManager):
+    def can_read_channel(self, user: User, channel: str):
+        if not user.has_perm("django_eventstream.view_event"):
+            return False
+
+        match parse_channel(channel):
+            case (user_id, ""):
+                return user_id == user.id or user.is_superuser
+            case (user_id, _):
+                # TODO(rmanach): check if the deployment belongs to the user
+                return user_id == user.id or user.is_superuser
+        return False
+
+
+class Event:
+    @staticmethod
+    def send_details(deployment: Deployment, progress: int):
+        send_event(
+            f"deployment_{deployment.user.id}_{deployment.id}",
+            "message",
+            {
+                "id": deployment.id,
+                "status": deployment.status,
+                "progress": progress,
+            },
+        )
+
+    @staticmethod
+    def send(deployment: Deployment):
+        send_event(
+            f"deployment_{deployment.user.id}",
+            "message",
+            {"id": deployment.id, "status": deployment.status},
+        )

deployment/signals.py

@@ -0,0 +1,27 @@
+from django.contrib.auth.models import User, Permission
+from django.db import transaction
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+
+from deployment.models import DeploymentUser
+
+
+@receiver(post_save, sender=User)
+def user_created_callback(sender, instance: User, created, **kwargs):
+    if created:
+        try:
+            with transaction.atomic():
+                if not instance.is_superuser:
+                    permission = Permission.objects.get(
+                        codename="view_event",
+                        content_type__app_label="django_eventstream",
+                    )
+                    instance.user_permissions.add(permission)
+                    permission.save()
+
+                user = DeploymentUser(
+                    user=instance, credits=-1 if instance.is_superuser else 3
+                )
+                user.save()
+        except Exception as e:
+            print(f"deployment user creation failed: {e}")