mas_tasks/
lib.rs

1// Copyright 2024, 2025 New Vector Ltd.
2// Copyright 2021-2024 The Matrix.org Foundation C.I.C.
3//
4// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
5// Please see LICENSE files in the repository root for full details.
6
7use std::sync::{Arc, LazyLock};
8
9use mas_data_model::SiteConfig;
10use mas_email::Mailer;
11use mas_matrix::HomeserverConnection;
12use mas_router::UrlBuilder;
13use mas_storage::{BoxClock, BoxRepository, RepositoryError, RepositoryFactory, SystemClock};
14use mas_storage_pg::PgRepositoryFactory;
15use new_queue::QueueRunnerError;
16use opentelemetry::metrics::Meter;
17use rand::SeedableRng;
18use sqlx::{Pool, Postgres};
19use tokio_util::{sync::CancellationToken, task::TaskTracker};
20
21mod database;
22mod email;
23mod matrix;
24mod new_queue;
25mod recovery;
26mod sessions;
27mod user;
28
29static METER: LazyLock<Meter> = LazyLock::new(|| {
30    let scope = opentelemetry::InstrumentationScope::builder(env!("CARGO_PKG_NAME"))
31        .with_version(env!("CARGO_PKG_VERSION"))
32        .with_schema_url(opentelemetry_semantic_conventions::SCHEMA_URL)
33        .build();
34
35    opentelemetry::global::meter_with_scope(scope)
36});
37
38#[derive(Clone)]
39struct State {
40    repository_factory: PgRepositoryFactory,
41    mailer: Mailer,
42    clock: SystemClock,
43    homeserver: Arc<dyn HomeserverConnection>,
44    url_builder: UrlBuilder,
45    site_config: SiteConfig,
46}
47
48impl State {
49    pub fn new(
50        repository_factory: PgRepositoryFactory,
51        clock: SystemClock,
52        mailer: Mailer,
53        homeserver: impl HomeserverConnection + 'static,
54        url_builder: UrlBuilder,
55        site_config: SiteConfig,
56    ) -> Self {
57        Self {
58            repository_factory,
59            mailer,
60            clock,
61            homeserver: Arc::new(homeserver),
62            url_builder,
63            site_config,
64        }
65    }
66
67    pub fn pool(&self) -> Pool<Postgres> {
68        self.repository_factory.pool()
69    }
70
71    pub fn clock(&self) -> BoxClock {
72        Box::new(self.clock.clone())
73    }
74
75    pub fn mailer(&self) -> &Mailer {
76        &self.mailer
77    }
78
79    // This is fine for now, we may move that to a trait at some point.
80    #[allow(clippy::unused_self, clippy::disallowed_methods)]
81    pub fn rng(&self) -> rand_chacha::ChaChaRng {
82        rand_chacha::ChaChaRng::from_rng(rand::thread_rng()).expect("failed to seed rng")
83    }
84
85    pub async fn repository(&self) -> Result<BoxRepository, RepositoryError> {
86        self.repository_factory.create().await
87    }
88
89    pub fn matrix_connection(&self) -> &dyn HomeserverConnection {
90        self.homeserver.as_ref()
91    }
92
93    pub fn url_builder(&self) -> &UrlBuilder {
94        &self.url_builder
95    }
96
97    pub fn site_config(&self) -> &SiteConfig {
98        &self.site_config
99    }
100}
101
102/// Initialise the workers.
103///
104/// # Errors
105///
106/// This function can fail if the database connection fails.
107pub async fn init(
108    repository_factory: PgRepositoryFactory,
109    mailer: &Mailer,
110    homeserver: impl HomeserverConnection + 'static,
111    url_builder: UrlBuilder,
112    site_config: &SiteConfig,
113    cancellation_token: CancellationToken,
114    task_tracker: &TaskTracker,
115) -> Result<(), QueueRunnerError> {
116    let state = State::new(
117        repository_factory,
118        SystemClock::default(),
119        mailer.clone(),
120        homeserver,
121        url_builder,
122        site_config.clone(),
123    );
124    let mut worker = self::new_queue::QueueWorker::new(state, cancellation_token).await?;
125
126    worker
127        .register_handler::<mas_storage::queue::CleanupExpiredTokensJob>()
128        .register_handler::<mas_storage::queue::DeactivateUserJob>()
129        .register_handler::<mas_storage::queue::DeleteDeviceJob>()
130        .register_handler::<mas_storage::queue::ProvisionDeviceJob>()
131        .register_handler::<mas_storage::queue::ProvisionUserJob>()
132        .register_handler::<mas_storage::queue::ReactivateUserJob>()
133        .register_handler::<mas_storage::queue::SendAccountRecoveryEmailsJob>()
134        .register_handler::<mas_storage::queue::SendEmailAuthenticationCodeJob>()
135        .register_handler::<mas_storage::queue::SyncDevicesJob>()
136        .register_handler::<mas_storage::queue::VerifyEmailJob>()
137        .register_handler::<mas_storage::queue::ExpireInactiveSessionsJob>()
138        .register_handler::<mas_storage::queue::ExpireInactiveCompatSessionsJob>()
139        .register_handler::<mas_storage::queue::ExpireInactiveOAuthSessionsJob>()
140        .register_handler::<mas_storage::queue::ExpireInactiveUserSessionsJob>()
141        .register_handler::<mas_storage::queue::PruneStalePolicyDataJob>()
142        .add_schedule(
143            "cleanup-expired-tokens",
144            "0 0 * * * *".parse()?,
145            mas_storage::queue::CleanupExpiredTokensJob,
146        )
147        .add_schedule(
148            "expire-inactive-sessions",
149            // Run this job every 15 minutes
150            "30 */15 * * * *".parse()?,
151            mas_storage::queue::ExpireInactiveSessionsJob,
152        )
153        .add_schedule(
154            "prune-stale-policy-data",
155            // Run once a day
156            "0 0 2 * * *".parse()?,
157            mas_storage::queue::PruneStalePolicyDataJob,
158        );
159
160    task_tracker.spawn(worker.run());
161
162    Ok(())
163}
mas_tasks/lib.rs

mas_tasks/
lib.rs
