Coverage for app/backend/src/couchers/servicers/threads.py: 86%
185 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-21 21:52 +0000
1import logging
3import grpc
4import sqlalchemy.exc
5from google.protobuf import empty_pb2
6from sqlalchemy import exists, select
7from sqlalchemy.orm import Session
8from sqlalchemy.sql import func
10from couchers.context import CouchersContext, make_background_user_context, make_notification_user_context
11from couchers.db import session_scope
12from couchers.helpers.completed_profile import has_completed_profile
13from couchers.jobs.enqueue import queue_job
14from couchers.models import (
15 Comment,
16 Discussion,
17 Event,
18 EventOccurrence,
19 ModerationObjectType,
20 Reply,
21 Thread,
22 User,
23)
24from couchers.models.discussions import CommentVersion, ContentChangeType, ReplyVersion
25from couchers.models.notifications import NotificationTopicAction
26from couchers.moderation.utils import create_moderation
27from couchers.notifications.notify import notify
28from couchers.proto import notification_data_pb2, threads_pb2, threads_pb2_grpc
29from couchers.proto.internal import jobs_pb2
30from couchers.servicers.api import user_model_to_pb
31from couchers.servicers.blocking import is_not_visible
32from couchers.sql import where_moderated_content_visible, where_users_column_visible
33from couchers.utils import Timestamp_from_datetime, now
35logger = logging.getLogger(__name__)
38# Since the API exposes a single ID space regardless of nesting level,
39# we construct the API id by appending the nesting level to the
40# database ID.
43def pack_thread_id(database_id: int, depth: int) -> int:
44 return database_id * 10 + depth
47def unpack_thread_id(thread_id: int) -> tuple[int, int]:
48 """Returns (database_id, depth) tuple."""
49 return divmod(thread_id, 10)
52def total_num_responses(session: Session, context: CouchersContext, database_id: int) -> int:
53 """Return the total number of visible, non-deleted comments and replies to the thread with
54 database id database_id.
55 """
56 comments = where_moderated_content_visible(
57 where_users_column_visible(
58 select(func.count())
59 .select_from(Comment)
60 .where(Comment.thread_id == database_id)
61 .where(Comment.deleted == None),
62 context,
63 Comment.author_user_id,
64 ),
65 context,
66 Comment,
67 is_list_operation=True,
68 )
69 replies = where_moderated_content_visible(
70 where_users_column_visible(
71 select(func.count())
72 .select_from(Reply)
73 .join(Comment, Comment.id == Reply.comment_id)
74 .where(Comment.thread_id == database_id)
75 .where(Reply.deleted == None),
76 context,
77 Reply.author_user_id,
78 ),
79 context,
80 Reply,
81 is_list_operation=True,
82 )
83 return session.execute(comments).scalar_one() + session.execute(replies).scalar_one()
86def thread_to_pb(session: Session, context: CouchersContext, database_id: int) -> threads_pb2.Thread:
87 return threads_pb2.Thread(
88 thread_id=pack_thread_id(database_id, 0),
89 num_responses=total_num_responses(session, context, database_id),
90 )
93def generate_reply_notifications(payload: jobs_pb2.GenerateReplyNotificationsPayload) -> None:
94 # Import here to avoid circular dependency
95 from couchers.servicers.discussions import discussion_to_pb # noqa: PLC0415
96 from couchers.servicers.events import event_to_pb # noqa: PLC0415
98 with session_scope() as session:
99 database_id, depth = unpack_thread_id(payload.thread_id)
100 if depth == 1:
101 # this is a top-level Comment on a Thread attached to event, discussion, etc
102 comment = session.execute(select(Comment).where(Comment.id == database_id)).scalar_one()
103 thread = session.execute(select(Thread).where(Thread.id == comment.thread_id)).scalar_one()
104 author_user = session.execute(select(User).where(User.id == comment.author_user_id)).scalar_one()
105 # reply object for notif
106 reply = threads_pb2.Reply(
107 thread_id=payload.thread_id,
108 content=comment.content,
109 author_user_id=comment.author_user_id,
110 created_time=Timestamp_from_datetime(comment.created),
111 num_replies=0,
112 )
113 # figure out if the thread is related to an event or discussion
114 event = session.execute(select(Event).where(Event.thread_id == thread.id)).scalar_one_or_none()
115 discussion = session.execute(
116 select(Discussion).where(Discussion.thread_id == thread.id)
117 ).scalar_one_or_none()
118 if event:
119 # thread is an event thread
120 occurrence = event.occurrences.order_by(EventOccurrence.id.desc()).limit(1).one()
121 subscribed_user_ids = [user.id for user in event.subscribers]
122 attending_user_ids = [user.user_id for user in occurrence.attendances]
124 for user_id in set(subscribed_user_ids + attending_user_ids):
125 if is_not_visible(session, user_id, comment.author_user_id): 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 continue
127 if user_id == comment.author_user_id: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 continue
129 context = make_notification_user_context(user_id=user_id)
130 notify(
131 session,
132 user_id=user_id,
133 topic_action=NotificationTopicAction.event__comment,
134 key=str(occurrence.id),
135 data=notification_data_pb2.EventComment(
136 reply=reply,
137 event=event_to_pb(session, occurrence, context),
138 author=user_model_to_pb(author_user, session, context),
139 ),
140 moderation_state_id=comment.moderation_state_id,
141 )
142 elif discussion: 142 ↛ 169line 142 didn't jump to line 169 because the condition on line 142 was always true
143 # community discussion thread
144 cluster = discussion.owner_cluster
146 if not cluster.is_official_cluster: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true
147 raise NotImplementedError("Shouldn't have discussions under groups, only communities")
149 for user_id in [discussion.creator_user_id]:
150 if is_not_visible(session, user_id, comment.author_user_id): 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true
151 continue
152 if user_id == comment.author_user_id: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 continue
155 context = make_notification_user_context(user_id=user_id)
156 notify(
157 session,
158 user_id=user_id,
159 topic_action=NotificationTopicAction.discussion__comment,
160 key=str(discussion.id),
161 data=notification_data_pb2.DiscussionComment(
162 reply=reply,
163 discussion=discussion_to_pb(session, discussion, context),
164 author=user_model_to_pb(author_user, session, context),
165 ),
166 moderation_state_id=comment.moderation_state_id,
167 )
168 else:
169 raise NotImplementedError("I can only do event and discussion threads for now")
170 elif depth == 2: 170 ↛ 245line 170 didn't jump to line 245 because the condition on line 170 was always true
171 # this is a second-level reply to a comment
172 db_reply = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one()
173 # the comment we're replying to
174 parent_comment = session.execute(select(Comment).where(Comment.id == db_reply.comment_id)).scalar_one()
175 context = make_background_user_context(user_id=db_reply.author_user_id)
176 thread_replies_author_user_ids = (
177 session.execute(
178 where_users_column_visible(
179 select(Reply.author_user_id).where(Reply.comment_id == parent_comment.id),
180 context,
181 Reply.author_user_id,
182 )
183 )
184 .scalars()
185 .all()
186 )
187 thread_user_ids = set(thread_replies_author_user_ids)
188 if not is_not_visible(session, parent_comment.author_user_id, db_reply.author_user_id): 188 ↛ 191line 188 didn't jump to line 191 because the condition on line 188 was always true
189 thread_user_ids.add(parent_comment.author_user_id)
191 author_user = session.execute(select(User).where(User.id == db_reply.author_user_id)).scalar_one()
193 user_ids_to_notify = set(thread_user_ids) - {db_reply.author_user_id}
195 reply = threads_pb2.Reply(
196 thread_id=payload.thread_id,
197 content=db_reply.content,
198 author_user_id=db_reply.author_user_id,
199 created_time=Timestamp_from_datetime(db_reply.created),
200 num_replies=0,
201 )
203 event = session.execute(
204 select(Event).where(Event.thread_id == parent_comment.thread_id)
205 ).scalar_one_or_none()
206 discussion = session.execute(
207 select(Discussion).where(Discussion.thread_id == parent_comment.thread_id)
208 ).scalar_one_or_none()
209 if event:
210 # thread is an event thread
211 occurrence = event.occurrences.order_by(EventOccurrence.id.desc()).limit(1).one()
212 for user_id in user_ids_to_notify:
213 context = make_notification_user_context(user_id=user_id)
214 notify(
215 session,
216 user_id=user_id,
217 topic_action=NotificationTopicAction.thread__reply,
218 key=str(occurrence.id),
219 data=notification_data_pb2.ThreadReply(
220 reply=reply,
221 event=event_to_pb(session, occurrence, context),
222 author=user_model_to_pb(author_user, session, context),
223 ),
224 moderation_state_id=db_reply.moderation_state_id,
225 )
226 elif discussion: 226 ↛ 243line 226 didn't jump to line 243 because the condition on line 226 was always true
227 # community discussion thread
228 for user_id in user_ids_to_notify:
229 context = make_notification_user_context(user_id=user_id)
230 notify(
231 session,
232 user_id=user_id,
233 topic_action=NotificationTopicAction.thread__reply,
234 key=str(discussion.id),
235 data=notification_data_pb2.ThreadReply(
236 reply=reply,
237 discussion=discussion_to_pb(session, discussion, context),
238 author=user_model_to_pb(author_user, session, context),
239 ),
240 moderation_state_id=db_reply.moderation_state_id,
241 )
242 else:
243 raise NotImplementedError("I can only do event and discussion threads for now")
244 else:
245 raise Exception("Unknown depth")
248class Threads(threads_pb2_grpc.ThreadsServicer):
249 def GetThread(
250 self, request: threads_pb2.GetThreadReq, context: CouchersContext, session: Session
251 ) -> threads_pb2.GetThreadRes:
252 database_id, depth = unpack_thread_id(request.thread_id)
253 page_size = request.page_size if 0 < request.page_size < 100000 else 1000
254 page_start = unpack_thread_id(int(request.page_token))[0] if request.page_token else 2**50
256 if depth == 0:
257 if not session.execute(select(Thread).where(Thread.id == database_id)).scalar_one_or_none(): 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
260 has_replies = exists().where((Reply.comment_id == Comment.id) & (Reply.deleted == None)).correlate(Comment)
261 visible_reply_count = (
262 where_moderated_content_visible(
263 where_users_column_visible(
264 select(func.count(Reply.id)).where(Reply.comment_id == Comment.id).where(Reply.deleted == None),
265 context,
266 Reply.author_user_id,
267 ),
268 context,
269 Reply,
270 is_list_operation=True,
271 )
272 .correlate(Comment)
273 .scalar_subquery()
274 )
276 res = session.execute(
277 where_moderated_content_visible(
278 where_users_column_visible(
279 select(Comment, visible_reply_count)
280 .where(Comment.thread_id == database_id)
281 .where((Comment.deleted == None) | has_replies)
282 .where(Comment.id < page_start)
283 .order_by(Comment.created.desc())
284 .limit(page_size + 1),
285 context,
286 Comment.author_user_id,
287 ),
288 context,
289 Comment,
290 is_list_operation=True,
291 )
292 ).all()
293 # Deleted comments are shown as stubs (thread_id, deleted, num_replies only)
294 # to preserve thread structure, but content and author are stripped.
295 replies = []
296 for r, n in res[:page_size]:
297 if r.deleted is not None:
298 replies.append(
299 threads_pb2.Reply(
300 thread_id=pack_thread_id(r.id, 1),
301 deleted=True,
302 num_replies=n,
303 )
304 )
305 else:
306 replies.append(
307 threads_pb2.Reply(
308 thread_id=pack_thread_id(r.id, 1),
309 content=r.content,
310 author_user_id=r.author_user_id,
311 created_time=Timestamp_from_datetime(r.created),
312 num_replies=n,
313 can_edit=(context.user_id == r.author_user_id),
314 last_edited=Timestamp_from_datetime(r.last_edited) if r.last_edited else None,
315 )
316 )
318 elif depth == 1:
319 if not session.execute(
320 where_moderated_content_visible(
321 where_users_column_visible(
322 select(Comment).where(Comment.id == database_id),
323 context,
324 Comment.author_user_id,
325 ),
326 context,
327 Comment,
328 )
329 ).scalar_one_or_none():
330 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
332 res = (
333 session.execute( # type: ignore[assignment]
334 where_moderated_content_visible(
335 where_users_column_visible(
336 select(Reply)
337 .where(Reply.comment_id == database_id)
338 .where(Reply.deleted == None)
339 .where(Reply.id < page_start)
340 .order_by(Reply.created.desc())
341 .limit(page_size + 1),
342 context,
343 Reply.author_user_id,
344 ),
345 context,
346 Reply,
347 is_list_operation=True,
348 )
349 )
350 .scalars()
351 .all()
352 )
353 replies = [
354 threads_pb2.Reply(
355 thread_id=pack_thread_id(r.id, 2),
356 content=r.content,
357 author_user_id=r.author_user_id,
358 created_time=Timestamp_from_datetime(r.created),
359 num_replies=0,
360 can_edit=(context.user_id == r.author_user_id),
361 last_edited=Timestamp_from_datetime(r.last_edited) if r.last_edited else None,
362 )
363 for r in res[:page_size]
364 ]
366 else:
367 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
369 if len(res) > page_size:
370 # There's more!
371 next_page_token = str(replies[-1].thread_id)
372 else:
373 next_page_token = ""
375 return threads_pb2.GetThreadRes(replies=replies, next_page_token=next_page_token)
377 def PostReply(
378 self, request: threads_pb2.PostReplyReq, context: CouchersContext, session: Session
379 ) -> threads_pb2.PostReplyRes:
380 content = request.content.strip()
382 if content == "":
383 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_comment")
385 database_id, depth = unpack_thread_id(request.thread_id)
386 if depth not in (0, 1):
387 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
389 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
390 if not has_completed_profile(session, user):
391 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_post_comment")
393 object_to_add: Comment | Reply | None = None
395 def create_object(moderation_state_id: int) -> int:
396 nonlocal object_to_add
397 if depth == 0:
398 object_to_add = Comment(
399 thread_id=database_id,
400 author_user_id=context.user_id,
401 content=content,
402 moderation_state_id=moderation_state_id,
403 )
404 else:
405 object_to_add = Reply(
406 comment_id=database_id,
407 author_user_id=context.user_id,
408 content=content,
409 moderation_state_id=moderation_state_id,
410 )
411 session.add(object_to_add)
412 try:
413 session.flush()
414 except sqlalchemy.exc.IntegrityError:
415 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
416 return object_to_add.id
418 create_moderation(
419 session=session,
420 object_type=ModerationObjectType.comment if depth == 0 else ModerationObjectType.reply,
421 object_id=create_object,
422 creator_user_id=context.user_id,
423 )
425 assert object_to_add is not None
426 thread_id = pack_thread_id(object_to_add.id, depth + 1)
428 queue_job(
429 session,
430 job=generate_reply_notifications,
431 payload=jobs_pb2.GenerateReplyNotificationsPayload(
432 thread_id=thread_id,
433 ),
434 )
436 return threads_pb2.PostReplyRes(thread_id=thread_id)
438 def UpdateReply(
439 self, request: threads_pb2.UpdateReplyReq, context: CouchersContext, session: Session
440 ) -> threads_pb2.Reply:
441 content = request.content.strip()
442 if not content: 442 ↛ 443line 442 didn't jump to line 443 because the condition on line 442 was never true
443 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_comment")
445 database_id, depth = unpack_thread_id(request.thread_id)
446 if depth == 1:
447 obj: Comment | Reply | None = session.execute(
448 select(Comment).where(Comment.id == database_id)
449 ).scalar_one_or_none()
450 elif depth == 2: 450 ↛ 453line 450 didn't jump to line 453 because the condition on line 450 was always true
451 obj = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one_or_none()
452 else:
453 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
455 if not obj: 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true
456 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
457 if obj.deleted is not None: 457 ↛ 458line 457 didn't jump to line 458 because the condition on line 457 was never true
458 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "reply_deleted")
459 if obj.author_user_id != context.user_id: 459 ↛ 460line 459 didn't jump to line 460 because the condition on line 459 was never true
460 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "reply_edit_permission_denied")
462 old_content = obj.content
464 if depth == 1:
465 session.add(
466 CommentVersion(
467 comment_id=database_id,
468 editor_user_id=context.user_id,
469 change_type=ContentChangeType.edit,
470 old_content=old_content,
471 new_content=content,
472 )
473 )
474 else:
475 session.add(
476 ReplyVersion(
477 reply_id=database_id,
478 editor_user_id=context.user_id,
479 change_type=ContentChangeType.edit,
480 old_content=old_content,
481 new_content=content,
482 )
483 )
485 obj.content = content
486 obj.last_edited = now()
488 return threads_pb2.Reply(
489 thread_id=request.thread_id,
490 content=obj.content,
491 author_user_id=obj.author_user_id,
492 created_time=Timestamp_from_datetime(obj.created),
493 num_replies=0,
494 can_edit=True,
495 last_edited=Timestamp_from_datetime(obj.last_edited),
496 )
498 def DeleteReply(
499 self, request: threads_pb2.DeleteReplyReq, context: CouchersContext, session: Session
500 ) -> empty_pb2.Empty:
501 database_id, depth = unpack_thread_id(request.thread_id)
502 if depth == 1:
503 obj: Comment | Reply | None = session.execute(
504 select(Comment).where(Comment.id == database_id)
505 ).scalar_one_or_none()
506 elif depth == 2: 506 ↛ 509line 506 didn't jump to line 509 because the condition on line 506 was always true
507 obj = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one_or_none()
508 else:
509 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
511 if not obj: 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true
512 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
513 if obj.deleted is not None: 513 ↛ 514line 513 didn't jump to line 514 because the condition on line 513 was never true
514 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "reply_deleted")
515 if obj.author_user_id != context.user_id: 515 ↛ 516line 515 didn't jump to line 516 because the condition on line 515 was never true
516 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "reply_delete_permission_denied")
518 if depth == 1:
519 session.add(
520 CommentVersion(
521 comment_id=database_id,
522 editor_user_id=context.user_id,
523 change_type=ContentChangeType.delete,
524 old_content=obj.content,
525 new_content=None,
526 )
527 )
528 else:
529 session.add(
530 ReplyVersion(
531 reply_id=database_id,
532 editor_user_id=context.user_id,
533 change_type=ContentChangeType.delete,
534 old_content=obj.content,
535 new_content=None,
536 )
537 )
539 obj.deleted = now()
541 return empty_pb2.Empty()