Coverage for slidge/group/bookmarks.py: 91%
174 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-18 04:30 +0000
1import logging
2from collections.abc import Iterator
3from typing import Any, Literal, overload
5from slixmpp import JID
6from slixmpp.exceptions import XMPPError
7from sqlalchemy.orm import Session as OrmSession
9from slidge.contact import LegacyContact
11from ..db.models import Contact, Room, Space
12from ..util.jid_escaping import EscapeMixin
13from ..util.lock import NamedLockMixin
14from ..util.types import (
15 AnyMUC,
16 AnySession,
17 LegacyMUCType,
18 SpaceMetadata,
19)
20from ..util.util import derive_wired_class
21from .room import LegacyMUC
24class LegacyBookmarks[LegacyMUCType: AnyMUC](
25 EscapeMixin,
26 NamedLockMixin,
27):
28 """
29 This is instantiated once per :class:`~slidge.BaseSession`
30 """
32 muc_cls: type[LegacyMUCType]
33 """
34 The concrete :class:`.LegacyMUC` subclass these bookmarks produce.
36 Derived automatically from the generic parameter, e.g.,
37 ``class Bookmarks(LegacyBookmarks[MUC])`` produces ``MUC`` instances.
38 """
40 def __init_subclass__(cls, **kwargs: object) -> None:
41 super().__init_subclass__(**kwargs)
42 derive_wired_class(cls, LegacyBookmarks, "muc_cls")
44 def __init__(self, session: AnySession) -> None:
45 self.session = session
46 self.xmpp = session.xmpp
47 self.user_jid = session.user_jid
49 self._user_nick: str = self.session.user_jid.node
51 super().__init__()
52 self.log = logging.getLogger(f"{self.user_jid.bare}:bookmarks")
53 self.ready = self.session.xmpp.loop.create_future()
54 if not self.xmpp.GROUPS:
55 self.ready.set_result(True)
57 @property
58 def user_nick(self) -> str:
59 return self._user_nick
61 @user_nick.setter
62 def user_nick(self, nick: str) -> None:
63 self._user_nick = nick
65 def orm(
66 self,
67 **kwargs: Any, # noqa:ANN401
68 ) -> OrmSession:
69 return self.session.xmpp.store.session(**kwargs)
71 def from_store(self, stored: Room) -> LegacyMUCType:
72 return self.muc_cls(self.session, stored)
74 def __iter__(self) -> Iterator[LegacyMUCType]:
75 with self.xmpp.store.session() as orm:
76 rooms = (
77 orm.query(Room).filter_by(user=self.session.user, updated=True).all()
78 )
79 for stored in rooms:
80 yield self.from_store(stored)
82 def __repr__(self) -> str:
83 return f"<Bookmarks of {self.user_jid}>"
85 async def legacy_id_to_jid_local_part(self, legacy_id: str) -> str:
86 return await self.legacy_id_to_jid_username(legacy_id)
88 async def jid_local_part_to_legacy_id(self, local_part: str) -> str:
89 return await self.jid_username_to_legacy_id(local_part)
91 async def by_jid(self, jid: JID) -> LegacyMUCType:
92 if jid.resource:
93 jid = JID(jid.bare)
94 async with self.lock(("bare", jid.bare)):
95 legacy_id = await self.jid_local_part_to_legacy_id(jid.node)
96 if self.get_lock(("legacy_id", legacy_id)):
97 self.session.log.debug("Already updating %s via by_legacy_id()", jid)
98 return await self.by_legacy_id(legacy_id)
100 with self.session.xmpp.store.session() as orm:
101 stored = (
102 orm.query(Room)
103 .filter_by(
104 user_account_id=self.session.user_pk, jid_localpart=jid.local
105 )
106 .one_or_none()
107 )
108 if stored is None:
109 stored = Room(
110 user_account_id=self.session.user_pk,
111 jid_localpart=jid.local,
112 legacy_id=legacy_id,
113 )
114 return await self.__update_if_needed(stored)
116 def by_jid_only_if_exists(self, jid: JID) -> LegacyMUCType | None:
117 with self.xmpp.store.session(expire_on_commit=False) as orm:
118 stored = (
119 orm.query(Room)
120 .filter_by(user=self.session.user, jid_localpart=jid.local)
121 .one_or_none()
122 )
123 if stored is not None and stored.updated:
124 return self.from_store(stored)
125 return None
127 @overload
128 async def by_legacy_id(self, legacy_id: str) -> "LegacyMUCType": ...
130 @overload
131 async def by_legacy_id(
132 self, legacy_id: str, create: Literal[False]
133 ) -> "LegacyMUCType | None": ...
135 @overload
136 async def by_legacy_id(
137 self, legacy_id: str, create: Literal[True]
138 ) -> "LegacyMUCType": ...
140 async def by_legacy_id(
141 self, legacy_id: str, create: bool = False
142 ) -> LegacyMUCType | None:
143 async with self.lock(("legacy_id", legacy_id)):
144 local = await self.legacy_id_to_jid_local_part(legacy_id)
145 jid = JID(f"{local}@{self.xmpp.boundjid}")
146 if self.get_lock(("bare", jid.bare)):
147 self.session.log.debug("Already updating %s via by_jid()", jid)
148 if create:
149 return await self.by_jid(jid)
150 else:
151 return self.by_jid_only_if_exists(jid)
153 with self.xmpp.store.session() as orm:
154 stored = (
155 orm.query(Room)
156 .filter_by(
157 user_account_id=self.session.user_pk,
158 legacy_id=str(legacy_id),
159 )
160 .one_or_none()
161 )
162 if stored is None:
163 stored = Room(
164 user_account_id=self.session.user_pk,
165 jid_localpart=local,
166 legacy_id=str(legacy_id),
167 )
168 return await self.__update_if_needed(stored)
170 async def __update_if_needed(self, stored: Room) -> LegacyMUCType:
171 muc = self.from_store(stored)
172 if muc.stored.updated:
173 return muc
175 with muc.updating_info():
176 try:
177 await muc.update_info()
178 except NotImplementedError:
179 pass
180 except XMPPError:
181 raise
182 except Exception as e: # noqa: BLE001
183 raise XMPPError("internal-server-error", str(e))
184 muc.archive.room = muc.stored
185 if self.ready.done() and muc.stored.space_id:
186 with self.orm() as orm:
187 orm.add(muc.stored)
188 self.xmpp.pubsub.broadcast_space(self.session, muc.stored.space)
189 return muc
191 async def fill(self) -> None:
192 """
193 Establish a user's known groups.
195 This has to be overridden in plugins with group support and at the
196 minimum, this should ``await self.by_legacy_id(group_id)`` for all
197 the groups a user is part of.
199 Slidge internals will call this on successful :meth:`BaseSession.login`
201 """
202 if self.xmpp.GROUPS:
203 raise NotImplementedError(
204 "The plugin advertised support for groups but"
205 " LegacyBookmarks.fill() was not overridden."
206 )
208 async def remove(
209 self,
210 muc: AnyMUC,
211 reason: str = "You left this group from the official client.",
212 kick: bool = True,
213 ) -> None:
214 """
215 Delete everything about a specific group.
217 This should be called when the user leaves the group from the official
218 app.
220 :param muc: The MUC to remove.
221 :param reason: Optionally, a reason why this group was removed.
222 :param kick: Whether the user should be kicked from this group. Set this
223 to False in case you do this somewhere else in your code, eg, on
224 receiving the confirmation that the group was deleted.
225 """
226 if kick:
227 user_participant = await muc.get_user_participant()
228 user_participant.kick(reason)
229 with self.xmpp.store.session() as orm:
230 orm.add(muc.stored)
231 orm.refresh(muc.stored)
232 orm.delete(muc.stored)
233 orm.commit()
235 async def update_space_if_needed(self, space: Space) -> Space:
236 async with self.lock(("space", space.legacy_id)):
237 with self.orm() as orm:
238 orm.add(space)
239 orm.refresh(space)
240 if space.updated:
241 return space
242 meta: SpaceMetadata = await self.fetch_space_metadata(space.legacy_id)
243 return await self.__update_space_metadata(space, meta)
245 async def __update_space_metadata(self, space: Space, meta: SpaceMetadata) -> Space:
246 creator = (
247 await self.__get_stored_contact(meta.creator_legacy_id)
248 if meta.creator_legacy_id
249 else None
250 )
251 owners: list[Contact] = []
252 for legacy_id in set(meta.owner_legacy_ids):
253 if legacy_id == meta.creator_legacy_id:
254 # We don't want to fetch any contact twice here to avoid:
255 # Can't attach instance <Contact at xxx>; another instance [...] is already present in this session.
256 continue
257 owner = await self.__get_stored_contact(legacy_id)
258 if owner is not None:
259 owners.append(owner)
260 if (
261 creator is not None
262 and meta.creator_legacy_id
263 and meta.creator_legacy_id in meta.owner_legacy_ids
264 ):
265 owners.append(creator)
266 with self.orm(expire_on_commit=False) as orm:
267 if creator is not None:
268 creator = orm.merge(creator)
269 owners = [orm.merge(owner) for owner in owners]
270 space = orm.merge(space)
271 space.name = meta.name or space.legacy_id
272 space.creator = creator
273 space.owners = owners
274 space.description = meta.description
275 space.member_count = meta.member_count
276 space.updated = True
277 orm.commit()
278 return space
280 async def __get_stored_contact(self, legacy_id_str: str) -> Contact | None:
281 try:
282 contact: LegacyContact = await self.session.contacts.by_legacy_id(
283 legacy_id_str
284 )
285 except Exception as e: # noqa: BLE001
286 self.log.warning("Could not get contact: %r", e)
287 return None
288 return contact.stored
290 async def fetch_space_metadata(self, legacy_id: str) -> SpaceMetadata:
291 """
292 Fetch metadata associated to a space.
294 This is called once per slidge runtime. It should return metadata
295 associated to the space identified by its ``legacy_id``.
296 If there are updates to this metata, they should be communicated to
297 slidge by calling :func:`LegacyBookmarks.update_space_metadata`.
299 :param legacy_id: Identifier of the space.
301 :return: Metadata associated to the space.
302 """
303 raise NotImplementedError
305 async def update_spaces_info(self) -> None:
306 with self.orm() as orm:
307 spaces = self.session.xmpp.store.spaces.get_unupdated(
308 orm, self.session.user_pk
309 )
310 for space in spaces:
311 await self.update_space_if_needed(space)
313 async def space_legacy_id_to_node(self, legacy_id: str) -> str:
314 return str(legacy_id)
316 async def space_node_to_legacy_id(self, node: str) -> str:
317 return node
319 async def update_space_metadata(
320 self,
321 legacy_id: str,
322 metadata: SpaceMetadata,
323 ) -> None:
324 """
325 Updates metadata associated to a space.
327 :param legacy_id: Identifier of the space.
328 :param name: Metadata associated to this space.
329 """
330 with self.orm(expire_on_commit=False) as orm:
331 space = self.session.xmpp.store.spaces.add_or_get(
332 orm,
333 self.session.user_pk,
334 str(legacy_id),
335 )
336 space = await self.__update_space_metadata(space, metadata)
337 self.xmpp.pubsub.broadcast_space_metadata(
338 self.session, space, await self.space_legacy_id_to_node(legacy_id)
339 )
342LegacyBookmarks.muc_cls = LegacyMUC # type:ignore[misc]