From e7917b1b636d098f83e1bfaccb3f72a28315b9cf Mon Sep 17 00:00:00 2001 From: Goffi Date: Thu, 9 Apr 2026 13:34:09 +0200 Subject: [PATCH] component pubsub: Implement remaining methods: - `retract`, `purge` and `delete` are now implemented - `features` list has been added - `publish` now returns the Item ID when it is generated - `subscribers` publish model is handled. rel 476 --- .../plugins/plugin_comp_pubsub/__init__.py | 269 +++++++++++++++--- 1 file changed, 226 insertions(+), 43 deletions(-) diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py index 7413474f..22d469f2 100644 --- a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py +++ b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py @@ -144,6 +144,27 @@ class PubsubSettings(ComponentSettings, plugin_info=PLUGIN_INFO): class PubsubResource(pubsub.PubSubResource): + """PubSub resource implementation for Libervia Backend. + + This resource handles all pubsub operations including node creation, + item publishing, subscriptions, affiliations, and node management. + """ + + features: list[str] = [ + "config-node", + "create-nodes", + "delete-nodes", + "item-ids", + "meta-data", + "publish", + "purge-nodes", + "retract-items", + "retrieve-affiliations", + "retrieve-default", + "retrieve-items", + "retrieve-subscriptions", + "subscribe", + ] def __init__(self, service: "PubsubService") -> None: super().__init__() @@ -230,16 +251,20 @@ class PubsubResource(pubsub.PubSubResource): """ return PubsubNodeConfig.get_field_definitions() - def publish(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: + def publish( + self, request: pubsub.PubSubRequest + ) -> defer.Deferred[None | domish.Element]: return defer.ensureDeferred(self._publish(request)) - async def _publish(self, request: pubsub.PubSubRequest) -> None: + async def _publish(self, request: pubsub.PubSubRequest) -> None | domish.Element: """Publish items to a node. - @raise StanzaError: with "bad_request" if no items are provided - @raise StanzaError: with "item_not_found" if the node does not exist + @param request: The publish request. + @return: A pubsub element containing the published items. + @raise StanzaError: with "bad-request" if no items are provided + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor lacks publish privileges - @raise StanzaError: with "not_allowed" if item format is invalid + @raise StanzaError: with "not-allowed" if item format is invalid """ client = self.client assert request.sender is not None @@ -248,7 +273,7 @@ class PubsubResource(pubsub.PubSubResource): assert request.nodeIdentifier is not None if request.items is None or len(request.items) == 0: - raise error.StanzaError("bad_request", text="no item to publish") + raise error.StanzaError("bad-request", text="no item to publish") node = await G.storage.get_pubsub_node( client, @@ -259,7 +284,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") affiliation = next( ( @@ -277,20 +302,23 @@ class PubsubResource(pubsub.PubSubResource): raise error.StanzaError("forbidden", text="Insufficient privileges") elif publish_model == PublishModel.subscribers: if affiliation not in (Affiliation.owner, Affiliation.publisher): - # FIXME - raise NotImplementedError("FIXME: need to handle subscribers") - # is_subscribed = any( - # sub.subscriber == requestor_jid and sub.state is not None - # for sub in node.subscriptions - # ) - # if not is_subscribed: - # raise error.StanzaError("forbidden", text="Insufficient privileges") + is_subscribed = any( + sub.subscriber == requestor_jid + and sub.state == SubscriptionState.SUBSCRIBED + for sub in node.subscriptions + ) + if not is_subscribed: + raise error.StanzaError("forbidden", text="Insufficient privileges") + + ret_pubsub_elt = domish.Element((NS_PUBSUB_EVENT, "pubsub")) + ret_publish_elt = ret_pubsub_elt.addElement("publish") + ret_publish_elt["node"] = request.nodeIdentifier items_data = [] for item in request.items: if item.name != "item" or item.uri != pubsub.NS_PUBSUB: raise error.StanzaError( - "not_allowed", text=f"Invalid item: {item.toXml()}" + "not-allowed", text=f"Invalid item: {item.toXml()}" ) item["publisher"] = request.sender.full() @@ -298,12 +326,17 @@ class PubsubResource(pubsub.PubSubResource): if not item.getAttribute("id"): item["id"] = f"{shortuuid.uuid()}" + ret_item_elt = ret_publish_elt.addElement("item") + ret_item_elt["id"] = item["id"] + items_data.append(item) await G.storage.cache_pubsub_items(client, node, items_data) await self._send_notifications(node, "items", items_data) + return ret_pubsub_elt + def subscribe( self, request: pubsub.PubSubRequest ) -> defer.Deferred[pubsub.Subscription]: @@ -315,7 +348,7 @@ class PubsubResource(pubsub.PubSubResource): @param request: The subscribe request containing node identifier, subscriber JID, and optional subscription options. @return: A Subscription object representing the resulting subscription state. - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor lacks subscribe privileges @raise StanzaError: with "not-authorized" if there's a pending subscription """ @@ -337,7 +370,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") affiliation = next( ( @@ -407,7 +440,7 @@ class PubsubResource(pubsub.PubSubResource): """Handle an unsubscribe request. @param request: The unsubscribe request - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor is not the subscriber """ client = self.client @@ -427,7 +460,7 @@ class PubsubResource(pubsub.PubSubResource): node = await G.storage.get_pubsub_node(client, client.jid, request.nodeIdentifier) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") await G.storage.set_pubsub_subscriptions( node, @@ -601,7 +634,7 @@ class PubsubResource(pubsub.PubSubResource): @param request: The configuration request. @return: Current configuration values for the node. - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor is not an owner or admin """ client = self.client @@ -615,7 +648,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") # Only admins and owner can get configuration. if not G.host.memory.is_admin_jid(requestor_jid): @@ -660,9 +693,9 @@ class PubsubResource(pubsub.PubSubResource): async def _configureSet(self, request: pubsub.PubSubRequest) -> None: """Set the configuration of a node. - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor is not an owner or admin - @raise StanzaError: with "not_acceptable" if configuration values are invalid + @raise StanzaError: with "not-acceptable" if configuration values are invalid """ client = self.client assert request.sender is not None @@ -675,7 +708,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") # Only admins and owner can modifify configuration. if not G.host.memory.is_admin_jid(requestor_jid): @@ -742,18 +775,20 @@ class PubsubResource(pubsub.PubSubResource): def items( self, request: pubsub.PubSubRequest, - ) -> defer.Deferred[tuple[list[domish.Element], rsm.RSMResponse | None]]: + ) -> defer.Deferred[ + list[domish.Element] | tuple[list[domish.Element], rsm.RSMResponse] + ]: return defer.ensureDeferred(self._items(request)) async def _items( self, request: pubsub.PubSubRequest - ) -> tuple[list[domish.Element], rsm.RSMResponse | None]: + ) -> list[domish.Element] | tuple[list[domish.Element], rsm.RSMResponse]: """Retrieve items from a pubsub node. @param request: The items request containing node identifier, item identifiers, max items limit, and optional RSM parameters. @return: A tuple of (list of item elements, RSM response or None). - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor lacks retrieve privileges """ client = self.client @@ -767,7 +802,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") affiliation = next( ( @@ -821,19 +856,164 @@ class PubsubResource(pubsub.PubSubResource): return defer.ensureDeferred(self._retract(request)) async def _retract(self, request: pubsub.PubSubRequest) -> None: - return + """Handle items retraction request. + + @param request: The retract request + @raise StanzaError: with "item-not-found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor lacks retract privileges + @raise StanzaError: with "bad-request" if no item identifiers are provided + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + assert request.nodeIdentifier is not None + + if not request.itemIdentifiers: + raise error.StanzaError("bad-request", text="No item identifiers provided") + + node = await G.storage.get_pubsub_node( + client, + client.jid, + request.nodeIdentifier, + with_affiliations=True, + with_subscriptions=True, + ) + + if node is None: + raise error.StanzaError("item-not-found", text="Node does not exist") + + items_data, __ = await G.storage.get_items( + node=node, + max_items=None, + item_ids=request.itemIdentifiers, + ) + + existing_items = [ + item for item in items_data if item.name in request.itemIdentifiers + ] + + if not existing_items: + raise error.StanzaError( + "item-not-found", + text="No items found with the specified identifiers", + ) + + if not G.host.memory.is_admin_jid(requestor_jid): + affiliation = next( + ( + aff_data.affiliation + for aff_data in node.affiliations + if aff_data.entity == requestor_jid + ), + None, + ) + + if affiliation not in (Affiliation.owner, Affiliation.publisher): + for item in existing_items: + publisher_jid = item.data.getAttribute("publisher") + if ( + not publisher_jid + or jid.JID(publisher_jid).userhostJID() != requestor_jid + ): + raise error.StanzaError( + "forbidden", + text="Insufficient privileges to retract this item", + ) + + await G.storage.delete_pubsub_items(node, request.itemIdentifiers) + + if request.notify: + item_elts = [item.data for item in existing_items] + await self._send_notifications(node, "retract", item_elts) def purge(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._purge(request)) async def _purge(self, request: pubsub.PubSubRequest) -> None: - return + """Handle a node purge request. + + @param request: The purge request. + @raise StanzaError: with "item-not-found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor is not an owner + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + assert request.nodeIdentifier is not None + + node = await G.storage.get_pubsub_node( + client, + client.jid, + request.nodeIdentifier, + with_affiliations=True, + with_subscriptions=True, + ) + + if node is None: + raise error.StanzaError("item-not-found", text="Node does not exist") + + if not G.host.memory.is_admin_jid(requestor_jid): + affiliation = next( + ( + aff_data.affiliation + for aff_data in node.affiliations + if aff_data.entity == requestor_jid + ), + None, + ) + + if affiliation != Affiliation.owner: + raise error.StanzaError("forbidden", text="Insufficient privileges") + + await G.storage.delete_pubsub_items(node, items_names=None) + await self._send_notifications(node, "purge", []) def delete(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._delete(request)) async def _delete(self, request: pubsub.PubSubRequest) -> None: - return + """Handle a node deletion request. + + @param request: The delete request containing the node identifier. + @raise StanzaError: with "item-not-found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor is not an owner + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + assert request.nodeIdentifier is not None + + node = await G.storage.get_pubsub_node( + client, + client.jid, + request.nodeIdentifier, + with_affiliations=True, + with_subscriptions=True, + ) + + if node is None: + raise error.StanzaError("item-not-found", text="Node does not exist") + + if not G.host.memory.is_admin_jid(requestor_jid): + affiliation = next( + ( + aff_data.affiliation + for aff_data in node.affiliations + if aff_data.entity == requestor_jid + ), + None, + ) + + if affiliation != Affiliation.owner: + raise error.StanzaError("forbidden", text="Insufficient privileges") + + await self._send_notifications(node, "delete", []) + await G.storage.delete_pubsub_node( + profiles=None, services=[client.jid], names=[node.name] + ) def affiliationsGet( self, request: pubsub.PubSubRequest @@ -845,7 +1025,7 @@ class PubsubResource(pubsub.PubSubResource): @param request: The affiliations get request with node identifier. @return: Mapping of entity JIDs to their affiliations. - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor is not an owner or admin """ client = self.client @@ -859,7 +1039,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") if not G.host.memory.is_admin_jid(requestor_jid): affiliation = next( @@ -887,9 +1067,9 @@ class PubsubResource(pubsub.PubSubResource): @param request: The affiliations set request with node identifier and affiliations dict. - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor is not an owner or admin - @raise StanzaError: with "not_acceptable" if affiliation values are invalid + @raise StanzaError: with "not-acceptable" if affiliation values are invalid """ client = self.client assert request.sender is not None @@ -902,7 +1082,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") if not G.host.memory.is_admin_jid(requestor_jid): affiliation = next( @@ -960,7 +1140,7 @@ class PubsubResource(pubsub.PubSubResource): @param request: The subscriptions get request with node identifier. @return: List of Subscription objects for all subscribers of the node. - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor is not an owner or admin """ client = self.client @@ -978,7 +1158,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") if not G.host.memory.is_admin_jid(requestor_jid): affiliation = next( @@ -1020,9 +1200,9 @@ class PubsubResource(pubsub.PubSubResource): @param request: The subscriptions set request with node identifier and subscriptions dict. - @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "item-not-found" if the node does not exist @raise StanzaError: with "forbidden" if the requestor is not an owner or admin - @raise StanzaError: with "not_acceptable" if subscription values are invalid + @raise StanzaError: with "not-acceptable" if subscription values are invalid """ client = self.client assert request.sender is not None @@ -1035,7 +1215,7 @@ class PubsubResource(pubsub.PubSubResource): ) if node is None: - raise error.StanzaError("item_not_found", text="Node does not exist") + raise error.StanzaError("item-not-found", text="Node does not exist") if not G.host.memory.is_admin_jid(requestor_jid): affiliation = next( @@ -1180,7 +1360,10 @@ class PubsubResource(pubsub.PubSubResource): for item_elt in item_elts: retract_elt = event_type_elt.addElement("retract") retract_elt["id"] = item_elt.getAttribute("id") - case "delete" | "purge" | "configuration": + case "delete" | "purge": + # Empty elements, no children needed + pass + case "configuration": raise NotImplementedError case _: raise exceptions.InternalError(