memory (sqla), bridge: New history_last_messages_metadata_get method:
This method is used to retrieve last messages and their timestamps. It's useful on frontend startup to display data on last discussions.
This commit is contained in:
parent
ac315a4300
commit
23cbd8ff9d
9 changed files with 219 additions and 1 deletions
|
|
@ -780,6 +780,18 @@ doc_param_4=filters: patterns to filter the history results, can be:
|
|||
doc_param_5=%(doc_profile)s
|
||||
doc_return=Ordered list (by timestamp) of data as in [message_new] (without final profile)
|
||||
|
||||
[history_last_messages_metadata_get]
|
||||
async=
|
||||
type=method
|
||||
category=core
|
||||
sig_in=ss
|
||||
sig_out=s
|
||||
param_1_default="@NONE@"
|
||||
doc=Get last message data per conversation peer
|
||||
doc_param_0=options: serialized HistoryLastOptions
|
||||
doc_param_1=%(doc_profile_key)s
|
||||
doc_return=serialized HistoryLastResult
|
||||
|
||||
[contact_add]
|
||||
type=method
|
||||
category=core
|
||||
|
|
|
|||
|
|
@ -111,6 +111,7 @@ class DBusObject(objects.DBusObject):
|
|||
Method("entity_data_get", arguments="sass", returns="a{ss}"),
|
||||
Method("features_get", arguments="s", returns="a{sa{ss}}"),
|
||||
Method("history_get", arguments="ssiba{ss}s", returns="a(sdssa{ss}a{ss}ss)"),
|
||||
Method("history_last_messages_metadata_get", arguments="ss", returns="s"),
|
||||
Method("image_check", arguments="s", returns="s"),
|
||||
Method("image_convert", arguments="ssss", returns="s"),
|
||||
Method("image_generate_preview", arguments="ss", returns="s"),
|
||||
|
|
@ -308,6 +309,9 @@ class DBusObject(objects.DBusObject):
|
|||
"history_get", from_jid, to_jid, limit, between, filters, profile
|
||||
)
|
||||
|
||||
def dbus_history_last_messages_metadata_get(self, options, profile_key="@NONE@"):
|
||||
return self._callback("history_last_messages_metadata_get", options, profile_key)
|
||||
|
||||
def dbus_image_check(self, arg_0):
|
||||
return self._callback("image_check", arg_0)
|
||||
|
||||
|
|
|
|||
|
|
@ -647,6 +647,16 @@ class _Bridge:
|
|||
d.addErrback(errback)
|
||||
return d
|
||||
|
||||
def history_last_messages_metadata_get(
|
||||
self, options, profile_key="@NONE@", callback=None, errback=None
|
||||
):
|
||||
d = self._methods_cbs["history_last_messages_metadata_get"](options, profile_key)
|
||||
if callback is not None:
|
||||
d.addCallback(callback)
|
||||
if errback is not None:
|
||||
d.addErrback(errback)
|
||||
return d
|
||||
|
||||
def image_check(self, arg_0, callback=None, errback=None):
|
||||
try:
|
||||
ret = self._methods_cbs["image_check"](arg_0)
|
||||
|
|
|
|||
|
|
@ -232,6 +232,10 @@ class LiberviaBackend(service.Service):
|
|||
"params_register_app", self.memory.params_register_app
|
||||
)
|
||||
self.bridge.register_method("history_get", self.memory._history_get)
|
||||
self.bridge.register_method(
|
||||
"history_last_messages_metadata_get",
|
||||
self.memory.history_last_messages_metadata_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)
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ from twisted.internet import defer, error, reactor
|
|||
from twisted.internet.base import DelayedCall
|
||||
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
|
||||
|
|
@ -48,6 +49,11 @@ from libervia.backend.memory.sqla import (
|
|||
NotificationType,
|
||||
Storage,
|
||||
)
|
||||
from libervia.backend.models.core import (
|
||||
HistoryLastOptions,
|
||||
HistoryLastResult,
|
||||
HistoryPeerInfo,
|
||||
)
|
||||
from libervia.backend.tools import config as tools_config
|
||||
from libervia.backend.tools.common import data_format
|
||||
from libervia.backend.tools.common import regex
|
||||
|
|
@ -693,6 +699,38 @@ class Memory:
|
|||
from_jid, to_jid, limit, between, filters, profile
|
||||
)
|
||||
|
||||
def history_last_messages_metadata_get(
|
||||
self,
|
||||
options_s: str,
|
||||
profile: str,
|
||||
) -> defer.Deferred[str]:
|
||||
"""Get last message timestamp per conversation peer.
|
||||
|
||||
@param options_s: Serialised HistoryLastOptions.
|
||||
@param profile: Profile.
|
||||
@return: Serialised HistoryLastResult.
|
||||
"""
|
||||
client = G.host.get_client(profile)
|
||||
options = HistoryLastOptions.model_validate_json(options_s)
|
||||
d = defer.ensureDeferred(
|
||||
self.storage.history_last_messages_metadata_get(
|
||||
client, options.types, options.since, options.limit
|
||||
)
|
||||
)
|
||||
d.addCallback(
|
||||
lambda last_data: HistoryLastResult(
|
||||
peers={
|
||||
peer: HistoryPeerInfo(
|
||||
timestamp=info["timestamp"],
|
||||
last_uid=info.get("uid", ""),
|
||||
last_message=info.get("message", ""),
|
||||
)
|
||||
for peer, info in last_data.items()
|
||||
}
|
||||
).model_dump_json()
|
||||
)
|
||||
return d
|
||||
|
||||
## Statuses ##
|
||||
|
||||
def _get_presence_statuses(self, profile_key):
|
||||
|
|
|
|||
|
|
@ -28,7 +28,20 @@ from typing import Any, Dict, List, Literal, Optional, Tuple, Union
|
|||
|
||||
from alembic import config as al_config, script as al_script
|
||||
from alembic.runtime import migration as al_migration
|
||||
from sqlalchemy import and_, delete, desc, event, exists, func, literal, or_, true, update
|
||||
from sqlalchemy import (
|
||||
and_,
|
||||
case,
|
||||
delete,
|
||||
desc,
|
||||
event,
|
||||
exists,
|
||||
func,
|
||||
literal,
|
||||
or_,
|
||||
over,
|
||||
true,
|
||||
update,
|
||||
)
|
||||
from sqlalchemy import Integer, literal_column, text
|
||||
from sqlalchemy.dialects.sqlite import insert
|
||||
from sqlalchemy.engine import Connection, Engine
|
||||
|
|
@ -738,6 +751,91 @@ class Storage:
|
|||
result.reverse()
|
||||
return [h.as_tuple() for h in result]
|
||||
|
||||
@aio
|
||||
async def history_last_messages_metadata_get(
|
||||
self,
|
||||
client: SatXMPPEntity,
|
||||
types: list[str] | None = None,
|
||||
since: float | None = None,
|
||||
limit: int | None = None,
|
||||
) -> dict[str, dict]:
|
||||
"""Get last message info per conversation peer.
|
||||
|
||||
For each peer, returns the timestamp, uid, and message text of the most recent
|
||||
message.
|
||||
|
||||
@param client: Client instance.
|
||||
@param types: Message types to include.
|
||||
@param since: Get messages after this timestamp.
|
||||
@param limit: Max number of peers to return.
|
||||
@return: Mapping of bare JID strings to {timestamp, uid, message}.
|
||||
"""
|
||||
own_userhost = client.jid.userhost()
|
||||
profile_id = self.profiles[client.profile]
|
||||
|
||||
# Filters
|
||||
where_clauses = [
|
||||
History.profile_id == profile_id,
|
||||
or_(History.source == own_userhost, History.dest == own_userhost),
|
||||
]
|
||||
if types:
|
||||
where_clauses.append(History.type.in_(types))
|
||||
if since is not None:
|
||||
where_clauses.append(History.timestamp >= since)
|
||||
|
||||
# Window function: rank history entries from newest to oldest within each peer
|
||||
peer_expr = case(
|
||||
(History.source == own_userhost, History.dest), else_=History.source
|
||||
)
|
||||
ranked = (
|
||||
select(
|
||||
peer_expr.label("peer"),
|
||||
History.timestamp.label("last_ts"),
|
||||
History.uid.label("last_uid"),
|
||||
over(
|
||||
func.row_number(),
|
||||
partition_by=peer_expr,
|
||||
order_by=text("timestamp DESC, uid DESC"),
|
||||
).label("row_rank"),
|
||||
)
|
||||
.where(*where_clauses)
|
||||
.subquery()
|
||||
)
|
||||
|
||||
# We only keep results with `row_rank == 1` (i.e. the last message)
|
||||
stmt = (
|
||||
select(ranked.c.peer, ranked.c.last_ts, ranked.c.last_uid)
|
||||
.where(ranked.c.row_rank == 1)
|
||||
.order_by(desc(ranked.c.last_ts))
|
||||
)
|
||||
if limit is not None:
|
||||
stmt = stmt.limit(limit)
|
||||
|
||||
async with self.session() as session:
|
||||
result = await session.execute(stmt)
|
||||
rows = {
|
||||
row[0]: {"timestamp": row[1], "uid": row[2], "message": ""}
|
||||
for row in result
|
||||
}
|
||||
|
||||
# We fetch message texts separately (only for rows that have uids)
|
||||
uids = [info["uid"] for info in rows.values() if info["uid"]]
|
||||
if uids:
|
||||
msg_stmt = select(Message.history_uid, Message.message).where(
|
||||
Message.history_uid.in_(uids)
|
||||
)
|
||||
msg_result = await session.execute(msg_stmt)
|
||||
msgs_by_uid = {}
|
||||
# Pick first non-NULL message per uid
|
||||
for hist_uid, msg in msg_result:
|
||||
if hist_uid not in msgs_by_uid and msg:
|
||||
msgs_by_uid[hist_uid] = msg
|
||||
# If no non-NULL message has been found, we use an empty string.
|
||||
for info in rows.values():
|
||||
info["message"] = msgs_by_uid.get(info["uid"], "")
|
||||
|
||||
return rows
|
||||
|
||||
@aio
|
||||
async def add_to_history(self, data: dict, profile: str) -> None:
|
||||
"""Store a new message in history
|
||||
|
|
|
|||
|
|
@ -23,6 +23,22 @@ class MessageData(dict):
|
|||
pass
|
||||
|
||||
|
||||
class HistoryLastOptions(BaseModel):
|
||||
types: list[str] | None = None
|
||||
since: float | None = None
|
||||
limit: int | None = None
|
||||
|
||||
|
||||
class HistoryPeerInfo(BaseModel):
|
||||
timestamp: float
|
||||
last_uid: str = ""
|
||||
last_message: str = ""
|
||||
|
||||
|
||||
class HistoryLastResult(BaseModel):
|
||||
peers: dict[str, HistoryPeerInfo]
|
||||
|
||||
|
||||
class MessageUpdateData(BaseModel):
|
||||
type: str
|
||||
|
||||
|
|
|
|||
|
|
@ -330,6 +330,9 @@ class bridge:
|
|||
"history_get", from_jid, to_jid, limit, between, filters, profile
|
||||
)
|
||||
|
||||
def history_last_messages_metadata_get(self, options, profile_key="@NONE@"):
|
||||
return self.call("history_last_messages_metadata_get", options, profile_key)
|
||||
|
||||
def image_check(self, arg_0):
|
||||
return self.call("image_check", arg_0)
|
||||
|
||||
|
|
@ -669,6 +672,9 @@ class AIOSignalsHandler(SignalsHandler):
|
|||
"history_get", from_jid, to_jid, limit, between, filters, profile
|
||||
)
|
||||
|
||||
async def history_last_messages_metadata_get(self, options, profile_key="@NONE@"):
|
||||
return await self.call("history_last_messages_metadata_get", options, profile_key)
|
||||
|
||||
async def image_check(self, arg_0):
|
||||
return await self.call("image_check", arg_0)
|
||||
|
||||
|
|
@ -1114,6 +1120,9 @@ class TwistedBridge(bridge):
|
|||
"history_get", from_jid, to_jid, limit, between, filters, profile
|
||||
)
|
||||
|
||||
def history_last_messages_metadata_get(self, options, profile_key="@NONE@"):
|
||||
return self.call("history_last_messages_metadata_get", options, profile_key)
|
||||
|
||||
def image_check(self, arg_0):
|
||||
return self.call("image_check", arg_0)
|
||||
|
||||
|
|
|
|||
|
|
@ -438,6 +438,19 @@ class bridge(object):
|
|||
else:
|
||||
d.addErrback(self._errback, ori_errback=errback)
|
||||
|
||||
def history_last_messages_metadata_get(
|
||||
self, options, profile_key="@NONE@", callback=None, errback=None
|
||||
):
|
||||
d = self.root.callRemote(
|
||||
"history_last_messages_metadata_get", options, 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 image_check(self, arg_0, callback=None, errback=None):
|
||||
d = self.root.callRemote("image_check", arg_0)
|
||||
if callback is not None:
|
||||
|
|
@ -1216,6 +1229,13 @@ class AIOBridge(bridge):
|
|||
d.addErrback(self._errback)
|
||||
return d.asFuture(asyncio.get_event_loop())
|
||||
|
||||
def history_last_messages_metadata_get(self, options, profile_key="@NONE@"):
|
||||
d = self.root.callRemote(
|
||||
"history_last_messages_metadata_get", options, profile_key
|
||||
)
|
||||
d.addErrback(self._errback)
|
||||
return d.asFuture(asyncio.get_event_loop())
|
||||
|
||||
def image_check(self, arg_0):
|
||||
d = self.root.callRemote("image_check", arg_0)
|
||||
d.addErrback(self._errback)
|
||||
|
|
@ -1699,6 +1719,13 @@ class TwistedBridge(bridge):
|
|||
d.addErrback(self._errback)
|
||||
return d
|
||||
|
||||
def history_last_messages_metadata_get(self, options, profile_key="@NONE@"):
|
||||
d = self.root.callRemote(
|
||||
"history_last_messages_metadata_get", options, profile_key
|
||||
)
|
||||
d.addErrback(self._errback)
|
||||
return d
|
||||
|
||||
def image_check(self, arg_0):
|
||||
d = self.root.callRemote("image_check", arg_0)
|
||||
d.addErrback(self._errback)
|
||||
|
|
|
|||
Loading…
Reference in a new issue