plugin XEP-0447: handle deferred sources:
XEP-0447 allow to send sources after the original message. Dino is doing that, and thus we can't display attachment immediately with only the original message. This patch handle that by updating the original message once the sources are received, and emitting a `message_update` signal to notify frontends.
This commit is contained in:
Goffi
parent
88ba99acd0
commit
dbc3973642
2 changed files with 121 additions and 16 deletions
|
|
@ -176,6 +176,7 @@ class Const:
|
|||
MESS_UPDATE_EDIT = "EDIT"
|
||||
MESS_UPDATE_REACTION = "REACTION"
|
||||
MESS_UPDATE_RETRACT = "RETRACT"
|
||||
MESS_UPDATE_SOURCES = "SOURCES"
|
||||
|
||||
MESS_EXTRA_INFO = "info_type"
|
||||
EXTRA_INFO_DECR_ERR = "DECRYPTION_ERROR"
|
||||
|
|
|
|||
|
|
@ -25,10 +25,12 @@ from typing import Any, ClassVar, NamedTuple, Self, cast
|
|||
|
||||
from pydantic import BaseModel, Field
|
||||
import shortuuid
|
||||
from sqlalchemy.orm.attributes import flag_modified
|
||||
from treq import collect
|
||||
from twisted.internet import defer
|
||||
from twisted.web.http_headers import Headers
|
||||
from twisted.web.iweb import IResponse
|
||||
from twisted.words.protocols.jabber import jid
|
||||
from twisted.words.xish import domish
|
||||
|
||||
from libervia.backend.core import exceptions
|
||||
|
|
@ -42,10 +44,11 @@ from libervia.backend.plugins.plugin_xep_0103 import URLData, XEP_0103
|
|||
from libervia.backend.plugins.plugin_xep_0334 import XEP_0334
|
||||
from libervia.backend.plugins.plugin_xep_0358 import JinglePub, XEP_0358
|
||||
from libervia.backend.plugins.plugin_xep_0363 import XEP_0363
|
||||
from libervia.backend.plugins.plugin_xep_0367 import XEP_0367
|
||||
from libervia.backend.plugins.plugin_xep_0367 import AttachedTo, XEP_0367
|
||||
from libervia.backend.plugins.plugin_xep_0428 import XEP_0428
|
||||
from libervia.backend.plugins.plugin_xep_0446 import FileMetadata, XEP_0446
|
||||
from libervia.backend.tools import stream
|
||||
from libervia.backend.tools.common import data_format
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
|
|
@ -483,7 +486,9 @@ class XEP_0447:
|
|||
if file_id is not None:
|
||||
data["id"] = file_id
|
||||
try:
|
||||
data["sources"] = self.parse_sources_elt(file_sharing_elt)
|
||||
data["sources"] = [
|
||||
source.model_dump() for source in self.parse_sources_elt(file_sharing_elt)
|
||||
]
|
||||
except exceptions.NotFound:
|
||||
# XEP-0447 allows <file-sharing/> without <sources/> (e.g. upload in progress)
|
||||
data["sources"] = []
|
||||
|
|
@ -630,7 +635,7 @@ class XEP_0447:
|
|||
|
||||
# <attach-to/>
|
||||
message_elt.addChild(
|
||||
self._ma.get_attach_to_elt(uid=main_msg_id, stanza_id=main_msg_id)
|
||||
AttachedTo(uid=main_msg_id, stanza_id=main_msg_id).to_element()
|
||||
)
|
||||
|
||||
# <sources/>
|
||||
|
|
@ -662,6 +667,11 @@ class XEP_0447:
|
|||
post_treat.addCallback(
|
||||
partial(self._add_file_sharing_attachments, client, message_elt)
|
||||
)
|
||||
post_treat.addCallback(
|
||||
lambda data: defer.ensureDeferred(
|
||||
self._handle_source_attachment(client, message_elt, data)
|
||||
)
|
||||
)
|
||||
return True
|
||||
|
||||
def _on_connection_closed(self, client: SatXMPPEntity, reason: Any) -> bool:
|
||||
|
|
@ -707,23 +717,117 @@ class XEP_0447:
|
|||
message_elt: domish.Element,
|
||||
data: dict[str, Any],
|
||||
) -> bool:
|
||||
"""Block backward-compatibility messages from continuing the workflow.
|
||||
"""Message parsing trigger.
|
||||
|
||||
If the message contains both <attach-to/> and <sources/>, it is a legacy
|
||||
compatibility message generated by our own multi-file sending. We raise
|
||||
CancelError to stop the message from being stored in history or sent to
|
||||
frontends.
|
||||
Ignore backward-compatibility messages emitted by our own device.
|
||||
"""
|
||||
ns_msg_att = "urn:xmpp:message-attaching:1"
|
||||
attach_to_elt = next(message_elt.elements(ns_msg_att, "attach-to"), None)
|
||||
attached_to = data["extra"].get("attached_to")
|
||||
sources_elt = next(message_elt.elements(NS_SFS, "sources"), None)
|
||||
|
||||
if attach_to_elt is not None and sources_elt is not None:
|
||||
log.debug(
|
||||
f"Ignoring backward-compatibility message for "
|
||||
f"attach-to id={attach_to_elt['id']!r}"
|
||||
)
|
||||
raise exceptions.CancelError(
|
||||
"Cancelled by XEP_0447 (backward-compatibility message)"
|
||||
)
|
||||
if attached_to is not None and sources_elt is not None:
|
||||
attach_to_id = attached_to["stanza_id"]
|
||||
# only block if this is our own backward-compatibility message (from our own
|
||||
# JID). Incoming source attachments from other clients should be processed
|
||||
# normally.
|
||||
if message_elt["from"] == client.jid.full():
|
||||
log.debug(
|
||||
f"Ignoring backward-compatibility message for "
|
||||
f"attach-to id={attach_to_id!r}"
|
||||
)
|
||||
raise exceptions.CancelError(
|
||||
"Cancelled by XEP_0447 (backward-compatibility message)"
|
||||
)
|
||||
return True
|
||||
|
||||
async def _handle_source_attachment(
|
||||
self, client: SatXMPPEntity, message_elt: domish.Element, data: dict[str, Any]
|
||||
) -> dict[str, Any]:
|
||||
"""Handle deferred source attachment.
|
||||
|
||||
When a message is attaching sources to an existing file-sharing message, find the
|
||||
original message, merge the new sources, update storage, and emit a signal.
|
||||
|
||||
@param client: Client instance.
|
||||
@param message_elt: The <message/> element.
|
||||
@param data: Parsed message data.
|
||||
@return: Updated message data.
|
||||
"""
|
||||
attached_to = data["extra"].get("attached_to")
|
||||
sources_elts = list(message_elt.elements(NS_SFS, "sources"))
|
||||
file_sharing_elts = list(message_elt.elements(NS_SFS, "file-sharing"))
|
||||
|
||||
# only process if we have attach-to + sources but no file-sharing
|
||||
if attached_to is None or not sources_elts or file_sharing_elts:
|
||||
return data
|
||||
|
||||
if attached_to.get("uid") is None:
|
||||
log.warning(
|
||||
f"Received source attachment with unresolved attach-to id "
|
||||
f"{attached_to['stanza_id']!r}, ignoring"
|
||||
)
|
||||
return data
|
||||
|
||||
# find the original message in history
|
||||
message_type = data.get("type")
|
||||
history = await self.host.memory.storage.get_history_from_xmpp_id(
|
||||
client, attached_to["stanza_id"], message_type
|
||||
)
|
||||
if history is None:
|
||||
log.warning(
|
||||
f"No history entry found for source attachment "
|
||||
f"attach-to={attached_to['stanza_id']!r}"
|
||||
)
|
||||
return data
|
||||
|
||||
# verify sender: sources can only be added by the original sender
|
||||
from_jid = jid.JID(message_elt["from"])
|
||||
if history.type == C.MESS_TYPE_GROUPCHAT:
|
||||
if history.source_jid != from_jid:
|
||||
log.warning(
|
||||
f"Ignoring source attachment from {from_jid.full()}: "
|
||||
f"not the original sender ({history.source_jid.full()})"
|
||||
)
|
||||
return data
|
||||
else:
|
||||
if history.source_jid.userhostJID() != from_jid.userhostJID():
|
||||
log.warning(
|
||||
f"Ignoring source attachment from {from_jid.full()}: "
|
||||
f"not the original sender ({history.source_jid.full()})"
|
||||
)
|
||||
return data
|
||||
|
||||
# merge new sources into existing attachments
|
||||
existing_attachments = history.extra.get(C.KEY_ATTACHMENTS, [])
|
||||
new_sources_added = False
|
||||
for sources_elt in sources_elts:
|
||||
sources = self.parse_sources_elt(sources_elt)
|
||||
file_id = sources_elt.getAttribute("id")
|
||||
# find matching attachment by file-sharing id
|
||||
for attachment in existing_attachments:
|
||||
if file_id is None or attachment.get("id") == file_id:
|
||||
for source in sources:
|
||||
source_data = source.model_dump()
|
||||
if source_data not in attachment["sources"]:
|
||||
attachment["sources"].append(source_data)
|
||||
new_sources_added = True
|
||||
break
|
||||
|
||||
if not new_sources_added:
|
||||
log.debug(
|
||||
f"No new sources to add for message uid={history.uid!r}, "
|
||||
f"skipping update"
|
||||
)
|
||||
else:
|
||||
flag_modified(history, "extra")
|
||||
await self.host.memory.storage.add(history)
|
||||
|
||||
# notify frontends
|
||||
update_data = MessageData(history.serialise())
|
||||
self.host.bridge.message_update(
|
||||
history.uid,
|
||||
C.MESS_UPDATE_SOURCES,
|
||||
data_format.serialise(update_data),
|
||||
client.profile,
|
||||
)
|
||||
# we stop the workflow here to avoid storage and bridge signal
|
||||
raise exceptions.CancelError("Source attachment merged, cancelling message")
|
||||
|
|
|
|||
Loading…
Reference in a new issue