Coverage for app/backend/src/couchers/servicers/conversations.py: 88%

1import logging 

2from collections.abc import Sequence 

3from datetime import timedelta 

4from typing import Any, cast 

5 

6import grpc 

7from google.protobuf import empty_pb2 

8from sqlalchemy import select 

9from sqlalchemy.orm import Session, contains_eager 

10from sqlalchemy.sql import and_, func, not_, or_ 

11 

12from couchers.constants import DATETIME_INFINITY, DATETIME_MINUS_INFINITY 

13from couchers.context import CouchersContext, make_background_user_context, make_notification_user_context 

14from couchers.db import session_scope 

15from couchers.event_log import log_event 

16from couchers.helpers.completed_profile import has_completed_profile 

17from couchers.helpers.group_chats import is_newest_subscription, is_unseen, mute_info, was_subscribed_at 

18from couchers.helpers.messages import message_to_pb 

19from couchers.jobs.enqueue import queue_job 

20from couchers.metrics import sent_messages_counter 

21from couchers.models import ( 

22 Conversation, 

23 GroupChat, 

24 GroupChatRole, 

25 GroupChatSubscription, 

26 Message, 

27 MessageType, 

28 ModerationObjectType, 

29 RateLimitAction, 

30 User, 

31) 

32from couchers.models.notifications import NotificationTopicAction 

33from couchers.moderation.utils import create_moderation 

34from couchers.notifications.notify import mark_notifications_seen, notify 

35from couchers.proto import conversations_pb2, conversations_pb2_grpc, notification_data_pb2 

36from couchers.proto.internal import jobs_pb2 

37from couchers.rate_limits.check import process_rate_limits_and_check_abort 

38from couchers.rate_limits.definitions import RATE_LIMIT_HOURS 

39from couchers.servicers.api import user_model_to_pb 

40from couchers.servicers.message_threads import list_message_threads, mark_all_threads_seen 

41from couchers.sql import to_bool, users_visible, where_moderated_content_visible, where_users_column_visible 

42from couchers.utils import Timestamp_from_datetime, now 

43 

44logger = logging.getLogger(__name__) 

45 

46# TODO: Still needs custom pagination: GetUpdates 

47DEFAULT_PAGINATION_LENGTH = 20 

48MAX_PAGE_SIZE = 50 

49 

50 

51# TODO(#7722): remove with the legacy conversations endpoints; the ListMessageThreads path filters 

52# preloaded subscriptions with message_threads._member_visible_to_viewer instead 

53def _get_visible_members_for_subscription(subscription: GroupChatSubscription) -> list[int]: 

54 """ 

55 If a user leaves a group chat, they shouldn't be able to see who's added 

56 after they left 

57 """ 

58 if not subscription.left: 

59 # still in the chat, we see everyone with a current subscription 

60 return [sub.user_id for sub in subscription.group_chat.subscriptions.where(GroupChatSubscription.left == None)] 

61 else: 

62 # not in chat anymore, see everyone who was in chat when we left 

63 return [ 

64 sub.user_id 

65 for sub in subscription.group_chat.subscriptions.where( 

66 was_subscribed_at(GroupChatSubscription, subscription.left) 

67 ) 

68 ] 

69 

70 

71def _get_visible_admins_for_subscription(subscription: GroupChatSubscription) -> list[int]: 

72 """ 

73 If a user leaves a group chat, they shouldn't be able to see who's added 

74 after they left 

75 """ 

76 if not subscription.left: 

77 # still in the chat, we see everyone with a current subscription 

78 return [ 

79 sub.user_id 

80 for sub in subscription.group_chat.subscriptions.where(GroupChatSubscription.left == None).where( 

81 GroupChatSubscription.role == GroupChatRole.admin 

82 ) 

83 ] 

84 else: 

85 # not in chat anymore, see everyone who was in chat when we left 

86 return [ 

87 sub.user_id 

88 for sub in subscription.group_chat.subscriptions.where( 

89 GroupChatSubscription.role == GroupChatRole.admin 

90 ).where(was_subscribed_at(GroupChatSubscription, subscription.left)) 

91 ] 

92 

93 

94def _user_can_message(session: Session, context: CouchersContext, group_chat: GroupChat) -> bool: 

95 """ 

96 If it is a true group chat (not a DM), user can always message. For a DM, user can message if the other participant 

97 - Is not deleted/banned 

98 - Has not been blocked by the user or is blocking the user 

99 - Has not left the chat 

100 """ 

101 if not group_chat.is_dm: 

102 return True 

103 

104 query = select( 

105 where_users_column_visible( 

106 select(GroupChatSubscription) 

107 .where(GroupChatSubscription.user_id != context.user_id) 

108 .where(GroupChatSubscription.group_chat_id == group_chat.conversation_id) 

109 .where(GroupChatSubscription.left == None), 

110 context=context, 

111 column=GroupChatSubscription.user_id, 

112 ).exists() 

113 ) 

114 return session.execute(query).scalar_one() 

115 

116 

117def generate_message_notifications(payload: jobs_pb2.GenerateMessageNotificationsPayload) -> None: 

118 """ 

119 Background job to generate notifications for a message sent to a group chat 

120 """ 

121 logger.info(f"Fanning notifications for message_id = {payload.message_id}") 

122 

123 with session_scope() as session: 

124 message, group_chat = session.execute( 

125 select(Message, GroupChat) 

126 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id) 

127 .where(Message.id == payload.message_id) 

128 ).one() 

129 

130 if message.message_type != MessageType.text: 

131 logger.info(f"Not a text message, not notifying. message_id = {payload.message_id}") 

132 return 

133 

134 context = make_background_user_context(user_id=message.author_id) 

135 user_ids_to_notify = ( 

136 session.execute( 

137 where_users_column_visible( 

138 select(GroupChatSubscription.user_id) 

139 .where(GroupChatSubscription.group_chat_id == message.conversation_id) 

140 .where(GroupChatSubscription.user_id != message.author_id) 

141 .where(was_subscribed_at(GroupChatSubscription, message.time)) 

142 .where(not_(GroupChatSubscription.is_muted)), 

143 context=context, 

144 column=GroupChatSubscription.user_id, 

145 ) 

146 ) 

147 .scalars() 

148 .all() 

149 ) 

150 

151 for user_id in user_ids_to_notify: 

152 notify( 

153 session, 

154 user_id=user_id, 

155 topic_action=NotificationTopicAction.chat__message, 

156 key=str(message.conversation_id), 

157 data=notification_data_pb2.ChatMessage( 

158 author=user_model_to_pb( 

159 message.author, 

160 session, 

161 make_notification_user_context(user_id=user_id), 

162 ), 

163 text=message.text, 

164 group_chat_id=message.conversation_id, 

165 group_chat_title=group_chat.title or None, 

166 # unseen_count irrelevant for this notification 

167 ), 

168 moderation_state_id=group_chat.moderation_state_id, 

169 ) 

170 

171 

172def _add_message_to_subscription(session: Session, subscription: GroupChatSubscription, **kwargs: Any) -> Message: 

173 """ 

174 Creates a new message for a subscription, from the user whose subscription that is. Updates last seen message id 

175 

176 Specify the keyword args for Message 

177 """ 

178 message = Message(conversation_id=subscription.group_chat.conversation.id, author_id=subscription.user_id, **kwargs) 

179 

180 session.add(message) 

181 session.flush() 

182 

183 subscription.last_seen_message_id = message.id 

184 

185 queue_job( 

186 session, 

187 job=generate_message_notifications, 

188 payload=jobs_pb2.GenerateMessageNotificationsPayload( 

189 message_id=message.id, 

190 ), 

191 ) 

192 

193 return message 

194 

195 

196def _create_chat( 

197 session: Session, 

198 creator_id: int, 

199 recipient_ids: Sequence[int], 

200 title: str | None = None, 

201 only_admins_invite: bool = True, 

202) -> GroupChat: 

203 conversation = Conversation() 

204 session.add(conversation) 

205 session.flush() 

206 

207 # Create moderation state for UMS (starts as SHADOWED) 

208 moderation_state = create_moderation( 

209 session=session, 

210 object_type=ModerationObjectType.group_chat, 

211 object_id=conversation.id, 

212 creator_user_id=creator_id, 

213 ) 

214 

215 chat = GroupChat( 

216 conversation_id=conversation.id, 

217 title=title, 

218 creator_id=creator_id, 

219 is_dm=True if len(recipient_ids) == 1 else False, 

220 only_admins_invite=only_admins_invite, 

221 moderation_state_id=moderation_state.id, 

222 ) 

223 session.add(chat) 

224 session.flush() 

225 

226 creator_subscription = GroupChatSubscription( 

227 user_id=creator_id, 

228 group_chat_id=chat.conversation_id, 

229 role=GroupChatRole.admin, 

230 ) 

231 session.add(creator_subscription) 

232 

233 for uid in recipient_ids: 

234 session.add( 

235 GroupChatSubscription( 

236 user_id=uid, 

237 group_chat_id=chat.conversation_id, 

238 role=GroupChatRole.participant, 

239 ) 

240 ) 

241 

242 return chat 

243 

244 

245def _get_message_subscription(session: Session, user_id: int, conversation_id: int) -> GroupChatSubscription: 

246 subscription = session.execute( 

247 select(GroupChatSubscription) 

248 .where(GroupChatSubscription.group_chat_id == conversation_id) 

249 .where(GroupChatSubscription.user_id == user_id) 

250 .where(GroupChatSubscription.left == None) 

251 ).scalar_one_or_none() 

252 

253 return cast(GroupChatSubscription, subscription) 

254 

255 

256def _get_visible_message_subscription( 

257 session: Session, context: CouchersContext, conversation_id: int, *, include_left: bool = False 

258) -> GroupChatSubscription: 

259 """ 

260 Get the user's newest subscription to the chat, with visibility filtering. Requires that they're 

261 still in the chat unless include_left, which is only for marking a chat seen: messages left unread 

262 when you leave keep counting towards the badge, so you need a way to clear it. 

263 """ 

264 subscription = session.execute( 

265 where_moderated_content_visible( 

266 select(GroupChatSubscription) 

267 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id) 

268 .where(GroupChatSubscription.group_chat_id == conversation_id) 

269 .where(GroupChatSubscription.user_id == context.user_id) 

270 .where(is_newest_subscription(context.user_id)) 

271 .where(or_(to_bool(include_left), GroupChatSubscription.left == None)), 

272 context, 

273 GroupChat, 

274 is_list_operation=False, 

275 ) 

276 ).scalar_one_or_none() 

277 

278 return cast(GroupChatSubscription, subscription) 

279 

280 

281def _unseen_message_count(session: Session, subscription_id: int) -> int: 

282 query = ( 

283 select(func.count()) 

284 .select_from(Message) 

285 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id) 

286 .where(GroupChatSubscription.id == subscription_id) 

287 .where(is_unseen(Message, GroupChatSubscription)) 

288 ) 

289 return session.execute(query).scalar_one() 

290 

291 

292class Conversations(conversations_pb2_grpc.ConversationsServicer): 

293 def ListMessageThreads( 

294 self, request: conversations_pb2.ListMessageThreadsReq, context: CouchersContext, session: Session 

295 ) -> conversations_pb2.ListMessageThreadsRes: 

296 return list_message_threads(request, context, session) 

297 

298 def MarkAllThreadsSeen( 

299 self, request: conversations_pb2.MarkAllThreadsSeenReq, context: CouchersContext, session: Session 

300 ) -> empty_pb2.Empty: 

301 return mark_all_threads_seen(request, context, session) 

302 

303 # TODO(#7722): remove after FE migrates to ListMessageThreads 

304 def ListGroupChats( 

305 self, request: conversations_pb2.ListGroupChatsReq, context: CouchersContext, session: Session 

306 ) -> conversations_pb2.ListGroupChatsRes: 

307 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH 

308 page_size = min(page_size, MAX_PAGE_SIZE) 

309 

310 # select group chats where you have a subscription, and for each of 

311 # these, the latest message from them 

312 

313 t = ( 

314 select( 

315 GroupChatSubscription.group_chat_id.label("group_chat_id"), 

316 func.max(GroupChatSubscription.id).label("group_chat_subscriptions_id"), 

317 func.max(Message.id).label("message_id"), 

318 ) 

319 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id) 

320 .where(GroupChatSubscription.user_id == context.user_id) 

321 .where(is_newest_subscription(context.user_id)) 

322 .where(was_subscribed_at(GroupChatSubscription, Message.time)) 

323 .where( 

324 or_( 

325 to_bool(request.HasField("only_archived") == False), 

326 GroupChatSubscription.is_archived == request.only_archived, 

327 ) 

328 ) 

329 .group_by(GroupChatSubscription.group_chat_id) 

330 .order_by(func.max(Message.id).desc()) 

331 .subquery() 

332 ) 

333 

334 results = session.execute( 

335 where_moderated_content_visible( 

336 select(t, GroupChat, GroupChatSubscription, Message) 

337 .join(Message, Message.id == t.c.message_id) 

338 .join(GroupChatSubscription, GroupChatSubscription.id == t.c.group_chat_subscriptions_id) 

339 .join(GroupChat, GroupChat.conversation_id == t.c.group_chat_id) 

340 .join(Conversation, Conversation.id == GroupChat.conversation_id) 

341 .options(contains_eager(GroupChat.conversation)) 

342 .where(or_(t.c.message_id < request.last_message_id, to_bool(request.last_message_id == 0))) 

343 .order_by(t.c.message_id.desc()) 

344 .limit(page_size + 1), 

345 context, 

346 GroupChat, 

347 is_list_operation=True, 

348 ) 

349 ).all() 

350 

351 # Batch: unseen message counts in one query instead of N individual queries 

352 subscription_ids = [r.GroupChatSubscription.id for r in results[:page_size]] 

353 unseen_counts: dict[int, int] = dict( 

354 session.execute( # type: ignore[arg-type] 

355 select(GroupChatSubscription.id, func.count(Message.id)) 

356 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id) 

357 .where(GroupChatSubscription.id.in_(subscription_ids)) 

358 .where(is_unseen(Message, GroupChatSubscription)) 

359 .group_by(GroupChatSubscription.id) 

360 ).all() 

361 ) 

362 

363 return conversations_pb2.ListGroupChatsRes( 

364 group_chats=[ 

365 conversations_pb2.GroupChat( 

366 group_chat_id=result.GroupChat.conversation_id, 

367 title=result.GroupChat.title, # TODO: proper title for DMs, etc 

368 member_user_ids=_get_visible_members_for_subscription(result.GroupChatSubscription), 

369 admin_user_ids=_get_visible_admins_for_subscription(result.GroupChatSubscription), 

370 only_admins_invite=result.GroupChat.only_admins_invite, 

371 is_dm=result.GroupChat.is_dm, 

372 created=Timestamp_from_datetime(result.GroupChat.conversation.created), 

373 unseen_message_count=unseen_counts.get(result.GroupChatSubscription.id, 0), 

374 last_seen_message_id=result.GroupChatSubscription.last_seen_message_id, 

375 latest_message=message_to_pb(result.Message) if result.Message else None, 

376 mute_info=mute_info(result.GroupChatSubscription), 

377 can_message=_user_can_message(session, context, result.GroupChat), 

378 is_archived=result.GroupChatSubscription.is_archived, 

379 ) 

380 for result in results[:page_size] 

381 ], 

382 last_message_id=( 

383 min(g.Message.id if g.Message else 1 for g in results[:page_size]) if len(results) > 0 else 0 

384 ), # TODO 

385 no_more=len(results) <= page_size, 

386 ) 

387 

388 def GetGroupChat( 

389 self, request: conversations_pb2.GetGroupChatReq, context: CouchersContext, session: Session 

390 ) -> conversations_pb2.GroupChat: 

391 result = session.execute( 

392 where_moderated_content_visible( 

393 select(GroupChat, GroupChatSubscription, Message) 

394 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id) 

395 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id) 

396 .join(Conversation, Conversation.id == GroupChat.conversation_id) 

397 .options(contains_eager(GroupChat.conversation)) 

398 .where(GroupChatSubscription.user_id == context.user_id) 

399 .where(GroupChatSubscription.group_chat_id == request.group_chat_id) 

400 .where(is_newest_subscription(context.user_id)) 

401 .where(was_subscribed_at(GroupChatSubscription, Message.time)) 

402 .order_by(Message.id.desc()) 

403 .limit(1), 

404 context, 

405 GroupChat, 

406 is_list_operation=False, 

407 ) 

408 ).one_or_none() 

409 

410 if not result: 

411 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found") 

412 

413 return conversations_pb2.GroupChat( 

414 group_chat_id=result.GroupChat.conversation_id, 

415 title=result.GroupChat.title, 

416 member_user_ids=_get_visible_members_for_subscription(result.GroupChatSubscription), 

417 admin_user_ids=_get_visible_admins_for_subscription(result.GroupChatSubscription), 

418 only_admins_invite=result.GroupChat.only_admins_invite, 

419 is_dm=result.GroupChat.is_dm, 

420 created=Timestamp_from_datetime(result.GroupChat.conversation.created), 

421 unseen_message_count=_unseen_message_count(session, result.GroupChatSubscription.id), 

422 last_seen_message_id=result.GroupChatSubscription.last_seen_message_id, 

423 latest_message=message_to_pb(result.Message) if result.Message else None, 

424 mute_info=mute_info(result.GroupChatSubscription), 

425 can_message=_user_can_message(session, context, result.GroupChat), 

426 is_archived=result.GroupChatSubscription.is_archived, 

427 ) 

428 

429 def GetDirectMessage( 

430 self, request: conversations_pb2.GetDirectMessageReq, context: CouchersContext, session: Session 

431 ) -> conversations_pb2.GroupChat: 

432 count = func.count(GroupChatSubscription.id).label("count") 

433 subquery = ( 

434 select(GroupChatSubscription.group_chat_id) 

435 .where( 

436 or_( 

437 GroupChatSubscription.user_id == context.user_id, 

438 GroupChatSubscription.user_id == request.user_id, 

439 ) 

440 ) 

441 .where(GroupChatSubscription.left == None) 

442 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id) 

443 .where(GroupChat.is_dm == True) 

444 .group_by(GroupChatSubscription.group_chat_id) 

445 .having(count == 2) 

446 .subquery() 

447 ) 

448 

449 result = session.execute( 

450 where_moderated_content_visible( 

451 select(subquery, GroupChat, GroupChatSubscription, Message) 

452 .join(subquery, subquery.c.group_chat_id == GroupChat.conversation_id) 

453 .join(Message, Message.conversation_id == GroupChat.conversation_id) 

454 .join(Conversation, Conversation.id == GroupChat.conversation_id) 

455 .options(contains_eager(GroupChat.conversation)) 

456 .where(GroupChatSubscription.user_id == context.user_id) 

457 .where(GroupChatSubscription.group_chat_id == GroupChat.conversation_id) 

458 .where(is_newest_subscription(context.user_id)) 

459 .where(was_subscribed_at(GroupChatSubscription, Message.time)) 

460 .order_by(Message.id.desc()) 

461 .limit(1), 

462 context, 

463 GroupChat, 

464 is_list_operation=False, 

465 ) 

466 ).one_or_none() 

467 

468 if not result: 

469 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found") 

470 

471 return conversations_pb2.GroupChat( 

472 group_chat_id=result.GroupChat.conversation_id, 

473 title=result.GroupChat.title, 

474 member_user_ids=_get_visible_members_for_subscription(result.GroupChatSubscription), 

475 admin_user_ids=_get_visible_admins_for_subscription(result.GroupChatSubscription), 

476 only_admins_invite=result.GroupChat.only_admins_invite, 

477 is_dm=result.GroupChat.is_dm, 

478 created=Timestamp_from_datetime(result.GroupChat.conversation.created), 

479 unseen_message_count=_unseen_message_count(session, result.GroupChatSubscription.id), 

480 last_seen_message_id=result.GroupChatSubscription.last_seen_message_id, 

481 latest_message=message_to_pb(result.Message) if result.Message else None, 

482 mute_info=mute_info(result.GroupChatSubscription), 

483 can_message=_user_can_message(session, context, result.GroupChat), 

484 is_archived=result.GroupChatSubscription.is_archived, 

485 ) 

486 

487 def GetUpdates( 

488 self, request: conversations_pb2.GetUpdatesReq, context: CouchersContext, session: Session 

489 ) -> conversations_pb2.GetUpdatesRes: 

490 results = ( 

491 session.execute( 

492 where_moderated_content_visible( 

493 select(Message) 

494 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id) 

495 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id) 

496 .where(GroupChatSubscription.user_id == context.user_id) 

497 .where(was_subscribed_at(GroupChatSubscription, Message.time)) 

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

499 .order_by(Message.id.asc()) 

500 .limit(DEFAULT_PAGINATION_LENGTH + 1), 

501 context, 

502 GroupChat, 

503 is_list_operation=False, 

504 ) 

505 ) 

506 .scalars() 

507 .all() 

508 ) 

509 

510 return conversations_pb2.GetUpdatesRes( 

511 updates=[ 

512 conversations_pb2.Update( 

513 group_chat_id=message.conversation_id, 

514 message=message_to_pb(message), 

515 ) 

516 for message in sorted(results, key=lambda message: message.id)[:DEFAULT_PAGINATION_LENGTH] 

517 ], 

518 no_more=len(results) <= DEFAULT_PAGINATION_LENGTH, 

519 ) 

520 

521 def GetGroupChatMessages( 

522 self, request: conversations_pb2.GetGroupChatMessagesReq, context: CouchersContext, session: Session 

523 ) -> conversations_pb2.GetGroupChatMessagesRes: 

524 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH 

525 page_size = min(page_size, MAX_PAGE_SIZE) 

526 

527 results = ( 

528 session.execute( 

529 where_moderated_content_visible( 

530 select(Message) 

531 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id) 

532 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id) 

533 .where(GroupChatSubscription.user_id == context.user_id) 

534 .where(GroupChatSubscription.group_chat_id == request.group_chat_id) 

535 .where(was_subscribed_at(GroupChatSubscription, Message.time)) 

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

537 .where( 

538 or_( 

539 and_( 

540 is_newest_subscription(context.user_id), 

541 is_unseen(Message, GroupChatSubscription), 

542 ), 

543 to_bool(request.only_unseen == 0), 

544 ) 

545 ) 

546 .order_by(Message.id.desc()) 

547 .limit(page_size + 1), 

548 context, 

549 GroupChat, 

550 is_list_operation=False, 

551 ) 

552 ) 

553 .scalars() 

554 .all() 

555 ) 

556 

557 return conversations_pb2.GetGroupChatMessagesRes( 

558 messages=[message_to_pb(message) for message in results[:page_size]], 

559 last_message_id=results[-2].id if len(results) > 1 else 0, # TODO 

560 no_more=len(results) <= page_size, 

561 ) 

562 

563 def MarkLastSeenGroupChat( 

564 self, request: conversations_pb2.MarkLastSeenGroupChatReq, context: CouchersContext, session: Session 

565 ) -> empty_pb2.Empty: 

566 subscription = _get_visible_message_subscription(session, context, request.group_chat_id, include_left=True) 

567 

568 if not subscription: 568 ↛ 569line 568 didn't jump to line 569 because the condition on line 568 was never true

569 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found") 

570 

571 if not subscription.last_seen_message_id <= request.last_seen_message_id: 571 ↛ 572line 571 didn't jump to line 572 because the condition on line 571 was never true

572 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_unsee_messages") 

573 

574 subscription.last_seen_message_id = request.last_seen_message_id 

575 

576 mark_notifications_seen( 

577 session, 

578 user_id=context.user_id, 

579 topic_actions_and_keys=[ 

580 ([NotificationTopicAction.chat__message], [str(request.group_chat_id)]), 

581 # chat__missed_messages is a summary across all chats, so it's keyed with an empty string 

582 # rather than a chat id: reading any chat counts as acting on it, and it gets marked seen 

583 ([NotificationTopicAction.chat__missed_messages], [""]), 

584 ], 

585 ) 

586 

587 return empty_pb2.Empty() 

588 

589 def MuteGroupChat( 

590 self, request: conversations_pb2.MuteGroupChatReq, context: CouchersContext, session: Session 

591 ) -> empty_pb2.Empty: 

592 subscription = _get_visible_message_subscription(session, context, request.group_chat_id) 

593 

594 if not subscription: 594 ↛ 595line 594 didn't jump to line 595 because the condition on line 594 was never true

595 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found") 

596 

597 if request.unmute: 

598 subscription.muted_until = DATETIME_MINUS_INFINITY 

599 elif request.forever: 

600 subscription.muted_until = DATETIME_INFINITY 

601 elif request.for_duration: 601 ↛ 607line 601 didn't jump to line 607 because the condition on line 601 was always true

602 duration = request.for_duration.ToTimedelta() 

603 if duration < timedelta(seconds=0): 603 ↛ 604line 603 didn't jump to line 604 because the condition on line 603 was never true

604 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cant_mute_past") 

605 subscription.muted_until = now() + duration 

606 

607 return empty_pb2.Empty() 

608 

609 def SetGroupChatArchiveStatus( 

610 self, request: conversations_pb2.SetGroupChatArchiveStatusReq, context: CouchersContext, session: Session 

611 ) -> conversations_pb2.SetGroupChatArchiveStatusRes: 

612 subscription = _get_visible_message_subscription(session, context, request.group_chat_id) 

613 

614 if not subscription: 

615 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found") 

616 

617 subscription.is_archived = request.is_archived 

618 

619 return conversations_pb2.SetGroupChatArchiveStatusRes( 

620 group_chat_id=request.group_chat_id, 

621 is_archived=request.is_archived, 

622 ) 

623 

624 def SearchMessages( 

625 self, request: conversations_pb2.SearchMessagesReq, context: CouchersContext, session: Session 

626 ) -> conversations_pb2.SearchMessagesRes: 

627 page_size = request.number if request.number != 0 else DEFAULT_PAGINATION_LENGTH 

628 page_size = min(page_size, MAX_PAGE_SIZE) 

629 

630 results = ( 

631 session.execute( 

632 where_moderated_content_visible( 

633 select(Message) 

634 .join(GroupChatSubscription, GroupChatSubscription.group_chat_id == Message.conversation_id) 

635 .join(GroupChat, GroupChat.conversation_id == Message.conversation_id) 

636 .where(GroupChatSubscription.user_id == context.user_id) 

637 .where(was_subscribed_at(GroupChatSubscription, Message.time)) 

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

639 .where(Message.text.ilike(f"%{request.query}%")) 

640 .order_by(Message.id.desc()) 

641 .limit(page_size + 1), 

642 context, 

643 GroupChat, 

644 is_list_operation=True, 

645 ) 

646 ) 

647 .scalars() 

648 .all() 

649 ) 

650 

651 return conversations_pb2.SearchMessagesRes( 

652 results=[ 

653 conversations_pb2.MessageSearchResult( 

654 group_chat_id=message.conversation_id, 

655 message=message_to_pb(message), 

656 ) 

657 for message in results[:page_size] 

658 ], 

659 last_message_id=results[-2].id if len(results) > 1 else 0, 

660 no_more=len(results) <= page_size, 

661 ) 

662 

663 def CreateGroupChat( 

664 self, request: conversations_pb2.CreateGroupChatReq, context: CouchersContext, session: Session 

665 ) -> conversations_pb2.GroupChat: 

666 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

667 if not has_completed_profile(session, user): 

668 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_send_message") 

669 

670 recipient_user_ids = list( 

671 session.execute( 

672 select(User.id).where(users_visible(context)).where(User.id.in_(request.recipient_user_ids)) 

673 ) 

674 .scalars() 

675 .all() 

676 ) 

677 

678 # make sure all requested users are visible 

679 if len(recipient_user_ids) != len(request.recipient_user_ids): 679 ↛ 680line 679 didn't jump to line 680 because the condition on line 679 was never true

680 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_not_found") 

681 

682 if not recipient_user_ids: 682 ↛ 683line 682 didn't jump to line 683 because the condition on line 682 was never true

683 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "no_recipients") 

684 

685 if len(recipient_user_ids) != len(set(recipient_user_ids)): 685 ↛ 687line 685 didn't jump to line 687 because the condition on line 685 was never true

686 # make sure there's no duplicate users 

687 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_recipients") 

688 

689 if context.user_id in recipient_user_ids: 689 ↛ 690line 689 didn't jump to line 690 because the condition on line 689 was never true

690 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_add_self") 

691 

692 if len(recipient_user_ids) == 1: 

693 # can only have one DM at a time between any two users 

694 other_user_id = recipient_user_ids[0] 

695 

696 # the following sql statement selects subscriptions that are DMs and have the same group_chat_id, and have 

697 # user_id either this user or the recipient user. If you find two subscriptions to the same DM group 

698 # chat, you know they already have a shared group chat 

699 count = func.count(GroupChatSubscription.id).label("count") 

700 if session.execute( 

701 where_moderated_content_visible( 

702 select(count) 

703 .where( 

704 or_( 

705 GroupChatSubscription.user_id == context.user_id, 

706 GroupChatSubscription.user_id == other_user_id, 

707 ) 

708 ) 

709 .where(GroupChatSubscription.left == None) 

710 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id) 

711 .where(GroupChat.is_dm == True) 

712 .group_by(GroupChatSubscription.group_chat_id) 

713 .having(count == 2), 

714 context, 

715 GroupChat, 

716 is_list_operation=False, 

717 ) 

718 ).scalar_one_or_none(): 

719 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "already_have_dm") 

720 

721 # Check if user has been initiating chats excessively 

722 if process_rate_limits_and_check_abort( 

723 session=session, user_id=context.user_id, action=RateLimitAction.chat_initiation 

724 ): 

725 context.abort_with_error_code( 

726 grpc.StatusCode.RESOURCE_EXHAUSTED, 

727 "chat_initiation_rate_limit2", 

728 substitutions={"count": RATE_LIMIT_HOURS}, 

729 ) 

730 

731 group_chat = _create_chat( 

732 session, 

733 creator_id=context.user_id, 

734 recipient_ids=request.recipient_user_ids, 

735 title=request.title.value, 

736 ) 

737 

738 your_subscription = _get_message_subscription(session, context.user_id, group_chat.conversation_id) 

739 

740 _add_message_to_subscription(session, your_subscription, message_type=MessageType.chat_created)