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