From e241075a34ea6e848da486063667b31b2936eed1 Mon Sep 17 00:00:00 2001 From: Goffi Date: Sat, 20 Jun 2026 15:11:06 +0200 Subject: [PATCH] core: Unseen message counter: This patches add handling for unseen messages counter, and last message received. The counters are retrieved from DB on startup, they updated in-memory. Frontend are expected to have their own in-memory counter too, the backend one is used for re-sync (e.g.; on page reload). Bridge methods are added to retrieve the state, and indicate which message has been seen last by user. A signal is sent on counter reset to all frontends can synchronise. This is currently local only, but it should be compatible with a future XEP-0490 (Message Displayed Synchronization) implementation. --- .../bridge_constructor/bridge_template.ini | 29 ++++ libervia/backend/bridge/dbus_bridge.py | 14 ++ libervia/backend/bridge/embedded.py | 42 ++++++ libervia/backend/bridge/pb.py | 3 + libervia/backend/core/constants.py | 1 + libervia/backend/core/core_types.py | 9 +- libervia/backend/core/main.py | 8 ++ libervia/backend/core/xmpp.py | 4 + libervia/backend/memory/memory.py | 136 +++++++++++++++++- libervia/backend/memory/sqla.py | 50 +++++++ libervia/backend/models/core.py | 19 ++- libervia/frontends/bridge/dbus_bridge.py | 20 +++ libervia/frontends/bridge/pb.py | 46 ++++++ 13 files changed, 377 insertions(+), 4 deletions(-) diff --git a/libervia/backend/bridge/bridge_constructor/bridge_template.ini b/libervia/backend/bridge/bridge_constructor/bridge_template.ini index 283b4f31..922f6302 100644 --- a/libervia/backend/bridge/bridge_constructor/bridge_template.ini +++ b/libervia/backend/bridge/bridge_constructor/bridge_template.ini @@ -49,6 +49,14 @@ doc_param_1=attributes: Dictionary of attributes where keys are: doc_param_2=groups: Roster's groups where the contact is doc_param_3=%(doc_profile)s +[messages_display_state_reset] +type=signal +category=core +sig_in=ss +doc=Display state for a peer was reset (unread count set to zero) +doc_param_0=peer_jid: bare JID of the peer +doc_param_1=%(doc_profile)s + [message_new] type=signal category=core @@ -792,6 +800,27 @@ doc_param_0=options: serialized HistoryLastOptions doc_param_1=%(doc_profile_key)s doc_return=serialized HistoryLastResult +[messages_display_state_set] +type=method +category=core +sig_in=sss +sig_out= +param_3_default="@NONE@" +doc=Store the last displayed message uid for a peer, reset unread count +doc_param_0=peer_jid: bare JID of the peer or room JID for MUC +doc_param_1=message_uid: internal uid of the last displayed message +doc_param_2=%(doc_profile_key)s + +[messages_display_state_get] +type=method +category=core +sig_in=ass +sig_out=s +doc=Get display state (unread count + last displayed UID) for peers +doc_param_0=jids: list of bare JIDs to filter, empty list to get all +doc_param_1=%(doc_profile_key)s +doc_return=serialized ``libervia.backend.models.core.MessageDisplayStatesResult`` + [contact_add] type=method category=core diff --git a/libervia/backend/bridge/dbus_bridge.py b/libervia/backend/bridge/dbus_bridge.py index 3af1b277..514f13ff 100644 --- a/libervia/backend/bridge/dbus_bridge.py +++ b/libervia/backend/bridge/dbus_bridge.py @@ -126,6 +126,8 @@ class DBusObject(objects.DBusObject): Method("message_encryption_start", arguments="ssbs", returns=""), Method("message_encryption_stop", arguments="ss", returns=""), Method("message_send", arguments="sa{ss}a{ss}sss", returns=""), + Method("messages_display_state_get", arguments="ass", returns="s"), + Method("messages_display_state_set", arguments="sss", returns=""), Method("namespaces_get", arguments="", returns="a{ss}"), Method("notification_add", arguments="ssssbbsdss", returns=""), Method("notification_delete", arguments="sbs", returns=""), @@ -176,6 +178,7 @@ class DBusObject(objects.DBusObject): Signal("message_encryption_stopped", "sa{ss}s"), Signal("message_new", "sdssa{ss}a{ss}sss"), Signal("message_update", "ssss"), + Signal("messages_display_state_reset", "ss"), Signal("messages_new", "a(sdssa{ss}a{ss}sss)s"), Signal("notification_deleted", "ss"), Signal("notification_new", "sdssssbidss"), @@ -374,6 +377,14 @@ class DBusObject(objects.DBusObject): "message_send", to_jid, message, subject, mess_type, extra, profile_key ) + def dbus_messages_display_state_get(self, jids, profile_key): + return self._callback("messages_display_state_get", jids, profile_key) + + def dbus_messages_display_state_set(self, peer_jid, message_uid, profile_key): + return self._callback( + "messages_display_state_set", peer_jid, message_uid, profile_key + ) + def dbus_namespaces_get( self, ): @@ -633,6 +644,9 @@ class bridge: def message_update(self, uid, message_type, message_data, profile): self._obj.emitSignal("message_update", uid, message_type, message_data, profile) + def messages_display_state_reset(self, peer_jid, profile): + self._obj.emitSignal("messages_display_state_reset", peer_jid, profile) + def messages_new(self, messages, profile): self._obj.emitSignal("messages_new", messages, profile) diff --git a/libervia/backend/bridge/embedded.py b/libervia/backend/bridge/embedded.py index bbfdd11e..0fa6b9d3 100644 --- a/libervia/backend/bridge/embedded.py +++ b/libervia/backend/bridge/embedded.py @@ -224,6 +224,16 @@ class _Bridge: else: cb(uid, message_type, message_data, profile) + def messages_display_state_reset(self, peer_jid, profile): + try: + cb = self._signals_cbs["core"]["messages_display_state_reset"] + except KeyError: + log.warning( + "ignoring signal messages_display_state_reset: no callback registered" + ) + else: + cb(peer_jid, profile) + def messages_new(self, messages, profile): try: cb = self._signals_cbs["core"]["messages_new"] @@ -842,6 +852,38 @@ class _Bridge: d.addErrback(errback) return d + def messages_display_state_get(self, jids, profile_key, callback=None, errback=None): + try: + ret = self._methods_cbs["messages_display_state_get"](jids, profile_key) + except Exception as e: + if errback is not None: + errback(e) + else: + raise e + else: + if callback is None: + return ret + else: + callback(ret) + + def messages_display_state_set( + self, peer_jid, message_uid, profile_key, callback=None, errback=None + ): + try: + ret = self._methods_cbs["messages_display_state_set"]( + peer_jid, message_uid, profile_key + ) + except Exception as e: + if errback is not None: + errback(e) + else: + raise e + else: + if callback is None: + return ret + else: + callback() + def namespaces_get(self, callback=None, errback=None): try: ret = self._methods_cbs["namespaces_get"]() diff --git a/libervia/backend/bridge/pb.py b/libervia/backend/bridge/pb.py index 62bdd19e..50889999 100644 --- a/libervia/backend/bridge/pb.py +++ b/libervia/backend/bridge/pb.py @@ -236,6 +236,9 @@ class bridge: def message_update(self, uid, message_type, message_data, profile): self.send_signal("message_update", uid, message_type, message_data, profile) + def messages_display_state_reset(self, peer_jid, profile): + self.send_signal("messages_display_state_reset", peer_jid, profile) + def messages_new(self, messages, profile): self.send_signal("messages_new", messages, profile) diff --git a/libervia/backend/core/constants.py b/libervia/backend/core/constants.py index c431d5da..969e6d1d 100644 --- a/libervia/backend/core/constants.py +++ b/libervia/backend/core/constants.py @@ -440,6 +440,7 @@ class Const: KEY_ORDER_BY = "order_by" KEY_USE_CACHE = "use_cache" KEY_DECRYPT = "decrypt" + KEY_LAST_DISP_MSG = "last_displayed_msg_id" ORDER_BY_CREATION = "creation" ORDER_BY_MODIFICATION = "modification" diff --git a/libervia/backend/core/core_types.py b/libervia/backend/core/core_types.py index 19d760b5..4f5a3e40 100644 --- a/libervia/backend/core/core_types.py +++ b/libervia/backend/core/core_types.py @@ -18,7 +18,13 @@ from abc import ABC, abstractmethod from collections import namedtuple -from typing import Any, Self, TYPE_CHECKING, Type, TypeVar +from typing import Any, Literal, Self, TYPE_CHECKING, Type, TypedDict, TypeVar + + +class DisplayStateData(TypedDict): + count: int + last_uid: str | None + from twisted.internet import defer from twisted.python import failure @@ -349,6 +355,7 @@ class SatXMPPEntity(ABC): class SatXMPPClient(SatXMPPEntity): roster: "LiberviaRosterProtocol" + display_state: dict[t_jid.JID, DisplayStateData] class SatXMPPComponent(SatXMPPEntity): diff --git a/libervia/backend/core/main.py b/libervia/backend/core/main.py index 718355fe..4ed8007c 100644 --- a/libervia/backend/core/main.py +++ b/libervia/backend/core/main.py @@ -236,6 +236,14 @@ class LiberviaBackend(service.Service): "history_last_messages_metadata_get", self.memory.history_last_messages_metadata_get, ) + self.bridge.register_method( + "messages_display_state_set", + self.memory._messages_display_state_set, + ) + self.bridge.register_method( + "messages_display_state_get", + self.memory._messages_display_state_get, + ) self.bridge.register_method("presence_set", self._set_presence) self.bridge.register_method("subscription", self.subscription) self.bridge.register_method("contact_add", self._add_contact) diff --git a/libervia/backend/core/xmpp.py b/libervia/backend/core/xmpp.py index b81f2806..7464ee71 100644 --- a/libervia/backend/core/xmpp.py +++ b/libervia/backend/core/xmpp.py @@ -504,6 +504,8 @@ class SatXMPPEntity(core_types.SatXMPPEntity): self.host_app.bridge.profile_connection_error(str(e), self.profile) raise + await self.host_app.memory._init_display_state(self) + self.host_app.bridge.profile_ready(profile) def _disconnection_cb(self, __): @@ -1175,6 +1177,8 @@ class SatXMPPClient(SatXMPPEntity, core_types.SatXMPPClient, tx_xmpp_client.XMPP self.presence = SatPresenceProtocol(self.host_app) self.presence.setHandlerParent(self) + self.display_state: dict[jid.JID, core_types.DisplayStateData] = {} + def _getConnection(self): if self.tor is not None: assert self.tor is not None diff --git a/libervia/backend/memory/memory.py b/libervia/backend/memory/memory.py index 379eb587..259aca92 100644 --- a/libervia/backend/memory/memory.py +++ b/libervia/backend/memory/memory.py @@ -35,7 +35,7 @@ from twisted.words.protocols.jabber import jid from libervia.backend import G 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.core_types import SatXMPPClient, SatXMPPEntity from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.memory.crypto import BlockCipher @@ -633,7 +633,37 @@ class Memory: ## History ## def add_to_history(self, client, data): - return self.storage.add_to_history(data, client.profile) + d = self.storage.add_to_history(data, client.profile) + # Update display state in memory + own_bare = client.jid.userhostJID() + from_bare = data["from"].userhostJID() + to_bare = data["to"].userhostJID() + is_outgoing = from_bare == own_bare + peer_jid = to_bare if is_outgoing else from_bare + state = client.display_state.setdefault(peer_jid, {"count": 0, "last_uid": None}) + if is_outgoing: + # Outgoing message: user is looking at this chat, reset counter and update + # last displayed UID to this message. + state["last_uid"] = data["uid"] + if state["count"] > 0: + state["count"] = 0 + self.host.bridge.messages_display_state_reset( + peer_jid.userhost(), client.profile + ) + return d.addCallback( + lambda _: defer.ensureDeferred( + self.storage.set_private_value( + C.KEY_LAST_DISP_MSG, + peer_jid.userhost(), + data["uid"], + profile=client.profile, + ) + ) + ) + else: + # Incoming message: increment counter + state["count"] += 1 + return d def _history_get_serialise(self, history_data): return [ @@ -731,6 +761,108 @@ class Memory: ) return d + ## Displayed State ## + + def _messages_display_state_set( + self, peer_jid_s: str, message_uid: str, profile: str + ) -> defer.Deferred[None]: + client = G.host.get_client(profile) + return self.messages_display_state_set(client, jid.JID(peer_jid_s), message_uid) + + def messages_display_state_set( + self, client: SatXMPPClient, peer_jid: jid.JID, message_uid: str + ) -> defer.Deferred[None]: + """Store the last displayed message uid for a peer. + + Also resets the unread count for this peer and notifies frontends. + + @param client: Client instance. + @param peer_jid: Bare JID of the peer (or room JID for MUC). + @param message_uid: Internal uid of the last displayed message. + """ + state = client.display_state.setdefault(peer_jid, {"count": 0, "last_uid": None}) + old_count = state["count"] + state["count"] = 0 + state["last_uid"] = message_uid + if old_count > 0: + self.host.bridge.messages_display_state_reset( + peer_jid.userhost(), client.profile + ) + return defer.ensureDeferred( + self.storage.set_private_value( + C.KEY_LAST_DISP_MSG, + peer_jid.userhost(), + message_uid, + profile=client.profile, + ) + ) + + def _messages_display_state_get( + self, jids_s: list[str], profile: str + ) -> defer.Deferred[str]: + client = G.host.get_client(profile) + jids = [jid.JID(j) for j in jids_s] if jids_s else [] + return self.messages_display_state_get(client, jids) + + async def _init_display_state(self, client: SatXMPPClient) -> None: + """Initialize display state from database. + + This method should be called once on profile connection to populate client.display_state. + The data is then updated in-memory. + + @param client: Client instance. + """ + uids = await self.storage.get_privates( + C.KEY_LAST_DISP_MSG, profile=client.profile + ) + counts = await self.storage.unread_counts_get(client, C.KEY_LAST_DISP_MSG) + + for peer_jid_s in uids: + if peer_jid_s not in counts: + log.warning( + f"Display state UID for {peer_jid_s!r} not found in history, " + "resetting display state for this peer" + ) + + client.display_state = { + jid.JID(peer_jid_s): { + "count": counts.get(peer_jid_s, 0), + "last_uid": uids.get(peer_jid_s) if peer_jid_s in counts else None, + } + for peer_jid_s in set(list(uids.keys()) + list(counts.keys())) + } + + def messages_display_state_get( + self, client: SatXMPPClient, jids: list[jid.JID] + ) -> defer.Deferred[str]: + """Get display state for peers. + + Returns cached unread count and last displayed UID per peer, + optionally filtered by JIDs. An empty list means return all peers. + + @param client: Client instance. + @param jids: List of bare JIDs to filter, or empty list for all. + @return: Serialized MessageDisplayStatesResult. + """ + display_state = client.display_state + if jids: + display_state = {k: v for k, v in display_state.items() if k in jids} + + from libervia.backend.models.core import ( + MessageDisplayState, + MessageDisplayStatesResult, + ) + + result = MessageDisplayStatesResult( + { + peer: MessageDisplayState( + count=state["count"], last_uid=state["last_uid"] + ) + for peer, state in display_state.items() + } + ) + return defer.succeed(result.model_dump_json()) + ## Statuses ## def _get_presence_statuses(self, profile_key): diff --git a/libervia/backend/memory/sqla.py b/libervia/backend/memory/sqla.py index 2c79a7ec..e1d23882 100644 --- a/libervia/backend/memory/sqla.py +++ b/libervia/backend/memory/sqla.py @@ -836,6 +836,56 @@ class Storage: return rows + @aio + async def unread_counts_get( + self, client: SatXMPPEntity, namespace: str + ) -> dict[str, int]: + """Compute unread message counts per peer. + + Joins PrivateInd (displayed state) with History to find the displayed timestamp + per peer, then counts messages after that timestamp. + + @param client: Client instance. + @param namespace: PrivateInd namespace for displayed state. + @return: Mapping of peer_jid to unread count. + """ + own_userhost = client.jid.userhost() + profile_id = self.profiles[client.profile] + + # Subquery to get timestamp of last seen message per peer JID + last_timestamps_subq = ( + select( + PrivateInd.key.label("peer_jid"), History.timestamp.label("msg_timestamp") + ) + .join( + History, + and_(PrivateInd.value == History.uid, History.profile_id == profile_id), + ) + .where(PrivateInd.namespace == namespace, PrivateInd.profile_id == profile_id) + .subquery() + ) + + # Be sure to use a peer JID, not our own. + peer_expr = case( + (History.source == own_userhost, History.dest), else_=History.source + ) + + # Count messages after displayed timestamp per peer + stmt = ( + select(peer_expr.label("peer_jid"), func.count().label("cnt")) + .join(last_timestamps_subq, peer_expr == last_timestamps_subq.c.peer_jid) + .where( + History.profile_id == profile_id, + History.timestamp > last_timestamps_subq.c.msg_timestamp, + ) + .group_by(peer_expr) + ) + + async with self.session() as session: + result = await session.execute(stmt) + + return {row[0]: row[1] for row in result} + @aio async def add_to_history(self, data: dict, profile: str) -> None: """Store a new message in history diff --git a/libervia/backend/models/core.py b/libervia/backend/models/core.py index 6dd3e867..fc21902c 100644 --- a/libervia/backend/models/core.py +++ b/libervia/backend/models/core.py @@ -16,7 +16,9 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, RootModel + +from .types import JIDType class MessageData(dict): @@ -62,6 +64,21 @@ class MessageEdition(BaseModel): extra: dict[str, str] = Field(default_factory=dict) +class MessageDisplayState(BaseModel): + """Display state for a single peer.""" + + count: int = Field(description="Unread message count") + last_uid: str | None = Field(description="Internal UID of the last displayed message") + + +class MessageDisplayStatesResult(RootModel[dict[JIDType, MessageDisplayState]]): + """Mapping of peer bare JID to display state.""" + + root: dict[JIDType, MessageDisplayState] = Field( + description="Mapping of peer bare JID to display state" + ) + + class DiscoIdentity(BaseModel): """A single disco identity diff --git a/libervia/frontends/bridge/dbus_bridge.py b/libervia/frontends/bridge/dbus_bridge.py index c7885bc0..3bac5f7d 100644 --- a/libervia/frontends/bridge/dbus_bridge.py +++ b/libervia/frontends/bridge/dbus_bridge.py @@ -395,6 +395,12 @@ class bridge: "message_send", to_jid, message, subject, mess_type, extra, profile_key ) + def messages_display_state_get(self, jids, profile_key): + return self.call("messages_display_state_get", jids, profile_key) + + def messages_display_state_set(self, peer_jid, message_uid, profile_key): + return self.call("messages_display_state_set", peer_jid, message_uid, profile_key) + def namespaces_get( self, ): @@ -737,6 +743,14 @@ class AIOSignalsHandler(SignalsHandler): "message_send", to_jid, message, subject, mess_type, extra, profile_key ) + async def messages_display_state_get(self, jids, profile_key): + return await self.call("messages_display_state_get", jids, profile_key) + + async def messages_display_state_set(self, peer_jid, message_uid, profile_key): + return await self.call( + "messages_display_state_set", peer_jid, message_uid, profile_key + ) + async def namespaces_get( self, ): @@ -1185,6 +1199,12 @@ class TwistedBridge(bridge): "message_send", to_jid, message, subject, mess_type, extra, profile_key ) + def messages_display_state_get(self, jids, profile_key): + return self.call("messages_display_state_get", jids, profile_key) + + def messages_display_state_set(self, peer_jid, message_uid, profile_key): + return self.call("messages_display_state_set", peer_jid, message_uid, profile_key) + def namespaces_get( self, ): diff --git a/libervia/frontends/bridge/pb.py b/libervia/frontends/bridge/pb.py index 70ca976e..15a8bf9a 100644 --- a/libervia/frontends/bridge/pb.py +++ b/libervia/frontends/bridge/pb.py @@ -614,6 +614,28 @@ class bridge(object): else: d.addErrback(self._errback, ori_errback=errback) + def messages_display_state_get(self, jids, profile_key, callback=None, errback=None): + d = self.root.callRemote("messages_display_state_get", jids, profile_key) + if callback is not None: + d.addCallback(callback) + if errback is None: + d.addErrback(self._generic_errback) + else: + d.addErrback(self._errback, ori_errback=errback) + + def messages_display_state_set( + self, peer_jid, message_uid, profile_key, callback=None, errback=None + ): + d = self.root.callRemote( + "messages_display_state_set", peer_jid, message_uid, profile_key + ) + if callback is not None: + d.addCallback(lambda __: callback()) + if errback is None: + d.addErrback(self._generic_errback) + else: + d.addErrback(self._errback, ori_errback=errback) + def namespaces_get(self, callback=None, errback=None): d = self.root.callRemote("namespaces_get") if callback is not None: @@ -1322,6 +1344,18 @@ class AIOBridge(bridge): d.addErrback(self._errback) return d.asFuture(asyncio.get_event_loop()) + def messages_display_state_get(self, jids, profile_key): + d = self.root.callRemote("messages_display_state_get", jids, profile_key) + d.addErrback(self._errback) + return d.asFuture(asyncio.get_event_loop()) + + def messages_display_state_set(self, peer_jid, message_uid, profile_key): + d = self.root.callRemote( + "messages_display_state_set", peer_jid, message_uid, profile_key + ) + d.addErrback(self._errback) + return d.asFuture(asyncio.get_event_loop()) + def namespaces_get(self): d = self.root.callRemote("namespaces_get") d.addErrback(self._errback) @@ -1812,6 +1846,18 @@ class TwistedBridge(bridge): d.addErrback(self._errback) return d + def messages_display_state_get(self, jids, profile_key): + d = self.root.callRemote("messages_display_state_get", jids, profile_key) + d.addErrback(self._errback) + return d + + def messages_display_state_set(self, peer_jid, message_uid, profile_key): + d = self.root.callRemote( + "messages_display_state_set", peer_jid, message_uid, profile_key + ) + d.addErrback(self._errback) + return d + def namespaces_get(self): d = self.root.callRemote("namespaces_get") d.addErrback(self._errback)