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 

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

45 

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

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

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

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

50 

51 # Unified Moderation System 

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

53 

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

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

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

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

58 

59 def __repr__(self) -> str: 

60 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})" 

61 

62 

63class GroupChatRole(enum.Enum): 

64 admin = enum.auto() 

65 participant = enum.auto() 

66 

67 

68class GroupChatSubscription(Base, kw_only=True): 

69 """ 

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

71 """ 

72 

73 __tablename__ = "group_chat_subscriptions" 

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

75 

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

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

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

79 

80 # timezones should always be UTC 

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

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

83 

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

85 

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

87 

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

89 

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

91 muted_until: Mapped[datetime] = mapped_column( 

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

93 ) 

94 

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

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

97 

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

99 """ 

100 Returns (muted, muted_until) display values: 

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

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

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

104 """ 

105 if self.muted_until < now(): 

106 return (False, None) 

107 elif self.muted_until == DATETIME_INFINITY: 

108 return (True, None) 

109 else: 

110 return (True, self.muted_until) 

111 

112 @hybrid_property 

113 def is_muted(self) -> Any: 

114 return self.muted_until > func.now() 

115 

116 def __repr__(self) -> str: 

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

118 

119 

120class MessageType(enum.Enum): 

121 text = enum.auto() 

122 # e.g. 

123 # image = 

124 # emoji = 

125 # ... 

126 chat_created = enum.auto() 

127 chat_edited = enum.auto() 

128 user_invited = enum.auto() 

129 user_left = enum.auto() 

130 user_made_admin = enum.auto() 

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

132 host_request_status_changed = enum.auto() 

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

134 

135 

136class Message(Base, kw_only=True): 

137 """ 

138 A message. 

139 

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

141 """ 

142 

143 __tablename__ = "messages" 

144 __table_args__ = ( 

145 Index( 

146 "ix_messages_conversation_id_id_text_only", 

147 "conversation_id", 

148 "id", 

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

150 ), 

151 ) 

152 

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

154 

155 # which conversation the message belongs in 

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

157 

158 # the user that sent the message/command 

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

160 

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

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

163 

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

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

166 

167 # time sent, timezone should always be UTC 

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

169 

170 # the plain-text message text if not control 

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

172 

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

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

175 

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

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

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

179 

180 @property 

181 def is_normal_message(self) -> bool: 

182 """ 

183 There's only one normal type atm, text 

184 """ 

185 return self.message_type == MessageType.text 

186 

187 def __repr__(self) -> str: 

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