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