From 766d49fb3856cafb9e230e856adb0430ce789f4a Mon Sep 17 00:00:00 2001 From: Goffi Date: Fri, 1 May 2026 20:31:49 +0200 Subject: [PATCH] plugin XEP-0313: Sync status tracking + resume workflow: - Add sync status tracking with bridge signals (`mam_sync_started`, `mam_sync_completed`, `mam_sync_error`), `mam_sync_statuses_get` method and helper methods that are called from `plugin_xep_0045`. - Update "resume" workflow: - fix settings of last stanza ID, so it's always set (was not set for our own messages) - own message now use `onMessage` full workflow instead of just store and signal. - add `flush_archive_batch` at the end of the process. - Sync status are sent during resumt - Modernisation (type hints, use of `get_plugin_data`) --- libervia/backend/plugins/plugin_xep_0045.py | 3 + libervia/backend/plugins/plugin_xep_0313.py | 571 +++++++++++++------- 2 files changed, 382 insertions(+), 192 deletions(-) diff --git a/libervia/backend/plugins/plugin_xep_0045.py b/libervia/backend/plugins/plugin_xep_0045.py index 482edc95..08b5d7bb 100644 --- a/libervia/backend/plugins/plugin_xep_0045.py +++ b/libervia/backend/plugins/plugin_xep_0045.py @@ -923,6 +923,7 @@ class XEP_0045: room_jid = room.roomJID total_messages = 0 + self._mam.sync_started(room_jid, client.profile) try: last_mess = await self.host.memory.history_get( room_jid, @@ -1015,9 +1016,11 @@ class XEP_0045: f"Retrieved {total_messages} message(s) from {room_jid} while" " offline.", ) + self._mam.sync_completed(room_jid, client.profile, total_messages) except Exception as e: log.exception(f"MAM history sync failed for {room_jid}") + self._mam.sync_error(room_jid, client.profile, str(e)) finally: client.messageProt.flush_archive_batch() diff --git a/libervia/backend/plugins/plugin_xep_0313.py b/libervia/backend/plugins/plugin_xep_0313.py index ee8c2ed7..a0c7fc49 100644 --- a/libervia/backend/plugins/plugin_xep_0313.py +++ b/libervia/backend/plugins/plugin_xep_0313.py @@ -1,7 +1,6 @@ #!/usr/bin/env python3 - -# SAT plugin for Message Archive Management (XEP-0313) +# Libervia plugin for Message Archive Management (XEP-0313) # Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) # Copyright (C) 2013-2016 Adrien Cossa (souliane@mailoo.org) @@ -18,6 +17,7 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +from collections.abc import Callable from datetime import datetime import uuid @@ -25,24 +25,33 @@ from dateutil import tz from twisted.internet import defer from twisted.words.protocols.jabber import jid from twisted.words.protocols.jabber.error import StanzaError +from libervia.backend import G from tx_xmpp import disco from tx_xmpp import data_form from tx_xmpp import rsm from tx_xmpp import mam +from twisted.words.xish import domish from zope.interface import implementer from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C +from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.tools import xml_tools from libervia.backend.tools.common import data_format +from typing import Any, Literal, TYPE_CHECKING + +if TYPE_CHECKING: + from libervia.backend.core.main import LiberviaBackend log = getLogger(__name__) +IMPORT_NAME = "XEP-0313" + PLUGIN_INFO = { C.PI_NAME: "Message Archive Management", - C.PI_IMPORT_NAME: "XEP-0313", + C.PI_IMPORT_NAME: IMPORT_NAME, C.PI_TYPE: "XEP", # XEP-0431 only defines a namespace, so we register it here C.PI_PROTOCOLS: ["XEP-0313", "XEP-0431"], @@ -60,8 +69,8 @@ MESSAGE_STANZA_ID = '/message/stanza-id[@xmlns="{ns_stanza_id}"]' NS_FTS = "urn:xmpp:fulltext:0" -class XEP_0313(object): - def __init__(self, host): +class XEP_0313: + def __init__(self, host: "LiberviaBackend") -> None: log.info(_("Message Archive Management plugin initialization")) self.host = host self.host.register_namespace("mam", mam.NS_MAM) @@ -71,6 +80,7 @@ class XEP_0313(object): # Deferred used to store last stanza id in order of reception self._last_stanza_id_d = defer.Deferred() self._last_stanza_id_d.callback(None) + self._sync_status: dict[jid.JID, bool] = {} host.bridge.add_method( "mam_get", ".plugin", @@ -79,137 +89,195 @@ class XEP_0313(object): method=self._get_archives, async_=True, ) - - async def resume(self, client): - """Retrieve one2one messages received since the last we have in local storage""" - stanza_id_data = await self.host.memory.storage.get_privates( - mam.NS_MAM, [KEY_LAST_STANZA_ID], profile=client.profile + host.bridge.add_method( + "mam_sync_statuses_get", + ".plugin", + in_sign="s", + out_sign="a{sb}", + method=self._sync_statuses_get, ) - stanza_id = stanza_id_data.get(KEY_LAST_STANZA_ID) - rsm_req = None - if stanza_id is None: - log.info("can't retrieve last stanza ID, checking history") - last_mess = await self.host.memory.history_get( - None, - None, - limit=1, - filters={"not_types": C.MESS_TYPE_GROUPCHAT, "last_stanza_id": True}, - profile=client.profile, + # args: entity_jid, profile + host.bridge.add_signal("mam_sync_started", ".plugin", signature="ss") + # args: entity_jid, profile, message_count + host.bridge.add_signal("mam_sync_completed", ".plugin", signature="ssi") + # args: entity_jid, profile, error_message + host.bridge.add_signal("mam_sync_error", ".plugin", signature="sss") + + def _sync_statuses_get(self, profile: str) -> dict[str, bool]: + """Get MAM sync status for all entities. + + @param profile: The profile name. + @return: Mapping of entity_jid to syncing status. + """ + return {str(jid): status for jid, status in self._sync_status.items()} + + def sync_started(self, entity_jid: jid.JID, profile: str) -> None: + """Set sync status and emit signal. + + @param entity_jid: The entity JID. + @param profile: The profile name. + """ + self._sync_status[entity_jid] = True + self.host.bridge.mam_sync_started(str(entity_jid), profile) + + def sync_completed(self, entity_jid: jid.JID, profile: str, count: int) -> None: + """Clear sync status and emit signal. + + @param entity_jid: The entity JID (jid.JID or str). + @param profile: The profile name. + @param count: Number of messages synced. + """ + self._sync_status.pop(entity_jid, None) + self.host.bridge.mam_sync_completed(str(entity_jid), profile, count) + + def sync_error(self, entity_jid: jid.JID, profile: str, error: str) -> None: + """Clear sync status and emit signal. + + @param entity_jid: The entity JID (jid.JID or str). + @param profile: The profile name. + @param error: Error message. + """ + self._sync_status.pop(entity_jid, None) + self.host.bridge.mam_sync_error(str(entity_jid), profile, error) + + async def resume(self, client: SatXMPPEntity) -> None: + """Retrieve one2one messages received since the last we have in local storage""" + entity_jid = client.jid.userhostJID() + self.sync_started(entity_jid, client.profile) + try: + stanza_id_data = await self.host.memory.storage.get_privates( + mam.NS_MAM, [KEY_LAST_STANZA_ID], profile=client.profile ) - if not last_mess: - log.info(_("It seems that we have no MAM history yet")) - stanza_id = None - rsm_req = rsm.RSMRequest(max_=50, before="") - else: - stanza_id = last_mess[0][-1]["stanza_id"] - if rsm_req is None: - rsm_req = rsm.RSMRequest(max_=100, after=stanza_id) - mam_req = mam.MAMRequest(rsm_=rsm_req) - complete = False - count = 0 - while not complete: - try: - mam_data = await self.get_archives( - client, mam_req, service=client.jid.userhostJID() + stanza_id = stanza_id_data.get(KEY_LAST_STANZA_ID) + rsm_req = None + if stanza_id is None: + log.info("can't retrieve last stanza ID, checking history") + last_mess = await self.host.memory.history_get( + None, + None, + limit=1, + filters={"not_types": C.MESS_TYPE_GROUPCHAT, "last_stanza_id": True}, + profile=client.profile, ) - except StanzaError as e: - log.warning( - f"Can't retrieve MAM archives: {e}\n" - f"{xml_tools.pp_elt(mam_req.toElement())}\n" - f"{xml_tools.pp_elt(e.stanza)}" - ) - return - except Exception as e: - log.exception(f"Can't retrieve retrieve MAM archive") - return - elt_list, rsm_response, mam_response = mam_data - complete = mam_response["complete"] - # we update MAM request for next iteration - mam_req.rsm.after = rsm_response.last - # before may be set if we had no previous history - mam_req.rsm.before = None - if not elt_list: - break - else: - count += len(elt_list) - - for idx, mess_elt in enumerate(elt_list): - try: - fwd_message_elt = self.get_message_from_result( - client, mess_elt, mam_req - ) - except exceptions.DataError: - continue - - try: - destinee = jid.JID(fwd_message_elt["to"]) - except KeyError: - log.warning(_('missing "to" attribute in forwarded message')) - destinee = client.jid - if destinee.userhostJID() == client.jid.userhostJID(): - # message to use, we insert the forwarded message in the normal - # workflow - client.xmlstream.dispatch(fwd_message_elt) + if not last_mess: + log.info(_("It seems that we have no MAM history yet")) + stanza_id = None + rsm_req = rsm.RSMRequest(max_=50, before="") else: - # this message should be from us, we just add it to history + stanza_id = last_mess[0][-1]["stanza_id"] + if rsm_req is None: + rsm_req = rsm.RSMRequest(max_=100, after=stanza_id) + mam_req = mam.MAMRequest(rsm_=rsm_req) + assert mam_req.rsm is not None + complete = False + count = 0 + while not complete: + try: + mam_data = await self.get_archives( + client, mam_req, service=client.jid.userhostJID() + ) + except StanzaError as e: + log.warning( + f"Can't retrieve MAM archives: {e}\n" + f"{xml_tools.pp_elt(mam_req.toElement())}\n" + f"{xml_tools.pp_elt(e.iq.toElement()) if e.iq else ''}" + ) + return + except Exception as e: + log.exception(f"Can't retrieve retrieve MAM archive") + return + elt_list, rsm_response, mam_response = mam_data + assert rsm_response is not None + complete = mam_response["complete"] + # we update MAM request for next iteration + mam_req.rsm.after = rsm_response.last + # before may be set if we had no previous history + mam_req.rsm.before = None + if not elt_list: + break + else: + count += len(elt_list) + + for idx, mess_elt in enumerate(elt_list): try: - from_jid = jid.JID(fwd_message_elt["from"]) - except KeyError: - log.warning(_('missing "from" attribute in forwarded message')) - from_jid = client.jid - if from_jid.userhostJID() != client.jid.userhostJID(): - log.warning( - _( - "was expecting a message sent by our jid, but this one if " - "from {from_jid}, ignoring\n{xml}" - ).format(from_jid=from_jid.full(), xml=mess_elt.toXml()) + fwd_message_elt = self.get_message_from_result( + client, mess_elt, mam_req ) + except exceptions.DataError: continue - # adding message to history - mess_data = client.messageProt.parse_message(fwd_message_elt) + try: - await client.messageProt.add_to_history(mess_data) - except exceptions.CancelError as e: - log.warning( - "message has not been added to history: {e}".format(e=e) - ) - except Exception as e: - log.error( - "can't add message to history: {e}\n{xml}".format( - e=e, xml=mess_elt.toXml() - ) - ) - if complete and idx == len(elt_list) - 1: - # We are at the last message from archive, we store the ID to not - # ask again the same messages next time. + destinee = jid.JID(fwd_message_elt["to"]) + except KeyError: + log.warning(_('missing "to" attribute in forwarded message')) + destinee = client.jid + if destinee.userhostJID() == client.jid.userhostJID(): + # message to use, we insert the forwarded message in the normal + # workflow + client.xmlstream.dispatch(fwd_message_elt) + else: + # this message should be from us, we just add it to history try: - stanza_id = mess_elt.result["id"] - await self.host.memory.storage.set_private_value( - namespace=mam.NS_MAM, - key=KEY_LAST_STANZA_ID, - value=stanza_id, - profile=client.profile, + from_jid = jid.JID(fwd_message_elt["from"]) + except KeyError: + log.warning( + _('missing "from" attribute in forwarded message') ) - except Exception: - log.exception("Can't store last stanza ID") + from_jid = client.jid + if from_jid.userhostJID() != client.jid.userhostJID(): + log.warning( + _( + "was expecting a message sent by our jid, but this " + "one if from {from_jid}, ignoring\n{xml}" + ).format(from_jid=from_jid.full(), xml=mess_elt.toXml()) + ) + continue + # use normal message flow for consistency + client.messageProt.onMessage(fwd_message_elt) - if not count: - log.info(_("We have received no message while offline")) - else: - log.info( - _("We have received {num_mess} message(s) while offline.").format( - num_mess=count + # We are at the last page, we store the ID to not ask again + # the same messages next time. + if complete and elt_list: + try: + last_result_elt = elt_list[-1].result + assert last_result_elt is not None + stanza_id = last_result_elt["id"] + await self.host.memory.storage.set_private_value( + namespace=mam.NS_MAM, + key=KEY_LAST_STANZA_ID, + value=stanza_id, + profile=client.profile, + ) + except Exception: + log.exception("Can't store last stanza ID") + + if not count: + log.info(_("We have received no message while offline")) + else: + log.info( + _("We have received {num_mess} message(s) while offline.").format( + num_mess=count + ) ) - ) + self.sync_completed(entity_jid, client.profile, count) - def profile_connected(self, client): + except Exception as e: + log.exception(f"MAM history sync failed for {entity_jid}") + self.sync_error(entity_jid, client.profile, str(e)) + finally: + client.messageProt.flush_archive_batch() + + def profile_connected(self, client: SatXMPPEntity) -> None: defer.ensureDeferred(self.resume(client)) - def get_handler(self, client): - mam_client = client._mam = LiberviaMAMClient(self) + def get_handler(self, client: SatXMPPEntity) -> "LiberviaMAMClient": + mam_client = LiberviaMAMClient(self) + client.set_plugin_data(IMPORT_NAME, mam_client) return mam_client - def parse_extra(self, extra, with_rsm=True): + def parse_extra( + self, extra: dict[str, Any], with_rsm: bool = True + ) -> mam.MAMRequest | None: """Parse extra dictionnary to retrieve MAM arguments @param extra(dict): data for parse @@ -269,16 +337,21 @@ class XEP_0313(object): return mam.MAMRequest(**mam_args) if mam_args else None - def get_message_from_result(self, client, mess_elt, mam_req, service=None): + def get_message_from_result( + self, + client: SatXMPPEntity, + mess_elt: domish.Element, + mam_req: mam.MAMRequest, + service: jid.JID | None = None, + ) -> domish.Element: """Extract usable from MAM query result The message will be validated, and stanza-id/delay will be added if necessary. - @param mess_elt(domish.Element): result element wrapping the message - to retrieve - @param mam_req(mam.MAMRequest): request used (needed to get query_id) - @param service(jid.JID, None): MAM service where the request has been sent + @param mess_elt: result element wrapping the message to retrieve + @param mam_req: request used (needed to get query_id) + @param service: MAM service where the request has been sent None if it's user server - @return domish.Element): that can be used directly with onMessage + @return: that can be used directly with onMessage """ if mess_elt.name != "message": log.warning( @@ -315,9 +388,8 @@ class XEP_0313(object): else: if not result_elt["queryid"] == mam_req.query_id: log.error( - "Unexpected query id (was expecting {query_id}): {xml}".format( - query_id=mam.query_id, xml=mess_elt.toXml() - ) + f"Unexpected query id (was expecting {mam_req.query_id}): " + f"{mess_elt.toXml()}" ) raise exceptions.DataError("Invalid element") stanza_id = self._sid.get_stanza_id(fwd_message_elt, service_jid) @@ -345,31 +417,69 @@ class XEP_0313(object): return fwd_message_elt - def queryFields(self, client, service=None): + def queryFields( + self, + client: SatXMPPEntity, + service: jid.JID | None = None, + ) -> defer.Deferred[data_form.Form | None]: """Ask the server about supported fields. + @param client: The client session. @param service: entity offering the MAM service (None for user archives) - @return (D(data_form.Form)): form with the implemented fields (cf XEP-0313 §4.1.5) + @return: data Form with the implemented fields (cf XEP-0313 §4.1.5), or None """ - return client._mam.queryFields(service) + mam_client = client.get_plugin_data(IMPORT_NAME, LiberviaMAMClient) + return mam_client.queryFields(service) - def queryArchive(self, client, mam_req, service=None): + def queryArchive( + self, + client: SatXMPPEntity, + mam_req: mam.MAMRequest, + service: jid.JID | None = None, + ) -> defer.Deferred[domish.Element]: """Query a user, MUC or pubsub archive. - @param mam_req(mam.MAMRequest): MAM query instance - @param service(jid.JID, None): entity offering the MAM service + @param client: The client session. + @param mam_req: MAM query instance + @param service: entity offering the MAM service None for user server - @return (D(domish.Element)): result + @return: result """ - return client._mam.queryArchive(mam_req, service) + mam_client = client.get_plugin_data(IMPORT_NAME, LiberviaMAMClient) + return mam_client.queryArchive(mam_req, service) - def _append_message(self, elt_list, message_cb, message_elt): + def _append_message( + self, + elt_list: list[domish.Element], + message_cb: Callable[[domish.Element], domish.Element] | None, + message_elt: domish.Element, + ) -> None: + """Append a message element to the list, optionally transforming it via callback. + + @param elt_list: List to append elements to. + @param message_cb: Optional callback to transform the element before appending. + @param message_elt: The message element to append. + """ if message_cb is not None: elt_list.append(message_cb(message_elt)) else: elt_list.append(message_elt) - def _query_finished(self, iq_result, client, elt_list, event): + def _query_finished( + self, + iq_result: domish.Element, + client: SatXMPPEntity, + elt_list: list[domish.Element], + event: str, + ) -> tuple[list[domish.Element], rsm.RSMResponse | None, dict[str, bool]]: + """Handle the completion of a MAM query. + + @param iq_result: The IQ result element. + @param client: The client session. + @param elt_list: List of collected message elements. + @param event: The event xpath string. + @return: Tuple of element list, RSM response, and MAM response dict. + """ client.xmlstream.removeObserver(event, self._append_message) try: fin_elt = next(iq_result.elements(mam.NS_MAM, "fin")) @@ -388,20 +498,43 @@ class XEP_0313(object): return (elt_list, rsm_response, mam_response) - def serialize_archive_result(self, data, client, mam_req, service): + async def serialize_archive_result( + self, + client: SatXMPPEntity, + data: tuple[list[domish.Element], rsm.RSMResponse | None, dict[str, bool]], + mam_req: mam.MAMRequest, + service: jid.JID | None, + ) -> tuple[list[dict[str, Any]], str, str]: + """Serialize MAM archive result for bridge communication. + + @param client: The client session. + @param data: Tuple of element list, RSM response, and MAM response. + @param mam_req: The MAM request used. + @param service: The MAM service JID. + @return: Tuple of message list, serialized metadata, and profile key. + """ elt_list, rsm_response, mam_response = data - mess_list = [] + mess_list: list[dict[str, Any]] = [] for elt in elt_list: fwd_message_elt = self.get_message_from_result( client, elt, mam_req, service=service ) - mess_data = client.messageProt.parse_message(fwd_message_elt) + mess_data = await client.messageProt.parse_message(fwd_message_elt) mess_list.append(client.message_get_bridge_args(mess_data)) - metadata = {"rsm": self._rsm.response2dict(rsm_response), "mam": mam_response} + metadata: dict[str, Any] = { + "rsm": self._rsm.response2dict(rsm_response), + "mam": mam_response, + } return mess_list, data_format.serialise(metadata), client.profile - def _get_archives(self, service, extra_ser, profile_key): - """ + def _get_archives( + self, service: str, extra_ser: str, profile_key: str + ) -> defer.Deferred[tuple[list[dict[str, Any]], str, str]]: + """Retrieve MAM archives via bridge. + + @param service: Service JID string or empty string for user server. + @param extra_ser: Serialized extra parameters dict. + @param profile_key: Profile key string. @return: tuple with: - list of message with same data as in bridge.message_new - response metadata with: @@ -410,23 +543,37 @@ class XEP_0313(object): - profile """ client = self.host.get_client(profile_key) - service = jid.JID(service) if service else None - extra = data_format.deserialise(extra_ser, {}) + service_jid: jid.JID | None = jid.JID(service) if service else None + extra: dict[str, Any] = data_format.deserialise(extra_ser, {}) mam_req = self.parse_extra(extra) + assert mam_req is not None - d = self.get_archives(client, mam_req, service=service) - d.addCallback(self.serialize_archive_result, client, mam_req, service) + d = self.get_archives(client, mam_req, service=service_jid) + d.addCallback( + lambda data: defer.ensureDeferred( + self.serialize_archive_result(client, data, mam_req, service_jid) + ) + ) return d - def get_archives(self, client, query, service=None, message_cb=None): - """Query archive and gather page result + def get_archives( + self, + client: SatXMPPEntity, + query: mam.MAMRequest, + service: jid.JID | None = None, + message_cb: Callable[[domish.Element], domish.Element] | None = None, + ) -> defer.Deferred[ + tuple[list[domish.Element], rsm.RSMResponse | None, dict[str, bool]] + ]: + """Query archive and gather page result. - @param query(mam.MAMRequest): MAM request - @param service(jid.JID, None): MAM service to use + @param client: The client session. + @param query: MAM request + @param service: MAM service to use None to use our own server - @param message_cb(callable, None): callback to use on each message + @param message_cb: callback to use on each message this method can be used to unwrap messages - @return (tuple[list[domish.Element], rsm.RSMResponse, dict): result data with: + @return: Deferred firing with result data: - list of found elements - RSM response - MAM response, which is a dict with following value: @@ -435,55 +582,82 @@ class XEP_0313(object): """ if query.query_id is None: query.query_id = str(uuid.uuid4()) - elt_list = [] + elt_list: list[domish.Element] = [] event = MESSAGE_RESULT.format(mam_ns=mam.NS_MAM, query_id=query.query_id) client.xmlstream.addObserver(event, self._append_message, 0, elt_list, message_cb) d = self.queryArchive(client, query, service) d.addCallback(self._query_finished, client, elt_list, event) return d - def get_prefs(self, client, service=None): + def get_prefs( + self, + client: SatXMPPEntity, + service: jid.JID | None = None, + ) -> defer.Deferred[domish.Element]: """Retrieve the current user preferences. + @param client: The client session. @param service: entity offering the MAM service (None for user archives) - @return: the server response as a Deferred domish.Element + @return: server response """ # http://xmpp.org/extensions/xep-0313.html#prefs - return client._mam.queryPrefs(service) + mam_client = client.get_plugin_data(IMPORT_NAME, LiberviaMAMClient) + return mam_client.queryPrefs(service) def _set_prefs( self, - service_s=None, - default="roster", - always=None, - never=None, - profile_key=C.PROF_KEY_NONE, - ): - service = jid.JID(service_s) if service_s else None - always_jid = [jid.JID(entity) for entity in always] - never_jid = [jid.JID(entity) for entity in never] + service_s: str | None = None, + default: Literal["always", "never", "roster"] = "roster", + always: list[str] | None = None, + never: list[str] | None = None, + profile_key: str = C.PROF_KEY_NONE, + ) -> defer.Deferred[domish.Element]: + """Set new user preferences via bridge. + + @param service_s: Service JID string or None for user server. + @param default: A value in ('always', 'never', 'roster'). + @param always: List of JID strings to always archive. + @param never: List of JID strings to never archive. + @param profile_key: Profile key string. + @return: server response + """ + client = G.host.get_client(profile_key) + service: jid.JID | None = jid.JID(service_s) if service_s else None + always_jid: list[jid.JID] = [jid.JID(entity) for entity in (always or [])] + never_jid: list[jid.JID] = [jid.JID(entity) for entity in (never or [])] # TODO: why not build here a MAMPrefs object instead of passing the args separately? - return self.setPrefs(service, default, always_jid, never_jid, profile_key) + return self.setPrefs(client, service, default, always_jid, never_jid) - def setPrefs(self, client, service=None, default="roster", always=None, never=None): - """Set news user preferences. + def setPrefs( + self, + client: SatXMPPEntity, + service: jid.JID | None = None, + default: Literal["always", "never", "roster"] = "roster", + always: list[jid.JID] | None = None, + never: list[jid.JID] | None = None, + ) -> defer.Deferred[domish.Element]: + """Set new user preferences. + @param client: The client session. @param service: entity offering the MAM service (None for user archives) - @param default (unicode): a value in ('always', 'never', 'roster') - @param always (list): a list of JID instances - @param never (list): a list of JID instances - @param profile_key (unicode): %(doc_profile_key)s - @return: the server response as a Deferred domish.Element + @param default: a value in ('always', 'never', 'roster') + @param always: a list of JID instances + @param never: a list of JID instances + @return: server response """ # http://xmpp.org/extensions/xep-0313.html#prefs - return client._mam.setPrefs(service, default, always, never) + mam_client = client.get_plugin_data(IMPORT_NAME, LiberviaMAMClient) + return mam_client.setPrefs(service, default, always, never) - def on_message_stanza_id(self, message_elt, client): - """Called when a message with a stanza-id is received + def on_message_stanza_id( + self, message_elt: domish.Element, client: SatXMPPEntity + ) -> None: + """Called when a message with a stanza-id is received. - the messages' stanza ids are stored when received, so the last one can be used - to retrieve missing history on next connection - @param message_elt(domish.Element): with a stanza-id + The messages' stanza ids are stored when received, so the last one can be used + to retrieve missing history on next connection. + @param message_elt: with a stanza-id + @param client: The client session. """ if message_elt.getAttribute("type") == C.MESS_TYPE_GROUPCHAT: # groupchat message MAM is handled by XEP-0045 @@ -508,23 +682,36 @@ class XEP_0313(object): @implementer(disco.IDisco) class LiberviaMAMClient(mam.MAMClient): - def __init__(self, plugin_parent): + def __init__(self, plugin_parent: XEP_0313) -> None: self.plugin_parent = plugin_parent - @property - def host(self): - return self.parent.host_app - - def connectionInitialized(self): - observer_xpath = MESSAGE_STANZA_ID.format( - ns_stanza_id=self.host.ns_map["stanza_id"] - ) + def connectionInitialized(self) -> None: + """Set up observers when the XMPP connection is initialized.""" + observer_xpath = MESSAGE_STANZA_ID.format(ns_stanza_id=G.host.ns_map["stanza_id"]) self.xmlstream.addObserver( observer_xpath, self.plugin_parent.on_message_stanza_id, client=self.parent ) - def getDiscoInfo(self, requestor, target, nodeIdentifier=""): + def getDiscoInfo( + self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" + ) -> list[disco.DiscoFeature]: + """Get disco info for MAM + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: optional node identifier + @return: list of disco features + """ return [disco.DiscoFeature(mam.NS_MAM)] - def getDiscoItems(self, requestor, target, nodeIdentifier=""): + def getDiscoItems( + self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" + ) -> list[disco.DiscoItem]: + """Get disco items for MAM + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: optional node identifier + @return: empty list (MAM doesn't have items) + """ return []