diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py index d3dee5bd..109a41dd 100644 --- a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py +++ b/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., , , 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