Coverage for app/backend/src/couchers/servicers/requests.py: 92%
341 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import logging
2from datetime import timedelta
4import grpc
5from google.protobuf import empty_pb2
6from sqlalchemy import exists, select
7from sqlalchemy.orm import Session, aliased
8from sqlalchemy.sql import and_, func, or_
10from couchers.constants import HOST_REQUEST_DUPLICATE_WINDOW_HOURS, HOST_REQUEST_MIN_LENGTH_UTF16
11from couchers.context import CouchersContext, make_notification_user_context
12from couchers.db import can_moderate_node
13from couchers.event_log import log_event
14from couchers.helpers.completed_profile import has_completed_profile
15from couchers.helpers.host_requests import HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS, unseen_host_request_message_count
16from couchers.helpers.messages import api2hostrequeststatus, hostrequeststatus2api, message_to_pb
17from couchers.materialized_views import UserResponseRate
18from couchers.metrics import (
19 account_age_on_host_request_create_histogram,
20 host_request_first_response_histogram,
21 host_request_responses_counter,
22 host_requests_sent_counter,
23 sent_messages_counter,
24)
25from couchers.models import (
26 Conversation,
27 HostRequest,
28 HostRequestFeedback,
29 HostRequestQuality,
30 HostRequestStatus,
31 Message,
32 MessageType,
33 ModerationObjectType,
34 RateLimitAction,
35 User,
36)
37from couchers.models.notifications import NotificationTopicAction
38from couchers.models.public_trips import PublicTrip, PublicTripStatus
39from couchers.moderation.utils import create_moderation
40from couchers.notifications.notify import mark_notifications_seen, notify
41from couchers.proto import (
42 messages_pb2,
43 notification_data_pb2,
44 requests_pb2,
45 requests_pb2_grpc,
46)
47from couchers.rate_limits.check import process_rate_limits_and_check_abort
48from couchers.rate_limits.definitions import RATE_LIMIT_HOURS
49from couchers.servicers.api import response_rate_to_pb, user_model_to_pb
50from couchers.sql import to_bool, users_visible, where_moderated_content_visible, where_users_column_visible
51from couchers.utils import (
52 Timestamp_from_datetime,
53 date_to_api,
54 get_coordinates,
55 now,
56 parse_date,
57 today_in_timezone,
58)
60logger = logging.getLogger(__name__)
62DEFAULT_PAGINATION_LENGTH = 10
63MAX_PAGE_SIZE = 50
66hostrequestquality2sql = {
67 requests_pb2.HOST_REQUEST_QUALITY_UNSPECIFIED: HostRequestQuality.high_quality,
68 requests_pb2.HOST_REQUEST_QUALITY_LOW: HostRequestQuality.okay_quality,
69 requests_pb2.HOST_REQUEST_QUALITY_OKAY: HostRequestQuality.low_quality,
70}
73def host_request_to_pb(
74 host_request: HostRequest, session: Session, context: CouchersContext
75) -> requests_pb2.HostRequest:
76 initial_message = session.execute(
77 select(Message)
78 .where(Message.conversation_id == host_request.conversation_id)
79 .order_by(Message.id.asc())
80 .limit(1)
81 ).scalar_one()
83 latest_message = session.execute(
84 select(Message)
85 .where(Message.conversation_id == host_request.conversation_id)
86 .order_by(Message.id.desc())
87 .limit(1)
88 ).scalar_one()
90 unseen_message_count = session.execute(
91 select(unseen_host_request_message_count(context.user_id)).where(
92 HostRequest.conversation_id == host_request.conversation_id
93 )
94 ).scalar_one()
96 lat, lng = get_coordinates(host_request.hosting_location)
98 need_feedback = False
99 if context.user_id == host_request.recipient_user_id and host_request.status == HostRequestStatus.rejected:
100 need_feedback = not session.execute(
101 select(
102 exists().where(
103 HostRequestFeedback.from_user_id == context.user_id,
104 HostRequestFeedback.host_request_id == host_request.conversation_id,
105 )
106 )
107 ).scalar_one()
109 return requests_pb2.HostRequest(
110 host_request_id=host_request.conversation_id,
111 surfer_user_id=host_request.surfer_user_id,
112 host_user_id=host_request.host_user_id,
113 status=hostrequeststatus2api[host_request.status],
114 created=Timestamp_from_datetime(initial_message.time),
115 from_date=date_to_api(host_request.from_date),
116 to_date=date_to_api(host_request.to_date),
117 last_seen_message_id=(
118 host_request.initiator_last_seen_message_id
119 if context.user_id == host_request.initiator_user_id
120 else host_request.recipient_last_seen_message_id
121 ),
122 latest_message=message_to_pb(latest_message),
123 hosting_city=host_request.hosting_city,
124 hosting_lat=lat,
125 hosting_lng=lng,
126 hosting_radius=host_request.hosting_radius,
127 need_host_request_feedback=need_feedback,
128 is_archived=(
129 host_request.is_recipient_archived
130 if context.user_id == host_request.recipient_user_id
131 else host_request.is_initiator_archived
132 ),
133 public_trip_id=host_request.public_trip_id,
134 unseen_message_count=unseen_message_count,
135 )
138def _possibly_observe_first_response_time(
139 session: Session, host_request: HostRequest, user_id: int, response_type: str
140) -> None:
141 # if this is the first response then there's nothing by this user yet
142 assert host_request.recipient_user_id == user_id
144 number_messages_by_host = session.execute(
145 select(func.count())
146 .where(Message.conversation_id == host_request.conversation_id)
147 .where(Message.author_id == user_id)
148 ).scalar_one_or_none()
150 if number_messages_by_host == 0:
151 host_gender = session.execute(select(User.gender).where(User.id == host_request.host_user_id)).scalar_one()
152 surfer_gender = session.execute(select(User.gender).where(User.id == host_request.surfer_user_id)).scalar_one()
153 host_request_first_response_histogram.labels(host_gender, surfer_gender, response_type).observe(
154 (now() - host_request.conversation.created).total_seconds()
155 )
158def _is_host_request_long_enough(text: str) -> bool:
159 # Python's len(str) does not match Javascript's string.length.
160 # e.g. len("é") == 2 but "é".length == 1.
161 # To match the frontend's validation, measure the string in utf16 code units.
162 text_length_utf16 = len(text.encode("utf-16-le")) // 2 # utf-16-le does not include a prefix BOM code unit.
163 return text_length_utf16 >= HOST_REQUEST_MIN_LENGTH_UTF16
166class Requests(requests_pb2_grpc.RequestsServicer):
167 def CreateHostRequest(
168 self, request: requests_pb2.CreateHostRequestReq, context: CouchersContext, session: Session
169 ) -> requests_pb2.CreateHostRequestRes:
170 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
171 if not has_completed_profile(session, user):
172 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_request")
174 if request.host_user_id == context.user_id:
175 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_request_self")
177 # just to check recipient exists and is visible
178 recipient = session.execute(
179 select(User).where(users_visible(context, User)).where(User.id == request.host_user_id)
180 ).scalar_one_or_none()
181 if not recipient:
182 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
184 from_date = parse_date(request.from_date)
185 to_date = parse_date(request.to_date)
187 if not from_date or not to_date:
188 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_date")
190 today = today_in_timezone(recipient.timezone)
192 # request starts from the past
193 if from_date < today:
194 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "date_from_before_today")
196 # from_date is not >= to_date
197 if from_date >= to_date:
198 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "date_from_after_to")
200 # No need to check today > to_date
202 if from_date - today > timedelta(days=365):
203 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "date_from_after_one_year")
205 if to_date - from_date > timedelta(days=365):
206 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "date_to_after_one_year")
208 # Check minimum length
209 if not _is_host_request_long_enough(request.text):
210 context.abort_with_error_code(
211 grpc.StatusCode.INVALID_ARGUMENT,
212 "host_request_too_short2",
213 substitutions={"count": HOST_REQUEST_MIN_LENGTH_UTF16},
214 )
216 # Check if user has been sending host requests excessively
217 if process_rate_limits_and_check_abort(
218 session=session, user_id=context.user_id, action=RateLimitAction.host_request
219 ):
220 context.abort_with_error_code(
221 grpc.StatusCode.RESOURCE_EXHAUSTED,
222 "host_request_rate_limit2",
223 substitutions={"count": RATE_LIMIT_HOURS},
224 )
226 # If this is an offer in response to a public trip, validate it
227 public_trip_id = request.public_trip_id if request.HasField("public_trip_id") else None
229 # Offers on public trips are deduplicated per trip further down instead
230 if public_trip_id is None:
231 recent_request = session.execute(
232 select(HostRequest.conversation_id)
233 .join(Conversation, HostRequest.conversation_id == Conversation.id)
234 .where(HostRequest.initiator_user_id == context.user_id)
235 .where(HostRequest.recipient_user_id == recipient.id)
236 .where(HostRequest.public_trip_id == None)
237 .where(Conversation.created >= now() - timedelta(hours=HOST_REQUEST_DUPLICATE_WINDOW_HOURS))
238 # overlapping nights, so back-to-back stays are still allowed
239 .where(HostRequest.from_date < to_date)
240 .where(HostRequest.to_date > from_date)
241 .limit(1)
242 ).scalar_one_or_none()
243 if recent_request is not None:
244 context.abort_with_error_code(
245 grpc.StatusCode.FAILED_PRECONDITION,
246 "duplicate_host_request",
247 substitutions={"count": HOST_REQUEST_DUPLICATE_WINDOW_HOURS},
248 )
250 if public_trip_id is not None:
251 public_trip = session.execute(
252 where_moderated_content_visible(select(PublicTrip), context, PublicTrip, is_list_operation=False).where(
253 PublicTrip.id == public_trip_id
254 )
255 ).scalar_one_or_none()
256 if not public_trip:
257 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "public_trip_not_found")
258 # The trip's traveler must be the recipient of this host request (role reversal)
259 if public_trip.user_id != recipient.id:
260 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "public_trip_user_mismatch")
261 # Trip must still be active
262 if public_trip.status != PublicTripStatus.searching_for_host:
263 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "public_trip_not_active")
264 # Offered dates must fall within the trip's window (host can shorten, not extend)
265 if from_date < public_trip.from_date or to_date > public_trip.to_date:
266 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "public_trip_dates_out_of_range")
267 # Enforce same_gender_only restriction (community moderators bypass)
268 if (
269 public_trip.same_gender_only
270 and not can_moderate_node(session, context.user_id, public_trip.node_id)
271 and user.gender != recipient.gender
272 ):
273 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "public_trip_same_gender_only")
274 # Prevent duplicate offers on the same trip
275 existing_offer = session.execute(
276 select(HostRequest)
277 .where(HostRequest.public_trip_id == public_trip_id)
278 .where(HostRequest.initiator_user_id == context.user_id)
279 ).scalar_one_or_none()
280 if existing_offer:
281 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "duplicate_host_request_for_trip")
283 # an offer on a public trip reverses the roles: the caller is the host and the recipient is the traveller
284 surfer_user, host_user = (recipient, user) if public_trip_id is not None else (user, recipient)
286 conversation = Conversation()
287 session.add(conversation)
288 session.flush()
290 session.add(
291 Message(
292 conversation_id=conversation.id,
293 author_id=context.user_id,
294 message_type=MessageType.chat_created,
295 )
296 )
298 message = Message(
299 conversation_id=conversation.id,
300 author_id=context.user_id,
301 text=request.text,
302 message_type=MessageType.text,
303 )
304 session.add(message)
305 session.flush()
307 # Create moderation state for UMS (starts as SHADOWED)
308 moderation_state = create_moderation(
309 session=session,
310 object_type=ModerationObjectType.host_request,
311 object_id=conversation.id,
312 creator_user_id=context.user_id,
313 )
315 host_request = HostRequest(
316 conversation_id=conversation.id,
317 initiator_user_id=context.user_id,
318 recipient_user_id=recipient.id,
319 moderation_state_id=moderation_state.id,
320 from_date=from_date,
321 to_date=to_date,
322 status=HostRequestStatus.pending,
323 initiator_last_seen_message_id=message.id,
324 # TODO: tz
325 # timezone=recipient.timezone,
326 hosting_city=host_user.city,
327 hosting_location=host_user.geom,
328 hosting_radius=host_user.geom_radius,
329 public_trip_id=public_trip_id,
330 )
331 session.add(host_request)
332 session.flush()
334 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
335 notify(
336 session,
337 user_id=host_request.recipient_user_id,
338 topic_action=NotificationTopicAction.host_request__create,
339 key=str(host_request.conversation_id),
340 data=notification_data_pb2.HostRequestCreate(
341 host_request=host_request_to_pb(host_request, session, recipient_context),
342 surfer=user_model_to_pb(host_request.initiator, session, recipient_context),
343 text=request.text,
344 ),
345 moderation_state_id=moderation_state.id,
346 )
348 host_requests_sent_counter.labels(user.gender, recipient.gender).inc()
349 sent_messages_counter.labels(user.gender, "host request send").inc()
350 account_age_on_host_request_create_histogram.labels(surfer_user.gender, host_user.gender).observe(
351 (now() - user.joined).total_seconds()
352 )
353 log_event(
354 context,
355 session,
356 "host_request.created",
357 {
358 "host_request_id": host_request.conversation_id,
359 "surfer_id": surfer_user.id,
360 "host_id": host_user.id,
361 "surfer_gender": surfer_user.gender,
362 "host_gender": host_user.gender,
363 "city": host_user.city,
364 "from_date": str(from_date),
365 "to_date": str(to_date),
366 "nights": (to_date - from_date).days,
367 },
368 )
370 return requests_pb2.CreateHostRequestRes(host_request_id=host_request.conversation_id)
372 def GetHostRequest(
373 self, request: requests_pb2.GetHostRequestReq, context: CouchersContext, session: Session
374 ) -> requests_pb2.HostRequest:
375 host_request = session.execute(
376 where_moderated_content_visible(
377 where_users_column_visible(
378 where_users_column_visible(
379 select(HostRequest),
380 context,
381 HostRequest.initiator_user_id,
382 ),
383 context,
384 HostRequest.recipient_user_id,
385 ),
386 context,
387 HostRequest,
388 is_list_operation=False,
389 )
390 .where(HostRequest.conversation_id == request.host_request_id)
391 .where(
392 or_(HostRequest.initiator_user_id == context.user_id, HostRequest.recipient_user_id == context.user_id)
393 )
394 ).scalar_one_or_none()
396 if not host_request:
397 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
399 return host_request_to_pb(host_request, session, context)
401 # TODO(#7722): remove after FE migrates to ListMessageThreads
402 def ListHostRequests(
403 self, request: requests_pb2.ListHostRequestsReq, context: CouchersContext, session: Session
404 ) -> requests_pb2.ListHostRequestsRes:
405 if request.only_sent and request.only_received: 405 ↛ 406line 405 didn't jump to line 406 because the condition on line 405 was never true
406 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "host_request_sent_or_received")
408 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
409 pagination = min(pagination, MAX_PAGE_SIZE)
411 # By outer joining messages on itself where the second id is bigger, only the highest IDs will have
412 # none as message_2.id. So just filter for these to get the highest messages only.
413 # See https://stackoverflow.com/a/27802817/6115336
414 message_2 = aliased(Message)
415 statement = where_moderated_content_visible(
416 where_users_column_visible(
417 where_users_column_visible(
418 select(
419 Message,
420 HostRequest,
421 Conversation,
422 unseen_host_request_message_count(context.user_id).label("unseen_message_count"),
423 )
424 .outerjoin(
425 message_2, and_(Message.conversation_id == message_2.conversation_id, Message.id < message_2.id)
426 )
427 .join(HostRequest, HostRequest.conversation_id == Message.conversation_id)
428 .join(Conversation, Conversation.id == HostRequest.conversation_id),
429 context,
430 HostRequest.initiator_user_id,
431 ),
432 context,
433 HostRequest.recipient_user_id,
434 ),
435 context,
436 HostRequest,
437 is_list_operation=True,
438 ).where(message_2.id == None)
440 sort_by_from_date = request.sort_by == requests_pb2.HOST_REQUEST_SORT_BY_FROM_DATE
442 if sort_by_from_date:
443 if request.page_token:
444 token_date_str, token_conv_id_str = request.page_token.split(":")
445 token_date = parse_date(token_date_str)
446 token_conv_id = int(token_conv_id_str)
447 statement = statement.where(
448 or_(
449 HostRequest.from_date > token_date,
450 and_(
451 HostRequest.from_date == token_date,
452 HostRequest.conversation_id > token_conv_id,
453 ),
454 )
455 )
456 else:
457 if request.page_token:
458 statement = statement.where(Message.id < int(request.page_token))
460 if request.only_sent:
461 statement = statement.where(HostRequest.initiator_user_id == context.user_id)
462 elif request.only_received:
463 statement = statement.where(HostRequest.recipient_user_id == context.user_id)
464 elif request.HasField("only_archived"):
465 statement = statement.where(
466 or_(
467 and_(
468 HostRequest.initiator_user_id == context.user_id,
469 HostRequest.is_initiator_archived == request.only_archived,
470 ),
471 and_(
472 HostRequest.recipient_user_id == context.user_id,
473 HostRequest.is_recipient_archived == request.only_archived,
474 ),
475 )
476 )
477 else:
478 statement = statement.where(
479 or_(HostRequest.recipient_user_id == context.user_id, HostRequest.initiator_user_id == context.user_id)
480 )
482 # TODO: I considered having the latest control message be the single source of truth for
483 # the HostRequest.status, but decided against it because of this filter.
484 # Another possibility is to filter in the python instead of SQL, but that's slower
485 if request.only_active:
486 statement = statement.where(
487 or_(
488 HostRequest.status == HostRequestStatus.pending,
489 HostRequest.status == HostRequestStatus.accepted,
490 HostRequest.status == HostRequestStatus.confirmed,
491 )
492 )
493 statement = statement.where(HostRequest.end_time >= func.now())
495 if request.status_in:
496 statement = statement.where(HostRequest.status.in_([api2hostrequeststatus[s] for s in request.status_in]))
498 if sort_by_from_date:
499 statement = statement.order_by(HostRequest.from_date.asc(), HostRequest.conversation_id.asc())
500 else:
501 statement = statement.order_by(Message.id.desc())
502 statement = statement.limit(pagination + 1)
503 results = session.execute(statement).all()
505 host_requests = []
506 for result in results[:pagination]:
507 lat, lng = get_coordinates(result.HostRequest.hosting_location)
508 host_requests.append(
509 requests_pb2.HostRequest(
510 host_request_id=result.HostRequest.conversation_id,
511 surfer_user_id=result.HostRequest.surfer_user_id,
512 host_user_id=result.HostRequest.host_user_id,
513 status=hostrequeststatus2api[result.HostRequest.status],
514 created=Timestamp_from_datetime(result.Conversation.created),
515 from_date=date_to_api(result.HostRequest.from_date),
516 to_date=date_to_api(result.HostRequest.to_date),
517 last_seen_message_id=(
518 result.HostRequest.initiator_last_seen_message_id
519 if context.user_id == result.HostRequest.initiator_user_id
520 else result.HostRequest.recipient_last_seen_message_id
521 ),
522 latest_message=message_to_pb(result.Message),
523 hosting_city=result.HostRequest.hosting_city,
524 hosting_lat=lat,
525 hosting_lng=lng,
526 hosting_radius=result.HostRequest.hosting_radius,
527 unseen_message_count=result.unseen_message_count,
528 )
529 )
531 no_more = len(results) <= pagination
533 if len(results) > pagination:
534 if sort_by_from_date:
535 last = results[pagination - 1]
536 next_page_token = f"{date_to_api(last.HostRequest.from_date)}:{last.HostRequest.conversation_id}"
537 else:
538 next_page_token = str(min(g.Message.id for g in results[:pagination]))
539 else:
540 next_page_token = None
542 return requests_pb2.ListHostRequestsRes(
543 next_page_token=next_page_token, no_more=no_more, host_requests=host_requests
544 )
546 def RespondHostRequest(
547 self, request: requests_pb2.RespondHostRequestReq, context: CouchersContext, session: Session
548 ) -> empty_pb2.Empty:
549 def count_host_response(other_user_id: int, response_type: str) -> None:
550 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
551 other_gender = session.execute(select(User.gender).where(User.id == other_user_id)).scalar_one()
552 host_request_responses_counter.labels(user_gender, other_gender, response_type).inc()
553 sent_messages_counter.labels(user_gender, "host request response").inc()
555 host_request = session.execute(
556 where_moderated_content_visible(
557 where_users_column_visible(
558 where_users_column_visible(
559 select(HostRequest),
560 context,
561 HostRequest.initiator_user_id,
562 ),
563 context,
564 HostRequest.recipient_user_id,
565 ),
566 context,
567 HostRequest,
568 is_list_operation=False,
569 ).where(HostRequest.conversation_id == request.host_request_id)
570 ).scalar_one_or_none()
572 if not host_request:
573 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
575 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id:
576 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
578 if request.status == messages_pb2.HOST_REQUEST_STATUS_PENDING:
579 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
581 if host_request.end_time < now(): 581 ↛ 582line 581 didn't jump to line 582 because the condition on line 581 was never true
582 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "host_request_in_past")
584 control_message = Message(
585 message_type=MessageType.host_request_status_changed,
586 conversation_id=host_request.conversation_id,
587 author_id=context.user_id,
588 )
590 if request.status == messages_pb2.HOST_REQUEST_STATUS_ACCEPTED:
591 # only host can accept
592 if context.user_id != host_request.recipient_user_id:
593 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "not_the_host")
594 # can't accept a cancelled or confirmed request (only reject), or already accepted
595 if ( 595 ↛ 600line 595 didn't jump to line 600 because the condition on line 595 was never true
596 host_request.status == HostRequestStatus.cancelled
597 or host_request.status == HostRequestStatus.confirmed
598 or host_request.status == HostRequestStatus.accepted
599 ):
600 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
601 _possibly_observe_first_response_time(session, host_request, context.user_id, "accepted")
602 control_message.host_request_status_target = HostRequestStatus.accepted
603 host_request.status = HostRequestStatus.accepted
604 session.flush()
606 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
607 notify(
608 session,
609 user_id=host_request.initiator_user_id,
610 topic_action=NotificationTopicAction.host_request__accept,
611 key=str(host_request.conversation_id),
612 data=notification_data_pb2.HostRequestAccept(
613 host_request=host_request_to_pb(host_request, session, recipient_context),
614 host=user_model_to_pb(host_request.recipient, session, recipient_context),
615 ),
616 moderation_state_id=host_request.moderation_state_id,
617 )
619 count_host_response(host_request.initiator_user_id, "accepted")
620 log_event(
621 context,
622 session,
623 "host_request.accepted",
624 {
625 "host_request_id": host_request.conversation_id,
626 "surfer_id": host_request.surfer_user_id,
627 "host_id": host_request.host_user_id,
628 "surfer_gender": host_request.surfer.gender,
629 "host_gender": host_request.host.gender,
630 "from_date": str(host_request.from_date),
631 "to_date": str(host_request.to_date),
632 "host_city": host_request.hosting_city,
633 },
634 )
636 if request.status == messages_pb2.HOST_REQUEST_STATUS_REJECTED:
637 # only host can reject
638 if context.user_id != host_request.recipient_user_id: 638 ↛ 639line 638 didn't jump to line 639 because the condition on line 638 was never true
639 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
640 # can't reject a cancelled or already rejected request
641 if host_request.status == HostRequestStatus.cancelled or host_request.status == HostRequestStatus.rejected: 641 ↛ 642line 641 didn't jump to line 642 because the condition on line 641 was never true
642 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
643 _possibly_observe_first_response_time(session, host_request, context.user_id, "rejected")
644 control_message.host_request_status_target = HostRequestStatus.rejected
645 host_request.status = HostRequestStatus.rejected
646 session.flush()
648 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
649 notify(
650 session,
651 user_id=host_request.initiator_user_id,
652 topic_action=NotificationTopicAction.host_request__reject,
653 key=str(host_request.conversation_id),
654 data=notification_data_pb2.HostRequestReject(
655 host_request=host_request_to_pb(host_request, session, recipient_context),
656 host=user_model_to_pb(host_request.recipient, session, recipient_context),
657 ),
658 moderation_state_id=host_request.moderation_state_id,
659 )
661 count_host_response(host_request.initiator_user_id, "rejected")
663 log_event(
664 context,
665 session,
666 "host_request.rejected",
667 {
668 "host_request_id": host_request.conversation_id,
669 "surfer_id": host_request.surfer_user_id,
670 "host_id": host_request.host_user_id,
671 "surfer_gender": host_request.surfer.gender,
672 "host_gender": host_request.host.gender,
673 "from_date": str(host_request.from_date),
674 "to_date": str(host_request.to_date),
675 "host_city": host_request.hosting_city,
676 },
677 )
679 if request.status == messages_pb2.HOST_REQUEST_STATUS_CONFIRMED:
680 # only surfer can confirm
681 if context.user_id != host_request.initiator_user_id:
682 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
683 # can only confirm an accepted request
684 if host_request.status != HostRequestStatus.accepted:
685 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
686 control_message.host_request_status_target = HostRequestStatus.confirmed
687 host_request.status = HostRequestStatus.confirmed
688 session.flush()
690 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
691 notify(
692 session,
693 user_id=host_request.recipient_user_id,
694 topic_action=NotificationTopicAction.host_request__confirm,
695 key=str(host_request.conversation_id),
696 data=notification_data_pb2.HostRequestConfirm(
697 host_request=host_request_to_pb(host_request, session, recipient_context),
698 surfer=user_model_to_pb(host_request.initiator, session, recipient_context),
699 ),
700 moderation_state_id=host_request.moderation_state_id,
701 )
703 count_host_response(host_request.recipient_user_id, "confirmed")
704 log_event(
705 context,
706 session,
707 "host_request.confirmed",
708 {
709 "host_request_id": host_request.conversation_id,
710 "surfer_id": host_request.surfer_user_id,
711 "host_id": host_request.host_user_id,
712 "surfer_gender": host_request.surfer.gender,
713 "host_gender": host_request.host.gender,
714 "from_date": str(host_request.from_date),
715 "to_date": str(host_request.to_date),
716 "host_city": host_request.hosting_city,
717 },
718 )
720 if request.status == messages_pb2.HOST_REQUEST_STATUS_CANCELLED:
721 # only surfer can cancel
722 if context.user_id != host_request.initiator_user_id:
723 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
724 # can't' cancel an already cancelled or rejected request
725 if host_request.status == HostRequestStatus.rejected or host_request.status == HostRequestStatus.cancelled: 725 ↛ 726line 725 didn't jump to line 726 because the condition on line 725 was never true
726 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
727 control_message.host_request_status_target = HostRequestStatus.cancelled
728 host_request.status = HostRequestStatus.cancelled
729 session.flush()
731 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
732 notify(
733 session,
734 user_id=host_request.recipient_user_id,
735 topic_action=NotificationTopicAction.host_request__cancel,
736 key=str(host_request.conversation_id),
737 data=notification_data_pb2.HostRequestCancel(
738 host_request=host_request_to_pb(host_request, session, recipient_context),
739 surfer=user_model_to_pb(host_request.initiator, session, recipient_context),
740 ),
741 moderation_state_id=host_request.moderation_state_id,
742 )
744 count_host_response(host_request.recipient_user_id, "cancelled")
745 log_event(
746 context,
747 session,
748 "host_request.cancelled",
749 {
750 "host_request_id": host_request.conversation_id,
751 "surfer_id": host_request.surfer_user_id,
752 "host_id": host_request.host_user_id,
753 "surfer_gender": host_request.surfer.gender,
754 "host_gender": host_request.host.gender,
755 "from_date": str(host_request.from_date),
756 "to_date": str(host_request.to_date),
757 "host_city": host_request.hosting_city,
758 },
759 )
761 session.add(control_message)
763 if request.text:
764 latest_message = Message(
765 conversation_id=host_request.conversation_id,
766 text=request.text,
767 author_id=context.user_id,
768 message_type=MessageType.text,
769 )
771 session.add(latest_message)
772 else:
773 latest_message = control_message
775 session.flush()
777 if host_request.initiator_user_id == context.user_id:
778 host_request.initiator_last_seen_message_id = latest_message.id
779 else:
780 host_request.recipient_last_seen_message_id = latest_message.id
781 session.commit()
783 return empty_pb2.Empty()
785 def GetHostRequestMessages(
786 self, request: requests_pb2.GetHostRequestMessagesReq, context: CouchersContext, session: Session
787 ) -> requests_pb2.GetHostRequestMessagesRes:
788 host_request = session.execute(
789 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False).where(
790 HostRequest.conversation_id == request.host_request_id
791 )
792 ).scalar_one_or_none()
794 if not host_request: 794 ↛ 795line 794 didn't jump to line 795 because the condition on line 794 was never true
795 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
797 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id: 797 ↛ 798line 797 didn't jump to line 798 because the condition on line 797 was never true
798 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
800 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
801 pagination = min(pagination, MAX_PAGE_SIZE)
803 messages = (
804 session.execute(
805 select(Message)
806 .where(Message.conversation_id == host_request.conversation_id)
807 .where(or_(Message.id < request.last_message_id, to_bool(request.last_message_id == 0)))
808 .order_by(Message.id.desc())
809 .limit(pagination + 1)
810 )
811 .scalars()
812 .all()
813 )
815 no_more = len(messages) <= pagination
817 last_message_id = min(m.id if m else 1 for m in messages[:pagination]) if len(messages) > 0 else 0
819 return requests_pb2.GetHostRequestMessagesRes(
820 last_message_id=last_message_id,
821 no_more=no_more,
822 messages=[message_to_pb(message) for message in messages[:pagination]],
823 )
825 def SendHostRequestMessage(
826 self, request: requests_pb2.SendHostRequestMessageReq, context: CouchersContext, session: Session
827 ) -> empty_pb2.Empty:
828 if request.text == "":
829 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
830 host_request = session.execute(
831 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False).where(
832 HostRequest.conversation_id == request.host_request_id
833 )
834 ).scalar_one_or_none()
836 if not host_request:
837 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
839 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id:
840 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
842 if host_request.recipient_user_id == context.user_id:
843 _possibly_observe_first_response_time(session, host_request, context.user_id, "message")
845 message = Message(
846 conversation_id=host_request.conversation_id,
847 author_id=context.user_id,
848 message_type=MessageType.text,
849 text=request.text,
850 )
852 session.add(message)
853 session.flush()
855 if host_request.initiator_user_id == context.user_id:
856 host_request.initiator_last_seen_message_id = message.id
858 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
859 notify(
860 session,
861 user_id=host_request.recipient_user_id,
862 topic_action=NotificationTopicAction.host_request__message,
863 key=str(host_request.conversation_id),
864 data=notification_data_pb2.HostRequestMessage(
865 host_request=host_request_to_pb(host_request, session, recipient_context),
866 user=user_model_to_pb(host_request.initiator, session, recipient_context),
867 text=request.text,
868 am_host=host_request.host_user_id == host_request.recipient_user_id,
869 ),
870 moderation_state_id=host_request.moderation_state_id,
871 )
873 else:
874 host_request.recipient_last_seen_message_id = message.id
876 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
877 notify(
878 session,
879 user_id=host_request.initiator_user_id,
880 topic_action=NotificationTopicAction.host_request__message,
881 key=str(host_request.conversation_id),
882 data=notification_data_pb2.HostRequestMessage(
883 host_request=host_request_to_pb(host_request, session, recipient_context),
884 user=user_model_to_pb(host_request.recipient, session, recipient_context),
885 text=request.text,
886 am_host=host_request.host_user_id == host_request.initiator_user_id,
887 ),
888 moderation_state_id=host_request.moderation_state_id,
889 )
891 session.commit()
893 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
894 sent_messages_counter.labels(user_gender, "host request").inc()
895 log_event(
896 context,
897 session,
898 "host_request.message_sent",
899 {
900 "host_request_id": host_request.conversation_id,
901 "surfer_id": host_request.surfer_user_id,
902 "host_id": host_request.host_user_id,
903 "role": "host" if context.user_id == host_request.host_user_id else "surfer",
904 "host_city": host_request.hosting_city,
905 },
906 )
908 return empty_pb2.Empty()
910 def GetHostRequestUpdates(
911 self, request: requests_pb2.GetHostRequestUpdatesReq, context: CouchersContext, session: Session
912 ) -> requests_pb2.GetHostRequestUpdatesRes:
913 if request.only_sent and request.only_received: 913 ↛ 914line 913 didn't jump to line 914 because the condition on line 913 was never true
914 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "host_request_sent_or_received")
916 if request.newest_message_id == 0:
917 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
919 if not session.execute(select(Message).where(Message.id == request.newest_message_id)).scalar_one_or_none(): 919 ↛ 920line 919 didn't jump to line 920 because the condition on line 919 was never true
920 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
922 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
923 pagination = min(pagination, MAX_PAGE_SIZE)
925 statement = where_moderated_content_visible(
926 select(
927 Message,
928 HostRequest.status.label("host_request_status"),
929 HostRequest.conversation_id.label("host_request_id"),
930 )
931 .join(HostRequest, HostRequest.conversation_id == Message.conversation_id)
932 .where(Message.id > request.newest_message_id),
933 context,
934 HostRequest,
935 is_list_operation=False,
936 )
938 if request.only_sent: 938 ↛ 939line 938 didn't jump to line 939 because the condition on line 938 was never true
939 statement = statement.where(HostRequest.initiator_user_id == context.user_id)
940 elif request.only_received: 940 ↛ 941line 940 didn't jump to line 941 because the condition on line 940 was never true
941 statement = statement.where(HostRequest.recipient_user_id == context.user_id)
942 else:
943 statement = statement.where(
944 or_(HostRequest.recipient_user_id == context.user_id, HostRequest.initiator_user_id == context.user_id)
945 )
947 statement = statement.order_by(Message.id.asc()).limit(pagination + 1)
948 res = session.execute(statement).all()
950 no_more = len(res) <= pagination
952 last_message_id = min(m.Message.id if m else 1 for m in res[:pagination]) if len(res) > 0 else 0 # TODO
954 return requests_pb2.GetHostRequestUpdatesRes(
955 no_more=no_more,
956 updates=[
957 requests_pb2.HostRequestUpdate(
958 host_request_id=result.host_request_id,
959 status=hostrequeststatus2api[result.host_request_status],
960 message=message_to_pb(result.Message),
961 )
962 for result in res[:pagination]
963 ],
964 )
966 def MarkLastSeenHostRequest(
967 self, request: requests_pb2.MarkLastSeenHostRequestReq, context: CouchersContext, session: Session
968 ) -> empty_pb2.Empty:
969 host_request = session.execute(
970 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False).where(
971 HostRequest.conversation_id == request.host_request_id
972 )
973 ).scalar_one_or_none()
975 if not host_request: 975 ↛ 976line 975 didn't jump to line 976 because the condition on line 975 was never true
976 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
978 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id: 978 ↛ 979line 978 didn't jump to line 979 because the condition on line 978 was never true
979 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
981 if host_request.initiator_user_id == context.user_id: 981 ↛ 982line 981 didn't jump to line 982 because the condition on line 981 was never true
982 if not host_request.initiator_last_seen_message_id <= request.last_seen_message_id:
983 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages")
984 host_request.initiator_last_seen_message_id = request.last_seen_message_id
985 else:
986 if not host_request.recipient_last_seen_message_id <= request.last_seen_message_id:
987 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages")
988 host_request.recipient_last_seen_message_id = request.last_seen_message_id
990 mark_notifications_seen(
991 session,
992 user_id=context.user_id,
993 topic_actions_and_keys=[
994 (HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS, [str(host_request.conversation_id)]),
995 ],
996 )
998 session.commit()
999 return empty_pb2.Empty()
1001 def SetHostRequestArchiveStatus(
1002 self, request: requests_pb2.SetHostRequestArchiveStatusReq, context: CouchersContext, session: Session
1003 ) -> requests_pb2.SetHostRequestArchiveStatusRes:
1004 host_request = session.execute(
1005 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False)
1006 .where(HostRequest.conversation_id == request.host_request_id)
1007 .where(
1008 or_(HostRequest.initiator_user_id == context.user_id, HostRequest.recipient_user_id == context.user_id)
1009 )
1010 ).scalar_one_or_none()
1012 if not host_request: 1012 ↛ 1013line 1012 didn't jump to line 1013 because the condition on line 1012 was never true
1013 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1015 if context.user_id == host_request.initiator_user_id: 1015 ↛ 1018line 1015 didn't jump to line 1018 because the condition on line 1015 was always true
1016 host_request.is_initiator_archived = request.is_archived
1017 else:
1018 host_request.is_recipient_archived = request.is_archived
1020 return requests_pb2.SetHostRequestArchiveStatusRes(
1021 host_request_id=host_request.conversation_id,
1022 is_archived=request.is_archived,
1023 )
1025 def GetResponseRate(
1026 self, request: requests_pb2.GetResponseRateReq, context: CouchersContext, session: Session
1027 ) -> requests_pb2.GetResponseRateRes:
1028 user_res = session.execute(
1029 select(User.id, UserResponseRate)
1030 .outerjoin(UserResponseRate, UserResponseRate.user_id == User.id)
1031 .where(users_visible(context, User))
1032 .where(User.id == request.user_id)
1033 ).one_or_none()
1035 # if user doesn't exist, return None
1036 if not user_res:
1037 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1039 user, response_rates = user_res
1040 return requests_pb2.GetResponseRateRes(**response_rate_to_pb(response_rates)) # type: ignore[arg-type]
1042 def SendHostRequestFeedback(
1043 self, request: requests_pb2.SendHostRequestFeedbackReq, context: CouchersContext, session: Session
1044 ) -> empty_pb2.Empty:
1045 host_request = session.execute(
1046 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False)
1047 .where(HostRequest.conversation_id == request.host_request_id)
1048 .where(HostRequest.recipient_user_id == context.user_id)
1049 ).scalar_one_or_none()
1051 if not host_request:
1052 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1054 feedback = session.execute(
1055 select(HostRequestFeedback)
1056 .where(HostRequestFeedback.host_request_id == host_request.conversation_id)
1057 .where(HostRequestFeedback.from_user_id == context.user_id)
1058 ).scalar_one_or_none()
1060 if feedback:
1061 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_left_host_request_feedback")
1063 session.add(
1064 HostRequestFeedback(
1065 host_request_id=host_request.conversation_id,
1066 from_user_id=host_request.recipient_user_id,
1067 to_user_id=host_request.initiator_user_id,
1068 request_quality=hostrequestquality2sql.get(request.host_request_quality),
1069 decline_reason=request.decline_reason,
1070 )
1071 )
1072 quality = hostrequestquality2sql.get(request.host_request_quality)
1073 log_event(
1074 context,
1075 session,
1076 "host_request.feedback_submitted",
1077 {
1078 "host_request_id": host_request.conversation_id,
1079 "surfer_id": host_request.surfer_user_id,
1080 "host_id": host_request.host_user_id,
1081 "request_quality": quality.name if quality else None,
1082 "has_decline_reason": bool(request.decline_reason),
1083 "host_city": host_request.hosting_city,
1084 },
1085 )
1087 return empty_pb2.Empty()