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

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 dataclasses import dataclass 

10from datetime import datetime 

11from functools import cache 

12from typing import TYPE_CHECKING, Protocol 

13 

14from sqlalchemy import BigInteger, ColumnElement, DateTime, Enum, ForeignKey, Index, String, func 

15from sqlalchemy.orm import Mapped, mapped_column, relationship 

16 

17from couchers.models.base import Base, moderation_seq 

18 

19if TYPE_CHECKING: 

20 from couchers.models.users import User 

21 

22 

23class ModerationVisibility(enum.Enum): 

24 # Only visible to moderators 

25 hidden = enum.auto() 

26 # Visible only to content author 

27 shadowed = enum.auto() 

28 # Visible to everyone, does not appear in listings 

29 unlisted = enum.auto() 

30 # Visible to everyone, appears in listings 

31 visible = enum.auto() 

32 

33 

34class ModerationTrigger(enum.Enum): 

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

36 

37 # New content requiring triage 

38 initial_review = enum.auto() 

39 # User reported/flagged content 

40 user_flag = enum.auto() 

41 # Automod flagged content 

42 machine_flag = enum.auto() 

43 # Moderator requested additional review 

44 moderator_review = enum.auto() 

45 

46 

47class ModerationAction(enum.Enum): 

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

49 

50 # Initial creation of moderation state 

51 create = enum.auto() 

52 # Approve content (make visible and listed) 

53 approve = enum.auto() 

54 # Hide content from everyone 

55 hide = enum.auto() 

56 # Flag for review 

57 flag = enum.auto() 

58 # Remove flag 

59 unflag = enum.auto() 

60 # Bulk visibility change applied to every item authored by a user 

61 bulk_set_visibility = enum.auto() 

62 

63 

64class ModerationObjectType(enum.Enum): 

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

66 

67 host_request = enum.auto() 

68 group_chat = enum.auto() 

69 friend_request = enum.auto() 

70 event_occurrence = enum.auto() 

71 comment = enum.auto() 

72 reply = enum.auto() 

73 discussion = enum.auto() 

74 reference = enum.auto() 

75 

76 

77class ModerationState(Base, kw_only=True): 

78 """ 

79 Moderation state for any moderatable object on the platform 

80 

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

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

83 """ 

84 

85 __tablename__ = "moderation_states" 

86 

87 id: Mapped[int] = mapped_column( 

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

89 ) 

90 

91 # Generic reference to the moderated object 

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

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

94 

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

96 

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

98 updated: Mapped[datetime] = mapped_column( 

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

100 ) 

101 

102 __table_args__ = ( 

103 # Each object can only have one moderation state 

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

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

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

107 # Fast filtering by object type and visibility 

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

109 ) 

110 

111 def __repr__(self) -> str: 

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

113 

114 

115class ModerationQueueItem(Base, kw_only=True): 

116 """ 

117 Action items in the moderation queue 

118 

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

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

121 """ 

122 

123 __tablename__ = "moderation_queue" 

124 

125 id: Mapped[int] = mapped_column( 

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

127 ) 

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

129 

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

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

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

133 

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

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

136 

137 # Relationships 

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

139 

140 __table_args__ = ( 

141 # Fast lookup of unresolved items 

142 Index( 

143 "ix_moderation_queue_unresolved", 

144 moderation_state_id, 

145 time_created, 

146 postgresql_where=resolved_by_log_id.is_(None), 

147 ), 

148 ) 

149 

150 def __repr__(self) -> str: 

151 return ( 

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

153 ) 

154 

155 

156class ModerationLog(Base, kw_only=True): 

157 """ 

158 History of moderation actions 

159 

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

161 including who performed the action and what changed. 

162 """ 

163 

164 __tablename__ = "moderation_log" 

165 

166 id: Mapped[int] = mapped_column( 

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

168 ) 

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

170 

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

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

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

174 

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

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

177 

178 # Explanation for the action 

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

180 

181 # Relationships 

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

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

184 

185 __table_args__ = ( 

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

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

188 ) 

189 

190 def __repr__(self) -> str: 

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

192 

193 

194class ModeratedContent(Protocol): 

195 """A model governed by the UMS, identified by the moderation metadata it declares as class attributes.""" 

196 

197 __moderation_object_type__: ModerationObjectType 

198 __moderation_author_column__: str 

199 

200 

201@dataclass(frozen=True) 

202class ModeratedModel: 

203 """A model governed by the UMS, with its moderation metadata resolved.""" 

204 

205 object_type: ModerationObjectType 

206 model: type[ModeratedContent] 

207 author_column: ColumnElement[int] 

208 object_id_column: ColumnElement[int] 

209 moderation_state_id_column: ColumnElement[int] 

210 

211 

212@cache 

213def get_moderated_models() -> dict[ModerationObjectType, ModeratedModel]: 

214 """ 

215 Maps each ModerationObjectType to its model and resolved moderation metadata. 

216 

217 Discovered from every mapped model that declares __moderation_object_type__, so the moderation 

218 metadata stays on the models themselves rather than in a separate hand-maintained list. 

219 """ 

220 models: dict[ModerationObjectType, ModeratedModel] = {} 

221 for mapper in Base.registry.mappers: 

222 cls = mapper.class_ 

223 if not hasattr(cls, "__moderation_object_type__"): 

224 continue 

225 model: type[ModeratedContent] = cls 

226 models[model.__moderation_object_type__] = ModeratedModel( 

227 object_type=model.__moderation_object_type__, 

228 model=model, 

229 author_column=mapper.columns[model.__moderation_author_column__], 

230 object_id_column=mapper.primary_key[0], 

231 moderation_state_id_column=mapper.columns["moderation_state_id"], 

232 ) 

233 return models