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, String, func 

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.utils import now 

14 

15if TYPE_CHECKING: 

16 from couchers.models import ModerationState, User 

17 

18 

19class Conversation(Base, kw_only=True): 

20 """ 

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

22 """ 

23 

24 __tablename__ = "conversations" 

25 

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

27 # timezone should always be UTC 

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

29 

30 def __repr__(self) -> str: 

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

32 

33 

34class GroupChat(Base, kw_only=True): 

35 """ 

36 Group chat 

37 """ 

38 

39 __tablename__ = "group_chats" 

40 __moderation_author_column__ = "creator_id" 

41 

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

43 

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

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

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

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

48 

49 # Unified Moderation System 

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

51 

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

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

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

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

56 

57 def __repr__(self) -> str: 

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

59 

60 

61class GroupChatRole(enum.Enum): 

62 admin = enum.auto() 

63 participant = enum.auto() 

64 

65 

66class GroupChatSubscription(Base, kw_only=True): 

67 """ 

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

69 """ 

70 

71 __tablename__ = "group_chat_subscriptions" 

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

73 

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

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

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

77 

78 # timezones should always be UTC 

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

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

81 

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

83 

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

85 

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

87 

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

89 muted_until: Mapped[datetime] = mapped_column( 

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

91 ) 

92 

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

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

95 

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

97 """ 

98 Returns (muted, muted_until) display values: 

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

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

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

102 """ 

103 if self.muted_until < now(): 

104 return (False, None) 

105 elif self.muted_until == DATETIME_INFINITY: 

106 return (True, None) 

107 else: 

108 return (True, self.muted_until) 

109 

110 @hybrid_property 

111 def is_muted(self) -> Any: 

112 return self.muted_until > func.now() 

113 

114 def __repr__(self) -> str: 

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

116 

117 

118class MessageType(enum.Enum): 

119 text = enum.auto() 

120 # e.g. 

121 # image = 

122 # emoji = 

123 # ... 

124 chat_created = enum.auto() 

125 chat_edited = enum.auto() 

126 user_invited = enum.auto() 

127 user_left = enum.auto() 

128 user_made_admin = enum.auto() 

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

130 host_request_status_changed = enum.auto() 

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

132 

133 

134class Message(Base, kw_only=True): 

135 """ 

136 A message. 

137 

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

139 """ 

140 

141 __tablename__ = "messages" 

142 

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

144 

145 # which conversation the message belongs in 

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

147 

148 # the user that sent the message/command 

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

150 

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

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

153 

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

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

156 

157 # time sent, timezone should always be UTC 

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

159 

160 # the plain-text message text if not control 

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

162 

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

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

165 

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

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

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

169 

170 @property 

171 def is_normal_message(self) -> bool: 

172 """ 

173 There's only one normal type atm, text 

174 """ 

175 return self.message_type == MessageType.text 

176 

177 def __repr__(self) -> str: 

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