component pubsub: implement notifications

rel 476

libervia/backend/plugins/plugin_comp_pubsub/__init__.py

@@ -40,11 +40,15 @@ from libervia.backend.memory.sqla_mapping import (
     Affiliation,
     PublishModel,
     PubsubAffiliation,
+    PubsubNode,
+    PubsubSub,
     SubscriptionState,
 )
 from libervia.backend.plugins.plugin_xep_0167 import NS_AV_CONFERENCES
 from libervia.backend.plugins.plugin_xep_0060 import NodeMetadata, XEP_0060
 from libervia.backend.tools.config import ComponentSettings
+from libervia.backend.tools.xml_tools import element_copy
+from tx_xmpp.pubsub import NS_PUBSUB_EVENT
 
 log = getLogger(__name__)
 
@@ -64,6 +68,8 @@ PLUGIN_INFO = {
     C.PI_DESCRIPTION: _("Libervia advanced Pubsub service."),
 }
 
+NotificationEventType = Literal["items", "retract", "delete", "purge", "configuration"]
+
 
 @dataclass
 class PubsubNodeConfig:
@@ -245,7 +251,11 @@ class PubsubResource(pubsub.PubSubResource):
             raise error.StanzaError("bad_request", text="no item to publish")
 
         node = await G.storage.get_pubsub_node(
-            client, client.jid, request.nodeIdentifier, with_affiliations=True
+            client,
+            client.jid,
+            request.nodeIdentifier,
+            with_affiliations=True,
+            with_subscriptions=True,
         )
 
         if node is None:
@@ -292,6 +302,8 @@ class PubsubResource(pubsub.PubSubResource):
 
         await G.storage.cache_pubsub_items(client, node, items_data)
 
+        await self._send_notifications(node, "items", items_data)
+
     def subscribe(
         self, request: pubsub.PubSubRequest
     ) -> defer.Deferred[pubsub.Subscription]:
@@ -931,7 +943,11 @@ class PubsubResource(pubsub.PubSubResource):
         assert request.nodeIdentifier is not None
 
         node = await G.storage.get_pubsub_node(
-            client, client.jid, request.nodeIdentifier, with_subscriptions=True, with_affiliations=True
+            client,
+            client.jid,
+            request.nodeIdentifier,
+            with_subscriptions=True,
+            with_affiliations=True,
         )
 
         if node is None:
@@ -1062,9 +1078,121 @@ class PubsubResource(pubsub.PubSubResource):
                         text=f"Invalid subscription state: {subscription_state!r}",
                     )
 
-        await G.storage.set_pubsub_subscriptions(
-            node, subscriptions_to_set
-        )
+        await G.storage.set_pubsub_subscriptions(node, subscriptions_to_set)
+
+    # Notifications
+
+    def _get_notification_recipients(
+        self, node: PubsubNode
+    ) -> dict[jid.JID, list[PubsubSub | None]]:
+        """Get all notification recipients for a node.
+
+        Filters the node subscriptions and affiliations to determine who should receive
+        notifications. Recipients include:
+        - All subscribers with SUBSCRIBED state
+        - All owners
+
+        @param node: The pubsub node (must have affiliations and subscriptions loaded).
+        @return: A mapping of recipient JIDs to their subscription list.
+            Recipients are grouped by JID to ensure only one notification per entity.
+            None is used for owner when there is no subscription.
+        """
+        recipients: dict[jid.JID, list[PubsubSub | None]] = {}
+
+        # Subscribers
+        if node.subscriptions:
+            for sub in node.subscriptions:
+                if sub.state == SubscriptionState.SUBSCRIBED:
+                    subscriber_jid = sub.subscriber
+                    subscriptions = recipients.setdefault(subscriber_jid, [])
+                    subscriptions.append(sub)
+
+        # Owners
+        if node.affiliations:
+            for aff in node.affiliations:
+                if aff.affiliation == Affiliation.owner:
+                    owner_jid = aff.entity
+                    subscriptions = recipients.setdefault(owner_jid, [])
+                    subscriptions.append(None)
+
+        return recipients
+
+    async def _send_notification(
+        self,
+        event_type: NotificationEventType,
+        node_id: str,
+        subscriber: jid.JID,
+        item_elts: list[domish.Element],
+    ) -> None:
+        """Send a single notification to a subscriber.
+
+        @param event_type: The type of event.
+        @param node_id: The node identifier.
+        @param subscriber: The subscriber JID.
+        @param item_elts: The items to include in the notification.
+        """
+        message_elt = domish.Element((None, "message"))
+        message_elt["from"] = self.client.jid.full()
+        message_elt["to"] = subscriber.full()
+        message_elt["id"] = f"{shortuuid.uuid()}"
+
+        event_elt = message_elt.addElement((NS_PUBSUB_EVENT, "event"))
+
+        # Create the event element (e.g., <items/>, <delete/>, etc.)
+        event_type_elt = event_elt.addElement(event_type)
+        event_type_elt["node"] = node_id
+
+        match event_type:
+            case "items":
+                for item_elt in item_elts:
+                    item_clone_elt = element_copy(item_elt)
+                    # We inherit from parents xmlns
+                    item_clone_elt.uri = item_clone_elt.defaultUri = None
+                    event_type_elt.addChild(item_clone_elt)
+            case "retract":
+                for item_elt in item_elts:
+                    retract_elt = event_type_elt.addElement("retract")
+                    retract_elt["id"] = item_elt.getAttribute("id")
+            case "delete" | "purge" | "configuration":
+                raise NotImplementedError
+            case _:
+                raise exceptions.InternalError(
+                    f"Unknown notification event type: {event_type!r}"
+                )
+
+        await self.client.a_send(message_elt)
+
+    async def _send_notifications(
+        self,
+        node: PubsubNode,
+        event_type: NotificationEventType,
+        item_elts: list[domish.Element],
+    ) -> None:
+        """Send notifications to all subscribers and owners of a node in parallel.
+
+        @param node: The pubsub node to send notifications for (must have affiliations and
+            subscriptions loaded).
+        @param event_type: The type of event ('items', 'delete', 'purge', etc.).
+        @param item_elts: The items to include in the notifications.
+        """
+        recipients = self._get_notification_recipients(node)
+
+        if not recipients:
+            return
+
+        deferreds = [
+            defer.ensureDeferred(
+                self._send_notification(event_type, node.name, recipient_jid, item_elts)
+            )
+            for recipient_jid, __ in recipients.items()
+        ]
+
+        # We send notifications in parallel
+        results = await defer.DeferredList(deferreds, consumeErrors=True)
+
+        for success, result in results:
+            if not success:
+                log.error(f"Failed to send notification: {result}")
 
 
 class PubsubService(rsm.PubSubService):
@@ -1079,7 +1207,7 @@ class PubsubService(rsm.PubSubService):
 
     @property
     def client(self) -> SatXMPPComponent:
-        return self.component._client
+        return self.component.client
 
 
 class PubsubComponent:
@@ -1090,6 +1218,7 @@ class PubsubComponent:
         self._clientt: SatXMPPComponent | None = None
         self.pubsub_service = PubsubService(self)
 
+    @property
     def client(self) -> SatXMPPComponent:
         assert self._client is not None
         return self._client