Coverage for app/backend/src/couchers/models/conversations.py: 96%

1import enum 

2from datetime import datetime 

3from typing import TYPE_CHECKING, Any 

4 

5from sqlalchemy import BigInteger, Boolean, DateTime, Enum, ForeignKey, Index, String, func, text 

6from sqlalchemy.ext.hybrid import hybrid_property 

7from sqlalchemy.orm import DynamicMapped, Mapped, mapped_column, relationship 

8from sqlalchemy.sql import expression 

9 

10from couchers.constants import DATETIME_INFINITY, DATETIME_MINUS_INFINITY 

11from couchers.models.base import Base 

12from couchers.models.host_requests import HostRequestStatus 

13from couchers.models.moderation import ModerationObjectType 

14from couchers.utils import now 

15 

16if TYPE_CHECKING: 

17 from couchers.models import ModerationState, User 

18 

19 

20class Conversation(Base, kw_only=True): 

21 """ 

22 Conversation brings together the different types of message/conversation types 

23 """ 

24 

25 __tablename__ = "conversations" 

26 

27 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False) 

28 # timezone should always be UTC 

29 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

30 

31 def __repr__(self) -> str: 

32 return f"Conversation(id={self.id}, created={self.created})" 

33 

34 

35class GroupChat(Base, kw_only=True): 

36 """ 

37 Group chat or direct message. 

38 """ 

39 

40 __tablename__ = "group_chats" 

41 __moderation_author_column__ = "creator_id" 

42 __moderation_object_type__ = ModerationObjectType.group_chat 

43 __moderation_has_own_visibility_mechanism__ = False 

44 

45 conversation_id: Mapped[int] = mapped_column("id", ForeignKey("conversations.id"), primary_key=True) 

46 

47 title: Mapped[str | None] = mapped_column(String, default=None) 

48 only_admins_invite: Mapped[bool] = mapped_column(Boolean, default=True) 

49 creator_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

50 is_dm: Mapped[bool] = mapped_column(Boolean) 

51 

52 # Unified Moderation System 

53 moderation_state_id: Mapped[int] = mapped_column(ForeignKey("moderation_states.id"), index=True) 

54 

55 conversation: Mapped[Conversation] = relationship(init=False, backref="group_chat") 

56 creator: Mapped[User] = relationship(init=False, backref="created_group_chats") 

57 moderation_state: Mapped[ModerationState] = relationship(init=False) 

58 subscriptions: DynamicMapped[GroupChatSubscription] = relationship(init=False, lazy="dynamic") 

59 

60 def __repr__(self) -> str: 

61 return f"GroupChat(conversation={self.conversation}, title={self.title or 'None'}, only_admins_invite={self.only_admins_invite}, creator={self.creator}, is_dm={self.is_dm})" 

62 

63 

64class GroupChatRole(enum.Enum): 

65 admin = enum.auto() 

66 participant = enum.auto() 

67 

68 

69class GroupChatSubscription(Base, kw_only=True): 

70 """ 

71 The recipient of a thread and information about when they joined/left/etc. 

72 """ 

73 

74 __tablename__ = "group_chat_subscriptions" 

75 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False) 

76 

77 # TODO: DB constraint on only one user+group_chat combo at a given time 

78 user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

79 group_chat_id: Mapped[int] = mapped_column(ForeignKey("group_chats.id"), index=True) 

80 

81 # timezones should always be UTC 

82 joined: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

83 left: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None) 

84 

85 role: Mapped[GroupChatRole] = mapped_column(Enum(GroupChatRole)) 

86 

87 last_seen_message_id: Mapped[int] = mapped_column(BigInteger, default=0) 

88 

89 is_archived: Mapped[bool] = mapped_column(Boolean, server_default=expression.false(), init=False) 

90 

91 # when this chat is muted until, DATETIME_INFINITY for "forever" 

92 muted_until: Mapped[datetime] = mapped_column( 

93 DateTime(timezone=True), server_default=DATETIME_MINUS_INFINITY.isoformat(), init=False 

94 ) 

95 

96 user: Mapped[User] = relationship(init=False, backref="group_chat_subscriptions") 

97 group_chat: Mapped[GroupChat] = relationship(init=False, back_populates="subscriptions") 

98 

99 def muted_display(self) -> tuple[bool, datetime | None]: 

100 """ 

101 Returns (muted, muted_until) display values: 

102 1. If not muted, returns (False, None) 

103 2. If muted forever, returns (True, None) 

104 3. If muted until a given datetime returns (True, dt) 

105 """ 

106 if self.muted_until < now(): 

107 return (False, None) 

108 elif self.muted_until == DATETIME_INFINITY: 

109 return (True, None) 

110 else: 

111 return (True, self.muted_until) 

112 

113 @hybrid_property 

114 def is_muted(self) -> Any: 

115 return self.muted_until > func.now() 

116 

117 def __repr__(self) -> str: 

118 return f"GroupChatSubscription(id={self.id}, user={self.user}, joined={self.joined}, left={self.left}, role={self.role}, group_chat={self.group_chat})" 

119 

120 

121class MessageType(enum.Enum): 

122 text = enum.auto() 

123 # e.g. 

124 # image = 

125 # emoji = 

126 # ... 

127 chat_created = enum.auto() 

128 chat_edited = enum.auto() 

129 user_invited = enum.auto() 

130 user_left = enum.auto() 

131 user_made_admin = enum.auto() 

132 user_removed_admin = enum.auto() # RemoveGroupChatAdmin: remove admin permission from a user in group chat 

133 host_request_status_changed = enum.auto() 

134 user_removed = enum.auto() # user is removed from group chat by amdin RemoveGroupChatUser 

135 

136 

137class Message(Base, kw_only=True): 

138 """ 

139 A message. 

140 

141 If message_type = text, then the message is a normal text message, otherwise, it's a special control message. 

142 """ 

143 

144 __tablename__ = "messages" 

145 __table_args__ = ( 

146 # serves "is there anything in this conversation newer than X", which the unread badge counts ask 

147 # once per host request. can't be the partial index below: those counts include control messages 

148 Index("ix_messages_conversation_id_id_time", "conversation_id", "id", postgresql_include=["time"]), 

149 # send_request_notifications only wakes users for text messages; time is included so it can decide 

150 # the "older than 5 minutes" cutoff without leaving the index 

151 Index( 

152 "ix_messages_conversation_id_id_time_text_only", 

153 "conversation_id", 

154 "id", 

155 postgresql_include=["time"], 

156 postgresql_where=text("message_type = 'text'"), 

157 ), 

158 ) 

159 

160 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False) 

161 

162 # which conversation the message belongs in 

163 conversation_id: Mapped[int] = mapped_column(ForeignKey("conversations.id")) 

164 

165 # the user that sent the message/command 

166 author_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

167 

168 # the message type, "text" is a text message, otherwise a "control message" 

169 message_type: Mapped[MessageType] = mapped_column(Enum(MessageType)) 

170 

171 # the target if a control message and requires target, e.g. if inviting a user, the user invited is the target 

172 target_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), index=True, default=None) 

173 

174 # time sent, timezone should always be UTC 

175 time: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

176 

177 # the plain-text message text if not control 

178 text: Mapped[str | None] = mapped_column(String, default=None) 

179 

180 # the new host request status if the message type is host_request_status_changed 

181 host_request_status_target: Mapped[HostRequestStatus | None] = mapped_column(Enum(HostRequestStatus), default=None) 

182 

183 conversation: Mapped[Conversation] = relationship(init=False, backref="messages", order_by="Message.time.desc()") 

184 author: Mapped[User] = relationship(init=False, foreign_keys="Message.author_id") 

185 target: Mapped[User | None] = relationship(init=False, foreign_keys="Message.target_id") 

186 

187 @property 

188 def is_normal_message(self) -> bool: 

189 """ 

190 There's only one normal type atm, text 

191 """ 

192 return self.message_type == MessageType.text 

193 

194 def __repr__(self) -> str: 

195 return f"Message(id={self.id}, time={self.time}, text={self.text}, author={self.author}, conversation={self.conversation})"