Coverage for app/backend/src/couchers/jobs/definitions.py: 100%

1from dataclasses import dataclass 

2from datetime import timedelta 

3from typing import Any, Protocol, cast, get_type_hints 

4 

5from couchers.jobs.handlers import ( 

6 add_users_to_email_list, 

7 auto_approve_moderation_queue, 

8 check_database_consistency, 

9 check_expo_push_receipts, 

10 check_mypostcard_jobs, 

11 enforce_community_membership, 

12 finalize_strong_verification, 

13 purge_account_deletion_tokens, 

14 purge_login_tokens, 

15 purge_password_reset_tokens, 

16 schedule_event_occurrences, 

17 send_activeness_probes, 

18 send_email, 

19 send_event_reminders, 

20 send_host_request_reminders, 

21 send_message_notifications, 

22 send_onboarding_emails, 

23 send_postal_verification_postcard, 

24 send_reference_reminders, 

25 send_request_notifications, 

26 update_badges, 

27 update_randomized_locations, 

28 update_recommendation_scores, 

29) 

30from couchers.materialized_views import refresh_materialized_views, refresh_materialized_views_rapid 

31from couchers.notifications.background import handle_email_digests, handle_notification 

32from couchers.notifications.send_raw_push_notification import send_raw_push_notification_v2 

33from couchers.servicers.conversations import generate_message_notifications 

34from couchers.servicers.discussions import generate_create_discussion_notifications 

35from couchers.servicers.editor import generate_new_blog_post_notifications 

36from couchers.servicers.events import ( 

37 generate_event_cancel_notifications, 

38 generate_event_create_notifications, 

39 generate_event_delete_notifications, 

40 generate_event_update_notifications, 

41) 

42from couchers.servicers.threads import generate_reply_notifications 

43 

44 

45class JobHandler[T](Protocol): 

46 def __call__(self, payload: T) -> None: ... 

47 

48 @property 

49 def __name__(self) -> str: ... 

50 

51 

52@dataclass(frozen=True, slots=True) 

53class Job[T]: 

54 """Definition of a background job.""" 

55 

56 handler: JobHandler[T] 

57 schedule: timedelta | None = None 

58 

59 @property 

60 def name(self) -> str: 

61 return self.handler.__name__ 

62 

63 @property 

64 def payload_type(self) -> type[T]: 

65 """Extracts protobuf type from type hints.""" 

66 return cast(type[T], get_type_hints(self.handler)["payload"]) 

67 

68 

69# Job registry - first create a list of all jobs. 

70_JOBS_LIST = [ 

71 Job(handle_notification), 

72 Job(send_raw_push_notification_v2), 

73 Job(handle_email_digests, schedule=timedelta(minutes=15)), 

74 Job(generate_message_notifications), 

75 Job(generate_reply_notifications), 

76 Job(generate_create_discussion_notifications), 

77 Job(generate_event_create_notifications), 

78 Job(generate_event_update_notifications), 

79 Job(generate_event_cancel_notifications), 

80 Job(generate_event_delete_notifications), 

81 Job(generate_new_blog_post_notifications), 

82 Job(refresh_materialized_views, schedule=timedelta(minutes=5)), 

83 Job(refresh_materialized_views_rapid, schedule=timedelta(seconds=30)), 

84 Job(send_email), 

85 Job(purge_login_tokens, schedule=timedelta(hours=24)), 

86 Job(purge_password_reset_tokens, schedule=timedelta(hours=24)), 

87 Job(purge_account_deletion_tokens, schedule=timedelta(hours=24)), 

88 Job(send_message_notifications, schedule=timedelta(minutes=3)), 

89 Job(send_request_notifications, schedule=timedelta(minutes=3)), 

90 Job(send_onboarding_emails, schedule=timedelta(hours=1)), 

91 Job(send_reference_reminders, schedule=timedelta(hours=1)), 

92 Job(send_host_request_reminders, schedule=timedelta(minutes=15)), 

93 Job(add_users_to_email_list, schedule=timedelta(hours=1)), 

94 Job(enforce_community_membership, schedule=timedelta(minutes=15)), 

95 Job(update_recommendation_scores, schedule=timedelta(hours=24)), 

96 Job(update_badges, schedule=timedelta(minutes=15)), 

97 Job(finalize_strong_verification), 

98 Job(send_activeness_probes, schedule=timedelta(minutes=60)), 

99 Job(update_randomized_locations, schedule=timedelta(hours=1)), 

100 Job(send_event_reminders, schedule=timedelta(hours=1)), 

101 Job(schedule_event_occurrences, schedule=timedelta(hours=1)), 

102 Job(check_expo_push_receipts, schedule=timedelta(minutes=5)), 

103 Job(send_postal_verification_postcard), 

104 Job(check_database_consistency, schedule=timedelta(hours=24)), 

105 Job(check_mypostcard_jobs, schedule=timedelta(hours=24)), 

106 Job(auto_approve_moderation_queue, schedule=timedelta(seconds=15)), 

107] 

108 

109# Map job names to job definitions 

110JOBS: dict[str, Job[Any]] = {job.name: job for job in cast(list[Job[Any]], _JOBS_LIST)}