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