Coverage for app/backend/src/couchers/helpers/group_chats.py: 100%

1""" 

2Rules for scoping a user's group chat messages, shared by the conversations API and the Ping badge. 

3""" 

4 

5from datetime import datetime 

6 

7from sqlalchemy import ColumnElement, SQLColumnExpression, and_, func, or_, select 

8 

9from couchers.models import GroupChatSubscription, Message 

10 

11 

12def was_subscribed_at( 

13 subscription: type[GroupChatSubscription], instant: SQLColumnExpression[datetime] | datetime 

14) -> ColumnElement[bool]: 

15 """ 

16 In the chat at the given instant: joined before it and hadn't left yet, or left after it. 

17 """ 

18 return and_( 

19 subscription.joined <= instant, 

20 or_(subscription.left == None, subscription.left >= instant), 

21 ) 

22 

23 

24def is_unseen(message: type[Message], subscription: type[GroupChatSubscription]) -> ColumnElement[bool]: 

25 """ 

26 Unseen by the subscriber and still within their reach: a message they can never open, because it 

27 was sent while they were out of the chat, is not unread. 

28 """ 

29 return and_( 

30 was_subscribed_at(subscription, message.time), 

31 message.id > subscription.last_seen_message_id, 

32 ) 

33 

34 

35def is_newest_subscription(user_id: SQLColumnExpression[int] | int) -> ColumnElement[bool]: 

36 """ 

37 Only the user's newest subscription to each chat they've been in. Rejoining a chat leaves the 

38 earlier subscription behind, and it's the newest one that carries the archived and last-seen state. 

39 """ 

40 newest_per_chat = ( 

41 select(func.max(GroupChatSubscription.id)) 

42 .where(GroupChatSubscription.user_id == user_id) 

43 .group_by(GroupChatSubscription.group_chat_id) 

44 # not aliased(): rebuilding the ORM proxy index per call is a hotspot, and Ping runs this on 

45 # every poll. correlate_except keeps this table local to the subquery so it doesn't bind to 

46 # the enclosing query's GroupChatSubscription 

47 .correlate_except(GroupChatSubscription) 

48 ) 

49 return GroupChatSubscription.id.in_(newest_per_chat)