plugin XEP-0045: MAM archive batching:

Message archives (MAM) are sent by batches to avoid flooding the bridge when the client is
retrieving big ones.

libervia/backend/bridge/bridge_constructor/bridge_template.ini

@@ -61,6 +61,14 @@ doc_param_7=extra: extra message information, can have data added by plugins and
   - info_type: subtype for info messages
 doc_param_8=%(doc_profile)s
 
+[messages_new]
+type=signal
+category=core
+sig_in=a(sdssa{ss}a{ss}sss)s
+doc=Batch of messages received (for MAM history sync)
+doc_param_0=messages: list of messages, each as for message_new (uid, timestamp, from_jid, to_jid, message, subject, mess_type, extra)
+doc_param_1=%(doc_profile)s
+
 [message_update]
 type=signal
 category=core

libervia/backend/bridge/dbus_bridge.py

@@ -175,6 +175,7 @@ class DBusObject(objects.DBusObject):
         Signal("message_encryption_stopped", "sa{ss}s"),
         Signal("message_new", "sdssa{ss}a{ss}sss"),
         Signal("message_update", "ssss"),
+        Signal("messages_new", "a(sdssa{ss}a{ss}sss)s"),
         Signal("notification_deleted", "ss"),
         Signal("notification_new", "sdssssbidss"),
         Signal("param_update", "ssss"),
@@ -627,6 +628,9 @@ class bridge:
     def message_update(self, uid, message_type, message_data, profile):
         self._obj.emitSignal("message_update", uid, message_type, message_data, profile)
 
+    def messages_new(self, messages, profile):
+        self._obj.emitSignal("messages_new", messages, profile)
+
     def notification_deleted(self, id, profile):
         self._obj.emitSignal("notification_deleted", id, profile)
 

libervia/backend/bridge/pb.py

@@ -236,6 +236,9 @@ class bridge:
     def message_update(self, uid, message_type, message_data, profile):
         self.send_signal("message_update", uid, message_type, message_data, profile)
 
+    def messages_new(self, messages, profile):
+        self.send_signal("messages_new", messages, profile)
+
     def notification_deleted(self, id, profile):
         self.send_signal("notification_deleted", id, profile)
 

libervia/backend/core/xmpp.py

@@ -460,9 +460,7 @@ class SatXMPPEntity(core_types.SatXMPPEntity):
                     f"  Plugins total={total_plugins:.2f}s real={total_real:.2f}s\n"
                 )
 
-            list_d.addCallback(
-                log_plugin_results
-            )
+            list_d.addCallback(log_plugin_results)
             # FIXME: we should have a timeout here, and a way to know if a plugin freeze
             await list_d
 
@@ -1568,6 +1566,7 @@ class SatMessageProtocol(xmppim.MessageProtocol):
     def __init__(self):
         xmppim.MessageProtocol.__init__(self)
         self.queue_manager = MessagesQueueManager(self.process_message)
+        self.archive_batch: list[dict] = []
 
     def setHandlerParent(self, parent):
         super().setHandlerParent(parent)
@@ -1806,6 +1805,12 @@ class SatMessageProtocol(xmppim.MessageProtocol):
                 )
         return data
 
+    def flush_archive_batch(self) -> None:
+        """Flush the archive message batch to the bridge."""
+        if self.archive_batch:
+            G.host.bridge.messages_new(self.archive_batch, self.parent.profile)
+            self.archive_batch.clear()
+
     def bridge_signal(self, data: MessageData) -> MessageData:
         """Send signal to frontends for the given message"""
         try:
@@ -1817,17 +1822,34 @@ class SatMessageProtocol(xmppim.MessageProtocol):
             data["extra"]["encrypted"] = True
         if data is not None:
             if self.parent.is_message_printable(data):
-                G.host.bridge.message_new(
-                    data["uid"],
-                    data["timestamp"],
-                    data["from"].full(),
-                    data["to"].full(),
-                    data["message"],
-                    data["subject"],
-                    data["type"],
-                    data_format.serialise(data["extra"]),
-                    profile=self.parent.profile,
-                )
+                if data["extra"].get("is_from_archive"):
+                    # Archives are sent in batches to avoid flooding the bridge.
+                    self.archive_batch.append(
+                        {
+                            "uid": data["uid"],
+                            "timestamp": data["timestamp"],
+                            "from_jid": data["from"].full(),
+                            "to_jid": data["to"].full(),
+                            "message": data["message"],
+                            "subject": data["subject"],
+                            "mess_type": data["type"],
+                            "extra": data_format.serialise(data["extra"]),
+                        }
+                    )
+                    if len(self.archive_batch) >= 50:
+                        self.flush_archive_batch()
+                else:
+                    G.host.bridge.message_new(
+                        data["uid"],
+                        data["timestamp"],
+                        data["from"].full(),
+                        data["to"].full(),
+                        data["message"],
+                        data["subject"],
+                        data["type"],
+                        data_format.serialise(data["extra"]),
+                        profile=self.parent.profile,
+                    )
             else:
                 log.debug(
                     "Discarding bridge signal for empty message: {data}".format(data=data)

libervia/backend/plugins/plugin_xep_0045.py

@@ -1018,6 +1018,8 @@ class XEP_0045:
 
         except Exception as e:
             log.exception(f"MAM history sync failed for {room_jid}")
+        finally:
+            client.messageProt.flush_archive_batch()
 
     def pop_rooms(self, client: SatXMPPEntity) -> list[tuple]:
         """Remove rooms and return data needed to re-join them.