Coverage for app/backend/src/couchers/servicers/conversations.py: 88%
315 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 22:54 +0000
1import logging
2from collections.abc import Sequence
3from datetime import timedelta
4from typing import Any, cast
6import grpc
7from google.protobuf import empty_pb2
8from sqlalchemy import select
9from sqlalchemy.orm import Session, contains_eager
10from sqlalchemy.sql import func, not_, or_
12from couchers.constants import DATETIME_INFINITY, DATETIME_MINUS_INFINITY
13from couchers.context import CouchersContext, make_background_user_context, make_notification_user_context
14from couchers.db import session_scope
15from couchers.event_log import log_event
16from couchers.helpers.completed_profile import has_completed_profile
17from couchers.jobs.enqueue import queue_job
18from couchers.metrics import sent_messages_counter
19from couchers.models import (
20 Conversation,
21 GroupChat,
22 GroupChatRole,
23 GroupChatSubscription,
24 Message,
25 MessageType,
26 ModerationObjectType,
27 RateLimitAction,
28 User,
29)
30from couchers.models.notifications import NotificationTopicAction
31from couchers.moderation.utils import create_moderation
32from couchers.notifications.notify import mark_notifications_seen, notify
33from couchers.proto import conversations_pb2, conversations_pb2_grpc, messages_pb2, notification_data_pb2
34from couchers.proto.internal import jobs_pb2
35from couchers.rate_limits.check import process_rate_limits_and_check_abort
36from couchers.rate_limits.definitions import RATE_LIMIT_HOURS
37from couchers.servicers.api import user_model_to_pb
38from couchers.sql import to_bool, users_visible, where_moderated_content_visible, where_users_column_visible
39from couchers.utils import Timestamp_from_datetime, now
41logger = logging.getLogger(__name__)
43# TODO: Still needs custom pagination: GetUpdates
44DEFAULT_PAGINATION_LENGTH = 20
45MAX_PAGE_SIZE = 50
48def _message_to_pb(message: Message) -> messages_pb2.Message:
49 """
50 Turns the given message to a protocol buffer
51 """
52 if message.is_normal_message:
53 return messages_pb2.Message(
54 message_id=message.id,
55 author_user_id=message.author_id,
56 time=Timestamp_from_datetime(message.time),
57 text=messages_pb2.MessageContentText(text=message.text),
58 )
59 else:
60 return messages_pb2.Message(
61 message_id=message.id,
62 author_user_id=message.author_id,
63 time=Timestamp_from_datetime(message.time),
64 chat_created=(
65 messages_pb2.MessageContentChatCreated() if message.message_type == MessageType.chat_created else None
66 ),
67 chat_edited=(
68 messages_pb2.MessageContentChatEdited() if message.message_type == MessageType.chat_edited else None
69 ),
70 user_invited=(
71 messages_pb2.MessageContentUserInvited(target_user_id=message.target_id)
72 if message.message_type == MessageType.user_invited
73 else None
74 ),
75 user_left=(
76 messages_pb2.MessageContentUserLeft() if message.message_type == MessageType.user_left else None
77 ),
78 user_made_admin=(
79 messages_pb2.MessageContentUserMadeAdmin(target_user_id=message.target_id)
80 if message.message_type == MessageType.user_made_admin
81 else None
82 ),
83 user_removed_admin=(
84 messages_pb2.MessageContentUserRemovedAdmin(target_user_id=message.target_id)
85 if message.message_type == MessageType.user_removed_admin
86 else None
87 ),
88 group_chat_user_removed=(
89 messages_pb2.MessageContentUserRemoved(target_user_id=message.target_id)
90 if message.message_type == MessageType.user_removed
91 else None
92 ),
93 )
96def _get_visible_members_for_subscription(subscription: GroupChatSubscription) -> list[int]:
97 """
98 If a user leaves a group chat, they shouldn't be able to see who's added
99 after they left
100 """
101 if not subscription.left:
102 # still in the chat, we see everyone with a current subscription
103 return [sub.user_id for sub in subscription.group_chat.subscriptions.where(GroupChatSubscription.left == None)]
104 else:
105 # not in chat anymore, see everyone who was in chat when we left
106 return [
107 sub.user_id
108 for sub in subscription.group_chat.subscriptions.where(
109 GroupChatSubscription.joined <= subscription.left
110 ).where(or_(GroupChatSubscription.left >= subscription.left, GroupChatSubscription.left == None))
111 ]
114def _get_visible_admins_for_subscription(subscription: GroupChatSubscription) -> list[int]:
115 """
116 If a user leaves a group chat, they shouldn't be able to see who's added
117 after they left
118 """
119 if not subscription.left:
120 # still in the chat, we see everyone with a current subscription
121 return [
122 sub.user_id
123 for sub in subscription.group_chat.subscriptions.where(GroupChatSubscription.left == None).where(
124 GroupChatSubscription.role == GroupChatRole.admin
125 )
126 ]
127 else:
128 # not in chat anymore, see everyone who was in chat when we left
129 return [
130 sub.user_id
131 for sub in subscription.group_chat.subscriptions.where(GroupChatSubscription.role == GroupChatRole.admin)
132 .where(GroupChatSubscription.joined <= subscription.left)
133 .where(or_(GroupChatSubscription.left >= subscription.left, GroupChatSubscription.left == None))
134 ]
137def _user_can_message(session: Session, context: CouchersContext, group_chat: GroupChat) -> bool:
138 """
139 If it is a true group chat (not a DM), user can always message. For a DM, user can message if the other participant
140 - Is not deleted/banned
141 - Has not been blocked by the user or is blocking the user
142 - Has not left the chat
143 """
144 if not group_chat.is_dm:
145 return True
147 query = select(
148 where_users_column_visible(
149 select(GroupChatSubscription)
150 .where(GroupChatSubscription.user_id != context.user_id)
151 .where(GroupChatSubscription.group_chat_id == group_chat.conversation_id)
152 .where(GroupChatSubscription.left == None),
153 context=context,
154 column=GroupChatSubscription.user_id,
155 ).exists()
156 )
157 return session.execute(query).scalar_one()
160def generate_message_notifications(payload: jobs_pb2.GenerateMessageNotificationsPayload) -> None:
161 """
162 Background job to generate notifications for a message sent to a group chat
163 """
164 logger.info(f"Fanning notifications for message_id = {payload.message_id}")
166 with session_scope() as session:
167 message, group_chat = session.execute(
168 select(Message, GroupChat)
169 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
170 .where(Message.id == payload.message_id)
171 ).one()
173 if message.message_type != MessageType.text:
174 logger.info(f"Not a text message, not notifying. message_id = {payload.message_id}")
175 return
177 context = make_background_user_context(user_id=message.author_id)
178 user_ids_to_notify = (
179 session.execute(
180 where_users_column_visible(
181 select(GroupChatSubscription.user_id)
182 .where(GroupChatSubscription.group_chat_id == message.conversation_id)
183 .where(GroupChatSubscription.user_id != message.author_id)
184 .where(GroupChatSubscription.joined <= message.time)
185 .where(or_(GroupChatSubscription.left == None, GroupChatSubscription.left >= message.time))
186 .where(not_(GroupChatSubscription.is_muted)),
187 context=context,
188 column=GroupChatSubscription.user_id,
189 )
190 )
191 .scalars()
192 .all()
193 )
195 for user_id in user_ids_to_notify:
196 notify(
197 session,
198 user_id=user_id,
199 topic_action=NotificationTopicAction.chat__message,
200 key=str(message.conversation_id),
201 data=notification_data_pb2.ChatMessage(
202 author=user_model_to_pb(
203 message.author,
204 session,
205 make_notification_user_context(user_id=user_id),
206 ),
207 text=message.text,
208 group_chat_id=message.conversation_id,
209 group_chat_title=group_chat.title or None,
210 # unseen_count irrelevant for this notification
211 ),
212 moderation_state_id=group_chat.moderation_state_id,
213 )
216def _add_message_to_subscription(session: Session, subscription: GroupChatSubscription, **kwargs: Any) -> Message:
217 """
218 Creates a new message for a subscription, from the user whose subscription that is. Updates last seen message id
220 Specify the keyword args for Message
221 """
222 message = Message(conversation_id=subscription.group_chat.conversation.id, author_id=subscription.user_id, **kwargs)
224 session.add(message)
225 session.flush()
227 subscription.last_seen_message_id = message.id
229 queue_job(
230 session,
231 job=generate_message_notifications,
232 payload=jobs_pb2.GenerateMessageNotificationsPayload(
233 message_id=message.id,
234 ),
235 )
237 return message
240def _create_chat(
241 session: Session,
242 creator_id: int,
243 recipient_ids: Sequence[int],
244 title: str | None = None,
245 only_admins_invite: bool = True,
246) -> GroupChat:
247 conversation = Conversation()
248 session.add(conversation)
249 session.flush()
251 # Create moderation state for UMS (starts as SHADOWED)
252 moderation_state = create_moderation(
253 session=session,
254 object_type=ModerationObjectType.group_chat,
255 object_id=conversation.id,
256 creator_user_id=creator_id,
257 )
259 chat = GroupChat(
260 conversation_id=conversation.id,
261 title=title,
262 creator_id=creator_id,
263 is_dm=True if len(recipient_ids) == 1 else False,
264 only_admins_invite=only_admins_invite,
265 moderation_state_id=moderation_state.id,
266 )
267 session.add(chat)
268 session.flush()
270 creator_subscription = GroupChatSubscription(
271 user_id=creator_id,
272 group_chat_id=chat.conversation_id,
273 role=GroupChatRole.admin,
274 )
275 session.add(creator_subscription)
277 for uid in recipient_ids:
278 session.add(
279 GroupChatSubscription(
280 user_id=uid,
281 group_chat_id=chat.conversation_id,
282 role=GroupChatRole.participant,
283 )
284 )
286 return chat
289def _get_message_subscription(session: Session, user_id: int, conversation_id: int) -> GroupChatSubscription:
290 subscription = session.execute(
291 select(GroupChatSubscription)
292 .where(GroupChatSubscription.group_chat_id == conversation_id)
293 .where(GroupChatSubscription.user_id == user_id)
294 .where(GroupChatSubscription.left == None)
295 ).scalar_one_or_none()
297 return cast(GroupChatSubscription, subscription)
300def _get_visible_message_subscription(
301 session: Session, context: CouchersContext, conversation_id: int
302) -> GroupChatSubscription:
303 """Get subscription with visibility filtering"""
304 subscription = session.execute(
305 where_moderated_content_visible(
306 select(GroupChatSubscription)
307 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
308 .where(GroupChatSubscription.group_chat_id == conversation_id)
309 .where(GroupChatSubscription.user_id == context.user_id)
310 .where(GroupChatSubscription.left == None),
311 context,
312 GroupChat,
313 is_list_operation=False,
314 )
315 ).scalar_one_or_none()
317 return cast(GroupChatSubscription, subscription)
320def _unseen_message_count(session: Session, subscription_id: int) -> int:
321 query = (
322 select(func.count())
323 .select_from(Message)
324 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
325 .where(GroupChatSubscription.id == subscription_id)
326 .where(Message.id > GroupChatSubscription.last_seen_message_id)
327 )
328 return session.execute(query).scalar_one()
331def _mute_info(subscription: GroupChatSubscription) -> conversations_pb2.MuteInfo:
332 (muted, muted_until) = subscription.muted_display()
333 return conversations_pb2.MuteInfo(
334 muted=muted,
335 muted_until=Timestamp_from_datetime(muted_until) if muted_until else None,
336 )
339class Conversations(conversations_pb2_grpc.ConversationsServicer):
340 def ListGroupChats(
341 self, request: conversations_pb2.ListGroupChatsReq, context: CouchersContext, session: Session
342 ) -> conversations_pb2.ListGroupChatsRes:
343 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH
344 page_size = min(page_size, MAX_PAGE_SIZE)
346 # select group chats where you have a subscription, and for each of
347 # these, the latest message from them
349 t = (
350 select(
351 GroupChatSubscription.group_chat_id.label("group_chat_id"),
352 func.max(GroupChatSubscription.id).label("group_chat_subscriptions_id"),
353 func.max(Message.id).label("message_id"),
354 )
355 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
356 .where(GroupChatSubscription.user_id == context.user_id)
357 .where(Message.time >= GroupChatSubscription.joined)
358 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
359 .where(
360 or_(
361 to_bool(request.HasField("only_archived") == False),
362 GroupChatSubscription.is_archived == request.only_archived,
363 )
364 )
365 .group_by(GroupChatSubscription.group_chat_id)
366 .order_by(func.max(Message.id).desc())
367 .subquery()
368 )
370 results = session.execute(
371 where_moderated_content_visible(
372 select(t, GroupChat, GroupChatSubscription, Message)
373 .join(Message, Message.id == t.c.message_id)
374 .join(GroupChatSubscription, GroupChatSubscription.id == t.c.group_chat_subscriptions_id)
375 .join(GroupChat, GroupChat.conversation_id == t.c.group_chat_id)
376 .join(Conversation, Conversation.id == GroupChat.conversation_id)
377 .options(contains_eager(GroupChat.conversation))
378 .where(or_(t.c.message_id < request.last_message_id, to_bool(request.last_message_id == 0)))
379 .order_by(t.c.message_id.desc())
380 .limit(page_size + 1),
381 context,
382 GroupChat,
383 is_list_operation=True,
384 )
385 ).all()
387 # Batch: unseen message counts in one query instead of N individual queries
388 subscription_ids = [r.GroupChatSubscription.id for r in results[:page_size]]
389 unseen_counts: dict[int, int] = dict(
390 session.execute( # type: ignore[arg-type]
391 select(GroupChatSubscription.id, func.count(Message.id))
392 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
393 .where(GroupChatSubscription.id.in_(subscription_ids))
394 .where(Message.id > GroupChatSubscription.last_seen_message_id)
395 .group_by(GroupChatSubscription.id)
396 ).all()
397 )
399 return conversations_pb2.ListGroupChatsRes(
400 group_chats=[
401 conversations_pb2.GroupChat(
402 group_chat_id=result.GroupChat.conversation_id,
403 title=result.GroupChat.title, # TODO: proper title for DMs, etc
404 member_user_ids=_get_visible_members_for_subscription(result.GroupChatSubscription),
405 admin_user_ids=_get_visible_admins_for_subscription(result.GroupChatSubscription),
406 only_admins_invite=result.GroupChat.only_admins_invite,
407 is_dm=result.GroupChat.is_dm,
408 created=Timestamp_from_datetime(result.GroupChat.conversation.created),
409 unseen_message_count=unseen_counts.get(result.GroupChatSubscription.id, 0),
410 last_seen_message_id=result.GroupChatSubscription.last_seen_message_id,
411 latest_message=_message_to_pb(result.Message) if result.Message else None,
412 mute_info=_mute_info(result.GroupChatSubscription),
413 can_message=_user_can_message(session, context, result.GroupChat),
414 is_archived=result.GroupChatSubscription.is_archived,
415 )
416 for result in results[:page_size]
417 ],
418 last_message_id=(
419 min(g.Message.id if g.Message else 1 for g in results[:page_size]) if len(results) > 0 else 0
420 ), # TODO
421 no_more=len(results) <= page_size,
422 )
424 def GetGroupChat(
425 self, request: conversations_pb2.GetGroupChatReq, context: CouchersContext, session: Session
426 ) -> conversations_pb2.GroupChat:
427 result = session.execute(
428 where_moderated_content_visible(
429 select(GroupChat, GroupChatSubscription, Message)
430 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
431 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
432 .join(Conversation, Conversation.id == GroupChat.conversation_id)
433 .options(contains_eager(GroupChat.conversation))
434 .where(GroupChatSubscription.user_id == context.user_id)
435 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
436 .where(Message.time >= GroupChatSubscription.joined)
437 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
438 .order_by(Message.id.desc())
439 .limit(1),
440 context,
441 GroupChat,
442 is_list_operation=False,
443 )
444 ).one_or_none()
446 if not result:
447 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
449 return conversations_pb2.GroupChat(
450 group_chat_id=result.GroupChat.conversation_id,
451 title=result.GroupChat.title,
452 member_user_ids=_get_visible_members_for_subscription(result.GroupChatSubscription),
453 admin_user_ids=_get_visible_admins_for_subscription(result.GroupChatSubscription),
454 only_admins_invite=result.GroupChat.only_admins_invite,
455 is_dm=result.GroupChat.is_dm,
456 created=Timestamp_from_datetime(result.GroupChat.conversation.created),
457 unseen_message_count=_unseen_message_count(session, result.GroupChatSubscription.id),
458 last_seen_message_id=result.GroupChatSubscription.last_seen_message_id,
459 latest_message=_message_to_pb(result.Message) if result.Message else None,
460 mute_info=_mute_info(result.GroupChatSubscription),
461 can_message=_user_can_message(session, context, result.GroupChat),
462 is_archived=result.GroupChatSubscription.is_archived,
463 )
465 def GetDirectMessage(
466 self, request: conversations_pb2.GetDirectMessageReq, context: CouchersContext, session: Session
467 ) -> conversations_pb2.GroupChat:
468 count = func.count(GroupChatSubscription.id).label("count")
469 subquery = (
470 select(GroupChatSubscription.group_chat_id)
471 .where(
472 or_(
473 GroupChatSubscription.user_id == context.user_id,
474 GroupChatSubscription.user_id == request.user_id,
475 )
476 )
477 .where(GroupChatSubscription.left == None)
478 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
479 .where(GroupChat.is_dm == True)
480 .group_by(GroupChatSubscription.group_chat_id)
481 .having(count == 2)
482 .subquery()
483 )
485 result = session.execute(
486 where_moderated_content_visible(
487 select(subquery, GroupChat, GroupChatSubscription, Message)
488 .join(subquery, subquery.c.group_chat_id == GroupChat.conversation_id)
489 .join(Message, Message.conversation_id == GroupChat.conversation_id)
490 .join(Conversation, Conversation.id == GroupChat.conversation_id)
491 .options(contains_eager(GroupChat.conversation))
492 .where(GroupChatSubscription.user_id == context.user_id)
493 .where(GroupChatSubscription.group_chat_id == GroupChat.conversation_id)
494 .where(Message.time >= GroupChatSubscription.joined)
495 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
496 .order_by(Message.id.desc())
497 .limit(1),
498 context,
499 GroupChat,
500 is_list_operation=False,
501 )
502 ).one_or_none()
504 if not result:
505 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
507 return conversations_pb2.GroupChat(
508 group_chat_id=result.GroupChat.conversation_id,
509 title=result.GroupChat.title,
510 member_user_ids=_get_visible_members_for_subscription(result.GroupChatSubscription),
511 admin_user_ids=_get_visible_admins_for_subscription(result.GroupChatSubscription),
512 only_admins_invite=result.GroupChat.only_admins_invite,
513 is_dm=result.GroupChat.is_dm,
514 created=Timestamp_from_datetime(result.GroupChat.conversation.created),
515 unseen_message_count=_unseen_message_count(session, result.GroupChatSubscription.id),
516 last_seen_message_id=result.GroupChatSubscription.last_seen_message_id,
517 latest_message=_message_to_pb(result.Message) if result.Message else None,
518 mute_info=_mute_info(result.GroupChatSubscription),
519 can_message=_user_can_message(session, context, result.GroupChat),
520 is_archived=result.GroupChatSubscription.is_archived,
521 )
523 def GetUpdates(
524 self, request: conversations_pb2.GetUpdatesReq, context: CouchersContext, session: Session
525 ) -> conversations_pb2.GetUpdatesRes:
526 results = (
527 session.execute(
528 where_moderated_content_visible(
529 select(Message)
530 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
531 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
532 .where(GroupChatSubscription.user_id == context.user_id)
533 .where(Message.time >= GroupChatSubscription.joined)
534 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
535 .where(Message.id > request.newest_message_id)
536 .order_by(Message.id.asc())
537 .limit(DEFAULT_PAGINATION_LENGTH + 1),
538 context,
539 GroupChat,
540 is_list_operation=False,
541 )
542 )
543 .scalars()
544 .all()
545 )
547 return conversations_pb2.GetUpdatesRes(
548 updates=[
549 conversations_pb2.Update(
550 group_chat_id=message.conversation_id,
551 message=_message_to_pb(message),
552 )
553 for message in sorted(results, key=lambda message: message.id)[:DEFAULT_PAGINATION_LENGTH]
554 ],
555 no_more=len(results) <= DEFAULT_PAGINATION_LENGTH,
556 )
558 def GetGroupChatMessages(
559 self, request: conversations_pb2.GetGroupChatMessagesReq, context: CouchersContext, session: Session
560 ) -> conversations_pb2.GetGroupChatMessagesRes:
561 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH
562 page_size = min(page_size, MAX_PAGE_SIZE)
564 results = (
565 session.execute(
566 where_moderated_content_visible(
567 select(Message)
568 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
569 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
570 .where(GroupChatSubscription.user_id == context.user_id)
571 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
572 .where(Message.time >= GroupChatSubscription.joined)
573 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
574 .where(or_(Message.id < request.last_message_id, to_bool(request.last_message_id == 0)))
575 .where(
576 or_(Message.id > GroupChatSubscription.last_seen_message_id, to_bool(request.only_unseen == 0))
577 )
578 .order_by(Message.id.desc())
579 .limit(page_size + 1),
580 context,
581 GroupChat,
582 is_list_operation=False,
583 )
584 )
585 .scalars()
586 .all()
587 )
589 return conversations_pb2.GetGroupChatMessagesRes(
590 messages=[_message_to_pb(message) for message in results[:page_size]],
591 last_message_id=results[-2].id if len(results) > 1 else 0, # TODO
592 no_more=len(results) <= page_size,
593 )
595 def MarkLastSeenGroupChat(
596 self, request: conversations_pb2.MarkLastSeenGroupChatReq, context: CouchersContext, session: Session
597 ) -> empty_pb2.Empty:
598 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
600 if not subscription: 600 ↛ 601line 600 didn't jump to line 601 because the condition on line 600 was never true
601 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
603 if not subscription.last_seen_message_id <= request.last_seen_message_id: 603 ↛ 604line 603 didn't jump to line 604 because the condition on line 603 was never true
604 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages")
606 subscription.last_seen_message_id = request.last_seen_message_id
608 mark_notifications_seen(
609 session,
610 user_id=context.user_id,
611 key=str(request.group_chat_id),
612 topic_actions=[
613 NotificationTopicAction.chat__message,
614 NotificationTopicAction.chat__missed_messages,
615 ],
616 )
618 return empty_pb2.Empty()
620 def MuteGroupChat(
621 self, request: conversations_pb2.MuteGroupChatReq, context: CouchersContext, session: Session
622 ) -> empty_pb2.Empty:
623 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
625 if not subscription: 625 ↛ 626line 625 didn't jump to line 626 because the condition on line 625 was never true
626 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
628 if request.unmute:
629 subscription.muted_until = DATETIME_MINUS_INFINITY
630 elif request.forever:
631 subscription.muted_until = DATETIME_INFINITY
632 elif request.for_duration: 632 ↛ 638line 632 didn't jump to line 638 because the condition on line 632 was always true
633 duration = request.for_duration.ToTimedelta()
634 if duration < timedelta(seconds=0): 634 ↛ 635line 634 didn't jump to line 635 because the condition on line 634 was never true
635 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_mute_past")
636 subscription.muted_until = now() + duration
638 return empty_pb2.Empty()
640 def SetGroupChatArchiveStatus(
641 self, request: conversations_pb2.SetGroupChatArchiveStatusReq, context: CouchersContext, session: Session
642 ) -> conversations_pb2.SetGroupChatArchiveStatusRes:
643 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
645 if not subscription:
646 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
648 subscription.is_archived = request.is_archived
650 return conversations_pb2.SetGroupChatArchiveStatusRes(
651 group_chat_id=request.group_chat_id,
652 is_archived=request.is_archived,
653 )
655 def SearchMessages(
656 self, request: conversations_pb2.SearchMessagesReq, context: CouchersContext, session: Session
657 ) -> conversations_pb2.SearchMessagesRes:
658 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH
659 page_size = min(page_size, MAX_PAGE_SIZE)
661 results = (
662 session.execute(
663 where_moderated_content_visible(
664 select(Message)
665 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
666 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
667 .where(GroupChatSubscription.user_id == context.user_id)
668 .where(Message.time >= GroupChatSubscription.joined)
669 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
670 .where(or_(Message.id < request.last_message_id, to_bool(request.last_message_id == 0)))
671 .where(Message.text.ilike(f"%{request.query}%"))
672 .order_by(Message.id.desc())
673 .limit(page_size + 1),
674 context,
675 GroupChat,
676 is_list_operation=True,
677 )
678 )
679 .scalars()
680 .all()
681 )
683 return conversations_pb2.SearchMessagesRes(
684 results=[
685 conversations_pb2.MessageSearchResult(
686 group_chat_id=message.conversation_id,
687 message=_message_to_pb(message),
688 )
689 for message in results[:page_size]
690 ],
691 last_message_id=results[-2].id if len(results) > 1 else 0,
692 no_more=len(results) <= page_size,
693 )
695 def CreateGroupChat(
696 self, request: conversations_pb2.CreateGroupChatReq, context: CouchersContext, session: Session
697 ) -> conversations_pb2.GroupChat:
698 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
699 if not has_completed_profile(session, user):
700 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_message")
702 recipient_user_ids = list(
703 session.execute(
704 select(User.id).where(users_visible(context)).where(User.id.in_(request.recipient_user_ids))
705 )
706 .scalars()
707 .all()
708 )
710 # make sure all requested users are visible
711 if len(recipient_user_ids) != len(request.recipient_user_ids): 711 ↛ 712line 711 didn't jump to line 712 because the condition on line 711 was never true
712 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_not_found")
714 if not recipient_user_ids: 714 ↛ 715line 714 didn't jump to line 715 because the condition on line 714 was never true
715 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "no_recipients")
717 if len(recipient_user_ids) != len(set(recipient_user_ids)): 717 ↛ 719line 717 didn't jump to line 719 because the condition on line 717 was never true
718 # make sure there's no duplicate users
719 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_recipients")
721 if context.user_id in recipient_user_ids: 721 ↛ 722line 721 didn't jump to line 722 because the condition on line 721 was never true
722 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_add_self")
724 if len(recipient_user_ids) == 1:
725 # can only have one DM at a time between any two users
726 other_user_id = recipient_user_ids[0]
728 # the following sql statement selects subscriptions that are DMs and have the same group_chat_id, and have
729 # user_id either this user or the recipient user. If you find two subscriptions to the same DM group
730 # chat, you know they already have a shared group chat
731 count = func.count(GroupChatSubscription.id).label("count")
732 if session.execute(
733 where_moderated_content_visible(
734 select(count)
735 .where(
736 or_(
737 GroupChatSubscription.user_id == context.user_id,
738 GroupChatSubscription.user_id == other_user_id,
739 )
740 )
741 .where(GroupChatSubscription.left == None)
742 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
743 .where(GroupChat.is_dm == True)
744 .group_by(GroupChatSubscription.group_chat_id)
745 .having(count == 2),
746 context,
747 GroupChat,
748 is_list_operation=False,
749 )
750 ).scalar_one_or_none():
751 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_have_dm")
753 # Check if user has been initiating chats excessively
754 if process_rate_limits_and_check_abort(
755 session=session, user_id=context.user_id, action=RateLimitAction.chat_initiation
756 ):
757 context.abort_with_error_code(
758 grpc.StatusCode.RESOURCE_EXHAUSTED,
759 "chat_initiation_rate_limit2",
760 substitutions={"count": RATE_LIMIT_HOURS},
761 )
763 group_chat = _create_chat(
764 session,
765 creator_id=context.user_id,
766 recipient_ids=request.recipient_user_ids,
767 title=request.title.value,
768 )
770 your_subscription = _get_message_subscription(session, context.user_id, group_chat.conversation_id)
772 _add_message_to_subscription(session, your_subscription, message_type=MessageType.chat_created)
774 session.flush()
776 log_event(
777 context,
778 session,
779 "group_chat.created",
780 {
781 "group_chat_id": group_chat.conversation_id,
782 "is_dm": group_chat.is_dm,
783 "recipient_count": len(request.recipient_user_ids),
784 },
785 )
787 return conversations_pb2.GroupChat(
788 group_chat_id=group_chat.conversation_id,
789 title=group_chat.title,
790 member_user_ids=_get_visible_members_for_subscription(your_subscription),
791 admin_user_ids=_get_visible_admins_for_subscription(your_subscription),
792 only_admins_invite=group_chat.only_admins_invite,
793 is_dm=group_chat.is_dm,
794 created=Timestamp_from_datetime(group_chat.conversation.created),
795 mute_info=_mute_info(your_subscription),
796 can_message=True,
797 )
799 def SendMessage(
800 self, request: conversations_pb2.SendMessageReq, context: CouchersContext, session: Session
801 ) -> empty_pb2.Empty:
802 if request.text == "": 802 ↛ 803line 802 didn't jump to line 803 because the condition on line 802 was never true
803 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
805 result = session.execute(
806 where_moderated_content_visible(
807 select(GroupChatSubscription, GroupChat)
808 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
809 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
810 .where(GroupChatSubscription.user_id == context.user_id)
811 .where(GroupChatSubscription.left == None),
812 context,
813 GroupChat,
814 is_list_operation=False,
815 )
816 ).one_or_none()
817 if not result:
818 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
820 subscription, group_chat = result._tuple()
821 if not _user_can_message(session, context, group_chat):
822 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_message_in_chat")
824 _add_message_to_subscription(session, subscription, message_type=MessageType.text, text=request.text)
826 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
827 sent_messages_counter.labels(
828 user_gender, "direct message" if subscription.group_chat.is_dm else "group chat"
829 ).inc()
830 log_event(
831 context,
832 session,
833 "message.sent",
834 {"group_chat_id": request.group_chat_id, "is_dm": subscription.group_chat.is_dm},
835 )
837 return empty_pb2.Empty()
839 def SendDirectMessage(
840 self, request: conversations_pb2.SendDirectMessageReq, context: CouchersContext, session: Session
841 ) -> conversations_pb2.SendDirectMessageRes:
842 user_id = context.user_id
843 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
845 recipient_id = request.recipient_user_id
847 if not has_completed_profile(session, user): 847 ↛ 848line 847 didn't jump to line 848 because the condition on line 847 was never true
848 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_message")
850 if not recipient_id: 850 ↛ 851line 850 didn't jump to line 851 because the condition on line 850 was never true
851 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "no_recipients")
853 recipient_user_id = session.execute(
854 select(User.id).where(users_visible(context)).where(User.id == recipient_id)
855 ).scalar_one_or_none()
857 if not recipient_user_id: 857 ↛ 858line 857 didn't jump to line 858 because the condition on line 857 was never true
858 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_not_found")
860 if user_id == recipient_id: 860 ↛ 861line 860 didn't jump to line 861 because the condition on line 860 was never true
861 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_add_self")
863 if request.text == "": 863 ↛ 864line 863 didn't jump to line 864 because the condition on line 863 was never true
864 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
866 # Look for an existing direct message (DM) chat between the two users
867 dm_chat_ids = (
868 select(GroupChatSubscription.group_chat_id)
869 .where(GroupChatSubscription.user_id.in_([user_id, recipient_id]))
870 .group_by(GroupChatSubscription.group_chat_id)
871 .having(func.count(GroupChatSubscription.user_id) == 2)
872 )
874 chat = session.execute(
875 where_moderated_content_visible(
876 select(GroupChat)
877 .where(GroupChat.is_dm == True)
878 .where(GroupChat.conversation_id.in_(dm_chat_ids))
879 .limit(1),
880 context,
881 GroupChat,
882 is_list_operation=False,
883 )
884 ).scalar_one_or_none()
886 if not chat:
887 if process_rate_limits_and_check_abort(
888 session=session, user_id=user_id, action=RateLimitAction.chat_initiation
889 ):
890 context.abort_with_error_code(
891 grpc.StatusCode.RESOURCE_EXHAUSTED,
892 "chat_initiation_rate_limit2",
893 substitutions={"count": RATE_LIMIT_HOURS},
894 )
895 chat = _create_chat(session, user_id, [recipient_id])
897 # Retrieve the sender's active subscription to the chat
898 subscription = _get_message_subscription(session, user_id, chat.conversation_id)
900 # Add the message to the conversation
901 _add_message_to_subscription(session, subscription, message_type=MessageType.text, text=request.text)
903 user_gender = session.execute(select(User.gender).where(User.id == user_id)).scalar_one()
904 sent_messages_counter.labels(user_gender, "direct message").inc()
905 log_event(
906 context,
907 session,
908 "message.sent",
909 {"group_chat_id": chat.conversation_id, "is_dm": True, "recipient_id": recipient_id},
910 )
912 session.flush()
914 return conversations_pb2.SendDirectMessageRes(group_chat_id=chat.conversation_id)
916 def EditGroupChat(
917 self, request: conversations_pb2.EditGroupChatReq, context: CouchersContext, session: Session
918 ) -> empty_pb2.Empty:
919 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
921 if not subscription:
922 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
924 if subscription.role != GroupChatRole.admin:
925 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_edit")
927 if request.HasField("title"):
928 subscription.group_chat.title = request.title.value
930 if request.HasField("only_admins_invite"): 930 ↛ 933line 930 didn't jump to line 933 because the condition on line 930 was always true
931 subscription.group_chat.only_admins_invite = request.only_admins_invite.value
933 _add_message_to_subscription(session, subscription, message_type=MessageType.chat_edited)
935 return empty_pb2.Empty()
937 def MakeGroupChatAdmin(
938 self, request: conversations_pb2.MakeGroupChatAdminReq, context: CouchersContext, session: Session
939 ) -> empty_pb2.Empty:
940 if not session.execute(
941 select(User).where(users_visible(context)).where(User.id == request.user_id)
942 ).scalar_one_or_none():
943 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
945 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
947 if not your_subscription: 947 ↛ 948line 947 didn't jump to line 948 because the condition on line 947 was never true
948 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
950 if your_subscription.role != GroupChatRole.admin:
951 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_make_admin")
953 if request.user_id == context.user_id: 953 ↛ 954line 953 didn't jump to line 954 because the condition on line 953 was never true
954 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_make_self_admin")
956 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
958 if not their_subscription: 958 ↛ 959line 958 didn't jump to line 959 because the condition on line 958 was never true
959 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_in_chat")
961 if their_subscription.role != GroupChatRole.participant:
962 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_admin")
964 their_subscription.role = GroupChatRole.admin
966 _add_message_to_subscription(
967 session, your_subscription, message_type=MessageType.user_made_admin, target_id=request.user_id
968 )
970 return empty_pb2.Empty()
972 def RemoveGroupChatAdmin(
973 self, request: conversations_pb2.RemoveGroupChatAdminReq, context: CouchersContext, session: Session
974 ) -> empty_pb2.Empty:
975 if not session.execute(
976 select(User).where(users_visible(context)).where(User.id == request.user_id)
977 ).scalar_one_or_none():
978 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
980 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
982 if not your_subscription: 982 ↛ 983line 982 didn't jump to line 983 because the condition on line 982 was never true
983 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
985 if request.user_id == context.user_id:
986 # Race condition!
987 other_admins_count = session.execute(
988 select(func.count())
989 .select_from(GroupChatSubscription)
990 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
991 .where(GroupChatSubscription.user_id != context.user_id)
992 .where(GroupChatSubscription.role == GroupChatRole.admin)
993 .where(GroupChatSubscription.left == None)
994 ).scalar_one()
995 if not other_admins_count > 0:
996 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_remove_last_admin")
998 if your_subscription.role != GroupChatRole.admin:
999 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_remove_admin")
1001 their_subscription = session.execute(
1002 select(GroupChatSubscription)
1003 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1004 .where(GroupChatSubscription.user_id == request.user_id)
1005 .where(GroupChatSubscription.left == None)
1006 .where(GroupChatSubscription.role == GroupChatRole.admin)
1007 ).scalar_one_or_none()
1009 if not their_subscription: 1009 ↛ 1010line 1009 didn't jump to line 1010 because the condition on line 1009 was never true
1010 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_admin")
1012 their_subscription.role = GroupChatRole.participant
1014 _add_message_to_subscription(
1015 session, your_subscription, message_type=MessageType.user_removed_admin, target_id=request.user_id
1016 )
1018 return empty_pb2.Empty()
1020 def InviteToGroupChat(
1021 self, request: conversations_pb2.InviteToGroupChatReq, context: CouchersContext, session: Session
1022 ) -> empty_pb2.Empty:
1023 if not session.execute(
1024 select(User).where(users_visible(context)).where(User.id == request.user_id)
1025 ).scalar_one_or_none():
1026 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1028 result = session.execute(
1029 where_moderated_content_visible(
1030 select(GroupChatSubscription, GroupChat)
1031 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
1032 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1033 .where(GroupChatSubscription.user_id == context.user_id)
1034 .where(GroupChatSubscription.left == None),
1035 context,
1036 GroupChat,
1037 is_list_operation=False,
1038 )
1039 ).one_or_none()
1041 if not result:
1042 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1044 your_subscription, group_chat = result._tuple()
1046 if request.user_id == context.user_id: 1046 ↛ 1047line 1046 didn't jump to line 1047 because the condition on line 1046 was never true
1047 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_invite_self")
1049 if your_subscription.role != GroupChatRole.admin and your_subscription.group_chat.only_admins_invite:
1050 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invite_permission_denied")
1052 if group_chat.is_dm:
1053 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_invite_to_dm")
1055 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
1057 if their_subscription: 1057 ↛ 1058line 1057 didn't jump to line 1058 because the condition on line 1057 was never true
1058 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_in_chat")
1060 # TODO: race condition!
1062 subscription = GroupChatSubscription(
1063 user_id=request.user_id,
1064 group_chat_id=your_subscription.group_chat.conversation_id,
1065 role=GroupChatRole.participant,
1066 )
1067 session.add(subscription)
1069 _add_message_to_subscription(
1070 session, your_subscription, message_type=MessageType.user_invited, target_id=request.user_id
1071 )
1073 return empty_pb2.Empty()
1075 def RemoveGroupChatUser(
1076 self, request: conversations_pb2.RemoveGroupChatUserReq, context: CouchersContext, session: Session
1077 ) -> empty_pb2.Empty:
1078 """
1079 1. Get admin info and check it's correct
1080 2. Get user data, check it's correct and remove user
1081 """
1082 # Admin info
1083 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1085 # if user info is missing
1086 if not your_subscription: 1086 ↛ 1087line 1086 didn't jump to line 1087 because the condition on line 1086 was never true
1087 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1089 # if user not admin
1090 if your_subscription.role != GroupChatRole.admin: 1090 ↛ 1091line 1090 didn't jump to line 1091 because the condition on line 1090 was never true
1091 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_remove_user")
1093 # if user wants to remove themselves
1094 if request.user_id == context.user_id: 1094 ↛ 1095line 1094 didn't jump to line 1095 because the condition on line 1094 was never true
1095 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_remove_self")
1097 # get user info
1098 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
1100 # user not found
1101 if not their_subscription:
1102 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_in_chat")
1104 _add_message_to_subscription(
1105 session, your_subscription, message_type=MessageType.user_removed, target_id=request.user_id
1106 )
1108 their_subscription.left = func.now()
1110 return empty_pb2.Empty()
1112 def LeaveGroupChat(
1113 self, request: conversations_pb2.LeaveGroupChatReq, context: CouchersContext, session: Session
1114 ) -> empty_pb2.Empty:
1115 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1117 if not subscription:
1118 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1120 if subscription.role == GroupChatRole.admin:
1121 other_admins_count = session.execute(
1122 select(func.count())
1123 .select_from(GroupChatSubscription)
1124 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1125 .where(GroupChatSubscription.user_id != context.user_id)
1126 .where(GroupChatSubscription.role == GroupChatRole.admin)
1127 .where(GroupChatSubscription.left == None)
1128 ).scalar_one()
1129 participants_count = session.execute(
1130 select(func.count())
1131 .select_from(GroupChatSubscription)
1132 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1133 .where(GroupChatSubscription.user_id != context.user_id)
1134 .where(GroupChatSubscription.role == GroupChatRole.participant)
1135 .where(GroupChatSubscription.left == None)
1136 ).scalar_one()
1137 if not (other_admins_count > 0 or participants_count == 0):
1138 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "last_admin_cant_leave")
1140 _add_message_to_subscription(session, subscription, message_type=MessageType.user_left)
1142 subscription.left = func.now()
1144 return empty_pb2.Empty()