Coverage for app/backend/src/couchers/servicers/conversations.py: 88%
316 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +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=[NotificationTopicAction.chat__message],
613 )
614 # chat__missed_messages is a summary across all chats, so it's keyed with an empty string
615 # rather than a chat id: reading any chat counts as acting on it, and it gets marked seen
616 mark_notifications_seen(
617 session,
618 user_id=context.user_id,
619 key="",
620 topic_actions=[NotificationTopicAction.chat__missed_messages],
621 )
623 return empty_pb2.Empty()
625 def MuteGroupChat(
626 self, request: conversations_pb2.MuteGroupChatReq, context: CouchersContext, session: Session
627 ) -> empty_pb2.Empty:
628 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
630 if not subscription: 630 ↛ 631line 630 didn't jump to line 631 because the condition on line 630 was never true
631 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
633 if request.unmute:
634 subscription.muted_until = DATETIME_MINUS_INFINITY
635 elif request.forever:
636 subscription.muted_until = DATETIME_INFINITY
637 elif request.for_duration: 637 ↛ 643line 637 didn't jump to line 643 because the condition on line 637 was always true
638 duration = request.for_duration.ToTimedelta()
639 if duration < timedelta(seconds=0): 639 ↛ 640line 639 didn't jump to line 640 because the condition on line 639 was never true
640 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_mute_past")
641 subscription.muted_until = now() + duration
643 return empty_pb2.Empty()
645 def SetGroupChatArchiveStatus(
646 self, request: conversations_pb2.SetGroupChatArchiveStatusReq, context: CouchersContext, session: Session
647 ) -> conversations_pb2.SetGroupChatArchiveStatusRes:
648 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
650 if not subscription:
651 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
653 subscription.is_archived = request.is_archived
655 return conversations_pb2.SetGroupChatArchiveStatusRes(
656 group_chat_id=request.group_chat_id,
657 is_archived=request.is_archived,
658 )
660 def SearchMessages(
661 self, request: conversations_pb2.SearchMessagesReq, context: CouchersContext, session: Session
662 ) -> conversations_pb2.SearchMessagesRes:
663 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH
664 page_size = min(page_size, MAX_PAGE_SIZE)
666 results = (
667 session.execute(
668 where_moderated_content_visible(
669 select(Message)
670 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
671 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
672 .where(GroupChatSubscription.user_id == context.user_id)
673 .where(Message.time >= GroupChatSubscription.joined)
674 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
675 .where(or_(Message.id < request.last_message_id, to_bool(request.last_message_id == 0)))
676 .where(Message.text.ilike(f"%{request.query}%"))
677 .order_by(Message.id.desc())
678 .limit(page_size + 1),
679 context,
680 GroupChat,
681 is_list_operation=True,
682 )
683 )
684 .scalars()
685 .all()
686 )
688 return conversations_pb2.SearchMessagesRes(
689 results=[
690 conversations_pb2.MessageSearchResult(
691 group_chat_id=message.conversation_id,
692 message=_message_to_pb(message),
693 )
694 for message in results[:page_size]
695 ],
696 last_message_id=results[-2].id if len(results) > 1 else 0,
697 no_more=len(results) <= page_size,
698 )
700 def CreateGroupChat(
701 self, request: conversations_pb2.CreateGroupChatReq, context: CouchersContext, session: Session
702 ) -> conversations_pb2.GroupChat:
703 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
704 if not has_completed_profile(session, user):
705 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_message")
707 recipient_user_ids = list(
708 session.execute(
709 select(User.id).where(users_visible(context)).where(User.id.in_(request.recipient_user_ids))
710 )
711 .scalars()
712 .all()
713 )
715 # make sure all requested users are visible
716 if len(recipient_user_ids) != len(request.recipient_user_ids): 716 ↛ 717line 716 didn't jump to line 717 because the condition on line 716 was never true
717 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_not_found")
719 if not recipient_user_ids: 719 ↛ 720line 719 didn't jump to line 720 because the condition on line 719 was never true
720 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "no_recipients")
722 if len(recipient_user_ids) != len(set(recipient_user_ids)): 722 ↛ 724line 722 didn't jump to line 724 because the condition on line 722 was never true
723 # make sure there's no duplicate users
724 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_recipients")
726 if context.user_id in recipient_user_ids: 726 ↛ 727line 726 didn't jump to line 727 because the condition on line 726 was never true
727 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_add_self")
729 if len(recipient_user_ids) == 1:
730 # can only have one DM at a time between any two users
731 other_user_id = recipient_user_ids[0]
733 # the following sql statement selects subscriptions that are DMs and have the same group_chat_id, and have
734 # user_id either this user or the recipient user. If you find two subscriptions to the same DM group
735 # chat, you know they already have a shared group chat
736 count = func.count(GroupChatSubscription.id).label("count")
737 if session.execute(
738 where_moderated_content_visible(
739 select(count)
740 .where(
741 or_(
742 GroupChatSubscription.user_id == context.user_id,
743 GroupChatSubscription.user_id == other_user_id,
744 )
745 )
746 .where(GroupChatSubscription.left == None)
747 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
748 .where(GroupChat.is_dm == True)
749 .group_by(GroupChatSubscription.group_chat_id)
750 .having(count == 2),
751 context,
752 GroupChat,
753 is_list_operation=False,
754 )
755 ).scalar_one_or_none():
756 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_have_dm")
758 # Check if user has been initiating chats excessively
759 if process_rate_limits_and_check_abort(
760 session=session, user_id=context.user_id, action=RateLimitAction.chat_initiation
761 ):
762 context.abort_with_error_code(
763 grpc.StatusCode.RESOURCE_EXHAUSTED,
764 "chat_initiation_rate_limit2",
765 substitutions={"count": RATE_LIMIT_HOURS},
766 )
768 group_chat = _create_chat(
769 session,
770 creator_id=context.user_id,
771 recipient_ids=request.recipient_user_ids,
772 title=request.title.value,
773 )
775 your_subscription = _get_message_subscription(session, context.user_id, group_chat.conversation_id)
777 _add_message_to_subscription(session, your_subscription, message_type=MessageType.chat_created)
779 session.flush()
781 log_event(
782 context,
783 session,
784 "group_chat.created",
785 {
786 "group_chat_id": group_chat.conversation_id,
787 "is_dm": group_chat.is_dm,
788 "recipient_count": len(request.recipient_user_ids),
789 },
790 )
792 return conversations_pb2.GroupChat(
793 group_chat_id=group_chat.conversation_id,
794 title=group_chat.title,
795 member_user_ids=_get_visible_members_for_subscription(your_subscription),
796 admin_user_ids=_get_visible_admins_for_subscription(your_subscription),
797 only_admins_invite=group_chat.only_admins_invite,
798 is_dm=group_chat.is_dm,
799 created=Timestamp_from_datetime(group_chat.conversation.created),
800 mute_info=_mute_info(your_subscription),
801 can_message=True,
802 )
804 def SendMessage(
805 self, request: conversations_pb2.SendMessageReq, context: CouchersContext, session: Session
806 ) -> empty_pb2.Empty:
807 if request.text == "": 807 ↛ 808line 807 didn't jump to line 808 because the condition on line 807 was never true
808 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
810 result = session.execute(
811 where_moderated_content_visible(
812 select(GroupChatSubscription, GroupChat)
813 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
814 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
815 .where(GroupChatSubscription.user_id == context.user_id)
816 .where(GroupChatSubscription.left == None),
817 context,
818 GroupChat,
819 is_list_operation=False,
820 )
821 ).one_or_none()
822 if not result:
823 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
825 subscription, group_chat = result._tuple()
826 if not _user_can_message(session, context, group_chat):
827 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_message_in_chat")
829 _add_message_to_subscription(session, subscription, message_type=MessageType.text, text=request.text)
831 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
832 sent_messages_counter.labels(
833 user_gender, "direct message" if subscription.group_chat.is_dm else "group chat"
834 ).inc()
835 log_event(
836 context,
837 session,
838 "message.sent",
839 {"group_chat_id": request.group_chat_id, "is_dm": subscription.group_chat.is_dm},
840 )
842 return empty_pb2.Empty()
844 def SendDirectMessage(
845 self, request: conversations_pb2.SendDirectMessageReq, context: CouchersContext, session: Session
846 ) -> conversations_pb2.SendDirectMessageRes:
847 user_id = context.user_id
848 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
850 recipient_id = request.recipient_user_id
852 if not has_completed_profile(session, user): 852 ↛ 853line 852 didn't jump to line 853 because the condition on line 852 was never true
853 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_message")
855 if not recipient_id: 855 ↛ 856line 855 didn't jump to line 856 because the condition on line 855 was never true
856 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "no_recipients")
858 recipient_user_id = session.execute(
859 select(User.id).where(users_visible(context)).where(User.id == recipient_id)
860 ).scalar_one_or_none()
862 if not recipient_user_id: 862 ↛ 863line 862 didn't jump to line 863 because the condition on line 862 was never true
863 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_not_found")
865 if user_id == recipient_id: 865 ↛ 866line 865 didn't jump to line 866 because the condition on line 865 was never true
866 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_add_self")
868 if request.text == "": 868 ↛ 869line 868 didn't jump to line 869 because the condition on line 868 was never true
869 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
871 # Look for an existing direct message (DM) chat between the two users
872 dm_chat_ids = (
873 select(GroupChatSubscription.group_chat_id)
874 .where(GroupChatSubscription.user_id.in_([user_id, recipient_id]))
875 .group_by(GroupChatSubscription.group_chat_id)
876 .having(func.count(GroupChatSubscription.user_id) == 2)
877 )
879 chat = session.execute(
880 where_moderated_content_visible(
881 select(GroupChat)
882 .where(GroupChat.is_dm == True)
883 .where(GroupChat.conversation_id.in_(dm_chat_ids))
884 .limit(1),
885 context,
886 GroupChat,
887 is_list_operation=False,
888 )
889 ).scalar_one_or_none()
891 if not chat:
892 if process_rate_limits_and_check_abort(
893 session=session, user_id=user_id, action=RateLimitAction.chat_initiation
894 ):
895 context.abort_with_error_code(
896 grpc.StatusCode.RESOURCE_EXHAUSTED,
897 "chat_initiation_rate_limit2",
898 substitutions={"count": RATE_LIMIT_HOURS},
899 )
900 chat = _create_chat(session, user_id, [recipient_id])
902 # Retrieve the sender's active subscription to the chat
903 subscription = _get_message_subscription(session, user_id, chat.conversation_id)
905 # Add the message to the conversation
906 _add_message_to_subscription(session, subscription, message_type=MessageType.text, text=request.text)
908 user_gender = session.execute(select(User.gender).where(User.id == user_id)).scalar_one()
909 sent_messages_counter.labels(user_gender, "direct message").inc()
910 log_event(
911 context,
912 session,
913 "message.sent",
914 {"group_chat_id": chat.conversation_id, "is_dm": True, "recipient_id": recipient_id},
915 )
917 session.flush()
919 return conversations_pb2.SendDirectMessageRes(group_chat_id=chat.conversation_id)
921 def EditGroupChat(
922 self, request: conversations_pb2.EditGroupChatReq, context: CouchersContext, session: Session
923 ) -> empty_pb2.Empty:
924 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
926 if not subscription:
927 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
929 if subscription.role != GroupChatRole.admin:
930 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_edit")
932 if request.HasField("title"):
933 subscription.group_chat.title = request.title.value
935 if request.HasField("only_admins_invite"): 935 ↛ 938line 935 didn't jump to line 938 because the condition on line 935 was always true
936 subscription.group_chat.only_admins_invite = request.only_admins_invite.value
938 _add_message_to_subscription(session, subscription, message_type=MessageType.chat_edited)
940 return empty_pb2.Empty()
942 def MakeGroupChatAdmin(
943 self, request: conversations_pb2.MakeGroupChatAdminReq, context: CouchersContext, session: Session
944 ) -> empty_pb2.Empty:
945 if not session.execute(
946 select(User).where(users_visible(context)).where(User.id == request.user_id)
947 ).scalar_one_or_none():
948 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
950 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
952 if not your_subscription: 952 ↛ 953line 952 didn't jump to line 953 because the condition on line 952 was never true
953 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
955 if your_subscription.role != GroupChatRole.admin:
956 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_make_admin")
958 if request.user_id == context.user_id: 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, "cant_make_self_admin")
961 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
963 if not their_subscription: 963 ↛ 964line 963 didn't jump to line 964 because the condition on line 963 was never true
964 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_in_chat")
966 if their_subscription.role != GroupChatRole.participant:
967 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_admin")
969 their_subscription.role = GroupChatRole.admin
971 _add_message_to_subscription(
972 session, your_subscription, message_type=MessageType.user_made_admin, target_id=request.user_id
973 )
975 return empty_pb2.Empty()
977 def RemoveGroupChatAdmin(
978 self, request: conversations_pb2.RemoveGroupChatAdminReq, context: CouchersContext, session: Session
979 ) -> empty_pb2.Empty:
980 if not session.execute(
981 select(User).where(users_visible(context)).where(User.id == request.user_id)
982 ).scalar_one_or_none():
983 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
985 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
987 if not your_subscription: 987 ↛ 988line 987 didn't jump to line 988 because the condition on line 987 was never true
988 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
990 if request.user_id == context.user_id:
991 # Race condition!
992 other_admins_count = session.execute(
993 select(func.count())
994 .select_from(GroupChatSubscription)
995 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
996 .where(GroupChatSubscription.user_id != context.user_id)
997 .where(GroupChatSubscription.role == GroupChatRole.admin)
998 .where(GroupChatSubscription.left == None)
999 ).scalar_one()
1000 if not other_admins_count > 0:
1001 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_remove_last_admin")
1003 if your_subscription.role != GroupChatRole.admin:
1004 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_remove_admin")
1006 their_subscription = session.execute(
1007 select(GroupChatSubscription)
1008 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1009 .where(GroupChatSubscription.user_id == request.user_id)
1010 .where(GroupChatSubscription.left == None)
1011 .where(GroupChatSubscription.role == GroupChatRole.admin)
1012 ).scalar_one_or_none()
1014 if not their_subscription: 1014 ↛ 1015line 1014 didn't jump to line 1015 because the condition on line 1014 was never true
1015 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_admin")
1017 their_subscription.role = GroupChatRole.participant
1019 _add_message_to_subscription(
1020 session, your_subscription, message_type=MessageType.user_removed_admin, target_id=request.user_id
1021 )
1023 return empty_pb2.Empty()
1025 def InviteToGroupChat(
1026 self, request: conversations_pb2.InviteToGroupChatReq, context: CouchersContext, session: Session
1027 ) -> empty_pb2.Empty:
1028 if not session.execute(
1029 select(User).where(users_visible(context)).where(User.id == request.user_id)
1030 ).scalar_one_or_none():
1031 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1033 result = session.execute(
1034 where_moderated_content_visible(
1035 select(GroupChatSubscription, GroupChat)
1036 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
1037 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1038 .where(GroupChatSubscription.user_id == context.user_id)
1039 .where(GroupChatSubscription.left == None),
1040 context,
1041 GroupChat,
1042 is_list_operation=False,
1043 )
1044 ).one_or_none()
1046 if not result:
1047 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1049 your_subscription, group_chat = result._tuple()
1051 if request.user_id == context.user_id: 1051 ↛ 1052line 1051 didn't jump to line 1052 because the condition on line 1051 was never true
1052 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_invite_self")
1054 if your_subscription.role != GroupChatRole.admin and your_subscription.group_chat.only_admins_invite:
1055 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invite_permission_denied")
1057 if group_chat.is_dm:
1058 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_invite_to_dm")
1060 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
1062 if their_subscription: 1062 ↛ 1063line 1062 didn't jump to line 1063 because the condition on line 1062 was never true
1063 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_in_chat")
1065 # TODO: race condition!
1067 subscription = GroupChatSubscription(
1068 user_id=request.user_id,
1069 group_chat_id=your_subscription.group_chat.conversation_id,
1070 role=GroupChatRole.participant,
1071 )
1072 session.add(subscription)
1074 _add_message_to_subscription(
1075 session, your_subscription, message_type=MessageType.user_invited, target_id=request.user_id
1076 )
1078 return empty_pb2.Empty()
1080 def RemoveGroupChatUser(
1081 self, request: conversations_pb2.RemoveGroupChatUserReq, context: CouchersContext, session: Session
1082 ) -> empty_pb2.Empty:
1083 """
1084 1. Get admin info and check it's correct
1085 2. Get user data, check it's correct and remove user
1086 """
1087 # Admin info
1088 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1090 # if user info is missing
1091 if not your_subscription: 1091 ↛ 1092line 1091 didn't jump to line 1092 because the condition on line 1091 was never true
1092 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1094 # if user not admin
1095 if your_subscription.role != GroupChatRole.admin: 1095 ↛ 1096line 1095 didn't jump to line 1096 because the condition on line 1095 was never true
1096 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_remove_user")
1098 # if user wants to remove themselves
1099 if request.user_id == context.user_id: 1099 ↛ 1100line 1099 didn't jump to line 1100 because the condition on line 1099 was never true
1100 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_remove_self")
1102 # get user info
1103 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
1105 # user not found
1106 if not their_subscription:
1107 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_in_chat")
1109 _add_message_to_subscription(
1110 session, your_subscription, message_type=MessageType.user_removed, target_id=request.user_id
1111 )
1113 their_subscription.left = func.now()
1115 return empty_pb2.Empty()
1117 def LeaveGroupChat(
1118 self, request: conversations_pb2.LeaveGroupChatReq, context: CouchersContext, session: Session
1119 ) -> empty_pb2.Empty:
1120 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1122 if not subscription:
1123 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1125 if subscription.role == GroupChatRole.admin:
1126 other_admins_count = session.execute(
1127 select(func.count())
1128 .select_from(GroupChatSubscription)
1129 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1130 .where(GroupChatSubscription.user_id != context.user_id)
1131 .where(GroupChatSubscription.role == GroupChatRole.admin)
1132 .where(GroupChatSubscription.left == None)
1133 ).scalar_one()
1134 participants_count = session.execute(
1135 select(func.count())
1136 .select_from(GroupChatSubscription)
1137 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1138 .where(GroupChatSubscription.user_id != context.user_id)
1139 .where(GroupChatSubscription.role == GroupChatRole.participant)
1140 .where(GroupChatSubscription.left == None)
1141 ).scalar_one()
1142 if not (other_admins_count > 0 or participants_count == 0):
1143 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "last_admin_cant_leave")
1145 _add_message_to_subscription(session, subscription, message_type=MessageType.user_left)
1147 subscription.left = func.now()
1149 return empty_pb2.Empty()