Coverage for app/backend/src/couchers/models/moderation.py: 97%
92 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1"""
2Unified Moderation System (UMS) models
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"""
8import enum
9from dataclasses import dataclass
10from datetime import datetime
11from functools import cache
12from typing import TYPE_CHECKING, Any, Protocol
14from sqlalchemy import (
15 BigInteger,
16 CheckConstraint,
17 ColumnElement,
18 DateTime,
19 Enum,
20 ForeignKey,
21 Index,
22 Integer,
23 String,
24 func,
25)
26from sqlalchemy.dialects.postgresql import JSONB
27from sqlalchemy.orm import Mapped, mapped_column, relationship
29from couchers.models.base import Base, moderation_seq
31if TYPE_CHECKING:
32 from couchers.models.users import User
35class ModerationVisibility(enum.Enum):
36 # Only visible to moderators
37 hidden = enum.auto()
38 # Visible only to content author
39 shadowed = enum.auto()
40 # Visible to everyone, does not appear in listings
41 unlisted = enum.auto()
42 # Visible to everyone, appears in listings
43 visible = enum.auto()
46class ModerationTrigger(enum.Enum):
47 """What triggered adding an item to the moderation queue"""
49 # New content requiring triage
50 initial_review = enum.auto()
51 # User reported/flagged content
52 user_flag = enum.auto()
53 # Automod flagged content
54 machine_flag = enum.auto()
55 # Moderator requested additional review
56 moderator_review = enum.auto()
59class ModerationAction(enum.Enum):
60 """Types of moderation actions that can be taken"""
62 # Initial creation of moderation state
63 create = enum.auto()
64 # Approve content (make visible and listed)
65 approve = enum.auto()
66 # Hide content from everyone
67 hide = enum.auto()
68 # Flag for review
69 flag = enum.auto()
70 # Remove flag
71 unflag = enum.auto()
72 # Change a flag's priority
73 set_priority = enum.auto()
74 # Bulk visibility change applied to every item authored by a user
75 bulk_set_visibility = enum.auto()
78class ModerationObjectType(enum.Enum):
79 """Types of objects that can be moderated"""
81 host_request = enum.auto()
82 group_chat = enum.auto()
83 friend_request = enum.auto()
84 event_occurrence = enum.auto()
85 comment = enum.auto()
86 reply = enum.auto()
87 discussion = enum.auto()
88 reference = enum.auto()
89 public_trip = enum.auto()
90 user = enum.auto()
93class ModerationState(Base, kw_only=True):
94 """
95 Moderation state for any moderatable object on the platform
97 This table tracks the visibility and listing state of content.
98 Notifications are linked directly via the moderation_state_id FK on Notification.
99 """
101 __tablename__ = "moderation_states"
103 id: Mapped[int] = mapped_column(
104 BigInteger, moderation_seq, primary_key=True, server_default=moderation_seq.next_value(), init=False
105 )
107 # Generic reference to the moderated object
108 object_type: Mapped[ModerationObjectType] = mapped_column(Enum(ModerationObjectType))
109 object_id: Mapped[int] = mapped_column(BigInteger)
111 visibility: Mapped[ModerationVisibility | None] = mapped_column(Enum(ModerationVisibility))
113 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False)
114 updated: Mapped[datetime] = mapped_column(
115 DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), init=False
116 )
118 __table_args__ = (
119 # Each object can only have one moderation state
120 Index("ix_moderation_states_object", object_type, object_id, unique=True),
121 # Covering index for visibility filtering - enables index-only scans in where_moderated_content_visible
122 Index("ix_moderation_states_id_visibility", id, visibility),
123 # Fast filtering by object type and visibility
124 Index("ix_moderation_states_type_visibility", object_type, visibility),
125 CheckConstraint(
126 "(object_type = 'user') = (visibility IS NULL)",
127 name="check_visibility_null_iff_own_mechanism",
128 ),
129 )
131 def __repr__(self) -> str:
132 return f"ModerationState(id={self.id}, type={self.object_type}, object_id={self.object_id}, visibility={self.visibility})"
135class ModerationQueueItem(Base, kw_only=True):
136 """
137 Action items in the moderation queue
139 This table tracks what moderators need to review. Items remain in the queue
140 until they are resolved (linked to a ModerationLog entry).
141 """
143 __tablename__ = "moderation_queue"
145 id: Mapped[int] = mapped_column(
146 BigInteger, moderation_seq, primary_key=True, server_default=moderation_seq.next_value(), init=False
147 )
148 moderation_state_id: Mapped[int] = mapped_column(ForeignKey("moderation_states.id"), index=True)
150 time_created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False)
151 trigger: Mapped[ModerationTrigger] = mapped_column(Enum(ModerationTrigger))
152 reason: Mapped[str] = mapped_column(String)
154 priority: Mapped[int] = mapped_column(Integer, nullable=False, server_default="0", default=0)
156 data: Mapped[Any | None] = mapped_column(JSONB(none_as_null=True), default=None)
158 # When resolved, this links to the log entry that resolved it
159 resolved_by_log_id: Mapped[int | None] = mapped_column(ForeignKey("moderation_log.id"), index=True, default=None)
161 # Relationships
162 moderation_state: Mapped[ModerationState] = relationship(init=False)
164 __table_args__ = (
165 # Fast lookup of unresolved items
166 Index(
167 "ix_moderation_queue_unresolved",
168 moderation_state_id,
169 time_created,
170 postgresql_where=resolved_by_log_id.is_(None),
171 ),
172 )
174 def __repr__(self) -> str:
175 return (
176 f"ModerationQueueItem(id={self.id}, trigger={self.trigger}, resolved={self.resolved_by_log_id is not None})"
177 )
180class ModerationLog(Base, kw_only=True):
181 """
182 History of moderation actions
184 This table provides a complete audit trail of all moderation actions taken,
185 including who performed the action and what changed.
186 """
188 __tablename__ = "moderation_log"
190 id: Mapped[int] = mapped_column(
191 BigInteger, moderation_seq, primary_key=True, server_default=moderation_seq.next_value(), init=False
192 )
193 moderation_state_id: Mapped[int] = mapped_column(ForeignKey("moderation_states.id"), index=True)
195 time: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False)
196 action: Mapped[ModerationAction] = mapped_column(Enum(ModerationAction))
197 moderator_user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
199 # State changes (nullable - only include fields that changed)
200 new_visibility: Mapped[ModerationVisibility | None] = mapped_column(Enum(ModerationVisibility), default=None)
201 new_priority: Mapped[int | None] = mapped_column(Integer, default=None)
203 # The queue item (flag) this action concerned, for flag-level actions
204 queue_item_id: Mapped[int | None] = mapped_column(ForeignKey("moderation_queue.id"), index=True, default=None)
206 # Explanation for the action
207 reason: Mapped[str] = mapped_column(String)
209 # Relationships
210 moderation_state: Mapped[ModerationState] = relationship(init=False)
211 moderator: Mapped[User] = relationship(init=False)
213 __table_args__ = (
214 # Fast lookup of log entries for a given state, ordered by time
215 Index("ix_moderation_log_state_time", moderation_state_id, time.desc()),
216 )
218 def __repr__(self) -> str:
219 return f"ModerationLog(id={self.id}, state_id={self.moderation_state_id}, action={self.action}, moderator={self.moderator_user_id}, time={self.time})"
222class ModeratedContent(Protocol):
223 """A model governed by the UMS, identified by the moderation metadata it declares as class attributes."""
225 __moderation_object_type__: ModerationObjectType
226 __moderation_author_column__: str
227 __moderation_has_own_visibility_mechanism__: bool
230@dataclass(frozen=True)
231class ModeratedModel:
232 """A model governed by the UMS, with its moderation metadata resolved."""
234 object_type: ModerationObjectType
235 model: type[ModeratedContent]
236 author_column: ColumnElement[int]
237 object_id_column: ColumnElement[int]
238 moderation_state_id_column: ColumnElement[int]
239 # Visibility not determined by the moderation state, they have some other visibility logic
240 has_own_visibility_mechanism: bool
243@cache
244def get_moderated_models() -> dict[ModerationObjectType, ModeratedModel]:
245 """
246 Maps each ModerationObjectType to its model and resolved moderation metadata.
248 Discovered from every mapped model that declares __moderation_object_type__, so the moderation
249 metadata stays on the models themselves rather than in a separate hand-maintained list.
251 Ordered by model class name. registry.mappers is a frozenset, so iterating it orders Mapper objects by id(), which
252 varies per process; callers build one OR branch per entry, so without sorting the same logical query is emitted
253 with its branches in a different order in every process. That splits it across pg_stat_statements entries.
254 """
255 models: dict[ModerationObjectType, ModeratedModel] = {}
256 for mapper in sorted(Base.registry.mappers, key=lambda m: m.class_.__name__):
257 cls = mapper.class_
258 if not hasattr(cls, "__moderation_object_type__"):
259 continue
260 model: type[ModeratedContent] = cls
261 models[model.__moderation_object_type__] = ModeratedModel(
262 object_type=model.__moderation_object_type__,
263 model=model,
264 author_column=mapper.columns[model.__moderation_author_column__],
265 object_id_column=mapper.primary_key[0],
266 moderation_state_id_column=mapper.columns["moderation_state_id"],
267 has_own_visibility_mechanism=model.__moderation_has_own_visibility_mechanism__,
268 )
269 return models