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 

66 

67class ModerationState(Base, kw_only=True): 

68 """ 

69 Moderation state for any moderatable object on the platform 

70 

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

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

73 """ 

74 

75 __tablename__ = "moderation_states" 

76 

77 id: Mapped[int] = mapped_column( 

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

79 ) 

80 

81 # Generic reference to the moderated object 

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

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

84 

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

86 

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

88 updated: Mapped[datetime] = mapped_column( 

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

90 ) 

91 

92 __table_args__ = ( 

93 # Each object can only have one moderation state 

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

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

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

97 ) 

98 

99 def __repr__(self) -> str: 

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

101 

102 

103class ModerationQueueItem(Base, kw_only=True): 

104 """ 

105 Action items in the moderation queue 

106 

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

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

109 """ 

110 

111 __tablename__ = "moderation_queue" 

112 

113 id: Mapped[int] = mapped_column( 

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

115 ) 

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

117 

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

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

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

121 

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

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

124 

125 # Relationships 

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

127 

128 __table_args__ = ( 

129 # Fast lookup of unresolved items 

130 Index( 

131 "ix_moderation_queue_unresolved", 

132 moderation_state_id, 

133 time_created, 

134 postgresql_where=resolved_by_log_id.is_(None), 

135 ), 

136 ) 

137 

138 def __repr__(self) -> str: 

139 return ( 

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

141 ) 

142 

143 

144class ModerationLog(Base, kw_only=True): 

145 """ 

146 History of moderation actions 

147 

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

149 including who performed the action and what changed. 

150 """ 

151 

152 __tablename__ = "moderation_log" 

153 

154 id: Mapped[int] = mapped_column( 

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

156 ) 

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

158 

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

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

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

162 

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

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

165 

166 # Explanation for the action 

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

168 

169 # Relationships 

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

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

172 

173 __table_args__ = ( 

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

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

176 ) 

177 

178 def __repr__(self) -> str: 

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