Coverage for slidge/db/models.py: 95%

1import logging 

2import warnings 

3from collections.abc import Callable 

4from datetime import UTC, date, datetime 

5from enum import IntEnum 

6from functools import lru_cache 

7from pathlib import Path 

8from typing import Any, Literal 

9from xml.etree import ElementTree as ET 

10 

11import slixmpp.plugins.xep_0234.stanza 

12import slixmpp.plugins.xep_0264.stanza 

13import slixmpp.plugins.xep_0300.stanza 

14import slixmpp.plugins.xep_0372.stanza 

15import slixmpp.plugins.xep_0385.stanza 

16import slixmpp.plugins.xep_0447.stanza 

17import sqlalchemy as sa 

18import sqlalchemy.orm 

19from slixmpp import JID, Message 

20from slixmpp.plugins.xep_0292.stanza import VCard4 

21from slixmpp.plugins.xep_0297 import Forwarded 

22from slixmpp.plugins.xep_0444 import XEP_0444 

23from slixmpp.types import MucAffiliation, MucRole 

24from sqlalchemy import JSON, Column, ForeignKey, Index, Table, UniqueConstraint 

25from sqlalchemy.orm import Mapped, mapped_column, relationship, sessionmaker 

26 

27from ..core import config 

28from ..util.types import AvatarMetadata, ClientType, Hat, MucType 

29from .meta import Base, JSONSerializable, JSONSerializableTypes 

30 

31 

32class ArchivedMessageSource(IntEnum): 

33 """ 

34 Whether an archived message comes from ``LegacyMUC.backfill()`` or was received 

35 as a "live" message. 

36 """ 

37 

38 LIVE = 1 

39 BACKFILL = 2 

40 

41 

42class _JidMixin: 

43 jid_localpart: Mapped[str] 

44 

45 @property 

46 def jid(self) -> JID: 

47 return JID(f"{self.jid_localpart}@{config.JID}") 

48 

49 

50class _EnsureLoadedMixin: 

51 id: int | None 

52 

53 def ensure_loaded( 

54 self, 

55 group: str, 

56 session_maker: sessionmaker[Any] | Callable[[], sa.orm.Session], 

57 ) -> None: 

58 if self.id is None: 

59 return 

60 

61 if sa.orm.object_session(self) is not None: 

62 return 

63 

64 state = sa.inspect(self) 

65 if state is None: 

66 return 

67 

68 group_keys = _group_keys( 

69 self.__class__, # type:ignore[arg-type] 

70 group, 

71 ) 

72 

73 unloaded = group_keys & state.unloaded 

74 

75 if not unloaded: 

76 return 

77 

78 cls = self.__class__ 

79 stmt = sa.select(*(getattr(cls, key) for key in unloaded)).where( 

80 cls.id == self.id # type:ignore[arg-type] 

81 ) 

82 with session_maker() as session, session.no_autoflush: 

83 row = session.execute(stmt).one() 

84 

85 for key, value in zip(unloaded, row): 

86 sa.orm.attributes.set_committed_value(self, key, value) 

87 

88 def ensure_relation( 

89 self, 

90 relation: str, 

91 session_maker: sessionmaker[Any] | Callable[[], sa.orm.Session], 

92 ) -> None: 

93 if self.id is None: 

94 return 

95 

96 if sa.orm.object_session(self) is not None: 

97 return 

98 

99 if relation not in sa.orm.attributes.instance_state(self).unloaded: 

100 return 

101 

102 fk = getattr(self, f"{relation}_id") 

103 

104 if fk is None: 

105 sa.orm.attributes.set_committed_value(self, relation, None) 

106 return 

107 

108 mapper = sa.inspect(type(self)) 

109 assert mapper is not None 

110 model = mapper.relationships[relation].mapper.class_ 

111 

112 with session_maker() as session, session.no_autoflush: 

113 stored = session.execute( 

114 sa.select(model).where(model.id == fk) 

115 ).scalar_one_or_none() 

116 sa.orm.attributes.set_committed_value(self, relation, stored) 

117 

118 

119@lru_cache 

120def _group_keys(cls: type[Base], group: str) -> frozenset[str]: 

121 keys = frozenset(a.key for a in sa.inspect(cls).column_attrs if a.group == group) 

122 if not keys: 

123 raise ValueError(f"{cls.__name__} has no deferred group {group!r}") 

124 return keys 

125 

126 

127class GatewayUser(Base): 

128 """ 

129 A user, registered to the gateway component. 

130 """ 

131 

132 __tablename__ = "user_account" 

133 id: Mapped[int] = mapped_column(primary_key=True) 

134 jid: Mapped[JID] = mapped_column(unique=True) 

135 registration_date: Mapped[datetime] = mapped_column( 

136 sa.DateTime, server_default=sa.func.now() 

137 ) 

138 

139 legacy_module_data: Mapped[JSONSerializable] = mapped_column(default={}) 

140 """ 

141 Arbitrary non-relational data that legacy modules can use 

142 """ 

143 preferences: Mapped[JSONSerializable] = mapped_column(default={}) 

144 avatar_hash: Mapped[str | None] = mapped_column(default=None) 

145 """ 

146 Hash of the user's avatar, to avoid re-publishing the same avatar on the 

147 legacy network 

148 """ 

149 

150 contacts: Mapped[list["Contact"]] = relationship( 

151 back_populates="user", cascade="all, delete-orphan" 

152 ) 

153 rooms: Mapped[list["Room"]] = relationship( 

154 back_populates="user", cascade="all, delete-orphan" 

155 ) 

156 attachments: Mapped[list["Attachment"]] = relationship(cascade="all, delete-orphan") 

157 spaces: Mapped[list["Space"]] = relationship( 

158 back_populates="user", cascade="all, delete-orphan" 

159 ) 

160 

161 def __repr__(self) -> str: 

162 return f"User(id={self.id!r}, jid={self.jid!r})" 

163 

164 def get(self, field: str, default: str = "") -> JSONSerializableTypes: 

165 # """ 

166 # Get fields from the registration form (required to comply with slixmpp backend protocol) 

167 # 

168 # :param field: Name of the field 

169 # :param default: Default value to return if the field is not present 

170 # 

171 # :return: Value of the field 

172 # """ 

173 return self.legacy_module_data.get(field, default) 

174 

175 @property 

176 def registration_form(self) -> dict[str, Any]: 

177 # Kept for retrocompat, should be 

178 # FIXME: delete me 

179 warnings.warn( 

180 "GatewayUser.registration_form is deprecated.", DeprecationWarning 

181 ) 

182 return self.legacy_module_data 

183 

184 

185class Avatar(Base): 

186 """ 

187 Avatars of contacts, rooms and participants. 

188 

189 To comply with XEPs, we convert them all to PNG before storing them. 

190 """ 

191 

192 __tablename__ = "avatar" 

193 

194 id: Mapped[int] = mapped_column(primary_key=True) 

195 

196 hash: Mapped[str | None] = mapped_column(unique=True, nullable=True) 

197 height: Mapped[int] = mapped_column() 

198 width: Mapped[int] = mapped_column() 

199 

200 legacy_id: Mapped[str | None] = mapped_column(unique=True, nullable=True) 

201 

202 http_id: Mapped[str | None] = mapped_column(nullable=True) 

203 http_type: Mapped[str | None] = mapped_column(nullable=True) 

204 http_bytes: Mapped[int | None] = mapped_column(nullable=True) 

205 http_url: Mapped[str | None] = mapped_column(nullable=True) 

206 http_height: Mapped[int | None] = mapped_column(nullable=True) 

207 http_width: Mapped[int | None] = mapped_column(nullable=True) 

208 

209 # this is only used when avatars are available as HTTP URLs and do not 

210 # have a legacy_id 

211 url: Mapped[str | None] = mapped_column(unique=True, default=None) 

212 etag: Mapped[str | None] = mapped_column(default=None) 

213 last_modified: Mapped[str | None] = mapped_column(default=None) 

214 

215 contacts: Mapped[list["Contact"]] = relationship(back_populates="avatar") 

216 rooms: Mapped[list["Room"]] = relationship(back_populates="avatar") 

217 

218 def set_http_metadata(self, meta: AvatarMetadata | None) -> None: 

219 if meta is None: 

220 return 

221 self.http_id = meta.id 

222 self.http_bytes = meta.bytes 

223 self.http_type = f"image/{meta.type}" 

224 self.http_url = meta.url 

225 self.http_height = meta.height 

226 self.http_width = meta.width 

227 

228 

229space_owner_association = Table( 

230 "space_owner_association", 

231 Base.metadata, 

232 Column("space_id", ForeignKey("space.id"), primary_key=True), 

233 Column("contact_id", ForeignKey("contact.id"), primary_key=True), 

234) 

235 

236 

237class Contact(Base, _EnsureLoadedMixin, _JidMixin): 

238 """ 

239 Legacy contacts 

240 """ 

241 

242 __tablename__ = "contact" 

243 __table_args__ = ( 

244 UniqueConstraint( 

245 "user_account_id", "legacy_id", name="uq_contact_user_account_id_legacy_id" 

246 ), 

247 UniqueConstraint( 

248 "user_account_id", 

249 "jid_localpart", 

250 name="uq_contact_user_account_id_jid_localpart", 

251 ), 

252 ) 

253 

254 id: Mapped[int] = mapped_column(primary_key=True) 

255 user_account_id: Mapped[int] = mapped_column( 

256 ForeignKey("user_account.id"), deferred=True 

257 ) 

258 user: Mapped[GatewayUser] = relationship(lazy=True, back_populates="contacts") 

259 legacy_id: Mapped[str] = mapped_column(nullable=False) 

260 

261 jid_localpart: Mapped[str] = mapped_column(nullable=False) 

262 

263 avatar_id: Mapped[int | None] = mapped_column( 

264 ForeignKey("avatar.id"), nullable=True 

265 ) 

266 avatar: Mapped[Avatar | None] = relationship(back_populates="contacts") 

267 

268 nick: Mapped[str | None] = mapped_column(nullable=True) 

269 

270 cached_presence: Mapped[bool] = mapped_column( 

271 default=False, deferred=True, deferred_group="presence" 

272 ) 

273 last_seen: Mapped[datetime | None] = mapped_column( 

274 nullable=True, deferred=True, deferred_group="presence" 

275 ) 

276 ptype: Mapped[str | None] = mapped_column( 

277 nullable=True, deferred=True, deferred_group="presence" 

278 ) 

279 pstatus: Mapped[str | None] = mapped_column( 

280 nullable=True, deferred=True, deferred_group="presence" 

281 ) 

282 pshow: Mapped[str | None] = mapped_column( 

283 nullable=True, deferred=True, deferred_group="presence" 

284 ) 

285 

286 is_friend: Mapped[bool] = mapped_column(default=False) 

287 added_to_roster: Mapped[bool] = mapped_column(default=False) 

288 sent_order: Mapped[list["ContactSent"]] = relationship( 

289 back_populates="contact", cascade="all, delete-orphan" 

290 ) 

291 

292 extra_attributes: Mapped[JSONSerializable | None] = mapped_column( 

293 default=None, nullable=True 

294 ) 

295 updated: Mapped[bool] = mapped_column(default=False) 

296 

297 vcard_fetched: Mapped[bool] = mapped_column(default=False) 

298 # vcard fields 

299 full_name: Mapped[str | None] = mapped_column( 

300 nullable=True, deferred=True, deferred_group="vcard" 

301 ) 

302 birthday: Mapped[date | None] = mapped_column( 

303 nullable=True, deferred=True, deferred_group="vcard" 

304 ) 

305 phones: Mapped[list[str] | None] = mapped_column( 

306 JSON, nullable=True, deferred=True, deferred_group="vcard" 

307 ) 

308 note: Mapped[str | None] = mapped_column( 

309 nullable=True, deferred=True, deferred_group="vcard" 

310 ) 

311 url: Mapped[str | None] = mapped_column( 

312 nullable=True, deferred=True, deferred_group="vcard" 

313 ) 

314 email: Mapped[str | None] = mapped_column( 

315 nullable=True, deferred=True, deferred_group="vcard" 

316 ) 

317 country: Mapped[str | None] = mapped_column( 

318 nullable=True, deferred=True, deferred_group="vcard" 

319 ) 

320 locality: Mapped[str | None] = mapped_column( 

321 nullable=True, deferred=True, deferred_group="vcard" 

322 ) 

323 pronouns: Mapped[str | None] = mapped_column( 

324 nullable=True, deferred=True, deferred_group="vcard" 

325 ) 

326 

327 # No "merge" cascade: committing a Contact must never write its Participants. 

328 # Otherwise a stale, transient Participant left in this (loaded) collection 

329 # by the back_populates of Participant.contact gets INSERTed a second time on 

330 # the next Contact.commit(). 

331 participants: Mapped[list["Participant"]] = relationship( 

332 back_populates="contact", cascade="save-update" 

333 ) 

334 

335 client_type: Mapped[ClientType] = mapped_column(nullable=False, default="pc") 

336 

337 messages: Mapped[list["DirectMessages"]] = relationship( 

338 cascade="all, delete-orphan" 

339 ) 

340 threads: Mapped[list["DirectThreads"]] = relationship(cascade="all, delete-orphan") 

341 

342 spaces_created: Mapped[list["Space"]] = relationship( 

343 back_populates="creator", 

344 cascade="all, delete-orphan", 

345 ) 

346 spaces_owned: Mapped[list["Space"]] = relationship( 

347 back_populates="owners", 

348 secondary=space_owner_association, 

349 ) 

350 

351 last_sent_msg_legacy_id: Mapped[str | None] = mapped_column( 

352 deferred=True, deferred_group="last_msg" 

353 ) 

354 last_sent_msg_date: Mapped[datetime | None] = mapped_column( 

355 deferred=True, deferred_group="last_msg" 

356 ) 

357 

358 def vcard(self, domain: str) -> VCard4: 

359 vcard = VCard4() 

360 vcard.add_impp(f"xmpp:{self.jid_localpart}@{domain}") 

361 

362 if n := self.nick: 

363 vcard.add_nickname(n) 

364 if self.full_name: 

365 vcard["full_name"] = self.full_name 

366 elif n: 

367 vcard["full_name"] = n 

368 

369 if self.birthday: 

370 vcard["birthday"] = self.birthday 

371 

372 if self.note: 

373 vcard.add_note(self.note) 

374 if self.url: 

375 vcard.add_url(self.url) 

376 if self.email: 

377 vcard.add_email(self.email) 

378 

379 if self.phones: 

380 for p in self.phones: 

381 vcard.add_tel(p) 

382 if self.country or self.locality: 

383 vcard.add_address(self.country, self.locality) 

384 if self.pronouns: 

385 vcard["pronouns"]["text"] = self.pronouns 

386 

387 return vcard 

388 

389 

390class ContactSent(Base): 

391 """ 

392 Keep track of XMPP msg ids sent by a specific contact for networks in which 

393 all messages need to be marked as read. 

394 

395 (XMPP displayed markers convey a "read up to here" semantic.) 

396 """ 

397 

398 __tablename__ = "contact_sent" 

399 __table_args__ = ( 

400 UniqueConstraint( 

401 "contact_id", "msg_id", name="uq_contact_sent_contact_id_msg_id" 

402 ), 

403 ) 

404 

405 id: Mapped[int] = mapped_column(primary_key=True) 

406 contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id")) 

407 contact: Mapped[Contact] = relationship(back_populates="sent_order") 

408 msg_id: Mapped[str] = mapped_column() 

409 

410 

411class Room(Base, _EnsureLoadedMixin, _JidMixin): 

412 """ 

413 Legacy room 

414 """ 

415 

416 __table_args__ = ( 

417 UniqueConstraint( 

418 "user_account_id", "legacy_id", name="uq_room_user_account_id_legacy_id" 

419 ), 

420 UniqueConstraint( 

421 "user_account_id", 

422 "jid_localpart", 

423 name="uq_room_user_account_id_jid_localpart", 

424 ), 

425 ) 

426 

427 __tablename__ = "room" 

428 id: Mapped[int] = mapped_column(primary_key=True) 

429 user_account_id: Mapped[int] = mapped_column( 

430 ForeignKey("user_account.id"), deferred=True 

431 ) 

432 user: Mapped[GatewayUser] = relationship(lazy=True, back_populates="rooms") 

433 legacy_id: Mapped[str] = mapped_column(nullable=False) 

434 

435 jid_localpart: Mapped[str] = mapped_column(nullable=False) 

436 

437 avatar_id: Mapped[int | None] = mapped_column( 

438 ForeignKey("avatar.id"), nullable=True 

439 ) 

440 avatar: Mapped[Avatar | None] = relationship(back_populates="rooms") 

441 

442 name: Mapped[str | None] = mapped_column(nullable=True) 

443 

444 description: Mapped[str | None] = mapped_column( 

445 nullable=True, deferred=True, deferred_group="disco" 

446 ) 

447 n_participants: Mapped[int | None] = mapped_column( 

448 default=None, deferred=True, deferred_group="disco" 

449 ) 

450 

451 subject: Mapped[str | None] = mapped_column( 

452 nullable=True, deferred=True, deferred_group="subject" 

453 ) 

454 subject_date: Mapped[datetime | None] = mapped_column( 

455 nullable=True, deferred=True, deferred_group="subject" 

456 ) 

457 subject_setter: Mapped[str | None] = mapped_column( 

458 nullable=True, deferred=True, deferred_group="subject" 

459 ) 

460 

461 muc_type: Mapped[MucType] = mapped_column(default=MucType.CHANNEL) 

462 

463 user_nick: Mapped[str | None] = mapped_column() 

464 user_resources: Mapped[str | None] = mapped_column(nullable=True) 

465 

466 participants_filled: Mapped[bool] = mapped_column(default=False) 

467 history_filled: Mapped[bool] = mapped_column(default=False) 

468 

469 extra_attributes: Mapped[JSONSerializable | None] = mapped_column(default=None) 

470 updated: Mapped[bool] = mapped_column(default=False) 

471 

472 participants: Mapped[list["Participant"]] = relationship( 

473 back_populates="room", 

474 primaryjoin="Participant.room_id == Room.id", 

475 cascade="all, delete-orphan", 

476 ) 

477 

478 archive: Mapped[list["ArchivedMessage"]] = relationship( 

479 cascade="all, delete-orphan" 

480 ) 

481 

482 space_id: Mapped[int | None] = mapped_column(ForeignKey("space.id"), nullable=True) 

483 space: Mapped["Space | None"] = relationship(back_populates="rooms") 

484 

485 def load_participants( 

486 self, orm: sessionmaker[Any] | Callable[[], sa.orm.Session], with_presence: bool 

487 ) -> None: 

488 if self.id is None: 

489 return 

490 

491 stmt = ( 

492 _STMT_ROOM_LOAD_PARTICIPANTS_WITH_PRESENCE 

493 if with_presence 

494 else _STMT_ROOM_LOAD_PARTICIPANTS 

495 ).where(Room.id == self.id) 

496 

497 if (existing_session := sa.orm.object_session(self)) is None: 

498 with orm() as session, session.no_autoflush: 

499 session.add(self) 

500 session.execute(stmt).all() 

501 else: 

502 existing_session.execute(stmt).all() 

503 

504 

505class ArchivedMessage(Base): 

506 """ 

507 Messages of rooms, that we store to act as a MAM server 

508 """ 

509 

510 __tablename__ = "mam" 

511 __table_args__ = ( 

512 UniqueConstraint("room_id", "stanza_id", name="uq_mam_room_id_stanza_id"), 

513 Index("ix_mam_room_id_legacy_id", "room_id", "legacy_id"), 

514 Index("ix_mam_room_id_timestamp", "room_id", "timestamp"), 

515 ) 

516 

517 id: Mapped[int] = mapped_column(primary_key=True) 

518 room_id: Mapped[int] = mapped_column(ForeignKey("room.id"), nullable=False) 

519 room: Mapped[Room] = relationship(lazy=True, back_populates="archive") 

520 

521 stanza_id: Mapped[str] = mapped_column(nullable=False) 

522 origin_id: Mapped[str | None] = mapped_column(nullable=True) 

523 timestamp: Mapped[datetime] = mapped_column(nullable=False) 

524 

525 author_resource: Mapped[str | None] = mapped_column(nullable=True) 

526 author_affiliation: Mapped[str | None] = mapped_column() 

527 author_role: Mapped[str | None] = mapped_column() 

528 author_jid_localpart: Mapped[str | None] = mapped_column() 

529 

530 from_user: Mapped[bool] = mapped_column() 

531 

532 occupant_id: Mapped[str | None] = mapped_column() 

533 

534 source: Mapped[ArchivedMessageSource] = mapped_column(nullable=False) 

535 legacy_id: Mapped[str | None] = mapped_column(nullable=True) 

536 thread_legacy_id: Mapped[str | None] = mapped_column(nullable=True) 

537 

538 # payload is for XML we don't have dedicated columns for 

539 payload: Mapped[str | None] = mapped_column(nullable=True) 

540 

541 displayed_id: Mapped[str | None] = mapped_column(nullable=True) 

542 reaction_id: Mapped[str | None] = mapped_column(nullable=True) 

543 reaction_emojis: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) 

544 body: Mapped[str | None] = mapped_column(nullable=True) 

545 thread: Mapped[str | None] = mapped_column(nullable=True) 

546 

547 displayed_by_user: Mapped[bool] = mapped_column(default=False, nullable=True) 

548 

549 def stanza( 

550 self, ns: str = "jabber:client", user_jid: JID | str | None = None 

551 ) -> Message: 

552 msg = Message( 

553 stype="groupchat", 

554 sfrom=f"{self.room.jid}/{self.author_resource}" 

555 if self.author_resource 

556 else self.room.jid, 

557 ) 

558 if ns != "jabber:client": 

559 msg.xml.tag = f"{{{ns}}}message" 

560 msg.namespace = ns 

561 

562 msg["stanza_id"]["id"] = self.stanza_id 

563 msg["stanza_id"]["by"] = self.room.jid 

564 

565 if self.origin_id: 

566 msg["id"] = msg["origin_id"]["id"] = self.origin_id 

567 else: 

568 msg["id"] = self.stanza_id 

569 

570 if self.occupant_id: 

571 msg["occupant-id"]["id"] = self.occupant_id 

572 elif self.author_jid_localpart: 

573 msg["occupant-id"]["id"] = f"{self.author_jid_localpart}@{config.JID}" 

574 else: 

575 log.warning("Cannot set an occupant ID for this archived message") 

576 

577 if self.author_affiliation: 

578 msg["muc"]["affiliation"] = self.author_affiliation 

579 if self.author_role: 

580 msg["muc"]["role"] = self.author_role 

581 if self.from_user: 

582 msg["muc"]["jid"] = str(user_jid) 

583 elif self.author_jid_localpart: 

584 msg["muc"]["jid"] = JID(f"{self.author_jid_localpart}@{config.JID}") 

585 

586 if self.payload: 

587 for child in ET.fromstring(f'<x xmlns="{ns}">{self.payload}</x>'): 

588 msg.append(child) 

589 if self.displayed_id: 

590 msg["displayed"]["id"] = self.displayed_id 

591 if self.reaction_id: 

592 XEP_0444.set_reactions(msg, self.reaction_id, self.reaction_emojis or []) 

593 if self.body: 

594 msg["body"] = self.body 

595 if self.thread: 

596 msg["thread"] = self.thread 

597 

598 return msg 

599 

600 def forwarded(self, user_jid: JID | str) -> Forwarded: 

601 forwarded = Forwarded() 

602 forwarded["delay"]["stamp"] = self.timestamp.replace(tzinfo=UTC) 

603 forwarded.append(self.stanza(user_jid=user_jid)) 

604 return forwarded 

605 

606 

607class _LegacyToXmppIdsBase: 

608 """ 

609 XMPP-client generated IDs, and mapping to the corresponding legacy IDs. 

610 

611 A single legacy ID can map to several XMPP ids. 

612 """ 

613 

614 id: Mapped[int] = mapped_column(primary_key=True) 

615 legacy_id: Mapped[str] = mapped_column(nullable=False) 

616 xmpp_id: Mapped[str] = mapped_column(nullable=False) 

617 

618 

619class DirectMessages(_LegacyToXmppIdsBase, Base): 

620 __tablename__ = "direct_msg" 

621 __table_args__ = (Index("ix_direct_msg_legacy_id", "legacy_id", "foreign_key"),) 

622 foreign_key: Mapped[int] = mapped_column(ForeignKey("contact.id"), nullable=False) 

623 

624 

625class DirectThreads(_LegacyToXmppIdsBase, Base): 

626 __tablename__ = "direct_thread" 

627 __table_args__ = (Index("ix_direct_direct_thread_id", "legacy_id", "foreign_key"),) 

628 foreign_key: Mapped[int] = mapped_column(ForeignKey("contact.id"), nullable=False) 

629 

630 

631class Attachment(Base): 

632 """ 

633 Legacy attachments 

634 """ 

635 

636 __tablename__ = "attachment" 

637 __table_args__ = ( 

638 UniqueConstraint( 

639 "user_account_id", 

640 "legacy_file_id", 

641 name="uq_attachment_user_account_id_legacy_file_id", 

642 ), 

643 ) 

644 

645 id: Mapped[int] = mapped_column(primary_key=True) 

646 user_account_id: Mapped[int] = mapped_column( 

647 ForeignKey("user_account.id"), deferred=True 

648 ) 

649 user: Mapped[GatewayUser] = relationship(back_populates="attachments") 

650 

651 legacy_file_id: Mapped[str | None] = mapped_column(index=True, nullable=True) 

652 url: Mapped[str] = mapped_column(index=True, nullable=False) 

653 

654 size: Mapped[int | None] = mapped_column(nullable=True) 

655 sha256: Mapped[str | None] = mapped_column(nullable=True) 

656 

657 thumbhash: Mapped[str | None] = mapped_column(nullable=True) 

658 width: Mapped[int | None] = mapped_column(nullable=True) 

659 height: Mapped[int | None] = mapped_column(nullable=True) 

660 content_type: Mapped[str | None] = mapped_column(nullable=True) 

661 desc: Mapped[str | None] = mapped_column(nullable=True) 

662 name: Mapped[str | None] = mapped_column(nullable=True) 

663 date: Mapped[datetime | None] = mapped_column(nullable=True) 

664 disposition: Mapped[Literal["inline", "attachment"] | None] = mapped_column( 

665 nullable=True 

666 ) 

667 

668 def _hash(self) -> slixmpp.plugins.xep_0300.stanza.Hash | None: 

669 if self.sha256 is None: 

670 return None 

671 hash_elem = slixmpp.plugins.xep_0300.stanza.Hash() 

672 hash_elem["algo"] = "sha-256" 

673 hash_elem["value"] = self.sha256 

674 return hash_elem 

675 

676 def _thumbnail(self) -> slixmpp.plugins.xep_0264.stanza.Thumbnail | None: 

677 if self.thumbhash is None: 

678 return None 

679 thumbnail = slixmpp.plugins.xep_0264.stanza.Thumbnail() 

680 thumbnail["width"] = self.width 

681 thumbnail["height"] = self.height 

682 thumbnail["media-type"] = "image/thumbhash" 

683 thumbnail["uri"] = "data:image/thumbhash;base64," + self.thumbhash 

684 return thumbnail 

685 

686 def sims(self) -> slixmpp.plugins.xep_0372.stanza.Reference: 

687 ref = slixmpp.plugins.xep_0372.stanza.Reference() 

688 ref["type"] = "data" 

689 ref["uri"] = self.url 

690 

691 sims = slixmpp.plugins.xep_0385.stanza.Sims() 

692 sims["sources"].append(ref) 

693 sims.enable("file") 

694 if self.content_type is not None: 

695 sims["file"]["media-type"] = self.content_type 

696 if self.desc is not None: 

697 sims["file"]["desc"] = self.desc 

698 if self.size is not None: 

699 sims["file"]["size"] = self.size 

700 if self.name is not None: 

701 sims["file"]["name"] = self.name 

702 if self.date is not None: 

703 sims["file"]["date"] = self.date 

704 if self.thumbhash is not None: 

705 sims["file"].append(self._thumbnail()) 

706 if self.sha256 is not None: 

707 sims["file"].append(self._hash()) 

708 

709 wrapper = slixmpp.plugins.xep_0372.stanza.Reference() 

710 wrapper["type"] = "data" 

711 wrapper.append(sims) 

712 

713 return wrapper 

714