Coverage for app/backend/src/couchers/jobs/handlers.py: 88%
497 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +0000
1"""
2Background job servicers
3"""
5import logging
6from collections.abc import Sequence
7from datetime import date, timedelta
8from math import cos, pi, sin, sqrt
9from random import sample
10from typing import Any
12import requests
13from google.protobuf import empty_pb2
14from sqlalchemy import ColumnElement, Float, Function, Integer, select
15from sqlalchemy.orm import InstrumentedAttribute, aliased
16from sqlalchemy.sql import (
17 and_,
18 case,
19 cast,
20 delete,
21 distinct,
22 exists,
23 extract,
24 func,
25 literal,
26 not_,
27 or_,
28 union_all,
29 update,
30)
32from couchers import experimentation
33from couchers.config import config
34from couchers.constants import (
35 ACTIVENESS_PROBE_EXPIRY_TIME,
36 ACTIVENESS_PROBE_INACTIVITY_PERIOD,
37 ACTIVENESS_PROBE_TIME_REMINDERS,
38 EVENT_REMINDER_TIMEDELTA,
39 HOST_REQUEST_MAX_REMINDERS,
40 HOST_REQUEST_REMINDER_INTERVAL,
41 MODERATION_AUTO_APPROVE_FLAG_PRIORITY,
42)
43from couchers.context import make_background_user_context, make_notification_user_context
44from couchers.crypto import (
45 USER_LOCATION_RANDOMIZATION_NAME,
46 asym_encrypt,
47 b64decode,
48 get_secret,
49 simple_decrypt,
50 stable_secure_uniform,
51)
52from couchers.db import session_scope
53from couchers.email.dev import print_dev_email
54from couchers.email.smtp import send_smtp_email
55from couchers.event_log import log_event
56from couchers.helpers.badges import user_add_badge, user_remove_badge
57from couchers.helpers.completed_profile import has_completed_profile_expression
58from couchers.materialized_views import (
59 UserResponseRate,
60)
61from couchers.metrics import (
62 moderation_auto_approved_counter,
63 postcards_sent_counter,
64 push_notification_counter,
65 strong_verification_completions_counter,
66)
67from couchers.models import (
68 AccountDeletionToken,
69 ActivenessProbe,
70 ActivenessProbeStatus,
71 Cluster,
72 ClusterRole,
73 ClusterSubscription,
74 EventOccurrence,
75 EventOccurrenceAttendee,
76 GroupChat,
77 GroupChatSubscription,
78 HostingStatus,
79 HostRequest,
80 HostRequestStatus,
81 LoginToken,
82 MeetupStatus,
83 Message,
84 MessageType,
85 ModerationAction,
86 ModerationLog,
87 ModerationObjectType,
88 ModerationQueueItem,
89 ModerationState,
90 ModerationTrigger,
91 PassportSex,
92 PasswordResetToken,
93 PhotoGallery,
94 PostalVerificationAttempt,
95 PostalVerificationStatus,
96 PushNotificationDeliveryAttempt,
97 PushNotificationSubscription,
98 Reference,
99 StrongVerificationAttempt,
100 StrongVerificationAttemptStatus,
101 User,
102 UserBadge,
103 Volunteer,
104)
105from couchers.models.notifications import NotificationTopicAction
106from couchers.notifications.expo_api import get_expo_push_receipts
107from couchers.notifications.notify import notify
108from couchers.postal.my_postcard import get_order_ids, send_postcard
109from couchers.proto import moderation_pb2, notification_data_pb2
110from couchers.proto.internal import internal_pb2, jobs_pb2
111from couchers.resources import get_badge_dict, get_static_badge_dict
112from couchers.sentry import report_message
113from couchers.servicers.api import user_model_to_pb
114from couchers.servicers.events import (
115 event_to_pb,
116)
117from couchers.servicers.moderation import Moderation
118from couchers.servicers.requests import host_request_to_pb
119from couchers.sql import (
120 users_visible_to_each_other,
121 where_moderated_content_visible,
122 where_moderated_content_visible_to_user_column,
123 where_user_columns_visible_to_each_other,
124 where_users_column_visible,
125)
126from couchers.tasks import enforce_community_memberships as tasks_enforce_community_memberships
127from couchers.tasks import send_duplicate_strong_verification_email
128from couchers.utils import (
129 Timestamp_from_datetime,
130 create_coordinate,
131 get_coordinates,
132 not_none,
133 now,
134)
136logger = logging.getLogger(__name__)
139def send_email(payload: jobs_pb2.SendEmailPayload) -> None:
140 logger.info(f"Sending email with subject '{payload.subject}' to '{payload.recipient}'")
141 # selects a "sender", which either prints the email to the logger or sends it out with SMTP
142 sender = send_smtp_email if config.ENABLE_EMAIL else print_dev_email
143 # the sender must return a models.Email object that can be added to the database
144 email = sender(payload)
145 with session_scope() as session:
146 session.add(email)
149def purge_login_tokens(payload: empty_pb2.Empty) -> None:
150 logger.info("Purging login tokens")
151 with session_scope() as session:
152 session.execute(delete(LoginToken).where(~LoginToken.is_valid).execution_options(synchronize_session=False))
155def purge_password_reset_tokens(payload: empty_pb2.Empty) -> None:
156 logger.info("Purging login tokens")
157 with session_scope() as session:
158 session.execute(
159 delete(PasswordResetToken).where(~PasswordResetToken.is_valid).execution_options(synchronize_session=False)
160 )
163def purge_account_deletion_tokens(payload: empty_pb2.Empty) -> None:
164 logger.info("Purging account deletion tokens")
165 with session_scope() as session:
166 session.execute(
167 delete(AccountDeletionToken)
168 .where(~AccountDeletionToken.is_valid)
169 .execution_options(synchronize_session=False)
170 )
173# how long a message must go unseen before we email the user about it
174MISSED_MESSAGES_DELAY = timedelta(minutes=5)
175# ... unless we could reach them by push, in which case they've already been told about it once
176MISSED_MESSAGES_DELAY_WITH_PUSH = timedelta(hours=24)
179def _message_unseen_long_enough(user_id_column: InstrumentedAttribute[int]) -> ColumnElement[bool]:
180 """
181 Whether `Message` has gone unseen long enough to email the given user about it.
183 Note that a subscription we can still push to is not proof the user sees those pushes: they may
184 have revoked notification permission in their OS settings without the token going stale. Such
185 users still get the email, just a day late.
186 """
187 has_active_push_subscription = (
188 select(PushNotificationSubscription.id)
189 .where(PushNotificationSubscription.user_id == user_id_column)
190 .where(PushNotificationSubscription.disabled_at > func.now())
191 .exists()
192 )
193 return Message.time < case(
194 (has_active_push_subscription, now() - MISSED_MESSAGES_DELAY_WITH_PUSH),
195 else_=now() - MISSED_MESSAGES_DELAY,
196 )
199def send_message_notifications(payload: empty_pb2.Empty) -> None:
200 """
201 Sends out email notifications for messages that have been unseen for a long enough time
202 """
203 # very crude and dumb algorithm
204 logger.info("Sending out email notifications for unseen messages")
206 with session_scope() as session:
207 # users who have unnotified messages older than 5 minutes in any group chat
208 users = (
209 session.execute(
210 where_moderated_content_visible_to_user_column(
211 select(User)
212 .join(GroupChatSubscription, GroupChatSubscription.user_id == User.id)
213 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
214 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id),
215 GroupChat,
216 User.id,
217 )
218 .where(not_(GroupChatSubscription.is_muted))
219 .where(User.is_visible)
220 .where(Message.time >= GroupChatSubscription.joined)
221 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
222 .where(Message.id > User.last_notified_message_id)
223 .where(Message.id > GroupChatSubscription.last_seen_message_id)
224 .where(_message_unseen_long_enough(User.id))
225 .where(Message.message_type == MessageType.text) # TODO: only text messages for now
226 )
227 .scalars()
228 .unique()
229 )
231 for user in users:
232 context = make_notification_user_context(user_id=user.id)
233 # now actually grab all the group chats, not just less than 5 min old
234 subquery = (
235 where_users_column_visible(
236 where_moderated_content_visible(
237 select(
238 GroupChatSubscription.group_chat_id.label("group_chat_id"),
239 func.max(GroupChatSubscription.id).label("group_chat_subscriptions_id"),
240 func.max(Message.id).label("message_id"),
241 func.count(Message.id).label("unseen_count"),
242 )
243 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
244 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id),
245 context,
246 GroupChat,
247 is_list_operation=True,
248 )
249 .where(GroupChatSubscription.user_id == user.id)
250 .where(not_(GroupChatSubscription.is_muted))
251 .where(Message.id > user.last_notified_message_id)
252 .where(Message.id > GroupChatSubscription.last_seen_message_id)
253 .where(Message.time >= GroupChatSubscription.joined)
254 .where(Message.message_type == MessageType.text) # TODO: only text messages for now
255 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None)),
256 context,
257 Message.author_id,
258 )
259 .group_by(GroupChatSubscription.group_chat_id)
260 .order_by(func.max(Message.id).desc())
261 .subquery()
262 )
264 unseen_messages = session.execute(
265 where_moderated_content_visible(
266 select(GroupChat, Message, subquery.c.unseen_count)
267 .join(subquery, subquery.c.message_id == Message.id)
268 .join(GroupChat, GroupChat.conversation_id == subquery.c.group_chat_id),
269 context,
270 GroupChat,
271 is_list_operation=True,
272 ).order_by(subquery.c.message_id.desc())
273 ).all()
275 if not unseen_messages:
276 continue
278 user.last_notified_message_id = max(message.id for _, message, _ in unseen_messages)
280 notify(
281 session,
282 user_id=user.id,
283 topic_action=NotificationTopicAction.chat__missed_messages,
284 key="",
285 data=notification_data_pb2.ChatMissedMessages(
286 messages=[
287 notification_data_pb2.ChatMessage(
288 author=user_model_to_pb(
289 message.author,
290 session,
291 context,
292 ),
293 text=message.text,
294 group_chat_id=message.conversation_id,
295 group_chat_title=group_chat.title or None,
296 unseen_count=unseen_count,
297 )
298 for group_chat, message, unseen_count in unseen_messages
299 ],
300 ),
301 )
302 session.commit()
305def send_request_notifications(payload: empty_pb2.Empty) -> None:
306 """
307 Sends out email notifications for unseen messages in host requests (as surfer or host)
308 """
309 logger.info("Sending out email notifications for unseen messages in host requests")
311 with session_scope() as session:
312 # Get all candidate users who might have unseen request messages.
313 # Drive from host_requests/messages (selective) rather than scanning all users (expensive).
314 surfer_ids = (
315 select(User.id)
316 .join(HostRequest, HostRequest.initiator_user_id == User.id)
317 .join(Message, Message.conversation_id == HostRequest.conversation_id)
318 .where(User.is_visible)
319 .where(Message.id > HostRequest.initiator_last_seen_message_id)
320 .where(Message.id > User.last_notified_request_message_id)
321 .where(_message_unseen_long_enough(User.id))
322 .where(Message.message_type == MessageType.text)
323 )
324 host_ids = (
325 select(User.id)
326 .join(HostRequest, HostRequest.recipient_user_id == User.id)
327 .join(Message, Message.conversation_id == HostRequest.conversation_id)
328 .where(User.is_visible)
329 .where(Message.id > HostRequest.recipient_last_seen_message_id)
330 .where(Message.id > User.last_notified_request_message_id)
331 .where(_message_unseen_long_enough(User.id))
332 .where(Message.message_type == MessageType.text)
333 )
334 candidate_user_ids = session.execute(union_all(surfer_ids, host_ids)).scalars().unique().all()
336 for user_id in candidate_user_ids:
337 context = make_notification_user_context(user_id=user_id)
339 # requests where this user is surfing
340 surfing_reqs = session.execute(
341 where_users_column_visible(
342 where_moderated_content_visible_to_user_column(
343 select(User, HostRequest, func.max(Message.id))
344 .where(User.id == user_id)
345 .join(HostRequest, HostRequest.initiator_user_id == User.id),
346 HostRequest,
347 HostRequest.initiator_user_id,
348 ),
349 context,
350 HostRequest.recipient_user_id,
351 )
352 .join(Message, Message.conversation_id == HostRequest.conversation_id)
353 .where(Message.id > HostRequest.initiator_last_seen_message_id)
354 .where(Message.id > User.last_notified_request_message_id)
355 .where(_message_unseen_long_enough(User.id))
356 .where(Message.message_type == MessageType.text)
357 .group_by(User, HostRequest) # type: ignore[arg-type]
358 ).all()
360 # where this user is hosting
361 hosting_reqs = session.execute(
362 where_users_column_visible(
363 where_moderated_content_visible_to_user_column(
364 select(User, HostRequest, func.max(Message.id))
365 .where(User.id == user_id)
366 .join(HostRequest, HostRequest.recipient_user_id == User.id),
367 HostRequest,
368 HostRequest.recipient_user_id,
369 ),
370 context,
371 HostRequest.initiator_user_id,
372 )
373 .join(Message, Message.conversation_id == HostRequest.conversation_id)
374 .where(Message.id > HostRequest.recipient_last_seen_message_id)
375 .where(Message.id > User.last_notified_request_message_id)
376 .where(_message_unseen_long_enough(User.id))
377 .where(Message.message_type == MessageType.text)
378 .group_by(User, HostRequest) # type: ignore[arg-type]
379 ).all()
381 for user, host_request, max_message_id in surfing_reqs:
382 user.last_notified_request_message_id = max(user.last_notified_request_message_id, max_message_id)
383 session.flush()
385 notify(
386 session,
387 user_id=user.id,
388 topic_action=NotificationTopicAction.host_request__missed_messages,
389 key=str(host_request.conversation_id),
390 data=notification_data_pb2.HostRequestMissedMessages(
391 host_request=host_request_to_pb(host_request, session, context),
392 user=user_model_to_pb(host_request.recipient, session, context),
393 am_host=False,
394 ),
395 )
397 for user, host_request, max_message_id in hosting_reqs:
398 user.last_notified_request_message_id = max(user.last_notified_request_message_id, max_message_id)
399 session.flush()
401 # When a host request is created, the recipient immediately receives a
402 # host_request__create notification that includes the initial message text.
403 # A few minutes later, this background job sees that same message as "unseen"
404 # (the recipient hasn't opened the request yet) and would send a duplicate
405 # missed_messages notification.
406 #
407 # To prevent this, we check if the only unseen text message in this host
408 # request is the very first text message in the conversation (i.e. the
409 # creation message). If so, we skip sending missed_messages — the user was
410 # already notified via host_request__create.
411 #
412 # Advancing last_notified_request_message_id above is safe even when we skip
413 # the notification: this watermark is only ever advanced when we process all
414 # unseen messages for the user, so skipping one notification doesn't cause us
415 # to miss future messages in other host requests.
416 only_creation_message = not session.execute(
417 select(
418 select(func.count())
419 .where(Message.conversation_id == host_request.conversation_id)
420 .where(Message.message_type == MessageType.text)
421 .scalar_subquery()
422 > 1
423 )
424 ).scalar_one()
425 if only_creation_message:
426 continue
428 notify(
429 session,
430 user_id=user.id,
431 topic_action=NotificationTopicAction.host_request__missed_messages,
432 key=str(host_request.conversation_id),
433 data=notification_data_pb2.HostRequestMissedMessages(
434 host_request=host_request_to_pb(host_request, session, context),
435 user=user_model_to_pb(host_request.initiator, session, context),
436 am_host=True,
437 ),
438 )
441def send_onboarding_emails(payload: empty_pb2.Empty) -> None:
442 """
443 Sends out onboarding emails
444 """
445 logger.info("Sending out onboarding emails")
447 with session_scope() as session:
448 # first onboarding email
449 users = (
450 session.execute(select(User).where(User.is_visible).where(User.onboarding_emails_sent == 0)).scalars().all()
451 )
453 for user in users:
454 notify(
455 session,
456 user_id=user.id,
457 topic_action=NotificationTopicAction.onboarding__reminder,
458 key="1",
459 )
460 user.onboarding_emails_sent = 1
461 user.last_onboarding_email_sent = now()
462 session.commit()
464 # second onboarding email
465 # sent after a week if the user has no profile or their "about me" section is less than 20 characters long
466 users = (
467 session.execute(
468 select(User)
469 .where(User.is_visible)
470 .where(User.onboarding_emails_sent == 1)
471 .where(now() - User.last_onboarding_email_sent > timedelta(days=7))
472 .where(~has_completed_profile_expression())
473 )
474 .scalars()
475 .all()
476 )
478 for user in users:
479 notify(
480 session,
481 user_id=user.id,
482 topic_action=NotificationTopicAction.onboarding__reminder,
483 key="2",
484 )
485 user.onboarding_emails_sent = 2
486 user.last_onboarding_email_sent = now()
487 session.commit()
490def send_reference_reminders(payload: empty_pb2.Empty) -> None:
491 """
492 Sends out reminders to write references after hosting/staying
493 """
494 logger.info("Sending out reference reminder emails")
496 # Keep this in chronological order!
497 reference_reminder_schedule = [
498 # (number, timedelta before we stop being able to write a ref, text for how long they have left to write the ref)
499 # the end time to write a reference is supposed to be midnight in the host's timezone
500 # 8 pm ish on the last day of the stay
501 (1, timedelta(days=15) - timedelta(hours=20), 14),
502 # 2 pm ish a week after stay
503 (2, timedelta(days=8) - timedelta(hours=14), 7),
504 # 10 am ish 3 days before end of time to write ref
505 (3, timedelta(days=4) - timedelta(hours=10), 3),
506 ]
508 with session_scope() as session:
509 # iterate the reminders in backwards order, so if we missed out on one we don't send duplicates
510 for reminder_number, reminder_time, reminder_days_left in reversed(reference_reminder_schedule):
511 user = aliased(User)
512 other_user = aliased(User)
513 # surfers needing to write a ref
514 q1 = (
515 select(literal(True), HostRequest, user, other_user)
516 .join(user, user.id == HostRequest.initiator_user_id)
517 .join(other_user, other_user.id == HostRequest.recipient_user_id)
518 .outerjoin(
519 Reference,
520 and_(
521 Reference.host_request_id == HostRequest.conversation_id,
522 # if no reference is found in this join, then the surfer has not written a ref
523 Reference.from_user_id == HostRequest.initiator_user_id,
524 ),
525 )
526 .where(Reference.id == None)
527 .where(HostRequest.can_write_reference)
528 .where(HostRequest.initiator_sent_reference_reminders < reminder_number)
529 .where(HostRequest.end_time_to_write_reference - reminder_time < now())
530 .where(HostRequest.initiator_reason_didnt_meetup == None)
531 .where(users_visible_to_each_other(self_user=user, other_user=other_user))
532 )
534 # hosts needing to write a ref
535 q2 = (
536 select(literal(False), HostRequest, user, other_user)
537 .join(user, user.id == HostRequest.recipient_user_id)
538 .join(other_user, other_user.id == HostRequest.initiator_user_id)
539 .outerjoin(
540 Reference,
541 and_(
542 Reference.host_request_id == HostRequest.conversation_id,
543 # if no reference is found in this join, then the host has not written a ref
544 Reference.from_user_id == HostRequest.recipient_user_id,
545 ),
546 )
547 .where(Reference.id == None)
548 .where(HostRequest.can_write_reference)
549 .where(HostRequest.recipient_sent_reference_reminders < reminder_number)
550 .where(HostRequest.end_time_to_write_reference - reminder_time < now())
551 .where(HostRequest.recipient_reason_didnt_meetup == None)
552 .where(users_visible_to_each_other(self_user=user, other_user=other_user))
553 )
555 union = union_all(q1, q2).subquery()
556 query = select(
557 union.c[0].label("surfed"),
558 aliased(HostRequest, union),
559 aliased(user, union),
560 aliased(other_user, union),
561 )
562 reference_reminders = session.execute(query).all()
564 for surfed, host_request, user, other_user in reference_reminders:
565 # visibility and blocking already checked in sql
566 assert user.is_visible
567 context = make_notification_user_context(user_id=user.id)
568 topic_action = (
569 NotificationTopicAction.reference__reminder_surfed
570 if surfed
571 else NotificationTopicAction.reference__reminder_hosted
572 )
573 notify(
574 session,
575 user_id=user.id,
576 topic_action=topic_action,
577 key=str(host_request.conversation_id),
578 data=notification_data_pb2.ReferenceReminder(
579 host_request_id=host_request.conversation_id,
580 other_user=user_model_to_pb(other_user, session, context),
581 days_left=reminder_days_left,
582 ),
583 )
584 if surfed:
585 host_request.initiator_sent_reference_reminders = reminder_number
586 else:
587 host_request.recipient_sent_reference_reminders = reminder_number
588 session.commit()
591def send_host_request_reminders(payload: empty_pb2.Empty) -> None:
592 with session_scope() as session:
593 host_has_sent_message = select(1).where(
594 Message.conversation_id == HostRequest.conversation_id, Message.author_id == HostRequest.recipient_user_id
595 )
597 requests = (
598 session.execute(
599 where_user_columns_visible_to_each_other(
600 where_moderated_content_visible_to_user_column(
601 select(HostRequest),
602 HostRequest,
603 HostRequest.recipient_user_id,
604 )
605 .where(HostRequest.status == HostRequestStatus.pending)
606 .where(HostRequest.recipient_sent_request_reminders < HOST_REQUEST_MAX_REMINDERS)
607 .where(HostRequest.start_time > func.now())
608 .where((func.now() - HostRequest.last_sent_request_reminder_time) >= HOST_REQUEST_REMINDER_INTERVAL)
609 .where(~exists(host_has_sent_message)),
610 self_column=HostRequest.recipient_user_id,
611 other_column=HostRequest.initiator_user_id,
612 )
613 )
614 .scalars()
615 .all()
616 )
618 for host_request in requests:
619 host_request.recipient_sent_request_reminders += 1
620 host_request.last_sent_request_reminder_time = now()
622 context = make_notification_user_context(user_id=host_request.recipient_user_id)
623 notify(
624 session,
625 user_id=host_request.recipient_user_id,
626 topic_action=NotificationTopicAction.host_request__reminder,
627 key=str(host_request.conversation_id),
628 data=notification_data_pb2.HostRequestReminder(
629 host_request=host_request_to_pb(host_request, session, context),
630 surfer=user_model_to_pb(host_request.initiator, session, context),
631 ),
632 moderation_state_id=host_request.moderation_state_id,
633 )
635 session.commit()
638def add_users_to_email_list(payload: empty_pb2.Empty) -> None:
639 if not experimentation.get_global_boolean_value("listmonk_enabled", default=False): 639 ↛ 640line 639 didn't jump to line 640 because the condition on line 639 was never true
640 logger.info("Not adding users to mailing list")
641 return
643 sess = requests.Session()
644 sess.auth = (config.LISTMONK_API_USERNAME, config.LISTMONK_API_KEY)
646 def sync_subscriber(user: User, status: str) -> None:
647 r = sess.post(
648 config.LISTMONK_BASE_URL + "/api/subscribers",
649 json={
650 "email": user.email,
651 "name": user.name,
652 "lists": [config.LISTMONK_LIST_ID],
653 "preconfirm_subscriptions": True,
654 "attribs": {"couchers_user_id": user.id},
655 "status": status,
656 },
657 timeout=10,
658 )
659 # the API returns 409 if the subscriber already exists
660 if r.status_code not in (200, 409): 660 ↛ 661line 660 didn't jump to line 661 because the condition on line 660 was never true
661 raise Exception("Failed to update user mailing list status")
663 logger.info("Adding users to mailing list")
665 while True:
666 with session_scope() as session:
667 user = session.execute(
668 select(User).where(User.is_visible).where(User.in_sync_with_newsletter == False).limit(1)
669 ).scalar_one_or_none()
670 if not user:
671 logger.info("Finished adding users to mailing list")
672 break
674 if not user.opt_out_of_newsletter:
675 sync_subscriber(user, "enabled")
677 user.in_sync_with_newsletter = True
678 session.commit()
680 if experimentation.get_global_boolean_value("remove_removed_users_from_mailing_list_enabled", default=False): 680 ↛ 681line 680 didn't jump to line 681 because the condition on line 680 was never true
681 with session_scope() as session:
682 session.execute(
683 update(User)
684 .where(~User.is_visible | User.is_shadowed)
685 .where(User.opt_out_of_newsletter == False)
686 .values(opt_out_of_newsletter=True, in_sync_with_newsletter=False)
687 )
688 session.commit()
690 while True:
691 with session_scope() as session:
692 user = session.execute(
693 select(User)
694 .where(~User.is_visible | User.is_shadowed)
695 .where(User.in_sync_with_newsletter == False)
696 .limit(1)
697 ).scalar_one_or_none()
698 if not user:
699 logger.info("Finished removing users from mailing list")
700 return
702 sync_subscriber(user, "blocklisted")
703 user.in_sync_with_newsletter = True
704 session.commit()
707def enforce_community_membership(payload: empty_pb2.Empty) -> None:
708 tasks_enforce_community_memberships()
711def update_recommendation_scores(payload: empty_pb2.Empty) -> None:
712 text_fields = [
713 User.hometown,
714 User.occupation,
715 User.education,
716 User.about_me,
717 User.things_i_like,
718 User.about_place,
719 User.additional_information,
720 User.pet_details,
721 User.kid_details,
722 User.housemate_details,
723 User.other_host_info,
724 User.sleeping_details,
725 User.area,
726 User.house_rules,
727 ]
728 home_fields = [User.about_place, User.other_host_info, User.sleeping_details, User.area, User.house_rules]
730 def poor_man_gaussian() -> ColumnElement[float] | float:
731 """
732 Produces an approximatley std normal random variate
733 """
734 trials = 5
735 return (sum([func.random() for _ in range(trials)]) - trials / 2) / sqrt(trials / 12)
737 def int_(stmt: Any) -> Function[int]:
738 return func.coalesce(cast(stmt, Integer), 0)
740 def float_(stmt: Any) -> Function[float]:
741 return func.coalesce(cast(stmt, Float), 0.0)
743 with session_scope() as session:
744 # profile
745 profile_text = ""
746 for field in text_fields:
747 profile_text += func.coalesce(field, "") # type: ignore[assignment]
748 text_length = func.length(profile_text)
749 home_text = ""
750 for field in home_fields:
751 home_text += func.coalesce(field, "") # type: ignore[assignment]
752 home_length = func.length(home_text)
754 filled_profile = int_(has_completed_profile_expression())
755 has_text = int_(text_length > 500)
756 long_text = int_(text_length > 2000)
757 can_host = int_(User.hosting_status == HostingStatus.can_host)
758 may_host = int_(User.hosting_status == HostingStatus.maybe)
759 cant_host = int_(User.hosting_status == HostingStatus.cant_host)
760 filled_home = int_(User.has_completed_my_home)
761 filled_home_lots = int_(home_length > 200)
762 hosting_status_points = 5 * can_host - 5 * may_host - 10 * cant_host
763 profile_points = 5 * filled_profile + 2 * has_text + 3 * long_text + 5 * filled_home + 10 * filled_home_lots
765 # references
766 left_ref_expr = int_(1).label("left_reference")
767 left_refs_subquery = (
768 select(Reference.from_user_id.label("user_id"), left_ref_expr).group_by(Reference.from_user_id).subquery()
769 )
770 left_reference = int_(left_refs_subquery.c.left_reference)
771 has_reference_expr = int_(func.count(Reference.id) >= 1).label("has_reference")
772 ref_count_expr = int_(func.count(Reference.id)).label("ref_count")
773 ref_avg_expr = func.avg(1.4 * (Reference.rating - 0.3)).label("ref_avg")
774 has_multiple_types_expr = int_(func.count(distinct(Reference.reference_type)) >= 2).label("has_multiple_types")
775 has_bad_ref_expr = int_(func.sum(int_((Reference.rating <= 0.2) | (~Reference.was_appropriate))) >= 1).label(
776 "has_bad_ref"
777 )
778 received_ref_subquery = (
779 select(
780 Reference.to_user_id.label("user_id"),
781 has_reference_expr,
782 has_multiple_types_expr,
783 has_bad_ref_expr,
784 ref_count_expr,
785 ref_avg_expr,
786 )
787 .group_by(Reference.to_user_id)
788 .subquery()
789 )
790 has_multiple_types = int_(received_ref_subquery.c.has_multiple_types)
791 has_reference = int_(received_ref_subquery.c.has_reference)
792 has_bad_reference = int_(received_ref_subquery.c.has_bad_ref)
793 rating_score = float_(
794 received_ref_subquery.c.ref_avg
795 * (
796 2 * func.least(received_ref_subquery.c.ref_count, 5)
797 + func.greatest(received_ref_subquery.c.ref_count - 5, 0)
798 )
799 )
800 ref_score = 2 * has_reference + has_multiple_types + left_reference - 5 * has_bad_reference + rating_score
802 # activeness
803 recently_active = int_(User.last_active >= now() - timedelta(days=180))
804 very_recently_active = int_(User.last_active >= now() - timedelta(days=14))
805 recently_messaged = int_(func.max(Message.time) > now() - timedelta(days=14))
806 messaged_lots = int_(func.count(Message.id) > 5)
807 messaging_points_subquery = (recently_messaged + messaged_lots).label("messaging_points")
808 messaging_subquery = (
809 select(Message.author_id.label("user_id"), messaging_points_subquery)
810 .where(Message.message_type == MessageType.text)
811 .group_by(Message.author_id)
812 .subquery()
813 )
814 activeness_points = recently_active + 2 * very_recently_active + int_(messaging_subquery.c.messaging_points)
816 # verification
817 cb_subquery = (
818 select(ClusterSubscription.user_id.label("user_id"), func.min(Cluster.parent_node_id).label("min_node_id"))
819 .join(Cluster, Cluster.id == ClusterSubscription.cluster_id)
820 .where(ClusterSubscription.role == ClusterRole.admin)
821 .where(Cluster.is_official_cluster)
822 .group_by(ClusterSubscription.user_id)
823 .subquery()
824 )
825 min_node_id = cb_subquery.c.min_node_id
826 cb = int_(min_node_id >= 1)
827 wcb = int_(min_node_id == 1)
828 badge_points = {
829 "founder": 100,
830 "board_member": 20,
831 "past_board_member": 5,
832 "strong_verification": 3,
833 "volunteer": 3,
834 "past_volunteer": 2,
835 "donor": 1,
836 "phone_verified": 1,
837 }
839 badge_subquery = (
840 select(
841 UserBadge.user_id.label("user_id"),
842 func.sum(case(badge_points, value=UserBadge.badge_id, else_=0)).label("badge_points"),
843 )
844 .group_by(UserBadge.user_id)
845 .subquery()
846 )
848 other_points = 0.0 + 10 * wcb + 5 * cb + int_(badge_subquery.c.badge_points)
850 # response rate
851 hr_subquery = select(
852 UserResponseRate.user_id,
853 float_(extract("epoch", UserResponseRate.response_time_33p) / 60.0).label("response_time_33p"),
854 float_(extract("epoch", UserResponseRate.response_time_66p) / 60.0).label("response_time_66p"),
855 ).subquery()
856 response_time_33p = hr_subquery.c.response_time_33p
857 response_time_66p = hr_subquery.c.response_time_66p
858 # be careful with nulls
859 response_rate_points = -10 * int_(response_time_33p > 60 * 96.0) + 5 * int_(response_time_66p < 60 * 96.0)
861 recommendation_score = (
862 hosting_status_points
863 + profile_points
864 + ref_score
865 + activeness_points
866 + other_points
867 + response_rate_points
868 + 2 * poor_man_gaussian()
869 )
871 scores = (
872 select(User.id.label("user_id"), recommendation_score.label("score"))
873 .outerjoin(messaging_subquery, messaging_subquery.c.user_id == User.id)
874 .outerjoin(left_refs_subquery, left_refs_subquery.c.user_id == User.id)
875 .outerjoin(badge_subquery, badge_subquery.c.user_id == User.id)
876 .outerjoin(received_ref_subquery, received_ref_subquery.c.user_id == User.id)
877 .outerjoin(cb_subquery, cb_subquery.c.user_id == User.id)
878 .outerjoin(hr_subquery, hr_subquery.c.user_id == User.id)
879 ).subquery()
881 session.execute(update(User).values(recommendation_score=scores.c.score).where(User.id == scores.c.user_id))
883 logger.info("Updated recommendation scores")
886def update_badges(payload: empty_pb2.Empty) -> None:
887 with session_scope() as session:
889 def update_badge(badge_id: str, members: Sequence[int]) -> None:
890 badge = get_badge_dict()[badge_id]
891 # this batch job has no per-user context to evaluate the gate against, so it's global
892 if badge.flag is not None and not experimentation.get_global_boolean_value(badge.flag, default=True):
893 members = []
894 user_ids = session.execute(select(UserBadge.user_id).where(UserBadge.badge_id == badge.id)).scalars().all()
895 # in case the user ids don't exist in the db
896 actual_members = session.execute(select(User.id).where(User.id.in_(members))).scalars().all()
897 # we should add the badge to these
898 add = set(actual_members) - set(user_ids)
899 # we should remove the badge from these
900 remove = set(user_ids) - set(actual_members)
901 for user_id in add:
902 user_add_badge(session, user_id, badge.id)
904 for user_id in remove:
905 user_remove_badge(session, user_id, badge.id)
907 update_badge("founder", get_static_badge_dict()["founder"])
908 update_badge("board_member", get_static_badge_dict()["board_member"])
909 update_badge("past_board_member", get_static_badge_dict()["past_board_member"])
910 update_badge("donor", session.execute(select(User.id).where(User.last_donated.is_not(None))).scalars().all())
911 update_badge("moderator", session.execute(select(User.id).where(User.is_superuser)).scalars().all())
912 update_badge("phone_verified", session.execute(select(User.id).where(User.phone_is_verified)).scalars().all())
913 # strong verification requires passport on file + gender/sex correspondence and date of birth match
914 update_badge(
915 "strong_verification",
916 session.execute(
917 select(User.id)
918 .join(StrongVerificationAttempt, StrongVerificationAttempt.user_id == User.id)
919 .where(StrongVerificationAttempt.has_strong_verification(User))
920 )
921 .scalars()
922 .all(),
923 )
924 # volunteer badge for active volunteers (stopped_volunteering is null)
925 update_badge(
926 "volunteer",
927 session.execute(select(Volunteer.user_id).where(Volunteer.stopped_volunteering.is_(None))).scalars().all(),
928 )
929 # past_volunteer badge for past volunteers (stopped_volunteering is not null)
930 update_badge(
931 "past_volunteer",
932 session.execute(select(Volunteer.user_id).where(Volunteer.stopped_volunteering.is_not(None)))
933 .scalars()
934 .all(),
935 )
938def finalize_strong_verification(payload: jobs_pb2.FinalizeStrongVerificationPayload) -> None:
939 with session_scope() as session:
940 verification_attempt = session.execute(
941 select(StrongVerificationAttempt)
942 .where(StrongVerificationAttempt.id == payload.verification_attempt_id)
943 .where(StrongVerificationAttempt.status == StrongVerificationAttemptStatus.in_progress_waiting_on_backend)
944 ).scalar_one()
945 response = requests.post(
946 "https://passportreader.app/api/v1/session.get",
947 auth=(config.IRIS_ID_PUBKEY, config.IRIS_ID_SECRET),
948 json={"id": verification_attempt.iris_session_id},
949 timeout=10,
950 verify="/etc/ssl/certs/ca-certificates.crt",
951 )
952 if response.status_code != 200: 952 ↛ 953line 952 didn't jump to line 953 because the condition on line 952 was never true
953 raise Exception(f"Iris didn't return 200: {response.text}")
954 json_data = response.json()
955 reference_payload = internal_pb2.VerificationReferencePayload.FromString(
956 simple_decrypt("iris_callback", b64decode(json_data["reference"]))
957 )
958 assert verification_attempt.user_id == reference_payload.user_id
959 assert verification_attempt.verification_attempt_token == reference_payload.verification_attempt_token
960 assert verification_attempt.iris_session_id == json_data["id"]
961 assert json_data["state"] == "APPROVED"
963 if json_data["document_type"] != "PASSPORT":
964 verification_attempt.status = StrongVerificationAttemptStatus.failed
965 notify(
966 session,
967 user_id=verification_attempt.user_id,
968 topic_action=NotificationTopicAction.verification__sv_fail,
969 key="",
970 data=notification_data_pb2.VerificationSVFail(
971 reason=notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT
972 ),
973 )
974 return
976 assert json_data["document_type"] == "PASSPORT"
978 expiry_date = date.fromisoformat(json_data["expiry_date"])
979 nationality = json_data["nationality"]
980 last_three_document_chars = json_data["document_number"][-3:]
982 existing_attempt = session.execute(
983 select(StrongVerificationAttempt)
984 .where(StrongVerificationAttempt.passport_expiry_date == expiry_date)
985 .where(StrongVerificationAttempt.passport_nationality == nationality)
986 .where(StrongVerificationAttempt.passport_last_three_document_chars == last_three_document_chars)
987 .order_by(StrongVerificationAttempt.id)
988 .limit(1)
989 ).scalar_one_or_none()
991 verification_attempt.has_minimal_data = True
992 verification_attempt.passport_expiry_date = expiry_date
993 verification_attempt.passport_nationality = nationality
994 verification_attempt.passport_last_three_document_chars = last_three_document_chars
996 if existing_attempt:
997 verification_attempt.status = StrongVerificationAttemptStatus.duplicate
999 if existing_attempt.user_id != verification_attempt.user_id:
1000 session.flush()
1001 send_duplicate_strong_verification_email(session, existing_attempt, verification_attempt)
1003 notify(
1004 session,
1005 user_id=verification_attempt.user_id,
1006 topic_action=NotificationTopicAction.verification__sv_fail,
1007 key="",
1008 data=notification_data_pb2.VerificationSVFail(reason=notification_data_pb2.SV_FAIL_REASON_DUPLICATE),
1009 )
1010 return
1012 verification_attempt.has_full_data = True
1013 verification_attempt.passport_encrypted_data = asym_encrypt(
1014 config.VERIFICATION_DATA_PUBLIC_KEY, response.text.encode("utf8")
1015 )
1016 verification_attempt.passport_date_of_birth = date.fromisoformat(json_data["date_of_birth"])
1017 verification_attempt.passport_sex = PassportSex[json_data["sex"].lower()]
1018 verification_attempt.status = StrongVerificationAttemptStatus.succeeded
1020 session.flush()
1022 strong_verification_completions_counter.inc()
1024 user = verification_attempt.user
1025 if verification_attempt.has_strong_verification(user): 1025 ↛ 1040line 1025 didn't jump to line 1040 because the condition on line 1025 was always true
1026 badge_id = "strong_verification"
1027 if session.execute(
1028 select(UserBadge).where(UserBadge.user_id == user.id, UserBadge.badge_id == badge_id)
1029 ).scalar_one_or_none():
1030 return
1032 user_add_badge(session, user.id, badge_id, do_notify=False)
1033 notify(
1034 session,
1035 user_id=verification_attempt.user_id,
1036 topic_action=NotificationTopicAction.verification__sv_success,
1037 key="",
1038 )
1039 else:
1040 notify(
1041 session,
1042 user_id=verification_attempt.user_id,
1043 topic_action=NotificationTopicAction.verification__sv_fail,
1044 key="",
1045 data=notification_data_pb2.VerificationSVFail(
1046 reason=notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE_OR_GENDER
1047 ),
1048 )
1051def send_activeness_probes(payload: empty_pb2.Empty) -> None:
1052 with session_scope() as session:
1053 ## Step 1: create new activeness probes for those who need it and don't have one (if enabled)
1055 if config.ACTIVENESS_PROBES_ENABLED:
1056 # current activeness probes
1057 subquery = select(ActivenessProbe.user_id).where(ActivenessProbe.responded == None).subquery()
1059 # users who we should send an activeness probe to
1060 new_probe_user_ids = (
1061 session.execute(
1062 select(User.id)
1063 .where(User.is_visible)
1064 .where(User.hosting_status == HostingStatus.can_host)
1065 .where(User.last_active < func.now() - ACTIVENESS_PROBE_INACTIVITY_PERIOD)
1066 .where(User.id.not_in(select(subquery.c.user_id)))
1067 )
1068 .scalars()
1069 .all()
1070 )
1072 total_users = session.execute(select(func.count()).select_from(User).where(User.is_visible)).scalar_one()
1073 probes_today = session.execute(
1074 select(func.count())
1075 .select_from(ActivenessProbe)
1076 .where(func.now() - ActivenessProbe.probe_initiated < timedelta(hours=24))
1077 ).scalar_one()
1079 # send probes to max 2% of users per day
1080 max_probes_per_day = 0.02 * total_users
1081 max_probe_size = int(max(min(max_probes_per_day - probes_today, max_probes_per_day / 24), 1))
1083 if len(new_probe_user_ids) > max_probe_size: 1083 ↛ 1084line 1083 didn't jump to line 1084 because the condition on line 1083 was never true
1084 new_probe_user_ids = sample(new_probe_user_ids, max_probe_size)
1086 for user_id in new_probe_user_ids:
1087 session.add(ActivenessProbe(user_id=user_id))
1089 session.commit()
1091 ## Step 2: actually send out probe notifications
1092 for probe_number_minus_1, delay in enumerate(ACTIVENESS_PROBE_TIME_REMINDERS):
1093 probes = (
1094 session.execute(
1095 select(ActivenessProbe)
1096 .where(ActivenessProbe.notifications_sent == probe_number_minus_1)
1097 .where(ActivenessProbe.probe_initiated + delay < func.now())
1098 .where(ActivenessProbe.is_pending)
1099 )
1100 .scalars()
1101 .all()
1102 )
1104 for probe in probes:
1105 probe.notifications_sent = probe_number_minus_1 + 1
1106 context = make_notification_user_context(user_id=probe.user.id)
1107 notify(
1108 session,
1109 user_id=probe.user.id,
1110 topic_action=NotificationTopicAction.activeness__probe,
1111 key=str(probe.id),
1112 data=notification_data_pb2.ActivenessProbe(
1113 reminder_number=probe_number_minus_1 + 1,
1114 deadline=Timestamp_from_datetime(probe.probe_initiated + ACTIVENESS_PROBE_EXPIRY_TIME),
1115 ),
1116 )
1117 session.commit()
1119 ## Step 3: for those who haven't responded, mark them as failed
1120 expired_probes = (
1121 session.execute(
1122 select(ActivenessProbe)
1123 .where(ActivenessProbe.notifications_sent == len(ACTIVENESS_PROBE_TIME_REMINDERS))
1124 .where(ActivenessProbe.is_pending)
1125 .where(ActivenessProbe.probe_initiated + ACTIVENESS_PROBE_EXPIRY_TIME < func.now())
1126 )
1127 .scalars()
1128 .all()
1129 )
1131 for probe in expired_probes:
1132 probe.responded = now()
1133 probe.response = ActivenessProbeStatus.expired
1134 if probe.user.hosting_status == HostingStatus.can_host: 1134 ↛ 1136line 1134 didn't jump to line 1136 because the condition on line 1134 was always true
1135 probe.user.hosting_status = HostingStatus.maybe
1136 if probe.user.meetup_status == MeetupStatus.wants_to_meetup: 1136 ↛ 1138line 1136 didn't jump to line 1138 because the condition on line 1136 was always true
1137 probe.user.meetup_status = MeetupStatus.open_to_meetup
1138 session.commit()
1141def update_randomized_locations(payload: empty_pb2.Empty) -> None:
1142 """
1143 We generate for each user a randomized location as follows:
1144 - Start from a strong random seed (based on the SECRET env var and our key derivation function)
1145 - For each user, mix in the user_id for randomness
1146 - Generate a radius from [0.02, 0.1] degrees (about 2-10km)
1147 - Generate an angle from [0, 360]
1148 - Randomized location is then a distance `radius` away at an angle `angle` from `geom`
1149 """
1150 randomization_secret = get_secret(USER_LOCATION_RANDOMIZATION_NAME)
1152 def gen_randomized_coords(user_id: int, lat: float, lng: float) -> tuple[float, float]:
1153 radius_u = stable_secure_uniform(randomization_secret, seed=bytes(f"{user_id}|radius", "ascii"))
1154 angle_u = stable_secure_uniform(randomization_secret, seed=bytes(f"{user_id}|angle", "ascii"))
1155 radius = 0.02 + 0.08 * radius_u
1156 angle_rad = 2 * pi * angle_u
1157 offset_lng = radius * cos(angle_rad)
1158 offset_lat = radius * sin(angle_rad)
1159 return lat + offset_lat, lng + offset_lng
1161 user_updates: list[dict[str, Any]] = []
1163 with session_scope() as session:
1164 users_to_update = session.execute(select(User.id, User.geom).where(User.randomized_geom == None)).all()
1166 for user_id, geom in users_to_update:
1167 lat, lng = get_coordinates(geom)
1168 user_updates.append(
1169 {"id": user_id, "randomized_geom": create_coordinate(*gen_randomized_coords(user_id, lat, lng))}
1170 )
1172 with session_scope() as session:
1173 session.execute(update(User), user_updates)
1176def send_event_reminders(payload: empty_pb2.Empty) -> None:
1177 """
1178 Sends reminders for events that are 24 hours away to users who marked themselves as attending.
1179 """
1180 logger.info("Sending event reminder emails")
1182 with session_scope() as session:
1183 occurrences = (
1184 session.execute(
1185 select(EventOccurrence)
1186 .where(EventOccurrence.start_time <= now() + EVENT_REMINDER_TIMEDELTA)
1187 .where(EventOccurrence.start_time >= now())
1188 .where(~EventOccurrence.is_cancelled)
1189 .where(~EventOccurrence.is_deleted)
1190 )
1191 .scalars()
1192 .all()
1193 )
1195 for occurrence in occurrences:
1196 results = session.execute(
1197 select(User, EventOccurrenceAttendee)
1198 .join(EventOccurrenceAttendee, EventOccurrenceAttendee.user_id == User.id)
1199 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id)
1200 .where(EventOccurrenceAttendee.reminder_sent == False)
1201 .where(User.is_visible)
1202 .where(~User.is_shadowed)
1203 ).all()
1205 for user, attendee in results:
1206 context = make_notification_user_context(user_id=user.id)
1208 notify(
1209 session,
1210 user_id=user.id,
1211 topic_action=NotificationTopicAction.event__reminder,
1212 key=str(occurrence.id),
1213 data=notification_data_pb2.EventReminder(
1214 event=event_to_pb(session, occurrence, context),
1215 user=user_model_to_pb(user, session, context),
1216 ),
1217 moderation_state_id=occurrence.moderation_state_id,
1218 )
1220 attendee.reminder_sent = True
1221 session.commit()
1224def check_expo_push_receipts(payload: empty_pb2.Empty) -> None:
1225 """
1226 Check Expo push receipts in batch and update delivery attempts.
1227 """
1228 MAX_ITERATIONS = 100 # Safety limit: 100 batches * 100 attempts = 10,000 max
1230 for iteration in range(MAX_ITERATIONS): 1230 ↛ 1292line 1230 didn't jump to line 1292 because the loop on line 1230 didn't complete
1231 with session_scope() as session:
1232 # Find all delivery attempts that need receipt checking
1233 # Wait 15 minutes per Expo's recommendation before checking receipts
1234 attempts = (
1235 session.execute(
1236 select(PushNotificationDeliveryAttempt)
1237 .where(PushNotificationDeliveryAttempt.expo_ticket_id != None)
1238 .where(PushNotificationDeliveryAttempt.receipt_checked_at == None)
1239 .where(PushNotificationDeliveryAttempt.time < now() - timedelta(minutes=15))
1240 .where(PushNotificationDeliveryAttempt.time > now() - timedelta(hours=24))
1241 .limit(100)
1242 )
1243 .scalars()
1244 .all()
1245 )
1247 if not attempts:
1248 logger.debug("No Expo receipts to check")
1249 return
1251 logger.info(f"Checking {len(attempts)} Expo push receipts")
1253 receipts = get_expo_push_receipts([not_none(attempt.expo_ticket_id) for attempt in attempts])
1255 for attempt in attempts:
1256 receipt = receipts.get(not_none(attempt.expo_ticket_id))
1258 # Always mark as checked to avoid infinite loops
1259 attempt.receipt_checked_at = now()
1261 if receipt is None:
1262 # Receipt not found after 15min - likely expired (>24h) or never existed
1263 # Per Expo docs: receipts should be available within 15 minutes
1264 attempt.receipt_status = "not_found"
1265 continue
1267 attempt.receipt_status = receipt.get("status")
1269 if receipt.get("status") == "error":
1270 details = receipt.get("details", {})
1271 error_code = details.get("error")
1272 attempt.receipt_error_code = error_code
1274 if error_code == "DeviceNotRegistered": 1274 ↛ 1289line 1274 didn't jump to line 1289 because the condition on line 1274 was always true
1275 # Device token is no longer valid - disable the subscription
1276 sub = session.execute(
1277 select(PushNotificationSubscription).where(
1278 PushNotificationSubscription.id == attempt.push_notification_subscription_id
1279 )
1280 ).scalar_one()
1282 if sub.disabled_at > now(): 1282 ↛ 1255line 1282 didn't jump to line 1255 because the condition on line 1282 was always true
1283 sub.disabled_at = now()
1284 logger.info(f"Disabled push sub {sub.id} due to DeviceNotRegistered in receipt")
1285 push_notification_counter.labels(
1286 platform="expo", outcome="permanent_subscription_failure_receipt"
1287 ).inc()
1288 else:
1289 logger.warning(f"Expo receipt error for ticket {attempt.expo_ticket_id}: {error_code}")
1291 # If we get here, we've exhausted MAX_ITERATIONS without finishing
1292 raise RuntimeError(
1293 f"check_expo_push_receipts exceeded {MAX_ITERATIONS} iterations - "
1294 "there may be an unusually large backlog of receipts to check"
1295 )
1298def send_postal_verification_postcard(payload: jobs_pb2.SendPostalVerificationPostcardPayload) -> None:
1299 """
1300 Sends the postcard via external API and updates attempt status.
1301 """
1302 with session_scope() as session:
1303 attempt = session.execute(
1304 select(PostalVerificationAttempt).where(
1305 PostalVerificationAttempt.id == payload.postal_verification_attempt_id
1306 )
1307 ).scalar_one_or_none()
1309 if not attempt or attempt.status != PostalVerificationStatus.in_progress: 1309 ↛ 1310line 1309 didn't jump to line 1310 because the condition on line 1309 was never true
1310 logger.warning(
1311 f"Postal verification attempt {payload.postal_verification_attempt_id} not found or wrong state"
1312 )
1313 return
1315 user_name = session.execute(select(User.name).where(User.id == attempt.user_id)).scalar_one()
1317 job_id = send_postcard(
1318 recipient_name=user_name,
1319 address_line_1=attempt.address_line_1,
1320 address_line_2=attempt.address_line_2,
1321 city=attempt.city,
1322 state=attempt.state,
1323 postal_code=attempt.postal_code,
1324 country=attempt.country_code,
1325 verification_code=not_none(attempt.verification_code),
1326 )
1328 attempt.mypostcard_job_id = job_id
1329 attempt.status = PostalVerificationStatus.awaiting_verification
1330 attempt.postcard_sent_at = func.now()
1332 postcards_sent_counter.labels(country_code=attempt.country_code).inc()
1334 context = make_background_user_context(attempt.user_id)
1335 log_event(
1336 context,
1337 session,
1338 "postcard.sent",
1339 {
1340 "attempt_id": attempt.id,
1341 "country": attempt.country_code,
1342 "city": attempt.city,
1343 "mypostcard_job_id": job_id,
1344 },
1345 )
1347 notify(
1348 session,
1349 user_id=attempt.user_id,
1350 topic_action=NotificationTopicAction.postal_verification__postcard_sent,
1351 key="",
1352 data=notification_data_pb2.PostalVerificationPostcardSent(
1353 city=attempt.city,
1354 country=attempt.country_code,
1355 ),
1356 )
1359def check_mypostcard_jobs(payload: empty_pb2.Empty) -> None:
1360 """
1361 Checks that all MyPostcard jobs from the last week are tied to a postal verification attempt.
1362 """
1363 if not experimentation.get_global_boolean_value("postal_verification_enabled", default=False):
1364 return
1366 with session_scope() as session:
1367 mypostcard_job_ids = set(
1368 get_order_ids(
1369 date_from=(now() - timedelta(days=7)).date(),
1370 date_to=now().date(),
1371 )
1372 )
1374 known_job_ids = set(
1375 session.execute(
1376 select(PostalVerificationAttempt.mypostcard_job_id).where(
1377 PostalVerificationAttempt.mypostcard_job_id.isnot(None),
1378 PostalVerificationAttempt.created >= now() - timedelta(days=14),
1379 )
1380 )
1381 .scalars()
1382 .all()
1383 )
1385 orphaned = mypostcard_job_ids - known_job_ids
1386 if orphaned:
1387 report_message(
1388 f"Found {len(orphaned)} orphaned MyPostcard jobs not tied to any verification attempt: {orphaned}"
1389 )
1392class DatabaseInconsistencyError(Exception):
1393 """Raised when database consistency checks fail"""
1395 pass
1398def check_database_consistency(payload: empty_pb2.Empty) -> None:
1399 """
1400 Checks database consistency and raises an exception if any issues are found.
1401 """
1402 logger.info("Checking database consistency")
1403 errors = []
1405 with session_scope() as session:
1406 # Check that all users have a profile gallery
1407 users_without_gallery = session.execute(
1408 select(User.id, User.username).where(User.profile_gallery_id.is_(None))
1409 ).all()
1410 if users_without_gallery:
1411 errors.append(f"Users without profile gallery: {users_without_gallery}")
1413 # Check that all profile galleries point to their owner
1414 mismatched_galleries = session.execute(
1415 select(User.id, User.username, User.profile_gallery_id, PhotoGallery.owner_user_id)
1416 .join(PhotoGallery, User.profile_gallery_id == PhotoGallery.id)
1417 .where(User.profile_gallery_id.is_not(None))
1418 .where(PhotoGallery.owner_user_id != User.id)
1419 ).all()
1420 if mismatched_galleries: 1420 ↛ 1421line 1420 didn't jump to line 1421 because the condition on line 1420 was never true
1421 errors.append(f"Profile galleries with mismatched owner: {mismatched_galleries}")
1423 # === Moderation System Consistency Checks ===
1425 # Check every ModerationState has at least one INITIAL_REVIEW queue item
1426 # Skip items with ID < 2000000 as they were created before this check was introduced
1427 states_without_initial_review = session.execute(
1428 select(ModerationState.id, ModerationState.object_type, ModerationState.object_id).where(
1429 ModerationState.id >= 2000000,
1430 ~exists(
1431 select(1)
1432 .where(ModerationQueueItem.moderation_state_id == ModerationState.id)
1433 .where(ModerationQueueItem.trigger == ModerationTrigger.initial_review)
1434 ),
1435 )
1436 ).all()
1437 if states_without_initial_review: 1437 ↛ 1438line 1437 didn't jump to line 1438 because the condition on line 1437 was never true
1438 errors.append(f"ModerationStates without INITIAL_REVIEW queue item: {states_without_initial_review}")
1440 # Check every ModerationState has a CREATE log entry
1441 # Skip items with ID < 2000000 as they were created before this check was introduced
1442 states_without_create_log = session.execute(
1443 select(ModerationState.id, ModerationState.object_type, ModerationState.object_id).where(
1444 ModerationState.id >= 2000000,
1445 ~exists(
1446 select(1)
1447 .where(ModerationLog.moderation_state_id == ModerationState.id)
1448 .where(ModerationLog.action == ModerationAction.create)
1449 ),
1450 )
1451 ).all()
1452 if states_without_create_log: 1452 ↛ 1453line 1452 didn't jump to line 1453 because the condition on line 1452 was never true
1453 errors.append(f"ModerationStates without CREATE log entry: {states_without_create_log}")
1455 # Check resolved queue items point to log entries for the same moderation state
1456 resolved_item_log_mismatches = session.execute(
1457 select(ModerationQueueItem.id, ModerationQueueItem.moderation_state_id, ModerationLog.moderation_state_id)
1458 .join(ModerationLog, ModerationQueueItem.resolved_by_log_id == ModerationLog.id)
1459 .where(ModerationQueueItem.resolved_by_log_id.is_not(None))
1460 .where(ModerationQueueItem.moderation_state_id != ModerationLog.moderation_state_id)
1461 ).all()
1462 if resolved_item_log_mismatches: 1462 ↛ 1463line 1462 didn't jump to line 1463 because the condition on line 1462 was never true
1463 errors.append(f"Resolved queue items with mismatched moderation_state_id: {resolved_item_log_mismatches}")
1465 # Check every HOST_REQUEST ModerationState has exactly one HostRequest pointing to it
1466 hr_states = (
1467 session.execute(
1468 select(ModerationState.id).where(ModerationState.object_type == ModerationObjectType.host_request)
1469 )
1470 .scalars()
1471 .all()
1472 )
1473 for state_id in hr_states: 1473 ↛ 1474line 1473 didn't jump to line 1474 because the loop on line 1473 never started
1474 hr_count = session.execute(
1475 select(func.count()).where(HostRequest.moderation_state_id == state_id)
1476 ).scalar_one()
1477 if hr_count != 1:
1478 errors.append(f"ModerationState {state_id} (HOST_REQUEST) has {hr_count} HostRequests (expected 1)")
1480 # Check every GROUP_CHAT ModerationState has exactly one GroupChat pointing to it
1481 gc_states = (
1482 session.execute(
1483 select(ModerationState.id).where(ModerationState.object_type == ModerationObjectType.group_chat)
1484 )
1485 .scalars()
1486 .all()
1487 )
1488 for state_id in gc_states:
1489 gc_count = session.execute(
1490 select(func.count()).where(GroupChat.moderation_state_id == state_id)
1491 ).scalar_one()
1492 if gc_count != 1: 1492 ↛ 1493line 1492 didn't jump to line 1493 because the condition on line 1492 was never true
1493 errors.append(f"ModerationState {state_id} (GROUP_CHAT) has {gc_count} GroupChats (expected 1)")
1495 # Check ModerationState.object_id matches the actual object's ID
1496 hr_object_id_mismatches = session.execute(
1497 select(ModerationState.id, ModerationState.object_id, HostRequest.conversation_id)
1498 .join(HostRequest, HostRequest.moderation_state_id == ModerationState.id)
1499 .where(ModerationState.object_type == ModerationObjectType.host_request)
1500 .where(ModerationState.object_id != HostRequest.conversation_id)
1501 ).all()
1502 if hr_object_id_mismatches: 1502 ↛ 1503line 1502 didn't jump to line 1503 because the condition on line 1502 was never true
1503 errors.append(f"ModerationState object_id mismatch for HOST_REQUEST: {hr_object_id_mismatches}")
1505 gc_object_id_mismatches = session.execute(
1506 select(ModerationState.id, ModerationState.object_id, GroupChat.conversation_id)
1507 .join(GroupChat, GroupChat.moderation_state_id == ModerationState.id)
1508 .where(ModerationState.object_type == ModerationObjectType.group_chat)
1509 .where(ModerationState.object_id != GroupChat.conversation_id)
1510 ).all()
1511 if gc_object_id_mismatches: 1511 ↛ 1512line 1511 didn't jump to line 1512 because the condition on line 1511 was never true
1512 errors.append(f"ModerationState object_id mismatch for GROUP_CHAT: {gc_object_id_mismatches}")
1514 # Check reverse mapping: HostRequest's moderation_state points to correct ModerationState
1515 hr_reverse_mismatches = session.execute(
1516 select(
1517 HostRequest.conversation_id,
1518 HostRequest.moderation_state_id,
1519 ModerationState.object_type,
1520 ModerationState.object_id,
1521 )
1522 .join(ModerationState, HostRequest.moderation_state_id == ModerationState.id)
1523 .where(
1524 (ModerationState.object_type != ModerationObjectType.host_request)
1525 | (ModerationState.object_id != HostRequest.conversation_id)
1526 )
1527 ).all()
1528 if hr_reverse_mismatches: 1528 ↛ 1529line 1528 didn't jump to line 1529 because the condition on line 1528 was never true
1529 errors.append(f"HostRequest points to ModerationState with wrong type/object_id: {hr_reverse_mismatches}")
1531 # Check reverse mapping: GroupChat's moderation_state points to correct ModerationState
1532 gc_reverse_mismatches = session.execute(
1533 select(
1534 GroupChat.conversation_id,
1535 GroupChat.moderation_state_id,
1536 ModerationState.object_type,
1537 ModerationState.object_id,
1538 )
1539 .join(ModerationState, GroupChat.moderation_state_id == ModerationState.id)
1540 .where(
1541 (ModerationState.object_type != ModerationObjectType.group_chat)
1542 | (ModerationState.object_id != GroupChat.conversation_id)
1543 )
1544 ).all()
1545 if gc_reverse_mismatches: 1545 ↛ 1546line 1545 didn't jump to line 1546 because the condition on line 1545 was never true
1546 errors.append(f"GroupChat points to ModerationState with wrong type/object_id: {gc_reverse_mismatches}")
1548 # Ensure auto-approve deadline isn't being exceeded by a significant margin
1549 # The auto-approver runs every 15s, so allow 5 minutes grace before alerting
1550 deadline_seconds = config.MODERATION_AUTO_APPROVE_DEADLINE_SECONDS
1551 if deadline_seconds > 0: 1551 ↛ 1552line 1551 didn't jump to line 1552 because the condition on line 1551 was never true
1552 grace_period = timedelta(minutes=5)
1553 stale_initial_review_items = session.execute(
1554 select(
1555 ModerationQueueItem.id,
1556 ModerationQueueItem.moderation_state_id,
1557 ModerationQueueItem.time_created,
1558 )
1559 .where(ModerationQueueItem.trigger == ModerationTrigger.initial_review)
1560 .where(ModerationQueueItem.resolved_by_log_id.is_(None))
1561 .where(ModerationQueueItem.time_created < now() - timedelta(seconds=deadline_seconds) - grace_period)
1562 ).all()
1563 if stale_initial_review_items:
1564 errors.append(
1565 f"INITIAL_REVIEW items exceeding auto-approve deadline by >5min: {stale_initial_review_items}"
1566 )
1568 if errors:
1569 raise DatabaseInconsistencyError("\n".join(errors))
1572def auto_approve_moderation_queue(payload: empty_pb2.Empty) -> None:
1573 """
1574 Dead man's switch: approves unresolved INITIAL_REVIEW content older than the deadline to VISIBLE, then
1575 re-flags it as a high-priority MACHINE_FLAG superseding only the INITIAL_REVIEW item. The switch only fires
1576 when moderators are behind, so every auto-approved item stays in the queue for a human to check. Other open
1577 flags are untouched, and items already actioned by moderators are left alone.
1578 """
1579 deadline_seconds = config.MODERATION_AUTO_APPROVE_DEADLINE_SECONDS
1580 if deadline_seconds <= 0:
1581 return
1583 with session_scope() as session:
1584 ctx = make_background_user_context(user_id=config.MODERATION_BOT_USER_ID)
1586 items = (
1587 Moderation()
1588 .GetModerationQueue(
1589 request=moderation_pb2.GetModerationQueueReq(
1590 triggers=[moderation_pb2.MODERATION_TRIGGER_INITIAL_REVIEW],
1591 unresolved_only=True,
1592 page_size=100,
1593 created_before=Timestamp_from_datetime(now() - timedelta(seconds=deadline_seconds)),
1594 ),
1595 context=ctx,
1596 session=session,
1597 )
1598 .queue_items
1599 )
1601 if not items:
1602 return
1604 # Skip items whose author is shadowed; their content stays in shadowed state indefinitely
1605 approvable = [item for item in items if not item.moderation_state.author.shadowed]
1606 if not approvable:
1607 return
1609 logger.info(f"Auto-approving {len(approvable)} moderation queue items")
1610 reason = f"Auto-approved: moderation deadline of {deadline_seconds} seconds exceeded."
1611 for item in approvable:
1612 Moderation().ModerateContent(
1613 request=moderation_pb2.ModerateContentReq(
1614 moderation_state_id=item.moderation_state_id,
1615 action=moderation_pb2.MODERATION_ACTION_APPROVE,
1616 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
1617 reason=reason,
1618 clear_flags=False,
1619 ),
1620 context=ctx,
1621 session=session,
1622 )
1623 Moderation().ModerateContent(
1624 request=moderation_pb2.ModerateContentReq(
1625 moderation_state_id=item.moderation_state_id,
1626 action=moderation_pb2.MODERATION_ACTION_FLAG,
1627 trigger=moderation_pb2.MODERATION_TRIGGER_MACHINE_FLAG,
1628 priority=MODERATION_AUTO_APPROVE_FLAG_PRIORITY,
1629 reason=reason,
1630 supersede_queue_item_id=item.queue_item_id,
1631 ),
1632 context=ctx,
1633 session=session,
1634 )
1635 moderation_auto_approved_counter.inc(len(approvable))