Coverage for app/backend/src/couchers/servicers/threads.py: 86%
188 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +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 EventOccurrence,
18 ModerationObjectType,
19 Reply,
20 Thread,
21 User,
22)
23from couchers.models.discussions import CommentVersion, ContentChangeType, ReplyVersion
24from couchers.models.notifications import NotificationTopicAction
25from couchers.moderation.utils import create_moderation
26from couchers.notifications.notify import notify
27from couchers.proto import notification_data_pb2, threads_pb2, threads_pb2_grpc
28from couchers.proto.internal import jobs_pb2
29from couchers.servicers.api import user_model_to_pb
30from couchers.servicers.blocking import is_not_visible
31from couchers.sql import where_moderated_content_visible, where_users_column_visible
32from couchers.utils import Timestamp_from_datetime, now
34logger = logging.getLogger(__name__)
37# Since the API exposes a single ID space regardless of nesting level,
38# we construct the API id by appending the nesting level to the
39# database ID.
42def pack_thread_id(database_id: int, depth: int) -> int:
43 return database_id * 10 + depth
46def unpack_thread_id(thread_id: int) -> tuple[int, int]:
47 """Returns (database_id, depth) tuple."""
48 return divmod(thread_id, 10)
51def total_num_responses(session: Session, context: CouchersContext, database_id: int) -> int:
52 """Return the total number of visible, non-deleted comments and replies to the thread with
53 database id database_id.
54 """
55 comments = where_moderated_content_visible(
56 where_users_column_visible(
57 select(func.count())
58 .select_from(Comment)
59 .where(Comment.thread_id == database_id)
60 .where(Comment.deleted == None),
61 context,
62 Comment.author_user_id,
63 ),
64 context,
65 Comment,
66 is_list_operation=True,
67 )
68 # the comment a reply hangs off is filtered too, but not on Comment.deleted: GetThread lists a
69 # deleted comment as a stub and still renders its replies
70 replies = (
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 )
77 replies = where_users_column_visible(replies, context, Reply.author_user_id)
78 replies = where_users_column_visible(replies, context, Comment.author_user_id)
79 replies = where_moderated_content_visible(replies, context, Reply, is_list_operation=True)
80 replies = where_moderated_content_visible(replies, context, Comment, is_list_operation=True)
81 return session.execute(comments).scalar_one() + session.execute(replies).scalar_one()
84def thread_to_pb(session: Session, context: CouchersContext, database_id: int) -> threads_pb2.Thread:
85 return threads_pb2.Thread(
86 thread_id=pack_thread_id(database_id, 0),
87 num_responses=total_num_responses(session, context, database_id),
88 )
91def generate_reply_notifications(payload: jobs_pb2.GenerateReplyNotificationsPayload) -> None:
92 # Import here to avoid circular dependency
93 from couchers.servicers.discussions import discussion_to_pb # noqa: PLC0415
94 from couchers.servicers.events import event_to_pb # noqa: PLC0415
96 with session_scope() as session:
97 database_id, depth = unpack_thread_id(payload.thread_id)
98 if depth == 1:
99 # this is a top-level Comment on a Thread attached to event, discussion, etc
100 comment = session.execute(select(Comment).where(Comment.id == database_id)).scalar_one()
101 thread = session.execute(select(Thread).where(Thread.id == comment.thread_id)).scalar_one()
102 author_user = session.execute(select(User).where(User.id == comment.author_user_id)).scalar_one()
103 # reply object for notif
104 reply = threads_pb2.Reply(
105 thread_id=payload.thread_id,
106 content=comment.content,
107 author_user_id=comment.author_user_id,
108 created_time=Timestamp_from_datetime(comment.created),
109 num_replies=0,
110 )
111 # figure out if the thread is related to an event occurrence or discussion
112 occurrence = session.execute(
113 select(EventOccurrence).where(EventOccurrence.thread_id == thread.id)
114 ).scalar_one_or_none()
115 discussion = session.execute(
116 select(Discussion).where(Discussion.thread_id == thread.id)
117 ).scalar_one_or_none()
118 if occurrence:
119 # thread is an event occurrence thread
120 event = occurrence.event
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 ↛ 244line 170 didn't jump to line 244 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 occurrence = session.execute(
204 select(EventOccurrence).where(EventOccurrence.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 occurrence:
210 # thread is an event occurrence thread
211 for user_id in user_ids_to_notify:
212 context = make_notification_user_context(user_id=user_id)
213 notify(
214 session,
215 user_id=user_id,
216 topic_action=NotificationTopicAction.thread__reply,
217 key=str(occurrence.id),
218 data=notification_data_pb2.ThreadReply(
219 reply=reply,
220 event=event_to_pb(session, occurrence, context),
221 author=user_model_to_pb(author_user, session, context),
222 ),
223 moderation_state_id=db_reply.moderation_state_id,
224 )
225 elif discussion: 225 ↛ 242line 225 didn't jump to line 242 because the condition on line 225 was always true
226 # community discussion thread
227 for user_id in user_ids_to_notify:
228 context = make_notification_user_context(user_id=user_id)
229 notify(
230 session,
231 user_id=user_id,
232 topic_action=NotificationTopicAction.thread__reply,
233 key=str(discussion.id),
234 data=notification_data_pb2.ThreadReply(
235 reply=reply,
236 discussion=discussion_to_pb(session, discussion, context),
237 author=user_model_to_pb(author_user, session, context),
238 ),
239 moderation_state_id=db_reply.moderation_state_id,
240 )
241 else:
242 raise NotImplementedError("I can only do event and discussion threads for now")
243 else:
244 raise Exception("Unknown depth")
247class Threads(threads_pb2_grpc.ThreadsServicer):
248 def GetThread(
249 self, request: threads_pb2.GetThreadReq, context: CouchersContext, session: Session
250 ) -> threads_pb2.GetThreadRes:
251 database_id, depth = unpack_thread_id(request.thread_id)
252 page_size = request.page_size if 0 < request.page_size < 100000 else 1000
253 page_start = unpack_thread_id(int(request.page_token))[0] if request.page_token else 2**50
255 if depth == 0:
256 if not session.execute(select(Thread).where(Thread.id == database_id)).scalar_one_or_none(): 256 ↛ 257line 256 didn't jump to line 257 because the condition on line 256 was never true
257 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
259 has_replies = exists().where((Reply.comment_id == Comment.id) & (Reply.deleted == None)).correlate(Comment)
260 visible_reply_count = (
261 where_moderated_content_visible(
262 where_users_column_visible(
263 select(func.count(Reply.id)).where(Reply.comment_id == Comment.id).where(Reply.deleted == None),
264 context,
265 Reply.author_user_id,
266 ),
267 context,
268 Reply,
269 is_list_operation=True,
270 )
271 .correlate(Comment)
272 .scalar_subquery()
273 )
275 res = session.execute(
276 where_moderated_content_visible(
277 where_users_column_visible(
278 select(Comment, visible_reply_count)
279 .where(Comment.thread_id == database_id)
280 .where((Comment.deleted == None) | has_replies)
281 .where(Comment.id < page_start)
282 .order_by(Comment.created.desc())
283 .limit(page_size + 1),
284 context,
285 Comment.author_user_id,
286 ),
287 context,
288 Comment,
289 is_list_operation=True,
290 )
291 ).all()
292 # Deleted comments are shown as stubs (thread_id, deleted, num_replies only)
293 # to preserve thread structure, but content and author are stripped.
294 replies = []
295 for r, n in res[:page_size]:
296 if r.deleted is not None:
297 replies.append(
298 threads_pb2.Reply(
299 thread_id=pack_thread_id(r.id, 1),
300 deleted=True,
301 num_replies=n,
302 )
303 )
304 else:
305 replies.append(
306 threads_pb2.Reply(
307 thread_id=pack_thread_id(r.id, 1),
308 content=r.content,
309 author_user_id=r.author_user_id,
310 created_time=Timestamp_from_datetime(r.created),
311 num_replies=n,
312 can_edit=(context.user_id == r.author_user_id),
313 last_edited=Timestamp_from_datetime(r.last_edited) if r.last_edited else None,
314 )
315 )
317 elif depth == 1:
318 if not session.execute(
319 where_moderated_content_visible(
320 where_users_column_visible(
321 select(Comment).where(Comment.id == database_id),
322 context,
323 Comment.author_user_id,
324 ),
325 context,
326 Comment,
327 )
328 ).scalar_one_or_none():
329 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
331 res = (
332 session.execute( # type: ignore[assignment]
333 where_moderated_content_visible(
334 where_users_column_visible(
335 select(Reply)
336 .where(Reply.comment_id == database_id)
337 .where(Reply.deleted == None)
338 .where(Reply.id < page_start)
339 .order_by(Reply.created.desc())
340 .limit(page_size + 1),
341 context,
342 Reply.author_user_id,
343 ),
344 context,
345 Reply,
346 is_list_operation=True,
347 )
348 )
349 .scalars()
350 .all()
351 )
352 replies = [
353 threads_pb2.Reply(
354 thread_id=pack_thread_id(r.id, 2),
355 content=r.content,
356 author_user_id=r.author_user_id,
357 created_time=Timestamp_from_datetime(r.created),
358 num_replies=0,
359 can_edit=(context.user_id == r.author_user_id),
360 last_edited=Timestamp_from_datetime(r.last_edited) if r.last_edited else None,
361 )
362 for r in res[:page_size]
363 ]
365 else:
366 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
368 if len(res) > page_size:
369 # There's more!
370 next_page_token = str(replies[-1].thread_id)
371 else:
372 next_page_token = ""
374 return threads_pb2.GetThreadRes(replies=replies, next_page_token=next_page_token)
376 def PostReply(
377 self, request: threads_pb2.PostReplyReq, context: CouchersContext, session: Session
378 ) -> threads_pb2.PostReplyRes:
379 content = request.content.strip()
381 if content == "":
382 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_comment")
384 database_id, depth = unpack_thread_id(request.thread_id)
385 if depth not in (0, 1):
386 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
388 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
389 if not has_completed_profile(session, user):
390 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_post_comment")
392 object_to_add: Comment | Reply | None = None
394 def create_object(moderation_state_id: int) -> int:
395 nonlocal object_to_add
396 if depth == 0:
397 object_to_add = Comment(
398 thread_id=database_id,
399 author_user_id=context.user_id,
400 content=content,
401 moderation_state_id=moderation_state_id,
402 )
403 else:
404 object_to_add = Reply(
405 comment_id=database_id,
406 author_user_id=context.user_id,
407 content=content,
408 moderation_state_id=moderation_state_id,
409 )
410 session.add(object_to_add)
411 try:
412 session.flush()
413 except sqlalchemy.exc.IntegrityError:
414 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
415 return object_to_add.id
417 create_moderation(
418 session=session,
419 object_type=ModerationObjectType.comment if depth == 0 else ModerationObjectType.reply,
420 object_id=create_object,
421 creator_user_id=context.user_id,
422 )
424 assert object_to_add is not None
425 thread_id = pack_thread_id(object_to_add.id, depth + 1)
427 queue_job(
428 session,
429 job=generate_reply_notifications,
430 payload=jobs_pb2.GenerateReplyNotificationsPayload(
431 thread_id=thread_id,
432 ),
433 )
435 return threads_pb2.PostReplyRes(thread_id=thread_id)
437 def UpdateReply(
438 self, request: threads_pb2.UpdateReplyReq, context: CouchersContext, session: Session
439 ) -> threads_pb2.Reply:
440 content = request.content.strip()
441 if not content: 441 ↛ 442line 441 didn't jump to line 442 because the condition on line 441 was never true
442 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_comment")
444 database_id, depth = unpack_thread_id(request.thread_id)
445 if depth == 1:
446 obj: Comment | Reply | None = session.execute(
447 select(Comment).where(Comment.id == database_id)
448 ).scalar_one_or_none()
449 elif depth == 2: 449 ↛ 452line 449 didn't jump to line 452 because the condition on line 449 was always true
450 obj = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one_or_none()
451 else:
452 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
454 if not obj: 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
456 if obj.deleted is not None: 456 ↛ 457line 456 didn't jump to line 457 because the condition on line 456 was never true
457 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "reply_deleted")
458 if obj.author_user_id != context.user_id: 458 ↛ 459line 458 didn't jump to line 459 because the condition on line 458 was never true
459 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "reply_edit_permission_denied")
461 old_content = obj.content
463 if depth == 1:
464 session.add(
465 CommentVersion(
466 comment_id=database_id,
467 editor_user_id=context.user_id,
468 change_type=ContentChangeType.edit,
469 old_content=old_content,
470 new_content=content,
471 )
472 )
473 else:
474 session.add(
475 ReplyVersion(
476 reply_id=database_id,
477 editor_user_id=context.user_id,
478 change_type=ContentChangeType.edit,
479 old_content=old_content,
480 new_content=content,
481 )
482 )
484 obj.content = content
485 obj.last_edited = now()
487 return threads_pb2.Reply(
488 thread_id=request.thread_id,
489 content=obj.content,
490 author_user_id=obj.author_user_id,
491 created_time=Timestamp_from_datetime(obj.created),
492 num_replies=0,
493 can_edit=True,
494 last_edited=Timestamp_from_datetime(obj.last_edited),
495 )
497 def DeleteReply(
498 self, request: threads_pb2.DeleteReplyReq, context: CouchersContext, session: Session
499 ) -> empty_pb2.Empty:
500 database_id, depth = unpack_thread_id(request.thread_id)
501 if depth == 1:
502 obj: Comment | Reply | None = session.execute(
503 select(Comment).where(Comment.id == database_id)
504 ).scalar_one_or_none()
505 elif depth == 2: 505 ↛ 508line 505 didn't jump to line 508 because the condition on line 505 was always true
506 obj = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one_or_none()
507 else:
508 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
510 if not obj: 510 ↛ 511line 510 didn't jump to line 511 because the condition on line 510 was never true
511 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
512 if obj.deleted is not None: 512 ↛ 513line 512 didn't jump to line 513 because the condition on line 512 was never true
513 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "reply_deleted")
514 if obj.author_user_id != context.user_id: 514 ↛ 515line 514 didn't jump to line 515 because the condition on line 514 was never true
515 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "reply_delete_permission_denied")
517 if depth == 1:
518 session.add(
519 CommentVersion(
520 comment_id=database_id,
521 editor_user_id=context.user_id,
522 change_type=ContentChangeType.delete,
523 old_content=obj.content,
524 new_content=None,
525 )
526 )
527 else:
528 session.add(
529 ReplyVersion(
530 reply_id=database_id,
531 editor_user_id=context.user_id,
532 change_type=ContentChangeType.delete,
533 old_content=obj.content,
534 new_content=None,
535 )
536 )
538 obj.deleted = now()
540 return empty_pb2.Empty()