Proactively provision users on registration & sync threepids

crates/tasks/Cargo.toml

@@ -20,9 +20,12 @@ tower = "0.4.13"
 tracing = "0.1.37"
 tracing-opentelemetry = "0.18.0"
 ulid = "1.0.0"
+url = "2.3.1"
 serde = { version = "1.0.159", features = ["derive"] }
 
+mas-axum-utils = { path = "../axum-utils" }
 mas-storage = { path = "../storage" }
 mas-storage-pg = { path = "../storage-pg" }
 mas-email = { path = "../email" }
+mas-http = { path = "../http" }
 mas-data-model = { path = "../data-model" }

crates/tasks/src/lib.rs

@@ -16,9 +16,13 @@
 #![deny(clippy::all, clippy::str_to_string, rustdoc::broken_intra_doc_links)]
 #![warn(clippy::pedantic)]
 
+use std::sync::Arc;
+
 use apalis_core::{executor::TokioExecutor, layers::extensions::Extension, monitor::Monitor};
 use apalis_sql::postgres::PostgresStorage;
+use mas_axum_utils::http_client_factory::HttpClientFactory;
 use mas_email::Mailer;
+use mas_http::{ClientInitError, ClientService, TracedClient};
 use mas_storage::{BoxClock, BoxRepository, Repository, SystemClock};
 use mas_storage_pg::{DatabaseError, PgRepository};
 use rand::SeedableRng;
@@ -28,20 +32,33 @@ use tracing::debug;
 mod database;
 mod email;
 mod layers;
+mod matrix;
+
+pub use self::matrix::HomeserverConnection;
 
 #[derive(Clone)]
 struct State {
     pool: Pool<Postgres>,
     mailer: Mailer,
     clock: SystemClock,
+    homeserver: Arc<HomeserverConnection>,
+    http_client_factory: HttpClientFactory,
 }
 
 impl State {
-    pub fn new(pool: Pool<Postgres>, clock: SystemClock, mailer: Mailer) -> Self {
+    pub fn new(
+        pool: Pool<Postgres>,
+        clock: SystemClock,
+        mailer: Mailer,
+        homeserver: HomeserverConnection,
+        http_client_factory: HttpClientFactory,
+    ) -> Self {
         Self {
             pool,
             mailer,
             clock,
+            homeserver: Arc::new(homeserver),
+            http_client_factory,
         }
     }
 
@@ -81,6 +98,21 @@ impl State {
 
         Ok(repo)
     }
+
+    pub fn matrix_connection(&self) -> &HomeserverConnection {
+        &self.homeserver
+    }
+
+    pub async fn http_client<B>(
+        &self,
+        operation: &'static str,
+    ) -> Result<ClientService<TracedClient<B>>, ClientInitError>
+    where
+        B: mas_axum_utils::axum::body::HttpBody + Send,
+        B::Data: Send,
+    {
+        self.http_client_factory.client(operation).await
+    }
 }
 
 trait JobContextExt {
@@ -96,11 +128,24 @@ impl JobContextExt for apalis_core::context::JobContext {
 }
 
 #[must_use]
-pub fn init(name: &str, pool: &Pool<Postgres>, mailer: &Mailer) -> Monitor<TokioExecutor> {
-    let state = State::new(pool.clone(), SystemClock::default(), mailer.clone());
+pub fn init(
+    name: &str,
+    pool: &Pool<Postgres>,
+    mailer: &Mailer,
+    homeserver: HomeserverConnection,
+    http_client_factory: &HttpClientFactory,
+) -> Monitor<TokioExecutor> {
+    let state = State::new(
+        pool.clone(),
+        SystemClock::default(),
+        mailer.clone(),
+        homeserver,
+        http_client_factory.clone(),
+    );
     let monitor = Monitor::new();
     let monitor = self::database::register(name, monitor, &state);
     let monitor = self::email::register(name, monitor, &state);
+    let monitor = self::matrix::register(name, monitor, &state);
     debug!(?monitor, "workers registered");
     monitor
 }

crates/tasks/src/matrix.rs

@@ -0,0 +1,184 @@
+// Copyright 2023 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use anyhow::Context;
+use apalis_core::{
+    builder::{WorkerBuilder, WorkerFactory},
+    context::JobContext,
+    executor::TokioExecutor,
+    job::Job,
+    job_fn::job_fn,
+    monitor::Monitor,
+    storage::builder::WithStorage,
+};
+use mas_axum_utils::axum::{
+    headers::{Authorization, HeaderMapExt},
+    http::{Request, StatusCode},
+};
+use mas_http::HttpServiceExt;
+use mas_storage::{
+    job::{JobWithSpanContext, ProvisionUserJob},
+    user::{UserEmailRepository, UserRepository},
+    RepositoryAccess,
+};
+use serde::{Deserialize, Serialize};
+use tower::{Service, ServiceExt};
+use tracing::{info, info_span, Instrument};
+use url::Url;
+
+use crate::{layers::TracingLayer, JobContextExt, State};
+
+pub struct HomeserverConnection {
+    homeserver: String,
+    endpoint: Url,
+    access_token: String,
+}
+
+impl HomeserverConnection {
+    pub fn new(homeserver: String, endpoint: Url, access_token: String) -> Self {
+        Self {
+            homeserver,
+            endpoint,
+            access_token,
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+struct ExternalID {
+    pub auth_provider: String,
+    pub external_id: String,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+enum ThreePIDMedium {
+    Email,
+    MSISDN,
+}
+
+#[derive(Serialize, Deserialize)]
+struct ThreePID {
+    pub medium: ThreePIDMedium,
+    pub address: String,
+}
+
+#[derive(Serialize, Deserialize)]
+struct UserRequest {
+    #[serde(rename = "displayname")]
+    pub display_name: String,
+
+    #[serde(rename = "threepids")]
+    pub three_pids: Vec<ThreePID>,
+
+    pub external_ids: Vec<ExternalID>,
+}
+
+#[tracing::instrument(
+    name = "job.provision_user"
+    fields(user.id = %job.user_id()),
+    skip_all,
+    err(Debug),
+)]
+async fn provision_user(
+    job: JobWithSpanContext<ProvisionUserJob>,
+    ctx: JobContext,
+) -> Result<(), anyhow::Error> {
+    let state = ctx.state();
+    let matrix = state.matrix_connection();
+    let mut client = state
+        .http_client("provision-matrix-user")
+        .await?
+        .request_bytes_to_body()
+        .json_request();
+    let mut repo = state.repository().await?;
+
+    let user = repo
+        .user()
+        .lookup(job.user_id())
+        .await?
+        .context("User not found")?;
+
+    let mxid = format!("@{}:{}", user.username, matrix.homeserver);
+
+    let three_pids = repo
+        .user_email()
+        .all(&user)
+        .await?
+        .into_iter()
+        .filter_map(|email| {
+            if email.confirmed_at.is_some() {
+                Some(ThreePID {
+                    medium: ThreePIDMedium::Email,
+                    address: email.email,
+                })
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    let display_name = user.username.clone();
+
+    let body = UserRequest {
+        display_name,
+        three_pids,
+        external_ids: vec![ExternalID {
+            auth_provider: "oauth-delegated".to_string(),
+            external_id: user.sub,
+        }],
+    };
+
+    repo.cancel().await?;
+
+    let mut req = Request::put(
+        matrix
+            .endpoint
+            .join("_synapse/admin/v2/users/")?
+            .join(&mxid)?
+            .as_str(),
+    );
+    req.headers_mut()
+        .context("Failed to get headers")?
+        .typed_insert(Authorization::bearer(&matrix.access_token)?);
+
+    let req = req.body(body).context("Failed to build request")?;
+
+    let span = info_span!("matrix.provision_user", %mxid);
+    let response = client.ready().await?.call(req).instrument(span).await?;
+
+    match response.status() {
+        StatusCode::CREATED => info!(%user.id, %mxid, "User created"),
+        StatusCode::OK => info!(%user.id, %mxid, "User updated"),
+        // TODO: Better error handling
+        code => anyhow::bail!("Failed to provision user. Status code: {code}"),
+    }
+
+    Ok(())
+}
+
+pub(crate) fn register(
+    suffix: &str,
+    monitor: Monitor<TokioExecutor>,
+    state: &State,
+) -> Monitor<TokioExecutor> {
+    let storage = state.store();
+    let worker_name = format!("{job}-{suffix}", job = ProvisionUserJob::NAME);
+    let worker = WorkerBuilder::new(worker_name)
+        .layer(state.inject())
+        .layer(TracingLayer::new())
+        .with_storage(storage)
+        .build(job_fn(provision_user));
+    monitor.register(worker)
+}