Implement AI workspace and mitigation workflow

core/views/queues.py

@@ -1,56 +1,63 @@
-from rest_framework.views import APIView
+from asgiref.sync import async_to_sync
 from django.contrib.auth.mixins import LoginRequiredMixin
-
-from rest_framework import status
-
+from django.db import transaction
 from django.http import HttpResponse
-from core.models import QueuedMessage, Message
+from mixins.views import ObjectCreate, ObjectDelete, ObjectList, ObjectUpdate
+from rest_framework import status
+from rest_framework.views import APIView
+
+from core.clients import signalapi
 from core.forms import QueueForm
+from core.models import Message, QueuedMessage
+from core.util import logs
 
-import requests
-import orjson
-from django.conf import settings
-import redis
-import msgpack
-
-from mixins.views import (
-    ObjectCreate,
-    ObjectDelete,
-    ObjectList,
-    ObjectUpdate,
-)
-
-# def start_typing(uuid):
-#     url = f"http://signal:8080/v1/typing_indicator/{settings.SIGNAL_NUMBER}"
-#     data = {
-#         "recipient": uuid,
-#     }
-
-#     response = requests.put(url, json=data)
-
-# def stop_typing(uuid):
-#     url = f"http://signal:8080/v1/typing_indicator/{settings.SIGNAL_NUMBER}"
-#     data = {
-#         "recipient": uuid,
-#     }
-
-#     response = requests.delete(url, json=data)
-
-r = redis.from_url("unix://var/run/gia-redis.sock", db=10)
+log = logs.get_logger("queue")
 
 
 class AcceptMessageAPI(LoginRequiredMixin, APIView):
     def get(self, request, message_id):
-        to_submit = {
-            "type": "def",
-            "method": "accept_message",
-            "user_id": request.user.id,
-            "message_id": message_id,
-        }
-        packed = msgpack.packb(to_submit, use_bin_type=True)
-        r.publish("processing", packed)
+        try:
+            queued = QueuedMessage.objects.select_related(
+                "session",
+                "session__identifier",
+                "session__user",
+            ).get(
+                user=request.user,
+                id=message_id,
+            )
+        except QueuedMessage.DoesNotExist:
+            return HttpResponse(status=status.HTTP_404_NOT_FOUND)
+
+        if queued.session.identifier.service != "signal":
+            log.warning(
+                "Queue accept failed: unsupported service '%s' for queued message %s",
+                queued.session.identifier.service,
+                queued.id,
+            )
+            return HttpResponse(status=status.HTTP_400_BAD_REQUEST)
+
+        ts = async_to_sync(signalapi.send_message_raw)(
+            queued.session.identifier.identifier,
+            queued.text or "",
+            [],
+        )
+        if not ts:
+            log.error("Queue accept send failed for queued message %s", queued.id)
+            return HttpResponse(status=status.HTTP_502_BAD_GATEWAY)
+
+        with transaction.atomic():
+            Message.objects.create(
+                user=queued.session.user,
+                session=queued.session,
+                custom_author=queued.custom_author or "BOT",
+                text=queued.text,
+                ts=ts,
+            )
+            queued.delete()
+
         return HttpResponse(status=status.HTTP_200_OK)
 
+
 class RejectMessageAPI(LoginRequiredMixin, APIView):
     def get(self, request, message_id):
         try:
@@ -64,11 +71,12 @@ class RejectMessageAPI(LoginRequiredMixin, APIView):
         message.delete()
 
         return HttpResponse(status=status.HTTP_200_OK)
-    
+
+
 class QueueList(LoginRequiredMixin, ObjectList):
     list_template = "partials/queue-list.html"
     model = QueuedMessage
-    page_title = "Queues"
+    page_title = "Queue"
 
     list_url_name = "queues"
     list_url_args = ["type"]