component AP gateway: convert Delete AP activities to corresponding Pubsub retract:
when a `Delete` actity is received, the item is removed from cache, and the corresponding item `retract` is sent to local XMPP subscribers. rel 367
This commit is contained in:
parent
4e0c1fe7bf
commit
408dc04ee6
3 changed files with 65 additions and 2 deletions
|
|
@ -1792,3 +1792,43 @@ class APGateway:
|
|||
[(subscription.subscriber, None, [item_elt])]
|
||||
)
|
||||
|
||||
async def newAPDeleteItem(
|
||||
self,
|
||||
client: SatXMPPEntity,
|
||||
destinee: Optional[jid.JID],
|
||||
node: str,
|
||||
item: dict,
|
||||
) -> None:
|
||||
"""Analyse, cache and send notification for received AP item
|
||||
|
||||
@param destinee: jid of the destinee,
|
||||
@param node: XMPP pubsub node
|
||||
@param item: AP object payload
|
||||
"""
|
||||
item_id = item.get("id")
|
||||
if not item_id:
|
||||
raise exceptions.DataError('"id" attribute is missing in item')
|
||||
if self.isLocalURL(item_id):
|
||||
raise ValueError("Local IDs should not be used")
|
||||
|
||||
cached_node = await self.host.memory.storage.getPubsubNode(
|
||||
client, client.jid, node, with_subscriptions=True
|
||||
)
|
||||
if cached_node is None:
|
||||
log.warning(
|
||||
f"Received an item retract for node {node!r} at {client.jid} which is "
|
||||
"not cached"
|
||||
)
|
||||
else:
|
||||
await self.host.memory.storage.deletePubsubItems(cached_node, [item_id])
|
||||
# notifyRetract is expecting domish.Element instances
|
||||
item_elt = domish.Element((None, "item"))
|
||||
item_elt["id"] = item_id
|
||||
for subscription in cached_node.subscriptions:
|
||||
if subscription.state != SubscriptionState.SUBSCRIBED:
|
||||
continue
|
||||
self.pubsub_service.notifyRetract(
|
||||
client.jid,
|
||||
node,
|
||||
[(subscription.subscriber, None, [item_elt])]
|
||||
)
|
||||
|
|
|
|||
|
|
@ -69,7 +69,8 @@ ACTIVITY_OBJECT_MANDATORY = (
|
|||
"Create", "Update", "Delete", "Follow", "Add", "Remove", "Like", "Block", "Undo"
|
||||
)
|
||||
ACTIVITY_TARGET_MANDATORY = ("Add", "Remove")
|
||||
# activities which can be used with Shared Inbox (i.e. with not account specified)
|
||||
ACTIVIY_NO_ACCOUNT_ALLOWED = ("create",)
|
||||
# activities which can be used with Shared Inbox (i.e. with no account specified)
|
||||
# must be lowercase
|
||||
ACTIVIY_NO_ACCOUNT_ALLOWED = ("create", "delete")
|
||||
# maximum number of parents to retrieve when comments_max_depth option is set
|
||||
COMMENTS_MAX_PARENTS = 100
|
||||
|
|
|
|||
|
|
@ -232,6 +232,28 @@ class HTTPAPGServer(web_resource.Resource):
|
|||
else:
|
||||
log.warning(f"Unmanaged accept type: {type_!r}")
|
||||
|
||||
async def handleDeleteActivity(
|
||||
self,
|
||||
request: "HTTPRequest",
|
||||
data: dict,
|
||||
account_jid: Optional[jid.JID],
|
||||
node: Optional[str],
|
||||
ap_account: Optional[str],
|
||||
ap_url: str,
|
||||
signing_actor: str
|
||||
):
|
||||
digest = request.getHeader("digest")
|
||||
if digest in self._seen_digest:
|
||||
log.debug(f"Ignoring duplicated request (digest: {digest!r})")
|
||||
return
|
||||
self._seen_digest.append(digest)
|
||||
if node is None:
|
||||
node = self.apg._m.namespace
|
||||
client = await self.apg.getVirtualClient(signing_actor)
|
||||
objects = await self.apg.apGetList(data, "object")
|
||||
for obj in objects:
|
||||
await self.apg.newAPDeleteItem(client, account_jid, node, obj)
|
||||
|
||||
async def handleCreateActivity(
|
||||
self,
|
||||
request: "HTTPRequest",
|
||||
|
|
|
|||
Loading…
Reference in a new issue