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