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 

78 

79class ModerationState(Base, kw_only=True): 

80 """ 

81 Moderation state for any moderatable object on the platform 

82 

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

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

85 """ 

86 

87 __tablename__ = "moderation_states" 

88 

89 id: Mapped[int] = mapped_column( 

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

91 ) 

92 

93 # Generic reference to the moderated object 

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

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

96 

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

98 

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

100 updated: Mapped[datetime] = mapped_column( 

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

102 ) 

103 

104 __table_args__ = ( 

105 # Each object can only have one moderation state 

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

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

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

109 # Fast filtering by object type and visibility 

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

111 ) 

112 

113 def __repr__(self) -> str: 

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

115 

116 

117class ModerationQueueItem(Base, kw_only=True): 

118 """ 

119 Action items in the moderation queue 

120 

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

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

123 """ 

124 

125 __tablename__ = "moderation_queue" 

126 

127 id: Mapped[int] = mapped_column( 

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

129 ) 

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

131 

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

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

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

135 

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

137 

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

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

140 

141 # Relationships 

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

143 

144 __table_args__ = ( 

145 # Fast lookup of unresolved items 

146 Index( 

147 "ix_moderation_queue_unresolved", 

148 moderation_state_id, 

149 time_created, 

150 postgresql_where=resolved_by_log_id.is_(None), 

151 ), 

152 ) 

153 

154 def __repr__(self) -> str: 

155 return ( 

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

157 ) 

158 

159 

160class ModerationLog(Base, kw_only=True): 

161 """ 

162 History of moderation actions 

163 

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

165 including who performed the action and what changed. 

166 """ 

167 

168 __tablename__ = "moderation_log" 

169 

170 id: Mapped[int] = mapped_column( 

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

172 ) 

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

174 

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

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

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

178 

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

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

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

182 

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

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

185 

186 # Explanation for the action 

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

188 

189 # Relationships 

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

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

192 

193 __table_args__ = ( 

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

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

196 ) 

197 

198 def __repr__(self) -> str: 

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

200 

201 

202class ModeratedContent(Protocol): 

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

204 

205 __moderation_object_type__: ModerationObjectType 

206 __moderation_author_column__: str 

207 

208 

209@dataclass(frozen=True) 

210class ModeratedModel: 

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

212 

213 object_type: ModerationObjectType 

214 model: type[ModeratedContent] 

215 author_column: ColumnElement[int] 

216 object_id_column: ColumnElement[int] 

217 moderation_state_id_column: ColumnElement[int] 

218 

219 

220@cache 

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

222 """ 

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

224 

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

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

227 """ 

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

229 for mapper in Base.registry.mappers: 

230 cls = mapper.class_ 

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

232 continue 

233 model: type[ModeratedContent] = cls 

234 models[model.__moderation_object_type__] = ModeratedModel( 

235 object_type=model.__moderation_object_type__, 

236 model=model, 

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

238 object_id_column=mapper.primary_key[0], 

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

240 ) 

241 return models