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, Integer, 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 # Change a flag's priority 

61 set_priority = enum.auto() 

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

63 bulk_set_visibility = enum.auto() 

64 

65 

66class ModerationObjectType(enum.Enum): 

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

68 

69 host_request = enum.auto() 

70 group_chat = enum.auto() 

71 friend_request = enum.auto() 

72 event_occurrence = enum.auto() 

73 comment = enum.auto() 

74 reply = enum.auto() 

75 discussion = enum.auto() 

76 reference = enum.auto() 

77 public_trip = enum.auto() 

78 

79 

80class ModerationState(Base, kw_only=True): 

81 """ 

82 Moderation state for any moderatable object on the platform 

83 

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

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

86 """ 

87 

88 __tablename__ = "moderation_states" 

89 

90 id: Mapped[int] = mapped_column( 

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

92 ) 

93 

94 # Generic reference to the moderated object 

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

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

97 

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

99 

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

101 updated: Mapped[datetime] = mapped_column( 

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

103 ) 

104 

105 __table_args__ = ( 

106 # Each object can only have one moderation state 

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

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

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

110 # Fast filtering by object type and visibility 

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

112 ) 

113 

114 def __repr__(self) -> str: 

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

116 

117 

118class ModerationQueueItem(Base, kw_only=True): 

119 """ 

120 Action items in the moderation queue 

121 

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

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

124 """ 

125 

126 __tablename__ = "moderation_queue" 

127 

128 id: Mapped[int] = mapped_column( 

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

130 ) 

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

132 

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

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

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

136 

137 priority: Mapped[int] = mapped_column(Integer, nullable=False, server_default="0", default=0) 

138 

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

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

141 

142 # Relationships 

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

144 

145 __table_args__ = ( 

146 # Fast lookup of unresolved items 

147 Index( 

148 "ix_moderation_queue_unresolved", 

149 moderation_state_id, 

150 time_created, 

151 postgresql_where=resolved_by_log_id.is_(None), 

152 ), 

153 ) 

154 

155 def __repr__(self) -> str: 

156 return ( 

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

158 ) 

159 

160 

161class ModerationLog(Base, kw_only=True): 

162 """ 

163 History of moderation actions 

164 

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

166 including who performed the action and what changed. 

167 """ 

168 

169 __tablename__ = "moderation_log" 

170 

171 id: Mapped[int] = mapped_column( 

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

173 ) 

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

175 

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

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

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

179 

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

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

182 new_priority: Mapped[int | None] = mapped_column(Integer, default=None) 

183 

184 # The queue item (flag) this action concerned, for flag-level actions 

185 queue_item_id: Mapped[int | None] = mapped_column(ForeignKey("moderation_queue.id"), index=True, default=None) 

186 

187 # Explanation for the action 

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

189 

190 # Relationships 

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

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

193 

194 __table_args__ = ( 

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

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

197 ) 

198 

199 def __repr__(self) -> str: 

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

201 

202 

203class ModeratedContent(Protocol): 

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

205 

206 __moderation_object_type__: ModerationObjectType 

207 __moderation_author_column__: str 

208 

209 

210@dataclass(frozen=True) 

211class ModeratedModel: 

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

213 

214 object_type: ModerationObjectType 

215 model: type[ModeratedContent] 

216 author_column: ColumnElement[int] 

217 object_id_column: ColumnElement[int] 

218 moderation_state_id_column: ColumnElement[int] 

219 

220 

221@cache 

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

223 """ 

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

225 

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

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

228 """ 

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

230 for mapper in Base.registry.mappers: 

231 cls = mapper.class_ 

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

233 continue 

234 model: type[ModeratedContent] = cls 

235 models[model.__moderation_object_type__] = ModeratedModel( 

236 object_type=model.__moderation_object_type__, 

237 model=model, 

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

239 object_id_column=mapper.primary_key[0], 

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

241 ) 

242 return models