mirror of
https://github.com/element-hq/matrix-authentication-service.git
synced 2026-10-22 21:54:07 +00:00
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) <noreply@anthropic.com>
This commit is contained in:
1 parent
4872e25929
commit
4ad436fb33
13 files changed
+598
-143
No files matched your search
Generated
+1
-1
@@ -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]]
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<ErrorResponse>")]
|
||||
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<dyn std::error::Error + Send + Sync + 'static>),
|
||||
}
|
||||
|
||||
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<String>,
|
||||
}
|
||||
|
||||
/// # 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<Invite>,
|
||||
}
|
||||
|
||||
/// # 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<Response>, _>(|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<BoxRng>,
|
||||
Json(params): Json<Request>,
|
||||
) -> Result<(StatusCode, Json<Response>), 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<String>, Option<String>, bool, Option<i32>)> =
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
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<String>, emails: Vec<String>) -> 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<String>, email: impl Into<String>) -> Self {
|
||||
pub fn new(room_id: impl Into<String>, 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
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
/// 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<R: RepositoryAccess + ?Sized>(
|
||||
repo: &mut R,
|
||||
rng: &mut (dyn RngCore + Send),
|
||||
clock: &dyn Clock,
|
||||
room_id: &str,
|
||||
invites: impl IntoIterator<Item = RoomInvite> + Send,
|
||||
) -> Result<usize, R::Error> {
|
||||
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())
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -152,7 +152,6 @@ pub async fn init(
|
||||
.register_handler::<mas_storage::queue::ReactivateUserJob>()
|
||||
.register_handler::<mas_storage::queue::SendAccountRecoveryEmailsJob>()
|
||||
.register_handler::<mas_storage::queue::SendEmailAuthenticationCodeJob>()
|
||||
.register_handler::<mas_storage::queue::SendRoomInviteEmailsJob>()
|
||||
.register_handler::<mas_storage::queue::SendRoomInviteEmailJob>()
|
||||
.register_handler::<mas_storage::queue::SyncDevicesJob>()
|
||||
.register_handler::<mas_storage::queue::VerifyEmailJob>()
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -154,8 +154,9 @@ $ mas-cli manage register-user --yes <username>
|
||||
|
||||
## `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:
|
||||
- `<room_id>`: The room to invite the recipients to.
|
||||
|
||||
Reference in new issue
Block a user