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