Coverage for app/backend/src/couchers/servicers/message_threads.py: 98%
130 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 15:29 +0000
1"""
2The unified message thread list (Conversations.ListMessageThreads / MarkAllThreadsSeen): one
3paginated list of all the viewer's conversations — group chats, DMs, host requests and public-trip
4offers — ordered by latest message.
6Each kind of conversation has its own select query, yielding one row per thread carrying only
7(conversation_id, latest_message_id, kind). ListMessageThreads unions them, so a single cursor on
8latest_message_id pages through all kinds as one list, then hydrates the page's worth of rows into
9protobufs, batched per kind. MarkAllThreadsSeen consumes the same queries as UPDATE targets.
10"""
12import logging
13from collections.abc import Sequence
14from dataclasses import dataclass
16import grpc
17from google.protobuf import empty_pb2
18from sqlalchemy import ColumnElement, Row, Select, exists, select, union_all, update
19from sqlalchemy.dialects.postgresql import aggregate_order_by
20from sqlalchemy.orm import Session, aliased
21from sqlalchemy.sql import and_, case, func, literal, or_
23from couchers.context import CouchersContext
24from couchers.crypto import decrypt_page_token, encrypt_page_token
25from couchers.helpers.group_chats import is_newest_subscription, is_unseen, mute_info, was_subscribed_at
26from couchers.helpers.host_requests import (
27 HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS,
28 has_unseen_host_request_messages,
29 is_hosting_party,
30 is_public_trip_offer_recipient,
31 is_surfing_party,
32 unseen_host_request_message_count,
33)
34from couchers.helpers.messages import hostrequeststatus2api, message_to_pb
35from couchers.models import (
36 Conversation,
37 GroupChat,
38 GroupChatRole,
39 GroupChatSubscription,
40 HostRequest,
41 HostRequestFeedback,
42 HostRequestStatus,
43 Message,
44)
45from couchers.models.notifications import NotificationTopicAction
46from couchers.notifications.notify import mark_notifications_seen
47from couchers.proto import conversations_pb2, requests_pb2
48from couchers.sql import to_bool, where_moderated_content_visible, where_users_column_visible
49from couchers.utils import Timestamp_from_datetime, date_to_api, get_coordinates
51logger = logging.getLogger(__name__)
53DEFAULT_PAGINATION_LENGTH = 20
54MAX_PAGE_SIZE = 50
56# discriminator on the unioned rows, telling us which table a conversation_id came from
57_KIND_GROUP_CHAT = "group_chat"
58_KIND_HOST_REQUEST = "host_request"
60# one thread as the select queries yield it: (conversation_id, latest_message_id, kind)
61_ThreadRow = Row[tuple[int, int, str]]
64@dataclass(frozen=True)
65class _ThreadFilters:
66 categories: frozenset[int]
67 only_archived: bool | None
68 only_unread: bool
71def _resolve_thread_filters(
72 context: CouchersContext,
73 request: conversations_pb2.ListMessageThreadsReq | conversations_pb2.MarkAllThreadsSeenReq,
74) -> _ThreadFilters:
75 """
76 An empty category list means all categories. MY_PUBLIC_TRIPS is dropped when the public-trips
77 flag is off; those offers still show up under SURFING.
78 """
79 if conversations_pb2.MESSAGE_THREAD_CATEGORY_UNSPECIFIED in request.categories:
80 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_thread_category")
81 categories = frozenset(
82 request.categories
83 or {
84 conversations_pb2.MESSAGE_THREAD_CATEGORY_CHATS,
85 conversations_pb2.MESSAGE_THREAD_CATEGORY_HOSTING,
86 conversations_pb2.MESSAGE_THREAD_CATEGORY_SURFING,
87 conversations_pb2.MESSAGE_THREAD_CATEGORY_MY_PUBLIC_TRIPS,
88 }
89 )
90 if not context.get_boolean_value("public_trips_enabled", False):
91 categories -= {conversations_pb2.MESSAGE_THREAD_CATEGORY_MY_PUBLIC_TRIPS}
92 return _ThreadFilters(
93 categories=categories,
94 only_archived=request.only_archived if request.HasField("only_archived") else None,
95 only_unread=request.only_unread,
96 )
99def _build_group_chat_select_query(
100 context: CouchersContext, only_archived: bool | None, unread: bool
101) -> Select[tuple[int, int, str]]:
102 """
103 The group chats (including DMs) the viewer should see, narrowed by the request's archived and
104 unread filters, along with the id of the newest message each one shows them.
106 The message join is windowed to the viewer's subscription, so a chat they were removed from ends
107 at the last message they could read rather than at the chat's newest.
108 """
109 return where_moderated_content_visible(
110 select(
111 GroupChatSubscription.group_chat_id.label("conversation_id"),
112 func.max(Message.id).label("latest_message_id"),
113 literal(_KIND_GROUP_CHAT).label("kind"),
114 )
115 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
116 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
117 .where(GroupChatSubscription.user_id == context.user_id)
118 .where(is_newest_subscription(context.user_id))
119 .where(was_subscribed_at(GroupChatSubscription, Message.time))
120 .where(or_(to_bool(only_archived is None), GroupChatSubscription.is_archived == only_archived))
121 .where(or_(to_bool(not unread), is_unseen(Message, GroupChatSubscription)))
122 .group_by(GroupChatSubscription.group_chat_id),
123 context,
124 GroupChat,
125 is_list_operation=True,
126 )
129def _build_host_request_select_query(
130 context: CouchersContext, role_filter: ColumnElement[bool], only_archived: bool | None, unread: bool
131) -> Select[tuple[int, int, str]]:
132 """
133 The host requests and public-trip offers the viewer should see, narrowed by the request's
134 archived and unread filters, along with the id of each one's newest message.
136 role_filter picks which side of the request the viewer is on: hosting, surfing, or offers on
137 their own public trips.
138 """
139 query = (
140 select(
141 HostRequest.conversation_id.label("conversation_id"),
142 # correlated, so it resolves per row, and never NULL: creating a request writes its
143 # first message
144 select(func.max(Message.id))
145 .where(Message.conversation_id == HostRequest.conversation_id)
146 .scalar_subquery()
147 .label("latest_message_id"),
148 literal(_KIND_HOST_REQUEST).label("kind"),
149 )
150 .where(
151 or_(
152 HostRequest.initiator_user_id == context.user_id,
153 HostRequest.recipient_user_id == context.user_id,
154 )
155 )
156 .where(role_filter)
157 .where(
158 or_(
159 to_bool(only_archived is None),
160 and_(
161 HostRequest.initiator_user_id == context.user_id,
162 HostRequest.is_initiator_archived == only_archived,
163 ),
164 and_(
165 HostRequest.recipient_user_id == context.user_id,
166 HostRequest.is_recipient_archived == only_archived,
167 ),
168 )
169 )
170 .where(or_(to_bool(not unread), has_unseen_host_request_messages(context.user_id)))
171 )
172 query = where_users_column_visible(query, context, HostRequest.initiator_user_id)
173 query = where_users_column_visible(query, context, HostRequest.recipient_user_id)
174 query = where_moderated_content_visible(query, context, HostRequest, is_list_operation=True)
175 return query
178def _build_host_request_role_filter(user_id: int, categories: frozenset[int]) -> ColumnElement[bool] | None:
179 """
180 The stay-roles the selected categories cover, or None if none of them is a host request category.
181 MY_PUBLIC_TRIPS is a subset of SURFING, so asking for both is redundant but harmless.
182 """
183 clauses = []
184 if conversations_pb2.MESSAGE_THREAD_CATEGORY_HOSTING in categories:
185 clauses.append(is_hosting_party(user_id))
186 if conversations_pb2.MESSAGE_THREAD_CATEGORY_SURFING in categories:
187 clauses.append(is_surfing_party(user_id))
188 if conversations_pb2.MESSAGE_THREAD_CATEGORY_MY_PUBLIC_TRIPS in categories:
189 clauses.append(is_public_trip_offer_recipient(user_id))
190 return or_(*clauses) if clauses else None
193def _host_request_thread_to_pb(
194 host_request: HostRequest,
195 conversation: Conversation,
196 message: Message,
197 user_id: int,
198 unseen_message_count: int,
199 need_host_request_feedback: bool,
200) -> requests_pb2.HostRequest:
201 """
202 Build the HostRequest protobuf for the unified thread list. Mirrors ListHostRequests (batched, so
203 no per-request queries like host_request_to_pb), with the same semantics: surfer/host are the
204 stay roles, which a public-trip offer reverses, while last_seen and archived key off the
205 conversation roles.
206 """
207 lat, lng = get_coordinates(host_request.hosting_location)
208 return requests_pb2.HostRequest(
209 host_request_id=host_request.conversation_id,
210 surfer_user_id=host_request.surfer_user_id,
211 host_user_id=host_request.host_user_id,
212 status=hostrequeststatus2api[host_request.status],
213 created=Timestamp_from_datetime(conversation.created),
214 from_date=date_to_api(host_request.from_date),
215 to_date=date_to_api(host_request.to_date),
216 last_seen_message_id=(
217 host_request.initiator_last_seen_message_id
218 if host_request.initiator_user_id == user_id
219 else host_request.recipient_last_seen_message_id
220 ),
221 latest_message=message_to_pb(message),
222 hosting_city=host_request.hosting_city,
223 hosting_lat=lat,
224 hosting_lng=lng,
225 hosting_radius=host_request.hosting_radius,
226 need_host_request_feedback=need_host_request_feedback,
227 is_archived=(
228 host_request.is_initiator_archived
229 if host_request.initiator_user_id == user_id
230 else host_request.is_recipient_archived
231 ),
232 public_trip_id=host_request.public_trip_id,
233 unseen_message_count=unseen_message_count,
234 )
237def _build_group_chats_pb(
238 session: Session, context: CouchersContext, threads: Sequence[_ThreadRow]
239) -> dict[int, conversations_pb2.GroupChat]:
240 """
241 Build GroupChat protobufs (with unseen counts) for the group chats on a page.
243 Each row already carries the id of the chat's latest message, windowed to the viewer's
244 subscription, so that rule doesn't have to be restated here.
245 """
246 if not threads:
247 return {}
248 group_chat_ids = [thread.conversation_id for thread in threads]
249 latest_message_ids = [thread.latest_message_id for thread in threads]
251 unseen_count_by_group_chat: dict[int, int] = dict(
252 session.execute( # type: ignore[arg-type]
253 select(GroupChatSubscription.group_chat_id, func.count(Message.id))
254 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
255 .where(GroupChatSubscription.group_chat_id.in_(group_chat_ids))
256 .where(GroupChatSubscription.user_id == context.user_id)
257 .where(is_newest_subscription(context.user_id))
258 .where(is_unseen(Message, GroupChatSubscription))
259 .group_by(GroupChatSubscription.group_chat_id)
260 ).all()
261 )
263 # the chat, its conversation, the viewer's own subscription, its latest message (id already
264 # known), its visible roster and whether the viewer can message it, in one query
265 member = aliased(GroupChatSubscription)
266 member_user_ids = func.array_agg(aggregate_order_by(member.user_id, member.user_id))
267 # same roster rule as _get_visible_members_for_subscription / _get_visible_admins_for_subscription
268 # in the conversations servicer: a viewer still in the chat sees everyone currently in it, and one
269 # who has left sees the roster frozen at the moment they left
270 member_visible = case(
271 (GroupChatSubscription.left.is_(None), member.left.is_(None)),
272 # the else_ branch only runs where left is non-NULL, which the annotation can't express
273 else_=was_subscribed_at(member, GroupChatSubscription.left), # type: ignore[arg-type]
274 )
275 # same rule as _user_can_message in the conversations servicer: a true group chat can always be
276 # messaged, a DM only while the other party is still in it and visible to the viewer
277 other_party = aliased(GroupChatSubscription)
278 can_message = or_(
279 ~GroupChat.is_dm,
280 where_users_column_visible(
281 select(1)
282 .select_from(other_party)
283 .where(other_party.group_chat_id == GroupChat.conversation_id)
284 .where(other_party.user_id != context.user_id)
285 .where(other_party.left.is_(None)),
286 context,
287 other_party.user_id,
288 )
289 .exists()
290 .correlate(GroupChat),
291 )
292 rows = session.execute(
293 select(
294 GroupChat,
295 Conversation,
296 GroupChatSubscription,
297 Message,
298 member_user_ids,
299 member_user_ids.filter(member.role == GroupChatRole.admin),
300 can_message,
301 )
302 .join(Conversation, Conversation.id == GroupChat.conversation_id)
303 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == GroupChat.conversation_id)
304 .join(Message, and_(Message.conversation_id == GroupChat.conversation_id, Message.id.in_(latest_message_ids)))
305 .join(member, and_(member.group_chat_id == GroupChat.conversation_id, member_visible))
306 .where(GroupChat.conversation_id.in_(group_chat_ids))
307 .where(GroupChatSubscription.user_id == context.user_id)
308 .where(is_newest_subscription(context.user_id))
309 .group_by(GroupChat.conversation_id, Conversation.id, GroupChatSubscription.id, Message.id)
310 ).all()
312 return {
313 group_chat.conversation_id: conversations_pb2.GroupChat(
314 group_chat_id=group_chat.conversation_id,
315 title=group_chat.title, # TODO: proper title for DMs, etc
316 member_user_ids=members,
317 # array_agg over an empty filter is NULL rather than an empty array
318 admin_user_ids=admins or [],
319 only_admins_invite=group_chat.only_admins_invite,
320 is_dm=group_chat.is_dm,
321 created=Timestamp_from_datetime(conversation.created),
322 unseen_message_count=unseen_count_by_group_chat.get(group_chat.conversation_id, 0),
323 last_seen_message_id=subscription.last_seen_message_id,
324 latest_message=message_to_pb(message),
325 mute_info=mute_info(subscription),
326 can_message=can_message,
327 is_archived=subscription.is_archived,
328 )
329 for group_chat, conversation, subscription, message, members, admins, can_message in rows
330 }
333def _build_host_request_threads_pb(
334 session: Session, context: CouchersContext, threads: Sequence[_ThreadRow]
335) -> dict[int, requests_pb2.HostRequest]:
336 """Build HostRequest protobufs (with unseen counts) for the host requests on a page."""
337 if not threads:
338 return {}
339 host_request_ids = [thread.conversation_id for thread in threads]
340 latest_message_ids = [thread.latest_message_id for thread in threads]
342 # same rule as host_request_to_pb: the host is asked for feedback once they've rejected a request
343 # and haven't given any yet.
344 # TODO(#9347): the recipient-based logic is wrong for public-trip offers, where the recipient is
345 # the traveller rather than the host — same for the response-rate observation.
346 need_host_request_feedback = and_(
347 HostRequest.recipient_user_id == context.user_id,
348 HostRequest.status == HostRequestStatus.rejected,
349 ~exists()
350 .where(HostRequestFeedback.from_user_id == context.user_id)
351 .where(HostRequestFeedback.host_request_id == HostRequest.conversation_id)
352 .correlate(HostRequest),
353 )
355 # the request, its conversation, its latest message (id already known), its unseen count and
356 # whether it's owed feedback, in one query
357 rows = session.execute(
358 select(
359 HostRequest,
360 Conversation,
361 Message,
362 unseen_host_request_message_count(context.user_id),
363 need_host_request_feedback,
364 )
365 .join(Conversation, Conversation.id == HostRequest.conversation_id)
366 .join(Message, and_(Message.conversation_id == HostRequest.conversation_id, Message.id.in_(latest_message_ids)))
367 .where(HostRequest.conversation_id.in_(host_request_ids))
368 ).all()
369 return {
370 host_request.conversation_id: _host_request_thread_to_pb(
371 host_request, conversation, message, context.user_id, unseen_message_count, needs_feedback
372 )
373 for host_request, conversation, message, unseen_message_count, needs_feedback in rows
374 }
377def list_message_threads(
378 request: conversations_pb2.ListMessageThreadsReq, context: CouchersContext, session: Session
379) -> conversations_pb2.ListMessageThreadsRes:
380 filters = _resolve_thread_filters(context, request)
382 queries = []
383 if conversations_pb2.MESSAGE_THREAD_CATEGORY_CHATS in filters.categories:
384 queries.append(_build_group_chat_select_query(context, filters.only_archived, filters.only_unread))
385 role_filter = _build_host_request_role_filter(context.user_id, filters.categories)
386 if role_filter is not None:
387 queries.append(
388 _build_host_request_select_query(context, role_filter, filters.only_archived, filters.only_unread)
389 )
391 # nothing to include: only reachable when MY_PUBLIC_TRIPS is requested alone and the
392 # public-trips flag is off. TODO: remove once public trips is live (flag always on)
393 if not queries:
394 return conversations_pb2.ListMessageThreadsRes()
396 page_size = min(request.page_size or DEFAULT_PAGINATION_LENGTH, MAX_PAGE_SIZE)
398 # unioned, so one cursor on latest_message_id pages through both kinds as a single list
399 threads = (queries[0] if len(queries) == 1 else union_all(*queries)).subquery()
400 page_query = select(threads.c.conversation_id, threads.c.latest_message_id, threads.c.kind)
401 if request.page_token:
402 page_query = page_query.where(threads.c.latest_message_id < int(decrypt_page_token(request.page_token)))
403 page_query = page_query.order_by(threads.c.latest_message_id.desc()).limit(page_size + 1)
404 rows = session.execute(page_query).all()
406 page_rows = rows[:page_size]
407 has_more = len(rows) > page_size
408 # rows are ordered by latest_message_id desc, so the last one on the page is the cursor
409 next_page_token = encrypt_page_token(str(page_rows[-1].latest_message_id)) if has_more else ""
411 # hydrate each kind in a batch, then re-assemble in the paginated order
412 group_chats_by_id = _build_group_chats_pb(
413 session, context, [row for row in page_rows if row.kind == _KIND_GROUP_CHAT]
414 )
415 host_request_threads_by_id = _build_host_request_threads_pb(
416 session, context, [row for row in page_rows if row.kind == _KIND_HOST_REQUEST]
417 )
419 message_threads = []
420 for row in page_rows:
421 if row.kind == _KIND_GROUP_CHAT:
422 group_chat = group_chats_by_id.get(row.conversation_id)
423 if group_chat is not None: 423 ↛ 420line 423 didn't jump to line 420 because the condition on line 423 was always true
424 message_threads.append(conversations_pb2.MessageThread(group_chat=group_chat))
425 else:
426 host_request_thread = host_request_threads_by_id.get(row.conversation_id)
427 if host_request_thread is not None: 427 ↛ 420line 427 didn't jump to line 420 because the condition on line 427 was always true
428 message_threads.append(conversations_pb2.MessageThread(host_request=host_request_thread))
430 return conversations_pb2.ListMessageThreadsRes(threads=message_threads, next_page_token=next_page_token)
433def mark_all_threads_seen(
434 request: conversations_pb2.MarkAllThreadsSeenReq, context: CouchersContext, session: Session
435) -> empty_pb2.Empty:
436 filters = _resolve_thread_filters(context, request)
438 # (topic actions, keys) groups for the notifications owned by the threads we mark seen
439 notification_groups: list[tuple[Sequence[NotificationTopicAction], Sequence[str]]] = []
441 if conversations_pb2.MESSAGE_THREAD_CATEGORY_CHATS in filters.categories: 441 ↛ 477line 441 didn't jump to line 477 because the condition on line 441 was always true
442 chat_query = _build_group_chat_select_query(context, filters.only_archived, filters.only_unread)
443 # correlated, so it resolves per row of the update: every subscription advances to its own
444 # chat's newest message without listing them out. windowed, so a subscription the viewer
445 # has left advances only to the last message they could read, matching its unseen count
446 latest_reachable_message_id = (
447 select(func.max(Message.id))
448 .where(Message.conversation_id == GroupChatSubscription.group_chat_id)
449 .where(was_subscribed_at(GroupChatSubscription, Message.time))
450 .scalar_subquery()
451 )
452 marked_group_chat_ids = (
453 session.execute(
454 update(GroupChatSubscription)
455 .where(GroupChatSubscription.user_id == context.user_id)
456 .where(is_newest_subscription(context.user_id))
457 .where(GroupChatSubscription.group_chat_id.in_(select(chat_query.subquery().c.conversation_id)))
458 .where(GroupChatSubscription.last_seen_message_id < latest_reachable_message_id)
459 .values(last_seen_message_id=latest_reachable_message_id)
460 .returning(GroupChatSubscription.group_chat_id)
461 .execution_options(synchronize_session=False)
462 )
463 .scalars()
464 .all()
465 )
466 if marked_group_chat_ids:
467 notification_groups.append(
468 (
469 [NotificationTopicAction.chat__message],
470 [str(group_chat_id) for group_chat_id in marked_group_chat_ids],
471 )
472 )
473 # chat__missed_messages is a summary across all chats, so it's keyed with an empty
474 # string rather than a chat id (same as MarkLastSeenGroupChat)
475 notification_groups.append(([NotificationTopicAction.chat__missed_messages], [""]))
477 role_filter = _build_host_request_role_filter(context.user_id, filters.categories)
478 if role_filter is not None:
479 host_request_query = _build_host_request_select_query(
480 context, role_filter, filters.only_archived, filters.only_unread
481 )
482 # the viewer's last-seen column depends on their role, so one update per role
483 # (a user is never both initiator and recipient of the same request, so these are disjoint)
484 matching_ids = select(host_request_query.subquery().c.conversation_id)
485 # correlated, so it resolves per row of the update: every request advances to its own newest
486 # message without listing them out
487 latest_message_id = (
488 select(func.max(Message.id)).where(Message.conversation_id == HostRequest.conversation_id).scalar_subquery()
489 )
490 marked_conversation_ids: list[int] = []
491 for user_id_column, last_seen_column in (
492 (HostRequest.initiator_user_id, HostRequest.initiator_last_seen_message_id),
493 (HostRequest.recipient_user_id, HostRequest.recipient_last_seen_message_id),
494 ):
495 marked_conversation_ids += (
496 session.execute(
497 update(HostRequest)
498 .where(user_id_column == context.user_id)
499 .where(HostRequest.conversation_id.in_(matching_ids))
500 .where(last_seen_column < latest_message_id)
501 .values({last_seen_column: latest_message_id})
502 .returning(HostRequest.conversation_id)
503 .execution_options(synchronize_session=False)
504 )
505 .scalars()
506 .all()
507 )
508 if marked_conversation_ids:
509 notification_groups.append(
510 (
511 HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS,
512 [str(conversation_id) for conversation_id in marked_conversation_ids],
513 )
514 )
516 mark_notifications_seen(session, user_id=context.user_id, topic_actions_and_keys=notification_groups)
518 return empty_pb2.Empty()