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 # Bulk visibility change applied to every item authored by a user 

59 bulk_set_visibility = enum.auto() 

60 

61 

62class ModerationObjectType(enum.Enum): 

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

64 

65 host_request = enum.auto() 

66 group_chat = enum.auto() 

67 friend_request = enum.auto() 

68 event_occurrence = enum.auto() 

69 

70 

71class ModerationState(Base, kw_only=True): 

72 """ 

73 Moderation state for any moderatable object on the platform 

74 

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

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

77 """ 

78 

79 __tablename__ = "moderation_states" 

80 

81 id: Mapped[int] = mapped_column( 

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

83 ) 

84 

85 # Generic reference to the moderated object 

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

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

88 

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

90 

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

92 updated: Mapped[datetime] = mapped_column( 

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

94 ) 

95 

96 __table_args__ = ( 

97 # Each object can only have one moderation state 

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

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

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

101 # Fast filtering by object type and visibility 

102 Index("ix_moderation_states_type_visibility", object_type, visibility), 

103 ) 

104 

105 def __repr__(self) -> str: 

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

107 

108 

109class ModerationQueueItem(Base, kw_only=True): 

110 """ 

111 Action items in the moderation queue 

112 

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

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

115 """ 

116 

117 __tablename__ = "moderation_queue" 

118 

119 id: Mapped[int] = mapped_column( 

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

121 ) 

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

123 

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

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

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

127 

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

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

130 

131 # Relationships 

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

133 

134 __table_args__ = ( 

135 # Fast lookup of unresolved items 

136 Index( 

137 "ix_moderation_queue_unresolved", 

138 moderation_state_id, 

139 time_created, 

140 postgresql_where=resolved_by_log_id.is_(None), 

141 ), 

142 ) 

143 

144 def __repr__(self) -> str: 

145 return ( 

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

147 ) 

148 

149 

150class ModerationLog(Base, kw_only=True): 

151 """ 

152 History of moderation actions 

153 

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

155 including who performed the action and what changed. 

156 """ 

157 

158 __tablename__ = "moderation_log" 

159 

160 id: Mapped[int] = mapped_column( 

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

162 ) 

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

164 

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

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

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

168 

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

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

171 

172 # Explanation for the action 

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

174 

175 # Relationships 

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

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

178 

179 __table_args__ = ( 

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

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

182 ) 

183 

184 def __repr__(self) -> str: 

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