Coverage for app/backend/src/couchers/servicers/threads.py: 86%
191 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +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 PendingNotification, notify_many
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 notifications = []
97 with session_scope() as session:
98 database_id, depth = unpack_thread_id(payload.thread_id)
99 if depth == 1:
100 # this is a top-level Comment on a Thread attached to event, discussion, etc
101 comment = session.execute(select(Comment).where(Comment.id == database_id)).scalar_one()
102 thread = session.execute(select(Thread).where(Thread.id == comment.thread_id)).scalar_one()
103 author_user = session.execute(select(User).where(User.id == comment.author_user_id)).scalar_one()
104 # reply object for notif
105 reply = threads_pb2.Reply(
106 thread_id=payload.thread_id,
107 content=comment.content,
108 author_user_id=comment.author_user_id,
109 created_time=Timestamp_from_datetime(comment.created),
110 num_replies=0,
111 )
112 # figure out if the thread is related to an event occurrence or discussion
113 occurrence = session.execute(
114 select(EventOccurrence).where(EventOccurrence.thread_id == thread.id)
115 ).scalar_one_or_none()
116 discussion = session.execute(
117 select(Discussion).where(Discussion.thread_id == thread.id)
118 ).scalar_one_or_none()
119 if occurrence:
120 # thread is an event occurrence thread
121 event = occurrence.event
122 subscribed_user_ids = [user.id for user in event.subscribers]
123 attending_user_ids = [user.user_id for user in occurrence.attendances]
125 for user_id in set(subscribed_user_ids + attending_user_ids):
126 if is_not_visible(session, user_id, comment.author_user_id): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 continue
128 if user_id == comment.author_user_id: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 continue
130 context = make_notification_user_context(user_id=user_id)
131 notifications.append(
132 PendingNotification(
133 user_id=user_id,
134 topic_action=NotificationTopicAction.event__comment,
135 key=str(occurrence.id),
136 data=notification_data_pb2.EventComment(
137 reply=reply,
138 event=event_to_pb(session, occurrence, context),
139 author=user_model_to_pb(author_user, session, context),
140 ),
141 moderation_state_id=comment.moderation_state_id,
142 )
143 )
144 elif discussion: 144 ↛ 172line 144 didn't jump to line 172 because the condition on line 144 was always true
145 # community discussion thread
146 cluster = discussion.owner_cluster
148 if not cluster.is_official_cluster: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true
149 raise NotImplementedError("Shouldn't have discussions under groups, only communities")
151 for user_id in [discussion.creator_user_id]:
152 if is_not_visible(session, 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
154 if user_id == comment.author_user_id: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 continue
157 context = make_notification_user_context(user_id=user_id)
158 notifications.append(
159 PendingNotification(
160 user_id=user_id,
161 topic_action=NotificationTopicAction.discussion__comment,
162 key=str(discussion.id),
163 data=notification_data_pb2.DiscussionComment(
164 reply=reply,
165 discussion=discussion_to_pb(session, discussion, context),
166 author=user_model_to_pb(author_user, session, context),
167 ),
168 moderation_state_id=comment.moderation_state_id,
169 )
170 )
171 else:
172 raise NotImplementedError("I can only do event and discussion threads for now")
173 elif depth == 2: 173 ↛ 249line 173 didn't jump to line 249 because the condition on line 173 was always true
174 # this is a second-level reply to a comment
175 db_reply = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one()
176 # the comment we're replying to
177 parent_comment = session.execute(select(Comment).where(Comment.id == db_reply.comment_id)).scalar_one()
178 context = make_background_user_context(user_id=db_reply.author_user_id)
179 thread_replies_author_user_ids = (
180 session.execute(
181 where_users_column_visible(
182 select(Reply.author_user_id).where(Reply.comment_id == parent_comment.id),
183 context,
184 Reply.author_user_id,
185 )
186 )
187 .scalars()
188 .all()
189 )
190 thread_user_ids = set(thread_replies_author_user_ids)
191 if not is_not_visible(session, parent_comment.author_user_id, db_reply.author_user_id): 191 ↛ 194line 191 didn't jump to line 194 because the condition on line 191 was always true
192 thread_user_ids.add(parent_comment.author_user_id)
194 author_user = session.execute(select(User).where(User.id == db_reply.author_user_id)).scalar_one()
196 user_ids_to_notify = set(thread_user_ids) - {db_reply.author_user_id}
198 reply = threads_pb2.Reply(
199 thread_id=payload.thread_id,
200 content=db_reply.content,
201 author_user_id=db_reply.author_user_id,
202 created_time=Timestamp_from_datetime(db_reply.created),
203 num_replies=0,
204 )
206 occurrence = session.execute(
207 select(EventOccurrence).where(EventOccurrence.thread_id == parent_comment.thread_id)
208 ).scalar_one_or_none()
209 discussion = session.execute(
210 select(Discussion).where(Discussion.thread_id == parent_comment.thread_id)
211 ).scalar_one_or_none()
212 if occurrence:
213 # thread is an event occurrence thread
214 for user_id in user_ids_to_notify:
215 context = make_notification_user_context(user_id=user_id)
216 notifications.append(
217 PendingNotification(
218 user_id=user_id,
219 topic_action=NotificationTopicAction.thread__reply,
220 key=str(occurrence.id),
221 data=notification_data_pb2.ThreadReply(
222 reply=reply,
223 event=event_to_pb(session, occurrence, context),
224 author=user_model_to_pb(author_user, session, context),
225 ),
226 moderation_state_id=db_reply.moderation_state_id,
227 )
228 )
229 elif discussion: 229 ↛ 247line 229 didn't jump to line 247 because the condition on line 229 was always true
230 # community discussion thread
231 for user_id in user_ids_to_notify:
232 context = make_notification_user_context(user_id=user_id)
233 notifications.append(
234 PendingNotification(
235 user_id=user_id,
236 topic_action=NotificationTopicAction.thread__reply,
237 key=str(discussion.id),
238 data=notification_data_pb2.ThreadReply(
239 reply=reply,
240 discussion=discussion_to_pb(session, discussion, context),
241 author=user_model_to_pb(author_user, session, context),
242 ),
243 moderation_state_id=db_reply.moderation_state_id,
244 )
245 )
246 else:
247 raise NotImplementedError("I can only do event and discussion threads for now")
248 else:
249 raise Exception("Unknown depth")
251 with session_scope() as session:
252 notify_many(session, notifications)
255class Threads(threads_pb2_grpc.ThreadsServicer):
256 def GetThread(
257 self, request: threads_pb2.GetThreadReq, context: CouchersContext, session: Session
258 ) -> threads_pb2.GetThreadRes:
259 database_id, depth = unpack_thread_id(request.thread_id)
260 page_size = request.page_size if 0 < request.page_size < 100000 else 1000
261 page_start = unpack_thread_id(int(request.page_token))[0] if request.page_token else 2**50
263 if depth == 0:
264 if not session.execute(select(Thread).where(Thread.id == database_id)).scalar_one_or_none(): 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
267 has_replies = exists().where((Reply.comment_id == Comment.id) & (Reply.deleted == None)).correlate(Comment)
268 visible_reply_count = (
269 where_moderated_content_visible(
270 where_users_column_visible(
271 select(func.count(Reply.id)).where(Reply.comment_id == Comment.id).where(Reply.deleted == None),
272 context,
273 Reply.author_user_id,
274 ),
275 context,
276 Reply,
277 is_list_operation=True,
278 )
279 .correlate(Comment)
280 .scalar_subquery()
281 )
283 res = session.execute(
284 where_moderated_content_visible(
285 where_users_column_visible(
286 select(Comment, visible_reply_count)
287 .where(Comment.thread_id == database_id)
288 .where((Comment.deleted == None) | has_replies)
289 .where(Comment.id < page_start)
290 .order_by(Comment.created.desc())
291 .limit(page_size + 1),
292 context,
293 Comment.author_user_id,
294 ),
295 context,
296 Comment,
297 is_list_operation=True,
298 )
299 ).all()
300 # Deleted comments are shown as stubs (thread_id, deleted, num_replies only)
301 # to preserve thread structure, but content and author are stripped.
302 replies = []
303 for r, n in res[:page_size]:
304 if r.deleted is not None:
305 replies.append(
306 threads_pb2.Reply(
307 thread_id=pack_thread_id(r.id, 1),
308 deleted=True,
309 num_replies=n,
310 )
311 )
312 else:
313 replies.append(
314 threads_pb2.Reply(
315 thread_id=pack_thread_id(r.id, 1),
316 content=r.content,
317 author_user_id=r.author_user_id,
318 created_time=Timestamp_from_datetime(r.created),
319 num_replies=n,
320 can_edit=(context.user_id == r.author_user_id),
321 last_edited=Timestamp_from_datetime(r.last_edited) if r.last_edited else None,
322 )
323 )
325 elif depth == 1:
326 if not session.execute(
327 where_moderated_content_visible(
328 where_users_column_visible(
329 select(Comment).where(Comment.id == database_id),
330 context,
331 Comment.author_user_id,
332 ),
333 context,
334 Comment,
335 )
336 ).scalar_one_or_none():
337 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
339 res = (
340 session.execute( # type: ignore[assignment]
341 where_moderated_content_visible(
342 where_users_column_visible(
343 select(Reply)
344 .where(Reply.comment_id == database_id)
345 .where(Reply.deleted == None)
346 .where(Reply.id < page_start)
347 .order_by(Reply.created.desc())
348 .limit(page_size + 1),
349 context,
350 Reply.author_user_id,
351 ),
352 context,
353 Reply,
354 is_list_operation=True,
355 )
356 )
357 .scalars()
358 .all()
359 )
360 replies = [
361 threads_pb2.Reply(
362 thread_id=pack_thread_id(r.id, 2),
363 content=r.content,
364 author_user_id=r.author_user_id,
365 created_time=Timestamp_from_datetime(r.created),
366 num_replies=0,
367 can_edit=(context.user_id == r.author_user_id),
368 last_edited=Timestamp_from_datetime(r.last_edited) if r.last_edited else None,
369 )
370 for r in res[:page_size]
371 ]
373 else:
374 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
376 if len(res) > page_size:
377 # There's more!
378 next_page_token = str(replies[-1].thread_id)
379 else:
380 next_page_token = ""
382 return threads_pb2.GetThreadRes(replies=replies, next_page_token=next_page_token)
384 def PostReply(
385 self, request: threads_pb2.PostReplyReq, context: CouchersContext, session: Session
386 ) -> threads_pb2.PostReplyRes:
387 content = request.content.strip()
389 if content == "":
390 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_comment")
392 database_id, depth = unpack_thread_id(request.thread_id)
393 if depth not in (0, 1):
394 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
396 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
397 if not has_completed_profile(session, user):
398 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_post_comment")
400 object_to_add: Comment | Reply | None = None
402 def create_object(moderation_state_id: int) -> int:
403 nonlocal object_to_add
404 if depth == 0:
405 object_to_add = Comment(
406 thread_id=database_id,
407 author_user_id=context.user_id,
408 content=content,
409 moderation_state_id=moderation_state_id,
410 )
411 else:
412 object_to_add = Reply(
413 comment_id=database_id,
414 author_user_id=context.user_id,
415 content=content,
416 moderation_state_id=moderation_state_id,
417 )
418 session.add(object_to_add)
419 try:
420 session.flush()
421 except sqlalchemy.exc.IntegrityError:
422 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
423 return object_to_add.id
425 create_moderation(
426 session=session,
427 object_type=ModerationObjectType.comment if depth == 0 else ModerationObjectType.reply,
428 object_id=create_object,
429 creator_user_id=context.user_id,
430 )
432 assert object_to_add is not None
433 thread_id = pack_thread_id(object_to_add.id, depth + 1)
435 queue_job(
436 session,
437 job=generate_reply_notifications,
438 payload=jobs_pb2.GenerateReplyNotificationsPayload(
439 thread_id=thread_id,
440 ),
441 )
443 return threads_pb2.PostReplyRes(thread_id=thread_id)
445 def UpdateReply(
446 self, request: threads_pb2.UpdateReplyReq, context: CouchersContext, session: Session
447 ) -> threads_pb2.Reply:
448 content = request.content.strip()
449 if not content: 449 ↛ 450line 449 didn't jump to line 450 because the condition on line 449 was never true
450 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_comment")
452 database_id, depth = unpack_thread_id(request.thread_id)
453 if depth == 1:
454 obj: Comment | Reply | None = session.execute(
455 select(Comment).where(Comment.id == database_id)
456 ).scalar_one_or_none()
457 elif depth == 2: 457 ↛ 460line 457 didn't jump to line 460 because the condition on line 457 was always true
458 obj = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one_or_none()
459 else:
460 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
462 if not obj: 462 ↛ 463line 462 didn't jump to line 463 because the condition on line 462 was never true
463 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
464 if obj.deleted is not None: 464 ↛ 465line 464 didn't jump to line 465 because the condition on line 464 was never true
465 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "reply_deleted")
466 if obj.author_user_id != context.user_id: 466 ↛ 467line 466 didn't jump to line 467 because the condition on line 466 was never true
467 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "reply_edit_permission_denied")
469 old_content = obj.content
471 if depth == 1:
472 session.add(
473 CommentVersion(
474 comment_id=database_id,
475 editor_user_id=context.user_id,
476 change_type=ContentChangeType.edit,
477 old_content=old_content,
478 new_content=content,
479 )
480 )
481 else:
482 session.add(
483 ReplyVersion(
484 reply_id=database_id,
485 editor_user_id=context.user_id,
486 change_type=ContentChangeType.edit,
487 old_content=old_content,
488 new_content=content,
489 )
490 )
492 obj.content = content
493 obj.last_edited = now()
495 return threads_pb2.Reply(
496 thread_id=request.thread_id,
497 content=obj.content,
498 author_user_id=obj.author_user_id,
499 created_time=Timestamp_from_datetime(obj.created),
500 num_replies=0,
501 can_edit=True,
502 last_edited=Timestamp_from_datetime(obj.last_edited),
503 )
505 def DeleteReply(
506 self, request: threads_pb2.DeleteReplyReq, context: CouchersContext, session: Session
507 ) -> empty_pb2.Empty:
508 database_id, depth = unpack_thread_id(request.thread_id)
509 if depth == 1:
510 obj: Comment | Reply | None = session.execute(
511 select(Comment).where(Comment.id == database_id)
512 ).scalar_one_or_none()
513 elif depth == 2: 513 ↛ 516line 513 didn't jump to line 516 because the condition on line 513 was always true
514 obj = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one_or_none()
515 else:
516 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
518 if not obj: 518 ↛ 519line 518 didn't jump to line 519 because the condition on line 518 was never true
519 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "thread_not_found")
520 if obj.deleted is not None: 520 ↛ 521line 520 didn't jump to line 521 because the condition on line 520 was never true
521 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "reply_deleted")
522 if obj.author_user_id != context.user_id: 522 ↛ 523line 522 didn't jump to line 523 because the condition on line 522 was never true
523 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "reply_delete_permission_denied")
525 if depth == 1:
526 session.add(
527 CommentVersion(
528 comment_id=database_id,
529 editor_user_id=context.user_id,
530 change_type=ContentChangeType.delete,
531 old_content=obj.content,
532 new_content=None,
533 )
534 )
535 else:
536 session.add(
537 ReplyVersion(
538 reply_id=database_id,
539 editor_user_id=context.user_id,
540 change_type=ContentChangeType.delete,
541 old_content=obj.content,
542 new_content=None,
543 )
544 )
546 obj.deleted = now()
548 return empty_pb2.Empty()