From 4ad436fb3393d0befade335ad4452549c3bb5b2d Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Mon, 28 Sep 2026 14:22:00 -0600 Subject: [PATCH] Add a `POST /api/admin/v1/room-invites` admin endpoint Mints a single-use, passwordless registration token per recipient, pinned to their address and optionally to a username, and schedules the invite emails. The links are only ever mailed to the recipients, so the response carries a count rather than the tokens. Minting moves out of the queue and into the caller so a username that is already taken is refused up front, rather than emailed out as a link nobody can register with. That leaves the fan-out job with nothing to do, so `SendRoomInviteEmailsJob` goes away and `SendRoomInviteEmailJob` carries the token: it takes both the address and the invite link from it, which replaces the `matrix.to` stub and keeps the link and its recipient together. `manage send-room-invites` and the endpoint share the minting through `mas_storage::room_invite`. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 2 +- crates/cli/src/commands/manage.rs | 15 +- crates/handlers/src/admin/v1/mod.rs | 5 + crates/handlers/src/admin/v1/room_invites.rs | 328 +++++++++++++++++++ crates/storage/Cargo.toml | 1 + crates/storage/src/lib.rs | 1 + crates/storage/src/queue/tasks.rs | 63 +--- crates/storage/src/room_invite.rs | 107 ++++++ crates/tasks/Cargo.toml | 1 - crates/tasks/src/lib.rs | 1 - crates/tasks/src/room_invite.rs | 109 ++---- docs/api/spec.json | 103 ++++++ docs/reference/cli/manage.md | 5 +- 13 files changed, 598 insertions(+), 143 deletions(-) create mode 100644 crates/handlers/src/admin/v1/room_invites.rs create mode 100644 crates/storage/src/room_invite.rs diff --git a/Cargo.lock b/Cargo.lock index 733b4e9b8..451452b15 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3745,6 +3745,7 @@ dependencies = [ "mas-jose", "oauth2-types", "opentelemetry", + "rand 0.8.5", "rand_core 0.6.4", "serde", "serde_json", @@ -3817,7 +3818,6 @@ dependencies = [ "tracing", "tracing-opentelemetry", "ulid", - "url", ] [[package]] diff --git a/crates/cli/src/commands/manage.rs b/crates/cli/src/commands/manage.rs index 4f604b8b4..595b90057 100644 --- a/crates/cli/src/commands/manage.rs +++ b/crates/cli/src/commands/manage.rs @@ -25,8 +25,9 @@ use mas_storage::{ oauth2::OAuth2SessionFilter, queue::{ DeactivateUserJob, ProvisionUserJob, QueueJobRepositoryExt as _, ReactivateUserJob, - SendRoomInviteEmailsJob, SyncDevicesJob, + SyncDevicesJob, }, + room_invite::{RoomInvite, schedule_room_invites}, user::{ BrowserSessionFilter, UserEmailRepository, UserFilter, UserPasswordRepository, UserRepository, @@ -592,12 +593,18 @@ impl Options { let txn = conn.begin().await?; let mut repo = PgRepository::from_conn(txn); - let job = SendRoomInviteEmailsJob::new(room_id, emails); - repo.queue_job().schedule_job(&mut rng, &clock, job).await?; + let invites = emails.into_iter().map(|email| RoomInvite { + email, + username: None, + }); + let scheduled = + schedule_room_invites(&mut repo, &mut rng, &clock, &room_id, invites).await?; repo.into_inner().commit().await?; - info!("Scheduled the room invite emails. A worker must be running to send them."); + info!( + "Scheduled {scheduled} room invite emails. A worker must be running to send them." + ); Ok(ExitCode::SUCCESS) } diff --git a/crates/handlers/src/admin/v1/mod.rs b/crates/handlers/src/admin/v1/mod.rs index b3e4a1540..38a8854d9 100644 --- a/crates/handlers/src/admin/v1/mod.rs +++ b/crates/handlers/src/admin/v1/mod.rs @@ -25,6 +25,7 @@ mod oauth2_clients; mod oauth2_sessions; mod personal_sessions; mod policy_data; +mod room_invites; mod site_config; mod upstream_oauth_links; mod upstream_oauth_providers; @@ -140,6 +141,10 @@ where "/policy-data/{id}", get_with(self::policy_data::get, self::policy_data::get_doc), ) + .api_route( + "/room-invites", + post_with(self::room_invites::handler, self::room_invites::doc), + ) .api_route( "/users", get_with(self::users::list, self::users::list_doc) diff --git a/crates/handlers/src/admin/v1/room_invites.rs b/crates/handlers/src/admin/v1/room_invites.rs new file mode 100644 index 000000000..7ff7e8c2c --- /dev/null +++ b/crates/handlers/src/admin/v1/room_invites.rs @@ -0,0 +1,328 @@ +// Copyright 2026 Element Creations Ltd. +// +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +// Please see LICENSE files in the repository root for full details. + +use std::{collections::HashSet, str::FromStr}; + +use aide::{NoApi, OperationIo, transform::TransformOperation}; +use axum::{Json, response::IntoResponse}; +use hyper::StatusCode; +use lettre::Address; +use mas_axum_utils::record_error; +use mas_data_model::BoxRng; +use mas_storage::room_invite::{RoomInvite, schedule_room_invites}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +use crate::{ + admin::{call_context::CallContext, response::ErrorResponse}, + impl_from_error_for_route, +}; + +#[derive(Debug, thiserror::Error, OperationIo)] +#[aide(output_with = "Json")] +pub enum RouteError { + #[error("Invalid email address {0:?}")] + InvalidEmail(String), + + #[error("Username {0:?} is given twice")] + DuplicateUsername(String), + + #[error("Username {0:?} is already taken")] + UsernameTaken(String), + + #[error(transparent)] + Internal(Box), +} + +impl_from_error_for_route!(mas_storage::RepositoryError); + +impl IntoResponse for RouteError { + fn into_response(self) -> axum::response::Response { + let error = ErrorResponse::from_error(&self); + let sentry_event_id = record_error!(self, Self::Internal(_)); + let status = match self { + Self::InvalidEmail(_) | Self::DuplicateUsername(_) => StatusCode::BAD_REQUEST, + Self::UsernameTaken(_) => StatusCode::CONFLICT, + Self::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR, + }; + (status, sentry_event_id, Json(error)).into_response() + } +} + +/// # A person to invite to the room +#[derive(Deserialize, JsonSchema)] +#[serde(rename = "RoomInvite")] +pub struct Invite { + /// The address to send the invite to. The recipient can only register with + /// this address. + #[schemars(example = &"alice@example.com")] + email: String, + + /// A username to impose on the recipient. If not set, they choose their + /// own. + #[schemars(example = &"guest-abcdef")] + username: Option, +} + +/// # JSON payload for the `POST /api/admin/v1/room-invites` endpoint +#[derive(Deserialize, JsonSchema)] +#[serde(rename = "AddRoomInvitesRequest")] +pub struct Request { + /// The room the recipients are invited to + #[schemars(example = &"!room:example.com")] + room_id: String, + + /// The people to invite. An address given more than once is invited once. + invites: Vec, +} + +/// # Response to the `POST /api/admin/v1/room-invites` endpoint +#[derive(Serialize, JsonSchema)] +#[serde(rename = "AddRoomInvitesResponse")] +pub struct Response { + /// How many recipients were scheduled an invite email, after dropping + /// repeated addresses + scheduled: usize, +} + +pub fn doc(operation: TransformOperation) -> TransformOperation { + operation + .id("addRoomInvites") + .summary("Invite a list of email addresses to a room") + .description( + "Mints a single-use, passwordless registration token for each recipient and emails \ + them a link to register with it. The links are only ever sent to the recipients, so \ + they are not in the response.", + ) + .tag("room-invite") + .response_with::<202, Json, _>(|t| { + t.description("The invite emails were scheduled") + .example(Response { scheduled: 2 }) + }) + .response_with::<400, RouteError, _>(|t| { + let response = + ErrorResponse::from_error(&RouteError::InvalidEmail("not-an-email".to_owned())); + t.description("The request has an invalid email or a repeated username") + .example(response) + }) + .response_with::<409, RouteError, _>(|t| { + let response = ErrorResponse::from_error(&RouteError::UsernameTaken("alice".to_owned())); + t.description("A username is already taken").example(response) + }) +} + +#[tracing::instrument( + name = "handler.admin.v1.room_invites.post", + fields(room.id = params.room_id, recipients.total = params.invites.len()), + skip_all, +)] +pub async fn handler( + CallContext { + mut repo, clock, .. + }: CallContext, + NoApi(mut rng): NoApi, + Json(params): Json, +) -> Result<(StatusCode, Json), RouteError> { + // Validate everything before minting anything, so that a rejected request + // sends no email at all + let mut usernames = HashSet::new(); + let mut invites = Vec::with_capacity(params.invites.len()); + for invite in params.invites { + if Address::from_str(&invite.email).is_err() { + return Err(RouteError::InvalidEmail(invite.email)); + } + + if let Some(username) = &invite.username { + if !usernames.insert(username.clone()) { + return Err(RouteError::DuplicateUsername(username.clone())); + } + + // A token pins the username, so the recipient of a token pinned to + // a name someone else already has can never finish registering + if repo.user().exists(username).await? { + return Err(RouteError::UsernameTaken(username.clone())); + } + } + + invites.push(RoomInvite { + email: invite.email, + username: invite.username, + }); + } + + let scheduled = + schedule_room_invites(&mut repo, &mut rng, &clock, ¶ms.room_id, invites).await?; + repo.save().await?; + + tracing::info!("Scheduled {scheduled} room invite emails"); + + Ok((StatusCode::ACCEPTED, Json(Response { scheduled }))) +} + +#[cfg(test)] +mod tests { + use chrono::Duration; + use hyper::{Request, StatusCode}; + use sqlx::PgPool; + + use crate::test_utils::{RequestBuilderExt, ResponseExt, TestState, setup}; + + /// Every recipient gets a single-use token pinned to their address, and an + /// email job of their own + #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")] + async fn test_invite(pool: PgPool) { + setup(); + let mut state = TestState::from_pool(pool.clone()).await.unwrap(); + let token = state.token_with_scope("urn:mas:admin").await; + + let request = Request::post("/api/admin/v1/room-invites") + .bearer(&token) + .json(serde_json::json!({ + "room_id": "!room:example.com", + "invites": [ + { "email": "alice@example.com", "username": "guest-alice" }, + { "email": "bob@example.com" }, + ], + })); + let response = state.request(request).await; + response.assert_status(StatusCode::ACCEPTED); + + let tokens: Vec<(String, Option, Option, bool, Option)> = + sqlx::query_as( + "SELECT token, username, email, passwordless, usage_limit + FROM user_registration_tokens ORDER BY email", + ) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(tokens.len(), 2); + assert_eq!( + ( + tokens[0].1.as_deref(), + tokens[0].2.as_deref(), + tokens[0].3, + tokens[0].4 + ), + (Some("guest-alice"), Some("alice@example.com"), true, Some(1)) + ); + // Without a username the recipient picks their own + assert_eq!( + (tokens[1].1.as_deref(), tokens[1].2.as_deref()), + (None, Some("bob@example.com")) + ); + + let jobs: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM queue_jobs WHERE queue_name = 'send-room-invite-email'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(jobs, 2); + + // Each job finds its token and renders an email out of it. The sends are + // spread out, so the second one only comes due later. + state.run_jobs_in_queue().await; + state.clock.advance(Duration::seconds(1)); + state.run_jobs_in_queue().await; + + let completed: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM queue_jobs + WHERE queue_name = 'send-room-invite-email' AND status = 'completed'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(completed, 2); + } + + /// An address given twice is invited once + #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")] + async fn test_invite_repeated_email(pool: PgPool) { + setup(); + let mut state = TestState::from_pool(pool.clone()).await.unwrap(); + let token = state.token_with_scope("urn:mas:admin").await; + + let request = Request::post("/api/admin/v1/room-invites") + .bearer(&token) + .json(serde_json::json!({ + "room_id": "!room:example.com", + "invites": [ + { "email": "alice@example.com" }, + { "email": "alice@example.com" }, + ], + })); + let response = state.request(request).await; + response.assert_status(StatusCode::ACCEPTED); + + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM user_registration_tokens") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 1); + } + + /// A token pinned to a username nobody can register with is refused, rather + /// than emailed out as a dead link + #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")] + async fn test_invite_username_taken(pool: PgPool) { + setup(); + let mut state = TestState::from_pool(pool.clone()).await.unwrap(); + let admin_token = state.token_with_scope("urn:mas:admin").await; + + let mut repo = state.repository().await.unwrap(); + repo.user() + .add(&mut state.rng(), &state.clock, "alice".to_owned()) + .await + .unwrap(); + repo.save().await.unwrap(); + + let request = Request::post("/api/admin/v1/room-invites") + .bearer(&admin_token) + .json(serde_json::json!({ + "room_id": "!room:example.com", + "invites": [{ "email": "alice@example.com", "username": "alice" }], + })); + let response = state.request(request).await; + response.assert_status(StatusCode::CONFLICT); + } + + /// A bad request mints nothing at all, so no part of the batch goes out + #[sqlx::test(migrator = "mas_storage_pg::MIGRATOR")] + async fn test_invite_invalid(pool: PgPool) { + setup(); + let mut state = TestState::from_pool(pool.clone()).await.unwrap(); + let token = state.token_with_scope("urn:mas:admin").await; + + let request = Request::post("/api/admin/v1/room-invites") + .bearer(&token) + .json(serde_json::json!({ + "room_id": "!room:example.com", + "invites": [ + { "email": "alice@example.com" }, + { "email": "not-an-email" }, + ], + })); + let response = state.request(request).await; + response.assert_status(StatusCode::BAD_REQUEST); + + let request = Request::post("/api/admin/v1/room-invites") + .bearer(&token) + .json(serde_json::json!({ + "room_id": "!room:example.com", + "invites": [ + { "email": "alice@example.com", "username": "guest-a" }, + { "email": "bob@example.com", "username": "guest-a" }, + ], + })); + let response = state.request(request).await; + response.assert_status(StatusCode::BAD_REQUEST); + + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM user_registration_tokens") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 0); + } +} diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index 07f4330c6..5fb7139ee 100644 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -21,6 +21,7 @@ async-trait.workspace = true chrono.workspace = true futures-util.workspace = true opentelemetry.workspace = true +rand.workspace = true rand_core.workspace = true serde_json.workspace = true serde.workspace = true diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index c5d5d0b26..8cc88e223 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -114,6 +114,7 @@ pub mod oauth2; pub mod personal; pub mod policy_data; pub mod queue; +pub mod room_invite; pub mod upstream_oauth2; pub mod user; diff --git a/crates/storage/src/queue/tasks.rs b/crates/storage/src/queue/tasks.rs index dafcd3870..4bce8fd19 100644 --- a/crates/storage/src/queue/tasks.rs +++ b/crates/storage/src/queue/tasks.rs @@ -652,55 +652,15 @@ impl InsertableJob for CleanupInactiveUserSessionIpsJob { const QUEUE_NAME: &'static str = "cleanup-inactive-user-session-ips"; } -/// Send room invite emails to a list of email addresses. +/// Send a room invite email for a registration token. /// -/// This job only fans out: it schedules one [`SendRoomInviteEmailJob`] per -/// unique address, so that each email is retried independently rather than -/// re-sending the whole batch. The fan-out commits as a single transaction, so -/// either every recipient got a job or none did. -#[derive(Serialize, Deserialize, Debug, Clone)] -pub struct SendRoomInviteEmailsJob { - room_id: String, - emails: Vec, -} - -impl SendRoomInviteEmailsJob { - /// Create a new job to send room invite emails - /// - /// # Parameters - /// - /// * `room_id` - The room to invite the recipients to - /// * `emails` - The addresses to invite - #[must_use] - pub fn new(room_id: impl Into, emails: Vec) -> Self { - Self { - room_id: room_id.into(), - emails, - } - } - - /// The room the recipients are invited to - #[must_use] - pub fn room_id(&self) -> &str { - &self.room_id - } - - /// The addresses to invite - #[must_use] - pub fn emails(&self) -> &[String] { - &self.emails - } -} - -impl InsertableJob for SendRoomInviteEmailsJob { - const QUEUE_NAME: &'static str = "send-room-invite-emails"; -} - -/// Send a room invite email to a single address +/// The recipient's address and the invite link both come from the token, so +/// that the link in the email can't be sent to anyone but the address the +/// token is pinned to. #[derive(Serialize, Deserialize, Debug, Clone)] pub struct SendRoomInviteEmailJob { room_id: String, - email: String, + registration_token_id: Ulid, } impl SendRoomInviteEmailJob { @@ -709,12 +669,13 @@ impl SendRoomInviteEmailJob { /// # Parameters /// /// * `room_id` - The room to invite the recipient to - /// * `email` - The address to invite + /// * `registration_token_id` - The registration token minted for the + /// recipient #[must_use] - pub fn new(room_id: impl Into, email: impl Into) -> Self { + pub fn new(room_id: impl Into, registration_token_id: Ulid) -> Self { Self { room_id: room_id.into(), - email: email.into(), + registration_token_id, } } @@ -724,10 +685,10 @@ impl SendRoomInviteEmailJob { &self.room_id } - /// The address to invite + /// The registration token minted for the recipient #[must_use] - pub fn email(&self) -> &str { - &self.email + pub fn registration_token_id(&self) -> Ulid { + self.registration_token_id } } diff --git a/crates/storage/src/room_invite.rs b/crates/storage/src/room_invite.rs new file mode 100644 index 000000000..03a36d2e2 --- /dev/null +++ b/crates/storage/src/room_invite.rs @@ -0,0 +1,107 @@ +// Copyright 2026 Element Creations Ltd. +// +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +// Please see LICENSE files in the repository root for full details. + +//! Inviting people to a room by email + +use std::collections::HashSet; + +use chrono::Duration; +use mas_data_model::Clock; +use rand::{ + RngCore, + distributions::{Alphanumeric, DistString}, +}; + +use crate::{ + RepositoryAccess, + queue::{QueueJobRepositoryExt as _, SendRoomInviteEmailJob}, +}; + +/// How much to space out the individual sends, so that a large batch doesn't +/// burst against the SMTP server. 100ms gives a ceiling of ~10 emails/second. +const SEND_INTERVAL: Duration = Duration::milliseconds(100); + +/// Length of a generated invite token, in alphanumeric characters +const TOKEN_LENGTH: usize = 12; + +/// Someone to invite to a room +pub struct RoomInvite { + /// The address to invite, which the recipient is pinned to + pub email: String, + + /// The localpart to impose on the recipient, if any + pub username: Option, +} + +/// Mint a registration token for each recipient and schedule the invite emails. +/// +/// The tokens are single-use and passwordless: the recipient is identified by +/// the address they were invited at, which the token pins, so there is nothing +/// left for them to choose that could point the invite somewhere else. +/// +/// Repeated addresses are invited once. Neither the tokens nor the jobs are +/// visible until the caller commits, so a batch either goes out whole or not +/// at all. +/// +/// Returns how many recipients were scheduled. +/// +/// # Parameters +/// +/// * `repo` - The repository to mint the tokens and schedule the jobs in +/// * `rng` - The random number generator used to generate the tokens +/// * `clock` - The clock used to generate timestamps +/// * `room_id` - The room to invite the recipients to +/// * `invites` - The recipients to invite +/// +/// # Errors +/// +/// Returns an error if the underlying repository fails. +pub async fn schedule_room_invites( + repo: &mut R, + rng: &mut (dyn RngCore + Send), + clock: &dyn Clock, + room_id: &str, + invites: impl IntoIterator + Send, +) -> Result { + let now = clock.now(); + let mut seen = HashSet::new(); + let mut delay = Duration::zero(); + + for invite in invites { + if !seen.insert(invite.email.clone()) { + continue; + } + + let token = Alphanumeric.sample_string(rng, TOKEN_LENGTH); + let token = repo + .user_registration_token() + .add( + rng, + clock, + token, + Some(1), + None, + invite.username, + Some(invite.email), + true, + ) + .await?; + + // One job per recipient, so that a bad address or a transient SMTP + // failure retries that single email rather than the whole batch. + repo.queue_job() + .schedule_job_later( + rng, + clock, + SendRoomInviteEmailJob::new(room_id, token.id), + now + delay, + ) + .await?; + + delay += SEND_INTERVAL; + } + + Ok(seen.len()) +} diff --git a/crates/tasks/Cargo.toml b/crates/tasks/Cargo.toml index 2c4e7fde9..5c2a31536 100644 --- a/crates/tasks/Cargo.toml +++ b/crates/tasks/Cargo.toml @@ -34,7 +34,6 @@ tokio.workspace = true tracing-opentelemetry.workspace = true tracing.workspace = true ulid.workspace = true -url.workspace = true mas-context.workspace = true mas-data-model.workspace = true diff --git a/crates/tasks/src/lib.rs b/crates/tasks/src/lib.rs index ffef4467a..3915f9ba3 100644 --- a/crates/tasks/src/lib.rs +++ b/crates/tasks/src/lib.rs @@ -152,7 +152,6 @@ pub async fn init( .register_handler::() .register_handler::() .register_handler::() - .register_handler::() .register_handler::() .register_handler::() .register_handler::() diff --git a/crates/tasks/src/room_invite.rs b/crates/tasks/src/room_invite.rs index 29355512f..6572995ed 100644 --- a/crates/tasks/src/room_invite.rs +++ b/crates/tasks/src/room_invite.rs @@ -5,119 +5,62 @@ //! Sending room invite emails -use std::collections::HashSet; - use async_trait::async_trait; -use chrono::Duration; use mas_email::{Address, Mailbox}; use mas_i18n::DataLocale; -use mas_storage::queue::{ - QueueJobRepositoryExt as _, SendRoomInviteEmailJob, SendRoomInviteEmailsJob, -}; +use mas_router::Invite; +use mas_storage::queue::SendRoomInviteEmailJob; use mas_templates::{EmailRoomInviteContext, TemplateContext as _}; use tracing::info; -use url::Url; use crate::{ State, new_queue::{JobContext, JobError, RunnableJob}, }; -/// How much to space out the individual sends, so that a large batch doesn't -/// burst against the SMTP server. 100ms gives a ceiling of ~10 emails/second. -const SEND_INTERVAL: Duration = Duration::milliseconds(100); - /// The locale used for recipients, who have no account and therefore no /// language preference we could look up. const DEFAULT_LOCALE: &str = "en"; -/// Build the link a recipient follows to accept the invite. -/// -/// XXX: this is a stub. A real implementation has to decide what the link -/// points at — a `matrix.to` URL, a client deep-link, or a MAS-hosted landing -/// page — and most likely needs to create the invite on the homeserver first -/// (a third-party invite keyed on the email address), so that following the -/// link actually lets the recipient in. -fn room_invite_link(room_id: &str) -> Url { - let mut url = Url::parse("https://matrix.to/").expect("static URL is valid"); - url.set_fragment(Some(&format!("/{room_id}"))); - url -} - -#[async_trait] -impl RunnableJob for SendRoomInviteEmailsJob { - #[tracing::instrument( - name = "job.send_room_invite_emails", - fields( - room.id = self.room_id(), - recipients.total = self.emails().len(), - recipients.scheduled, - ), - skip_all, - )] - async fn run(&self, state: &State, _context: JobContext) -> Result<(), JobError> { - let clock = state.clock(); - let mut rng = state.rng(); - let now = clock.now(); - let mut repo = state.repository().await.map_err(JobError::retry)?; - - // One job per recipient, so that a bad address or a transient SMTP - // failure retries that single email rather than the whole batch. - let mut seen = HashSet::new(); - let mut delay = Duration::zero(); - - for email in self.emails() { - if !seen.insert(email.as_str()) { - continue; - } - - repo.queue_job() - .schedule_job_later( - &mut rng, - clock, - SendRoomInviteEmailJob::new(self.room_id(), email), - now + delay, - ) - .await - .map_err(JobError::retry)?; - - delay += SEND_INTERVAL; - } - - tracing::Span::current().record("recipients.scheduled", seen.len()); - - // Committing the whole fan-out at once is what makes this job safe to - // retry: either every recipient got a job, or none did. - repo.save().await.map_err(JobError::retry)?; - - info!( - "Scheduled {} room invite emails, spread over {}s", - seen.len(), - delay.num_seconds() - ); - - Ok(()) - } -} - #[async_trait] impl RunnableJob for SendRoomInviteEmailJob { #[tracing::instrument( name = "job.send_room_invite_email", - fields(room.id = self.room_id()), + fields( + room.id = self.room_id(), + user_registration_token.id = %self.registration_token_id(), + ), skip_all, )] async fn run(&self, state: &State, _context: JobContext) -> Result<(), JobError> { let mailer = state.mailer(); + let url_builder = state.url_builder(); + let mut repo = state.repository().await.map_err(JobError::retry)?; + + let token = repo + .user_registration_token() + .lookup(self.registration_token_id()) + .await + .map_err(JobError::retry)? + .ok_or_else(|| JobError::fail(anyhow::anyhow!("Registration token not found")))?; + repo.cancel().await.map_err(JobError::retry)?; + + // The address is the one the token is pinned to, so the invite link can + // only ever be sent to the recipient it was minted for + let email = token.email.ok_or_else(|| { + JobError::fail(anyhow::anyhow!( + "Registration token has no email address pinned to it" + )) + })?; // A malformed address will never parse, however many times we retry - let address: Address = self.email().parse().map_err(JobError::fail)?; + let address: Address = email.parse().map_err(JobError::fail)?; let mailbox = Mailbox::new(None, address); let lang: DataLocale = DEFAULT_LOCALE.parse().map_err(JobError::fail)?; let context = EmailRoomInviteContext::new( self.room_id().to_owned(), - room_invite_link(self.room_id()), + url_builder.absolute_url_for(&Invite::new(token.token)), ) .with_language(lang); diff --git a/docs/api/spec.json b/docs/api/spec.json index 5f4842659..26b0e8d6e 100644 --- a/docs/api/spec.json +++ b/docs/api/spec.json @@ -2005,6 +2005,65 @@ } } }, + "/api/admin/v1/room-invites": { + "post": { + "tags": [ + "room-invite" + ], + "summary": "Invite a list of email addresses to a room", + "description": "Mints a single-use, passwordless registration token for each recipient and emails them a link to register with it. The links are only ever sent to the recipients, so they are not in the response.", + "operationId": "addRoomInvites", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/AddRoomInvitesRequest" + } + } + }, + "required": true + }, + "responses": { + "202": { + "description": "The invite emails were scheduled" + }, + "400": { + "description": "The request has an invalid email or a repeated username", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + }, + "example": { + "errors": [ + { + "title": "Invalid email address \"not-an-email\"" + } + ] + } + } + } + }, + "409": { + "description": "A username is already taken", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + }, + "example": { + "errors": [ + { + "title": "Username \"alice\" is already taken" + } + ] + } + } + } + } + } + } + }, "/api/admin/v1/users": { "get": { "tags": [ @@ -6494,6 +6553,50 @@ "data" ] }, + "AddRoomInvitesRequest": { + "title": "JSON payload for the `POST /api/admin/v1/room-invites` endpoint", + "type": "object", + "properties": { + "room_id": { + "description": "The room the recipients are invited to", + "type": "string", + "example": "!room:example.com" + }, + "invites": { + "description": "The people to invite. An address given more than once is invited once.", + "type": "array", + "items": { + "$ref": "#/components/schemas/RoomInvite" + } + } + }, + "required": [ + "room_id", + "invites" + ] + }, + "RoomInvite": { + "title": "A person to invite to the room", + "type": "object", + "properties": { + "email": { + "description": "The address to send the invite to. The recipient can only register with\nthis address.", + "type": "string", + "example": "alice@example.com" + }, + "username": { + "description": "A username to impose on the recipient. If not set, they choose their\nown.", + "type": [ + "string", + "null" + ], + "example": "guest-abcdef" + } + }, + "required": [ + "email" + ] + }, "UserFilter": { "type": "object", "properties": { diff --git a/docs/reference/cli/manage.md b/docs/reference/cli/manage.md index 85cc6756c..3b1fb71b1 100644 --- a/docs/reference/cli/manage.md +++ b/docs/reference/cli/manage.md @@ -154,8 +154,9 @@ $ mas-cli manage register-user --yes ## `manage send-room-invites` -Send a room invite email to a list of addresses. This schedules the job; a -worker must be running to actually send the emails. +Send a room invite email to a list of addresses. Each address gets a single-use, +passwordless registration token pinned to it, and a link to register with it. +This schedules the emails; a worker must be running to actually send them. Arguments: - ``: The room to invite the recipients to.