db73d56
ZhenYi
refactor: add events dependency and enhance event handling for improved clarity and organization
db73d56
use std::sync::LazyLock;
use std::time::{Duration, Instant};
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use futures_util::future;
use models::event_outbox;
use opentelemetry::metrics::Gauge;
use opentelemetry::{KeyValue, global};
use queue::{DeliveryClass, EVENT_PREFIX, Envelope, Publisher, SCHEMA_VERSION};
use sea_orm::ActiveValue::Set;
use sea_orm::{
ColumnTrait, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, DbErr,
EntityTrait, QueryFilter, Statement,
};
use serde::Serialize;
use serde_json::Value;
use tokio::task::{JoinHandle, JoinSet};
use tokio_util::sync::CancellationToken;
use tracing::Instrument;
use uuid::Uuid;
use crate::handler::DURABLE_DOMAINS;
use crate::{EventHandler, Labels, label, topic};
const BATCH_SIZE: i64 = 16;
const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);
const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
const POLL_INTERVAL: Duration = Duration::from_millis(500);
const MAX_ENVELOPE_BYTES: usize = queue::DEFAULT_MAX_MESSAGE_BYTES - 4096;
const BACKLOG_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);
/// 单条事件的重试上限。超过后丢弃并打 ERROR：否则一条毒消息会永久堵住它所在的
/// ordering_key（`auth:{user_id}` 之类），后面的事件永远发不出去。
const MAX_PUBLISH_ATTEMPTS: i32 = 50;
/// 已投递行的保留时长与清理间隔。outbox 只是投递队列，不承担归档职责。
const PURGE_INTERVAL: Duration = Duration::from_secs(600);
const PURGED_RETENTION_SECS: i64 = 7 * 24 * 60 * 60;
const OUTBOX_ENTITY_PREFIX: &str = "models::event_outbox";
static BACKLOG_AGE: LazyLock<Gauge<f64>> = LazyLock::new(|| {
global::meter("events")
.f64_gauge("event.outbox.oldest.ready.age")
.with_unit("s")
.build()
});
const CLAIM: &str = "
WITH pending AS (
SELECT candidate.id FROM event_outbox AS candidate
WHERE candidate.published_at IS NULL AND candidate.available_at <= CURRENT_TIMESTAMP
AND (candidate.leased_until IS NULL OR candidate.leased_until <= CURRENT_TIMESTAMP)
AND NOT EXISTS (
SELECT 1 FROM event_outbox AS earlier
WHERE earlier.ordering_key = candidate.ordering_key AND earlier.published_at IS NULL
AND (earlier.occurred_at, earlier.id) < (candidate.occurred_at, candidate.id)
)
ORDER BY candidate.occurred_at, candidate.id
LIMIT $1 FOR UPDATE OF candidate SKIP LOCKED
)
UPDATE event_outbox AS events
SET lease_id = $2, leased_until = CURRENT_TIMESTAMP + INTERVAL '60 seconds',
attempts = LEAST(events.attempts, 2147483646) + 1
FROM pending WHERE events.id = pending.id
RETURNING events.id, events.subject, events.kind, events.envelope, events.attempts
";
/// 建表与索引。走 `EntityFirst` 是为了复用启动迁移的 advisory lock：
/// 多实例同时启动时不会并发 `CREATE TABLE`。
pub async fn sync_schema(db: &DatabaseConnection) -> Result<(), DbErr> {
db::migrate::EntityFirst::new(&[OUTBOX_ENTITY_PREFIX])
.after([
"CREATE INDEX IF NOT EXISTS idx_event_outbox_pending ON event_outbox (available_at, occurred_at, id) WHERE published_at IS NULL",
"CREATE INDEX IF NOT EXISTS idx_event_outbox_ordering ON event_outbox (ordering_key, occurred_at, id) WHERE published_at IS NULL",
"CREATE INDEX IF NOT EXISTS idx_event_outbox_published ON event_outbox (published_at) WHERE published_at IS NOT NULL",
])
.run(db)
.await
}
fn domain(kind: &str) -> &str {
kind.split('.').next().unwrap_or_default()
}
fn is_durable_kind(kind: &str) -> bool {
DURABLE_DOMAINS.contains(&domain(kind))
}
/// 串行化粒度：同一 `ordering_key` 的事件严格按写入顺序投递。
///
/// 组织事件按组织串行，认证事件按用户串行（没有用户主体时退回请求 id，
/// 相当于不串行），邮件事件按发信请求串行。其余域沿用资源 id。
fn ordering_key(kind: &str, labels: &Labels) -> Result<String, DbErr> {
if let Some(explicit) = labels
.get(label::ORDERING_KEY)
.filter(|key| !key.is_empty())
{
return Ok(explicit.to_string());
}
let domain = domain(kind);
let key = match domain {
"guild" => labels.get(label::GUILD_ID),
"git" => labels.get(label::REPOSITORY_ID),
"pull" => labels
.get("pull_id")
.or_else(|| labels.get(label::PULL_REQUEST_ID))
.or_else(|| labels.get(label::REPOSITORY_ID)),
"pull_request" => labels
.get(label::PULL_REQUEST_ID)
.or_else(|| labels.get(label::REPOSITORY_ID)),
"issue" => labels
.get(label::ISSUE_ID)
.or_else(|| labels.get(label::BOARD_ID))
.or_else(|| labels.get(label::PULL_REQUEST_ID))
.or_else(|| labels.get(label::GUILD_ID)),
"organization" => labels.get(label::ORGANIZATION_ID),
"auth" => labels
.get(label::USER_ID)
.or_else(|| labels.get(label::REQUEST_ID)),
"email" => labels.get(label::REQUEST_ID),
_ => labels
.get(label::CHANNEL_ID)
.or_else(|| labels.get(label::GUILD_ID))
.or_else(|| labels.get(label::ROOM_ID))
.or_else(|| labels.get(label::ORGANIZATION_ID))
.or_else(|| labels.get(label::USER_ID))
.or_else(|| labels.get(label::REQUEST_ID)),
};
key.filter(|key| !key.is_empty())
.map(|key| format!("{domain}:{key}"))
.ok_or_else(|| {
DbErr::Custom(format!(
"event `{kind}` requires a user, organization or request id label"
))
})
}
#[tracing::instrument(skip_all, fields(kind, request_id = %_request_id.unwrap_or_default()))]
pub async fn enqueue<T: Serialize>(
txn: &DatabaseTransaction,
kind: &str,
labels: Labels,
payload: T,
trace_id: &str,
_request_id: Option<&str>,
) -> Result<Uuid, DbErr> {
enqueue_at(
txn,
kind,
labels,
payload,
trace_id,
Utc::now(),
_request_id,
)
.await
}
#[tracing::instrument(skip_all, fields(kind, request_id = %_request_id.unwrap_or_default()))]
pub async fn enqueue_at<T: Serialize>(
conn: &impl ConnectionTrait,
kind: &str,
labels: Labels,
payload: T,
trace_id: &str,
available_at: DateTime<Utc>,
_request_id: Option<&str>,
) -> Result<Uuid, DbErr> {
if !is_durable_kind(kind) {
return Err(DbErr::Custom(format!(
"event `{kind}` requires an outbox-enabled business domain"
)));
}
let (id, model) = build_model(kind, labels, payload, trace_id, available_at)?;
event_outbox::Entity::insert(model)
.exec_without_returning(conn)
.await?;
Ok(id)
}
pub async fn enqueue_now<T: Serialize>(
conn: &impl ConnectionTrait,
kind: &str,
labels: Labels,
payload: T,
trace_id: &str,
request_id: Option<&str>,
) -> Result<Uuid, DbErr> {
enqueue_at(
conn,
kind,
labels,
payload,
trace_id,
Utc::now(),
request_id,
)
.await
}
fn build_model<T: Serialize>(
kind: &str,
labels: Labels,
payload: T,
trace_id: &str,
available_at: DateTime<Utc>,
) -> Result<(Uuid, event_outbox::ActiveModel), DbErr> {
if !is_durable_kind(kind) {
return Err(DbErr::Custom(format!(
"event `{kind}` requires an outbox-enabled business domain"
)));
}
let ordering_key = ordering_key(kind, &labels)?;
let labels = labels.without(label::ORDERING_KEY);
let mut envelope = Envelope::new(kind, payload)
.with_delivery(DeliveryClass::BusinessEvent)
.with_source(domain(kind))
.with_labels(labels);
if !trace_id.is_empty() {
envelope = envelope.with_trace_id(trace_id);
}
let subject = format!("{EVENT_PREFIX}{}", envelope.kind);
let encoded = envelope
.encode()
.map_err(|error| DbErr::Custom(format!("encoding outbox event: {error}")))?;
if encoded.len() > MAX_ENVELOPE_BYTES {
return Err(DbErr::Custom(
"outbox event exceeds the size limit".to_owned(),
));
}
let stored = serde_json::from_slice(&encoded)
.map_err(|error| DbErr::Custom(format!("storing outbox event: {error}")))?;
let model = event_outbox::ActiveModel {
id: Set(envelope.id),
subject: Set(subject),
kind: Set(envelope.kind.clone()),
ordering_key: Set(ordering_key),
envelope: Set(stored),
occurred_at: Set(envelope.occurred_at),
available_at: Set(available_at),
..Default::default()
};
Ok((envelope.id, model))
}
fn chat_labels(room_id: Uuid, request_id: Option<&str>) -> Labels {
let mut labels = Labels::new()
.with(label::ROOM_ID, room_id)
.with(label::ORDERING_KEY, format!("chat:room:{room_id}"));
if let Some(request_id) = request_id {
labels.insert(label::REQUEST_ID, request_id);
}
labels
}
pub async fn enqueue_room<T: Serialize>(
conn: &impl ConnectionTrait,
kind: &str,
room_id: Uuid,
payload: T,
request_id: Option<&str>,
) -> Result<Uuid, DbErr> {
enqueue_at(
conn,
kind,
chat_labels(room_id, request_id),
payload,
request_id.unwrap_or_default(),
Utc::now(),
request_id,
)
.await
}
pub async fn enqueue_chat_users<T: Serialize>(
conn: &impl ConnectionTrait,
kind: &str,
uids: &[i64],
room_id: Uuid,
payload: &T,
request_id: Option<&str>,
) -> Result<(), DbErr> {
if uids.is_empty() {
return Ok(());
}
if !is_durable_kind(kind) {
return Err(DbErr::Custom(format!(
"event `{kind}` requires an outbox-enabled business domain"
)));
}
let mut models = Vec::with_capacity(uids.len());
for uid in uids {
let labels = chat_labels(room_id, request_id)
.with(label::USER_ID, *uid)
.with(label::ORDERING_KEY, format!("chat:user:{uid}"));
let (_, model) = build_model(
kind,
labels,
payload,
request_id.unwrap_or_default(),
Utc::now(),
)?;
models.push(model);
}
event_outbox::Entity::insert_many(models).exec(conn).await?;
Ok(())
}
pub async fn release(db: &DatabaseConnection, ids: &[Uuid]) -> Result<(), DbErr> {
if ids.is_empty() {
return Ok(());
}
event_outbox::Entity::update_many()
.set(event_outbox::ActiveModel {
available_at: Set(Utc::now()),
lease_id: Set(None),
leased_until: Set(None),
..Default::default()
})
.filter(event_outbox::Column::Id.is_in(ids.to_vec()))
.filter(event_outbox::Column::PublishedAt.is_null())
.exec(db)
.await?;
Ok(())
}
pub async fn park(db: &DatabaseConnection, ids: &[Uuid], grace_secs: i64) -> Result<(), DbErr> {
if ids.is_empty() {
return Ok(());
}
event_outbox::Entity::update_many()
.set(event_outbox::ActiveModel {
available_at: Set(Utc::now() + ChronoDuration::seconds(grace_secs.max(0))),
lease_id: Set(None),
leased_until: Set(None),
..Default::default()
})
.filter(event_outbox::Column::Id.is_in(ids.to_vec()))
.filter(event_outbox::Column::PublishedAt.is_null())
.exec(db)
.await?;
Ok(())
}
pub async fn discard(db: &DatabaseConnection, ids: &[Uuid]) -> Result<(), DbErr> {
if ids.is_empty() {
return Ok(());
}
event_outbox::Entity::update_many()
.set(event_outbox::ActiveModel {
published_at: Set(Some(Utc::now())),
lease_id: Set(None),
leased_until: Set(None),
..Default::default()
})
.filter(event_outbox::Column::Id.is_in(ids.to_vec()))
.filter(event_outbox::Column::PublishedAt.is_null())
.exec(db)
.await?;
Ok(())
}
struct PendingEvent {
id: Uuid,
subject: String,
kind: String,
envelope: Value,
attempts: i32,
}
fn retry_delay(attempts: i32) -> f64 {
f64::from(2_u32.pow(attempts.clamp(1, 8) as u32))
}
/// identity 事件与 apihost 事件共用一张 outbox，按 kind 的域段决定投递管线。
fn is_identity_kind(kind: &str) -> bool {
matches!(domain(kind), "auth" | "organization" | "email")
}
async fn publish_identity(
handler: Option<&EventHandler>,
row: &PendingEvent,
) -> Result<(), String> {
let handler = handler.ok_or_else(|| "identity event handler is not configured".to_owned())?;
let request_id = row
.envelope
.get("labels")
.and_then(|labels| labels.get(label::REQUEST_ID))
.or_else(|| {
row.envelope
.get("payload")
.and_then(|payload| payload.get(label::REQUEST_ID))
})
.and_then(Value::as_str)
.unwrap_or_default();
let envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())
.map_err(|error| format!("invalid outbox envelope: {error}"))?;
if envelope.id != row.id || envelope.schema_version != SCHEMA_VERSION {
return Err("outbox event identity or schema is invalid".to_owned());
}
handler
.publish_durable(&row.subject, &envelope, Some(request_id))
.await
.map_err(|error| error.to_string())
}
async fn publish_app(publisher: Option<&Publisher>, row: &PendingEvent) -> Result<(), String> {
let publisher = publisher.ok_or_else(|| "app event publisher is not configured".to_owned())?;
let envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())
.map_err(|error| format!("invalid outbox envelope: {error}"))?;
if envelope.id != row.id
|| envelope.kind != row.kind
|| envelope.schema_version != SCHEMA_VERSION
{
return Err("outbox event identity is invalid".to_owned());
}
let topic = publisher.topic(topic::of(&envelope.kind));
let body = envelope.encode().map_err(|error| error.to_string())?;
let payload = serde_json::to_value(&envelope.payload).ok();
let tag = label::tag(&envelope.kind, payload.as_ref());
publisher
.publish(
&topic,
tag,
vec![row.id.to_string()],
envelope.properties(payload.as_ref(), None),
body,
)
.await
.map_err(|error| error.to_string())
}
async fn publish(
db: &DatabaseConnection,
identity: Option<&EventHandler>,
app: Option<&Publisher>,
lease_id: Uuid,
row: PendingEvent,
) -> Result<(), DbErr> {
let request_id = row
.envelope
.get("labels")
.and_then(|labels| labels.get(label::REQUEST_ID))
.or_else(|| {
row.envelope
.get("payload")
.and_then(|payload| payload.get(label::REQUEST_ID))
})
.and_then(Value::as_str)
.unwrap_or_default()
.to_owned();
let span = tracing::info_span!(
"outbox.publish",
event_id = %row.id,
subject = %row.subject,
kind = %row.kind,
request_id = %request_id,
);
let result = async {
if is_identity_kind(&row.kind) {
publish_identity(identity, &row).await
} else {
publish_app(app, &row).await
}
}
.instrument(span);
let result = tokio::time::timeout(PUBLISH_TIMEOUT, result)
.await
.unwrap_or_else(|_| Err("outbox publish timed out".to_owned()));
let statement = match result {
Ok(()) => Statement::from_sql_and_values(
DbBackend::Postgres,
"UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
[row.id.into(), lease_id.into()],
),
Err(error) => {
let error_text: String = error.chars().take(2048).collect();
if row.attempts >= MAX_PUBLISH_ATTEMPTS {
tracing::error!(
event_id = %row.id,
subject = %row.subject,
kind = %row.kind,
attempt = row.attempts,
error = %error_text,
"outbox event exceeded the retry budget and was discarded"
);
Statement::from_sql_and_values(
DbBackend::Postgres,
"UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = $3 WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
[row.id.into(), lease_id.into(), error_text.into()],
)
} else {
tracing::warn!(event_id = %row.id, subject = %row.subject, kind = %row.kind, attempt = row.attempts, error = %error_text, "outbox publish will retry");
let delay = retry_delay(row.attempts) + f64::from(row.id.as_bytes()[15] % 5);
Statement::from_sql_and_values(
DbBackend::Postgres,
"UPDATE event_outbox SET available_at = CURRENT_TIMESTAMP + ($3 * INTERVAL '1 second'), lease_id = NULL, leased_until = NULL, last_error = $4 WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
[
row.id.into(),
lease_id.into(),
delay.into(),
error_text.into(),
],
)
}
}
};
let updated = db.execute_raw(statement).await?;
if updated.rows_affected() == 0 {
tracing::warn!(event_id = %row.id, "outbox lease was reclaimed by another relay");
}
Ok(())
}
#[tracing::instrument(skip_all, fields(request_id = %_request_id.unwrap_or_default()))]
async fn run_batch(
db: &DatabaseConnection,
identity: Option<&EventHandler>,
app: Option<&Publisher>,
_request_id: Option<&str>,
) -> Result<usize, DbErr> {
let lease_id = Uuid::now_v7();
let claimed = db
.query_all_raw(Statement::from_sql_and_values(
DbBackend::Postgres,
CLAIM,
[BATCH_SIZE.into(), lease_id.into()],
))
.await?;
let mut pending = Vec::with_capacity(claimed.len());
for row in claimed {
pending.push(PendingEvent {
id: row.try_get("", "id")?,
subject: row.try_get("", "subject")?,
kind: row.try_get("", "kind")?,
envelope: row.try_get("", "envelope")?,
attempts: row.try_get("", "attempts")?,
});
}
let count = pending.len();
let results = future::join_all(
pending
.into_iter()
.map(|row| publish(db, identity, app, lease_id, row)),
)
.await;
for result in results {
result?;
}
Ok(count)
}
async fn purge_published(db: &DatabaseConnection) {
match db
.execute_unprepared(&format!(
"DELETE FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds'"
))
.await
{
Ok(result) if result.rows_affected() > 0 => {
tracing::info!(
rows = result.rows_affected(),
"purged published outbox events"
);
}
Ok(_) => {}
Err(error) => tracing::warn!(%error, "outbox purge failed"),
}
}
async fn run_database(
domain: &'static str,
db: DatabaseConnection,
identity: Option<EventHandler>,
app: Option<Publisher>,
cancel: CancellationToken,
) {
let mut sample_at = Instant::now();
let mut purge_at = Instant::now() + PURGE_INTERVAL;
loop {
if cancel.is_cancelled() {
return;
}
if Instant::now() >= purge_at {
purge_at = Instant::now() + PURGE_INTERVAL;
purge_published(&db).await;
}
if Instant::now() >= sample_at {
sample_at = Instant::now() + BACKLOG_SAMPLE_INTERVAL;
let sample = db.query_one_raw(Statement::from_string(
DbBackend::Postgres,
"SELECT COALESCE(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - (SELECT available_at FROM event_outbox WHERE published_at IS NULL AND available_at <= CURRENT_TIMESTAMP ORDER BY available_at, occurred_at, id LIMIT 1))), 0)::double precision AS age",
)).await;
match sample {
Ok(Some(row)) => match row.try_get::<f64>("", "age") {
Ok(age) => BACKLOG_AGE.record(age.max(0.0), &[KeyValue::new("domain", domain)]),
Err(error) => {
tracing::warn!(%domain, %error, "outbox backlog sample is invalid")
}
},
Ok(None) => {}
Err(error) => tracing::warn!(%domain, %error, "outbox backlog sample failed"),
}
}
match run_batch(&db, identity.as_ref(), app.as_ref(), None).await {
Ok(count) if count > 0 => continue,
Ok(_) => {}
Err(error) => {
tracing::error!(%domain, %error, "outbox relay database operation failed")
}
}
tokio::select! {
_ = cancel.cancelled() => return,
_ = tokio::time::sleep(POLL_INTERVAL) => {}
}
}
}
pub struct OutboxRelay {
cancel: CancellationToken,
join: Option<JoinHandle<Result<(), DbErr>>>,
}
impl OutboxRelay {
pub fn start(
identity: Option<EventHandler>,
app: Option<Publisher>,
databases: Vec<(&'static str, DatabaseConnection)>,
) -> Self {
let cancel = CancellationToken::new();
let join = if identity.is_some() || app.is_some() {
let cancel = cancel.clone();
Some(tokio::spawn(async move {
let mut workers = JoinSet::new();
for (domain, db) in databases {
workers.spawn(run_database(
domain,
db,
identity.clone(),
app.clone(),
cancel.clone(),
));
}
while let Some(result) = workers.join_next().await {
result.map_err(|error| {
DbErr::Custom(format!("outbox relay task failed: {error}"))
})?;
if !cancel.is_cancelled() {
return Err(DbErr::Custom(
"outbox relay stopped unexpectedly".to_owned(),
));
}
}
Ok(())
}))
} else {
None
};
if join.is_none() {
tracing::warn!(
"event forwarding is disabled; reliable events remain pending in the database outbox"
);
}
Self { cancel, join }
}
pub async fn wait(&mut self) -> Result<(), DbErr> {
let Some(join) = self.join.as_mut() else {
return std::future::pending().await;
};
let result = join.await;
self.join.take();
result.map_err(|error| DbErr::Custom(format!("outbox relay task failed: {error}")))??;
Err(DbErr::Custom(
"outbox relay stopped unexpectedly".to_owned(),
))
}
pub async fn shutdown(mut self) -> Result<(), DbErr> {
self.cancel.cancel();
let Some(mut join) = self.join.take() else {
return Ok(());
};
match tokio::time::timeout(SHUTDOWN_TIMEOUT, &mut join).await {
Ok(result) => result
.map_err(|error| DbErr::Custom(format!("outbox relay task failed: {error}")))?,
Err(_) => {
join.abort();
let _ = join.await;
Err(DbErr::Custom(
"outbox relay shutdown timed out; pending leases will recover".to_owned(),
))
}
}
}
}
impl Drop for OutboxRelay {
fn drop(&mut self) {
self.cancel.cancel();
}
}
#[cfg(test)]
mod tests {
use super::ordering_key;
use crate::{Labels, label};
#[test]
fn auth_events_serialize_per_user_and_fall_back_to_request_id() {
let user = Labels::new()
.with(label::USER_ID, 42)
.with(label::REQUEST_ID, "req-1");
assert_eq!(
ordering_key(label::AUTH_LOGIN_CREATE, &user).unwrap(),
"auth:42"
);
let anonymous = Labels::new().with(label::REQUEST_ID, "req-2");
assert_eq!(
ordering_key(label::AUTH_LOGIN_CREATE, &anonymous).unwrap(),
"auth:req-2"
);
}
#[test]
fn organization_and_email_use_their_own_resource_ids() {
let org = Labels::new()
.with(label::ORGANIZATION_ID, "org-1")
.with(label::ACTOR_USER_ID, 7);
assert_eq!(
ordering_key(label::ORGANIZATION_LIFE_CYCLE_CREATE, &org).unwrap(),
"organization:org-1"
);
let mail = Labels::new().with(label::REQUEST_ID, "mail-1");
assert_eq!(
ordering_key(label::EMAIL_DELIVERY_CREATE, &mail).unwrap(),
"email:mail-1"
);
}
#[test]
fn events_without_any_correlation_label_are_rejected() {
let error = ordering_key(label::AUTH_LOGIN_CREATE, &Labels::new()).unwrap_err();
assert!(
error
.to_string()
.contains("requires a user, organization or request id")
);
}
}