Coverage for app/backend/src/couchers/servicers/notifications.py: 90%
132 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import functools
2import json
3import logging
5import grpc
6from google.protobuf import empty_pb2
7from sqlalchemy import select, update
8from sqlalchemy.orm import Session
9from sqlalchemy.sql import or_
11from couchers.config import config
12from couchers.constants import DATETIME_INFINITY
13from couchers.context import CouchersContext
14from couchers.helpers.hosting_meetup_status import record_hosting_meetup_status
15from couchers.i18n import LocalizationContext
16from couchers.models import (
17 DeviceType,
18 HostingMeetupStatusSource,
19 HostingStatus,
20 MeetupStatus,
21 Notification,
22 NotificationDeliveryType,
23 PushNotificationPlatform,
24 PushNotificationSubscription,
25 User,
26)
27from couchers.notifications.push import PushNotificationContent, push_to_subscription, push_to_user
28from couchers.notifications.render_push import render_adhoc_push_notification, render_push_notification
29from couchers.notifications.send_raw_push_notification import is_known_invalid_endpoint
30from couchers.notifications.settings import (
31 PreferenceNotUserEditableError,
32 get_topic_actions_by_delivery_type,
33 get_user_setting_groups,
34 set_preference,
35)
36from couchers.notifications.utils import enum_from_topic_action
37from couchers.notifications.web_push_api import decode_key, get_vapid_public_key_from_private_key
38from couchers.proto import notifications_pb2, notifications_pb2_grpc
39from couchers.sql import moderation_state_column_visible, to_bool
40from couchers.utils import Timestamp_from_datetime, now
42logger = logging.getLogger(__name__)
43MAX_PAGINATION_LENGTH = 100
46@functools.cache
47def get_vapid_public_key() -> str:
48 return get_vapid_public_key_from_private_key(config.PUSH_NOTIFICATIONS_VAPID_PRIVATE_KEY)
51def notification_to_pb(user: User, notification: Notification) -> notifications_pb2.Notification:
52 content = render_push_notification(notification, LocalizationContext.from_user(user))
53 return notifications_pb2.Notification(
54 notification_id=notification.id,
55 created=Timestamp_from_datetime(notification.created),
56 topic=notification.topic_action.topic,
57 action=notification.topic_action.action,
58 key=notification.key,
59 title=content.title,
60 body=content.body,
61 icon=content.icon_url,
62 url=content.action_url,
63 is_seen=notification.is_seen,
64 )
67class Notifications(notifications_pb2_grpc.NotificationsServicer):
68 def GetNotificationSettings(
69 self, request: notifications_pb2.GetNotificationSettingsReq, context: CouchersContext, session: Session
70 ) -> notifications_pb2.GetNotificationSettingsRes:
71 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
72 return notifications_pb2.GetNotificationSettingsRes(
73 do_not_email_enabled=user.do_not_email,
74 groups=get_user_setting_groups(user.id, context.localization),
75 )
77 def SetNotificationSettings(
78 self, request: notifications_pb2.SetNotificationSettingsReq, context: CouchersContext, session: Session
79 ) -> notifications_pb2.GetNotificationSettingsRes:
80 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
81 user.do_not_email = request.enable_do_not_email
82 if request.enable_do_not_email:
83 user.hosting_status = HostingStatus.cant_host
84 user.meetup_status = MeetupStatus.does_not_want_to_meetup
85 record_hosting_meetup_status(session, user, HostingMeetupStatusSource.do_not_email)
86 for preference in request.preferences:
87 topic_action = enum_from_topic_action.get((preference.topic, preference.action), None)
88 if not topic_action: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true
89 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "invalid_notification_preference")
90 delivery_types = {t.name for t in NotificationDeliveryType}
91 if preference.delivery_method not in delivery_types: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true
92 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "invalid_delivery_method")
93 delivery_type = NotificationDeliveryType[preference.delivery_method]
94 try:
95 set_preference(session, user.id, topic_action, delivery_type, preference.enabled)
96 except PreferenceNotUserEditableError:
97 context.abort_with_error_code(
98 grpc.StatusCode.FAILED_PRECONDITION, "cannot_edit_that_notification_preference"
99 )
100 return notifications_pb2.GetNotificationSettingsRes(
101 do_not_email_enabled=user.do_not_email,
102 groups=get_user_setting_groups(user.id, context.localization),
103 )
105 def ListNotifications(
106 self, request: notifications_pb2.ListNotificationsReq, context: CouchersContext, session: Session
107 ) -> notifications_pb2.ListNotificationsRes:
108 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
109 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
110 next_notification_id = int(request.page_token) if request.page_token else 2**50
111 notifications = (
112 session.execute(
113 select(Notification)
114 .where(Notification.user_id == context.user_id)
115 .where(Notification.id <= next_notification_id)
116 .where(or_(to_bool(request.only_unread == False), Notification.is_seen == False))
117 .where(
118 Notification.topic_action.in_(
119 get_topic_actions_by_delivery_type(session, user.id, NotificationDeliveryType.push)
120 )
121 )
122 .where(moderation_state_column_visible(context, Notification.moderation_state_id))
123 .order_by(Notification.id.desc())
124 .limit(page_size + 1)
125 )
126 .scalars()
127 .all()
128 )
129 return notifications_pb2.ListNotificationsRes(
130 notifications=[notification_to_pb(user, notification) for notification in notifications[:page_size]],
131 next_page_token=str(notifications[-1].id) if len(notifications) > page_size else None,
132 )
134 def MarkNotificationSeen(
135 self, request: notifications_pb2.MarkNotificationSeenReq, context: CouchersContext, session: Session
136 ) -> empty_pb2.Empty:
137 notification = (
138 session.execute(
139 select(Notification)
140 .where(Notification.user_id == context.user_id)
141 .where(Notification.id == request.notification_id)
142 )
143 .scalars()
144 .one_or_none()
145 )
146 if not notification: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true
147 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "notification_not_found")
148 notification.is_seen = request.set_seen
149 return empty_pb2.Empty()
151 def MarkAllNotificationsSeen(
152 self, request: notifications_pb2.MarkAllNotificationsSeenReq, context: CouchersContext, session: Session
153 ) -> empty_pb2.Empty:
154 session.execute(
155 update(Notification)
156 .values(is_seen=True)
157 .where(Notification.user_id == context.user_id)
158 .where(Notification.id <= request.latest_notification_id)
159 )
160 return empty_pb2.Empty()
162 def GetVapidPublicKey(
163 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
164 ) -> notifications_pb2.GetVapidPublicKeyRes:
165 if not config.PUSH_NOTIFICATIONS_ENABLED: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true
166 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "push_notifications_disabled")
168 return notifications_pb2.GetVapidPublicKeyRes(vapid_public_key=get_vapid_public_key())
170 def RegisterPushNotificationSubscription(
171 self,
172 request: notifications_pb2.RegisterPushNotificationSubscriptionReq,
173 context: CouchersContext,
174 session: Session,
175 ) -> empty_pb2.Empty:
176 if not config.PUSH_NOTIFICATIONS_ENABLED: 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true
177 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "push_notifications_disabled")
179 data = json.loads(request.full_subscription_json)
180 if is_known_invalid_endpoint(data["endpoint"]):
181 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_endpoint")
183 subscription = PushNotificationSubscription(
184 user_id=context.user_id,
185 platform=PushNotificationPlatform.web_push,
186 endpoint=data["endpoint"],
187 p256dh_key=decode_key(data["keys"]["p256dh"]),
188 auth_key=decode_key(data["keys"]["auth"]),
189 full_subscription_info=request.full_subscription_json,
190 user_agent=request.user_agent,
191 )
192 session.add(subscription)
193 session.flush()
194 push_to_subscription(
195 session,
196 push_notification_subscription_id=subscription.id,
197 user_id=context.user_id,
198 topic_action="adhoc:setup",
199 content=PushNotificationContent(
200 title="Push notifications test",
201 ios_title="Push Notifications Test",
202 body="Hi, thanks for enabling push notifications!",
203 ),
204 )
206 return empty_pb2.Empty()
208 def SendTestPushNotification(
209 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
210 ) -> empty_pb2.Empty:
211 if not config.PUSH_NOTIFICATIONS_ENABLED: 211 ↛ 212line 211 didn't jump to line 212 because the condition on line 211 was never true
212 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "push_notifications_disabled")
214 push_to_user(
215 session,
216 user_id=context.user_id,
217 topic_action="adhoc:testing",
218 content=PushNotificationContent(
219 title="Push notifications test",
220 ios_title="Push Notifications Test",
221 body="If you see this, then it's working :)",
222 ),
223 )
225 return empty_pb2.Empty()
227 def RegisterMobilePushNotificationSubscription(
228 self,
229 request: notifications_pb2.RegisterMobilePushNotificationSubscriptionReq,
230 context: CouchersContext,
231 session: Session,
232 ) -> empty_pb2.Empty:
233 if not config.PUSH_NOTIFICATIONS_ENABLED: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "push_notifications_disabled")
236 # Check for existing subscription with this token
237 existing = session.execute(
238 select(PushNotificationSubscription).where(PushNotificationSubscription.token == request.token)
239 ).scalar_one_or_none()
241 if existing:
242 # Re-enable if disabled
243 if existing.disabled_at < now():
244 existing.disabled_at = DATETIME_INFINITY
245 existing.device_name = request.device_name or existing.device_name
246 if request.device_type: 246 ↛ 248line 246 didn't jump to line 248 because the condition on line 246 was always true
247 existing.device_type = DeviceType[request.device_type]
248 logger.info(f"Re-enabled mobile push sub {existing.id} for user {context.user_id}")
249 return empty_pb2.Empty()
251 # Parse device_type if provided
252 device_type = DeviceType[request.device_type] if request.device_type else None
254 subscription = PushNotificationSubscription(
255 user_id=context.user_id,
256 platform=PushNotificationPlatform.expo,
257 token=request.token,
258 device_name=request.device_name if request.device_name else None,
259 device_type=device_type,
260 )
261 session.add(subscription)
262 session.flush()
264 push_content = render_adhoc_push_notification("push_enabled", context.localization)
265 push_to_subscription(
266 session,
267 push_notification_subscription_id=subscription.id,
268 user_id=context.user_id,
269 topic_action="adhoc:push_enabled",
270 content=push_content,
271 )
273 return empty_pb2.Empty()
275 def SendTestMobilePushNotification(
276 self, request: empty_pb2.Empty, context: CouchersContext, session: Session
277 ) -> empty_pb2.Empty:
278 if not config.PUSH_NOTIFICATIONS_ENABLED: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true
279 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "push_notifications_disabled")
281 push_to_user(
282 session,
283 user_id=context.user_id,
284 topic_action="adhoc:testing",
285 content=PushNotificationContent(
286 title="Mobile notifications test",
287 ios_title="Mobile Notifications Test",
288 body="If you see this on your phone, everything is wired up correctly 🎉",
289 ),
290 )
292 return empty_pb2.Empty()
294 def SendDevPushNotification(
295 self, request: notifications_pb2.SendDevPushNotificationReq, context: CouchersContext, session: Session
296 ) -> empty_pb2.Empty:
297 if not config.ENABLE_DEV_APIS:
298 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "dev_apis_disabled")
300 if not config.PUSH_NOTIFICATIONS_ENABLED:
301 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "push_notifications_disabled")
303 push_to_user(
304 session,
305 user_id=context.user_id,
306 topic_action="adhoc:testing",
307 content=PushNotificationContent(
308 title=request.title,
309 ios_title=request.title,
310 body=request.body,
311 action_url=request.url or None,
312 icon_url=request.icon or None,
313 ),
314 key=request.key or None,
315 ttl=request.ttl,
316 )
318 return empty_pb2.Empty()
320 def DebugRedeliverPushNotification(
321 self,
322 request: notifications_pb2.DebugRedeliverPushNotificationReq,
323 context: CouchersContext,
324 session: Session,
325 ) -> empty_pb2.Empty:
326 if not config.ENABLE_DEV_APIS:
327 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "dev_apis_disabled")
329 if not config.PUSH_NOTIFICATIONS_ENABLED:
330 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "push_notifications_disabled")
332 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
334 notification = session.execute(
335 select(Notification)
336 .where(Notification.id == request.notification_id)
337 .where(Notification.user_id == context.user_id)
338 ).scalar_one_or_none()
340 if not notification:
341 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "notification_not_found")
343 push_to_user(
344 session,
345 user_id=context.user_id,
346 topic_action=notification.topic_action.display,
347 content=render_push_notification(notification, LocalizationContext.from_user(user)),
348 key=notification.key,
349 )
351 return empty_pb2.Empty()