Coverage for app/backend/src/couchers/jobs/handlers.py: 88%
499 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-25 17:29 +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 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 )
173def send_message_notifications(payload: empty_pb2.Empty) -> None:
174 """
175 Sends out email notifications for messages that have been unseen for a long enough time
176 """
177 # very crude and dumb algorithm
178 logger.info("Sending out email notifications for unseen messages")
180 with session_scope() as session:
181 # users who have unnotified messages older than 5 minutes in any group chat
182 users = (
183 session.execute(
184 where_moderated_content_visible_to_user_column(
185 select(User)
186 .join(GroupChatSubscription, GroupChatSubscription.user_id == User.id)
187 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
188 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id),
189 GroupChat,
190 User.id,
191 )
192 .where(not_(GroupChatSubscription.is_muted))
193 .where(User.is_visible)
194 .where(Message.time >= GroupChatSubscription.joined)
195 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None))
196 .where(Message.id > User.last_notified_message_id)
197 .where(Message.id > GroupChatSubscription.last_seen_message_id)
198 .where(Message.time < now() - timedelta(minutes=5))
199 .where(Message.message_type == MessageType.text) # TODO: only text messages for now
200 )
201 .scalars()
202 .unique()
203 )
205 for user in users:
206 context = make_notification_user_context(user_id=user.id)
207 # now actually grab all the group chats, not just less than 5 min old
208 subquery = (
209 where_users_column_visible(
210 where_moderated_content_visible(
211 select(
212 GroupChatSubscription.group_chat_id.label("group_chat_id"),
213 func.max(GroupChatSubscription.id).label("group_chat_subscriptions_id"),
214 func.max(Message.id).label("message_id"),
215 func.count(Message.id).label("unseen_count"),
216 )
217 .join(Message, Message.conversation_id == GroupChatSubscription.group_chat_id)
218 .join(GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id),
219 context,
220 GroupChat,
221 is_list_operation=True,
222 )
223 .where(GroupChatSubscription.user_id == user.id)
224 .where(not_(GroupChatSubscription.is_muted))
225 .where(Message.id > user.last_notified_message_id)
226 .where(Message.id > GroupChatSubscription.last_seen_message_id)
227 .where(Message.time >= GroupChatSubscription.joined)
228 .where(Message.message_type == MessageType.text) # TODO: only text messages for now
229 .where(or_(Message.time <= GroupChatSubscription.left, GroupChatSubscription.left == None)),
230 context,
231 Message.author_id,
232 )
233 .group_by(GroupChatSubscription.group_chat_id)
234 .order_by(func.max(Message.id).desc())
235 .subquery()
236 )
238 unseen_messages = session.execute(
239 where_moderated_content_visible(
240 select(GroupChat, Message, subquery.c.unseen_count)
241 .join(subquery, subquery.c.message_id == Message.id)
242 .join(GroupChat, GroupChat.conversation_id == subquery.c.group_chat_id),
243 context,
244 GroupChat,
245 is_list_operation=True,
246 ).order_by(subquery.c.message_id.desc())
247 ).all()
249 if not unseen_messages:
250 continue
252 user.last_notified_message_id = max(message.id for _, message, _ in unseen_messages)
254 notify(
255 session,
256 user_id=user.id,
257 topic_action=NotificationTopicAction.chat__missed_messages,
258 key="",
259 data=notification_data_pb2.ChatMissedMessages(
260 messages=[
261 notification_data_pb2.ChatMessage(
262 author=user_model_to_pb(
263 message.author,
264 session,
265 context,
266 ),
267 text=message.text,
268 group_chat_id=message.conversation_id,
269 group_chat_title=group_chat.title or None,
270 unseen_count=unseen_count,
271 )
272 for group_chat, message, unseen_count in unseen_messages
273 ],
274 ),
275 )
276 session.commit()
279def send_request_notifications(payload: empty_pb2.Empty) -> None:
280 """
281 Sends out email notifications for unseen messages in host requests (as surfer or host)
282 """
283 logger.info("Sending out email notifications for unseen messages in host requests")
285 with session_scope() as session:
286 # Get all candidate users who might have unseen request messages.
287 # Drive from host_requests/messages (selective) rather than scanning all users (expensive).
288 surfer_ids = (
289 select(User.id)
290 .join(HostRequest, HostRequest.initiator_user_id == User.id)
291 .join(Message, Message.conversation_id == HostRequest.conversation_id)
292 .where(User.is_visible)
293 .where(Message.id > HostRequest.initiator_last_seen_message_id)
294 .where(Message.id > User.last_notified_request_message_id)
295 .where(Message.time < now() - timedelta(minutes=5))
296 .where(Message.message_type == MessageType.text)
297 )
298 host_ids = (
299 select(User.id)
300 .join(HostRequest, HostRequest.recipient_user_id == User.id)
301 .join(Message, Message.conversation_id == HostRequest.conversation_id)
302 .where(User.is_visible)
303 .where(Message.id > HostRequest.recipient_last_seen_message_id)
304 .where(Message.id > User.last_notified_request_message_id)
305 .where(Message.time < now() - timedelta(minutes=5))
306 .where(Message.message_type == MessageType.text)
307 )
308 candidate_user_ids = session.execute(union_all(surfer_ids, host_ids)).scalars().unique().all()
310 for user_id in candidate_user_ids:
311 context = make_notification_user_context(user_id=user_id)
313 # requests where this user is surfing
314 surfing_reqs = session.execute(
315 where_users_column_visible(
316 where_moderated_content_visible_to_user_column(
317 select(User, HostRequest, func.max(Message.id))
318 .where(User.id == user_id)
319 .join(HostRequest, HostRequest.initiator_user_id == User.id),
320 HostRequest,
321 HostRequest.initiator_user_id,
322 ),
323 context,
324 HostRequest.recipient_user_id,
325 )
326 .join(Message, Message.conversation_id == HostRequest.conversation_id)
327 .where(Message.id > HostRequest.initiator_last_seen_message_id)
328 .where(Message.id > User.last_notified_request_message_id)
329 .where(Message.time < now() - timedelta(minutes=5))
330 .where(Message.message_type == MessageType.text)
331 .group_by(User, HostRequest) # type: ignore[arg-type]
332 ).all()
334 # where this user is hosting
335 hosting_reqs = session.execute(
336 where_users_column_visible(
337 where_moderated_content_visible_to_user_column(
338 select(User, HostRequest, func.max(Message.id))
339 .where(User.id == user_id)
340 .join(HostRequest, HostRequest.recipient_user_id == User.id),
341 HostRequest,
342 HostRequest.recipient_user_id,
343 ),
344 context,
345 HostRequest.initiator_user_id,
346 )
347 .join(Message, Message.conversation_id == HostRequest.conversation_id)
348 .where(Message.id > HostRequest.recipient_last_seen_message_id)
349 .where(Message.id > User.last_notified_request_message_id)
350 .where(Message.time < now() - timedelta(minutes=5))
351 .where(Message.message_type == MessageType.text)
352 .group_by(User, HostRequest) # type: ignore[arg-type]
353 ).all()
355 for user, host_request, max_message_id in surfing_reqs:
356 user.last_notified_request_message_id = max(user.last_notified_request_message_id, max_message_id)
357 session.flush()
359 notify(
360 session,
361 user_id=user.id,
362 topic_action=NotificationTopicAction.host_request__missed_messages,
363 key=str(host_request.conversation_id),
364 data=notification_data_pb2.HostRequestMissedMessages(
365 host_request=host_request_to_pb(host_request, session, context),
366 user=user_model_to_pb(host_request.recipient, session, context),
367 am_host=False,
368 ),
369 )
371 for user, host_request, max_message_id in hosting_reqs:
372 user.last_notified_request_message_id = max(user.last_notified_request_message_id, max_message_id)
373 session.flush()
375 # When a host request is created, the recipient immediately receives a
376 # host_request__create notification that includes the initial message text.
377 # A few minutes later, this background job sees that same message as "unseen"
378 # (the recipient hasn't opened the request yet) and would send a duplicate
379 # missed_messages notification.
380 #
381 # To prevent this, we check if the only unseen text message in this host
382 # request is the very first text message in the conversation (i.e. the
383 # creation message). If so, we skip sending missed_messages — the user was
384 # already notified via host_request__create.
385 #
386 # Advancing last_notified_request_message_id above is safe even when we skip
387 # the notification: this watermark is only ever advanced when we process all
388 # unseen messages for the user, so skipping one notification doesn't cause us
389 # to miss future messages in other host requests.
390 only_creation_message = not session.execute(
391 select(
392 select(func.count())
393 .where(Message.conversation_id == host_request.conversation_id)
394 .where(Message.message_type == MessageType.text)
395 .scalar_subquery()
396 > 1
397 )
398 ).scalar_one()
399 if only_creation_message:
400 continue
402 notify(
403 session,
404 user_id=user.id,
405 topic_action=NotificationTopicAction.host_request__missed_messages,
406 key=str(host_request.conversation_id),
407 data=notification_data_pb2.HostRequestMissedMessages(
408 host_request=host_request_to_pb(host_request, session, context),
409 user=user_model_to_pb(host_request.initiator, session, context),
410 am_host=True,
411 ),
412 )
415def send_onboarding_emails(payload: empty_pb2.Empty) -> None:
416 """
417 Sends out onboarding emails
418 """
419 logger.info("Sending out onboarding emails")
421 with session_scope() as session:
422 # first onboarding email
423 users = (
424 session.execute(select(User).where(User.is_visible).where(User.onboarding_emails_sent == 0)).scalars().all()
425 )
427 for user in users:
428 notify(
429 session,
430 user_id=user.id,
431 topic_action=NotificationTopicAction.onboarding__reminder,
432 key="1",
433 )
434 user.onboarding_emails_sent = 1
435 user.last_onboarding_email_sent = now()
436 session.commit()
438 # second onboarding email
439 # sent after a week if the user has no profile or their "about me" section is less than 20 characters long
440 users = (
441 session.execute(
442 select(User)
443 .where(User.is_visible)
444 .where(User.onboarding_emails_sent == 1)
445 .where(now() - User.last_onboarding_email_sent > timedelta(days=7))
446 .where(~has_completed_profile_expression())
447 )
448 .scalars()
449 .all()
450 )
452 for user in users:
453 notify(
454 session,
455 user_id=user.id,
456 topic_action=NotificationTopicAction.onboarding__reminder,
457 key="2",
458 )
459 user.onboarding_emails_sent = 2
460 user.last_onboarding_email_sent = now()
461 session.commit()
464def send_reference_reminders(payload: empty_pb2.Empty) -> None:
465 """
466 Sends out reminders to write references after hosting/staying
467 """
468 logger.info("Sending out reference reminder emails")
470 # Keep this in chronological order!
471 reference_reminder_schedule = [
472 # (number, timedelta before we stop being able to write a ref, text for how long they have left to write the ref)
473 # the end time to write a reference is supposed to be midnight in the host's timezone
474 # 8 pm ish on the last day of the stay
475 (1, timedelta(days=15) - timedelta(hours=20), 14),
476 # 2 pm ish a week after stay
477 (2, timedelta(days=8) - timedelta(hours=14), 7),
478 # 10 am ish 3 days before end of time to write ref
479 (3, timedelta(days=4) - timedelta(hours=10), 3),
480 ]
482 with session_scope() as session:
483 # iterate the reminders in backwards order, so if we missed out on one we don't send duplicates
484 for reminder_number, reminder_time, reminder_days_left in reversed(reference_reminder_schedule):
485 user = aliased(User)
486 other_user = aliased(User)
487 # surfers needing to write a ref
488 q1 = (
489 select(literal(True), HostRequest, user, other_user)
490 .join(user, user.id == HostRequest.initiator_user_id)
491 .join(other_user, other_user.id == HostRequest.recipient_user_id)
492 .outerjoin(
493 Reference,
494 and_(
495 Reference.host_request_id == HostRequest.conversation_id,
496 # if no reference is found in this join, then the surfer has not written a ref
497 Reference.from_user_id == HostRequest.initiator_user_id,
498 ),
499 )
500 .where(Reference.id == None)
501 .where(HostRequest.can_write_reference)
502 .where(HostRequest.initiator_sent_reference_reminders < reminder_number)
503 .where(HostRequest.end_time_to_write_reference - reminder_time < now())
504 .where(HostRequest.initiator_reason_didnt_meetup == None)
505 .where(users_visible_to_each_other(self_user=user, other_user=other_user))
506 )
508 # hosts needing to write a ref
509 q2 = (
510 select(literal(False), HostRequest, user, other_user)
511 .join(user, user.id == HostRequest.recipient_user_id)
512 .join(other_user, other_user.id == HostRequest.initiator_user_id)
513 .outerjoin(
514 Reference,
515 and_(
516 Reference.host_request_id == HostRequest.conversation_id,
517 # if no reference is found in this join, then the host has not written a ref
518 Reference.from_user_id == HostRequest.recipient_user_id,
519 ),
520 )
521 .where(Reference.id == None)
522 .where(HostRequest.can_write_reference)
523 .where(HostRequest.recipient_sent_reference_reminders < reminder_number)
524 .where(HostRequest.end_time_to_write_reference - reminder_time < now())
525 .where(HostRequest.recipient_reason_didnt_meetup == None)
526 .where(users_visible_to_each_other(self_user=user, other_user=other_user))
527 )
529 union = union_all(q1, q2).subquery()
530 query = select(
531 union.c[0].label("surfed"),
532 aliased(HostRequest, union),
533 aliased(user, union),
534 aliased(other_user, union),
535 )
536 reference_reminders = session.execute(query).all()
538 for surfed, host_request, user, other_user in reference_reminders:
539 # visibility and blocking already checked in sql
540 assert user.is_visible
541 context = make_notification_user_context(user_id=user.id)
542 topic_action = (
543 NotificationTopicAction.reference__reminder_surfed
544 if surfed
545 else NotificationTopicAction.reference__reminder_hosted
546 )
547 notify(
548 session,
549 user_id=user.id,
550 topic_action=topic_action,
551 key=str(host_request.conversation_id),
552 data=notification_data_pb2.ReferenceReminder(
553 host_request_id=host_request.conversation_id,
554 other_user=user_model_to_pb(other_user, session, context),
555 days_left=reminder_days_left,
556 ),
557 )
558 if surfed:
559 host_request.initiator_sent_reference_reminders = reminder_number
560 else:
561 host_request.recipient_sent_reference_reminders = reminder_number
562 session.commit()
565def send_host_request_reminders(payload: empty_pb2.Empty) -> None:
566 with session_scope() as session:
567 host_has_sent_message = select(1).where(
568 Message.conversation_id == HostRequest.conversation_id, Message.author_id == HostRequest.recipient_user_id
569 )
571 requests = (
572 session.execute(
573 where_user_columns_visible_to_each_other(
574 where_moderated_content_visible_to_user_column(
575 select(HostRequest),
576 HostRequest,
577 HostRequest.recipient_user_id,
578 )
579 .where(HostRequest.status == HostRequestStatus.pending)
580 .where(HostRequest.recipient_sent_request_reminders < HOST_REQUEST_MAX_REMINDERS)
581 .where(HostRequest.start_time > func.now())
582 .where((func.now() - HostRequest.last_sent_request_reminder_time) >= HOST_REQUEST_REMINDER_INTERVAL)
583 .where(~exists(host_has_sent_message)),
584 self_column=HostRequest.recipient_user_id,
585 other_column=HostRequest.initiator_user_id,
586 )
587 )
588 .scalars()
589 .all()
590 )
592 for host_request in requests:
593 host_request.recipient_sent_request_reminders += 1
594 host_request.last_sent_request_reminder_time = now()
596 context = make_notification_user_context(user_id=host_request.recipient_user_id)
597 notify(
598 session,
599 user_id=host_request.recipient_user_id,
600 topic_action=NotificationTopicAction.host_request__reminder,
601 key=str(host_request.conversation_id),
602 data=notification_data_pb2.HostRequestReminder(
603 host_request=host_request_to_pb(host_request, session, context),
604 surfer=user_model_to_pb(host_request.initiator, session, context),
605 ),
606 moderation_state_id=host_request.moderation_state_id,
607 )
609 session.commit()
612def add_users_to_email_list(payload: empty_pb2.Empty) -> None:
613 if not experimentation.get_global_boolean_value("listmonk_enabled", default=False): 613 ↛ 614line 613 didn't jump to line 614 because the condition on line 613 was never true
614 logger.info("Not adding users to mailing list")
615 return
617 sess = requests.Session()
618 sess.auth = (config.LISTMONK_API_USERNAME, config.LISTMONK_API_KEY)
620 def sync_subscriber(user: User, status: str) -> None:
621 r = sess.post(
622 config.LISTMONK_BASE_URL + "/api/subscribers",
623 json={
624 "email": user.email,
625 "name": user.name,
626 "lists": [config.LISTMONK_LIST_ID],
627 "preconfirm_subscriptions": True,
628 "attribs": {"couchers_user_id": user.id},
629 "status": status,
630 },
631 timeout=10,
632 )
633 # the API returns 409 if the subscriber already exists
634 if r.status_code not in (200, 409): 634 ↛ 635line 634 didn't jump to line 635 because the condition on line 634 was never true
635 raise Exception("Failed to update user mailing list status")
637 logger.info("Adding users to mailing list")
639 while True:
640 with session_scope() as session:
641 user = session.execute(
642 select(User).where(User.is_visible).where(User.in_sync_with_newsletter == False).limit(1)
643 ).scalar_one_or_none()
644 if not user:
645 logger.info("Finished adding users to mailing list")
646 break
648 if not user.opt_out_of_newsletter:
649 sync_subscriber(user, "enabled")
651 user.in_sync_with_newsletter = True
652 session.commit()
654 if experimentation.get_global_boolean_value("remove_removed_users_from_mailing_list_enabled", default=False): 654 ↛ 655line 654 didn't jump to line 655 because the condition on line 654 was never true
655 with session_scope() as session:
656 session.execute(
657 update(User)
658 .where(~User.is_visible | User.is_shadowed)
659 .where(User.opt_out_of_newsletter == False)
660 .values(opt_out_of_newsletter=True, in_sync_with_newsletter=False)
661 )
662 session.commit()
664 while True:
665 with session_scope() as session:
666 user = session.execute(
667 select(User)
668 .where(~User.is_visible | User.is_shadowed)
669 .where(User.in_sync_with_newsletter == False)
670 .limit(1)
671 ).scalar_one_or_none()
672 if not user:
673 logger.info("Finished removing users from mailing list")
674 return
676 sync_subscriber(user, "blocklisted")
677 user.in_sync_with_newsletter = True
678 session.commit()
681def enforce_community_membership(payload: empty_pb2.Empty) -> None:
682 tasks_enforce_community_memberships()
685def update_recommendation_scores(payload: empty_pb2.Empty) -> None:
686 text_fields = [
687 User.hometown,
688 User.occupation,
689 User.education,
690 User.about_me,
691 User.things_i_like,
692 User.about_place,
693 User.additional_information,
694 User.pet_details,
695 User.kid_details,
696 User.housemate_details,
697 User.other_host_info,
698 User.sleeping_details,
699 User.area,
700 User.house_rules,
701 ]
702 home_fields = [User.about_place, User.other_host_info, User.sleeping_details, User.area, User.house_rules]
704 def poor_man_gaussian() -> ColumnElement[float] | float:
705 """
706 Produces an approximatley std normal random variate
707 """
708 trials = 5
709 return (sum([func.random() for _ in range(trials)]) - trials / 2) / sqrt(trials / 12)
711 def int_(stmt: Any) -> Function[int]:
712 return func.coalesce(cast(stmt, Integer), 0)
714 def float_(stmt: Any) -> Function[float]:
715 return func.coalesce(cast(stmt, Float), 0.0)
717 with session_scope() as session:
718 # profile
719 profile_text = ""
720 for field in text_fields:
721 profile_text += func.coalesce(field, "") # type: ignore[assignment]
722 text_length = func.length(profile_text)
723 home_text = ""
724 for field in home_fields:
725 home_text += func.coalesce(field, "") # type: ignore[assignment]
726 home_length = func.length(home_text)
728 filled_profile = int_(has_completed_profile_expression())
729 has_text = int_(text_length > 500)
730 long_text = int_(text_length > 2000)
731 can_host = int_(User.hosting_status == HostingStatus.can_host)
732 may_host = int_(User.hosting_status == HostingStatus.maybe)
733 cant_host = int_(User.hosting_status == HostingStatus.cant_host)
734 filled_home = int_(User.has_completed_my_home)
735 filled_home_lots = int_(home_length > 200)
736 hosting_status_points = 5 * can_host - 5 * may_host - 10 * cant_host
737 profile_points = 5 * filled_profile + 2 * has_text + 3 * long_text + 5 * filled_home + 10 * filled_home_lots
739 # references
740 left_ref_expr = int_(1).label("left_reference")
741 left_refs_subquery = (
742 select(Reference.from_user_id.label("user_id"), left_ref_expr).group_by(Reference.from_user_id).subquery()
743 )
744 left_reference = int_(left_refs_subquery.c.left_reference)
745 has_reference_expr = int_(func.count(Reference.id) >= 1).label("has_reference")
746 ref_count_expr = int_(func.count(Reference.id)).label("ref_count")
747 ref_avg_expr = func.avg(1.4 * (Reference.rating - 0.3)).label("ref_avg")
748 has_multiple_types_expr = int_(func.count(distinct(Reference.reference_type)) >= 2).label("has_multiple_types")
749 has_bad_ref_expr = int_(func.sum(int_((Reference.rating <= 0.2) | (~Reference.was_appropriate))) >= 1).label(
750 "has_bad_ref"
751 )
752 received_ref_subquery = (
753 select(
754 Reference.to_user_id.label("user_id"),
755 has_reference_expr,
756 has_multiple_types_expr,
757 has_bad_ref_expr,
758 ref_count_expr,
759 ref_avg_expr,
760 )
761 .group_by(Reference.to_user_id)
762 .subquery()
763 )
764 has_multiple_types = int_(received_ref_subquery.c.has_multiple_types)
765 has_reference = int_(received_ref_subquery.c.has_reference)
766 has_bad_reference = int_(received_ref_subquery.c.has_bad_ref)
767 rating_score = float_(
768 received_ref_subquery.c.ref_avg
769 * (
770 2 * func.least(received_ref_subquery.c.ref_count, 5)
771 + func.greatest(received_ref_subquery.c.ref_count - 5, 0)
772 )
773 )
774 ref_score = 2 * has_reference + has_multiple_types + left_reference - 5 * has_bad_reference + rating_score
776 # activeness
777 recently_active = int_(User.last_active >= now() - timedelta(days=180))
778 very_recently_active = int_(User.last_active >= now() - timedelta(days=14))
779 recently_messaged = int_(func.max(Message.time) > now() - timedelta(days=14))
780 messaged_lots = int_(func.count(Message.id) > 5)
781 messaging_points_subquery = (recently_messaged + messaged_lots).label("messaging_points")
782 messaging_subquery = (
783 select(Message.author_id.label("user_id"), messaging_points_subquery)
784 .where(Message.message_type == MessageType.text)
785 .group_by(Message.author_id)
786 .subquery()
787 )
788 activeness_points = recently_active + 2 * very_recently_active + int_(messaging_subquery.c.messaging_points)
790 # verification
791 cb_subquery = (
792 select(ClusterSubscription.user_id.label("user_id"), func.min(Cluster.parent_node_id).label("min_node_id"))
793 .join(Cluster, Cluster.id == ClusterSubscription.cluster_id)
794 .where(ClusterSubscription.role == ClusterRole.admin)
795 .where(Cluster.is_official_cluster)
796 .group_by(ClusterSubscription.user_id)
797 .subquery()
798 )
799 min_node_id = cb_subquery.c.min_node_id
800 cb = int_(min_node_id >= 1)
801 wcb = int_(min_node_id == 1)
802 badge_points = {
803 "founder": 100,
804 "board_member": 20,
805 "past_board_member": 5,
806 "strong_verification": 3,
807 "volunteer": 3,
808 "past_volunteer": 2,
809 "donor": 1,
810 "phone_verified": 1,
811 }
813 badge_subquery = (
814 select(
815 UserBadge.user_id.label("user_id"),
816 func.sum(case(badge_points, value=UserBadge.badge_id, else_=0)).label("badge_points"),
817 )
818 .group_by(UserBadge.user_id)
819 .subquery()
820 )
822 other_points = 0.0 + 10 * wcb + 5 * cb + int_(badge_subquery.c.badge_points)
824 # response rate
825 hr_subquery = select(
826 UserResponseRate.user_id,
827 float_(extract("epoch", UserResponseRate.response_time_33p) / 60.0).label("response_time_33p"),
828 float_(extract("epoch", UserResponseRate.response_time_66p) / 60.0).label("response_time_66p"),
829 ).subquery()
830 response_time_33p = hr_subquery.c.response_time_33p
831 response_time_66p = hr_subquery.c.response_time_66p
832 # be careful with nulls
833 response_rate_points = -10 * int_(response_time_33p > 60 * 96.0) + 5 * int_(response_time_66p < 60 * 96.0)
835 recommendation_score = (
836 hosting_status_points
837 + profile_points
838 + ref_score
839 + activeness_points
840 + other_points
841 + response_rate_points
842 + 2 * poor_man_gaussian()
843 )
845 scores = (
846 select(User.id.label("user_id"), recommendation_score.label("score"))
847 .outerjoin(messaging_subquery, messaging_subquery.c.user_id == User.id)
848 .outerjoin(left_refs_subquery, left_refs_subquery.c.user_id == User.id)
849 .outerjoin(badge_subquery, badge_subquery.c.user_id == User.id)
850 .outerjoin(received_ref_subquery, received_ref_subquery.c.user_id == User.id)
851 .outerjoin(cb_subquery, cb_subquery.c.user_id == User.id)
852 .outerjoin(hr_subquery, hr_subquery.c.user_id == User.id)
853 ).subquery()
855 session.execute(update(User).values(recommendation_score=scores.c.score).where(User.id == scores.c.user_id))
857 logger.info("Updated recommendation scores")
860def update_badges(payload: empty_pb2.Empty) -> None:
861 with session_scope() as session:
863 def update_badge(badge_id: str, members: Sequence[int]) -> None:
864 badge = get_badge_dict()[badge_id]
865 # this batch job has no per-user context to evaluate the gate against, so it's global
866 if badge.flag is not None and not experimentation.get_global_boolean_value(badge.flag, default=True):
867 members = []
868 user_ids = session.execute(select(UserBadge.user_id).where(UserBadge.badge_id == badge.id)).scalars().all()
869 # in case the user ids don't exist in the db
870 actual_members = session.execute(select(User.id).where(User.id.in_(members))).scalars().all()
871 # we should add the badge to these
872 add = set(actual_members) - set(user_ids)
873 # we should remove the badge from these
874 remove = set(user_ids) - set(actual_members)
875 for user_id in add:
876 user_add_badge(session, user_id, badge.id)
878 for user_id in remove:
879 user_remove_badge(session, user_id, badge.id)
881 update_badge("founder", get_static_badge_dict()["founder"])
882 update_badge("board_member", get_static_badge_dict()["board_member"])
883 update_badge("past_board_member", get_static_badge_dict()["past_board_member"])
884 update_badge("donor", session.execute(select(User.id).where(User.last_donated.is_not(None))).scalars().all())
885 update_badge("moderator", session.execute(select(User.id).where(User.is_superuser)).scalars().all())
886 update_badge("phone_verified", session.execute(select(User.id).where(User.phone_is_verified)).scalars().all())
887 # strong verification requires passport on file + gender/sex correspondence and date of birth match
888 update_badge(
889 "strong_verification",
890 session.execute(
891 select(User.id)
892 .join(StrongVerificationAttempt, StrongVerificationAttempt.user_id == User.id)
893 .where(StrongVerificationAttempt.has_strong_verification(User))
894 )
895 .scalars()
896 .all(),
897 )
898 # volunteer badge for active volunteers (stopped_volunteering is null)
899 update_badge(
900 "volunteer",
901 session.execute(select(Volunteer.user_id).where(Volunteer.stopped_volunteering.is_(None))).scalars().all(),
902 )
903 # past_volunteer badge for past volunteers (stopped_volunteering is not null)
904 update_badge(
905 "past_volunteer",
906 session.execute(select(Volunteer.user_id).where(Volunteer.stopped_volunteering.is_not(None)))
907 .scalars()
908 .all(),
909 )
912def finalize_strong_verification(payload: jobs_pb2.FinalizeStrongVerificationPayload) -> None:
913 with session_scope() as session:
914 verification_attempt = session.execute(
915 select(StrongVerificationAttempt)
916 .where(StrongVerificationAttempt.id == payload.verification_attempt_id)
917 .where(StrongVerificationAttempt.status == StrongVerificationAttemptStatus.in_progress_waiting_on_backend)
918 ).scalar_one()
919 response = requests.post(
920 "https://passportreader.app/api/v1/session.get",
921 auth=(config.IRIS_ID_PUBKEY, config.IRIS_ID_SECRET),
922 json={"id": verification_attempt.iris_session_id},
923 timeout=10,
924 verify="/etc/ssl/certs/ca-certificates.crt",
925 )
926 if response.status_code != 200: 926 ↛ 927line 926 didn't jump to line 927 because the condition on line 926 was never true
927 raise Exception(f"Iris didn't return 200: {response.text}")
928 json_data = response.json()
929 reference_payload = internal_pb2.VerificationReferencePayload.FromString(
930 simple_decrypt("iris_callback", b64decode(json_data["reference"]))
931 )
932 assert verification_attempt.user_id == reference_payload.user_id
933 assert verification_attempt.verification_attempt_token == reference_payload.verification_attempt_token
934 assert verification_attempt.iris_session_id == json_data["id"]
935 assert json_data["state"] == "APPROVED"
937 if json_data["document_type"] != "PASSPORT":
938 verification_attempt.status = StrongVerificationAttemptStatus.failed
939 notify(
940 session,
941 user_id=verification_attempt.user_id,
942 topic_action=NotificationTopicAction.verification__sv_fail,
943 key="",
944 data=notification_data_pb2.VerificationSVFail(
945 reason=notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT
946 ),
947 )
948 return
950 assert json_data["document_type"] == "PASSPORT"
952 expiry_date = date.fromisoformat(json_data["expiry_date"])
953 nationality = json_data["nationality"]
954 last_three_document_chars = json_data["document_number"][-3:]
956 existing_attempt = session.execute(
957 select(StrongVerificationAttempt)
958 .where(StrongVerificationAttempt.passport_expiry_date == expiry_date)
959 .where(StrongVerificationAttempt.passport_nationality == nationality)
960 .where(StrongVerificationAttempt.passport_last_three_document_chars == last_three_document_chars)
961 .order_by(StrongVerificationAttempt.id)
962 .limit(1)
963 ).scalar_one_or_none()
965 verification_attempt.has_minimal_data = True
966 verification_attempt.passport_expiry_date = expiry_date
967 verification_attempt.passport_nationality = nationality
968 verification_attempt.passport_last_three_document_chars = last_three_document_chars
970 if existing_attempt:
971 verification_attempt.status = StrongVerificationAttemptStatus.duplicate
973 if existing_attempt.user_id != verification_attempt.user_id:
974 session.flush()
975 send_duplicate_strong_verification_email(session, existing_attempt, verification_attempt)
977 notify(
978 session,
979 user_id=verification_attempt.user_id,
980 topic_action=NotificationTopicAction.verification__sv_fail,
981 key="",
982 data=notification_data_pb2.VerificationSVFail(reason=notification_data_pb2.SV_FAIL_REASON_DUPLICATE),
983 )
984 return
986 verification_attempt.has_full_data = True
987 verification_attempt.passport_encrypted_data = asym_encrypt(
988 config.VERIFICATION_DATA_PUBLIC_KEY, response.text.encode("utf8")
989 )
990 verification_attempt.passport_date_of_birth = date.fromisoformat(json_data["date_of_birth"])
991 verification_attempt.passport_sex = PassportSex[json_data["sex"].lower()]
992 verification_attempt.status = StrongVerificationAttemptStatus.succeeded
994 session.flush()
996 strong_verification_completions_counter.inc()
998 user = verification_attempt.user
999 if verification_attempt.has_strong_verification(user):
1000 badge_id = "strong_verification"
1001 if session.execute(
1002 select(UserBadge).where(UserBadge.user_id == user.id, UserBadge.badge_id == badge_id)
1003 ).scalar_one_or_none():
1004 return
1006 user_add_badge(session, user.id, badge_id, do_notify=False)
1007 notify(
1008 session,
1009 user_id=verification_attempt.user_id,
1010 topic_action=NotificationTopicAction.verification__sv_success,
1011 key="",
1012 )
1013 else:
1014 birthdate_ok = verification_attempt.matches_birthdate(user)
1015 gender_ok = verification_attempt.matches_gender(user)
1016 if not birthdate_ok and not gender_ok:
1017 reason = notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE_OR_GENDER
1018 elif not birthdate_ok:
1019 reason = notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE
1020 else:
1021 reason = notification_data_pb2.SV_FAIL_REASON_WRONG_GENDER
1022 notify(
1023 session,
1024 user_id=verification_attempt.user_id,
1025 topic_action=NotificationTopicAction.verification__sv_fail,
1026 key="",
1027 data=notification_data_pb2.VerificationSVFail(reason=reason),
1028 )
1031def send_activeness_probes(payload: empty_pb2.Empty) -> None:
1032 with session_scope() as session:
1033 ## Step 1: create new activeness probes for those who need it and don't have one (if enabled)
1035 if config.ACTIVENESS_PROBES_ENABLED:
1036 # current activeness probes
1037 subquery = select(ActivenessProbe.user_id).where(ActivenessProbe.responded == None).subquery()
1039 # users who we should send an activeness probe to
1040 new_probe_user_ids = (
1041 session.execute(
1042 select(User.id)
1043 .where(User.is_visible)
1044 .where(User.hosting_status == HostingStatus.can_host)
1045 .where(User.last_active < func.now() - ACTIVENESS_PROBE_INACTIVITY_PERIOD)
1046 .where(User.id.not_in(select(subquery.c.user_id)))
1047 )
1048 .scalars()
1049 .all()
1050 )
1052 total_users = session.execute(select(func.count()).select_from(User).where(User.is_visible)).scalar_one()
1053 probes_today = session.execute(
1054 select(func.count())
1055 .select_from(ActivenessProbe)
1056 .where(func.now() - ActivenessProbe.probe_initiated < timedelta(hours=24))
1057 ).scalar_one()
1059 # send probes to max 2% of users per day
1060 max_probes_per_day = 0.02 * total_users
1061 max_probe_size = int(max(min(max_probes_per_day - probes_today, max_probes_per_day / 24), 1))
1063 if len(new_probe_user_ids) > max_probe_size: 1063 ↛ 1064line 1063 didn't jump to line 1064 because the condition on line 1063 was never true
1064 new_probe_user_ids = sample(new_probe_user_ids, max_probe_size)
1066 for user_id in new_probe_user_ids:
1067 session.add(ActivenessProbe(user_id=user_id))
1069 session.commit()
1071 ## Step 2: actually send out probe notifications
1072 for probe_number_minus_1, delay in enumerate(ACTIVENESS_PROBE_TIME_REMINDERS):
1073 probes = (
1074 session.execute(
1075 select(ActivenessProbe)
1076 .where(ActivenessProbe.notifications_sent == probe_number_minus_1)
1077 .where(ActivenessProbe.probe_initiated + delay < func.now())
1078 .where(ActivenessProbe.is_pending)
1079 )
1080 .scalars()
1081 .all()
1082 )
1084 for probe in probes:
1085 probe.notifications_sent = probe_number_minus_1 + 1
1086 context = make_notification_user_context(user_id=probe.user.id)
1087 notify(
1088 session,
1089 user_id=probe.user.id,
1090 topic_action=NotificationTopicAction.activeness__probe,
1091 key=str(probe.id),
1092 data=notification_data_pb2.ActivenessProbe(
1093 reminder_number=probe_number_minus_1 + 1,
1094 deadline=Timestamp_from_datetime(probe.probe_initiated + ACTIVENESS_PROBE_EXPIRY_TIME),
1095 ),
1096 )
1097 session.commit()
1099 ## Step 3: for those who haven't responded, mark them as failed
1100 expired_probes = (
1101 session.execute(
1102 select(ActivenessProbe)
1103 .where(ActivenessProbe.notifications_sent == len(ACTIVENESS_PROBE_TIME_REMINDERS))
1104 .where(ActivenessProbe.is_pending)
1105 .where(ActivenessProbe.probe_initiated + ACTIVENESS_PROBE_EXPIRY_TIME < func.now())
1106 )
1107 .scalars()
1108 .all()
1109 )
1111 for probe in expired_probes:
1112 probe.responded = now()
1113 probe.response = ActivenessProbeStatus.expired
1114 if probe.user.hosting_status == HostingStatus.can_host: 1114 ↛ 1116line 1114 didn't jump to line 1116 because the condition on line 1114 was always true
1115 probe.user.hosting_status = HostingStatus.maybe
1116 if probe.user.meetup_status == MeetupStatus.wants_to_meetup: 1116 ↛ 1118line 1116 didn't jump to line 1118 because the condition on line 1116 was always true
1117 probe.user.meetup_status = MeetupStatus.open_to_meetup
1118 session.commit()
1121def update_randomized_locations(payload: empty_pb2.Empty) -> None:
1122 """
1123 We generate for each user a randomized location as follows:
1124 - Start from a strong random seed (based on the SECRET env var and our key derivation function)
1125 - For each user, mix in the user_id for randomness
1126 - Generate a radius from [0.02, 0.1] degrees (about 2-10km)
1127 - Generate an angle from [0, 360]
1128 - Randomized location is then a distance `radius` away at an angle `angle` from `geom`
1129 """
1130 randomization_secret = get_secret(USER_LOCATION_RANDOMIZATION_NAME)
1132 def gen_randomized_coords(user_id: int, lat: float, lng: float) -> tuple[float, float]:
1133 radius_u = stable_secure_uniform(randomization_secret, seed=bytes(f"{user_id}|radius", "ascii"))
1134 angle_u = stable_secure_uniform(randomization_secret, seed=bytes(f"{user_id}|angle", "ascii"))
1135 radius = 0.02 + 0.08 * radius_u
1136 angle_rad = 2 * pi * angle_u
1137 offset_lng = radius * cos(angle_rad)
1138 offset_lat = radius * sin(angle_rad)
1139 return lat + offset_lat, lng + offset_lng
1141 user_updates: list[dict[str, Any]] = []
1143 with session_scope() as session:
1144 users_to_update = session.execute(select(User.id, User.geom).where(User.randomized_geom == None)).all()
1146 for user_id, geom in users_to_update:
1147 lat, lng = get_coordinates(geom)
1148 user_updates.append(
1149 {"id": user_id, "randomized_geom": create_coordinate(*gen_randomized_coords(user_id, lat, lng))}
1150 )
1152 with session_scope() as session:
1153 session.execute(update(User), user_updates)
1156def send_event_reminders(payload: empty_pb2.Empty) -> None:
1157 """
1158 Sends reminders for events that are 24 hours away to users who marked themselves as attending.
1159 """
1160 logger.info("Sending event reminder emails")
1162 with session_scope() as session:
1163 occurrences = (
1164 session.execute(
1165 select(EventOccurrence)
1166 .where(EventOccurrence.start_time <= now() + EVENT_REMINDER_TIMEDELTA)
1167 .where(EventOccurrence.start_time >= now())
1168 .where(~EventOccurrence.is_cancelled)
1169 .where(~EventOccurrence.is_deleted)
1170 )
1171 .scalars()
1172 .all()
1173 )
1175 for occurrence in occurrences:
1176 results = session.execute(
1177 select(User, EventOccurrenceAttendee)
1178 .join(EventOccurrenceAttendee, EventOccurrenceAttendee.user_id == User.id)
1179 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id)
1180 .where(EventOccurrenceAttendee.reminder_sent == False)
1181 .where(User.is_visible)
1182 .where(~User.is_shadowed)
1183 ).all()
1185 for user, attendee in results:
1186 context = make_notification_user_context(user_id=user.id)
1188 notify(
1189 session,
1190 user_id=user.id,
1191 topic_action=NotificationTopicAction.event__reminder,
1192 key=str(occurrence.id),
1193 data=notification_data_pb2.EventReminder(
1194 event=event_to_pb(session, occurrence, context),
1195 user=user_model_to_pb(user, session, context),
1196 ),
1197 moderation_state_id=occurrence.moderation_state_id,
1198 )
1200 attendee.reminder_sent = True
1201 session.commit()
1204def check_expo_push_receipts(payload: empty_pb2.Empty) -> None:
1205 """
1206 Check Expo push receipts in batch and update delivery attempts.
1207 """
1208 MAX_ITERATIONS = 100 # Safety limit: 100 batches * 100 attempts = 10,000 max
1210 for iteration in range(MAX_ITERATIONS): 1210 ↛ 1272line 1210 didn't jump to line 1272 because the loop on line 1210 didn't complete
1211 with session_scope() as session:
1212 # Find all delivery attempts that need receipt checking
1213 # Wait 15 minutes per Expo's recommendation before checking receipts
1214 attempts = (
1215 session.execute(
1216 select(PushNotificationDeliveryAttempt)
1217 .where(PushNotificationDeliveryAttempt.expo_ticket_id != None)
1218 .where(PushNotificationDeliveryAttempt.receipt_checked_at == None)
1219 .where(PushNotificationDeliveryAttempt.time < now() - timedelta(minutes=15))
1220 .where(PushNotificationDeliveryAttempt.time > now() - timedelta(hours=24))
1221 .limit(100)
1222 )
1223 .scalars()
1224 .all()
1225 )
1227 if not attempts:
1228 logger.debug("No Expo receipts to check")
1229 return
1231 logger.info(f"Checking {len(attempts)} Expo push receipts")
1233 receipts = get_expo_push_receipts([not_none(attempt.expo_ticket_id) for attempt in attempts])
1235 for attempt in attempts:
1236 receipt = receipts.get(not_none(attempt.expo_ticket_id))
1238 # Always mark as checked to avoid infinite loops
1239 attempt.receipt_checked_at = now()
1241 if receipt is None:
1242 # Receipt not found after 15min - likely expired (>24h) or never existed
1243 # Per Expo docs: receipts should be available within 15 minutes
1244 attempt.receipt_status = "not_found"
1245 continue
1247 attempt.receipt_status = receipt.get("status")
1249 if receipt.get("status") == "error":
1250 details = receipt.get("details", {})
1251 error_code = details.get("error")
1252 attempt.receipt_error_code = error_code
1254 if error_code == "DeviceNotRegistered": 1254 ↛ 1269line 1254 didn't jump to line 1269 because the condition on line 1254 was always true
1255 # Device token is no longer valid - disable the subscription
1256 sub = session.execute(
1257 select(PushNotificationSubscription).where(
1258 PushNotificationSubscription.id == attempt.push_notification_subscription_id
1259 )
1260 ).scalar_one()
1262 if sub.disabled_at > now(): 1262 ↛ 1235line 1262 didn't jump to line 1235 because the condition on line 1262 was always true
1263 sub.disabled_at = now()
1264 logger.info(f"Disabled push sub {sub.id} due to DeviceNotRegistered in receipt")
1265 push_notification_counter.labels(
1266 platform="expo", outcome="permanent_subscription_failure_receipt"
1267 ).inc()
1268 else:
1269 logger.warning(f"Expo receipt error for ticket {attempt.expo_ticket_id}: {error_code}")
1271 # If we get here, we've exhausted MAX_ITERATIONS without finishing
1272 raise RuntimeError(
1273 f"check_expo_push_receipts exceeded {MAX_ITERATIONS} iterations - "
1274 "there may be an unusually large backlog of receipts to check"
1275 )
1278def send_postal_verification_postcard(payload: jobs_pb2.SendPostalVerificationPostcardPayload) -> None:
1279 """
1280 Sends the postcard via external API and updates attempt status.
1281 """
1282 with session_scope() as session:
1283 attempt = session.execute(
1284 select(PostalVerificationAttempt).where(
1285 PostalVerificationAttempt.id == payload.postal_verification_attempt_id
1286 )
1287 ).scalar_one_or_none()
1289 if not attempt or attempt.status != PostalVerificationStatus.in_progress: 1289 ↛ 1290line 1289 didn't jump to line 1290 because the condition on line 1289 was never true
1290 logger.warning(
1291 f"Postal verification attempt {payload.postal_verification_attempt_id} not found or wrong state"
1292 )
1293 return
1295 user_name = session.execute(select(User.name).where(User.id == attempt.user_id)).scalar_one()
1297 job_id = send_postcard(
1298 recipient_name=user_name,
1299 address_line_1=attempt.address_line_1,
1300 address_line_2=attempt.address_line_2,
1301 city=attempt.city,
1302 state=attempt.state,
1303 postal_code=attempt.postal_code,
1304 country=attempt.country_code,
1305 verification_code=not_none(attempt.verification_code),
1306 )
1308 attempt.mypostcard_job_id = job_id
1309 attempt.status = PostalVerificationStatus.awaiting_verification
1310 attempt.postcard_sent_at = func.now()
1312 postcards_sent_counter.labels(country_code=attempt.country_code).inc()
1314 context = make_background_user_context(attempt.user_id)
1315 log_event(
1316 context,
1317 session,
1318 "postcard.sent",
1319 {
1320 "attempt_id": attempt.id,
1321 "country": attempt.country_code,
1322 "city": attempt.city,
1323 "mypostcard_job_id": job_id,
1324 },
1325 )
1327 notify(
1328 session,
1329 user_id=attempt.user_id,
1330 topic_action=NotificationTopicAction.postal_verification__postcard_sent,
1331 key="",
1332 data=notification_data_pb2.PostalVerificationPostcardSent(
1333 city=attempt.city,
1334 country=attempt.country_code,
1335 ),
1336 )
1339def check_mypostcard_jobs(payload: empty_pb2.Empty) -> None:
1340 """
1341 Checks that all MyPostcard jobs from the last week are tied to a postal verification attempt.
1342 """
1343 if not experimentation.get_global_boolean_value("postal_verification_enabled", default=False):
1344 return
1346 with session_scope() as session:
1347 mypostcard_job_ids = set(
1348 get_order_ids(
1349 date_from=(now() - timedelta(days=7)).date(),
1350 date_to=now().date(),
1351 )
1352 )
1354 known_job_ids = set(
1355 session.execute(
1356 select(PostalVerificationAttempt.mypostcard_job_id).where(
1357 PostalVerificationAttempt.mypostcard_job_id.isnot(None),
1358 PostalVerificationAttempt.created >= now() - timedelta(days=14),
1359 )
1360 )
1361 .scalars()
1362 .all()
1363 )
1365 orphaned = mypostcard_job_ids - known_job_ids
1366 if orphaned:
1367 report_message(
1368 f"Found {len(orphaned)} orphaned MyPostcard jobs not tied to any verification attempt: {orphaned}"
1369 )
1372class DatabaseInconsistencyError(Exception):
1373 """Raised when database consistency checks fail"""
1375 pass
1378def check_database_consistency(payload: empty_pb2.Empty) -> None:
1379 """
1380 Checks database consistency and raises an exception if any issues are found.
1381 """
1382 logger.info("Checking database consistency")
1383 errors = []
1385 with session_scope() as session:
1386 # Check that all users have a profile gallery
1387 users_without_gallery = session.execute(
1388 select(User.id, User.username).where(User.profile_gallery_id.is_(None))
1389 ).all()
1390 if users_without_gallery:
1391 errors.append(f"Users without profile gallery: {users_without_gallery}")
1393 # Check that all profile galleries point to their owner
1394 mismatched_galleries = session.execute(
1395 select(User.id, User.username, User.profile_gallery_id, PhotoGallery.owner_user_id)
1396 .join(PhotoGallery, User.profile_gallery_id == PhotoGallery.id)
1397 .where(User.profile_gallery_id.is_not(None))
1398 .where(PhotoGallery.owner_user_id != User.id)
1399 ).all()
1400 if mismatched_galleries: 1400 ↛ 1401line 1400 didn't jump to line 1401 because the condition on line 1400 was never true
1401 errors.append(f"Profile galleries with mismatched owner: {mismatched_galleries}")
1403 # === Moderation System Consistency Checks ===
1405 # Check every ModerationState has at least one INITIAL_REVIEW queue item
1406 # Skip items with ID < 2000000 as they were created before this check was introduced
1407 states_without_initial_review = session.execute(
1408 select(ModerationState.id, ModerationState.object_type, ModerationState.object_id).where(
1409 ModerationState.id >= 2000000,
1410 ~exists(
1411 select(1)
1412 .where(ModerationQueueItem.moderation_state_id == ModerationState.id)
1413 .where(ModerationQueueItem.trigger == ModerationTrigger.initial_review)
1414 ),
1415 )
1416 ).all()
1417 if states_without_initial_review: 1417 ↛ 1418line 1417 didn't jump to line 1418 because the condition on line 1417 was never true
1418 errors.append(f"ModerationStates without INITIAL_REVIEW queue item: {states_without_initial_review}")
1420 # Check every ModerationState has a CREATE log entry
1421 # Skip items with ID < 2000000 as they were created before this check was introduced
1422 states_without_create_log = session.execute(
1423 select(ModerationState.id, ModerationState.object_type, ModerationState.object_id).where(
1424 ModerationState.id >= 2000000,
1425 ~exists(
1426 select(1)
1427 .where(ModerationLog.moderation_state_id == ModerationState.id)
1428 .where(ModerationLog.action == ModerationAction.create)
1429 ),
1430 )
1431 ).all()
1432 if states_without_create_log: 1432 ↛ 1433line 1432 didn't jump to line 1433 because the condition on line 1432 was never true
1433 errors.append(f"ModerationStates without CREATE log entry: {states_without_create_log}")
1435 # Check resolved queue items point to log entries for the same moderation state
1436 resolved_item_log_mismatches = session.execute(
1437 select(ModerationQueueItem.id, ModerationQueueItem.moderation_state_id, ModerationLog.moderation_state_id)
1438 .join(ModerationLog, ModerationQueueItem.resolved_by_log_id == ModerationLog.id)
1439 .where(ModerationQueueItem.resolved_by_log_id.is_not(None))
1440 .where(ModerationQueueItem.moderation_state_id != ModerationLog.moderation_state_id)
1441 ).all()
1442 if resolved_item_log_mismatches: 1442 ↛ 1443line 1442 didn't jump to line 1443 because the condition on line 1442 was never true
1443 errors.append(f"Resolved queue items with mismatched moderation_state_id: {resolved_item_log_mismatches}")
1445 # Check every HOST_REQUEST ModerationState has exactly one HostRequest pointing to it
1446 hr_states = (
1447 session.execute(
1448 select(ModerationState.id).where(ModerationState.object_type == ModerationObjectType.host_request)
1449 )
1450 .scalars()
1451 .all()
1452 )
1453 for state_id in hr_states: 1453 ↛ 1454line 1453 didn't jump to line 1454 because the loop on line 1453 never started
1454 hr_count = session.execute(
1455 select(func.count()).where(HostRequest.moderation_state_id == state_id)
1456 ).scalar_one()
1457 if hr_count != 1:
1458 errors.append(f"ModerationState {state_id} (HOST_REQUEST) has {hr_count} HostRequests (expected 1)")
1460 # Check every GROUP_CHAT ModerationState has exactly one GroupChat pointing to it
1461 gc_states = (
1462 session.execute(
1463 select(ModerationState.id).where(ModerationState.object_type == ModerationObjectType.group_chat)
1464 )
1465 .scalars()
1466 .all()
1467 )
1468 for state_id in gc_states:
1469 gc_count = session.execute(
1470 select(func.count()).where(GroupChat.moderation_state_id == state_id)
1471 ).scalar_one()
1472 if gc_count != 1: 1472 ↛ 1473line 1472 didn't jump to line 1473 because the condition on line 1472 was never true
1473 errors.append(f"ModerationState {state_id} (GROUP_CHAT) has {gc_count} GroupChats (expected 1)")
1475 # Check ModerationState.object_id matches the actual object's ID
1476 hr_object_id_mismatches = session.execute(
1477 select(ModerationState.id, ModerationState.object_id, HostRequest.conversation_id)
1478 .join(HostRequest, HostRequest.moderation_state_id == ModerationState.id)
1479 .where(ModerationState.object_type == ModerationObjectType.host_request)
1480 .where(ModerationState.object_id != HostRequest.conversation_id)
1481 ).all()
1482 if hr_object_id_mismatches: 1482 ↛ 1483line 1482 didn't jump to line 1483 because the condition on line 1482 was never true
1483 errors.append(f"ModerationState object_id mismatch for HOST_REQUEST: {hr_object_id_mismatches}")
1485 gc_object_id_mismatches = session.execute(
1486 select(ModerationState.id, ModerationState.object_id, GroupChat.conversation_id)
1487 .join(GroupChat, GroupChat.moderation_state_id == ModerationState.id)
1488 .where(ModerationState.object_type == ModerationObjectType.group_chat)
1489 .where(ModerationState.object_id != GroupChat.conversation_id)
1490 ).all()
1491 if gc_object_id_mismatches: 1491 ↛ 1492line 1491 didn't jump to line 1492 because the condition on line 1491 was never true
1492 errors.append(f"ModerationState object_id mismatch for GROUP_CHAT: {gc_object_id_mismatches}")
1494 # Check reverse mapping: HostRequest's moderation_state points to correct ModerationState
1495 hr_reverse_mismatches = session.execute(
1496 select(
1497 HostRequest.conversation_id,
1498 HostRequest.moderation_state_id,
1499 ModerationState.object_type,
1500 ModerationState.object_id,
1501 )
1502 .join(ModerationState, HostRequest.moderation_state_id == ModerationState.id)
1503 .where(
1504 (ModerationState.object_type != ModerationObjectType.host_request)
1505 | (ModerationState.object_id != HostRequest.conversation_id)
1506 )
1507 ).all()
1508 if hr_reverse_mismatches: 1508 ↛ 1509line 1508 didn't jump to line 1509 because the condition on line 1508 was never true
1509 errors.append(f"HostRequest points to ModerationState with wrong type/object_id: {hr_reverse_mismatches}")
1511 # Check reverse mapping: GroupChat's moderation_state points to correct ModerationState
1512 gc_reverse_mismatches = session.execute(
1513 select(
1514 GroupChat.conversation_id,
1515 GroupChat.moderation_state_id,
1516 ModerationState.object_type,
1517 ModerationState.object_id,
1518 )
1519 .join(ModerationState, GroupChat.moderation_state_id == ModerationState.id)
1520 .where(
1521 (ModerationState.object_type != ModerationObjectType.group_chat)
1522 | (ModerationState.object_id != GroupChat.conversation_id)
1523 )
1524 ).all()
1525 if gc_reverse_mismatches: 1525 ↛ 1526line 1525 didn't jump to line 1526 because the condition on line 1525 was never true
1526 errors.append(f"GroupChat points to ModerationState with wrong type/object_id: {gc_reverse_mismatches}")
1528 # Ensure auto-approve deadline isn't being exceeded by a significant margin
1529 # The auto-approver runs every 15s, so allow 5 minutes grace before alerting
1530 deadline_seconds = config.MODERATION_AUTO_APPROVE_DEADLINE_SECONDS
1531 if deadline_seconds > 0: 1531 ↛ 1532line 1531 didn't jump to line 1532 because the condition on line 1531 was never true
1532 grace_period = timedelta(minutes=5)
1533 stale_initial_review_items = session.execute(
1534 select(
1535 ModerationQueueItem.id,
1536 ModerationQueueItem.moderation_state_id,
1537 ModerationQueueItem.time_created,
1538 )
1539 .where(ModerationQueueItem.trigger == ModerationTrigger.initial_review)
1540 .where(ModerationQueueItem.resolved_by_log_id.is_(None))
1541 .where(ModerationQueueItem.time_created < now() - timedelta(seconds=deadline_seconds) - grace_period)
1542 ).all()
1543 if stale_initial_review_items:
1544 errors.append(
1545 f"INITIAL_REVIEW items exceeding auto-approve deadline by >5min: {stale_initial_review_items}"
1546 )
1548 if errors:
1549 raise DatabaseInconsistencyError("\n".join(errors))
1552def auto_approve_moderation_queue(payload: empty_pb2.Empty) -> None:
1553 """
1554 Dead man's switch: approves unresolved INITIAL_REVIEW content older than the deadline to VISIBLE, then
1555 re-flags it as a high-priority MACHINE_FLAG superseding only the INITIAL_REVIEW item. The switch only fires
1556 when moderators are behind, so every auto-approved item stays in the queue for a human to check. Other open
1557 flags are untouched, and items already actioned by moderators are left alone.
1558 """
1559 deadline_seconds = config.MODERATION_AUTO_APPROVE_DEADLINE_SECONDS
1560 if deadline_seconds <= 0:
1561 return
1563 with session_scope() as session:
1564 ctx = make_background_user_context(user_id=config.MODERATION_BOT_USER_ID)
1566 items = (
1567 Moderation()
1568 .GetModerationQueue(
1569 request=moderation_pb2.GetModerationQueueReq(
1570 triggers=[moderation_pb2.MODERATION_TRIGGER_INITIAL_REVIEW],
1571 unresolved_only=True,
1572 page_size=100,
1573 created_before=Timestamp_from_datetime(now() - timedelta(seconds=deadline_seconds)),
1574 ),
1575 context=ctx,
1576 session=session,
1577 )
1578 .queue_items
1579 )
1581 if not items:
1582 return
1584 # Skip items whose author is shadowed; their content stays in shadowed state indefinitely
1585 approvable = [item for item in items if not item.moderation_state.author.shadowed]
1586 if not approvable:
1587 return
1589 logger.info(f"Auto-approving {len(approvable)} moderation queue items")
1590 reason = f"Auto-approved: moderation deadline of {deadline_seconds} seconds exceeded."
1591 for item in approvable:
1592 Moderation().ModerateContent(
1593 request=moderation_pb2.ModerateContentReq(
1594 moderation_state_id=item.moderation_state_id,
1595 action=moderation_pb2.MODERATION_ACTION_APPROVE,
1596 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
1597 reason=reason,
1598 clear_flags=False,
1599 ),
1600 context=ctx,
1601 session=session,
1602 )
1603 Moderation().ModerateContent(
1604 request=moderation_pb2.ModerateContentReq(
1605 moderation_state_id=item.moderation_state_id,
1606 action=moderation_pb2.MODERATION_ACTION_FLAG,
1607 trigger=moderation_pb2.MODERATION_TRIGGER_MACHINE_FLAG,
1608 priority=MODERATION_AUTO_APPROVE_FLAG_PRIORITY,
1609 reason=reason,
1610 supersede_queue_item_id=item.queue_item_id,
1611 ),
1612 context=ctx,
1613 session=session,
1614 )
1615 moderation_auto_approved_counter.inc(len(approvable))