Coverage for app/backend/src/couchers/servicers/conversations.py: 90%
452 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-22 16:01 +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 ColumnElement, Select, select, union_all
9from sqlalchemy.orm import Session, contains_eager
10from sqlalchemy.sql import and_, case, func, literal, 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.crypto import decrypt_page_token, encrypt_page_token
15from couchers.db import session_scope
16from couchers.event_log import log_event
17from couchers.helpers.completed_profile import has_completed_profile
18from couchers.jobs.enqueue import queue_job
19from couchers.metrics import sent_messages_counter
20from couchers.models import (
21 Conversation,
22 GroupChat,
23 GroupChatRole,
24 GroupChatSubscription,
25 HostRequest,
26 Message,
27 MessageType,
28 ModerationObjectType,
29 RateLimitAction,
30 User,
31)
32from couchers.models.notifications import NotificationTopicAction
33from couchers.moderation.utils import create_moderation
34from couchers.notifications.notify import mark_notifications_seen, notify
35from couchers.proto import conversations_pb2, conversations_pb2_grpc, notification_data_pb2
36from couchers.proto.internal import jobs_pb2
37from couchers.rate_limits.check import process_rate_limits_and_check_abort
38from couchers.rate_limits.definitions import RATE_LIMIT_HOURS
39from couchers.servicers.api import user_model_to_pb
40from couchers.servicers.requests import hostrequeststatus2api
41from couchers.sql import to_bool, users_visible, where_moderated_content_visible, where_users_column_visible
42from couchers.utils import Timestamp_from_datetime, date_to_api, now
44# topic actions whose notifications are marked seen alongside a host request being read
45_HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS = [
46 NotificationTopicAction.host_request__create,
47 NotificationTopicAction.host_request__accept,
48 NotificationTopicAction.host_request__reject,
49 NotificationTopicAction.host_request__confirm,
50 NotificationTopicAction.host_request__cancel,
51 NotificationTopicAction.host_request__message,
52 NotificationTopicAction.host_request__missed_messages,
53 NotificationTopicAction.host_request__reminder,
54]
56logger = logging.getLogger(__name__)
58# TODO: Still needs custom pagination: GetUpdates
59DEFAULT_PAGINATION_LENGTH = 20
60MAX_PAGE_SIZE = 50
63def _message_to_pb(message: Message) -> conversations_pb2.Message:
64 """
65 Turns the given message to a protocol buffer
66 """
67 if message.is_normal_message:
68 return conversations_pb2.Message(
69 message_id=message.id,
70 author_user_id=message.author_id,
71 time=Timestamp_from_datetime(message.time),
72 text=conversations_pb2.MessageContentText(text=message.text),
73 )
74 else:
75 return conversations_pb2.Message(
76 message_id=message.id,
77 author_user_id=message.author_id,
78 time=Timestamp_from_datetime(message.time),
79 chat_created=(
80 conversations_pb2.MessageContentChatCreated()
81 if message.message_type == MessageType.chat_created
82 else None
83 ),
84 chat_edited=(
85 conversations_pb2.MessageContentChatEdited()
86 if message.message_type == MessageType.chat_edited
87 else None
88 ),
89 user_invited=(
90 conversations_pb2.MessageContentUserInvited(target_user_id=message.target_id)
91 if message.message_type == MessageType.user_invited
92 else None
93 ),
94 user_left=(
95 conversations_pb2.MessageContentUserLeft() if message.message_type == MessageType.user_left else None
96 ),
97 user_made_admin=(
98 conversations_pb2.MessageContentUserMadeAdmin(target_user_id=message.target_id)
99 if message.message_type == MessageType.user_made_admin
100 else None
101 ),
102 user_removed_admin=(
103 conversations_pb2.MessageContentUserRemovedAdmin(target_user_id=message.target_id)
104 if message.message_type == MessageType.user_removed_admin
105 else None
106 ),
107 group_chat_user_removed=(
108 conversations_pb2.MessageContentUserRemoved(target_user_id=message.target_id)
109 if message.message_type == MessageType.user_removed
110 else None
111 ),
112 host_request_status_changed=(
113 conversations_pb2.MessageContentHostRequestStatusChanged(
114 status=hostrequeststatus2api[message.host_request_status_target] # type: ignore[index]
115 )
116 if message.message_type == MessageType.host_request_status_changed
117 else None
118 ),
119 )
122def _get_visible_members_for_subscription(subscription: GroupChatSubscription) -> list[int]:
123 """
124 If a user leaves a group chat, they shouldn't be able to see who's added
125 after they left
126 """
127 if not subscription.left:
128 # still in the chat, we see everyone with a current subscription
129 return [sub.user_id for sub in subscription.group_chat.subscriptions.where(GroupChatSubscription.left == None)]
130 else:
131 # not in chat anymore, see everyone who was in chat when we left
132 return [
133 sub.user_id
134 for sub in subscription.group_chat.subscriptions.where(
135 GroupChatSubscription.joined <= subscription.left
136 ).where(or_(GroupChatSubscription.left >= subscription.left, GroupChatSubscription.left == None))
137 ]
140def _get_visible_admins_for_subscription(subscription: GroupChatSubscription) -> list[int]:
141 """
142 If a user leaves a group chat, they shouldn't be able to see who's added
143 after they left
144 """
145 if not subscription.left:
146 # still in the chat, we see everyone with a current subscription
147 return [
148 sub.user_id
149 for sub in subscription.group_chat.subscriptions.where(GroupChatSubscription.left == None).where(
150 GroupChatSubscription.role == GroupChatRole.admin
151 )
152 ]
153 else:
154 # not in chat anymore, see everyone who was in chat when we left
155 return [
156 sub.user_id
157 for sub in subscription.group_chat.subscriptions.where(GroupChatSubscription.role == GroupChatRole.admin)
158 .where(GroupChatSubscription.joined <= subscription.left)
159 .where(or_(GroupChatSubscription.left >= subscription.left, GroupChatSubscription.left == None))
160 ]
163def _user_can_message(session: Session, context: CouchersContext, group_chat: GroupChat) -> bool:
164 """
165 If it is a true group chat (not a DM), user can always message. For a DM, user can message if the other participant
166 - Is not deleted/banned
167 - Has not been blocked by the user or is blocking the user
168 - Has not left the chat
169 """
170 if not group_chat.is_dm:
171 return True
173 query = select(
174 where_users_column_visible(
175 select(GroupChatSubscription)
176 .where(GroupChatSubscription.user_id != context.user_id)
177 .where(GroupChatSubscription.group_chat_id == group_chat.conversation_id)
178 .where(GroupChatSubscription.left == None),
179 context=context,
180 column=GroupChatSubscription.user_id,
181 ).exists()
182 )
183 return session.execute(query).scalar_one()
186def generate_message_notifications(payload: jobs_pb2.GenerateMessageNotificationsPayload) -> None:
187 """
188 Background job to generate notifications for a message sent to a group chat
189 """
190 logger.info(f"Fanning notifications for message_id = {payload.message_id}")
192 with session_scope() as session:
193 message, group_chat = session.execute(
194 select(Message, GroupChat)
195 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
196 .where(Message.id == payload.message_id)
197 ).one()
199 if message.message_type != MessageType.text:
200 logger.info(f"Not a text message, not notifying. message_id = {payload.message_id}")
201 return
203 context = make_background_user_context(user_id=message.author_id)
204 user_ids_to_notify = (
205 session.execute(
206 where_users_column_visible(
207 select(GroupChatSubscription.user_id)
208 .where(GroupChatSubscription.group_chat_id == message.conversation_id)
209 .where(GroupChatSubscription.user_id != message.author_id)
210 .where(GroupChatSubscription.joined <= message.time)
211 .where(or_(GroupChatSubscription.left == None, GroupChatSubscription.left >= message.time))
212 .where(not_(GroupChatSubscription.is_muted)),
213 context=context,
214 column=GroupChatSubscription.user_id,
215 )
216 )
217 .scalars()
218 .all()
219 )
221 for user_id in user_ids_to_notify:
222 notify(
223 session,
224 user_id=user_id,
225 topic_action=NotificationTopicAction.chat__message,
226 key=str(message.conversation_id),
227 data=notification_data_pb2.ChatMessage(
228 author=user_model_to_pb(
229 message.author,
230 session,
231 make_notification_user_context(user_id=user_id),
232 ),
233 text=message.text,
234 group_chat_id=message.conversation_id,
235 group_chat_title=group_chat.title or None,
236 # unseen_count irrelevant for this notification
237 ),
238 moderation_state_id=group_chat.moderation_state_id,
239 )
242def _add_message_to_subscription(session: Session, subscription: GroupChatSubscription, **kwargs: Any) -> Message:
243 """
244 Creates a new message for a subscription, from the user whose subscription that is. Updates last seen message id
246 Specify the keyword args for Message
247 """
248 message = Message(conversation_id=subscription.group_chat.conversation.id, author_id=subscription.user_id, **kwargs)
250 session.add(message)
251 session.flush()
253 subscription.last_seen_message_id = message.id
255 queue_job(
256 session,
257 job=generate_message_notifications,
258 payload=jobs_pb2.GenerateMessageNotificationsPayload(
259 message_id=message.id,
260 ),
261 )
263 return message
266def _create_chat(
267 session: Session,
268 creator_id: int,
269 recipient_ids: Sequence[int],
270 title: str | None = None,
271 only_admins_invite: bool = True,
272) -> GroupChat:
273 conversation = Conversation()
274 session.add(conversation)
275 session.flush()
277 # Create moderation state for UMS (starts as SHADOWED)
278 moderation_state = create_moderation(
279 session=session,
280 object_type=ModerationObjectType.group_chat,
281 object_id=conversation.id,
282 creator_user_id=creator_id,
283 )
285 chat = GroupChat(
286 conversation_id=conversation.id,
287 title=title,
288 creator_id=creator_id,
289 is_dm=True if len(recipient_ids) == 1 else False,
290 only_admins_invite=only_admins_invite,
291 moderation_state_id=moderation_state.id,
292 )
293 session.add(chat)
294 session.flush()
296 creator_subscription = GroupChatSubscription(
297 user_id=creator_id,
298 group_chat_id=chat.conversation_id,
299 role=GroupChatRole.admin,
300 )
301 session.add(creator_subscription)
303 for uid in recipient_ids:
304 session.add(
305 GroupChatSubscription(
306 user_id=uid,
307 group_chat_id=chat.conversation_id,
308 role=GroupChatRole.participant,
309 )
310 )
312 return chat
315def _get_message_subscription(session: Session, user_id: int, conversation_id: int) -> GroupChatSubscription:
316 subscription = session.execute(
317 select(GroupChatSubscription)
318 .where(GroupChatSubscription.group_chat_id == conversation_id)
319 .where(GroupChatSubscription.user_id == user_id)
320 .where(GroupChatSubscription.left == None)
321 ).scalar_one_or_none()
323 return cast(GroupChatSubscription, subscription)
326def _get_visible_message_subscription(
327 session: Session, context: CouchersContext, conversation_id: int
328) -> GroupChatSubscription:
329 """Get subscription with visibility filtering"""
330 subscription = session.execute(
331 where_moderated_content_visible(
332 select(GroupChatSubscription)
333 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
334 .where(GroupChatSubscription.group_chat_id == conversation_id)
335 .where(GroupChatSubscription.user_id == context.user_id)
336 .where(GroupChatSubscription.left == None),
337 context,
338 GroupChat,
339 is_list_operation=False,
340 )
341 ).scalar_one_or_none()
343 return cast(GroupChatSubscription, subscription)
346def _unseen_message_count(session: Session, subscription_id: int) -> int:
347 query = (
348 select(func.count())
349 .select_from(Message)
350 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
351 .where(GroupChatSubscription.id == subscription_id)
352 .where(Message.id > GroupChatSubscription.last_seen_message_id)
353 )
354 return session.execute(query).scalar_one()
357def _mute_info(subscription: GroupChatSubscription) -> conversations_pb2.MuteInfo:
358 (muted, muted_until) = subscription.muted_display()
359 return conversations_pb2.MuteInfo(
360 muted=muted,
361 muted_until=Timestamp_from_datetime(muted_until) if muted_until else None,
362 )
365def _group_chat_to_pb(
366 session: Session,
367 context: CouchersContext,
368 group_chat: GroupChat,
369 subscription: GroupChatSubscription,
370 message: Message | None,
371 unseen_message_count: int,
372) -> conversations_pb2.GroupChat:
373 """Build the GroupChat protobuf from a (chat, subscription, latest message) row."""
374 return conversations_pb2.GroupChat(
375 group_chat_id=group_chat.conversation_id,
376 title=group_chat.title, # TODO: proper title for DMs, etc
377 member_user_ids=_get_visible_members_for_subscription(subscription),
378 admin_user_ids=_get_visible_admins_for_subscription(subscription),
379 only_admins_invite=group_chat.only_admins_invite,
380 is_dm=group_chat.is_dm,
381 created=Timestamp_from_datetime(group_chat.conversation.created),
382 unseen_message_count=unseen_message_count,
383 last_seen_message_id=subscription.last_seen_message_id,
384 latest_message=_message_to_pb(message) if message else None,
385 mute_info=_mute_info(subscription),
386 can_message=_user_can_message(session, context, group_chat),
387 is_archived=subscription.is_archived,
388 )
391def _host_request_viewer_last_seen(host_request: HostRequest, user_id: int) -> int:
392 """The viewer's role-specific last-seen message id for a host request."""
393 if host_request.initiator_user_id == user_id:
394 return host_request.initiator_last_seen_message_id
395 if host_request.recipient_user_id == user_id: 395 ↛ 397line 395 didn't jump to line 397 because the condition on line 395 was always true
396 return host_request.recipient_last_seen_message_id
397 raise ValueError(f"User {user_id} is not a party to host request {host_request.conversation_id}")
400def _host_request_thread_to_pb(
401 host_request: HostRequest,
402 conversation: Conversation,
403 message: Message | None,
404 user_id: int,
405 unseen_message_count: int,
406) -> conversations_pb2.HostRequestThread:
407 """
408 Build the HostRequestThread protobuf. surfer/host are ROLE-based: for a
409 public-trip offer the roles are reversed relative to initiator/recipient
410 (initiator = offering host, recipient = traveller).
411 """
412 if host_request.public_trip_id is not None:
413 surfer_user_id = host_request.recipient_user_id
414 host_user_id = host_request.initiator_user_id
415 else:
416 surfer_user_id = host_request.initiator_user_id
417 host_user_id = host_request.recipient_user_id
418 return conversations_pb2.HostRequestThread(
419 host_request_id=host_request.conversation_id,
420 surfer_user_id=surfer_user_id,
421 host_user_id=host_user_id,
422 status=hostrequeststatus2api[host_request.status],
423 created=Timestamp_from_datetime(conversation.created),
424 from_date=date_to_api(host_request.from_date),
425 to_date=date_to_api(host_request.to_date),
426 last_seen_message_id=_host_request_viewer_last_seen(host_request, user_id),
427 latest_message=_message_to_pb(message) if message else None,
428 hosting_city=host_request.hosting_city,
429 is_archived=(
430 host_request.is_initiator_archived
431 if host_request.initiator_user_id == user_id
432 else host_request.is_recipient_archived
433 ),
434 public_trip_id=host_request.public_trip_id,
435 unseen_message_count=unseen_message_count,
436 )
439def _group_chat_candidate_query(
440 context: CouchersContext, only_archived: bool | None, unread: bool
441) -> Select[tuple[int, int, str]]:
442 """
443 Candidate (conversation_id, latest_message_id, kind) rows for the user's group
444 chats (incl. DMs), with moderation + archived + unread filtering applied.
445 """
446 latest_by_group_chat = (
447 select(
448 GroupChatSubscription.group_chat_id.label("conversation_id"),
449 func.max(Message.id).label("latest_message_id"),
450 )
451 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
452 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
453 .where(GroupChatSubscription.user_id == context.user_id)
454 .where(Message.time >= GroupChatSubscription.joined)
455 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
456 .group_by(GroupChatSubscription.group_chat_id)
457 )
458 if only_archived is not None:
459 latest_by_group_chat = latest_by_group_chat.where(GroupChatSubscription.is_archived == only_archived)
460 if unread:
461 # restrict to chats with at least one message newer than the user's last-seen
462 latest_by_group_chat = latest_by_group_chat.where(Message.id > GroupChatSubscription.last_seen_message_id)
463 visible_group_chats = where_moderated_content_visible(
464 latest_by_group_chat, context, GroupChat, is_list_operation=True
465 ).subquery()
467 return select(
468 visible_group_chats.c.conversation_id.label("conversation_id"),
469 visible_group_chats.c.latest_message_id.label("latest_message_id"),
470 literal("group_chat").label("kind"),
471 )
474def _host_request_candidate_query(
475 context: CouchersContext, party_predicate: ColumnElement[bool], only_archived: bool | None, unread: bool
476) -> Select[tuple[int, int, str]]:
477 """
478 Candidate (conversation_id, latest_message_id, kind) rows for the user's host
479 requests / public-trip offers matching party_predicate, with visibility +
480 moderation + archived + unread filtering applied.
481 """
482 viewer_last_seen_message_id = case(
483 (HostRequest.initiator_user_id == context.user_id, HostRequest.initiator_last_seen_message_id),
484 else_=HostRequest.recipient_last_seen_message_id,
485 )
486 candidate_query = (
487 select(
488 HostRequest.conversation_id.label("conversation_id"),
489 func.max(Message.id).label("latest_message_id"),
490 literal("host_request").label("kind"),
491 )
492 .join(Message, Message.conversation_id == HostRequest.conversation_id)
493 .where(party_predicate)
494 .group_by(HostRequest.conversation_id)
495 )
496 if only_archived is not None:
497 candidate_query = candidate_query.where(
498 or_(
499 and_(
500 HostRequest.initiator_user_id == context.user_id,
501 HostRequest.is_initiator_archived == only_archived,
502 ),
503 and_(
504 HostRequest.recipient_user_id == context.user_id,
505 HostRequest.is_recipient_archived == only_archived,
506 ),
507 )
508 )
509 if unread:
510 candidate_query = candidate_query.having(func.max(Message.id) > viewer_last_seen_message_id)
511 candidate_query = where_users_column_visible(candidate_query, context, HostRequest.initiator_user_id)
512 candidate_query = where_users_column_visible(candidate_query, context, HostRequest.recipient_user_id)
513 candidate_query = where_moderated_content_visible(candidate_query, context, HostRequest, is_list_operation=True)
514 return candidate_query
517def _host_request_party_predicate(context: CouchersContext, thread_filter: int) -> ColumnElement[bool]:
518 """Role-based membership predicate for the host-request leg, per filter."""
519 viewer_id = context.user_id
520 if thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_HOSTING:
521 # I am the host of the stay: normal request I received, or an offer I sent
522 return or_(
523 and_(HostRequest.public_trip_id.is_(None), HostRequest.recipient_user_id == viewer_id),
524 and_(HostRequest.public_trip_id.isnot(None), HostRequest.initiator_user_id == viewer_id),
525 )
526 if thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_SURFING:
527 # I am the guest of the stay: normal request I sent, or an offer I received
528 return or_(
529 and_(HostRequest.public_trip_id.is_(None), HostRequest.initiator_user_id == viewer_id),
530 and_(HostRequest.public_trip_id.isnot(None), HostRequest.recipient_user_id == viewer_id),
531 )
532 if thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_MY_PUBLIC_TRIPS:
533 # incoming offers on my public trips (I'm the traveller = recipient)
534 return and_(HostRequest.public_trip_id.isnot(None), HostRequest.recipient_user_id == viewer_id)
535 # ALL / UNREAD: any host request I'm party to
536 return or_(HostRequest.initiator_user_id == viewer_id, HostRequest.recipient_user_id == viewer_id)
539def _build_group_chats_pb(
540 session: Session, context: CouchersContext, group_chat_ids: list[int]
541) -> dict[int, conversations_pb2.GroupChat]:
542 """Build GroupChat protobufs (with unseen counts) for a page of group-chat ids."""
543 if not group_chat_ids:
544 return {}
545 # For each group chat, find the viewer's subscription and its latest message.
546 latest_per_group_chat = (
547 select(
548 GroupChatSubscription.group_chat_id.label("group_chat_id"),
549 func.max(GroupChatSubscription.id).label("subscription_id"),
550 func.max(Message.id).label("message_id"),
551 )
552 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
553 .where(GroupChatSubscription.user_id == context.user_id)
554 .where(GroupChatSubscription.group_chat_id.in_(group_chat_ids))
555 .where(Message.time >= GroupChatSubscription.joined)
556 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
557 .group_by(GroupChatSubscription.group_chat_id)
558 .subquery()
559 )
560 group_chat_rows = session.execute(
561 select(GroupChat, GroupChatSubscription, Message)
562 .select_from(latest_per_group_chat)
563 .join(Message, Message.id == latest_per_group_chat.c.message_id)
564 .join(GroupChatSubscription, GroupChatSubscription.id == latest_per_group_chat.c.subscription_id)
565 .join(GroupChat, GroupChat.conversation_id == latest_per_group_chat.c.group_chat_id)
566 .join(Conversation, Conversation.id == GroupChat.conversation_id)
567 .options(contains_eager(GroupChat.conversation))
568 ).all()
570 subscription_ids = [row.GroupChatSubscription.id for row in group_chat_rows]
571 unseen_count_by_subscription: dict[int, int] = dict(
572 session.execute( # type: ignore[arg-type]
573 select(GroupChatSubscription.id, func.count(Message.id))
574 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
575 .where(GroupChatSubscription.id.in_(subscription_ids))
576 .where(Message.id > GroupChatSubscription.last_seen_message_id)
577 .group_by(GroupChatSubscription.id)
578 ).all()
579 )
580 return {
581 row.GroupChat.conversation_id: _group_chat_to_pb(
582 session,
583 context,
584 row.GroupChat,
585 row.GroupChatSubscription,
586 row.Message,
587 unseen_count_by_subscription.get(row.GroupChatSubscription.id, 0),
588 )
589 for row in group_chat_rows
590 }
593def _build_host_request_threads_pb(
594 session: Session,
595 context: CouchersContext,
596 host_request_ids: list[int],
597 latest_message_id_by_conversation: dict[int, int],
598) -> dict[int, conversations_pb2.HostRequestThread]:
599 """Build HostRequestThread protobufs (with role-based roles + unseen counts) for a page."""
600 if not host_request_ids:
601 return {}
602 host_request_rows = session.execute(
603 select(HostRequest, Conversation)
604 .join(Conversation, Conversation.id == HostRequest.conversation_id)
605 .where(HostRequest.conversation_id.in_(host_request_ids))
606 ).all()
608 latest_message_ids = [latest_message_id_by_conversation[conversation_id] for conversation_id in host_request_ids]
609 message_by_id = {
610 message.id: message
611 for message in session.execute(select(Message).where(Message.id.in_(latest_message_ids))).scalars().all()
612 }
614 viewer_last_seen_message_id = case(
615 (HostRequest.initiator_user_id == context.user_id, HostRequest.initiator_last_seen_message_id),
616 else_=HostRequest.recipient_last_seen_message_id,
617 )
618 unseen_count_by_conversation: dict[int, int] = dict(
619 session.execute( # type: ignore[arg-type]
620 select(HostRequest.conversation_id, func.count(Message.id))
621 .join(Message, Message.conversation_id == HostRequest.conversation_id)
622 .where(HostRequest.conversation_id.in_(host_request_ids))
623 .where(Message.id > viewer_last_seen_message_id)
624 .group_by(HostRequest.conversation_id)
625 ).all()
626 )
627 return {
628 row.HostRequest.conversation_id: _host_request_thread_to_pb(
629 row.HostRequest,
630 row.Conversation,
631 message_by_id.get(latest_message_id_by_conversation[row.HostRequest.conversation_id]),
632 context.user_id,
633 unseen_count_by_conversation.get(row.HostRequest.conversation_id, 0),
634 )
635 for row in host_request_rows
636 }
639class Conversations(conversations_pb2_grpc.ConversationsServicer):
640 # TODO(#7722): remove after FE migrates to ListMessageThreads
641 def ListGroupChats(
642 self, request: conversations_pb2.ListGroupChatsReq, context: CouchersContext, session: Session
643 ) -> conversations_pb2.ListGroupChatsRes:
644 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH
645 page_size = min(page_size, MAX_PAGE_SIZE)
647 # select group chats where you have a subscription, and for each of
648 # these, the latest message from them
650 t = (
651 select(
652 GroupChatSubscription.group_chat_id.label("group_chat_id"),
653 func.max(GroupChatSubscription.id).label("group_chat_subscriptions_id"),
654 func.max(Message.id).label("message_id"),
655 )
656 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
657 .where(GroupChatSubscription.user_id == context.user_id)
658 .where(Message.time >= GroupChatSubscription.joined)
659 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
660 .where(
661 or_(
662 to_bool(request.HasField("only_archived") == False),
663 GroupChatSubscription.is_archived == request.only_archived,
664 )
665 )
666 .group_by(GroupChatSubscription.group_chat_id)
667 .order_by(func.max(Message.id).desc())
668 .subquery()
669 )
671 results = session.execute(
672 where_moderated_content_visible(
673 select(t, GroupChat, GroupChatSubscription, Message)
674 .join(Message, Message.id == t.c.message_id)
675 .join(GroupChatSubscription, GroupChatSubscription.id == t.c.group_chat_subscriptions_id)
676 .join(GroupChat, GroupChat.conversation_id == t.c.group_chat_id)
677 .join(Conversation, Conversation.id == GroupChat.conversation_id)
678 .options(contains_eager(GroupChat.conversation))
679 .where(or_(t.c.message_id < request.last_message_id, to_bool(request.last_message_id == 0)))
680 .order_by(t.c.message_id.desc())
681 .limit(page_size + 1),
682 context,
683 GroupChat,
684 is_list_operation=True,
685 )
686 ).all()
688 # Batch: unseen message counts in one query instead of N individual queries
689 subscription_ids = [r.GroupChatSubscription.id for r in results[:page_size]]
690 unseen_counts: dict[int, int] = dict(
691 session.execute( # type: ignore[arg-type]
692 select(GroupChatSubscription.id, func.count(Message.id))
693 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
694 .where(GroupChatSubscription.id.in_(subscription_ids))
695 .where(Message.id > GroupChatSubscription.last_seen_message_id)
696 .group_by(GroupChatSubscription.id)
697 ).all()
698 )
700 return conversations_pb2.ListGroupChatsRes(
701 group_chats=[
702 _group_chat_to_pb(
703 session,
704 context,
705 result.GroupChat,
706 result.GroupChatSubscription,
707 result.Message,
708 unseen_counts.get(result.GroupChatSubscription.id, 0),
709 )
710 for result in results[:page_size]
711 ],
712 last_message_id=(
713 min(g.Message.id if g.Message else 1 for g in results[:page_size]) if len(results) > 0 else 0
714 ), # TODO
715 no_more=len(results) <= page_size,
716 )
718 def ListMessageThreads(
719 self, request: conversations_pb2.ListMessageThreadsReq, context: CouchersContext, session: Session
720 ) -> conversations_pb2.ListMessageThreadsRes:
721 thread_filter = request.filter
723 if thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_UNSPECIFIED:
724 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_thread_filter")
726 if thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_MY_PUBLIC_TRIPS and not context.get_boolean_value(
727 "public_trips_enabled", False
728 ):
729 return conversations_pb2.ListMessageThreadsRes()
731 page_size = request.page_size if request.page_size != 0 else DEFAULT_PAGINATION_LENGTH
732 page_size = min(page_size, MAX_PAGE_SIZE)
733 only_archived = request.only_archived if request.HasField("only_archived") else None
734 unread = thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_UNREAD
736 include_chats = thread_filter in (
737 conversations_pb2.MESSAGE_THREAD_FILTER_ALL,
738 conversations_pb2.MESSAGE_THREAD_FILTER_UNREAD,
739 conversations_pb2.MESSAGE_THREAD_FILTER_CHATS,
740 )
741 include_host_requests = thread_filter in (
742 conversations_pb2.MESSAGE_THREAD_FILTER_ALL,
743 conversations_pb2.MESSAGE_THREAD_FILTER_UNREAD,
744 conversations_pb2.MESSAGE_THREAD_FILTER_HOSTING,
745 conversations_pb2.MESSAGE_THREAD_FILTER_SURFING,
746 conversations_pb2.MESSAGE_THREAD_FILTER_MY_PUBLIC_TRIPS,
747 )
749 # Build one candidate subquery per included thread kind, each yielding
750 # (conversation_id, latest_message_id, kind), then union them so a single
751 # global cursor (Message.id) can paginate across all kinds.
752 candidate_queries = []
753 if include_chats:
754 candidate_queries.append(_group_chat_candidate_query(context, only_archived, unread))
755 if include_host_requests:
756 candidate_queries.append(
757 _host_request_candidate_query(
758 context, _host_request_party_predicate(context, thread_filter), only_archived, unread
759 )
760 )
762 combined_candidates = (
763 candidate_queries[0] if len(candidate_queries) == 1 else union_all(*candidate_queries)
764 ).subquery()
765 page_query = select(
766 combined_candidates.c.conversation_id,
767 combined_candidates.c.latest_message_id,
768 combined_candidates.c.kind,
769 )
770 if request.page_token:
771 page_query = page_query.where(
772 combined_candidates.c.latest_message_id < int(decrypt_page_token(request.page_token))
773 )
774 page_query = page_query.order_by(combined_candidates.c.latest_message_id.desc()).limit(page_size + 1)
775 candidate_rows = session.execute(page_query).all()
777 page_rows = candidate_rows[:page_size]
778 has_more = len(candidate_rows) > page_size
779 next_page_token = encrypt_page_token(str(min(row.latest_message_id for row in page_rows))) if has_more else ""
781 latest_message_id_by_conversation = {row.conversation_id: row.latest_message_id for row in page_rows}
782 group_chat_ids = [row.conversation_id for row in page_rows if row.kind == "group_chat"]
783 host_request_ids = [row.conversation_id for row in page_rows if row.kind == "host_request"]
785 # Hydrate each kind in a batch, then re-assemble in the paginated order.
786 group_chats_by_id = _build_group_chats_pb(session, context, group_chat_ids)
787 host_request_threads_by_id = _build_host_request_threads_pb(
788 session, context, host_request_ids, latest_message_id_by_conversation
789 )
791 threads = []
792 for row in page_rows:
793 if row.kind == "group_chat":
794 group_chat = group_chats_by_id.get(row.conversation_id)
795 if group_chat is not None: 795 ↛ 792line 795 didn't jump to line 792 because the condition on line 795 was always true
796 threads.append(conversations_pb2.MessageThread(group_chat=group_chat))
797 else:
798 host_request_thread = host_request_threads_by_id.get(row.conversation_id)
799 if host_request_thread is not None: 799 ↛ 792line 799 didn't jump to line 792 because the condition on line 799 was always true
800 threads.append(conversations_pb2.MessageThread(host_request=host_request_thread))
802 return conversations_pb2.ListMessageThreadsRes(threads=threads, next_page_token=next_page_token)
804 def MarkAllThreadsSeen(
805 self, request: conversations_pb2.MarkAllThreadsSeenReq, context: CouchersContext, session: Session
806 ) -> empty_pb2.Empty:
807 thread_filter = request.filter
809 if thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_UNSPECIFIED:
810 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_thread_filter")
812 if thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_MY_PUBLIC_TRIPS and not context.get_boolean_value( 812 ↛ 815line 812 didn't jump to line 815 because the condition on line 812 was never true
813 "public_trips_enabled", False
814 ):
815 return empty_pb2.Empty()
817 only_archived = request.only_archived if request.HasField("only_archived") else None
818 unread = thread_filter == conversations_pb2.MESSAGE_THREAD_FILTER_UNREAD
820 include_chats = thread_filter in (
821 conversations_pb2.MESSAGE_THREAD_FILTER_ALL,
822 conversations_pb2.MESSAGE_THREAD_FILTER_UNREAD,
823 conversations_pb2.MESSAGE_THREAD_FILTER_CHATS,
824 )
825 include_host_requests = thread_filter in (
826 conversations_pb2.MESSAGE_THREAD_FILTER_ALL,
827 conversations_pb2.MESSAGE_THREAD_FILTER_UNREAD,
828 conversations_pb2.MESSAGE_THREAD_FILTER_HOSTING,
829 conversations_pb2.MESSAGE_THREAD_FILTER_SURFING,
830 conversations_pb2.MESSAGE_THREAD_FILTER_MY_PUBLIC_TRIPS,
831 )
833 if include_chats: 833 ↛ 863line 833 didn't jump to line 863 because the condition on line 833 was always true
834 latest_message_id_by_conversation = {
835 row.conversation_id: row.latest_message_id
836 for row in session.execute(_group_chat_candidate_query(context, only_archived, unread)).all()
837 }
838 if latest_message_id_by_conversation:
839 subscriptions = (
840 session.execute(
841 select(GroupChatSubscription)
842 .where(GroupChatSubscription.user_id == context.user_id)
843 .where(GroupChatSubscription.group_chat_id.in_(latest_message_id_by_conversation.keys()))
844 .where(GroupChatSubscription.left == None)
845 )
846 .scalars()
847 .all()
848 )
849 for subscription in subscriptions:
850 latest_message_id = latest_message_id_by_conversation[subscription.group_chat_id]
851 if subscription.last_seen_message_id < latest_message_id:
852 subscription.last_seen_message_id = latest_message_id
853 mark_notifications_seen(
854 session,
855 user_id=context.user_id,
856 key=str(subscription.group_chat_id),
857 topic_actions=[
858 NotificationTopicAction.chat__message,
859 NotificationTopicAction.chat__missed_messages,
860 ],
861 )
863 if include_host_requests:
864 party_predicate = _host_request_party_predicate(context, thread_filter)
865 latest_message_id_by_conversation = {
866 row.conversation_id: row.latest_message_id
867 for row in session.execute(
868 _host_request_candidate_query(context, party_predicate, only_archived, unread)
869 ).all()
870 }
871 if latest_message_id_by_conversation: 871 ↛ 895line 871 didn't jump to line 895 because the condition on line 871 was always true
872 host_requests = (
873 session.execute(
874 select(HostRequest).where(
875 HostRequest.conversation_id.in_(latest_message_id_by_conversation.keys())
876 )
877 )
878 .scalars()
879 .all()
880 )
881 for host_request in host_requests:
882 latest_message_id = latest_message_id_by_conversation[host_request.conversation_id]
883 if host_request.initiator_user_id == context.user_id: 883 ↛ 884line 883 didn't jump to line 884 because the condition on line 883 was never true
884 if host_request.initiator_last_seen_message_id < latest_message_id:
885 host_request.initiator_last_seen_message_id = latest_message_id
886 elif host_request.recipient_last_seen_message_id < latest_message_id: 886 ↛ 888line 886 didn't jump to line 888 because the condition on line 886 was always true
887 host_request.recipient_last_seen_message_id = latest_message_id
888 mark_notifications_seen(
889 session,
890 user_id=context.user_id,
891 key=str(host_request.conversation_id),
892 topic_actions=_HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS,
893 )
895 return empty_pb2.Empty()
897 def GetGroupChat(
898 self, request: conversations_pb2.GetGroupChatReq, context: CouchersContext, session: Session
899 ) -> conversations_pb2.GroupChat:
900 result = session.execute(
901 where_moderated_content_visible(
902 select(GroupChat, GroupChatSubscription, Message)
903 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
904 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
905 .join(Conversation, Conversation.id == GroupChat.conversation_id)
906 .options(contains_eager(GroupChat.conversation))
907 .where(GroupChatSubscription.user_id == context.user_id)
908 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
909 .where(Message.time >= GroupChatSubscription.joined)
910 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
911 .order_by(Message.id.desc())
912 .limit(1),
913 context,
914 GroupChat,
915 is_list_operation=False,
916 )
917 ).one_or_none()
919 if not result:
920 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
922 return conversations_pb2.GroupChat(
923 group_chat_id=result.GroupChat.conversation_id,
924 title=result.GroupChat.title,
925 member_user_ids=_get_visible_members_for_subscription(result.GroupChatSubscription),
926 admin_user_ids=_get_visible_admins_for_subscription(result.GroupChatSubscription),
927 only_admins_invite=result.GroupChat.only_admins_invite,
928 is_dm=result.GroupChat.is_dm,
929 created=Timestamp_from_datetime(result.GroupChat.conversation.created),
930 unseen_message_count=_unseen_message_count(session, result.GroupChatSubscription.id),
931 last_seen_message_id=result.GroupChatSubscription.last_seen_message_id,
932 latest_message=_message_to_pb(result.Message) if result.Message else None,
933 mute_info=_mute_info(result.GroupChatSubscription),
934 can_message=_user_can_message(session, context, result.GroupChat),
935 is_archived=result.GroupChatSubscription.is_archived,
936 )
938 def GetDirectMessage(
939 self, request: conversations_pb2.GetDirectMessageReq, context: CouchersContext, session: Session
940 ) -> conversations_pb2.GroupChat:
941 count = func.count(GroupChatSubscription.id).label("count")
942 subquery = (
943 select(GroupChatSubscription.group_chat_id)
944 .where(
945 or_(
946 GroupChatSubscription.user_id == context.user_id,
947 GroupChatSubscription.user_id == request.user_id,
948 )
949 )
950 .where(GroupChatSubscription.left == None)
951 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
952 .where(GroupChat.is_dm == True)
953 .group_by(GroupChatSubscription.group_chat_id)
954 .having(count == 2)
955 .subquery()
956 )
958 result = session.execute(
959 where_moderated_content_visible(
960 select(subquery, GroupChat, GroupChatSubscription, Message)
961 .join(subquery, subquery.c.group_chat_id == GroupChat.conversation_id)
962 .join(Message, Message.conversation_id == GroupChat.conversation_id)
963 .join(Conversation, Conversation.id == GroupChat.conversation_id)
964 .options(contains_eager(GroupChat.conversation))
965 .where(GroupChatSubscription.user_id == context.user_id)
966 .where(GroupChatSubscription.group_chat_id == GroupChat.conversation_id)
967 .where(Message.time >= GroupChatSubscription.joined)
968 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
969 .order_by(Message.id.desc())
970 .limit(1),
971 context,
972 GroupChat,
973 is_list_operation=False,
974 )
975 ).one_or_none()
977 if not result:
978 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
980 return conversations_pb2.GroupChat(
981 group_chat_id=result.GroupChat.conversation_id,
982 title=result.GroupChat.title,
983 member_user_ids=_get_visible_members_for_subscription(result.GroupChatSubscription),
984 admin_user_ids=_get_visible_admins_for_subscription(result.GroupChatSubscription),
985 only_admins_invite=result.GroupChat.only_admins_invite,
986 is_dm=result.GroupChat.is_dm,
987 created=Timestamp_from_datetime(result.GroupChat.conversation.created),
988 unseen_message_count=_unseen_message_count(session, result.GroupChatSubscription.id),
989 last_seen_message_id=result.GroupChatSubscription.last_seen_message_id,
990 latest_message=_message_to_pb(result.Message) if result.Message else None,
991 mute_info=_mute_info(result.GroupChatSubscription),
992 can_message=_user_can_message(session, context, result.GroupChat),
993 is_archived=result.GroupChatSubscription.is_archived,
994 )
996 def GetUpdates(
997 self, request: conversations_pb2.GetUpdatesReq, context: CouchersContext, session: Session
998 ) -> conversations_pb2.GetUpdatesRes:
999 results = (
1000 session.execute(
1001 where_moderated_content_visible(
1002 select(Message)
1003 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
1004 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
1005 .where(GroupChatSubscription.user_id == context.user_id)
1006 .where(Message.time >= GroupChatSubscription.joined)
1007 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
1008 .where(Message.id > request.newest_message_id)
1009 .order_by(Message.id.asc())
1010 .limit(DEFAULT_PAGINATION_LENGTH + 1),
1011 context,
1012 GroupChat,
1013 is_list_operation=False,
1014 )
1015 )
1016 .scalars()
1017 .all()
1018 )
1020 return conversations_pb2.GetUpdatesRes(
1021 updates=[
1022 conversations_pb2.Update(
1023 group_chat_id=message.conversation_id,
1024 message=_message_to_pb(message),
1025 )
1026 for message in sorted(results, key=lambda message: message.id)[:DEFAULT_PAGINATION_LENGTH]
1027 ],
1028 no_more=len(results) <= DEFAULT_PAGINATION_LENGTH,
1029 )
1031 def GetGroupChatMessages(
1032 self, request: conversations_pb2.GetGroupChatMessagesReq, context: CouchersContext, session: Session
1033 ) -> conversations_pb2.GetGroupChatMessagesRes:
1034 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH
1035 page_size = min(page_size, MAX_PAGE_SIZE)
1037 results = (
1038 session.execute(
1039 where_moderated_content_visible(
1040 select(Message)
1041 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
1042 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
1043 .where(GroupChatSubscription.user_id == context.user_id)
1044 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1045 .where(Message.time >= GroupChatSubscription.joined)
1046 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
1047 .where(or_(Message.id < request.last_message_id, to_bool(request.last_message_id == 0)))
1048 .where(
1049 or_(Message.id > GroupChatSubscription.last_seen_message_id, to_bool(request.only_unseen == 0))
1050 )
1051 .order_by(Message.id.desc())
1052 .limit(page_size + 1),
1053 context,
1054 GroupChat,
1055 is_list_operation=False,
1056 )
1057 )
1058 .scalars()
1059 .all()
1060 )
1062 return conversations_pb2.GetGroupChatMessagesRes(
1063 messages=[_message_to_pb(message) for message in results[:page_size]],
1064 last_message_id=results[-2].id if len(results) > 1 else 0, # TODO
1065 no_more=len(results) <= page_size,
1066 )
1068 def MarkLastSeenGroupChat(
1069 self, request: conversations_pb2.MarkLastSeenGroupChatReq, context: CouchersContext, session: Session
1070 ) -> empty_pb2.Empty:
1071 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1073 if not subscription: 1073 ↛ 1074line 1073 didn't jump to line 1074 because the condition on line 1073 was never true
1074 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1076 if not subscription.last_seen_message_id <= request.last_seen_message_id: 1076 ↛ 1077line 1076 didn't jump to line 1077 because the condition on line 1076 was never true
1077 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages")
1079 subscription.last_seen_message_id = request.last_seen_message_id
1081 mark_notifications_seen(
1082 session,
1083 user_id=context.user_id,
1084 key=str(request.group_chat_id),
1085 topic_actions=[
1086 NotificationTopicAction.chat__message,
1087 NotificationTopicAction.chat__missed_messages,
1088 ],
1089 )
1091 return empty_pb2.Empty()
1093 def MuteGroupChat(
1094 self, request: conversations_pb2.MuteGroupChatReq, context: CouchersContext, session: Session
1095 ) -> empty_pb2.Empty:
1096 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1098 if not subscription: 1098 ↛ 1099line 1098 didn't jump to line 1099 because the condition on line 1098 was never true
1099 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1101 if request.unmute:
1102 subscription.muted_until = DATETIME_MINUS_INFINITY
1103 elif request.forever:
1104 subscription.muted_until = DATETIME_INFINITY
1105 elif request.for_duration: 1105 ↛ 1111line 1105 didn't jump to line 1111 because the condition on line 1105 was always true
1106 duration = request.for_duration.ToTimedelta()
1107 if duration < timedelta(seconds=0): 1107 ↛ 1108line 1107 didn't jump to line 1108 because the condition on line 1107 was never true
1108 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_mute_past")
1109 subscription.muted_until = now() + duration
1111 return empty_pb2.Empty()
1113 def SetGroupChatArchiveStatus(
1114 self, request: conversations_pb2.SetGroupChatArchiveStatusReq, context: CouchersContext, session: Session
1115 ) -> conversations_pb2.SetGroupChatArchiveStatusRes:
1116 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1118 if not subscription:
1119 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1121 subscription.is_archived = request.is_archived
1123 return conversations_pb2.SetGroupChatArchiveStatusRes(
1124 group_chat_id=request.group_chat_id,
1125 is_archived=request.is_archived,
1126 )
1128 def SearchMessages(
1129 self, request: conversations_pb2.SearchMessagesReq, context: CouchersContext, session: Session
1130 ) -> conversations_pb2.SearchMessagesRes:
1131 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH
1132 page_size = min(page_size, MAX_PAGE_SIZE)
1134 results = (
1135 session.execute(
1136 where_moderated_content_visible(
1137 select(Message)
1138 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id)
1139 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id)
1140 .where(GroupChatSubscription.user_id == context.user_id)
1141 .where(Message.time >= GroupChatSubscription.joined)
1142 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
1143 .where(or_(Message.id < request.last_message_id, to_bool(request.last_message_id == 0)))
1144 .where(Message.text.ilike(f"%{request.query}%"))
1145 .order_by(Message.id.desc())
1146 .limit(page_size + 1),
1147 context,
1148 GroupChat,
1149 is_list_operation=True,
1150 )
1151 )
1152 .scalars()
1153 .all()
1154 )
1156 return conversations_pb2.SearchMessagesRes(
1157 results=[
1158 conversations_pb2.MessageSearchResult(
1159 group_chat_id=message.conversation_id,
1160 message=_message_to_pb(message),
1161 )
1162 for message in results[:page_size]
1163 ],
1164 last_message_id=results[-2].id if len(results) > 1 else 0,
1165 no_more=len(results) <= page_size,
1166 )
1168 def CreateGroupChat(
1169 self, request: conversations_pb2.CreateGroupChatReq, context: CouchersContext, session: Session
1170 ) -> conversations_pb2.GroupChat:
1171 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
1172 if not has_completed_profile(session, user):
1173 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_message")
1175 recipient_user_ids = list(
1176 session.execute(
1177 select(User.id).where(users_visible(context)).where(User.id.in_(request.recipient_user_ids))
1178 )
1179 .scalars()
1180 .all()
1181 )
1183 # make sure all requested users are visible
1184 if len(recipient_user_ids) != len(request.recipient_user_ids): 1184 ↛ 1185line 1184 didn't jump to line 1185 because the condition on line 1184 was never true
1185 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_not_found")
1187 if not recipient_user_ids: 1187 ↛ 1188line 1187 didn't jump to line 1188 because the condition on line 1187 was never true
1188 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "no_recipients")
1190 if len(recipient_user_ids) != len(set(recipient_user_ids)): 1190 ↛ 1192line 1190 didn't jump to line 1192 because the condition on line 1190 was never true
1191 # make sure there's no duplicate users
1192 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_recipients")
1194 if context.user_id in recipient_user_ids: 1194 ↛ 1195line 1194 didn't jump to line 1195 because the condition on line 1194 was never true
1195 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_add_self")
1197 if len(recipient_user_ids) == 1:
1198 # can only have one DM at a time between any two users
1199 other_user_id = recipient_user_ids[0]
1201 # the following sql statement selects subscriptions that are DMs and have the same group_chat_id, and have
1202 # user_id either this user or the recipient user. If you find two subscriptions to the same DM group
1203 # chat, you know they already have a shared group chat
1204 count = func.count(GroupChatSubscription.id).label("count")
1205 if session.execute(
1206 where_moderated_content_visible(
1207 select(count)
1208 .where(
1209 or_(
1210 GroupChatSubscription.user_id == context.user_id,
1211 GroupChatSubscription.user_id == other_user_id,
1212 )
1213 )
1214 .where(GroupChatSubscription.left == None)
1215 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
1216 .where(GroupChat.is_dm == True)
1217 .group_by(GroupChatSubscription.group_chat_id)
1218 .having(count == 2),
1219 context,
1220 GroupChat,
1221 is_list_operation=False,
1222 )
1223 ).scalar_one_or_none():
1224 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_have_dm")
1226 # Check if user has been initiating chats excessively
1227 if process_rate_limits_and_check_abort(
1228 session=session, user_id=context.user_id, action=RateLimitAction.chat_initiation
1229 ):
1230 context.abort_with_error_code(
1231 grpc.StatusCode.RESOURCE_EXHAUSTED,
1232 "chat_initiation_rate_limit2",
1233 substitutions={"count": RATE_LIMIT_HOURS},
1234 )
1236 group_chat = _create_chat(
1237 session,
1238 creator_id=context.user_id,
1239 recipient_ids=request.recipient_user_ids,
1240 title=request.title.value,
1241 )
1243 your_subscription = _get_message_subscription(session, context.user_id, group_chat.conversation_id)
1245 _add_message_to_subscription(session, your_subscription, message_type=MessageType.chat_created)
1247 session.flush()
1249 log_event(
1250 context,
1251 session,
1252 "group_chat.created",
1253 {
1254 "group_chat_id": group_chat.conversation_id,
1255 "is_dm": group_chat.is_dm,
1256 "recipient_count": len(request.recipient_user_ids),
1257 },
1258 )
1260 return conversations_pb2.GroupChat(
1261 group_chat_id=group_chat.conversation_id,
1262 title=group_chat.title,
1263 member_user_ids=_get_visible_members_for_subscription(your_subscription),
1264 admin_user_ids=_get_visible_admins_for_subscription(your_subscription),
1265 only_admins_invite=group_chat.only_admins_invite,
1266 is_dm=group_chat.is_dm,
1267 created=Timestamp_from_datetime(group_chat.conversation.created),
1268 mute_info=_mute_info(your_subscription),
1269 can_message=True,
1270 )
1272 def SendMessage(
1273 self, request: conversations_pb2.SendMessageReq, context: CouchersContext, session: Session
1274 ) -> empty_pb2.Empty:
1275 if request.text == "": 1275 ↛ 1276line 1275 didn't jump to line 1276 because the condition on line 1275 was never true
1276 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
1278 result = session.execute(
1279 where_moderated_content_visible(
1280 select(GroupChatSubscription, GroupChat)
1281 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
1282 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1283 .where(GroupChatSubscription.user_id == context.user_id)
1284 .where(GroupChatSubscription.left == None),
1285 context,
1286 GroupChat,
1287 is_list_operation=False,
1288 )
1289 ).one_or_none()
1290 if not result:
1291 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1293 subscription, group_chat = result._tuple()
1294 if not _user_can_message(session, context, group_chat):
1295 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_message_in_chat")
1297 _add_message_to_subscription(session, subscription, message_type=MessageType.text, text=request.text)
1299 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
1300 sent_messages_counter.labels(
1301 user_gender, "direct message" if subscription.group_chat.is_dm else "group chat"
1302 ).inc()
1303 log_event(
1304 context,
1305 session,
1306 "message.sent",
1307 {"group_chat_id": request.group_chat_id, "is_dm": subscription.group_chat.is_dm},
1308 )
1310 return empty_pb2.Empty()
1312 def SendDirectMessage(
1313 self, request: conversations_pb2.SendDirectMessageReq, context: CouchersContext, session: Session
1314 ) -> conversations_pb2.SendDirectMessageRes:
1315 user_id = context.user_id
1316 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
1318 recipient_id = request.recipient_user_id
1320 if not has_completed_profile(session, user): 1320 ↛ 1321line 1320 didn't jump to line 1321 because the condition on line 1320 was never true
1321 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_message")
1323 if not recipient_id: 1323 ↛ 1324line 1323 didn't jump to line 1324 because the condition on line 1323 was never true
1324 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "no_recipients")
1326 recipient_user_id = session.execute(
1327 select(User.id).where(users_visible(context)).where(User.id == recipient_id)
1328 ).scalar_one_or_none()
1330 if not recipient_user_id: 1330 ↛ 1331line 1330 didn't jump to line 1331 because the condition on line 1330 was never true
1331 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_not_found")
1333 if user_id == recipient_id: 1333 ↛ 1334line 1333 didn't jump to line 1334 because the condition on line 1333 was never true
1334 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_add_self")
1336 if request.text == "": 1336 ↛ 1337line 1336 didn't jump to line 1337 because the condition on line 1336 was never true
1337 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
1339 # Look for an existing direct message (DM) chat between the two users
1340 dm_chat_ids = (
1341 select(GroupChatSubscription.group_chat_id)
1342 .where(GroupChatSubscription.user_id.in_([user_id, recipient_id]))
1343 .group_by(GroupChatSubscription.group_chat_id)
1344 .having(func.count(GroupChatSubscription.user_id) == 2)
1345 )
1347 chat = session.execute(
1348 where_moderated_content_visible(
1349 select(GroupChat)
1350 .where(GroupChat.is_dm == True)
1351 .where(GroupChat.conversation_id.in_(dm_chat_ids))
1352 .limit(1),
1353 context,
1354 GroupChat,
1355 is_list_operation=False,
1356 )
1357 ).scalar_one_or_none()
1359 if not chat:
1360 if process_rate_limits_and_check_abort(
1361 session=session, user_id=user_id, action=RateLimitAction.chat_initiation
1362 ):
1363 context.abort_with_error_code(
1364 grpc.StatusCode.RESOURCE_EXHAUSTED,
1365 "chat_initiation_rate_limit2",
1366 substitutions={"count": RATE_LIMIT_HOURS},
1367 )
1368 chat = _create_chat(session, user_id, [recipient_id])
1370 # Retrieve the sender's active subscription to the chat
1371 subscription = _get_message_subscription(session, user_id, chat.conversation_id)
1373 # Add the message to the conversation
1374 _add_message_to_subscription(session, subscription, message_type=MessageType.text, text=request.text)
1376 user_gender = session.execute(select(User.gender).where(User.id == user_id)).scalar_one()
1377 sent_messages_counter.labels(user_gender, "direct message").inc()
1378 log_event(
1379 context,
1380 session,
1381 "message.sent",
1382 {"group_chat_id": chat.conversation_id, "is_dm": True, "recipient_id": recipient_id},
1383 )
1385 session.flush()
1387 return conversations_pb2.SendDirectMessageRes(group_chat_id=chat.conversation_id)
1389 def EditGroupChat(
1390 self, request: conversations_pb2.EditGroupChatReq, context: CouchersContext, session: Session
1391 ) -> empty_pb2.Empty:
1392 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1394 if not subscription:
1395 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1397 if subscription.role != GroupChatRole.admin:
1398 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_edit")
1400 if request.HasField("title"):
1401 subscription.group_chat.title = request.title.value
1403 if request.HasField("only_admins_invite"): 1403 ↛ 1406line 1403 didn't jump to line 1406 because the condition on line 1403 was always true
1404 subscription.group_chat.only_admins_invite = request.only_admins_invite.value
1406 _add_message_to_subscription(session, subscription, message_type=MessageType.chat_edited)
1408 return empty_pb2.Empty()
1410 def MakeGroupChatAdmin(
1411 self, request: conversations_pb2.MakeGroupChatAdminReq, context: CouchersContext, session: Session
1412 ) -> empty_pb2.Empty:
1413 if not session.execute(
1414 select(User).where(users_visible(context)).where(User.id == request.user_id)
1415 ).scalar_one_or_none():
1416 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1418 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1420 if not your_subscription: 1420 ↛ 1421line 1420 didn't jump to line 1421 because the condition on line 1420 was never true
1421 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1423 if your_subscription.role != GroupChatRole.admin:
1424 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_make_admin")
1426 if request.user_id == context.user_id: 1426 ↛ 1427line 1426 didn't jump to line 1427 because the condition on line 1426 was never true
1427 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_make_self_admin")
1429 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
1431 if not their_subscription: 1431 ↛ 1432line 1431 didn't jump to line 1432 because the condition on line 1431 was never true
1432 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_in_chat")
1434 if their_subscription.role != GroupChatRole.participant:
1435 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_admin")
1437 their_subscription.role = GroupChatRole.admin
1439 _add_message_to_subscription(
1440 session, your_subscription, message_type=MessageType.user_made_admin, target_id=request.user_id
1441 )
1443 return empty_pb2.Empty()
1445 def RemoveGroupChatAdmin(
1446 self, request: conversations_pb2.RemoveGroupChatAdminReq, context: CouchersContext, session: Session
1447 ) -> empty_pb2.Empty:
1448 if not session.execute(
1449 select(User).where(users_visible(context)).where(User.id == request.user_id)
1450 ).scalar_one_or_none():
1451 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1453 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1455 if not your_subscription: 1455 ↛ 1456line 1455 didn't jump to line 1456 because the condition on line 1455 was never true
1456 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1458 if request.user_id == context.user_id:
1459 # Race condition!
1460 other_admins_count = session.execute(
1461 select(func.count())
1462 .select_from(GroupChatSubscription)
1463 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1464 .where(GroupChatSubscription.user_id != context.user_id)
1465 .where(GroupChatSubscription.role == GroupChatRole.admin)
1466 .where(GroupChatSubscription.left == None)
1467 ).scalar_one()
1468 if not other_admins_count > 0:
1469 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_remove_last_admin")
1471 if your_subscription.role != GroupChatRole.admin:
1472 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_remove_admin")
1474 their_subscription = session.execute(
1475 select(GroupChatSubscription)
1476 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1477 .where(GroupChatSubscription.user_id == request.user_id)
1478 .where(GroupChatSubscription.left == None)
1479 .where(GroupChatSubscription.role == GroupChatRole.admin)
1480 ).scalar_one_or_none()
1482 if not their_subscription: 1482 ↛ 1483line 1482 didn't jump to line 1483 because the condition on line 1482 was never true
1483 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_admin")
1485 their_subscription.role = GroupChatRole.participant
1487 _add_message_to_subscription(
1488 session, your_subscription, message_type=MessageType.user_removed_admin, target_id=request.user_id
1489 )
1491 return empty_pb2.Empty()
1493 def InviteToGroupChat(
1494 self, request: conversations_pb2.InviteToGroupChatReq, context: CouchersContext, session: Session
1495 ) -> empty_pb2.Empty:
1496 if not session.execute(
1497 select(User).where(users_visible(context)).where(User.id == request.user_id)
1498 ).scalar_one_or_none():
1499 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1501 result = session.execute(
1502 where_moderated_content_visible(
1503 select(GroupChatSubscription, GroupChat)
1504 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
1505 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1506 .where(GroupChatSubscription.user_id == context.user_id)
1507 .where(GroupChatSubscription.left == None),
1508 context,
1509 GroupChat,
1510 is_list_operation=False,
1511 )
1512 ).one_or_none()
1514 if not result:
1515 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1517 your_subscription, group_chat = result._tuple()
1519 if request.user_id == context.user_id: 1519 ↛ 1520line 1519 didn't jump to line 1520 because the condition on line 1519 was never true
1520 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_invite_self")
1522 if your_subscription.role != GroupChatRole.admin and your_subscription.group_chat.only_admins_invite:
1523 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invite_permission_denied")
1525 if group_chat.is_dm:
1526 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_invite_to_dm")
1528 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
1530 if their_subscription: 1530 ↛ 1531line 1530 didn't jump to line 1531 because the condition on line 1530 was never true
1531 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_in_chat")
1533 # TODO: race condition!
1535 subscription = GroupChatSubscription(
1536 user_id=request.user_id,
1537 group_chat_id=your_subscription.group_chat.conversation_id,
1538 role=GroupChatRole.participant,
1539 )
1540 session.add(subscription)
1542 _add_message_to_subscription(
1543 session, your_subscription, message_type=MessageType.user_invited, target_id=request.user_id
1544 )
1546 return empty_pb2.Empty()
1548 def RemoveGroupChatUser(
1549 self, request: conversations_pb2.RemoveGroupChatUserReq, context: CouchersContext, session: Session
1550 ) -> empty_pb2.Empty:
1551 """
1552 1. Get admin info and check it's correct
1553 2. Get user data, check it's correct and remove user
1554 """
1555 # Admin info
1556 your_subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1558 # if user info is missing
1559 if not your_subscription: 1559 ↛ 1560line 1559 didn't jump to line 1560 because the condition on line 1559 was never true
1560 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1562 # if user not admin
1563 if your_subscription.role != GroupChatRole.admin: 1563 ↛ 1564line 1563 didn't jump to line 1564 because the condition on line 1563 was never true
1564 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "only_admin_can_remove_user")
1566 # if user wants to remove themselves
1567 if request.user_id == context.user_id: 1567 ↛ 1568line 1567 didn't jump to line 1568 because the condition on line 1567 was never true
1568 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_remove_self")
1570 # get user info
1571 their_subscription = _get_message_subscription(session, request.user_id, request.group_chat_id)
1573 # user not found
1574 if not their_subscription:
1575 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "user_not_in_chat")
1577 _add_message_to_subscription(
1578 session, your_subscription, message_type=MessageType.user_removed, target_id=request.user_id
1579 )
1581 their_subscription.left = func.now()
1583 return empty_pb2.Empty()
1585 def LeaveGroupChat(
1586 self, request: conversations_pb2.LeaveGroupChatReq, context: CouchersContext, session: Session
1587 ) -> empty_pb2.Empty:
1588 subscription = _get_visible_message_subscription(session, context, request.group_chat_id)
1590 if not subscription:
1591 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found")
1593 if subscription.role == GroupChatRole.admin:
1594 other_admins_count = session.execute(
1595 select(func.count())
1596 .select_from(GroupChatSubscription)
1597 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1598 .where(GroupChatSubscription.user_id != context.user_id)
1599 .where(GroupChatSubscription.role == GroupChatRole.admin)
1600 .where(GroupChatSubscription.left == None)
1601 ).scalar_one()
1602 participants_count = session.execute(
1603 select(func.count())
1604 .select_from(GroupChatSubscription)
1605 .where(GroupChatSubscription.group_chat_id == request.group_chat_id)
1606 .where(GroupChatSubscription.user_id != context.user_id)
1607 .where(GroupChatSubscription.role == GroupChatRole.participant)
1608 .where(GroupChatSubscription.left == None)
1609 ).scalar_one()
1610 if not (other_admins_count > 0 or participants_count == 0):
1611 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "last_admin_cant_leave")
1613 _add_message_to_subscription(session, subscription, message_type=MessageType.user_left)
1615 subscription.left = func.now()
1617 return empty_pb2.Empty()