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