diff --git a/crates/features/src/relay/messages.rs b/crates/features/src/relay/messages.rs new file mode 100644 index 00000000..dc379c53 --- /dev/null +++ b/crates/features/src/relay/messages.rs @@ -0,0 +1,53 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! `inbuxa:RelayMessage` ids (relay spec, "Registry objects"): a relay's +//! `Inbound` and `Outbound` messages, as this server names them. Nothing +//! is stored: the relay keeps its queues, and each id is made from the +//! relay's id and the relay's own message id, so every node names a +//! message the same way. The top 16 bits are the relay, the rest a hash +//! of the message's id at the relay, which a lookup checks against the +//! relay's queues. + +use xxhash_rust::xxh3::xxh3_64; + +const HASH_BITS: u32 = 48; +const HASH_MASK: u64 = (1 << HASH_BITS) - 1; + +/// Relays numbered above this have no message ids. Relays are numbered +/// from 1, one more for each made, so there are never that many. +pub const MAX_RELAY_ID: u32 = u16::MAX as u32; + +/// The id of a relay's message, from its id at the relay. +pub fn message_id(relay_id: u32, relay_message_id: &str) -> Option { + (relay_id <= MAX_RELAY_ID).then(|| { + (u64::from(relay_id) << HASH_BITS) | (xxh3_64(relay_message_id.as_bytes()) & HASH_MASK) + }) +} + +/// The relay a message id is at. +pub fn relay_of(id: u64) -> u32 { + (id >> HASH_BITS) as u32 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ids_name_their_relay() { + let id = message_id(7, "o0123456789abcdef01234567").unwrap(); + assert_eq!(relay_of(id), 7); + assert_eq!(message_id(7, "o0123456789abcdef01234567"), Some(id)); + assert_ne!(message_id(8, "o0123456789abcdef01234567"), Some(id)); + assert_ne!(message_id(7, "i0123456789abcdef01234567"), Some(id)); + assert_eq!( + relay_of(message_id(MAX_RELAY_ID, "x").unwrap()), + MAX_RELAY_ID + ); + assert_eq!(message_id(MAX_RELAY_ID + 1, "x"), None); + } +} diff --git a/crates/features/src/relay/mod.rs b/crates/features/src/relay/mod.rs index 6c17f230..f28138e3 100644 --- a/crates/features/src/relay/mod.rs +++ b/crates/features/src/relay/mod.rs @@ -24,6 +24,7 @@ //! Numbers are big-endian. There are few relays, so they're read whole. pub mod link; +pub mod messages; pub mod outbound; use inbuxa_relay_proto::{ diff --git a/crates/features/src/relay/outbound.rs b/crates/features/src/relay/outbound.rs index a44eca2f..736e2e95 100644 --- a/crates/features/src/relay/outbound.rs +++ b/crates/features/src/relay/outbound.rs @@ -47,10 +47,16 @@ pub enum Refused { pub type HandOver<'a> = Pin> + Send + 'a>>; +pub type Call<'a> = Pin> + Send + 'a>>; + /// One relay link on this node, able to take a message: uploads it and /// creates an `Outbound`, whose id it returns. pub trait Sender: Send + Sync { fn hand_over(&self, message: Vec, envelope: Envelope) -> HandOver<'_>; + + /// One method on the relay's API, on its account: the answer's + /// arguments, or why there are none (the relay's error included). + fn call(&self, method: &'static str, args: serde_json::Value) -> Call<'_>; } /// A relay this node is linked to. @@ -121,6 +127,17 @@ impl Senders { .insert(linked.relay_id, linked); } + /// This node's link to a relay, while it is up, whether the relay + /// sends mail or not. + pub fn linked(&self, relay_id: u32) -> Option> { + self.state + .lock() + .expect("senders") + .linked + .get(&relay_id) + .cloned() + } + /// Forgets a link, unless a newer one to the same relay replaced it. pub fn link_down(&self, linked: &Arc) { let mut state = self.state.lock().expect("senders"); @@ -245,6 +262,10 @@ mod tests { fn hand_over(&self, _: Vec, _: Envelope) -> HandOver<'_> { Box::pin(async { Err(Refused::Temporary("no".into())) }) } + + fn call(&self, _: &'static str, _: serde_json::Value) -> Call<'_> { + Box::pin(async { Err("no".into()) }) + } } fn relay(id: u32, send_through: bool, inbound_only: bool) -> Relay { diff --git a/crates/jmap-proto/src/object/inbuxa_relay_message.rs b/crates/jmap-proto/src/object/inbuxa_relay_message.rs new file mode 100644 index 00000000..c2fd9837 --- /dev/null +++ b/crates/jmap-proto/src/object/inbuxa_relay_message.rs @@ -0,0 +1,268 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! `inbuxa:RelayMessage/get`, `/query` and `/set` under `urn:inbuxa:jmap`: +//! what waits in a relay's `Inbound` and `Outbound` queues, fetched over +//! this node's link when asked (docs/spec/features/relay.md, "Registry +//! objects"). Never message content. Set only destroys: discards held +//! mail, or cancels an outbound message. The set call's `reason`, if +//! given, goes into the audit log with the change. + +use crate::{ + object::{AnyId, JmapObject, JmapObjectId}, + request::deserialize::DeserializeArguments, +}; +use jmap_tools::{Element, Key, Property}; +use std::{borrow::Cow, str::FromStr}; +use types::id::Id; + +#[derive(Debug, Clone, Default)] +pub struct RelayMessage; + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum RelayMessageProperty { + Id, + /// The `inbuxa:Relay` it is at. + RelayId, + /// `inbound` or `outbound`. + Direction, + /// Its id at the relay. + RelayMessageId, + CreatedAt, + Size, + /// RFC 8621 `Envelope`. + Envelope, + /// Inbound: `held`, `pending` and so on, as the relay says. + State, + ClaimedUntil, + /// Inbound: the SMTP session it came in on. + Session, + DoorChecks, + /// Outbound: RFC 8621 `DeliveryStatus` per recipient. + DeliveryStatus, + NextAttemptAt, + Attempts, + ExpiresAt, + /// Outbound: the message in this server's queue it was handed from. + QueueId, +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum RelayMessageValue { + Id(Id), +} + +impl Property for RelayMessageProperty { + fn try_parse(parent: Option<&Key<'_, Self>>, value: &str) -> Option { + // Keys inside the envelope, session and statuses stay plain keys + match parent { + None => RelayMessageProperty::parse(value), + Some(_) => None, + } + } + + fn to_cow(&self) -> Cow<'static, str> { + match self { + RelayMessageProperty::Id => "id", + RelayMessageProperty::RelayId => "relayId", + RelayMessageProperty::Direction => "direction", + RelayMessageProperty::RelayMessageId => "relayMessageId", + RelayMessageProperty::CreatedAt => "createdAt", + RelayMessageProperty::Size => "size", + RelayMessageProperty::Envelope => "envelope", + RelayMessageProperty::State => "state", + RelayMessageProperty::ClaimedUntil => "claimedUntil", + RelayMessageProperty::Session => "session", + RelayMessageProperty::DoorChecks => "doorChecks", + RelayMessageProperty::DeliveryStatus => "deliveryStatus", + RelayMessageProperty::NextAttemptAt => "nextAttemptAt", + RelayMessageProperty::Attempts => "attempts", + RelayMessageProperty::ExpiresAt => "expiresAt", + RelayMessageProperty::QueueId => "queueId", + } + .into() + } +} + +impl RelayMessageProperty { + fn parse(value: &str) -> Option { + hashify::tiny_map!(value.as_bytes(), + b"id" => RelayMessageProperty::Id, + b"relayId" => RelayMessageProperty::RelayId, + b"direction" => RelayMessageProperty::Direction, + b"relayMessageId" => RelayMessageProperty::RelayMessageId, + b"createdAt" => RelayMessageProperty::CreatedAt, + b"size" => RelayMessageProperty::Size, + b"envelope" => RelayMessageProperty::Envelope, + b"state" => RelayMessageProperty::State, + b"claimedUntil" => RelayMessageProperty::ClaimedUntil, + b"session" => RelayMessageProperty::Session, + b"doorChecks" => RelayMessageProperty::DoorChecks, + b"deliveryStatus" => RelayMessageProperty::DeliveryStatus, + b"nextAttemptAt" => RelayMessageProperty::NextAttemptAt, + b"attempts" => RelayMessageProperty::Attempts, + b"expiresAt" => RelayMessageProperty::ExpiresAt, + b"queueId" => RelayMessageProperty::QueueId, + ) + } +} + +impl FromStr for RelayMessageProperty { + type Err = (); + + fn from_str(s: &str) -> Result { + RelayMessageProperty::parse(s).ok_or(()) + } +} + +impl Element for RelayMessageValue { + type Property = RelayMessageProperty; + + fn try_parse

(key: &Key<'_, Self::Property>, value: &str) -> Option { + match key { + Key::Property( + RelayMessageProperty::Id + | RelayMessageProperty::RelayId + | RelayMessageProperty::QueueId, + ) => Id::from_str(value).ok().map(RelayMessageValue::Id), + _ => None, + } + } + + fn to_cow(&self) -> Cow<'static, str> { + match self { + RelayMessageValue::Id(id) => id.to_string().into(), + } + } +} + +/// One condition of an `inbuxa:RelayMessage/query` filter. Several in one +/// filter object must all hold. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RelayMessageFilter { + RelayId(Id), + /// `inbound` or `outbound`. + Direction(String), + _T(String), +} + +impl Default for RelayMessageFilter { + fn default() -> Self { + RelayMessageFilter::_T(String::new()) + } +} + +impl<'de> DeserializeArguments<'de> for RelayMessageFilter { + fn deserialize_argument(&mut self, key: &str, map: &mut A) -> Result<(), A::Error> + where + A: serde::de::MapAccess<'de>, + { + match key { + "relayId" => *self = RelayMessageFilter::RelayId(map.next_value()?), + "direction" => *self = RelayMessageFilter::Direction(map.next_value()?), + _ => { + *self = RelayMessageFilter::_T(key.to_string()); + let _ = map.next_value::()?; + } + } + Ok(()) + } +} + +/// The set call's own argument: why, for the audit log. +#[derive(Debug, Clone, Default)] +pub struct RelayMessageSetArguments { + pub reason: Option, +} + +impl<'de> DeserializeArguments<'de> for RelayMessageSetArguments { + fn deserialize_argument(&mut self, key: &str, map: &mut A) -> Result<(), A::Error> + where + A: serde::de::MapAccess<'de>, + { + if key == "reason" { + self.reason = map.next_value()?; + } else { + let _ = map.next_value::()?; + } + Ok(()) + } +} + +impl JmapObject for RelayMessage { + type Property = RelayMessageProperty; + + type Element = RelayMessageValue; + + type Id = Id; + + type Filter = RelayMessageFilter; + + type Comparator = (); + + type GetArguments = (); + + type SetArguments<'de> = RelayMessageSetArguments; + + type QueryArguments = (); + + type CopyArguments = (); + + type ParseArguments = (); + + const ID_PROPERTY: Self::Property = RelayMessageProperty::Id; +} + +impl From for RelayMessageValue { + fn from(id: Id) -> Self { + RelayMessageValue::Id(id) + } +} + +impl JmapObjectId for RelayMessageValue { + fn as_id(&self) -> Option { + match self { + RelayMessageValue::Id(id) => Some(*id), + } + } + + fn as_any_id(&self) -> Option { + match self { + RelayMessageValue::Id(id) => Some(AnyId::Id(*id)), + } + } + + fn as_id_ref(&self) -> Option<&str> { + None + } + + fn try_set_id(&mut self, new_id: AnyId) -> bool { + if let AnyId::Id(id) = new_id { + *self = RelayMessageValue::Id(id); + true + } else { + false + } + } +} + +impl JmapObjectId for RelayMessageProperty { + fn as_id(&self) -> Option { + None + } + + fn as_any_id(&self) -> Option { + None + } + + fn as_id_ref(&self) -> Option<&str> { + None + } + + fn try_set_id(&mut self, _: AnyId) -> bool { + false + } +} diff --git a/crates/jmap-proto/src/object/mod.rs b/crates/jmap-proto/src/object/mod.rs index ad828cd8..b158c737 100644 --- a/crates/jmap-proto/src/object/mod.rs +++ b/crates/jmap-proto/src/object/mod.rs @@ -43,6 +43,7 @@ pub mod inbuxa_log_settings; // inbuxa: personal-data catalog, D1 pub mod inbuxa_mail_rule; // inbuxa: DLP and mail flow rules pub mod inbuxa_protocol_policy; // inbuxa: legacy protocols off pub mod inbuxa_relay; // inbuxa: inbuxa Relay +pub mod inbuxa_relay_message; // inbuxa: a relay's queues pub mod inbuxa_report_export; // inbuxa: scheduled reports, RP-19 pub mod inbuxa_scheduled_report; // inbuxa: scheduled reports pub mod inbuxa_scheduled_report_settings; // inbuxa: scheduled reports diff --git a/crates/jmap-proto/src/references/eval.rs b/crates/jmap-proto/src/references/eval.rs index 193e234b..e1af0805 100644 --- a/crates/jmap-proto/src/references/eval.rs +++ b/crates/jmap-proto/src/references/eval.rs @@ -91,6 +91,9 @@ impl Response<'_> { GetResponseMethod::Relay(response) => { response.eval_jptr(path, &mut results) } + GetResponseMethod::RelayMessage(response) => { + response.eval_jptr(path, &mut results) + } GetResponseMethod::SecurityAcceptance(response) => { response.eval_jptr(path, &mut results) } diff --git a/crates/jmap-proto/src/references/resolve.rs b/crates/jmap-proto/src/references/resolve.rs index 40200f36..5c7bdafb 100644 --- a/crates/jmap-proto/src/references/resolve.rs +++ b/crates/jmap-proto/src/references/resolve.rs @@ -56,6 +56,7 @@ impl Response<'_> { GetRequestMethod::LegalHold(request) => request.resolve_references(self)?, GetRequestMethod::MailRule(request) => request.resolve_references(self)?, GetRequestMethod::Relay(request) => request.resolve_references(self)?, + GetRequestMethod::RelayMessage(request) => request.resolve_references(self)?, GetRequestMethod::SecurityAcceptance(request) => { request.resolve_references(self)? } @@ -151,6 +152,9 @@ impl Response<'_> { request.resolve_references(self, 1, false)? } SetRequestMethod::Relay(request) => request.resolve_references(self, 1, false)?, + SetRequestMethod::RelayMessage(request) => { + request.resolve_references(self, 1, false)? + } SetRequestMethod::SecurityAcceptance(request) => { request.resolve_references(self, 1, false)? } diff --git a/crates/jmap-proto/src/request/method.rs b/crates/jmap-proto/src/request/method.rs index ddc2dcc5..adca09cc 100644 --- a/crates/jmap-proto/src/request/method.rs +++ b/crates/jmap-proto/src/request/method.rs @@ -69,6 +69,7 @@ pub enum MethodObject { // inbuxa: DLP and mail flow rules MailRule, Relay, + RelayMessage, // inbuxa: accepted security to-do items SecurityAcceptance, // inbuxa: the deliverability check @@ -126,6 +127,7 @@ impl MethodObject { | MethodObject::HoldExport | MethodObject::MailRule | MethodObject::Relay + | MethodObject::RelayMessage | MethodObject::SecurityAcceptance | MethodObject::HeldMessage | MethodObject::DeliverabilityReport @@ -340,6 +342,9 @@ impl MethodName { (MethodFunction::Set, MethodObject::MailRule) => "inbuxa:MailRule/set", (MethodFunction::Get, MethodObject::Relay) => "inbuxa:Relay/get", (MethodFunction::Set, MethodObject::Relay) => "inbuxa:Relay/set", + (MethodFunction::Get, MethodObject::RelayMessage) => "inbuxa:RelayMessage/get", + (MethodFunction::Query, MethodObject::RelayMessage) => "inbuxa:RelayMessage/query", + (MethodFunction::Set, MethodObject::RelayMessage) => "inbuxa:RelayMessage/set", (MethodFunction::Get, MethodObject::SecurityAcceptance) => { "inbuxa:SecurityAcceptance/get" } @@ -540,6 +545,9 @@ impl MethodName { "inbuxa:MailRule/set" => (MethodObject::MailRule, MethodFunction::Set), "inbuxa:Relay/get" => (MethodObject::Relay, MethodFunction::Get), "inbuxa:Relay/set" => (MethodObject::Relay, MethodFunction::Set), + "inbuxa:RelayMessage/get" => (MethodObject::RelayMessage, MethodFunction::Get), + "inbuxa:RelayMessage/query" => (MethodObject::RelayMessage, MethodFunction::Query), + "inbuxa:RelayMessage/set" => (MethodObject::RelayMessage, MethodFunction::Set), "inbuxa:SecurityAcceptance/get" => (MethodObject::SecurityAcceptance, MethodFunction::Get), "inbuxa:SecurityAcceptance/set" => (MethodObject::SecurityAcceptance, MethodFunction::Set), "inbuxa:DeliverabilityReport/get" => (MethodObject::DeliverabilityReport, MethodFunction::Get), @@ -637,6 +645,7 @@ impl Display for MethodObject { MethodObject::LegalHold => "inbuxa:LegalHold", MethodObject::MailRule => "inbuxa:MailRule", MethodObject::Relay => "inbuxa:Relay", + MethodObject::RelayMessage => "inbuxa:RelayMessage", MethodObject::SecurityAcceptance => "inbuxa:SecurityAcceptance", MethodObject::DeliverabilityReport => "inbuxa:DeliverabilityReport", MethodObject::DeliverabilitySettings => "inbuxa:DeliverabilitySettings", diff --git a/crates/jmap-proto/src/request/mod.rs b/crates/jmap-proto/src/request/mod.rs index 807b13ec..01bde4cf 100644 --- a/crates/jmap-proto/src/request/mod.rs +++ b/crates/jmap-proto/src/request/mod.rs @@ -128,6 +128,7 @@ pub enum GetRequestMethod { LegalHold(Box>), MailRule(Box>), Relay(Box>), + RelayMessage(Box>), SecurityAcceptance( Box>, ), @@ -185,6 +186,7 @@ pub enum SetRequestMethod<'x> { LegalHold(Box>), MailRule(Box>), Relay(Box>), + RelayMessage(Box>), SecurityAcceptance( Box>, ), @@ -246,6 +248,7 @@ pub enum QueryRequestMethod { Registry(Box>), AuditEvent(Box>), JournalEntry(Box>), + RelayMessage(Box>), } #[derive(Debug)] diff --git a/crates/jmap-proto/src/request/parser.rs b/crates/jmap-proto/src/request/parser.rs index a598c0bf..3d4144fb 100644 --- a/crates/jmap-proto/src/request/parser.rs +++ b/crates/jmap-proto/src/request/parser.rs @@ -704,6 +704,27 @@ impl<'de> Visitor<'de> for CallVisitor { return Err(de::Error::invalid_length(1, &self)); } }, + (MethodFunction::Get, MethodObject::RelayMessage) => match seq.next_element() { + Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::RelayMessage(value)), + Err(err) => RequestMethod::invalid(err), + Ok(None) => { + return Err(de::Error::invalid_length(1, &self)); + } + }, + (MethodFunction::Query, MethodObject::RelayMessage) => match seq.next_element() { + Ok(Some(value)) => RequestMethod::Query(QueryRequestMethod::RelayMessage(value)), + Err(err) => RequestMethod::invalid(err), + Ok(None) => { + return Err(de::Error::invalid_length(1, &self)); + } + }, + (MethodFunction::Set, MethodObject::RelayMessage) => match seq.next_element() { + Ok(Some(value)) => RequestMethod::Set(SetRequestMethod::RelayMessage(value)), + Err(err) => RequestMethod::invalid(err), + Ok(None) => { + return Err(de::Error::invalid_length(1, &self)); + } + }, // inbuxa: accepted security to-do items (MethodFunction::Get, MethodObject::SecurityAcceptance) => match seq.next_element() { Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::SecurityAcceptance(value)), diff --git a/crates/jmap-proto/src/response/mod.rs b/crates/jmap-proto/src/response/mod.rs index ccbab85e..68f3d69f 100644 --- a/crates/jmap-proto/src/response/mod.rs +++ b/crates/jmap-proto/src/response/mod.rs @@ -113,6 +113,7 @@ pub enum GetResponseMethod { LegalHold(GetResponse), MailRule(GetResponse), Relay(GetResponse), + RelayMessage(GetResponse), SecurityAcceptance(GetResponse), DeliverabilityReport( GetResponse, @@ -168,6 +169,7 @@ pub enum SetResponseMethod { LegalHold(Box>), MailRule(Box>), Relay(Box>), + RelayMessage(Box>), SecurityAcceptance( Box>, ), @@ -1047,6 +1049,22 @@ impl<'x> From> for ResponseMetho } } +impl<'x> From> + for ResponseMethod<'x> +{ + fn from(value: GetResponse) -> Self { + ResponseMethod::Get(GetResponseMethod::RelayMessage(value)) + } +} + +impl<'x> From> + for ResponseMethod<'x> +{ + fn from(value: SetResponse) -> Self { + ResponseMethod::Set(SetResponseMethod::RelayMessage(Box::new(value))) + } +} + // inbuxa: journaling impl<'x> From> for ResponseMethod<'x> diff --git a/crates/jmap/src/api/auth.rs b/crates/jmap/src/api/auth.rs index b1040249..985ace27 100644 --- a/crates/jmap/src/api/auth.rs +++ b/crates/jmap/src/api/auth.rs @@ -118,6 +118,7 @@ impl JmapAuthorization for AccessToken { } // inbuxa: inbuxa Relay GetRequestMethod::Relay(_) => Permission::SysRelayGet, + GetRequestMethod::RelayMessage(_) => Permission::SysRelayQueueGet, // inbuxa: journaling (JR-18) GetRequestMethod::Journal(_) => Permission::SysJournalGet, GetRequestMethod::JournalEntry(_) => Permission::SysJournalSearch, @@ -320,6 +321,14 @@ impl JmapAuthorization for AccessToken { Permission::SysRelayUpdate, Permission::SysRelayUpdate, ), + // inbuxa: a relay's queues; set only destroys + SetRequestMethod::RelayMessage(s) => validate_set( + s, + self, + Permission::SysRelayQueueUpdate, + Permission::SysRelayQueueUpdate, + Permission::SysRelayQueueUpdate, + ), // inbuxa: journaling (JR-18) SetRequestMethod::Journal(s) => validate_set( s, @@ -577,6 +586,7 @@ impl JmapAuthorization for AccessToken { | MethodObject::HoldExport | MethodObject::MailRule | MethodObject::Relay + | MethodObject::RelayMessage | MethodObject::SecurityAcceptance | MethodObject::DeliverabilityReport | MethodObject::DeliverabilitySettings @@ -650,6 +660,8 @@ impl JmapAuthorization for AccessToken { QueryRequestMethod::AuditEvent(_) => Permission::SysAuditGet, // inbuxa: journaling (JR-15) QueryRequestMethod::JournalEntry(_) => Permission::SysJournalSearch, + // inbuxa: a relay's queues + QueryRequestMethod::RelayMessage(_) => Permission::SysRelayQueueGet, QueryRequestMethod::Registry(_) => { let MethodObject::Registry(object_type) = object else { unreachable!() diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index dd528df2..e770c653 100644 --- a/crates/jmap/src/api/request.rs +++ b/crates/jmap/src/api/request.rs @@ -293,6 +293,9 @@ impl RequestHandler for Server { SetResponseMethod::Relay(set_response) => { set_response.update_created_ids(&mut response); } + SetResponseMethod::RelayMessage(set_response) => { + set_response.update_created_ids(&mut response); + } SetResponseMethod::SecurityAcceptance(set_response) => { set_response.update_created_ids(&mut response); } @@ -566,6 +569,13 @@ impl RequestHandler for Server { .await? .into() } + // inbuxa: a relay's queues, fetched over the link + GetRequestMethod::RelayMessage(mut req) => { + resolve_account_id(&mut req.account_id, method_name.obj, access_token)?; + crate::inbuxa::relay_message::get(self, access_token, *req) + .await? + .into() + } // inbuxa: accepted security to-do items GetRequestMethod::SecurityAcceptance(mut req) => { resolve_account_id(&mut req.account_id, method_name.obj, access_token)?; @@ -835,6 +845,13 @@ impl RequestHandler for Server { .await? .into() } + // inbuxa: a relay's queues + QueryRequestMethod::RelayMessage(mut req) => { + resolve_account_id(&mut req.account_id, method_name.obj, access_token)?; + crate::inbuxa::relay_message::query(self, access_token, *req) + .await? + .into() + } QueryRequestMethod::Registry(mut req) => { resolve_account_id(&mut req.account_id, method_name.obj, access_token)?; assert_registry_account(self, method_name.obj, access_token, req.account_id) @@ -1138,6 +1155,22 @@ impl RequestHandler for Server { .await? .into() } + SetRequestMethod::RelayMessage(mut req) => { + resolve_account_id(&mut req.account_id, method_name.obj, access_token)?; + let reason = req.arguments.reason.clone(); + crate::inbuxa::audit::recorded( + self, + access_token, + session, + &method_name.obj.to_string(), + None, + reason, + *req, + |req| Box::pin(crate::inbuxa::relay_message::set(self, access_token, req)), + ) + .await? + .into() + } // inbuxa: SS-26, every acceptance made or removed is in the // audit log SetRequestMethod::SecurityAcceptance(mut req) => { diff --git a/crates/jmap/src/changes/get.rs b/crates/jmap/src/changes/get.rs index bf105c9c..263f9ddd 100644 --- a/crates/jmap/src/changes/get.rs +++ b/crates/jmap/src/changes/get.rs @@ -432,6 +432,7 @@ impl IntermediateChangesResponse { | MethodObject::HoldExport | MethodObject::MailRule | MethodObject::Relay + | MethodObject::RelayMessage | MethodObject::SecurityAcceptance | MethodObject::DeliverabilityReport | MethodObject::DeliverabilitySettings diff --git a/crates/jmap/src/inbuxa/mod.rs b/crates/jmap/src/inbuxa/mod.rs index cfb20291..2fb0e7eb 100644 --- a/crates/jmap/src/inbuxa/mod.rs +++ b/crates/jmap/src/inbuxa/mod.rs @@ -33,6 +33,7 @@ pub mod masked_secrets; // inbuxa: "****" is never saved as a secret pub mod oidc_directory; // inbuxa: http-directory-services-13 pub mod protocol_policy; pub mod relay; // inbuxa: inbuxa Relay +pub mod relay_message; // inbuxa: a relay's queues pub mod scheduled_reports; // inbuxa: scheduled reports and the weekly digest pub mod security_acceptance; pub mod sharing_policy; diff --git a/crates/jmap/src/inbuxa/relay.rs b/crates/jmap/src/inbuxa/relay.rs index 64f303b0..eede336a 100644 --- a/crates/jmap/src/inbuxa/relay.rs +++ b/crates/jmap/src/inbuxa/relay.rs @@ -72,7 +72,7 @@ const SERVER_SET: &[P] = &[ P::UpdatedAt, ]; -fn server_level(access_token: &AccessToken) -> trc::Result<()> { +pub(super) fn server_level(access_token: &AccessToken) -> trc::Result<()> { if access_token.tenant_id().is_some() { Err(trc::JmapEvent::Forbidden .into_err() @@ -82,7 +82,12 @@ fn server_level(access_token: &AccessToken) -> trc::Result<()> { } } -fn json_to_value(json: serde_json::Value) -> RValue { +/// A JSON value as a JMAP one, its object keys kept as plain keys. +pub(super) fn json_to_value(json: serde_json::Value) -> Value<'static, P, E> +where + P: Property, + E: jmap_tools::Element, +{ match json { serde_json::Value::Null => Value::Null, serde_json::Value::Bool(b) => Value::Bool(b), diff --git a/crates/jmap/src/inbuxa/relay_message.rs b/crates/jmap/src/inbuxa/relay_message.rs new file mode 100644 index 00000000..07c2d36d --- /dev/null +++ b/crates/jmap/src/inbuxa/relay_message.rs @@ -0,0 +1,476 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! `inbuxa:RelayMessage` (docs/spec/features/relay.md, "Registry +//! objects"): what waits in each relay's `Inbound` and `Outbound` queues, +//! fetched over this node's link when asked, and never the message itself. +//! Destroying one discards held mail at the relay, or cancels an outbound +//! message: its recipients fail here, with no DSN, as a delivery cancelled +//! in this server's queue does, and the relay drops its copy (at its next +//! status, if it is sending it just then). The request layer records every +//! destroy in the audit log. + +use super::relay::{json_to_value, server_level}; +use common::{Server, auth::AccessToken}; +use inbuxa_features::relay::{ + self as relays, Relay as StoredRelay, + messages::{message_id, relay_of}, + outbound::{self, Handed, Linked}, +}; +use inbuxa_relay_proto::jmap::method; +use jmap_proto::{ + error::set::SetError, + method::{ + get::{GetRequest, GetResponse}, + query::{Filter as QueryFilter, QueryRequest, QueryResponse}, + set::{SetRequest, SetResponse}, + }, + object::inbuxa_relay_message::{ + RelayMessage, RelayMessageFilter, RelayMessageProperty as P, RelayMessageValue, + }, + request::IntoValid, + types::state::State, +}; +use jmap_tools::{Key, Map, Value}; +use serde_json::json; +use smtp::outbound::relay::{Applied, RelayedStatus, apply_relayed}; +use std::{collections::HashMap, sync::Arc}; +use types::id::Id; + +type RValue = Value<'static, P, RelayMessageValue>; + +const ALL: &[P] = &[ + P::Id, + P::RelayId, + P::Direction, + P::RelayMessageId, + P::CreatedAt, + P::Size, + P::Envelope, + P::State, + P::ClaimedUntil, + P::Session, + P::DoorChecks, + P::DeliveryStatus, + P::NextAttemptAt, + P::Attempts, + P::ExpiresAt, + P::QueueId, +]; + +/// The most ids one query returns. +const MAX_QUERY_LIMIT: usize = 1_000; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Direction { + Inbound, + Outbound, +} + +impl Direction { + fn as_str(self) -> &'static str { + match self { + Direction::Inbound => "inbound", + Direction::Outbound => "outbound", + } + } +} + +/// One message at a relay, as the relay describes it. +struct Found { + id: u64, + relay_id: u32, + direction: Direction, + relay_message_id: String, + created_at: String, + object: serde_json::Value, +} + +/// One relay's queues, read over this node's link. +struct Queues { + linked: Arc, + messages: Vec, + /// Its outbound messages' places in this server's queue. + handed: HashMap, +} + +fn unavailable(relay: &StoredRelay, why: impl std::fmt::Display) -> trc::Error { + trc::EventType::Relay(trc::RelayEvent::LinkDown) + .into_err() + .details(format!("Relay {}: {why}", relay.name)) + .ctx(trc::Key::Id, Id::from(relay.id).to_string()) +} + +async fn queues(server: &Server, relay: &StoredRelay) -> trc::Result { + let linked = server + .inner + .data + .relay_senders + .linked(relay.id) + .ok_or_else(|| unavailable(relay, "this node has no link to it now"))?; + let mut messages = Vec::new(); + for (direction, get) in [ + (Direction::Inbound, method::INBOUND_GET), + (Direction::Outbound, method::OUTBOUND_GET), + ] { + let answer = linked + .sender + .call(get, json!({"ids": null})) + .await + .map_err(|why| unavailable(relay, why))?; + for object in answer["list"].as_array().into_iter().flatten() { + let Some(relay_message_id) = object["id"].as_str() else { + continue; + }; + let Some(id) = message_id(relay.id, relay_message_id) else { + continue; + }; + let created_at = match direction { + Direction::Inbound => &object["receivedAt"], + Direction::Outbound => &object["createdAt"], + }; + messages.push(Found { + id, + relay_id: relay.id, + direction, + relay_message_id: relay_message_id.to_string(), + created_at: created_at.as_str().unwrap_or_default().to_string(), + object: object.clone(), + }); + } + } + let handed = outbound::handed(server.store(), relay.id) + .await? + .into_iter() + .map(|h| (h.id.clone(), h)) + .collect(); + Ok(Queues { + linked, + messages, + handed, + }) +} + +fn to_value(found: &Found, handed: &HashMap, properties: &[P]) -> RValue { + let field = |name: &str| -> RValue { + match found.object.get(name) { + Some(value) if !value.is_null() => json_to_value(value.clone()), + _ => Value::Null, + } + }; + let inbound = found.direction == Direction::Inbound; + let mut out = Map::with_capacity(properties.len()); + for property in properties { + let value = match property { + P::Id => Value::Element(RelayMessageValue::Id(Id::new(found.id))), + P::RelayId => Value::Element(RelayMessageValue::Id(Id::from(found.relay_id))), + P::Direction => Value::Str(found.direction.as_str().into()), + P::RelayMessageId => Value::Str(found.relay_message_id.clone().into()), + P::CreatedAt => Value::Str(found.created_at.clone().into()), + P::Size => field("size"), + P::Envelope => field("envelope"), + P::State if inbound => field("state"), + P::ClaimedUntil if inbound => field("claimedUntil"), + P::Session if inbound => field("session"), + P::DoorChecks if inbound => field("doorChecks"), + P::DeliveryStatus if !inbound => field("deliveryStatus"), + P::NextAttemptAt if !inbound => field("nextAttemptAt"), + P::Attempts if !inbound => field("attempts"), + P::ExpiresAt if !inbound => field("expiresAt"), + P::QueueId if !inbound => handed + .get(&found.relay_message_id) + .map_or(Value::Null, |h| { + Value::Element(RelayMessageValue::Id(Id::new(h.queue_id))) + }), + _ => Value::Null, + }; + out.insert_unchecked(Key::Property(property.clone()), value); + } + Value::Object(out) +} + +/// The relays with these messages, or every enabled relay. +fn wanted_relays(all: &[StoredRelay], ids: Option<&[Id]>) -> Vec { + all.iter() + .filter(|relay| match ids { + Some(ids) => ids.iter().any(|id| relay_of(id.id()) == relay.id), + None => relay.enabled, + }) + .cloned() + .collect() +} + +pub async fn get( + server: &Server, + access_token: &AccessToken, + mut request: GetRequest, +) -> trc::Result> { + server_level(access_token)?; + let properties = request.unwrap_properties(ALL); + let (ids, not_found) = request.unwrap_ids(server.core.jmap.get_max_objects)?; + let mut response = GetResponse { + account_id: request.account_id.into(), + state: None, + list: Vec::new(), + not_found, + }; + let all = relays::all(server.store()).await?; + let mut fetched = Vec::new(); + for relay in wanted_relays(&all, ids.as_deref()) { + match queues(server, &relay).await { + Ok(queues) => fetched.push(queues), + // Asked for all: a relay this node can't reach now is left out + Err(_) if ids.is_none() => {} + Err(err) => return Err(err), + } + } + match ids { + None => { + for queues in &fetched { + for found in &queues.messages { + response + .list + .push(to_value(found, &queues.handed, &properties)); + } + } + } + Some(ids) => { + for id in ids { + match fetched.iter().find_map(|queues| { + queues + .messages + .iter() + .find(|found| found.id == id.id()) + .map(|found| (found, &queues.handed)) + }) { + Some((found, handed)) => { + response.list.push(to_value(found, handed, &properties)) + } + None => response.push_not_found(id), + } + } + } + } + Ok(response) +} + +/// `inbuxa:RelayMessage/query`: newest first, of every enabled relay this +/// node is linked to, or of the one asked for, which must be. +pub async fn query( + server: &Server, + access_token: &AccessToken, + request: QueryRequest, +) -> trc::Result { + server_level(access_token)?; + let unsupported = |why: String| trc::JmapEvent::UnsupportedFilter.into_err().details(why); + let mut relay_id = None; + let mut direction = None; + for condition in request.filter { + match condition { + QueryFilter::Property(RelayMessageFilter::RelayId(id)) => relay_id = Some(id), + QueryFilter::Property(RelayMessageFilter::Direction(d)) => { + direction = Some(match d.as_str() { + "inbound" => Direction::Inbound, + "outbound" => Direction::Outbound, + _ => { + return Err(unsupported("direction is inbound or outbound.".to_string())); + } + }) + } + QueryFilter::Property(RelayMessageFilter::_T(other)) => { + return Err(unsupported(format!("Unknown filter property {other}."))); + } + QueryFilter::And | QueryFilter::Close => {} + QueryFilter::Or | QueryFilter::Not => { + return Err(unsupported( + "Conditions must all hold; OR and NOT aren't supported.".to_string(), + )); + } + } + } + let position = request.position.unwrap_or(0); + if position < 0 || request.anchor.is_some() { + return Err(unsupported( + "Relay messages page by a position from the start.".to_string(), + )); + } + let limit = request + .limit + .unwrap_or(MAX_QUERY_LIMIT) + .min(MAX_QUERY_LIMIT); + + let all = relays::all(server.store()).await?; + let mut found = Vec::new(); + match relay_id { + Some(id) => { + if let Some(relay) = all.iter().find(|r| u64::from(r.id) == id.id()) { + found.extend(queues(server, relay).await?.messages); + } + } + None => { + for relay in all.iter().filter(|r| r.enabled) { + if let Ok(queues) = queues(server, relay).await { + found.extend(queues.messages); + } + } + } + } + found.retain(|f| direction.is_none_or(|d| f.direction == d)); + found.sort_by(|a, b| b.created_at.cmp(&a.created_at).then(b.id.cmp(&a.id))); + let total = found.len(); + Ok(QueryResponse { + account_id: request.account_id, + query_state: State::Initial, + can_calculate_changes: false, + position, + ids: found + .into_iter() + .skip(position as usize) + .take(limit) + .map(|f| Id::new(f.id)) + .collect(), + total: request.calculate_total.unwrap_or(false).then_some(total), + limit: Some(limit), + }) +} + +/// What a relay's destroy did with one id. +enum Destroyed { + Yes, + /// It was gone already. + NotFound, + Refused(SetError

), +} + +fn destroyed(answer: &serde_json::Value, id: &str) -> Destroyed { + let Some(error) = answer["notDestroyed"].get(id) else { + return Destroyed::Yes; + }; + match error["type"].as_str() { + Some("notFound") => Destroyed::NotFound, + _ => Destroyed::Refused( + SetError::forbidden().with_description( + error["description"] + .as_str() + .unwrap_or("The relay refused.") + .to_string(), + ), + ), + } +} + +/// `inbuxa:RelayMessage/set`: destroy only. +pub async fn set( + server: &Server, + access_token: &AccessToken, + mut request: SetRequest<'_, RelayMessage>, +) -> trc::Result> { + server_level(access_token)?; + let mut response = SetResponse::from_request(&request, server.core.jmap.set_max_objects)?; + for (client_id, _) in request.unwrap_create() { + response.not_created.append( + client_id, + SetError::forbidden().with_description("Only a relay makes its messages."), + ); + } + for (id, _) in request.unwrap_update().into_valid() { + response.not_updated.append( + id, + SetError::forbidden() + .with_description("A relay's message can't be changed, only destroyed."), + ); + } + + let data = server.store(); + let all = relays::all(data).await?; + let mut fetched: HashMap = HashMap::new(); + for id in request.unwrap_destroy().into_valid() { + let Some(relay) = all.iter().find(|r| r.id == relay_of(id.id())) else { + response.not_destroyed.append(id, SetError::not_found()); + continue; + }; + if let std::collections::hash_map::Entry::Vacant(entry) = fetched.entry(relay.id) { + entry.insert(queues(server, relay).await?); + } + let queues = &fetched[&relay.id]; + let Some(found) = queues.messages.iter().find(|f| f.id == id.id()) else { + response.not_destroyed.append(id, SetError::not_found()); + continue; + }; + let rid = found.relay_message_id.as_str(); + match found.direction { + Direction::Inbound => { + // Being taken just now: it is this server's mail already + if found.object["state"].as_str() == Some("claimed") { + response.not_destroyed.append( + id, + SetError::forbidden() + .with_description("This server is taking it now; try again."), + ); + continue; + } + let answer = queues + .linked + .sender + .call(method::INBOUND_SET, json!({"destroy": [rid]})) + .await + .map_err(|why| unavailable(relay, why))?; + match destroyed(&answer, rid) { + Destroyed::Yes => response.destroyed.push(id), + Destroyed::NotFound => response.not_destroyed.append(id, SetError::not_found()), + Destroyed::Refused(error) => response.not_destroyed.append(id, error), + } + } + Direction::Outbound => { + // Cancelled here first, so the queue never hands it over + // again once the relay drops it + let handed = queues.handed.get(rid); + if let Some(handed) = handed { + let cancelled: Vec<(String, RelayedStatus)> = handed + .recipients + .iter() + .map(|r| (r.clone(), RelayedStatus::Cancelled)) + .collect(); + if apply_relayed( + server, + handed.queue_id, + &handed.queue_name, + &relay.name, + &cancelled, + ) + .await + == Applied::Locked + { + response.not_destroyed.append( + id, + SetError::forbidden().with_description( + "This server's queue is working on the message; try again.", + ), + ); + continue; + } + } + let answer = queues + .linked + .sender + .call(method::OUTBOUND_SET, json!({"destroy": [rid]})) + .await + .map_err(|why| unavailable(relay, why))?; + match destroyed(&answer, rid) { + // Cancelled here; the relay drops it after this attempt + Destroyed::Refused(error) => response.not_destroyed.append(id, error), + Destroyed::Yes | Destroyed::NotFound => { + if handed.is_some() { + outbound::forget_handed(data, relay.id, &[rid.to_string()]).await?; + } + response.destroyed.push(id); + } + } + } + } + } + Ok(response) +} diff --git a/crates/services/src/inbuxa_relay_link.rs b/crates/services/src/inbuxa_relay_link.rs index c62f7f5a..a67154a4 100644 --- a/crates/services/src/inbuxa_relay_link.rs +++ b/crates/services/src/inbuxa_relay_link.rs @@ -46,7 +46,7 @@ use inbuxa_features::{ RecipientPlan, forget_ingested, is_inbound_id, issue_node_certificate, pushed_settings, record_ingested, record_join, set_status, was_ingested, }, - outbound::{self, HandOver, Handed, Linked, Refused, Sender}, + outbound::{self, Call, HandOver, Handed, Linked, Refused, Sender}, }, }; use inbuxa_relay_proto::{ @@ -1214,6 +1214,32 @@ impl Sender for LinkSender { Err(refusal(&args["notCreated"]["m"])) }) } + + fn call(&self, method: &'static str, mut args: Value) -> Call<'_> { + Box::pin(async move { + if args.get("accountId").is_none() { + args["accountId"] = ACCOUNT_ID.into(); + } + let answer = self + .dialer + .api(vec![Invocation::new(method, args, "c")]) + .await + .map_err(|err| err.message)? + .into_iter() + .next() + .ok_or_else(|| format!("The relay didn't answer {method}."))?; + if answer.name() == "error" { + return Err(format!( + "The relay refused {method}: {}", + answer.args()["description"] + .as_str() + .or(answer.args()["type"].as_str()) + .unwrap_or("no reason") + )); + } + Ok(Value::Object(answer.1)) + }) + } } /// A relay's `notCreated` for a message, as this server's refusal. diff --git a/crates/smtp/src/outbound/relay.rs b/crates/smtp/src/outbound/relay.rs index 054d1e6c..8ba26f18 100644 --- a/crates/smtp/src/outbound/relay.rs +++ b/crates/smtp/src/outbound/relay.rs @@ -17,8 +17,8 @@ use super::DeliveryResult; use crate::queue::{ - Error, ErrorDetails, HostResponse, MessageWrapper, RCPT_RELAYED, Status, UnexpectedResponse, - spool::SmtpSpool, + Error, ErrorDetails, HostResponse, MessageWrapper, RCPT_DSN_SENT, RCPT_RELAYED, Status, + UnexpectedResponse, spool::SmtpSpool, }; use crate::{queue::RecipientDomain, reporting::send::MtaReportSend}; use common::{Server, config::smtp::queue::QueueName, ipc::QueueEvent}; @@ -274,6 +274,10 @@ pub enum RelayedStatus { }, /// The relay no longer has the message: hand it over again. Lost, + /// An administrator cancelled it (`inbuxa:RelayMessage/set` destroy): + /// failed, like a delivery cancelled in this server's queue, and with + /// no DSN. + Cancelled, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -281,7 +285,9 @@ pub enum Applied { Done, /// The queue is working on the message: try again later. Locked, - /// The message is no longer queued here. + /// The message is no longer queued here, or none of these recipients + /// waits on the relay any more (cancelled, or expired here): the relay + /// can stop. Gone, } @@ -308,6 +314,7 @@ pub async fn apply_relayed( let now = now(); let mut changed = false; let mut wake = false; + let mut waiting = false; for rcpt in message.message.recipients.iter_mut() { // Expired here meanwhile: this server's word stands if rcpt.flags & RCPT_RELAYED == 0 @@ -324,6 +331,7 @@ pub async fn apply_relayed( else { continue; }; + waiting = true; let entity = |mx: &Option| -> Box { mx.as_deref().unwrap_or(relay).into() }; let new = match status { RelayedStatus::Delivered { reply, mx } => Status::Completed(HostResponse { @@ -351,10 +359,20 @@ pub async fn apply_relayed( entity: relay.into(), details: Error::Io("The relay no longer has the message.".into()), }), + RelayedStatus::Cancelled => { + rcpt.flags |= RCPT_DSN_SENT; + Status::PermanentFailure(ErrorDetails { + entity: relay.into(), + details: Error::Io("Delivery canceled.".into()), + }) + } }; if matches!( status, - RelayedStatus::Delivered { .. } | RelayedStatus::Refused { .. } | RelayedStatus::Lost + RelayedStatus::Delivered { .. } + | RelayedStatus::Refused { .. } + | RelayedStatus::Lost + | RelayedStatus::Cancelled ) { rcpt.flags &= !RCPT_RELAYED; rcpt.retry.due = now; @@ -365,6 +383,10 @@ pub async fn apply_relayed( changed = true; } } + if !waiting { + server.unlock_event(queue_id, queue_name).await; + return Applied::Gone; + } if changed || wake { message .save_changes(server, None, wake.then_some(now)) diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index 1f208771..f1e2cfbc 100644 --- a/docs/spec/features/relay.md +++ b/docs/spec/features/relay.md @@ -508,7 +508,7 @@ When no link is up, or after RL-13: | Object | What it holds | |---|---| | `inbuxa:Relay` | One per relay: `name`, `address`, `linkPort`, `enabled`, the settings pushed to the relay, `sendThrough` (route outbound mail through relays), `inboundOnly` (takes mail in but sends none, e.g. while its IP isn't warmed up or isn't in SPF yet), and read-only status mirrored from the relay: link state per node, version, queue counts, `lastSeen`. Create, or an update with `rejoin: true`, returns `joinToken` once | -| `inbuxa:RelayMessage` | Read-only view of a relay's `Inbound` and `Outbound` queues, fetched over the link when asked. Never message content | +| `inbuxa:RelayMessage` | Read-only view of a relay's `Inbound` and `Outbound` queues, fetched over the link when asked. Never message content. `/get`, `/query` (filter `relayId`, `direction`; newest first) and `/set` destroy only: discards held mail, or cancels an outbound message, whose recipients fail here with no DSN as a delivery cancelled in the queue does. Each id is the relay's id and a hash of the relay's own message id, so nothing is stored | Permissions: `sysRelayGet`, `sysRelayUpdate`, `sysRelayQueueGet`, `sysRelayQueueUpdate` (cancel an outbound message, discard held mail). The @@ -858,8 +858,17 @@ In progress. Done so far: `crates/common`. The server keeps its caches, trace events, policy fetch and test mocks around them, so its behavior is unchanged. The relay will use the same code to deliver. -- Not done yet: `inbuxa:RelayMessage`, which comes with inbuxa-admin's - Relays page (step 5), once `Outbound` exists to show beside `Inbound`. +- `inbuxa:RelayMessage` (step 5, for inbuxa-admin's Relays page): get, + query and destroy, over the asking node's link, with `sysRelayQueueGet` + and `sysRelayQueueUpdate`, and every destroy audited. A relay this node + can't reach is left out of a query of all relays; asked for by name, it + is an error (`serverUnavailable`). An outbound message cancelled while + the relay is sending it is dropped at the relay's next status. Tested + end to end: the deferred message read back with the remote server's + reply and its queue id, then cancelled, gone from both queues, with no + DSN. +- Still to come in step 5: inbuxa-admin's Relays page, certificates + (RL-27), DNS (RL-25) and upgrades (RL-29). Order: diff --git a/resources/privacy/catalog.toml b/resources/privacy/catalog.toml index d48a3035..8d8a1720 100644 --- a/resources/privacy/catalog.toml +++ b/resources/privacy/catalog.toml @@ -131,6 +131,20 @@ joinToken = ["credential"] links = ["network"] createdBy = ["identifier"] +# Kept at the relay, read over the link when asked; nothing stored here +[object."inbuxa:RelayMessage"] +file = "inbuxa_relay_message.rs" +default = "none" +whose = ["holder", "correspondent"] +where = ["external"] +scope = "server" +retention = "object-life" +[object."inbuxa:RelayMessage".properties] +envelope = ["identifier", "contact"] +session = ["network", "identifier"] +deliveryStatus = ["identifier", "network"] +doorChecks = ["network"] + [object."inbuxa:Journal"] file = "inbuxa_journal.rs" default = "none" diff --git a/tests/src/system/relay.rs b/tests/src/system/relay.rs index 6cf5809b..2a4e61dd 100644 --- a/tests/src/system/relay.rs +++ b/tests/src/system/relay.rs @@ -1026,6 +1026,92 @@ pub async fn link(test: &mut TestServer) { tokio::time::sleep(std::time::Duration::from_millis(500)).await; } + // The relay's queues, read over the link (inbuxa:RelayMessage): the + // deferred message, with the remote server's reply and its place in + // this server's queue, and never the message itself + let (name, response) = call( + admin, + "inbuxa:RelayMessage/query", + json!({"filter": {"relayId": &id, "direction": "outbound"}, "calculateTotal": true}), + ) + .await; + assert_eq!(name, "inbuxa:RelayMessage/query", "{response}"); + let ids = response["ids"].as_array().cloned().unwrap_or_default(); + assert!(ids.len() >= 2, "{response}"); + let (_, response) = call(admin, "inbuxa:RelayMessage/get", json!({"ids": ids})).await; + assert!(!response.to_string().contains("blobId"), "{response}"); + let later = response["list"] + .as_array() + .into_iter() + .flatten() + .find(|m| m["envelope"]["rcptTo"][0]["email"] == "later@remote.test") + .cloned() + .unwrap_or_else(|| panic!("the deferred message: {response}")); + assert_eq!(later["relayId"], json!(&id), "{later}"); + assert_eq!(later["direction"], "outbound", "{later}"); + assert!( + later["deliveryStatus"]["later@remote.test"]["smtpReply"] + .as_str() + .is_some_and(|r| r.contains("Try again later")), + "{later}" + ); + assert!(later["queueId"].is_string(), "{later}"); + assert!( + later["state"].is_null() && later["session"].is_null(), + "{later}" + ); + // Cancelled: gone from the relay and from this server's queue, with no + // DSN (the count below) + let (_, response) = call( + admin, + "inbuxa:RelayMessage/set", + json!({"destroy": [&later["id"]], "reason": "Sent by mistake"}), + ) + .await; + assert_eq!(response["destroyed"], json!([&later["id"]]), "{response}"); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let (_, queued) = call(admin, "x:QueuedMessage/get", json!({"ids": null})).await; + let (_, at_relay) = call( + admin, + "inbuxa:RelayMessage/get", + json!({"ids": [&later["id"]]}), + ) + .await; + let here = queued.to_string().contains("later@remote.test"); + let there = at_relay["notFound"] != json!([&later["id"]]); + if !here && !there { + break; + } + assert!( + std::time::Instant::now() < deadline, + "queued here: {here}, at the relay: {there}" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + // A relay message that isn't there, and one that can't be changed + let other = ids + .iter() + .filter_map(|i| i.as_str()) + .find(|i| *i != later["id"]) + .unwrap() + .to_string(); + let (_, response) = call( + admin, + "inbuxa:RelayMessage/set", + json!({"destroy": [&later["id"]], "update": {&other: {"size": 1}}}), + ) + .await; + assert_eq!( + response["notDestroyed"][later["id"].as_str().unwrap()]["type"], + "notFound", + "{response}" + ); + assert_eq!( + response["notUpdated"][&other]["type"], "forbidden", + "{response}" + ); + // A route naming relays (RL-24): with sendThrough off, only the // domains routed to it go through a relay, and only the relays named. // One named relay that isn't there: the mail waits, saying so