Implement XMPP relaying

core/lib/deferred.py

@@ -1,25 +1,35 @@
 # Deferred processing library
 from core.util import logs
 from pydantic import BaseModel
-from typing import Annotated
+from typing import Annotated, Optional
 from uuid import UUID
 from pydantic import ValidationError
-from core.models import QueuedMessage, Message
+from core.models import QueuedMessage, Message, PersonIdentifier, User
 from core.clients import signal
 from core.lib.prompts.functions import delete_messages
 from asgiref.sync import sync_to_async
+from django.conf import settings
 
 
 log = logs.get_logger("deferred")
 
+class DeferredDetail(BaseModel):
+    reply_to_self: bool
+    reply_to_others: bool
+    is_outgoing_message: bool
 
 class DeferredRequest(BaseModel):
     type: str
     method: str
-    user_id: int
-    message_id: Annotated[str, UUID]
+    user_id: Optional[int] = None
+    message_id: Optional[Annotated[str, UUID]] = None
+    identifier: Optional[str] = None
+    msg: Optional[str] = None
+    service: Optional[str] = None
+    detail: Optional[DeferredDetail] = None
 
-async def process_deferred(data: dict):
+
+async def process_deferred(data: dict, **kwargs):
     try:
         validated_data = DeferredRequest(**data)
         log.info(f"Validated Data: {validated_data}")
@@ -32,23 +42,40 @@ async def process_deferred(data: dict):
     user_id = validated_data.user_id
     message_id = validated_data.message_id
 
-    try:
-        message = await sync_to_async(QueuedMessage.objects.get)(
-            user_id=user_id,
-            id=message_id,
-        )
-        log.info(f"Got {message}")
-    except QueuedMessage.DoesNotExist:
-        log.info(f"Didn't get message from {message_id}")
-        return
 
-    if message.session.identifier.service == "signal":
-        log.info(f"Is sisngla")
-        if method == "accept_message":
-            await signal.send_message(message)
-        else:
-            log.warning(f"Method not yet supported: {method}")
+    if method == "accept_message":
+        try:
+            message = await sync_to_async(QueuedMessage.objects.get)(
+                user_id=user_id,
+                id=message_id,
+            )
+            log.info(f"Got {message}")
+        except QueuedMessage.DoesNotExist:
+            log.info(f"Didn't get message from {message_id}")
             return
+
+        if message.session.identifier.service == "signal":
+            await signal.send_message(message)
+
+        else:
+            log.warning(f"Protocol not supported: {message.session.identifier.service}")
+            return
+    elif method == "xmpp":
+        xmpp = kwargs.get("xmpp")
+        service = validated_data.service
+        msg = validated_data.msg
+
+        # Get User from identifier
+        identifiers = PersonIdentifier.objects.filter(
+            identifier=validated_data.identifier,
+            service=service,
+        )
+
+        for identifier in identifiers:
+            # Fair is fair, we can have multiple
+            log.info(f"Sending {msg} from {identifier}")
+            xmpp.send_from_external(identifier, msg, validated_data.detail)
+    
     else:
-        log.warning(f"Protocol not supported: {message.session.identifier.service}")
-        return
+        log.warning(f"Method not yet supported: {method}")
+        return