Coverage for src/couchers/servicers/requests.py: 92%

1import logging 

2from datetime import timedelta 

3 

4import grpc 

5from google.protobuf import empty_pb2 

6from sqlalchemy import Float 

7from sqlalchemy.orm import aliased 

8from sqlalchemy.sql import and_, func, or_ 

9from sqlalchemy.sql.functions import percentile_disc 

10 

11from couchers import errors, urls 

12from couchers.db import session_scope 

13from couchers.models import Conversation, HostRequest, HostRequestStatus, Message, MessageType, User 

14from couchers.notifications.notify import notify 

15from couchers.sql import couchers_select as select 

16from couchers.tasks import ( 

17 send_host_request_accepted_email_to_guest, 

18 send_host_request_cancelled_email_to_host, 

19 send_host_request_confirmed_email_to_host, 

20 send_host_request_rejected_email_to_guest, 

21 send_new_host_request_email, 

22) 

23from couchers.utils import ( 

24 Duration_from_timedelta, 

25 Timestamp_from_datetime, 

26 date_to_api, 

27 now, 

28 parse_date, 

29 today_in_timezone, 

30) 

31from proto import conversations_pb2, requests_pb2, requests_pb2_grpc 

32 

33logger = logging.getLogger(__name__) 

34 

35DEFAULT_PAGINATION_LENGTH = 10 

36MAX_PAGE_SIZE = 50 

37 

38 

39hostrequeststatus2api = { 

40 HostRequestStatus.pending: conversations_pb2.HOST_REQUEST_STATUS_PENDING, 

41 HostRequestStatus.accepted: conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED, 

42 HostRequestStatus.rejected: conversations_pb2.HOST_REQUEST_STATUS_REJECTED, 

43 HostRequestStatus.confirmed: conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED, 

44 HostRequestStatus.cancelled: conversations_pb2.HOST_REQUEST_STATUS_CANCELLED, 

45} 

46 

47 

48def message_to_pb(message: Message): 

49 """ 

50 Turns the given message to a protocol buffer 

51 """ 

52 if message.is_normal_message: 

53 return conversations_pb2.Message( 

54 message_id=message.id, 

55 author_user_id=message.author_id, 

56 time=Timestamp_from_datetime(message.time), 

57 text=conversations_pb2.MessageContentText(text=message.text), 

58 ) 

59 else: 

60 return conversations_pb2.Message( 

61 message_id=message.id, 

62 author_user_id=message.author_id, 

63 time=Timestamp_from_datetime(message.time), 

64 chat_created=( 

65 conversations_pb2.MessageContentChatCreated() 

66 if message.message_type == MessageType.chat_created 

67 else None 

68 ), 

69 host_request_status_changed=( 

70 conversations_pb2.MessageContentHostRequestStatusChanged( 

71 status=hostrequeststatus2api[message.host_request_status_target] 

72 ) 

73 if message.message_type == MessageType.host_request_status_changed 

74 else None 

75 ), 

76 ) 

77 

78 

79class Requests(requests_pb2_grpc.RequestsServicer): 

80 def CreateHostRequest(self, request, context): 

81 with session_scope() as session: 

82 if request.host_user_id == context.user_id: 

83 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.CANT_REQUEST_SELF) 

84 

85 # just to check host exists and is visible 

86 host = session.execute( 

87 select(User).where_users_visible(context).where(User.id == request.host_user_id) 

88 ).scalar_one_or_none() 

89 if not host: 

90 context.abort(grpc.StatusCode.NOT_FOUND, errors.USER_NOT_FOUND) 

91 

92 from_date = parse_date(request.from_date) 

93 to_date = parse_date(request.to_date) 

94 

95 if not from_date or not to_date: 

96 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.INVALID_DATE) 

97 

98 today = today_in_timezone(host.timezone) 

99 

100 # request starts from the past 

101 if from_date < today: 

102 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.DATE_FROM_BEFORE_TODAY) 

103 

104 # from_date is not >= to_date 

105 if from_date >= to_date: 

106 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.DATE_FROM_AFTER_TO) 

107 

108 # No need to check today > to_date 

109 

110 if from_date - today > timedelta(days=365): 

111 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.DATE_FROM_AFTER_ONE_YEAR) 

112 

113 if to_date - from_date > timedelta(days=365): 

114 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.DATE_TO_AFTER_ONE_YEAR) 

115 

116 conversation = Conversation() 

117 session.add(conversation) 

118 session.flush() 

119 

120 session.add( 

121 Message( 

122 conversation_id=conversation.id, 

123 author_id=context.user_id, 

124 message_type=MessageType.chat_created, 

125 ) 

126 ) 

127 

128 message = Message( 

129 conversation_id=conversation.id, 

130 author_id=context.user_id, 

131 text=request.text, 

132 message_type=MessageType.text, 

133 ) 

134 session.add(message) 

135 session.flush() 

136 

137 host_request = HostRequest( 

138 conversation_id=conversation.id, 

139 surfer_user_id=context.user_id, 

140 host_user_id=host.id, 

141 from_date=from_date, 

142 to_date=to_date, 

143 status=HostRequestStatus.pending, 

144 surfer_last_seen_message_id=message.id, 

145 # TODO: tz 

146 # timezone=host.timezone, 

147 ) 

148 session.add(host_request) 

149 session.commit() 

150 

151 send_new_host_request_email(host_request) 

152 

153 notify( 

154 user_id=host_request.host_user_id, 

155 topic="host_request", 

156 action="create", 

157 key=str(host_request.surfer_user_id), 

158 avatar_key=host_request.surfer.avatar.thumbnail_url if host_request.surfer.avatar else None, 

159 title=f"**{host_request.surfer.name}** sent you a hosting request", 

160 content=request.text, 

161 link=urls.host_request_link_host(), 

162 ) 

163 

164 return requests_pb2.CreateHostRequestRes(host_request_id=host_request.conversation_id) 

165 

166 def GetHostRequest(self, request, context): 

167 with session_scope() as session: 

168 host_request = session.execute( 

169 select(HostRequest) 

170 .where_users_column_visible(context, HostRequest.surfer_user_id) 

171 .where_users_column_visible(context, HostRequest.host_user_id) 

172 .where(HostRequest.conversation_id == request.host_request_id) 

173 .where(or_(HostRequest.surfer_user_id == context.user_id, HostRequest.host_user_id == context.user_id)) 

174 ).scalar_one_or_none() 

175 

176 if not host_request: 

177 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

178 

179 initial_message = session.execute( 

180 select(Message) 

181 .where(Message.conversation_id == host_request.conversation_id) 

182 .order_by(Message.id.asc()) 

183 .limit(1) 

184 ).scalar_one() 

185 

186 latest_message = session.execute( 

187 select(Message) 

188 .where(Message.conversation_id == host_request.conversation_id) 

189 .order_by(Message.id.desc()) 

190 .limit(1) 

191 ).scalar_one() 

192 

193 return requests_pb2.HostRequest( 

194 host_request_id=host_request.conversation_id, 

195 surfer_user_id=host_request.surfer_user_id, 

196 host_user_id=host_request.host_user_id, 

197 status=hostrequeststatus2api[host_request.status], 

198 created=Timestamp_from_datetime(initial_message.time), 

199 from_date=date_to_api(host_request.from_date), 

200 to_date=date_to_api(host_request.to_date), 

201 last_seen_message_id=( 

202 host_request.surfer_last_seen_message_id 

203 if context.user_id == host_request.surfer_user_id 

204 else host_request.host_last_seen_message_id 

205 ), 

206 latest_message=message_to_pb(latest_message), 

207 ) 

208 

209 def ListHostRequests(self, request, context): 

210 if request.only_sent and request.only_received: 

211 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.HOST_REQUEST_SENT_OR_RECEIVED) 

212 

213 with session_scope() as session: 

214 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH 

215 pagination = min(pagination, MAX_PAGE_SIZE) 

216 

217 # By outer joining messages on itself where the second id is bigger, only the highest IDs will have 

218 # none as message_2.id. So just filter for these ones to get highest messages only. 

219 # See https://stackoverflow.com/a/27802817/6115336 

220 message_2 = aliased(Message) 

221 statement = ( 

222 select(Message, HostRequest, Conversation) 

223 .outerjoin( 

224 message_2, and_(Message.conversation_id == message_2.conversation_id, Message.id < message_2.id) 

225 ) 

226 .join(HostRequest, HostRequest.conversation_id == Message.conversation_id) 

227 .join(Conversation, Conversation.id == HostRequest.conversation_id) 

228 .where_users_column_visible(context, HostRequest.surfer_user_id) 

229 .where_users_column_visible(context, HostRequest.host_user_id) 

230 .where(message_2.id == None) 

231 .where(or_(Message.id < request.last_request_id, request.last_request_id == 0)) 

232 ) 

233 

234 if request.only_sent: 

235 statement = statement.where(HostRequest.surfer_user_id == context.user_id) 

236 elif request.only_received: 

237 statement = statement.where(HostRequest.host_user_id == context.user_id) 

238 else: 

239 statement = statement.where( 

240 or_(HostRequest.host_user_id == context.user_id, HostRequest.surfer_user_id == context.user_id) 

241 ) 

242 

243 # TODO: I considered having the latest control message be the single source of truth for 

244 # the HostRequest.status, but decided against it because of this filter. 

245 # Another possibility is to filter in the python instead of SQL, but that's slower 

246 if request.only_active: 

247 statement = statement.where( 

248 or_( 

249 HostRequest.status == HostRequestStatus.pending, 

250 HostRequest.status == HostRequestStatus.accepted, 

251 HostRequest.status == HostRequestStatus.confirmed, 

252 ) 

253 ) 

254 statement = statement.where(HostRequest.end_time <= func.now()) 

255 

256 statement = statement.order_by(Message.id.desc()).limit(pagination + 1) 

257 results = session.execute(statement).all() 

258 

259 host_requests = [ 

260 requests_pb2.HostRequest( 

261 host_request_id=result.HostRequest.conversation_id, 

262 surfer_user_id=result.HostRequest.surfer_user_id, 

263 host_user_id=result.HostRequest.host_user_id, 

264 status=hostrequeststatus2api[result.HostRequest.status], 

265 created=Timestamp_from_datetime(result.Conversation.created), 

266 from_date=date_to_api(result.HostRequest.from_date), 

267 to_date=date_to_api(result.HostRequest.to_date), 

268 last_seen_message_id=( 

269 result.HostRequest.surfer_last_seen_message_id 

270 if context.user_id == result.HostRequest.surfer_user_id 

271 else result.HostRequest.host_last_seen_message_id 

272 ), 

273 latest_message=message_to_pb(result.Message), 

274 ) 

275 for result in results[:pagination] 

276 ] 

277 last_request_id = ( 

278 min(g.Message.id for g in results[:pagination]) if len(results) > pagination else 0 

279 ) # TODO 

280 no_more = len(results) <= pagination 

281 

282 return requests_pb2.ListHostRequestsRes( 

283 last_request_id=last_request_id, no_more=no_more, host_requests=host_requests 

284 ) 

285 

286 def RespondHostRequest(self, request, context): 

287 with session_scope() as session: 

288 host_request = session.execute( 

289 select(HostRequest) 

290 .where_users_column_visible(context, HostRequest.surfer_user_id) 

291 .where_users_column_visible(context, HostRequest.host_user_id) 

292 .where(HostRequest.conversation_id == request.host_request_id) 

293 ).scalar_one_or_none() 

294 

295 if not host_request: 

296 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

297 

298 if host_request.surfer_user_id != context.user_id and host_request.host_user_id != context.user_id: 

299 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

300 

301 if request.status == conversations_pb2.HOST_REQUEST_STATUS_PENDING: 

302 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.INVALID_HOST_REQUEST_STATUS) 

303 

304 if host_request.end_time < now(): 

305 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.HOST_REQUEST_IN_PAST) 

306 

307 control_message = Message() 

308 

309 if request.status == conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED: 

310 # only host can accept 

311 if context.user_id != host_request.host_user_id: 

312 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.NOT_THE_HOST) 

313 # can't accept a cancelled or confirmed request (only reject), or already accepted 

314 if ( 

315 host_request.status == HostRequestStatus.cancelled 

316 or host_request.status == HostRequestStatus.confirmed 

317 or host_request.status == HostRequestStatus.accepted 

318 ): 

319 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.INVALID_HOST_REQUEST_STATUS) 

320 control_message.host_request_status_target = HostRequestStatus.accepted 

321 host_request.status = HostRequestStatus.accepted 

322 

323 send_host_request_accepted_email_to_guest(host_request) 

324 

325 notify( 

326 user_id=host_request.surfer_user_id, 

327 topic="host_request", 

328 action="accept", 

329 key=str(host_request.host_user_id), 

330 avatar_key=host_request.host.avatar.thumbnail_url if host_request.host.avatar else None, 

331 title=f"**{host_request.host.name}** accepted your host request", 

332 link=urls.host_request_link_guest(), 

333 ) 

334 

335 if request.status == conversations_pb2.HOST_REQUEST_STATUS_REJECTED: 

336 # only host can reject 

337 if context.user_id != host_request.host_user_id: 

338 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.INVALID_HOST_REQUEST_STATUS) 

339 # can't reject a cancelled or already rejected request 

340 if ( 

341 host_request.status == HostRequestStatus.cancelled 

342 or host_request.status == HostRequestStatus.rejected 

343 ): 

344 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.INVALID_HOST_REQUEST_STATUS) 

345 control_message.host_request_status_target = HostRequestStatus.rejected 

346 host_request.status = HostRequestStatus.rejected 

347 

348 send_host_request_rejected_email_to_guest(host_request) 

349 

350 notify( 

351 user_id=host_request.surfer_user_id, 

352 topic="host_request", 

353 action="reject", 

354 key=str(host_request.host_user_id), 

355 avatar_key=host_request.host.avatar.thumbnail_url if host_request.host.avatar else None, 

356 title=f"**{host_request.host.name}** rejected your host request", 

357 link=urls.host_request_link_guest(), 

358 ) 

359 

360 if request.status == conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED: 

361 # only hostee can confirm 

362 if context.user_id != host_request.surfer_user_id: 

363 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.INVALID_HOST_REQUEST_STATUS) 

364 # can only confirm an accepted request 

365 if host_request.status != HostRequestStatus.accepted: 

366 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.INVALID_HOST_REQUEST_STATUS) 

367 control_message.host_request_status_target = HostRequestStatus.confirmed 

368 host_request.status = HostRequestStatus.confirmed 

369 

370 send_host_request_confirmed_email_to_host(host_request) 

371 

372 notify( 

373 user_id=host_request.host_user_id, 

374 topic="host_request", 

375 action="confirm", 

376 key=str(host_request.surfer_user_id), 

377 avatar_key=host_request.surfer.avatar.thumbnail_url if host_request.surfer.avatar else None, 

378 title=f"**{host_request.surfer.name}** confirmed their host request", 

379 link=urls.host_request_link_host(), 

380 ) 

381 

382 if request.status == conversations_pb2.HOST_REQUEST_STATUS_CANCELLED: 

383 # only hostee can cancel 

384 if context.user_id != host_request.surfer_user_id: 

385 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.INVALID_HOST_REQUEST_STATUS) 

386 # can't' cancel an already cancelled or rejected request 

387 if ( 

388 host_request.status == HostRequestStatus.rejected 

389 or host_request.status == HostRequestStatus.cancelled 

390 ): 

391 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.INVALID_HOST_REQUEST_STATUS) 

392 control_message.host_request_status_target = HostRequestStatus.cancelled 

393 host_request.status = HostRequestStatus.cancelled 

394 

395 send_host_request_cancelled_email_to_host(host_request) 

396 

397 notify( 

398 user_id=host_request.host_user_id, 

399 topic="host_request", 

400 action="cancel", 

401 key=str(host_request.surfer_user_id), 

402 avatar_key=host_request.surfer.avatar.thumbnail_url if host_request.surfer.avatar else None, 

403 title=f"**{host_request.surfer.name}** cancelled their host request", 

404 link=urls.host_request_link_host(), 

405 ) 

406 

407 control_message.message_type = MessageType.host_request_status_changed 

408 control_message.conversation_id = host_request.conversation_id 

409 control_message.author_id = context.user_id 

410 session.add(control_message) 

411 

412 if request.text: 

413 latest_message = Message() 

414 latest_message.conversation_id = host_request.conversation_id 

415 latest_message.text = request.text 

416 latest_message.author_id = context.user_id 

417 latest_message.message_type = MessageType.text 

418 session.add(latest_message) 

419 else: 

420 latest_message = control_message 

421 

422 session.flush() 

423 

424 if host_request.surfer_user_id == context.user_id: 

425 host_request.surfer_last_seen_message_id = latest_message.id 

426 else: 

427 host_request.host_last_seen_message_id = latest_message.id 

428 session.commit() 

429 

430 return empty_pb2.Empty() 

431 

432 def GetHostRequestMessages(self, request, context): 

433 with session_scope() as session: 

434 host_request = session.execute( 

435 select(HostRequest).where(HostRequest.conversation_id == request.host_request_id) 

436 ).scalar_one_or_none() 

437 

438 if not host_request: 

439 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

440 

441 if host_request.surfer_user_id != context.user_id and host_request.host_user_id != context.user_id: 

442 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

443 

444 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH 

445 pagination = min(pagination, MAX_PAGE_SIZE) 

446 

447 messages = ( 

448 session.execute( 

449 select(Message) 

450 .where(Message.conversation_id == host_request.conversation_id) 

451 .where(or_(Message.id < request.last_message_id, request.last_message_id == 0)) 

452 .order_by(Message.id.desc()) 

453 .limit(pagination + 1) 

454 ) 

455 .scalars() 

456 .all() 

457 ) 

458 

459 no_more = len(messages) <= pagination 

460 

461 last_message_id = min(map(lambda m: m.id if m else 1, messages[:pagination])) if len(messages) > 0 else 0 

462 

463 return requests_pb2.GetHostRequestMessagesRes( 

464 last_message_id=last_message_id, 

465 no_more=no_more, 

466 messages=[message_to_pb(message) for message in messages[:pagination]], 

467 ) 

468 

469 def SendHostRequestMessage(self, request, context): 

470 if request.text == "": 

471 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.INVALID_MESSAGE) 

472 with session_scope() as session: 

473 host_request = session.execute( 

474 select(HostRequest).where(HostRequest.conversation_id == request.host_request_id) 

475 ).scalar_one_or_none() 

476 

477 if not host_request: 

478 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

479 

480 if host_request.surfer_user_id != context.user_id and host_request.host_user_id != context.user_id: 

481 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

482 

483 if host_request.status == HostRequestStatus.rejected or host_request.status == HostRequestStatus.cancelled: 

484 context.abort(grpc.StatusCode.PERMISSION_DENIED, errors.HOST_REQUEST_CLOSED) 

485 

486 if host_request.end_time < now(): 

487 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.HOST_REQUEST_IN_PAST) 

488 

489 message = Message() 

490 message.conversation_id = host_request.conversation_id 

491 message.author_id = context.user_id 

492 message.message_type = MessageType.text 

493 message.text = request.text 

494 session.add(message) 

495 session.flush() 

496 

497 if host_request.surfer_user_id == context.user_id: 

498 host_request.surfer_last_seen_message_id = message.id 

499 

500 notify( 

501 user_id=host_request.host_user_id, 

502 topic="host_request", 

503 action="message", 

504 key=str(host_request.surfer_user_id), 

505 avatar_key=host_request.surfer.avatar.thumbnail_url if host_request.surfer.avatar else None, 

506 title=f"**{host_request.surfer.name}** sent a message in their host request", 

507 link=urls.host_request_link_host(), 

508 ) 

509 

510 else: 

511 host_request.host_last_seen_message_id = message.id 

512 

513 notify( 

514 user_id=host_request.surfer_user_id, 

515 topic="host_request", 

516 action="message", 

517 key=str(host_request.host_user_id), 

518 avatar_key=host_request.host.avatar.thumbnail_url if host_request.host.avatar else None, 

519 title=f"**{host_request.host.name}** sent a message in your host request", 

520 link=urls.host_request_link_guest(), 

521 ) 

522 

523 session.commit() 

524 

525 return empty_pb2.Empty() 

526 

527 def GetHostRequestUpdates(self, request, context): 

528 if request.only_sent and request.only_received: 

529 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.HOST_REQUEST_SENT_OR_RECEIVED) 

530 

531 with session_scope() as session: 

532 if request.newest_message_id == 0: 

533 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.INVALID_MESSAGE) 

534 

535 if not session.execute(select(Message).where(Message.id == request.newest_message_id)).scalar_one_or_none(): 

536 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.INVALID_MESSAGE) 

537 

538 pagination = request.number if request.number > 0 else DEFAULT_PAGINATION_LENGTH 

539 pagination = min(pagination, MAX_PAGE_SIZE) 

540 

541 statement = ( 

542 select( 

543 Message, 

544 HostRequest.status.label("host_request_status"), 

545 HostRequest.conversation_id.label("host_request_id"), 

546 ) 

547 .join(HostRequest, HostRequest.conversation_id == Message.conversation_id) 

548 .where(Message.id > request.newest_message_id) 

549 ) 

550 

551 if request.only_sent: 

552 statement = statement.where(HostRequest.surfer_user_id == context.user_id) 

553 elif request.only_received: 

554 statement = statement.where(HostRequest.host_user_id == context.user_id) 

555 else: 

556 statement = statement.where( 

557 or_(HostRequest.host_user_id == context.user_id, HostRequest.surfer_user_id == context.user_id) 

558 ) 

559 

560 statement = statement.order_by(Message.id.asc()).limit(pagination + 1) 

561 res = session.execute(statement).all() 

562 

563 no_more = len(res) <= pagination 

564 

565 last_message_id = ( 

566 min(map(lambda m: m.Message.id if m else 1, res[:pagination])) if len(res) > 0 else 0 

567 ) # TODO 

568 

569 return requests_pb2.GetHostRequestUpdatesRes( 

570 no_more=no_more, 

571 updates=[ 

572 requests_pb2.HostRequestUpdate( 

573 host_request_id=result.host_request_id, 

574 status=hostrequeststatus2api[result.host_request_status], 

575 message=message_to_pb(result.Message), 

576 ) 

577 for result in res[:pagination] 

578 ], 

579 ) 

580 

581 def MarkLastSeenHostRequest(self, request, context): 

582 with session_scope() as session: 

583 host_request = session.execute( 

584 select(HostRequest).where(HostRequest.conversation_id == request.host_request_id) 

585 ).scalar_one_or_none() 

586 

587 if not host_request: 

588 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

589 

590 if host_request.surfer_user_id != context.user_id and host_request.host_user_id != context.user_id: 

591 context.abort(grpc.StatusCode.NOT_FOUND, errors.HOST_REQUEST_NOT_FOUND) 

592 

593 if host_request.surfer_user_id == context.user_id: 

594 if not host_request.surfer_last_seen_message_id <= request.last_seen_message_id: 

595 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.CANT_UNSEE_MESSAGES) 

596 host_request.surfer_last_seen_message_id = request.last_seen_message_id 

597 else: 

598 if not host_request.host_last_seen_message_id <= request.last_seen_message_id: 

599 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.CANT_UNSEE_MESSAGES) 

600 host_request.host_last_seen_message_id = request.last_seen_message_id 

601 

602 session.commit() 

603 return empty_pb2.Empty() 

604 

605 def GetResponseRate(self, request, context): 

606 with session_scope() as session: 

607 # this subquery gets the time that the request was sent 

608 t = ( 

609 select(Message.conversation_id, Message.time) 

610 .where(Message.message_type == MessageType.chat_created) 

611 .subquery() 

612 ) 

613 # this subquery gets the time that the user responded to the request 

614 s = ( 

615 select(Message.conversation_id, func.min(Message.time).label("time")) 

616 .where(Message.author_id == request.user_id) 

617 .group_by(Message.conversation_id) 

618 .subquery() 

619 ) 

620 

621 res = session.execute( 

622 select( 

623 User.id, 

624 # number of requests received 

625 func.count().label("n"), 

626 # percentage of requests responded to 

627 (func.count(s.c.time) / func.cast(func.greatest(func.count(t.c.time), 1.0), Float)).label( 

628 "response_rate" 

629 ), 

630 # the 33rd percentile response time 

631 percentile_disc(0.33) 

632 .within_group(func.coalesce(s.c.time - t.c.time, timedelta(days=1000))) 

633 .label("response_time_p33"), 

634 # the 66th percentile response time 

635 percentile_disc(0.66) 

636 .within_group(func.coalesce(s.c.time - t.c.time, timedelta(days=1000))) 

637 .label("response_time_p66"), 

638 ) 

639 .where_users_visible(context) 

640 .where(User.id == request.user_id) 

641 .outerjoin(HostRequest, HostRequest.host_user_id == User.id) 

642 .outerjoin(t, t.c.conversation_id == HostRequest.conversation_id) 

643 .outerjoin(s, s.c.conversation_id == HostRequest.conversation_id) 

644 .group_by(User.id) 

645 ).one_or_none() 

646 

647 if not res: 

648 context.abort(grpc.StatusCode.NOT_FOUND, errors.USER_NOT_FOUND) 

649 

650 _, n, response_rate, response_time_p33, response_time_p66 = res 

651 

652 if n < 3: 

653 return requests_pb2.GetResponseRateRes( 

654 insufficient_data=requests_pb2.ResponseRateInsufficientData(), 

655 ) 

656 

657 if response_rate <= 0.33: 

658 return requests_pb2.GetResponseRateRes( 

659 low=requests_pb2.ResponseRateLow(), 

660 ) 

661 

662 response_time_p33_coarsened = Duration_from_timedelta( 

663 timedelta(seconds=round(response_time_p33.total_seconds() / 60) * 60) 

664 ) 

665 

666 if response_rate <= 0.66: 

667 return requests_pb2.GetResponseRateRes( 

668 some=requests_pb2.ResponseRateSome(response_time_p33=response_time_p33_coarsened), 

669 ) 

670 

671 response_time_p66_coarsened = Duration_from_timedelta( 

672 timedelta(seconds=round(response_time_p66.total_seconds() / 60) * 60) 

673 ) 

674 

675 if response_rate <= 0.90: 

676 return requests_pb2.GetResponseRateRes( 

677 most=requests_pb2.ResponseRateMost( 

678 response_time_p33=response_time_p33_coarsened, response_time_p66=response_time_p66_coarsened 

679 ), 

680 ) 

681 else: 

682 return requests_pb2.GetResponseRateRes( 

683 almost_all=requests_pb2.ResponseRateAlmostAll( 

684 response_time_p33=response_time_p33_coarsened, response_time_p66=response_time_p66_coarsened 

685 ), 

686 )