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