Coverage for app/backend/src/couchers/servicers/requests.py: 92%
343 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +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.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
267 # Offers on public trips are deduplicated per trip further down instead
268 if public_trip_id is None:
269 recent_request = session.execute(
270 select(HostRequest.conversation_id)
271 .join(Conversation, HostRequest.conversation_id == Conversation.id)
272 .where(HostRequest.initiator_user_id == context.user_id)
273 .where(HostRequest.recipient_user_id == recipient.id)
274 .where(HostRequest.public_trip_id == None)
275 .where(Conversation.created >= now() - timedelta(hours=HOST_REQUEST_DUPLICATE_WINDOW_HOURS))
276 # overlapping nights, so back-to-back stays are still allowed
277 .where(HostRequest.from_date < to_date)
278 .where(HostRequest.to_date > from_date)
279 .limit(1)
280 ).scalar_one_or_none()
281 if recent_request is not None:
282 context.abort_with_error_code(
283 grpc.StatusCode.FAILED_PRECONDITION,
284 "duplicate_host_request",
285 substitutions={"count": HOST_REQUEST_DUPLICATE_WINDOW_HOURS},
286 )
288 if public_trip_id is not None:
289 public_trip = session.execute(
290 where_moderated_content_visible(select(PublicTrip), context, PublicTrip, is_list_operation=False).where(
291 PublicTrip.id == public_trip_id
292 )
293 ).scalar_one_or_none()
294 if not public_trip:
295 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "public_trip_not_found")
296 # The trip's traveler must be the recipient of this host request (role reversal)
297 if public_trip.user_id != recipient.id:
298 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "public_trip_user_mismatch")
299 # Trip must still be active
300 if public_trip.status != PublicTripStatus.searching_for_host:
301 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "public_trip_not_active")
302 # Offered dates must fall within the trip's window (host can shorten, not extend)
303 if from_date < public_trip.from_date or to_date > public_trip.to_date:
304 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "public_trip_dates_out_of_range")
305 # Enforce same_gender_only restriction (community moderators bypass)
306 if (
307 public_trip.same_gender_only
308 and not can_moderate_node(session, context.user_id, public_trip.node_id)
309 and user.gender != recipient.gender
310 ):
311 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "public_trip_same_gender_only")
312 # Prevent duplicate offers on the same trip
313 existing_offer = session.execute(
314 select(HostRequest)
315 .where(HostRequest.public_trip_id == public_trip_id)
316 .where(HostRequest.initiator_user_id == context.user_id)
317 ).scalar_one_or_none()
318 if existing_offer:
319 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "duplicate_host_request_for_trip")
321 conversation = Conversation()
322 session.add(conversation)
323 session.flush()
325 session.add(
326 Message(
327 conversation_id=conversation.id,
328 author_id=context.user_id,
329 message_type=MessageType.chat_created,
330 )
331 )
333 message = Message(
334 conversation_id=conversation.id,
335 author_id=context.user_id,
336 text=request.text,
337 message_type=MessageType.text,
338 )
339 session.add(message)
340 session.flush()
342 # Create moderation state for UMS (starts as SHADOWED)
343 moderation_state = create_moderation(
344 session=session,
345 object_type=ModerationObjectType.host_request,
346 object_id=conversation.id,
347 creator_user_id=context.user_id,
348 )
350 host_request = HostRequest(
351 conversation_id=conversation.id,
352 initiator_user_id=context.user_id,
353 recipient_user_id=recipient.id,
354 moderation_state_id=moderation_state.id,
355 from_date=from_date,
356 to_date=to_date,
357 status=HostRequestStatus.pending,
358 initiator_last_seen_message_id=message.id,
359 # TODO: tz
360 # timezone=recipient.timezone,
361 hosting_city=recipient.city,
362 hosting_location=recipient.geom,
363 hosting_radius=recipient.geom_radius,
364 public_trip_id=public_trip_id,
365 )
366 session.add(host_request)
367 session.flush()
369 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
370 notify(
371 session,
372 user_id=host_request.recipient_user_id,
373 topic_action=NotificationTopicAction.host_request__create,
374 key=str(host_request.conversation_id),
375 data=notification_data_pb2.HostRequestCreate(
376 host_request=host_request_to_pb(host_request, session, recipient_context),
377 surfer=user_model_to_pb(host_request.initiator, session, recipient_context),
378 text=request.text,
379 ),
380 moderation_state_id=moderation_state.id,
381 )
383 host_requests_sent_counter.labels(user.gender, recipient.gender).inc()
384 sent_messages_counter.labels(user.gender, "host request send").inc()
385 account_age_on_host_request_create_histogram.labels(user.gender, recipient.gender).observe(
386 (now() - user.joined).total_seconds()
387 )
388 log_event(
389 context,
390 session,
391 "host_request.created",
392 {
393 "host_request_id": host_request.conversation_id,
394 "host_id": recipient.id,
395 "surfer_gender": user.gender,
396 "host_gender": recipient.gender,
397 "city": recipient.city,
398 "from_date": str(from_date),
399 "to_date": str(to_date),
400 "nights": (to_date - from_date).days,
401 },
402 )
404 return requests_pb2.CreateHostRequestRes(host_request_id=host_request.conversation_id)
406 def GetHostRequest(
407 self, request: requests_pb2.GetHostRequestReq, context: CouchersContext, session: Session
408 ) -> requests_pb2.HostRequest:
409 host_request = session.execute(
410 where_moderated_content_visible(
411 where_users_column_visible(
412 where_users_column_visible(
413 select(HostRequest),
414 context,
415 HostRequest.initiator_user_id,
416 ),
417 context,
418 HostRequest.recipient_user_id,
419 ),
420 context,
421 HostRequest,
422 is_list_operation=False,
423 )
424 .where(HostRequest.conversation_id == request.host_request_id)
425 .where(
426 or_(HostRequest.initiator_user_id == context.user_id, HostRequest.recipient_user_id == context.user_id)
427 )
428 ).scalar_one_or_none()
430 if not host_request:
431 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
433 return host_request_to_pb(host_request, session, context)
435 def ListHostRequests(
436 self, request: requests_pb2.ListHostRequestsReq, context: CouchersContext, session: Session
437 ) -> requests_pb2.ListHostRequestsRes:
438 if request.only_sent and request.only_received: 438 ↛ 439line 438 didn't jump to line 439 because the condition on line 438 was never true
439 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "host_request_sent_or_received")
441 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
442 pagination = min(pagination, MAX_PAGE_SIZE)
444 # By outer joining messages on itself where the second id is bigger, only the highest IDs will have
445 # none as message_2.id. So just filter for these to get the highest messages only.
446 # See https://stackoverflow.com/a/27802817/6115336
447 message_2 = aliased(Message)
448 statement = where_moderated_content_visible(
449 where_users_column_visible(
450 where_users_column_visible(
451 select(Message, HostRequest, Conversation)
452 .outerjoin(
453 message_2, and_(Message.conversation_id == message_2.conversation_id, Message.id < message_2.id)
454 )
455 .join(HostRequest, HostRequest.conversation_id == Message.conversation_id)
456 .join(Conversation, Conversation.id == HostRequest.conversation_id),
457 context,
458 HostRequest.initiator_user_id,
459 ),
460 context,
461 HostRequest.recipient_user_id,
462 ),
463 context,
464 HostRequest,
465 is_list_operation=True,
466 ).where(message_2.id == None)
468 sort_by_from_date = request.sort_by == requests_pb2.HOST_REQUEST_SORT_BY_FROM_DATE
470 if sort_by_from_date:
471 if request.page_token:
472 token_date_str, token_conv_id_str = request.page_token.split(":")
473 token_date = parse_date(token_date_str)
474 token_conv_id = int(token_conv_id_str)
475 statement = statement.where(
476 or_(
477 HostRequest.from_date > token_date,
478 and_(
479 HostRequest.from_date == token_date,
480 HostRequest.conversation_id > token_conv_id,
481 ),
482 )
483 )
484 else:
485 if request.page_token:
486 statement = statement.where(Message.id < int(request.page_token))
488 if request.only_sent:
489 statement = statement.where(HostRequest.initiator_user_id == context.user_id)
490 elif request.only_received:
491 statement = statement.where(HostRequest.recipient_user_id == context.user_id)
492 elif request.HasField("only_archived"):
493 statement = statement.where(
494 or_(
495 and_(
496 HostRequest.initiator_user_id == context.user_id,
497 HostRequest.is_initiator_archived == request.only_archived,
498 ),
499 and_(
500 HostRequest.recipient_user_id == context.user_id,
501 HostRequest.is_recipient_archived == request.only_archived,
502 ),
503 )
504 )
505 else:
506 statement = statement.where(
507 or_(HostRequest.recipient_user_id == context.user_id, HostRequest.initiator_user_id == context.user_id)
508 )
510 # TODO: I considered having the latest control message be the single source of truth for
511 # the HostRequest.status, but decided against it because of this filter.
512 # Another possibility is to filter in the python instead of SQL, but that's slower
513 if request.only_active:
514 statement = statement.where(
515 or_(
516 HostRequest.status == HostRequestStatus.pending,
517 HostRequest.status == HostRequestStatus.accepted,
518 HostRequest.status == HostRequestStatus.confirmed,
519 )
520 )
521 statement = statement.where(HostRequest.end_time >= func.now())
523 if request.status_in:
524 statement = statement.where(HostRequest.status.in_([api2hostrequeststatus[s] for s in request.status_in]))
526 if sort_by_from_date:
527 statement = statement.order_by(HostRequest.from_date.asc(), HostRequest.conversation_id.asc())
528 else:
529 statement = statement.order_by(Message.id.desc())
530 statement = statement.limit(pagination + 1)
531 results = session.execute(statement).all()
533 host_requests = []
534 for result in results[:pagination]:
535 lat, lng = get_coordinates(result.HostRequest.hosting_location)
536 host_requests.append(
537 requests_pb2.HostRequest(
538 host_request_id=result.HostRequest.conversation_id,
539 surfer_user_id=result.HostRequest.initiator_user_id,
540 host_user_id=result.HostRequest.recipient_user_id,
541 status=hostrequeststatus2api[result.HostRequest.status],
542 created=Timestamp_from_datetime(result.Conversation.created),
543 from_date=date_to_api(result.HostRequest.from_date),
544 to_date=date_to_api(result.HostRequest.to_date),
545 last_seen_message_id=(
546 result.HostRequest.initiator_last_seen_message_id
547 if context.user_id == result.HostRequest.initiator_user_id
548 else result.HostRequest.recipient_last_seen_message_id
549 ),
550 latest_message=message_to_pb(result.Message),
551 hosting_city=result.HostRequest.hosting_city,
552 hosting_lat=lat,
553 hosting_lng=lng,
554 hosting_radius=result.HostRequest.hosting_radius,
555 )
556 )
558 no_more = len(results) <= pagination
560 if len(results) > pagination:
561 if sort_by_from_date:
562 last = results[pagination - 1]
563 next_page_token = f"{date_to_api(last.HostRequest.from_date)}:{last.HostRequest.conversation_id}"
564 else:
565 next_page_token = str(min(g.Message.id for g in results[:pagination]))
566 else:
567 next_page_token = None
569 return requests_pb2.ListHostRequestsRes(
570 next_page_token=next_page_token, no_more=no_more, host_requests=host_requests
571 )
573 def RespondHostRequest(
574 self, request: requests_pb2.RespondHostRequestReq, context: CouchersContext, session: Session
575 ) -> empty_pb2.Empty:
576 def count_host_response(other_user_id: int, response_type: str) -> None:
577 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
578 other_gender = session.execute(select(User.gender).where(User.id == other_user_id)).scalar_one()
579 host_request_responses_counter.labels(user_gender, other_gender, response_type).inc()
580 sent_messages_counter.labels(user_gender, "host request response").inc()
582 host_request = session.execute(
583 where_moderated_content_visible(
584 where_users_column_visible(
585 where_users_column_visible(
586 select(HostRequest),
587 context,
588 HostRequest.initiator_user_id,
589 ),
590 context,
591 HostRequest.recipient_user_id,
592 ),
593 context,
594 HostRequest,
595 is_list_operation=False,
596 ).where(HostRequest.conversation_id == request.host_request_id)
597 ).scalar_one_or_none()
599 if not host_request:
600 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
602 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id:
603 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
605 if request.status == messages_pb2.HOST_REQUEST_STATUS_PENDING:
606 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
608 if host_request.end_time < now(): 608 ↛ 609line 608 didn't jump to line 609 because the condition on line 608 was never true
609 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "host_request_in_past")
611 control_message = Message(
612 message_type=MessageType.host_request_status_changed,
613 conversation_id=host_request.conversation_id,
614 author_id=context.user_id,
615 )
617 if request.status == messages_pb2.HOST_REQUEST_STATUS_ACCEPTED:
618 # only host can accept
619 if context.user_id != host_request.recipient_user_id:
620 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "not_the_host")
621 # can't accept a cancelled or confirmed request (only reject), or already accepted
622 if ( 622 ↛ 627line 622 didn't jump to line 627 because the condition on line 622 was never true
623 host_request.status == HostRequestStatus.cancelled
624 or host_request.status == HostRequestStatus.confirmed
625 or host_request.status == HostRequestStatus.accepted
626 ):
627 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
628 _possibly_observe_first_response_time(session, host_request, context.user_id, "accepted")
629 control_message.host_request_status_target = HostRequestStatus.accepted
630 host_request.status = HostRequestStatus.accepted
631 session.flush()
633 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
634 notify(
635 session,
636 user_id=host_request.initiator_user_id,
637 topic_action=NotificationTopicAction.host_request__accept,
638 key=str(host_request.conversation_id),
639 data=notification_data_pb2.HostRequestAccept(
640 host_request=host_request_to_pb(host_request, session, recipient_context),
641 host=user_model_to_pb(host_request.recipient, session, recipient_context),
642 ),
643 moderation_state_id=host_request.moderation_state_id,
644 )
646 count_host_response(host_request.initiator_user_id, "accepted")
647 log_event(
648 context,
649 session,
650 "host_request.accepted",
651 {
652 "host_request_id": host_request.conversation_id,
653 "surfer_id": host_request.initiator_user_id,
654 "host_id": host_request.recipient_user_id,
655 "surfer_gender": host_request.initiator.gender,
656 "host_gender": host_request.recipient.gender,
657 "from_date": str(host_request.from_date),
658 "to_date": str(host_request.to_date),
659 "host_city": host_request.hosting_city,
660 },
661 )
663 if request.status == messages_pb2.HOST_REQUEST_STATUS_REJECTED:
664 # only host can reject
665 if context.user_id != host_request.recipient_user_id: 665 ↛ 666line 665 didn't jump to line 666 because the condition on line 665 was never true
666 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
667 # can't reject a cancelled or already rejected request
668 if host_request.status == HostRequestStatus.cancelled or host_request.status == HostRequestStatus.rejected: 668 ↛ 669line 668 didn't jump to line 669 because the condition on line 668 was never true
669 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
670 _possibly_observe_first_response_time(session, host_request, context.user_id, "rejected")
671 control_message.host_request_status_target = HostRequestStatus.rejected
672 host_request.status = HostRequestStatus.rejected
673 session.flush()
675 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
676 notify(
677 session,
678 user_id=host_request.initiator_user_id,
679 topic_action=NotificationTopicAction.host_request__reject,
680 key=str(host_request.conversation_id),
681 data=notification_data_pb2.HostRequestReject(
682 host_request=host_request_to_pb(host_request, session, recipient_context),
683 host=user_model_to_pb(host_request.recipient, session, recipient_context),
684 ),
685 moderation_state_id=host_request.moderation_state_id,
686 )
688 count_host_response(host_request.initiator_user_id, "rejected")
690 log_event(
691 context,
692 session,
693 "host_request.rejected",
694 {
695 "host_request_id": host_request.conversation_id,
696 "surfer_id": host_request.initiator_user_id,
697 "host_id": host_request.recipient_user_id,
698 "surfer_gender": host_request.initiator.gender,
699 "host_gender": host_request.recipient.gender,
700 "from_date": str(host_request.from_date),
701 "to_date": str(host_request.to_date),
702 "host_city": host_request.hosting_city,
703 },
704 )
706 if request.status == messages_pb2.HOST_REQUEST_STATUS_CONFIRMED:
707 # only surfer can confirm
708 if context.user_id != host_request.initiator_user_id:
709 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
710 # can only confirm an accepted request
711 if host_request.status != HostRequestStatus.accepted:
712 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
713 control_message.host_request_status_target = HostRequestStatus.confirmed
714 host_request.status = HostRequestStatus.confirmed
715 session.flush()
717 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
718 notify(
719 session,
720 user_id=host_request.recipient_user_id,
721 topic_action=NotificationTopicAction.host_request__confirm,
722 key=str(host_request.conversation_id),
723 data=notification_data_pb2.HostRequestConfirm(
724 host_request=host_request_to_pb(host_request, session, recipient_context),
725 surfer=user_model_to_pb(host_request.initiator, session, recipient_context),
726 ),
727 moderation_state_id=host_request.moderation_state_id,
728 )
730 count_host_response(host_request.recipient_user_id, "confirmed")
731 log_event(
732 context,
733 session,
734 "host_request.confirmed",
735 {
736 "host_request_id": host_request.conversation_id,
737 "surfer_id": host_request.initiator_user_id,
738 "host_id": host_request.recipient_user_id,
739 "surfer_gender": host_request.initiator.gender,
740 "host_gender": host_request.recipient.gender,
741 "from_date": str(host_request.from_date),
742 "to_date": str(host_request.to_date),
743 "host_city": host_request.hosting_city,
744 },
745 )
747 if request.status == messages_pb2.HOST_REQUEST_STATUS_CANCELLED:
748 # only surfer can cancel
749 if context.user_id != host_request.initiator_user_id:
750 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
751 # can't' cancel an already cancelled or rejected request
752 if host_request.status == HostRequestStatus.rejected or host_request.status == HostRequestStatus.cancelled: 752 ↛ 753line 752 didn't jump to line 753 because the condition on line 752 was never true
753 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
754 control_message.host_request_status_target = HostRequestStatus.cancelled
755 host_request.status = HostRequestStatus.cancelled
756 session.flush()
758 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
759 notify(
760 session,
761 user_id=host_request.recipient_user_id,
762 topic_action=NotificationTopicAction.host_request__cancel,
763 key=str(host_request.conversation_id),
764 data=notification_data_pb2.HostRequestCancel(
765 host_request=host_request_to_pb(host_request, session, recipient_context),
766 surfer=user_model_to_pb(host_request.initiator, session, recipient_context),
767 ),
768 moderation_state_id=host_request.moderation_state_id,
769 )
771 count_host_response(host_request.recipient_user_id, "cancelled")
772 log_event(
773 context,
774 session,
775 "host_request.cancelled",
776 {
777 "host_request_id": host_request.conversation_id,
778 "surfer_id": host_request.initiator_user_id,
779 "host_id": host_request.recipient_user_id,
780 "surfer_gender": host_request.initiator.gender,
781 "host_gender": host_request.recipient.gender,
782 "from_date": str(host_request.from_date),
783 "to_date": str(host_request.to_date),
784 "host_city": host_request.hosting_city,
785 },
786 )
788 session.add(control_message)
790 if request.text:
791 latest_message = Message(
792 conversation_id=host_request.conversation_id,
793 text=request.text,
794 author_id=context.user_id,
795 message_type=MessageType.text,
796 )
798 session.add(latest_message)
799 else:
800 latest_message = control_message
802 session.flush()
804 if host_request.initiator_user_id == context.user_id:
805 host_request.initiator_last_seen_message_id = latest_message.id
806 else:
807 host_request.recipient_last_seen_message_id = latest_message.id
808 session.commit()
810 return empty_pb2.Empty()
812 def GetHostRequestMessages(
813 self, request: requests_pb2.GetHostRequestMessagesReq, context: CouchersContext, session: Session
814 ) -> requests_pb2.GetHostRequestMessagesRes:
815 host_request = session.execute(
816 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False).where(
817 HostRequest.conversation_id == request.host_request_id
818 )
819 ).scalar_one_or_none()
821 if not host_request: 821 ↛ 822line 821 didn't jump to line 822 because the condition on line 821 was never true
822 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
824 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id: 824 ↛ 825line 824 didn't jump to line 825 because the condition on line 824 was never true
825 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
827 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
828 pagination = min(pagination, MAX_PAGE_SIZE)
830 messages = (
831 session.execute(
832 select(Message)
833 .where(Message.conversation_id == host_request.conversation_id)
834 .where(or_(Message.id < request.last_message_id, to_bool(request.last_message_id == 0)))
835 .order_by(Message.id.desc())
836 .limit(pagination + 1)
837 )
838 .scalars()
839 .all()
840 )
842 no_more = len(messages) <= pagination
844 last_message_id = min(m.id if m else 1 for m in messages[:pagination]) if len(messages) > 0 else 0
846 return requests_pb2.GetHostRequestMessagesRes(
847 last_message_id=last_message_id,
848 no_more=no_more,
849 messages=[message_to_pb(message) for message in messages[:pagination]],
850 )
852 def SendHostRequestMessage(
853 self, request: requests_pb2.SendHostRequestMessageReq, context: CouchersContext, session: Session
854 ) -> empty_pb2.Empty:
855 if request.text == "":
856 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
857 host_request = session.execute(
858 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False).where(
859 HostRequest.conversation_id == request.host_request_id
860 )
861 ).scalar_one_or_none()
863 if not host_request:
864 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
866 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id:
867 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
869 if host_request.recipient_user_id == context.user_id:
870 _possibly_observe_first_response_time(session, host_request, context.user_id, "message")
872 message = Message(
873 conversation_id=host_request.conversation_id,
874 author_id=context.user_id,
875 message_type=MessageType.text,
876 text=request.text,
877 )
879 session.add(message)
880 session.flush()
882 if host_request.initiator_user_id == context.user_id:
883 host_request.initiator_last_seen_message_id = message.id
885 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
886 notify(
887 session,
888 user_id=host_request.recipient_user_id,
889 topic_action=NotificationTopicAction.host_request__message,
890 key=str(host_request.conversation_id),
891 data=notification_data_pb2.HostRequestMessage(
892 host_request=host_request_to_pb(host_request, session, recipient_context),
893 user=user_model_to_pb(host_request.initiator, session, recipient_context),
894 text=request.text,
895 am_host=True,
896 ),
897 moderation_state_id=host_request.moderation_state_id,
898 )
900 else:
901 host_request.recipient_last_seen_message_id = message.id
903 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
904 notify(
905 session,
906 user_id=host_request.initiator_user_id,
907 topic_action=NotificationTopicAction.host_request__message,
908 key=str(host_request.conversation_id),
909 data=notification_data_pb2.HostRequestMessage(
910 host_request=host_request_to_pb(host_request, session, recipient_context),
911 user=user_model_to_pb(host_request.recipient, session, recipient_context),
912 text=request.text,
913 am_host=False,
914 ),
915 moderation_state_id=host_request.moderation_state_id,
916 )
918 session.commit()
920 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
921 sent_messages_counter.labels(user_gender, "host request").inc()
922 log_event(
923 context,
924 session,
925 "host_request.message_sent",
926 {
927 "host_request_id": host_request.conversation_id,
928 "surfer_id": host_request.initiator_user_id,
929 "host_id": host_request.recipient_user_id,
930 "role": "host" if context.user_id == host_request.recipient_user_id else "surfer",
931 "host_city": host_request.hosting_city,
932 },
933 )
935 return empty_pb2.Empty()
937 def GetHostRequestUpdates(
938 self, request: requests_pb2.GetHostRequestUpdatesReq, context: CouchersContext, session: Session
939 ) -> requests_pb2.GetHostRequestUpdatesRes:
940 if request.only_sent and request.only_received: 940 ↛ 941line 940 didn't jump to line 941 because the condition on line 940 was never true
941 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "host_request_sent_or_received")
943 if request.newest_message_id == 0:
944 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
946 if not session.execute(select(Message).where(Message.id == request.newest_message_id)).scalar_one_or_none(): 946 ↛ 947line 946 didn't jump to line 947 because the condition on line 946 was never true
947 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
949 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
950 pagination = min(pagination, MAX_PAGE_SIZE)
952 statement = where_moderated_content_visible(
953 select(
954 Message,
955 HostRequest.status.label("host_request_status"),
956 HostRequest.conversation_id.label("host_request_id"),
957 )
958 .join(HostRequest, HostRequest.conversation_id == Message.conversation_id)
959 .where(Message.id > request.newest_message_id),
960 context,
961 HostRequest,
962 is_list_operation=False,
963 )
965 if request.only_sent: 965 ↛ 966line 965 didn't jump to line 966 because the condition on line 965 was never true
966 statement = statement.where(HostRequest.initiator_user_id == context.user_id)
967 elif request.only_received: 967 ↛ 968line 967 didn't jump to line 968 because the condition on line 967 was never true
968 statement = statement.where(HostRequest.recipient_user_id == context.user_id)
969 else:
970 statement = statement.where(
971 or_(HostRequest.recipient_user_id == context.user_id, HostRequest.initiator_user_id == context.user_id)
972 )
974 statement = statement.order_by(Message.id.asc()).limit(pagination + 1)
975 res = session.execute(statement).all()
977 no_more = len(res) <= pagination
979 last_message_id = min(m.Message.id if m else 1 for m in res[:pagination]) if len(res) > 0 else 0 # TODO
981 return requests_pb2.GetHostRequestUpdatesRes(
982 no_more=no_more,
983 updates=[
984 requests_pb2.HostRequestUpdate(
985 host_request_id=result.host_request_id,
986 status=hostrequeststatus2api[result.host_request_status],
987 message=message_to_pb(result.Message),
988 )
989 for result in res[:pagination]
990 ],
991 )
993 def MarkLastSeenHostRequest(
994 self, request: requests_pb2.MarkLastSeenHostRequestReq, context: CouchersContext, session: Session
995 ) -> empty_pb2.Empty:
996 host_request = session.execute(
997 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False).where(
998 HostRequest.conversation_id == request.host_request_id
999 )
1000 ).scalar_one_or_none()
1002 if not host_request: 1002 ↛ 1003line 1002 didn't jump to line 1003 because the condition on line 1002 was never true
1003 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1005 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id: 1005 ↛ 1006line 1005 didn't jump to line 1006 because the condition on line 1005 was never true
1006 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1008 if host_request.initiator_user_id == context.user_id: 1008 ↛ 1009line 1008 didn't jump to line 1009 because the condition on line 1008 was never true
1009 if not host_request.initiator_last_seen_message_id <= request.last_seen_message_id:
1010 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages")
1011 host_request.initiator_last_seen_message_id = request.last_seen_message_id
1012 else:
1013 if not host_request.recipient_last_seen_message_id <= request.last_seen_message_id:
1014 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages")
1015 host_request.recipient_last_seen_message_id = request.last_seen_message_id
1017 mark_notifications_seen(
1018 session,
1019 user_id=context.user_id,
1020 key=str(host_request.conversation_id),
1021 topic_actions=[
1022 NotificationTopicAction.host_request__create,
1023 NotificationTopicAction.host_request__accept,
1024 NotificationTopicAction.host_request__reject,
1025 NotificationTopicAction.host_request__confirm,
1026 NotificationTopicAction.host_request__cancel,
1027 NotificationTopicAction.host_request__message,
1028 NotificationTopicAction.host_request__missed_messages,
1029 NotificationTopicAction.host_request__reminder,
1030 ],
1031 )
1033 session.commit()
1034 return empty_pb2.Empty()
1036 def SetHostRequestArchiveStatus(
1037 self, request: requests_pb2.SetHostRequestArchiveStatusReq, context: CouchersContext, session: Session
1038 ) -> requests_pb2.SetHostRequestArchiveStatusRes:
1039 host_request = session.execute(
1040 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False)
1041 .where(HostRequest.conversation_id == request.host_request_id)
1042 .where(
1043 or_(HostRequest.initiator_user_id == context.user_id, HostRequest.recipient_user_id == context.user_id)
1044 )
1045 ).scalar_one_or_none()
1047 if not host_request: 1047 ↛ 1048line 1047 didn't jump to line 1048 because the condition on line 1047 was never true
1048 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1050 if context.user_id == host_request.initiator_user_id: 1050 ↛ 1053line 1050 didn't jump to line 1053 because the condition on line 1050 was always true
1051 host_request.is_initiator_archived = request.is_archived
1052 else:
1053 host_request.is_recipient_archived = request.is_archived
1055 return requests_pb2.SetHostRequestArchiveStatusRes(
1056 host_request_id=host_request.conversation_id,
1057 is_archived=request.is_archived,
1058 )
1060 def GetResponseRate(
1061 self, request: requests_pb2.GetResponseRateReq, context: CouchersContext, session: Session
1062 ) -> requests_pb2.GetResponseRateRes:
1063 user_res = session.execute(
1064 select(User.id, UserResponseRate)
1065 .outerjoin(UserResponseRate, UserResponseRate.user_id == User.id)
1066 .where(users_visible(context, User))
1067 .where(User.id == request.user_id)
1068 ).one_or_none()
1070 # if user doesn't exist, return None
1071 if not user_res:
1072 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1074 user, response_rates = user_res
1075 return requests_pb2.GetResponseRateRes(**response_rate_to_pb(response_rates)) # type: ignore[arg-type]
1077 def SendHostRequestFeedback(
1078 self, request: requests_pb2.SendHostRequestFeedbackReq, context: CouchersContext, session: Session
1079 ) -> empty_pb2.Empty:
1080 host_request = session.execute(
1081 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False)
1082 .where(HostRequest.conversation_id == request.host_request_id)
1083 .where(HostRequest.recipient_user_id == context.user_id)
1084 ).scalar_one_or_none()
1086 if not host_request:
1087 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1089 feedback = session.execute(
1090 select(HostRequestFeedback)
1091 .where(HostRequestFeedback.host_request_id == host_request.conversation_id)
1092 .where(HostRequestFeedback.from_user_id == context.user_id)
1093 ).scalar_one_or_none()
1095 if feedback:
1096 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_left_host_request_feedback")
1098 session.add(
1099 HostRequestFeedback(
1100 host_request_id=host_request.conversation_id,
1101 from_user_id=host_request.recipient_user_id,
1102 to_user_id=host_request.initiator_user_id,
1103 request_quality=hostrequestquality2sql.get(request.host_request_quality),
1104 decline_reason=request.decline_reason,
1105 )
1106 )
1107 quality = hostrequestquality2sql.get(request.host_request_quality)
1108 log_event(
1109 context,
1110 session,
1111 "host_request.feedback_submitted",
1112 {
1113 "host_request_id": host_request.conversation_id,
1114 "surfer_id": host_request.initiator_user_id,
1115 "host_id": host_request.recipient_user_id,
1116 "request_quality": quality.name if quality else None,
1117 "has_decline_reason": bool(request.decline_reason),
1118 "host_city": host_request.hosting_city,
1119 },
1120 )
1122 return empty_pb2.Empty()