component pubsub: publish and items retrieval implementation:

rel 476

libervia/backend/plugins/plugin_comp_pubsub/__init__.py

@@ -46,6 +46,7 @@ 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.utils import ensure_deferred
+from libervia.backend.tools.xml_tools import element_copy
 
 log = getLogger(__name__)
 
@@ -229,7 +230,69 @@ class PubsubResource(pubsub.PubSubResource):
         return defer.ensureDeferred(self._publish(request))
 
     async def _publish(self, request: rsm.PubSubRequest) -> None:
-        return
+        """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
+        @raise StanzaError: with "forbidden" if the requestor lacks publish privileges
+        @raise StanzaError: with "not_allowed" if item format is invalid
+        """
+        client = self.client
+        assert request.sender is not None
+        requestor_jid = request.sender.userhostJID()
+
+        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")
+
+        node = await G.storage.get_pubsub_node(
+            client, client.jid, request.nodeIdentifier, with_affiliations=True
+        )
+
+        if node is None:
+            raise error.StanzaError("item_not_found", text="Node does not exist")
+
+        affiliation = next(
+            (
+                aff_data.affiliation
+                for aff_data in node.affiliations
+                if aff_data.entity == requestor_jid
+            ),
+            None,
+        )
+
+        publish_model = node.publish_model or PublishModel.publishers
+
+        if publish_model == PublishModel.publishers:
+            if affiliation not in (Affiliation.owner, Affiliation.publisher):
+                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")
+
+        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()}"
+                )
+
+            item["publisher"] = request.sender.full()
+
+            if not item.getAttribute("id"):
+                item["id"] = f"{shortuuid.uuid()}"
+
+            items_data.append(item)
+
+        await G.storage.cache_pubsub_items(client, node, items_data)
 
     def subscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
         return defer.ensureDeferred(self._subscribe(request))
@@ -470,13 +533,74 @@ class PubsubResource(pubsub.PubSubResource):
     async def _items(
         self, request: rsm.PubSubRequest
     ) -> tuple[list[domish.Element], rsm.RSMResponse | None]:
+        """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 "forbidden" if the requestor lacks retrieve privileges
+        """
         client = self.client
         assert request.sender is not None
         requestor_jid = request.sender.userhostJID()
-        if not client.is_local(requestor_jid):
-            raise error.StanzaError("forbidden")
 
-        return [], None
+        assert request.nodeIdentifier is not None
+
+        node = await G.storage.get_pubsub_node(
+            client, client.jid, request.nodeIdentifier, with_affiliations=True
+        )
+
+        if node is None:
+            raise error.StanzaError("item_not_found", text="Node does not exist")
+
+        affiliation = next(
+            (
+                aff_data.affiliation
+                for aff_data in node.affiliations
+                if aff_data.entity == requestor_jid
+            ),
+            None,
+        )
+
+        access_model = node.access_model or AccessModel.open
+
+        if access_model == AccessModel.open:
+            pass
+        elif access_model == AccessModel.whitelist:
+            if affiliation not in (
+                Affiliation.owner,
+                Affiliation.publisher,
+                Affiliation.member,
+            ):
+                raise error.StanzaError("forbidden", text="Insufficient privileges")
+        else:
+            log.error(
+                f"Unknown access model {access_model!r} for node"
+                f" {request.nodeIdentifier!r}"
+            )
+            raise error.StanzaError(
+                "internal-server-error",
+                text=f"Unknown access model {access_model!r}",
+            )
+
+        items_data, metadata = await G.storage.get_items(
+            node=node,
+            max_items=request.rsm.max if request.rsm else request.maxItems,
+            item_ids=request.itemIdentifiers or None,
+            before=request.rsm.before if request.rsm else None,
+            after=request.rsm.after if request.rsm else None,
+            from_index=request.rsm.index if request.rsm else None,
+            desc=request.rsm is None,
+            force_rsm=request.rsm is not None,
+        )
+
+        if "rsm" in metadata:
+            rsm_response = rsm.RSMResponse(**metadata["rsm"])
+        else:
+            rsm_response = None
+
+        return [item.data for item in items_data], rsm_response
 
     def retract(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
         return defer.ensureDeferred(self._retract(request))