Coverage for slidge/group/bookmarks.py: 90%
235 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-29 05:05 +0000
1import logging
2import warnings
3from collections.abc import Iterable, Iterator
4from typing import Any, Literal, overload
6from slixmpp import JID
7from slixmpp.exceptions import XMPPError
8from sqlalchemy.orm import Session as OrmSession
10from slidge.contact import LegacyContact
11from slidge.db.avatar import avatar_cache
12from slidge.db.meta import modified_attributes
14from ..db.models import Avatar as AvatarModel
15from ..db.models import Contact, Room, Space
16from ..util.jid_escaping import EscapeMixin
17from ..util.lock import NamedLockMixin
18from ..util.types import (
19 AnyMUC,
20 AnySession,
21 Avatar,
22 LegacyMUCType,
23 SpaceMetadata,
24 Unset,
25)
26from ..util.util import derive_wired_class
27from .room import LegacyMUC
30class LegacyBookmarks[LegacyMUCType: AnyMUC](
31 EscapeMixin,
32 NamedLockMixin,
33):
34 """
35 This is instantiated once per :class:`~slidge.BaseSession`
36 """
38 muc_cls: type[LegacyMUCType]
39 """
40 The concrete :class:`.LegacyMUC` subclass these bookmarks produce.
42 Derived automatically from the generic parameter, e.g.,
43 ``class Bookmarks(LegacyBookmarks[MUC])`` produces ``MUC`` instances.
44 """
46 def __init_subclass__(cls, **kwargs: object) -> None:
47 super().__init_subclass__(**kwargs)
48 derive_wired_class(cls, LegacyBookmarks, "muc_cls")
50 def __init__(self, session: AnySession) -> None:
51 self.session = session
52 self.xmpp = session.xmpp
53 self.user_jid = session.user_jid
55 self._user_nick: str = self.session.user_jid.node
57 super().__init__()
58 self.log = logging.getLogger(f"{self.user_jid.bare}:bookmarks")
59 self.ready = self.session.xmpp.loop.create_future()
60 if not self.xmpp.GROUPS:
61 self.ready.set_result(True)
63 @property
64 def user_nick(self) -> str:
65 return self._user_nick
67 @user_nick.setter
68 def user_nick(self, nick: str) -> None:
69 self._user_nick = nick
71 def orm(
72 self,
73 **kwargs: Any, # noqa:ANN401
74 ) -> OrmSession:
75 return self.session.xmpp.store.session(**kwargs)
77 def from_store(self, stored: Room) -> LegacyMUCType:
78 return self.muc_cls(self.session, stored)
80 def __iter__(self) -> Iterator[LegacyMUCType]:
81 with self.xmpp.store.session() as orm:
82 rooms = (
83 orm.query(Room).filter_by(user=self.session.user, updated=True).all()
84 )
85 for stored in rooms:
86 yield self.from_store(stored)
88 def __repr__(self) -> str:
89 return f"<Bookmarks of {self.user_jid}>"
91 async def legacy_id_to_jid_local_part(self, legacy_id: str) -> str:
92 return await self.legacy_id_to_jid_username(legacy_id)
94 async def jid_local_part_to_legacy_id(self, local_part: str) -> str:
95 return await self.jid_username_to_legacy_id(local_part)
97 async def by_jid(self, jid: JID, *update_info_args: object) -> LegacyMUCType:
98 if jid.resource:
99 jid = JID(jid.bare)
100 async with self.lock(("bare", jid.bare)):
101 legacy_id = await self.jid_local_part_to_legacy_id(jid.node)
102 if self.get_lock(("legacy_id", legacy_id)):
103 self.session.log.debug("Already updating %s via by_legacy_id()", jid)
104 return await self.by_legacy_id(legacy_id)
106 with self.session.xmpp.store.session() as orm:
107 stored = (
108 orm.query(Room)
109 .filter_by(
110 user_account_id=self.session.user_pk, jid_localpart=jid.local
111 )
112 .one_or_none()
113 )
114 if stored is None:
115 stored = Room(
116 user_account_id=self.session.user_pk,
117 jid_localpart=jid.local,
118 legacy_id=legacy_id,
119 )
120 return await self.__update_if_needed(stored, *update_info_args)
122 def by_jid_only_if_exists(self, jid: JID) -> LegacyMUCType | None:
123 with self.xmpp.store.session(expire_on_commit=False) as orm:
124 stored = (
125 orm.query(Room)
126 .filter_by(user=self.session.user, jid_localpart=jid.local)
127 .one_or_none()
128 )
129 if stored is not None and stored.updated:
130 return self.from_store(stored)
131 return None
133 @overload
134 async def by_legacy_id(
135 self, /, legacy_id: str, *update_info_args: object
136 ) -> "LegacyMUCType": ...
138 @overload
139 async def by_legacy_id(
140 self, /, legacy_id: str, *update_info_args: object, create: Literal[False]
141 ) -> "LegacyMUCType | None": ...
143 @overload
144 async def by_legacy_id(
145 self, /, legacy_id: str, *update_info_args: object, create: Literal[True]
146 ) -> "LegacyMUCType": ...
148 async def by_legacy_id(
149 self, /, legacy_id: str, *update_info_args: object, create: bool = True
150 ) -> LegacyMUCType | None:
151 async with self.lock(("legacy_id", legacy_id)):
152 local = await self.legacy_id_to_jid_local_part(legacy_id)
153 jid = JID(f"{local}@{self.xmpp.boundjid}")
154 if self.get_lock(("bare", jid.bare)):
155 self.session.log.debug("Already updating %s via by_jid()", jid)
156 if create:
157 return await self.by_jid(jid, *update_info_args)
158 else:
159 if update_info_args:
160 self.log.warning(
161 "By legacy ID called with `create=False`, "
162 "`update_info_args` will not be used."
163 )
164 return self.by_jid_only_if_exists(jid)
166 with self.xmpp.store.session() as orm:
167 stored = (
168 orm.query(Room)
169 .filter_by(
170 user_account_id=self.session.user_pk,
171 legacy_id=str(legacy_id),
172 )
173 .one_or_none()
174 )
175 if stored is None:
176 if not create:
177 return None
178 stored = Room(
179 user_account_id=self.session.user_pk,
180 jid_localpart=local,
181 legacy_id=str(legacy_id),
182 )
183 return await self.__update_if_needed(stored, *update_info_args)
185 async def __update_if_needed(
186 self, stored: Room, *update_info_args: object
187 ) -> LegacyMUCType:
188 muc = self.from_store(stored)
189 if muc.stored.updated and not update_info_args:
190 return muc
192 with muc.updating_info():
193 try:
194 await muc.update_info(*update_info_args)
195 except NotImplementedError:
196 pass
197 except XMPPError:
198 raise
199 except Exception as e: # noqa: BLE001
200 raise XMPPError("internal-server-error", str(e))
201 muc.archive.room = muc.stored
202 if self.ready.done() and muc.stored.space_id:
203 assert muc.stored.space is not None
204 node = await self.space_legacy_id_to_node(muc.stored.space.legacy_id)
205 with self.orm() as orm:
206 orm.add(muc.stored)
207 self.xmpp.pubsub.broadcast_space(
208 self.session, muc.stored.space, node, [str(muc.jid)]
209 )
210 return muc
212 async def fill(self) -> None:
213 """
214 Establish a user's known groups.
216 This has to be overridden in plugins with group support and at the
217 minimum, this should ``await self.by_legacy_id(group_id)`` for all
218 the groups a user is part of.
220 Slidge internals will call this on successful :meth:`BaseSession.login`
222 """
223 if self.xmpp.GROUPS:
224 raise NotImplementedError(
225 "The plugin advertised support for groups but"
226 " LegacyBookmarks.fill() was not overridden."
227 )
229 async def remove(
230 self,
231 muc: AnyMUC,
232 reason: str = "You left this group from the official client.",
233 kick: bool = True,
234 ) -> None:
235 """
236 Delete everything about a specific group.
238 This should be called when the user leaves the group from the official
239 app.
241 :param muc: The MUC to remove.
242 :param reason: Optionally, a reason why this group was removed.
243 :param kick: Whether the user should be kicked from this group. Set this
244 to False in case you do this somewhere else in your code, eg, on
245 receiving the confirmation that the group was deleted.
246 """
247 if kick:
248 user_participant = await muc.get_user_participant()
249 user_participant.kick(reason)
250 with self.xmpp.store.session() as orm:
251 orm.add(muc.stored)
252 orm.refresh(muc.stored)
253 orm.delete(muc.stored)
254 orm.commit()
256 async def update_space_if_needed(self, space: Space) -> Space:
257 async with self.lock(("space", space.legacy_id)):
258 with self.orm() as orm:
259 orm.add(space)
260 orm.refresh(space)
261 if space.updated:
262 return space
263 orm.refresh(space, ["avatar", "banner"])
264 meta = await self.fetch_space_metadata(space.legacy_id)
265 return (await self.__update_space_metadata(space, meta))[0]
267 async def get_updated_space(
268 self,
269 legacy_id: str,
270 room_legacy_id_filter: Iterable[str] | None = None,
271 ) -> Space | None:
272 with self.orm() as orm:
273 space = self.xmpp.store.spaces.get_by_legacy_id(
274 orm,
275 self.session.user_pk,
276 legacy_id,
277 images=True,
278 room_legacy_id_filter=room_legacy_id_filter,
279 )
280 if space is None:
281 return None
282 if space.updated:
283 return space
284 async with self.lock(("space", space.legacy_id)):
285 meta = await self.fetch_space_metadata(space.legacy_id)
286 return (await self.__update_space_metadata(space, meta))[0]
288 async def __update_space_metadata(
289 self, space: Space, meta: SpaceMetadata
290 ) -> tuple[Space, set[str]]:
291 creator = (
292 await self.__get_stored_contact(meta.creator_legacy_id)
293 if meta.creator_legacy_id
294 else None
295 )
296 owners: list[Contact] = []
297 if not isinstance(meta.owner_legacy_ids, Unset):
298 for legacy_id in set(meta.owner_legacy_ids or ()):
299 if legacy_id == meta.creator_legacy_id:
300 # We don't want to fetch any contact twice here to avoid:
301 # Can't attach instance <Contact at xxx>; another instance [...] is already present in this session.
302 continue
303 owner = await self.__get_stored_contact(legacy_id)
304 if owner is not None:
305 owners.append(owner)
306 if (
307 creator is not None
308 and meta.creator_legacy_id
309 and (
310 isinstance(meta.owner_legacy_ids, Unset)
311 or meta.creator_legacy_id in meta.owner_legacy_ids
312 )
313 ):
314 owners.append(creator)
316 changed = set()
318 for attr in "avatar", "banner":
319 try:
320 if await self.__update_space_image(space, attr, getattr(meta, attr)):
321 changed.add(attr)
322 except Exception:
323 self.log.exception("Can't set %s for %s", attr, space)
325 with self.orm(expire_on_commit=False) as orm:
326 if creator is not None:
327 creator = orm.merge(creator)
328 owners = [orm.merge(owner) for owner in owners]
329 space = orm.merge(space)
330 if isinstance(meta.name, Unset):
331 if not space.name:
332 space.name = space.legacy_id
333 else:
334 space.name = meta.name or space.name or space.legacy_id
336 if not isinstance(meta.creator_legacy_id, Unset):
337 space.creator = creator
338 if not isinstance(meta.owner_legacy_ids, Unset):
339 space.owners = owners
340 if not isinstance(meta.description, Unset):
341 space.description = meta.description
342 if not isinstance(meta.member_count, Unset):
343 space.member_count = meta.member_count
345 changed |= modified_attributes(space)
346 self.log.debug("Changed space attributes: %s", changed)
348 if not space.updated:
349 # setting .updated to True means that fetch_space_metadata()
350 # should not be called for this specific space, but does not
351 # mean that we have to broadcast a pubsub#metadata change
352 space.updated = True
354 if orm.is_modified(space, include_collections=True):
355 orm.commit()
356 return space, changed # update_img, bool(changed)
358 async def __update_space_image(
359 self,
360 space: Space,
361 attr: Literal["avatar", "banner"],
362 new: Avatar | Unset | None,
363 ) -> bool:
364 stored: AvatarModel | None = getattr(space, attr)
366 if isinstance(new, Unset):
367 return False
369 if new is None:
370 setattr(space, attr, None)
371 return stored is not None
373 cached_avatar = await avatar_cache.get(new, session=self.session, convert=False)
375 if cached_avatar.stored == stored:
376 return False
378 if cached_avatar.stored.http_url:
379 setattr(space, attr, cached_avatar.stored)
380 return True
381 else:
382 warnings.warn(
383 "Space avatar can only be served via HTTP. "
384 "Consider using 'no-upload' or 'use-attachment-original-urls'.",
385 UserWarning,
386 )
387 return False
389 async def __get_stored_contact(self, legacy_id_str: str) -> Contact | None:
390 try:
391 contact: LegacyContact = await self.session.contacts.by_legacy_id(
392 legacy_id_str
393 )
394 except Exception as e: # noqa: BLE001
395 self.log.warning("Could not get contact: %r", e)
396 return None
397 return contact.stored
399 async def fetch_space_metadata(self, legacy_id: str) -> SpaceMetadata:
400 """
401 Fetch metadata associated to a space.
403 This is called once per slidge runtime. It should return metadata
404 associated to the space identified by its ``legacy_id``.
405 If there are updates to this metata, they should be communicated to
406 slidge by calling :func:`LegacyBookmarks.update_space_metadata`.
408 :param legacy_id: Identifier of the space.
410 :return: Metadata associated to the space.
411 """
412 raise NotImplementedError
414 async def update_spaces_info(self) -> None:
415 with self.orm() as orm:
416 spaces = self.session.xmpp.store.spaces.get_unupdated(
417 orm, self.session.user_pk
418 )
419 for space in spaces:
420 await self.update_space_if_needed(space)
422 async def space_legacy_id_to_node(self, legacy_id: str) -> str:
423 return legacy_id
425 async def space_node_to_legacy_id(self, node: str) -> str:
426 return node
428 async def update_space_metadata(
429 self,
430 legacy_id: str,
431 metadata: SpaceMetadata,
432 ) -> None:
433 """
434 Updates metadata associated to a space.
436 :param legacy_id: Identifier of the space.
437 :param name: Metadata associated to this space.
438 """
439 with self.orm(expire_on_commit=False) as orm:
440 space = self.session.xmpp.store.spaces.add_or_get(
441 orm,
442 self.session.user_pk,
443 str(legacy_id),
444 )
445 space, changes = await self.__update_space_metadata(space, metadata)
446 node = await self.space_legacy_id_to_node(legacy_id)
447 if changes - {"avatar", "banner"}:
448 self.xmpp.pubsub.broadcast_space_metadata(self.session, space, node)
449 item_changes = changes & {"avatar", "banner"}
450 if item_changes:
451 # avatar and banner are pubsub *items*, not part of the
452 # pubsub#metadata form; broadcasting their update has a
453 # different protocol
454 items_ids = [f"urn:xmpp:spaces:{attr}:metadata:0" for attr in item_changes]
455 self.xmpp.pubsub.broadcast_space(self.session, space, node, items_ids)
458LegacyBookmarks.muc_cls = LegacyMUC # type:ignore[misc]