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