erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

core: Unseen message counter:

Browse source 
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.
This commit is contained in:
Goffi 2026-06-20 15:11:06 +02:00
commit e241075a34
13 changed files with 377 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

29
libervia/backend/bridge/bridge_constructor/bridge_template.ini
View file

@ -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

14
libervia/backend/bridge/dbus_bridge.py
View file

@ -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)

42
libervia/backend/bridge/embedded.py
View file

@ -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"]()

3
libervia/backend/bridge/pb.py
View file

@ -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)

1
libervia/backend/core/constants.py
View file

@ -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"

9
libervia/backend/core/core_types.py
View file

@ -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):

8
libervia/backend/core/main.py
View file

@ -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)

4
libervia/backend/core/xmpp.py
View file

@ -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

136
libervia/backend/memory/memory.py
View file

@ -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):

50
libervia/backend/memory/sqla.py
View file

@ -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

19
libervia/backend/models/core.py
View file

@ -16,7 +16,9 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
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

20
libervia/frontends/bridge/dbus_bridge.py
View file

@ -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,
):

46
libervia/frontends/bridge/pb.py
View file

@ -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)