Coverage for 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): 

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() 

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()) 

88 updated: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) 

89 

90 __table_args__ = ( 

91 # Each object can only have one moderation state 

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

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

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

95 ) 

96 

97 def __repr__(self) -> str: 

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

99 

100 

101class ModerationQueueItem(Base): 

102 """ 

103 Action items in the moderation queue 

104 

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

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

107 """ 

108 

109 __tablename__ = "moderation_queue" 

110 

111 id: Mapped[int] = mapped_column( 

112 BigInteger, moderation_seq, primary_key=True, server_default=moderation_seq.next_value() 

113 ) 

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

115 

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

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

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

119 

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

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

122 

123 # Relationships 

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

125 

126 __table_args__ = ( 

127 # Fast lookup of unresolved items 

128 Index( 

129 "ix_moderation_queue_unresolved", 

130 moderation_state_id, 

131 time_created, 

132 postgresql_where=resolved_by_log_id.is_(None), 

133 ), 

134 ) 

135 

136 def __repr__(self) -> str: 

137 return ( 

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

139 ) 

140 

141 

142class ModerationLog(Base): 

143 """ 

144 History of moderation actions 

145 

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

147 including who performed the action and what changed. 

148 """ 

149 

150 __tablename__ = "moderation_log" 

151 

152 id: Mapped[int] = mapped_column( 

153 BigInteger, moderation_seq, primary_key=True, server_default=moderation_seq.next_value() 

154 ) 

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

156 

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

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

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

160 

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

162 new_visibility: Mapped[ModerationVisibility | None] = mapped_column(Enum(ModerationVisibility), nullable=True) 

163 

164 # Explanation for the action 

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

166 

167 # Relationships 

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

169 moderator: Mapped["User"] = relationship("User") 

170 

171 __table_args__ = ( 

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

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

174 ) 

175 

176 def __repr__(self) -> str: 

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