Coverage for slidge/group/bookmarks.py: 90%
231 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-10 04:45 +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 node = await self.space_legacy_id_to_node(muc.stored.space.legacy_id)
204 with self.orm() as orm:
205 orm.add(muc.stored)
206 self.xmpp.pubsub.broadcast_space(
207 self.session, muc.stored.space, node, [str(muc.jid)]
208 )
209 return muc
211 async def fill(self) -> None:
212 """
213 Establish a user's known groups.
215 This has to be overridden in plugins with group support and at the
216 minimum, this should ``await self.by_legacy_id(group_id)`` for all
217 the groups a user is part of.
219 Slidge internals will call this on successful :meth:`BaseSession.login`
221 """
222 if self.xmpp.GROUPS:
223 raise NotImplementedError(
224 "The plugin advertised support for groups but"
225 " LegacyBookmarks.fill() was not overridden."
226 )
228 async def remove(
229 self,
230 muc: AnyMUC,
231 reason: str = "You left this group from the official client.",
232 kick: bool = True,
233 ) -> None:
234 """
235 Delete everything about a specific group.
237 This should be called when the user leaves the group from the official
238 app.
240 :param muc: The MUC to remove.
241 :param reason: Optionally, a reason why this group was removed.
242 :param kick: Whether the user should be kicked from this group. Set this
243 to False in case you do this somewhere else in your code, eg, on
244 receiving the confirmation that the group was deleted.
245 """
246 if kick:
247 user_participant = await muc.get_user_participant()
248 user_participant.kick(reason)
249 with self.xmpp.store.session() as orm:
250 orm.add(muc.stored)
251 orm.refresh(muc.stored)
252 orm.delete(muc.stored)
253 orm.commit()
255 async def update_space_if_needed(self, space: Space) -> Space:
256 async with self.lock(("space", space.legacy_id)):
257 with self.orm() as orm:
258 orm.add(space)
259 orm.refresh(space)
260 if space.updated:
261 return space
262 orm.refresh(space, ["avatar", "banner"])
263 meta = await self.fetch_space_metadata(space.legacy_id)
264 return (await self.__update_space_metadata(space, meta))[0]
266 async def get_updated_space(
267 self,
268 legacy_id: str,
269 room_legacy_id_filter: Iterable[str] | None = None,
270 ) -> Space | None:
271 with self.orm() as orm:
272 space = self.xmpp.store.spaces.get_by_legacy_id(
273 orm,
274 self.session.user_pk,
275 legacy_id,
276 images=True,
277 room_legacy_id_filter=room_legacy_id_filter,
278 )
279 if space is None:
280 return None
281 if space.updated:
282 return space
283 async with self.lock(("space", space.legacy_id)):
284 meta = await self.fetch_space_metadata(space.legacy_id)
285 return (await self.__update_space_metadata(space, meta))[0]
287 async def __update_space_metadata(
288 self, space: Space, meta: SpaceMetadata
289 ) -> tuple[Space, set[str]]:
290 creator = (
291 await self.__get_stored_contact(meta.creator_legacy_id)
292 if meta.creator_legacy_id
293 else None
294 )
295 owners: list[Contact] = []
296 if not isinstance(meta.owner_legacy_ids, Unset):
297 for legacy_id in set(meta.owner_legacy_ids or ()):
298 if legacy_id == meta.creator_legacy_id:
299 # We don't want to fetch any contact twice here to avoid:
300 # Can't attach instance <Contact at xxx>; another instance [...] is already present in this session.
301 continue
302 owner = await self.__get_stored_contact(legacy_id)
303 if owner is not None:
304 owners.append(owner)
305 if (
306 creator is not None
307 and meta.creator_legacy_id
308 and (
309 isinstance(meta.owner_legacy_ids, Unset)
310 or meta.creator_legacy_id in meta.owner_legacy_ids
311 )
312 ):
313 owners.append(creator)
315 changed = set()
316 for attr in "avatar", "banner":
317 if await self.__update_space_image(space, attr, getattr(meta, attr)):
318 changed.add(attr)
319 with self.orm(expire_on_commit=False) as orm:
320 if creator is not None:
321 creator = orm.merge(creator)
322 owners = [orm.merge(owner) for owner in owners]
323 space = orm.merge(space)
324 if isinstance(meta.name, Unset):
325 if not space.name:
326 space.name = space.legacy_id
327 else:
328 space.name = meta.name or space.name or space.legacy_id
330 if not isinstance(meta.creator_legacy_id, Unset):
331 space.creator = creator
332 if not isinstance(meta.owner_legacy_ids, Unset):
333 space.owners = owners
334 if not isinstance(meta.description, Unset):
335 space.description = meta.description
336 if not isinstance(meta.member_count, Unset):
337 space.member_count = meta.member_count
339 changed |= modified_attributes(space)
340 self.log.debug("Changed space attributes: %s", changed)
342 if not space.updated:
343 # setting .updated to True means that fetch_space_metadata()
344 # should not be called for this specific space, but does not
345 # mean that we have to broadcast a pubsub#metadata change
346 space.updated = True
348 if orm.is_modified(space, include_collections=True):
349 orm.commit()
350 return space, changed # update_img, bool(changed)
352 async def __update_space_image(
353 self,
354 space: Space,
355 attr: Literal["avatar", "banner"],
356 new: Avatar | Unset | None,
357 ) -> bool:
358 stored: AvatarModel | None = getattr(space, attr)
360 if isinstance(new, Unset):
361 return False
363 if new is None:
364 setattr(space, attr, None)
365 return stored is not None
367 cached_avatar = await avatar_cache.get(new, session=self.session, convert=False)
369 if cached_avatar.stored == stored:
370 return False
372 if cached_avatar.stored.http_url:
373 setattr(space, attr, cached_avatar.stored)
374 return True
375 else:
376 warnings.warn(
377 "Space avatar can only be served via HTTP. "
378 "Consider using 'no-upload' or 'use-attachment-original-urls'.",
379 UserWarning,
380 )
381 return False
383 async def __get_stored_contact(self, legacy_id_str: str) -> Contact | None:
384 try:
385 contact: LegacyContact = await self.session.contacts.by_legacy_id(
386 legacy_id_str
387 )
388 except Exception as e: # noqa: BLE001
389 self.log.warning("Could not get contact: %r", e)
390 return None
391 return contact.stored
393 async def fetch_space_metadata(self, legacy_id: str) -> SpaceMetadata:
394 """
395 Fetch metadata associated to a space.
397 This is called once per slidge runtime. It should return metadata
398 associated to the space identified by its ``legacy_id``.
399 If there are updates to this metata, they should be communicated to
400 slidge by calling :func:`LegacyBookmarks.update_space_metadata`.
402 :param legacy_id: Identifier of the space.
404 :return: Metadata associated to the space.
405 """
406 raise NotImplementedError
408 async def update_spaces_info(self) -> None:
409 with self.orm() as orm:
410 spaces = self.session.xmpp.store.spaces.get_unupdated(
411 orm, self.session.user_pk
412 )
413 for space in spaces:
414 await self.update_space_if_needed(space)
416 async def space_legacy_id_to_node(self, legacy_id: str) -> str:
417 return legacy_id
419 async def space_node_to_legacy_id(self, node: str) -> str:
420 return node
422 async def update_space_metadata(
423 self,
424 legacy_id: str,
425 metadata: SpaceMetadata,
426 ) -> None:
427 """
428 Updates metadata associated to a space.
430 :param legacy_id: Identifier of the space.
431 :param name: Metadata associated to this space.
432 """
433 with self.orm(expire_on_commit=False) as orm:
434 space = self.session.xmpp.store.spaces.add_or_get(
435 orm,
436 self.session.user_pk,
437 str(legacy_id),
438 )
439 space, changes = await self.__update_space_metadata(space, metadata)
440 node = await self.space_legacy_id_to_node(legacy_id)
441 if changes - {"avatar", "banner"}:
442 self.xmpp.pubsub.broadcast_space_metadata(self.session, space, node)
443 item_changes = changes & {"avatar", "banner"}
444 if item_changes:
445 # avatar and banner are pubsub *items*, not part of the
446 # pubsub#metadata form; broadcasting their update has a
447 # different protocol
448 items_ids = [f"urn:xmpp:spaces:{attr}:metadata:0" for attr in item_changes]
449 self.xmpp.pubsub.broadcast_space(self.session, space, node, items_ids)
452LegacyBookmarks.muc_cls = LegacyMUC # type:ignore[misc]