Coverage for app/backend/src/couchers/servicers/message_threads.py: 96%
65 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-08 23:56 +0000
1"""
2Bulk operations over the viewer's message threads — group chats, DMs, host requests and public-trip
3offers — selected by category (Conversations.MarkAllThreadsSeen).
4"""
6import logging
7from collections.abc import Sequence
8from dataclasses import dataclass
10import grpc
11from google.protobuf import empty_pb2
12from sqlalchemy import ColumnElement, Select, select, update
13from sqlalchemy.orm import Session
14from sqlalchemy.sql import and_, func, or_
16from couchers.context import CouchersContext
17from couchers.helpers.group_chats import is_newest_subscription, is_unseen, was_subscribed_at
18from couchers.helpers.host_requests import (
19 HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS,
20 has_unseen_host_request_messages,
21 is_hosting_party,
22 is_public_trip_offer_recipient,
23 is_surfing_party,
24)
25from couchers.models import (
26 GroupChat,
27 GroupChatSubscription,
28 HostRequest,
29 Message,
30)
31from couchers.models.notifications import NotificationTopicAction
32from couchers.notifications.notify import mark_notifications_seen
33from couchers.proto import conversations_pb2
34from couchers.sql import to_bool, where_moderated_content_visible, where_users_column_visible
36logger = logging.getLogger(__name__)
39@dataclass(frozen=True)
40class _ThreadFilters:
41 categories: frozenset[int]
42 only_archived: bool | None
43 only_unread: bool
46def _resolve_thread_filters(
47 context: CouchersContext, request: conversations_pb2.MarkAllThreadsSeenReq
48) -> _ThreadFilters:
49 """
50 An empty category list means all categories. MY_PUBLIC_TRIPS is dropped when the public-trips
51 flag is off; those offers still show up under SURFING.
52 """
53 if conversations_pb2.MESSAGE_THREAD_CATEGORY_UNSPECIFIED in request.categories:
54 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_thread_category")
55 categories = frozenset(
56 request.categories
57 or {
58 conversations_pb2.MESSAGE_THREAD_CATEGORY_CHATS,
59 conversations_pb2.MESSAGE_THREAD_CATEGORY_HOSTING,
60 conversations_pb2.MESSAGE_THREAD_CATEGORY_SURFING,
61 conversations_pb2.MESSAGE_THREAD_CATEGORY_MY_PUBLIC_TRIPS,
62 }
63 )
64 if not context.get_boolean_value("public_trips_enabled", False): 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true
65 categories -= {conversations_pb2.MESSAGE_THREAD_CATEGORY_MY_PUBLIC_TRIPS}
66 return _ThreadFilters(
67 categories=categories,
68 only_archived=request.only_archived if request.HasField("only_archived") else None,
69 only_unread=request.only_unread,
70 )
73def _build_group_chat_select_query(
74 context: CouchersContext, only_archived: bool | None, unread: bool
75) -> Select[tuple[int]]:
76 """
77 The ids of the group chats (including DMs) the viewer should see, narrowed by the request's
78 archived and unread filters.
79 """
80 return where_moderated_content_visible(
81 select(GroupChatSubscription.group_chat_id.label("conversation_id"))
82 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
83 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id)
84 .where(GroupChatSubscription.user_id == context.user_id)
85 .where(is_newest_subscription(context.user_id))
86 .where(was_subscribed_at(GroupChatSubscription, Message.time))
87 .where(or_(to_bool(only_archived is None), GroupChatSubscription.is_archived == only_archived))
88 .where(or_(to_bool(not unread), is_unseen(Message, GroupChatSubscription)))
89 .group_by(GroupChatSubscription.group_chat_id),
90 context,
91 GroupChat,
92 is_list_operation=True,
93 )
96def _build_host_request_select_query(
97 context: CouchersContext, role_filter: ColumnElement[bool], only_archived: bool | None, unread: bool
98) -> Select[tuple[int]]:
99 """
100 The ids of the host requests and public-trip offers the viewer should see, narrowed by the
101 request's archived and unread filters.
103 role_filter picks which side of the request the viewer is on: hosting, surfing, or offers on
104 their own public trips.
105 """
106 query = (
107 select(HostRequest.conversation_id.label("conversation_id"))
108 .where(
109 or_(
110 HostRequest.initiator_user_id == context.user_id,
111 HostRequest.recipient_user_id == context.user_id,
112 )
113 )
114 .where(role_filter)
115 .where(
116 or_(
117 to_bool(only_archived is None),
118 and_(
119 HostRequest.initiator_user_id == context.user_id,
120 HostRequest.is_initiator_archived == only_archived,
121 ),
122 and_(
123 HostRequest.recipient_user_id == context.user_id,
124 HostRequest.is_recipient_archived == only_archived,
125 ),
126 )
127 )
128 .where(or_(to_bool(not unread), has_unseen_host_request_messages(context.user_id)))
129 )
130 query = where_users_column_visible(query, context, HostRequest.initiator_user_id)
131 query = where_users_column_visible(query, context, HostRequest.recipient_user_id)
132 query = where_moderated_content_visible(query, context, HostRequest, is_list_operation=True)
133 return query
136def _build_host_request_role_filter(user_id: int, categories: frozenset[int]) -> ColumnElement[bool] | None:
137 """
138 The stay-roles the selected categories cover, or None if none of them is a host request category.
139 MY_PUBLIC_TRIPS is a subset of SURFING, so asking for both is redundant but harmless.
140 """
141 clauses = []
142 if conversations_pb2.MESSAGE_THREAD_CATEGORY_HOSTING in categories:
143 clauses.append(is_hosting_party(user_id))
144 if conversations_pb2.MESSAGE_THREAD_CATEGORY_SURFING in categories:
145 clauses.append(is_surfing_party(user_id))
146 if conversations_pb2.MESSAGE_THREAD_CATEGORY_MY_PUBLIC_TRIPS in categories:
147 clauses.append(is_public_trip_offer_recipient(user_id))
148 return or_(*clauses) if clauses else None
151def mark_all_threads_seen(
152 request: conversations_pb2.MarkAllThreadsSeenReq, context: CouchersContext, session: Session
153) -> empty_pb2.Empty:
154 filters = _resolve_thread_filters(context, request)
156 # (topic actions, keys) groups for the notifications owned by the threads we mark seen
157 notification_groups: list[tuple[Sequence[NotificationTopicAction], Sequence[str]]] = []
159 if conversations_pb2.MESSAGE_THREAD_CATEGORY_CHATS in filters.categories: 159 ↛ 195line 159 didn't jump to line 195 because the condition on line 159 was always true
160 chat_query = _build_group_chat_select_query(context, filters.only_archived, filters.only_unread)
161 # correlated, so it resolves per row of the update: every subscription advances to its own
162 # chat's newest message without listing them out. windowed, so a subscription the viewer
163 # has left advances only to the last message they could read, matching its unseen count
164 latest_message_id = (
165 select(func.max(Message.id))
166 .where(Message.conversation_id == GroupChatSubscription.group_chat_id)
167 .where(was_subscribed_at(GroupChatSubscription, Message.time))
168 .scalar_subquery()
169 )
170 marked_group_chat_ids = (
171 session.execute(
172 update(GroupChatSubscription)
173 .where(GroupChatSubscription.user_id == context.user_id)
174 .where(is_newest_subscription(context.user_id))
175 .where(GroupChatSubscription.group_chat_id.in_(select(chat_query.subquery().c.conversation_id)))
176 .where(GroupChatSubscription.last_seen_message_id < latest_message_id)
177 .values(last_seen_message_id=latest_message_id)
178 .returning(GroupChatSubscription.group_chat_id)
179 .execution_options(synchronize_session=False)
180 )
181 .scalars()
182 .all()
183 )
184 if marked_group_chat_ids:
185 notification_groups.append(
186 (
187 [NotificationTopicAction.chat__message],
188 [str(group_chat_id) for group_chat_id in marked_group_chat_ids],
189 )
190 )
191 # chat__missed_messages is a summary across all chats, so it's keyed with an empty
192 # string rather than a chat id (same as MarkLastSeenGroupChat)
193 notification_groups.append(([NotificationTopicAction.chat__missed_messages], [""]))
195 role_filter = _build_host_request_role_filter(context.user_id, filters.categories)
196 if role_filter is not None:
197 host_request_query = _build_host_request_select_query(
198 context, role_filter, filters.only_archived, filters.only_unread
199 )
200 # the viewer's last-seen column depends on their role, so one update per role
201 # (a user is never both initiator and recipient of the same request, so these are disjoint)
202 matching_ids = select(host_request_query.subquery().c.conversation_id)
203 latest_message_id = (
204 select(func.max(Message.id)).where(Message.conversation_id == HostRequest.conversation_id).scalar_subquery()
205 )
206 marked_conversation_ids: list[int] = []
207 for user_id_column, last_seen_column in (
208 (HostRequest.initiator_user_id, HostRequest.initiator_last_seen_message_id),
209 (HostRequest.recipient_user_id, HostRequest.recipient_last_seen_message_id),
210 ):
211 marked_conversation_ids += (
212 session.execute(
213 update(HostRequest)
214 .where(user_id_column == context.user_id)
215 .where(HostRequest.conversation_id.in_(matching_ids))
216 .where(last_seen_column < latest_message_id)
217 .values({last_seen_column: latest_message_id})
218 .returning(HostRequest.conversation_id)
219 .execution_options(synchronize_session=False)
220 )
221 .scalars()
222 .all()
223 )
224 if marked_conversation_ids:
225 notification_groups.append(
226 (
227 HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS,
228 [str(conversation_id) for conversation_id in marked_conversation_ids],
229 )
230 )
232 mark_notifications_seen(session, user_id=context.user_id, topic_actions_and_keys=notification_groups)
234 return empty_pb2.Empty()