Coverage for app / backend / src / couchers / models / moderation.py: 95%

1""" 

2Unified Moderation System (UMS) models 

3 

4These models provide a flexible, generic moderation system that can be applied 

5to any moderatable content on the platform (host requests, discussions, events, etc.) 

6""" 

7 

8import enum 

9from datetime import datetime 

10from typing import TYPE_CHECKING 

11 

12from sqlalchemy import BigInteger, DateTime, Enum, ForeignKey, Index, String, func 

13from sqlalchemy.orm import Mapped, mapped_column, relationship 

14 

15from couchers.models.base import Base, moderation_seq 

16 

17if TYPE_CHECKING: 

18 from couchers.models.users import User 

19 

20 

21class ModerationVisibility(enum.Enum): 

22 # Only visible to moderators 

23 hidden = enum.auto() 

24 # Visible only to content author 

25 shadowed = enum.auto() 

26 # Visible to everyone, does not appear in listings 

27 unlisted = enum.auto() 

28 # Visible to everyone, appears in listings 

29 visible = enum.auto() 

30 

31 

32class ModerationTrigger(enum.Enum): 

33 """What triggered adding an item to the moderation queue""" 

34 

35 # New content requiring triage 

36 initial_review = enum.auto() 

37 # User reported/flagged content 

38 user_flag = enum.auto() 

39 # Automod flagged content 

40 machine_flag = enum.auto() 

41 # Moderator requested additional review 

42 moderator_review = enum.auto() 

43 

44 

45class ModerationAction(enum.Enum): 

46 """Types of moderation actions that can be taken""" 

47 

48 # Initial creation of moderation state 

49 create = enum.auto() 

50 # Approve content (make visible and listed) 

51 approve = enum.auto() 

52 # Hide content from everyone 

53 hide = enum.auto() 

54 # Flag for review 

55 flag = enum.auto() 

56 # Remove flag 

57 unflag = enum.auto() 

58 

59 

60class ModerationObjectType(enum.Enum): 

61 """Types of objects that can be moderated""" 

62 

63 host_request = enum.auto() 

64 group_chat = enum.auto() 

65 friend_request = enum.auto() 

66 event_occurrence = enum.auto() 

67 

68 

69class ModerationState(Base, kw_only=True): 

70 """ 

71 Moderation state for any moderatable object on the platform 

72 

73 This table tracks the visibility and listing state of content. 

74 Notifications are linked directly via the moderation_state_id FK on Notification. 

75 """ 

76 

77 __tablename__ = "moderation_states" 

78 

79 id: Mapped[int] = mapped_column( 

80 BigInteger, moderation_seq, primary_key=True, server_default=moderation_seq.next_value(), init=False 

81 ) 

82 

83 # Generic reference to the moderated object 

84 object_type: Mapped[ModerationObjectType] = mapped_column(Enum(ModerationObjectType)) 

85 object_id: Mapped[int] = mapped_column(BigInteger) 

86 

87 visibility: Mapped[ModerationVisibility] = mapped_column(Enum(ModerationVisibility)) 

88 

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

90 updated: Mapped[datetime] = mapped_column( 

91 DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), init=False 

92 ) 

93 

94 __table_args__ = ( 

95 # Each object can only have one moderation state 

96 Index("ix_moderation_states_object", object_type, object_id, unique=True), 

97 # Covering index for visibility filtering - enables index-only scans in where_moderated_content_visible 

98 Index("ix_moderation_states_id_visibility", id, visibility), 

99 ) 

100 

101 def __repr__(self) -> str: 

102 return f"ModerationState(id={self.id}, type={self.object_type}, object_id={self.object_id}, visibility={self.visibility})" 

103 

104 

105class ModerationQueueItem(Base, kw_only=True): 

106 """ 

107 Action items in the moderation queue 

108 

109 This table tracks what moderators need to review. Items remain in the queue 

110 until they are resolved (linked to a ModerationLog entry). 

111 """ 

112 

113 __tablename__ = "moderation_queue" 

114 

115 id: Mapped[int] = mapped_column( 

116 BigInteger, moderation_seq, primary_key=True, server_default=moderation_seq.next_value(), init=False 

117 ) 

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

119 

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

121 trigger: Mapped[ModerationTrigger] = mapped_column(Enum(ModerationTrigger)) 

122 reason: Mapped[str] = mapped_column(String) 

123 

124 # When resolved, this links to the log entry that resolved it 

125 resolved_by_log_id: Mapped[int | None] = mapped_column(ForeignKey("moderation_log.id"), index=True, default=None) 

126 

127 # Relationships 

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

129 

130 __table_args__ = ( 

131 # Fast lookup of unresolved items 

132 Index( 

133 "ix_moderation_queue_unresolved", 

134 moderation_state_id, 

135 time_created, 

136 postgresql_where=resolved_by_log_id.is_(None), 

137 ), 

138 ) 

139 

140 def __repr__(self) -> str: 

141 return ( 

142 f"ModerationQueueItem(id={self.id}, trigger={self.trigger}, resolved={self.resolved_by_log_id is not None})" 

143 ) 

144 

145 

146class ModerationLog(Base, kw_only=True): 

147 """ 

148 History of moderation actions 

149 

150 This table provides a complete audit trail of all moderation actions taken, 

151 including who performed the action and what changed. 

152 """ 

153 

154 __tablename__ = "moderation_log" 

155 

156 id: Mapped[int] = mapped_column( 

157 BigInteger, moderation_seq, primary_key=True, server_default=moderation_seq.next_value(), init=False 

158 ) 

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

160 

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

162 action: Mapped[ModerationAction] = mapped_column(Enum(ModerationAction)) 

163 moderator_user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) 

164 

165 # State changes (nullable - only include fields that changed) 

166 new_visibility: Mapped[ModerationVisibility | None] = mapped_column(Enum(ModerationVisibility), default=None) 

167 

168 # Explanation for the action 

169 reason: Mapped[str] = mapped_column(String) 

170 

171 # Relationships 

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

173 moderator: Mapped[User] = relationship(init=False) 

174 

175 __table_args__ = ( 

176 # Fast lookup of log entries for a given state, ordered by time 

177 Index("ix_moderation_log_state_time", moderation_state_id, time.desc()), 

178 ) 

179 

180 def __repr__(self) -> str: 

181 return f"ModerationLog(id={self.id}, state_id={self.moderation_state_id}, action={self.action}, moderator={self.moderator_user_id}, time={self.time})"