Coverage for slidge/group/bookmarks.py: 91%
171 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-09 03:59 +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 Generic[LegacyMUCType],
26 EscapeMixin,
27 NamedLockMixin,
28 SubclassableOnce,
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(user_account_id=self.session.user_pk, jid=jid)
96 .one_or_none()
97 )
98 if stored is None:
99 stored = Room(
100 user_account_id=self.session.user_pk,
101 jid=jid,
102 legacy_id=legacy_id,
103 )
104 return await self.__update_if_needed(stored)
106 def by_jid_only_if_exists(self, jid: JID) -> LegacyMUCType | None:
107 with self.xmpp.store.session(expire_on_commit=False) as orm:
108 stored = (
109 orm.query(Room).filter_by(user=self.session.user, jid=jid).one_or_none()
110 )
111 if stored is not None and stored.updated:
112 return self.from_store(stored)
113 return None
115 @overload
116 async def by_legacy_id(self, legacy_id: str) -> "LegacyMUCType": ...
118 @overload
119 async def by_legacy_id(
120 self, legacy_id: str, create: Literal[False]
121 ) -> "LegacyMUCType | None": ...
123 @overload
124 async def by_legacy_id(
125 self, legacy_id: str, create: Literal[True]
126 ) -> "LegacyMUCType": ...
128 async def by_legacy_id(
129 self, legacy_id: str, create: bool = False
130 ) -> LegacyMUCType | None:
131 async with self.lock(("legacy_id", legacy_id)):
132 local = await self.legacy_id_to_jid_local_part(legacy_id)
133 jid = JID(f"{local}@{self.xmpp.boundjid}")
134 if self.get_lock(("bare", jid.bare)):
135 self.session.log.debug("Already updating %s via by_jid()", jid)
136 if create:
137 return await self.by_jid(jid)
138 else:
139 return self.by_jid_only_if_exists(jid)
141 with self.xmpp.store.session() as orm:
142 stored = (
143 orm.query(Room)
144 .filter_by(
145 user_account_id=self.session.user_pk,
146 legacy_id=str(legacy_id),
147 )
148 .one_or_none()
149 )
150 if stored is None:
151 stored = Room(
152 user_account_id=self.session.user_pk,
153 jid=jid,
154 legacy_id=str(legacy_id),
155 )
156 return await self.__update_if_needed(stored)
158 async def __update_if_needed(self, stored: Room) -> LegacyMUCType:
159 muc = self.from_store(stored)
160 if muc.stored.updated:
161 return muc
163 with muc.updating_info():
164 try:
165 await muc.update_info()
166 except NotImplementedError:
167 pass
168 except XMPPError:
169 raise
170 except Exception as e:
171 raise XMPPError("internal-server-error", str(e))
172 muc.archive.room = muc.stored
173 if self.ready.done() and muc.stored.space_id:
174 with self.orm() as orm:
175 orm.add(muc.stored)
176 self.xmpp.pubsub.broadcast_space(self.session, muc.stored.space)
177 return muc
179 async def fill(self) -> None:
180 """
181 Establish a user's known groups.
183 This has to be overridden in plugins with group support and at the
184 minimum, this should ``await self.by_legacy_id(group_id)`` for all
185 the groups a user is part of.
187 Slidge internals will call this on successful :meth:`BaseSession.login`
189 """
190 if self.xmpp.GROUPS:
191 raise NotImplementedError(
192 "The plugin advertised support for groups but"
193 " LegacyBookmarks.fill() was not overridden."
194 )
196 async def remove(
197 self,
198 muc: AnyMUC,
199 reason: str = "You left this group from the official client.",
200 kick: bool = True,
201 ) -> None:
202 """
203 Delete everything about a specific group.
205 This should be called when the user leaves the group from the official
206 app.
208 :param muc: The MUC to remove.
209 :param reason: Optionally, a reason why this group was removed.
210 :param kick: Whether the user should be kicked from this group. Set this
211 to False in case you do this somewhere else in your code, eg, on
212 receiving the confirmation that the group was deleted.
213 """
214 if kick:
215 user_participant = await muc.get_user_participant()
216 user_participant.kick(reason)
217 with self.xmpp.store.session() as orm:
218 orm.add(muc.stored)
219 orm.refresh(muc.stored)
220 orm.delete(muc.stored)
221 orm.commit()
223 async def update_space_if_needed(self, space: Space) -> Space:
224 async with self.lock(("space", space.legacy_id)):
225 with self.orm() as orm:
226 orm.add(space)
227 orm.refresh(space)
228 if space.updated:
229 return space
230 meta: SpaceMetadata = await self.fetch_space_metadata(space.legacy_id)
231 return await self.__update_space_metadata(space, meta)
233 async def __update_space_metadata(self, space: Space, meta: SpaceMetadata) -> Space:
234 creator = (
235 await self.__get_stored_contact(meta.creator_legacy_id)
236 if meta.creator_legacy_id
237 else None
238 )
239 owners: list[Contact] = []
240 for legacy_id in set(meta.owner_legacy_ids):
241 if legacy_id == meta.creator_legacy_id:
242 # We don't want to fetch any contact twice here to avoid:
243 # Can't attach instance <Contact at xxx>; another instance [...] is already present in this session.
244 continue
245 owner = await self.__get_stored_contact(legacy_id)
246 if owner is not None:
247 owners.append(owner)
248 if (
249 creator is not None
250 and meta.creator_legacy_id
251 and meta.creator_legacy_id in meta.owner_legacy_ids
252 ):
253 owners.append(creator)
254 with self.orm(expire_on_commit=False) as orm:
255 if creator is not None:
256 creator = orm.merge(creator)
257 owners = [orm.merge(owner) for owner in owners]
258 space = orm.merge(space)
259 space.name = meta.name or space.legacy_id
260 space.creator = creator
261 space.owners = owners
262 space.description = meta.description
263 space.member_count = meta.member_count
264 space.updated = True
265 orm.commit()
266 return space
268 async def __get_stored_contact(self, legacy_id_str: str) -> Contact | None:
269 try:
270 contact: LegacyContact = await self.session.contacts.by_legacy_id(
271 legacy_id_str
272 )
273 except Exception as e:
274 self.log.warning("Could not get contact: %r", e)
275 return None
276 return contact.stored
278 @abc.abstractmethod
279 async def fetch_space_metadata(self, legacy_id: str) -> SpaceMetadata:
280 """
281 Fetch metadata associated to a space.
283 This is called once per slidge runtime. It should return metadata
284 associated to the space identified by its ``legacy_id``.
285 If there are updates to this metata, they should be communicated to
286 slidge by calling :func:`LegacyBookmarks.update_space_metadata`.
288 :param legacy_id: Identifier of the space.
290 :return: Metadata associated to the space.
291 """
292 raise NotImplementedError
294 async def update_spaces_info(self) -> None:
295 with self.orm() as orm:
296 spaces = self.session.xmpp.store.spaces.get_unupdated(
297 orm, self.session.user_pk
298 )
299 for space in spaces:
300 await self.update_space_if_needed(space)
302 async def space_legacy_id_to_node(self, legacy_id: str) -> str:
303 return str(legacy_id)
305 async def space_node_to_legacy_id(self, node: str) -> str:
306 return node
308 async def update_space_metadata(
309 self,
310 legacy_id: str,
311 metadata: SpaceMetadata,
312 ) -> None:
313 """
314 Updates metadata associated to a space.
316 :param legacy_id: Identifier of the space.
317 :param name: Metadata associated to this space.
318 """
319 with self.orm(expire_on_commit=False) as orm:
320 space = self.session.xmpp.store.spaces.add_or_get(
321 orm,
322 self.session.user_pk,
323 str(legacy_id),
324 )
325 space = await self.__update_space_metadata(space, metadata)
326 self.xmpp.pubsub.broadcast_space_metadata(
327 self.session, space, await self.space_legacy_id_to_node(legacy_id)
328 )