Coverage for app/backend/src/couchers/servicers/requests.py: 93%
364 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +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 (
11 HOST_REQUEST_DUPLICATE_WINDOW_HOURS,
12 HOST_REQUEST_MIN_LENGTH_UTF16,
13 PUBLIC_TRIP_OFFER_MIN_LENGTH_UTF16,
14)
15from couchers.context import CouchersContext, make_notification_user_context
16from couchers.db import can_moderate_node
17from couchers.event_log import log_event
18from couchers.helpers.completed_profile import has_completed_profile
19from couchers.helpers.host_requests import HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS, unseen_host_request_message_count
20from couchers.helpers.messages import api2hostrequeststatus, hostrequeststatus2api, message_to_pb
21from couchers.materialized_views import UserResponseRate
22from couchers.metrics import (
23 account_age_on_host_request_create_histogram,
24 host_request_first_response_histogram,
25 host_request_responses_counter,
26 host_requests_sent_counter,
27 public_trip_offers_sent_counter,
28 sent_messages_counter,
29)
30from couchers.models import (
31 Conversation,
32 HostRequest,
33 HostRequestFeedback,
34 HostRequestQuality,
35 HostRequestStatus,
36 Message,
37 MessageType,
38 ModerationObjectType,
39 RateLimitAction,
40 StayProposalKind,
41 User,
42)
43from couchers.models.notifications import NotificationTopicAction
44from couchers.models.public_trips import PublicTrip, PublicTripStatus
45from couchers.moderation.utils import create_moderation
46from couchers.notifications.notify import mark_notifications_seen, notify
47from couchers.proto import (
48 messages_pb2,
49 notification_data_pb2,
50 requests_pb2,
51 requests_pb2_grpc,
52)
53from couchers.rate_limits.check import process_rate_limits_and_check_abort
54from couchers.rate_limits.definitions import RATE_LIMIT_HOURS
55from couchers.servicers.api import response_rate_to_pb, user_model_to_pb
56from couchers.sql import to_bool, users_visible, where_moderated_content_visible, where_users_column_visible
57from couchers.utils import (
58 Timestamp_from_datetime,
59 date_id_from_page_token,
60 date_id_to_page_token,
61 date_to_api,
62 get_coordinates,
63 now,
64 parse_date,
65 today_in_timezone,
66)
68logger = logging.getLogger(__name__)
70DEFAULT_PAGINATION_LENGTH = 10
71MAX_PAGE_SIZE = 50
74hostrequestquality2sql = {
75 requests_pb2.HOST_REQUEST_QUALITY_UNSPECIFIED: HostRequestQuality.high_quality,
76 requests_pb2.HOST_REQUEST_QUALITY_LOW: HostRequestQuality.okay_quality,
77 requests_pb2.HOST_REQUEST_QUALITY_OKAY: HostRequestQuality.low_quality,
78}
81def host_request_to_pb(
82 host_request: HostRequest, session: Session, context: CouchersContext
83) -> requests_pb2.HostRequest:
84 initial_message = session.execute(
85 select(Message)
86 .where(Message.conversation_id == host_request.conversation_id)
87 .order_by(Message.id.asc())
88 .limit(1)
89 ).scalar_one()
91 latest_message = session.execute(
92 select(Message)
93 .where(Message.conversation_id == host_request.conversation_id)
94 .order_by(Message.id.desc())
95 .limit(1)
96 ).scalar_one()
98 unseen_message_count = session.execute(
99 select(unseen_host_request_message_count(context.user_id)).where(
100 HostRequest.conversation_id == host_request.conversation_id
101 )
102 ).scalar_one()
104 lat, lng = get_coordinates(host_request.hosting_location)
106 # public-trip offers aren't asked for feedback: their recipient is the traveller, not a host
107 need_feedback = False
108 if (
109 context.user_id == host_request.recipient_user_id
110 and host_request.status == HostRequestStatus.rejected
111 and host_request.proposal_kind == StayProposalKind.request
112 ):
113 need_feedback = not session.execute(
114 select(
115 exists().where(
116 HostRequestFeedback.from_user_id == context.user_id,
117 HostRequestFeedback.host_request_id == host_request.conversation_id,
118 )
119 )
120 ).scalar_one()
122 return requests_pb2.HostRequest(
123 host_request_id=host_request.conversation_id,
124 surfer_user_id=host_request.surfer_user_id,
125 host_user_id=host_request.host_user_id,
126 status=hostrequeststatus2api[host_request.status],
127 created=Timestamp_from_datetime(initial_message.time),
128 from_date=date_to_api(host_request.from_date),
129 to_date=date_to_api(host_request.to_date),
130 timezone=host_request.timezone,
131 last_seen_message_id=(
132 host_request.initiator_last_seen_message_id
133 if context.user_id == host_request.initiator_user_id
134 else host_request.recipient_last_seen_message_id
135 ),
136 latest_message=message_to_pb(latest_message),
137 hosting_city=host_request.hosting_city,
138 hosting_lat=lat,
139 hosting_lng=lng,
140 hosting_radius=host_request.hosting_radius,
141 need_host_request_feedback=need_feedback,
142 is_archived=(
143 host_request.is_recipient_archived
144 if context.user_id == host_request.recipient_user_id
145 else host_request.is_initiator_archived
146 ),
147 public_trip_id=host_request.public_trip_id,
148 unseen_message_count=unseen_message_count,
149 )
152def _possibly_observe_first_response_time(
153 session: Session, host_request: HostRequest, user_id: int, response_type: str
154) -> None:
155 # if this is the first response then there's nothing by this user yet
156 assert host_request.recipient_user_id == user_id
158 # on a public trip offer the recipient is the traveller, so this isn't a host's response time
159 if host_request.proposal_kind == StayProposalKind.invitation:
160 return
162 number_messages_by_host = session.execute(
163 select(func.count())
164 .where(Message.conversation_id == host_request.conversation_id)
165 .where(Message.author_id == user_id)
166 ).scalar_one_or_none()
168 if number_messages_by_host == 0:
169 host_gender = session.execute(select(User.gender).where(User.id == host_request.host_user_id)).scalar_one()
170 surfer_gender = session.execute(select(User.gender).where(User.id == host_request.surfer_user_id)).scalar_one()
171 host_request_first_response_histogram.labels(host_gender, surfer_gender, response_type).observe(
172 (now() - host_request.conversation.created).total_seconds()
173 )
176def _validate_request_text(text: str, is_public_trip_offer: bool, context: CouchersContext) -> None:
177 min_length, error_code = (
178 (PUBLIC_TRIP_OFFER_MIN_LENGTH_UTF16, "public_trip_offer_too_short")
179 if is_public_trip_offer
180 else (HOST_REQUEST_MIN_LENGTH_UTF16, "host_request_too_short2")
181 )
182 # Python's len(str) does not match Javascript's string.length.
183 # e.g. len("é") == 2 but "é".length == 1.
184 # To match the frontend's validation, measure the string in utf16 code units.
185 text_length_utf16 = len(text.encode("utf-16-le")) // 2 # utf-16-le does not include a prefix BOM code unit.
186 if text_length_utf16 < min_length:
187 context.abort_with_error_code(
188 grpc.StatusCode.INVALID_ARGUMENT,
189 error_code,
190 substitutions={"count": min_length},
191 )
194class Requests(requests_pb2_grpc.RequestsServicer):
195 def CreateHostRequest(
196 self, request: requests_pb2.CreateHostRequestReq, context: CouchersContext, session: Session
197 ) -> requests_pb2.CreateHostRequestRes:
198 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
199 if not has_completed_profile(session, user):
200 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_request")
202 if request.host_user_id == context.user_id:
203 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_request_self")
205 # just to check recipient exists and is visible
206 recipient = session.execute(
207 select(User).where(users_visible(context, User)).where(User.id == request.host_user_id)
208 ).scalar_one_or_none()
209 if not recipient:
210 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
212 from_date = parse_date(request.from_date)
213 to_date = parse_date(request.to_date)
215 if not from_date or not to_date:
216 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_date")
218 # If this is an offer in response to a public trip, it's validated further down
219 proposal_kind = StayProposalKind.invitation if request.HasField("public_trip_id") else StayProposalKind.request
220 public_trip_id = request.public_trip_id if proposal_kind == StayProposalKind.invitation else None
222 # an offer on a public trip reverses the roles: the caller is the host and the recipient is the traveller
223 surfer_user, host_user = (user, recipient) if proposal_kind == StayProposalKind.request else (recipient, user)
225 # the dates are in the timezone at the host's location
226 today = today_in_timezone(host_user.timezone or "Etc/UTC")
228 # request starts from the past
229 if from_date < today:
230 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "date_from_before_today")
232 # from_date is not >= to_date
233 if from_date >= to_date:
234 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "date_from_after_to")
236 # No need to check today > to_date
238 if from_date - today > timedelta(days=365):
239 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "date_from_after_one_year")
241 if to_date - from_date > timedelta(days=365):
242 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "date_to_after_one_year")
244 _validate_request_text(request.text, proposal_kind == StayProposalKind.invitation, context)
246 # Check if user has been sending host requests excessively
247 if process_rate_limits_and_check_abort(
248 session=session, user_id=context.user_id, action=RateLimitAction.host_request
249 ):
250 context.abort_with_error_code(
251 grpc.StatusCode.RESOURCE_EXHAUSTED,
252 "host_request_rate_limit2",
253 substitutions={"count": RATE_LIMIT_HOURS},
254 )
256 if proposal_kind == StayProposalKind.request:
257 recent_request = session.execute(
258 select(HostRequest.conversation_id)
259 .join(Conversation, HostRequest.conversation_id == Conversation.id)
260 .where(HostRequest.initiator_user_id == context.user_id)
261 .where(HostRequest.recipient_user_id == recipient.id)
262 .where(HostRequest.proposal_kind == StayProposalKind.request)
263 .where(Conversation.created >= now() - timedelta(hours=HOST_REQUEST_DUPLICATE_WINDOW_HOURS))
264 # overlapping nights, so back-to-back stays are still allowed
265 .where(HostRequest.from_date < to_date)
266 .where(HostRequest.to_date > from_date)
267 .limit(1)
268 ).scalar_one_or_none()
269 if recent_request is not None:
270 context.abort_with_error_code(
271 grpc.StatusCode.FAILED_PRECONDITION,
272 "duplicate_host_request",
273 substitutions={"count": HOST_REQUEST_DUPLICATE_WINDOW_HOURS},
274 )
275 else:
276 public_trip = session.execute(
277 where_moderated_content_visible(select(PublicTrip), context, PublicTrip, is_list_operation=False).where(
278 PublicTrip.id == public_trip_id
279 )
280 ).scalar_one_or_none()
281 if not public_trip:
282 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "public_trip_not_found")
283 # The trip's traveler must be the recipient of this host request (role reversal)
284 if public_trip.user_id != recipient.id:
285 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "public_trip_user_mismatch")
286 # Trip must still be active
287 if public_trip.status != PublicTripStatus.searching_for_host:
288 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "public_trip_not_active")
289 # Offered dates must fall within the trip's window (host can shorten, not extend)
290 if from_date < public_trip.from_date or to_date > public_trip.to_date:
291 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "public_trip_dates_out_of_range")
292 # Enforce same_gender_only restriction (community moderators bypass)
293 if (
294 public_trip.same_gender_only
295 and not can_moderate_node(session, context.user_id, public_trip.node_id)
296 and user.gender != recipient.gender
297 ):
298 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "public_trip_same_gender_only")
299 # Prevent duplicate offers on the same trip. A withdrawn offer doesn't count, so a host
300 # can offer again after withdrawing, e.g. to renegotiate dates.
301 existing_offer = session.execute(
302 select(HostRequest.conversation_id)
303 .where(HostRequest.public_trip_id == public_trip_id)
304 .where(HostRequest.initiator_user_id == context.user_id)
305 .where(HostRequest.status != HostRequestStatus.cancelled)
306 .limit(1)
307 ).scalar_one_or_none()
308 if existing_offer is not None:
309 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "duplicate_host_request_for_trip")
311 conversation = Conversation()
312 session.add(conversation)
313 session.flush()
315 session.add(
316 Message(
317 conversation_id=conversation.id,
318 author_id=context.user_id,
319 message_type=MessageType.chat_created,
320 )
321 )
323 message = Message(
324 conversation_id=conversation.id,
325 author_id=context.user_id,
326 text=request.text,
327 message_type=MessageType.text,
328 )
329 session.add(message)
330 session.flush()
332 # Create moderation state for UMS (starts as SHADOWED)
333 moderation_state = create_moderation(
334 session=session,
335 object_type=ModerationObjectType.host_request,
336 object_id=conversation.id,
337 creator_user_id=context.user_id,
338 )
340 host_request = HostRequest(
341 conversation_id=conversation.id,
342 initiator_user_id=context.user_id,
343 recipient_user_id=recipient.id,
344 moderation_state_id=moderation_state.id,
345 from_date=from_date,
346 to_date=to_date,
347 status=HostRequestStatus.pending,
348 initiator_last_seen_message_id=message.id,
349 hosting_city=host_user.city,
350 hosting_location=host_user.geom,
351 hosting_radius=host_user.geom_radius,
352 proposal_kind=proposal_kind,
353 public_trip_id=public_trip_id,
354 )
355 session.add(host_request)
356 session.flush()
358 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
359 notify(
360 session,
361 user_id=host_request.recipient_user_id,
362 topic_action=NotificationTopicAction.host_request__create,
363 key=str(host_request.conversation_id),
364 data=notification_data_pb2.HostRequestCreate(
365 host_request=host_request_to_pb(host_request, session, recipient_context),
366 initiator=user_model_to_pb(host_request.initiator, session, recipient_context),
367 text=request.text,
368 ),
369 moderation_state_id=moderation_state.id,
370 )
372 # the host request metrics assume the sender is the surfer, which is backwards on a public trip offer
373 if proposal_kind == StayProposalKind.request:
374 host_requests_sent_counter.labels(user.gender, recipient.gender).inc()
375 account_age_on_host_request_create_histogram.labels(surfer_user.gender, host_user.gender).observe(
376 (now() - user.joined).total_seconds()
377 )
378 else:
379 public_trip_offers_sent_counter.labels(user.gender, recipient.gender).inc()
380 sent_messages_counter.labels(user.gender, "host request send").inc()
381 log_event(
382 context,
383 session,
384 "host_request.created",
385 {
386 "host_request_id": host_request.conversation_id,
387 "surfer_id": surfer_user.id,
388 "host_id": host_user.id,
389 "surfer_gender": surfer_user.gender,
390 "host_gender": host_user.gender,
391 "city": host_user.city,
392 "from_date": str(from_date),
393 "to_date": str(to_date),
394 "nights": (to_date - from_date).days,
395 },
396 )
398 return requests_pb2.CreateHostRequestRes(host_request_id=host_request.conversation_id)
400 def GetHostRequest(
401 self, request: requests_pb2.GetHostRequestReq, context: CouchersContext, session: Session
402 ) -> requests_pb2.HostRequest:
403 host_request = session.execute(
404 where_moderated_content_visible(
405 where_users_column_visible(
406 where_users_column_visible(
407 select(HostRequest),
408 context,
409 HostRequest.initiator_user_id,
410 ),
411 context,
412 HostRequest.recipient_user_id,
413 ),
414 context,
415 HostRequest,
416 is_list_operation=False,
417 )
418 .where(HostRequest.conversation_id == request.host_request_id)
419 .where(
420 or_(HostRequest.initiator_user_id == context.user_id, HostRequest.recipient_user_id == context.user_id)
421 )
422 ).scalar_one_or_none()
424 if not host_request:
425 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
427 return host_request_to_pb(host_request, session, context)
429 # TODO(#7722): remove after FE migrates to ListMessageThreads
430 def ListHostRequests(
431 self, request: requests_pb2.ListHostRequestsReq, context: CouchersContext, session: Session
432 ) -> requests_pb2.ListHostRequestsRes:
433 if request.only_sent and request.only_received: 433 ↛ 434line 433 didn't jump to line 434 because the condition on line 433 was never true
434 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "host_request_sent_or_received")
436 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
437 pagination = min(pagination, MAX_PAGE_SIZE)
439 # By outer joining messages on itself where the second id is bigger, only the highest IDs will have
440 # none as message_2.id. So just filter for these to get the highest messages only.
441 # See https://stackoverflow.com/a/27802817/6115336
442 message_2 = aliased(Message)
443 statement = where_moderated_content_visible(
444 where_users_column_visible(
445 where_users_column_visible(
446 select(
447 Message,
448 HostRequest,
449 Conversation,
450 unseen_host_request_message_count(context.user_id).label("unseen_message_count"),
451 )
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 # filter by stay role, so a public-trip offer (started by the host) lands on the right side
489 if request.only_sent:
490 statement = statement.where(HostRequest.surfer_user_id == context.user_id)
491 elif request.only_received:
492 statement = statement.where(HostRequest.host_user_id == context.user_id)
493 elif request.HasField("only_archived"):
494 statement = statement.where(
495 or_(
496 and_(
497 HostRequest.initiator_user_id == context.user_id,
498 HostRequest.is_initiator_archived == request.only_archived,
499 ),
500 and_(
501 HostRequest.recipient_user_id == context.user_id,
502 HostRequest.is_recipient_archived == request.only_archived,
503 ),
504 )
505 )
506 else:
507 statement = statement.where(
508 or_(HostRequest.recipient_user_id == context.user_id, HostRequest.initiator_user_id == context.user_id)
509 )
511 # TODO: I considered having the latest control message be the single source of truth for
512 # the HostRequest.status, but decided against it because of this filter.
513 # Another possibility is to filter in the python instead of SQL, but that's slower
514 if request.only_active:
515 statement = statement.where(
516 or_(
517 HostRequest.status == HostRequestStatus.pending,
518 HostRequest.status == HostRequestStatus.accepted,
519 HostRequest.status == HostRequestStatus.confirmed,
520 )
521 )
522 statement = statement.where(HostRequest.end_time >= func.now())
524 if request.status_in:
525 statement = statement.where(HostRequest.status.in_([api2hostrequeststatus[s] for s in request.status_in]))
527 if sort_by_from_date:
528 statement = statement.order_by(HostRequest.from_date.asc(), HostRequest.conversation_id.asc())
529 else:
530 statement = statement.order_by(Message.id.desc())
531 statement = statement.limit(pagination + 1)
532 results = session.execute(statement).all()
534 host_requests = []
535 for result in results[:pagination]:
536 lat, lng = get_coordinates(result.HostRequest.hosting_location)
537 host_requests.append(
538 requests_pb2.HostRequest(
539 host_request_id=result.HostRequest.conversation_id,
540 surfer_user_id=result.HostRequest.surfer_user_id,
541 host_user_id=result.HostRequest.host_user_id,
542 status=hostrequeststatus2api[result.HostRequest.status],
543 created=Timestamp_from_datetime(result.Conversation.created),
544 from_date=date_to_api(result.HostRequest.from_date),
545 to_date=date_to_api(result.HostRequest.to_date),
546 timezone=result.HostRequest.timezone,
547 last_seen_message_id=(
548 result.HostRequest.initiator_last_seen_message_id
549 if context.user_id == result.HostRequest.initiator_user_id
550 else result.HostRequest.recipient_last_seen_message_id
551 ),
552 latest_message=message_to_pb(result.Message),
553 hosting_city=result.HostRequest.hosting_city,
554 hosting_lat=lat,
555 hosting_lng=lng,
556 hosting_radius=result.HostRequest.hosting_radius,
557 public_trip_id=result.HostRequest.public_trip_id,
558 unseen_message_count=result.unseen_message_count,
559 )
560 )
562 no_more = len(results) <= pagination
564 if len(results) > pagination:
565 if sort_by_from_date:
566 last = results[pagination - 1]
567 next_page_token = f"{date_to_api(last.HostRequest.from_date)}:{last.HostRequest.conversation_id}"
568 else:
569 next_page_token = str(min(g.Message.id for g in results[:pagination]))
570 else:
571 next_page_token = None
573 return requests_pb2.ListHostRequestsRes(
574 next_page_token=next_page_token, no_more=no_more, host_requests=host_requests
575 )
577 def ListStays(
578 self, request: requests_pb2.ListStaysReq, context: CouchersContext, session: Session
579 ) -> requests_pb2.ListStaysRes:
580 if request.role == requests_pb2.STAY_ROLE_HOSTING:
581 role_column = HostRequest.host_user_id
582 elif request.role == requests_pb2.STAY_ROLE_SURFING:
583 role_column = HostRequest.surfer_user_id
584 else:
585 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_stay_role")
587 page_size = min(MAX_PAGE_SIZE, request.page_size or DEFAULT_PAGINATION_LENGTH)
589 statement = where_moderated_content_visible(
590 where_users_column_visible(
591 where_users_column_visible(select(HostRequest), context, HostRequest.initiator_user_id),
592 context,
593 HostRequest.recipient_user_id,
594 ),
595 context,
596 HostRequest,
597 is_list_operation=True,
598 ).where(
599 role_column == context.user_id,
600 HostRequest.status.in_([HostRequestStatus.accepted, HostRequestStatus.confirmed]),
601 HostRequest.end_time >= func.now(),
602 )
604 if request.page_token:
605 token_date, token_id = date_id_from_page_token(request.page_token)
606 statement = statement.where(
607 or_(
608 HostRequest.from_date > token_date,
609 and_(HostRequest.from_date == token_date, HostRequest.conversation_id > token_id),
610 )
611 )
613 host_requests = (
614 session.execute(statement.order_by(HostRequest.from_date, HostRequest.conversation_id).limit(page_size + 1))
615 .scalars()
616 .all()
617 )
619 next_page_token = ""
620 if len(host_requests) > page_size:
621 last = host_requests[page_size - 1]
622 next_page_token = date_id_to_page_token(last.from_date, last.conversation_id)
624 return requests_pb2.ListStaysRes(
625 host_requests=[
626 host_request_to_pb(host_request, session, context) for host_request in host_requests[:page_size]
627 ],
628 next_page_token=next_page_token,
629 )
631 def RespondHostRequest(
632 self, request: requests_pb2.RespondHostRequestReq, context: CouchersContext, session: Session
633 ) -> empty_pb2.Empty:
634 def count_host_response(host_request: HostRequest, other_user_id: int, response_type: str) -> None:
635 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
636 other_gender = session.execute(select(User.gender).where(User.id == other_user_id)).scalar_one()
637 # the response types assume the recipient is the host, which is backwards on a public trip offer
638 if host_request.proposal_kind == StayProposalKind.request:
639 host_request_responses_counter.labels(user_gender, other_gender, response_type).inc()
640 sent_messages_counter.labels(user_gender, "host request response").inc()
642 host_request = session.execute(
643 where_moderated_content_visible(
644 where_users_column_visible(
645 where_users_column_visible(
646 select(HostRequest),
647 context,
648 HostRequest.initiator_user_id,
649 ),
650 context,
651 HostRequest.recipient_user_id,
652 ),
653 context,
654 HostRequest,
655 is_list_operation=False,
656 ).where(HostRequest.conversation_id == request.host_request_id)
657 ).scalar_one_or_none()
659 if not host_request:
660 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
662 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id:
663 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
665 if request.status == messages_pb2.HOST_REQUEST_STATUS_PENDING:
666 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
668 if host_request.end_time < now(): 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.INVALID_ARGUMENT, "host_request_in_past")
671 control_message = Message(
672 message_type=MessageType.host_request_status_changed,
673 conversation_id=host_request.conversation_id,
674 author_id=context.user_id,
675 )
677 if request.status == messages_pb2.HOST_REQUEST_STATUS_ACCEPTED:
678 # only host can accept
679 if context.user_id != host_request.recipient_user_id:
680 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "not_the_host")
681 # can't accept a cancelled or confirmed request (only reject), or already accepted
682 if ( 682 ↛ 687line 682 didn't jump to line 687 because the condition on line 682 was never true
683 host_request.status == HostRequestStatus.cancelled
684 or host_request.status == HostRequestStatus.confirmed
685 or host_request.status == HostRequestStatus.accepted
686 ):
687 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
688 _possibly_observe_first_response_time(session, host_request, context.user_id, "accepted")
689 control_message.host_request_status_target = HostRequestStatus.accepted
690 host_request.status = HostRequestStatus.accepted
691 session.flush()
693 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
694 notify(
695 session,
696 user_id=host_request.initiator_user_id,
697 topic_action=NotificationTopicAction.host_request__accept,
698 key=str(host_request.conversation_id),
699 data=notification_data_pb2.HostRequestAccept(
700 host_request=host_request_to_pb(host_request, session, recipient_context),
701 recipient=user_model_to_pb(host_request.recipient, session, recipient_context),
702 ),
703 moderation_state_id=host_request.moderation_state_id,
704 )
706 count_host_response(host_request, host_request.initiator_user_id, "accepted")
707 log_event(
708 context,
709 session,
710 "host_request.accepted",
711 {
712 "host_request_id": host_request.conversation_id,
713 "surfer_id": host_request.surfer_user_id,
714 "host_id": host_request.host_user_id,
715 "surfer_gender": host_request.surfer.gender,
716 "host_gender": host_request.host.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_REJECTED:
724 # only host can reject
725 if context.user_id != host_request.recipient_user_id: 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 # can't reject a cancelled or already rejected request
728 if host_request.status == HostRequestStatus.cancelled or host_request.status == HostRequestStatus.rejected: 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 _possibly_observe_first_response_time(session, host_request, context.user_id, "rejected")
731 control_message.host_request_status_target = HostRequestStatus.rejected
732 host_request.status = HostRequestStatus.rejected
733 session.flush()
735 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
736 notify(
737 session,
738 user_id=host_request.initiator_user_id,
739 topic_action=NotificationTopicAction.host_request__reject,
740 key=str(host_request.conversation_id),
741 data=notification_data_pb2.HostRequestReject(
742 host_request=host_request_to_pb(host_request, session, recipient_context),
743 recipient=user_model_to_pb(host_request.recipient, session, recipient_context),
744 ),
745 moderation_state_id=host_request.moderation_state_id,
746 )
748 count_host_response(host_request, host_request.initiator_user_id, "rejected")
750 log_event(
751 context,
752 session,
753 "host_request.rejected",
754 {
755 "host_request_id": host_request.conversation_id,
756 "surfer_id": host_request.surfer_user_id,
757 "host_id": host_request.host_user_id,
758 "surfer_gender": host_request.surfer.gender,
759 "host_gender": host_request.host.gender,
760 "from_date": str(host_request.from_date),
761 "to_date": str(host_request.to_date),
762 "host_city": host_request.hosting_city,
763 },
764 )
766 if request.status == messages_pb2.HOST_REQUEST_STATUS_CONFIRMED:
767 # only surfer can confirm
768 if context.user_id != host_request.initiator_user_id:
769 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
770 # can only confirm an accepted request
771 if host_request.status != HostRequestStatus.accepted:
772 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
773 control_message.host_request_status_target = HostRequestStatus.confirmed
774 host_request.status = HostRequestStatus.confirmed
775 session.flush()
777 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
778 notify(
779 session,
780 user_id=host_request.recipient_user_id,
781 topic_action=NotificationTopicAction.host_request__confirm,
782 key=str(host_request.conversation_id),
783 data=notification_data_pb2.HostRequestConfirm(
784 host_request=host_request_to_pb(host_request, session, recipient_context),
785 initiator=user_model_to_pb(host_request.initiator, session, recipient_context),
786 ),
787 moderation_state_id=host_request.moderation_state_id,
788 )
790 count_host_response(host_request, host_request.recipient_user_id, "confirmed")
791 log_event(
792 context,
793 session,
794 "host_request.confirmed",
795 {
796 "host_request_id": host_request.conversation_id,
797 "surfer_id": host_request.surfer_user_id,
798 "host_id": host_request.host_user_id,
799 "surfer_gender": host_request.surfer.gender,
800 "host_gender": host_request.host.gender,
801 "from_date": str(host_request.from_date),
802 "to_date": str(host_request.to_date),
803 "host_city": host_request.hosting_city,
804 },
805 )
807 if request.status == messages_pb2.HOST_REQUEST_STATUS_CANCELLED:
808 # only surfer can cancel
809 if context.user_id != host_request.initiator_user_id:
810 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
811 # can't' cancel an already cancelled or rejected request
812 if host_request.status == HostRequestStatus.rejected or host_request.status == HostRequestStatus.cancelled: 812 ↛ 813line 812 didn't jump to line 813 because the condition on line 812 was never true
813 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "invalid_host_request_status")
814 control_message.host_request_status_target = HostRequestStatus.cancelled
815 host_request.status = HostRequestStatus.cancelled
816 session.flush()
818 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
819 notify(
820 session,
821 user_id=host_request.recipient_user_id,
822 topic_action=NotificationTopicAction.host_request__cancel,
823 key=str(host_request.conversation_id),
824 data=notification_data_pb2.HostRequestCancel(
825 host_request=host_request_to_pb(host_request, session, recipient_context),
826 initiator=user_model_to_pb(host_request.initiator, session, recipient_context),
827 ),
828 moderation_state_id=host_request.moderation_state_id,
829 )
831 count_host_response(host_request, host_request.recipient_user_id, "cancelled")
832 log_event(
833 context,
834 session,
835 "host_request.cancelled",
836 {
837 "host_request_id": host_request.conversation_id,
838 "surfer_id": host_request.surfer_user_id,
839 "host_id": host_request.host_user_id,
840 "surfer_gender": host_request.surfer.gender,
841 "host_gender": host_request.host.gender,
842 "from_date": str(host_request.from_date),
843 "to_date": str(host_request.to_date),
844 "host_city": host_request.hosting_city,
845 },
846 )
848 session.add(control_message)
850 if request.text:
851 latest_message = Message(
852 conversation_id=host_request.conversation_id,
853 text=request.text,
854 author_id=context.user_id,
855 message_type=MessageType.text,
856 )
858 session.add(latest_message)
859 else:
860 latest_message = control_message
862 session.flush()
864 if host_request.initiator_user_id == context.user_id:
865 host_request.initiator_last_seen_message_id = latest_message.id
866 else:
867 host_request.recipient_last_seen_message_id = latest_message.id
868 session.commit()
870 return empty_pb2.Empty()
872 def GetHostRequestMessages(
873 self, request: requests_pb2.GetHostRequestMessagesReq, context: CouchersContext, session: Session
874 ) -> requests_pb2.GetHostRequestMessagesRes:
875 host_request = session.execute(
876 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False).where(
877 HostRequest.conversation_id == request.host_request_id
878 )
879 ).scalar_one_or_none()
881 if not host_request: 881 ↛ 882line 881 didn't jump to line 882 because the condition on line 881 was never true
882 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
884 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id: 884 ↛ 885line 884 didn't jump to line 885 because the condition on line 884 was never true
885 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
887 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
888 pagination = min(pagination, MAX_PAGE_SIZE)
890 messages = (
891 session.execute(
892 select(Message)
893 .where(Message.conversation_id == host_request.conversation_id)
894 .where(or_(Message.id < request.last_message_id, to_bool(request.last_message_id == 0)))
895 .order_by(Message.id.desc())
896 .limit(pagination + 1)
897 )
898 .scalars()
899 .all()
900 )
902 no_more = len(messages) <= pagination
904 last_message_id = min(m.id if m else 1 for m in messages[:pagination]) if len(messages) > 0 else 0
906 return requests_pb2.GetHostRequestMessagesRes(
907 last_message_id=last_message_id,
908 no_more=no_more,
909 messages=[message_to_pb(message) for message in messages[:pagination]],
910 )
912 def SendHostRequestMessage(
913 self, request: requests_pb2.SendHostRequestMessageReq, context: CouchersContext, session: Session
914 ) -> empty_pb2.Empty:
915 if request.text == "":
916 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
917 host_request = session.execute(
918 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False).where(
919 HostRequest.conversation_id == request.host_request_id
920 )
921 ).scalar_one_or_none()
923 if not host_request:
924 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
926 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id:
927 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
929 if host_request.recipient_user_id == context.user_id:
930 _possibly_observe_first_response_time(session, host_request, context.user_id, "message")
932 message = Message(
933 conversation_id=host_request.conversation_id,
934 author_id=context.user_id,
935 message_type=MessageType.text,
936 text=request.text,
937 )
939 session.add(message)
940 session.flush()
942 if host_request.initiator_user_id == context.user_id:
943 host_request.initiator_last_seen_message_id = message.id
945 recipient_context = make_notification_user_context(user_id=host_request.recipient_user_id)
946 notify(
947 session,
948 user_id=host_request.recipient_user_id,
949 topic_action=NotificationTopicAction.host_request__message,
950 key=str(host_request.conversation_id),
951 data=notification_data_pb2.HostRequestMessage(
952 host_request=host_request_to_pb(host_request, session, recipient_context),
953 counterparty=user_model_to_pb(host_request.initiator, session, recipient_context),
954 text=request.text,
955 counterparty_is_initiator=True,
956 ),
957 moderation_state_id=host_request.moderation_state_id,
958 )
960 else:
961 host_request.recipient_last_seen_message_id = message.id
963 recipient_context = make_notification_user_context(user_id=host_request.initiator_user_id)
964 notify(
965 session,
966 user_id=host_request.initiator_user_id,
967 topic_action=NotificationTopicAction.host_request__message,
968 key=str(host_request.conversation_id),
969 data=notification_data_pb2.HostRequestMessage(
970 host_request=host_request_to_pb(host_request, session, recipient_context),
971 counterparty=user_model_to_pb(host_request.recipient, session, recipient_context),
972 text=request.text,
973 counterparty_is_initiator=False,
974 ),
975 moderation_state_id=host_request.moderation_state_id,
976 )
978 session.commit()
980 user_gender = session.execute(select(User.gender).where(User.id == context.user_id)).scalar_one()
981 sent_messages_counter.labels(user_gender, "host request").inc()
982 log_event(
983 context,
984 session,
985 "host_request.message_sent",
986 {
987 "host_request_id": host_request.conversation_id,
988 "surfer_id": host_request.surfer_user_id,
989 "host_id": host_request.host_user_id,
990 "role": "host" if context.user_id == host_request.host_user_id else "surfer",
991 "host_city": host_request.hosting_city,
992 },
993 )
995 return empty_pb2.Empty()
997 def GetHostRequestUpdates(
998 self, request: requests_pb2.GetHostRequestUpdatesReq, context: CouchersContext, session: Session
999 ) -> requests_pb2.GetHostRequestUpdatesRes:
1000 if request.only_sent and request.only_received: 1000 ↛ 1001line 1000 didn't jump to line 1001 because the condition on line 1000 was never true
1001 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "host_request_sent_or_received")
1003 if request.newest_message_id == 0:
1004 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
1006 if not session.execute(select(Message).where(Message.id == request.newest_message_id)).scalar_one_or_none(): 1006 ↛ 1007line 1006 didn't jump to line 1007 because the condition on line 1006 was never true
1007 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_message")
1009 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH
1010 pagination = min(pagination, MAX_PAGE_SIZE)
1012 statement = where_moderated_content_visible(
1013 select(
1014 Message,
1015 HostRequest.status.label("host_request_status"),
1016 HostRequest.conversation_id.label("host_request_id"),
1017 )
1018 .join(HostRequest, HostRequest.conversation_id == Message.conversation_id)
1019 .where(Message.id > request.newest_message_id),
1020 context,
1021 HostRequest,
1022 is_list_operation=False,
1023 )
1025 if request.only_sent: 1025 ↛ 1026line 1025 didn't jump to line 1026 because the condition on line 1025 was never true
1026 statement = statement.where(HostRequest.initiator_user_id == context.user_id)
1027 elif request.only_received: 1027 ↛ 1028line 1027 didn't jump to line 1028 because the condition on line 1027 was never true
1028 statement = statement.where(HostRequest.recipient_user_id == context.user_id)
1029 else:
1030 statement = statement.where(
1031 or_(HostRequest.recipient_user_id == context.user_id, HostRequest.initiator_user_id == context.user_id)
1032 )
1034 statement = statement.order_by(Message.id.asc()).limit(pagination + 1)
1035 res = session.execute(statement).all()
1037 no_more = len(res) <= pagination
1039 last_message_id = min(m.Message.id if m else 1 for m in res[:pagination]) if len(res) > 0 else 0 # TODO
1041 return requests_pb2.GetHostRequestUpdatesRes(
1042 no_more=no_more,
1043 updates=[
1044 requests_pb2.HostRequestUpdate(
1045 host_request_id=result.host_request_id,
1046 status=hostrequeststatus2api[result.host_request_status],
1047 message=message_to_pb(result.Message),
1048 )
1049 for result in res[:pagination]
1050 ],
1051 )
1053 def MarkLastSeenHostRequest(
1054 self, request: requests_pb2.MarkLastSeenHostRequestReq, 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).where(
1058 HostRequest.conversation_id == request.host_request_id
1059 )
1060 ).scalar_one_or_none()
1062 if not host_request: 1062 ↛ 1063line 1062 didn't jump to line 1063 because the condition on line 1062 was never true
1063 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1065 if host_request.initiator_user_id != context.user_id and host_request.recipient_user_id != context.user_id: 1065 ↛ 1066line 1065 didn't jump to line 1066 because the condition on line 1065 was never true
1066 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1068 if host_request.initiator_user_id == context.user_id: 1068 ↛ 1069line 1068 didn't jump to line 1069 because the condition on line 1068 was never true
1069 if not host_request.initiator_last_seen_message_id <= request.last_seen_message_id:
1070 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages")
1071 host_request.initiator_last_seen_message_id = request.last_seen_message_id
1072 else:
1073 if not host_request.recipient_last_seen_message_id <= request.last_seen_message_id:
1074 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages")
1075 host_request.recipient_last_seen_message_id = request.last_seen_message_id
1077 mark_notifications_seen(
1078 session,
1079 user_id=context.user_id,
1080 topic_actions_and_keys=[
1081 (HOST_REQUEST_NOTIFICATION_TOPIC_ACTIONS, [str(host_request.conversation_id)]),
1082 ],
1083 )
1085 session.commit()
1086 return empty_pb2.Empty()
1088 def SetHostRequestArchiveStatus(
1089 self, request: requests_pb2.SetHostRequestArchiveStatusReq, context: CouchersContext, session: Session
1090 ) -> requests_pb2.SetHostRequestArchiveStatusRes:
1091 host_request = session.execute(
1092 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False)
1093 .where(HostRequest.conversation_id == request.host_request_id)
1094 .where(
1095 or_(HostRequest.initiator_user_id == context.user_id, HostRequest.recipient_user_id == context.user_id)
1096 )
1097 ).scalar_one_or_none()
1099 if not host_request: 1099 ↛ 1100line 1099 didn't jump to line 1100 because the condition on line 1099 was never true
1100 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1102 if context.user_id == host_request.initiator_user_id: 1102 ↛ 1105line 1102 didn't jump to line 1105 because the condition on line 1102 was always true
1103 host_request.is_initiator_archived = request.is_archived
1104 else:
1105 host_request.is_recipient_archived = request.is_archived
1107 return requests_pb2.SetHostRequestArchiveStatusRes(
1108 host_request_id=host_request.conversation_id,
1109 is_archived=request.is_archived,
1110 )
1112 def GetResponseRate(
1113 self, request: requests_pb2.GetResponseRateReq, context: CouchersContext, session: Session
1114 ) -> requests_pb2.GetResponseRateRes:
1115 user_res = session.execute(
1116 select(User.id, UserResponseRate)
1117 .outerjoin(UserResponseRate, UserResponseRate.user_id == User.id)
1118 .where(users_visible(context, User))
1119 .where(User.id == request.user_id)
1120 ).one_or_none()
1122 # if user doesn't exist, return None
1123 if not user_res:
1124 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1126 user, response_rates = user_res
1127 return requests_pb2.GetResponseRateRes(**response_rate_to_pb(response_rates)) # type: ignore[arg-type]
1129 def SendHostRequestFeedback(
1130 self, request: requests_pb2.SendHostRequestFeedbackReq, context: CouchersContext, session: Session
1131 ) -> empty_pb2.Empty:
1132 host_request = session.execute(
1133 where_moderated_content_visible(select(HostRequest), context, HostRequest, is_list_operation=False)
1134 .where(HostRequest.conversation_id == request.host_request_id)
1135 .where(HostRequest.recipient_user_id == context.user_id)
1136 # public-trip offers don't take feedback
1137 .where(HostRequest.proposal_kind == StayProposalKind.request)
1138 ).scalar_one_or_none()
1140 if not host_request:
1141 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "host_request_not_found")
1143 feedback = session.execute(
1144 select(HostRequestFeedback)
1145 .where(HostRequestFeedback.host_request_id == host_request.conversation_id)
1146 .where(HostRequestFeedback.from_user_id == context.user_id)
1147 ).scalar_one_or_none()
1149 if feedback:
1150 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_left_host_request_feedback")
1152 session.add(
1153 HostRequestFeedback(
1154 host_request_id=host_request.conversation_id,
1155 from_user_id=host_request.recipient_user_id,
1156 to_user_id=host_request.initiator_user_id,
1157 request_quality=hostrequestquality2sql.get(request.host_request_quality),
1158 decline_reason=request.decline_reason,
1159 )
1160 )
1161 quality = hostrequestquality2sql.get(request.host_request_quality)
1162 log_event(
1163 context,
1164 session,
1165 "host_request.feedback_submitted",
1166 {
1167 "host_request_id": host_request.conversation_id,
1168 "surfer_id": host_request.surfer_user_id,
1169 "host_id": host_request.host_user_id,
1170 "request_quality": quality.name if quality else None,
1171 "has_decline_reason": bool(request.decline_reason),
1172 "host_city": host_request.hosting_city,
1173 },
1174 )
1176 return empty_pb2.Empty()