Coverage for src/couchers/models/conversations.py: 95%

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 Mapped, backref, mapped_column, relationship 

8 

9from couchers.constants import DATETIME_INFINITY, DATETIME_MINUS_INFINITY 

10from couchers.models.base import Base 

11from couchers.models.host_requests import HostRequestStatus 

12from couchers.utils import now 

13 

14if TYPE_CHECKING: 

15 from couchers.models.moderation import ModerationState 

16 

17 

18class Conversation(Base): 

19 """ 

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

21 """ 

22 

23 __tablename__ = "conversations" 

24 

25 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

26 # timezone should always be UTC 

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

28 

29 def __repr__(self) -> str: 

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

31 

32 

33class GroupChat(Base): 

34 """ 

35 Group chat 

36 """ 

37 

38 __tablename__ = "group_chats" 

39 __moderation_author_column__ = "creator_id" 

40 

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

42 

43 title: Mapped[str | None] = mapped_column(String, nullable=True) 

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

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

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

47 

48 # Unified Moderation System 

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

50 

51 conversation = relationship("Conversation", backref="group_chat") 

52 creator = relationship("User", backref="created_group_chats") 

53 moderation_state: Mapped["ModerationState"] = relationship("ModerationState") 

54 

55 def __repr__(self) -> str: 

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

57 

58 

59class GroupChatRole(enum.Enum): 

60 admin = enum.auto() 

61 participant = enum.auto() 

62 

63 

64class GroupChatSubscription(Base): 

65 """ 

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

67 """ 

68 

69 __tablename__ = "group_chat_subscriptions" 

70 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

71 

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

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

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

75 

76 # timezones should always be UTC 

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

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

79 

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

81 

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

83 

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

85 muted_until: Mapped[datetime] = mapped_column( 

86 DateTime(timezone=True), server_default=DATETIME_MINUS_INFINITY.isoformat() 

87 ) 

88 

89 user = relationship("User", backref="group_chat_subscriptions") 

90 group_chat = relationship("GroupChat", backref=backref("subscriptions", lazy="dynamic")) 

91 

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

93 """ 

94 Returns (muted, muted_until) display values: 

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

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

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

98 """ 

99 if self.muted_until < now(): 

100 return (False, None) 

101 elif self.muted_until == DATETIME_INFINITY: 

102 return (True, None) 

103 else: 

104 return (True, self.muted_until) 

105 

106 @hybrid_property 

107 def is_muted(self) -> Any: 

108 return self.muted_until > func.now() 

109 

110 def __repr__(self) -> str: 

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

112 

113 

114class MessageType(enum.Enum): 

115 text = enum.auto() 

116 # e.g. 

117 # image = 

118 # emoji = 

119 # ... 

120 chat_created = enum.auto() 

121 chat_edited = enum.auto() 

122 user_invited = enum.auto() 

123 user_left = enum.auto() 

124 user_made_admin = enum.auto() 

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

126 host_request_status_changed = enum.auto() 

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

128 

129 

130class Message(Base): 

131 """ 

132 A message. 

133 

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

135 """ 

136 

137 __tablename__ = "messages" 

138 

139 id: Mapped[int] = mapped_column(BigInteger, primary_key=True) 

140 

141 # which conversation the message belongs in 

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

143 

144 # the user that sent the message/command 

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

146 

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

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

149 

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

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

152 

153 # time sent, timezone should always be UTC 

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

155 

156 # the plain-text message text if not control 

157 text: Mapped[str | None] = mapped_column(String, nullable=True) 

158 

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

160 host_request_status_target: Mapped[HostRequestStatus | None] = mapped_column(Enum(HostRequestStatus), nullable=True) 

161 

162 conversation = relationship("Conversation", backref="messages", order_by="Message.time.desc()") 

163 author = relationship("User", foreign_keys="Message.author_id") 

164 target = relationship("User", foreign_keys="Message.target_id") 

165 

166 @property 

167 def is_normal_message(self) -> bool: 

168 """ 

169 There's only one normal type atm, text 

170 """ 

171 return self.message_type == MessageType.text 

172 

173 def __repr__(self) -> str: 

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