Coverage for slidge/room/archive.py: 97%

1import logging 

2import uuid 

3import warnings 

4from collections.abc import Collection, Iterator 

5from copy import copy 

6from datetime import UTC, datetime 

7 

8from slixmpp import Iq, Message 

9 

10from ..db.models import ArchivedMessage, ArchivedMessageSource, Room 

11from ..db.store import SlidgeStore 

12from ..util.types import AnyParticipant, HoleBound 

13 

14 

15class MessageArchive: 

16 def __init__(self, room: Room, store: SlidgeStore) -> None: 

17 self.room = room 

18 self.__rooms_store = store.rooms 

19 self.__store = store.mam 

20 

21 def add( 

22 self, 

23 msg: Message, 

24 participant: "AnyParticipant | None" = None, 

25 backfill: bool = False, 

26 legacy_msg_id: str | None = None, 

27 ) -> None: 

28 """ 

29 Add a message to the archive if it is deemed archivable 

30 

31 :param msg: 

32 :param participant: 

33 :param legacy_msg_id: 

34 """ 

35 if not archivable(msg): 

36 return 

37 new_msg = copy(msg) 

38 if participant and not participant._room.is_anonymous: 

39 new_msg["muc"]["role"] = participant.role or "participant" 

40 new_msg["muc"]["affiliation"] = participant.affiliation or "member" 

41 if participant.contact: 

42 new_msg["muc"]["jid"] = participant.contact.jid.bare 

43 elif participant.is_user: 

44 new_msg["muc"]["jid"] = participant.user_jid.bare 

45 elif participant.is_system: 

46 new_msg["muc"]["jid"] = participant._room.jid 

47 else: 

48 warnings.warn( 

49 f"No real JID for participant '{participant.nickname}' in '{self.room.name}'" 

50 ) 

51 new_msg["muc"]["jid"] = ( 

52 f"{uuid.uuid4()}@{participant.xmpp.boundjid.bare}" 

53 ) 

54 

55 if "replace" in msg: 

56 legacy_msg_id = None 

57 

58 with self.__store.session(expire_on_commit=False) as orm: 

59 if self.room.id is None: 

60 self.room = self.__rooms_store.get( 

61 orm, self.room.user_account_id, self.room.legacy_id 

62 ) 

63 self.__store.add_message( 

64 orm, 

65 self.room.id, 

66 new_msg, 

67 backfill, 

68 None if legacy_msg_id is None else str(legacy_msg_id), 

69 from_user=participant is not None and participant.is_user, 

70 ) 

71 orm.commit() 

72 

73 def __iter__(self) -> Iterator[ArchivedMessage]: 

74 return iter(self.get_all()) 

75 

76 @staticmethod 

77 def __to_bound(stored: ArchivedMessage) -> HoleBound: 

78 return HoleBound( 

79 stored.legacy_id, # type:ignore 

80 stored.timestamp.replace(tzinfo=UTC), 

81 ) 

82 

83 def get_hole_bounds(self) -> tuple[HoleBound | None, HoleBound | None]: 

84 with self.__store.session() as orm: 

85 most_recent = self.__store.get_most_recent_with_legacy_id(orm, self.room.id) 

86 if most_recent is None: 

87 return None, None 

88 if most_recent.source == ArchivedMessageSource.BACKFILL: 

89 # most recent = only backfill, fetch everything since last backfill 

90 return self.__to_bound(most_recent), None 

91 

92 most_recent_back_filled = self.__store.get_most_recent_with_legacy_id( 

93 orm, self.room.id, ArchivedMessageSource.BACKFILL 

94 ) 

95 if most_recent_back_filled is None: 

96 # group was never back-filled, fetch everything before first live 

97 least_recent_live = self.__store.get_first(orm, self.room.id, True) 

98 assert least_recent_live is not None 

99 return None, self.__to_bound(least_recent_live) 

100 

101 assert most_recent_back_filled.legacy_id is not None 

102 least_recent_live = self.__store.get_least_recent_with_legacy_id_after( 

103 orm, self.room.id, most_recent_back_filled.legacy_id 

104 ) 

105 assert least_recent_live is not None 

106 # this is a hole caused by slidge downtime 

107 return self.__to_bound(most_recent_back_filled), self.__to_bound( 

108 least_recent_live 

109 ) 

110 

111 def get_all( 

112 self, 

113 start_date: datetime | None = None, 

114 end_date: datetime | None = None, 

115 before_id: str | None = None, 

116 after_id: str | None = None, 

117 ids: Collection[str] = (), 

118 last_page_n: int | None = None, 

119 author_resource: str | None = None, 

120 flip: bool = False, 

121 ) -> Iterator[ArchivedMessage]: 

122 with self.__store.session() as orm: 

123 yield from self.__store.get_messages( 

124 orm, 

125 self.room.id, 

126 before_id=before_id, 

127 after_id=after_id, 

128 ids=ids, 

129 last_page_n=last_page_n, 

130 author_resource=author_resource, 

131 start_date=start_date, 

132 end_date=end_date, 

133 flip=flip, 

134 ) 

135 

136 async def send_metadata(self, iq: Iq) -> None: 

137 """ 

138 Send archive extent, as per the spec 

139 

140 :param iq: 

141 :return: 

142 """ 

143 reply = iq.reply() 

144 with self.__store.session() as orm: 

145 messages = self.__store.get_first_and_last(orm, self.room.id) 

146 if messages: 

147 for x, m in [("start", messages[0]), ("end", messages[-1])]: 

148 reply["mam_metadata"][x]["id"] = m.id 

149 reply["mam_metadata"][x]["timestamp"] = m.sent_on.replace(tzinfo=UTC) 

150 else: 

151 reply.enable("mam_metadata") 

152 reply.send() 

153 

154 

155def archivable(msg: Message) -> bool: 

156 """ 

157 Determine if a message stanza is worth archiving, ie, convey meaningful 

158 info 

159 

160 :param msg: 

161 :return: 

162 """ 

163 

164 if "no-store" in msg: 

165 return False 

166 

167 if "no-permanent-store" in msg: 

168 return False 

169 

170 if "store" in msg: 

171 return True 

172 

173 if msg["body"]: 

174 return True 

175 

176 if "retract" in msg: 

177 return True 

178 

179 if "reactions" in msg: 

180 return True 

181 

182 if "displayed" in msg: 

183 return True 

184 

185 return bool(msg["thread"] and msg["subject"]) 

186 

187 

188log = logging.getLogger(__name__)