Implement attachment view

core/modules/router.py

@@ -1,4 +1,7 @@
+import asyncio
+
 from asgiref.sync import sync_to_async
+from django.conf import settings
 
 from core.clients import transport
 from core.clients.instagram import InstagramClient
@@ -7,6 +10,7 @@ from core.clients.whatsapp import WhatsAppClient
 from core.clients.xmpp import XMPPClient
 from core.messaging import history
 from core.models import PersonIdentifier
+from core.realtime.typing_state import set_person_typing_state
 from core.util import logs
 
 
@@ -17,6 +21,10 @@ class UnifiedRouter(object):
 
     def __init__(self, loop):
         self.loop = loop
+        self.typing_auto_stop_seconds = int(
+            getattr(settings, "XMPP_TYPING_AUTO_STOP_SECONDS", 3)
+        )
+        self._typing_stop_tasks = {}
 
         self.log = logs.get_logger("router")
         self.log.info("Initialised Unified Router Interface.")
@@ -26,6 +34,42 @@ class UnifiedRouter(object):
         self.whatsapp = WhatsAppClient(self, loop, "whatsapp")
         self.instagram = InstagramClient(self, loop, "instagram")
 
+    def _typing_task_key(self, target):
+        return (
+            int(target.user_id),
+            int(target.person_id),
+            str(target.service),
+            str(target.identifier),
+        )
+
+    def _cancel_typing_timer(self, key):
+        existing = self._typing_stop_tasks.pop(key, None)
+        if existing and not existing.done():
+            existing.cancel()
+
+    def _schedule_typing_auto_stop(self, target):
+        key = self._typing_task_key(target)
+        self._cancel_typing_timer(key)
+        delay = max(1, int(self.typing_auto_stop_seconds))
+
+        async def _timer():
+            try:
+                await asyncio.sleep(delay)
+                await transport.stop_typing(target.service, target.identifier)
+            except asyncio.CancelledError:
+                return
+            except Exception as exc:
+                self.log.warning(
+                    "Typing auto-stop failed for %s/%s: %s",
+                    target.service,
+                    target.identifier,
+                    exc,
+                )
+            finally:
+                self._typing_stop_tasks.pop(key, None)
+
+        self._typing_stop_tasks[key] = self.loop.create_task(_timer())
+
     def _start(self):
         print("UR _start")
         self.xmpp.start()
@@ -86,6 +130,23 @@ class UnifiedRouter(object):
         identifier = kwargs.get("identifier")
         identifiers = await self._resolve_identifier_objects(protocol, identifier)
         for src in identifiers:
+            if protocol != "xmpp":
+                set_person_typing_state(
+                    user_id=src.user_id,
+                    person_id=src.person_id,
+                    started=True,
+                    source_service=protocol,
+                    display_name=src.person.name,
+                )
+                try:
+                    await self.xmpp.start_typing_for_person(src.user, src)
+                except Exception as exc:
+                    self.log.warning(
+                        "Failed to relay typing-start to XMPP for %s: %s",
+                        src.identifier,
+                        exc,
+                    )
+
             targets = await sync_to_async(list)(
                 PersonIdentifier.objects.filter(
                     user=src.user,
@@ -93,13 +154,34 @@ class UnifiedRouter(object):
                 ).exclude(service=protocol)
             )
             for target in targets:
+                if target.service == "xmpp":
+                    continue
                 await transport.start_typing(target.service, target.identifier)
+                if protocol == "xmpp":
+                    self._schedule_typing_auto_stop(target)
 
     async def stopped_typing(self, protocol, *args, **kwargs):
         self.log.info(f"Stopped typing ({protocol}) {args} {kwargs}")
         identifier = kwargs.get("identifier")
         identifiers = await self._resolve_identifier_objects(protocol, identifier)
         for src in identifiers:
+            if protocol != "xmpp":
+                set_person_typing_state(
+                    user_id=src.user_id,
+                    person_id=src.person_id,
+                    started=False,
+                    source_service=protocol,
+                    display_name=src.person.name,
+                )
+                try:
+                    await self.xmpp.stop_typing_for_person(src.user, src)
+                except Exception as exc:
+                    self.log.warning(
+                        "Failed to relay typing-stop to XMPP for %s: %s",
+                        src.identifier,
+                        exc,
+                    )
+
             targets = await sync_to_async(list)(
                 PersonIdentifier.objects.filter(
                     user=src.user,
@@ -107,6 +189,9 @@ class UnifiedRouter(object):
                 ).exclude(service=protocol)
             )
             for target in targets:
+                if target.service == "xmpp":
+                    continue
+                self._cancel_typing_timer(self._typing_task_key(target))
                 await transport.stop_typing(target.service, target.identifier)
 
     async def reacted(self, protocol, *args, **kwargs):