56494b8
use std::collections::BTreeMap;
use bytes::Bytes;
use chrono::{DateTime, Utc};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;
use crate::error::{QueueError, QueueResult};
use crate::subject;
pub const PROPERTY_USER: &str = "user";
pub const PROPERTY_ORG: &str = "org";
pub const PROPERTY_REQUEST_ID: &str = "request_id";
pub const SCHEMA_VERSION: u16 = 1;
pub const EVENT_PREFIX: &str = "events.";
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct Labels(BTreeMap<String, String>);
impl Labels {
pub fn new() -> Self {
Self::default()
}
pub fn with(mut self, key: impl Into<String>, value: impl std::fmt::Display) -> Self {
self.insert(key, value);
self
}
pub fn with_opt(
mut self,
key: impl Into<String>,
value: Option<impl std::fmt::Display>,
) -> Self {
if let Some(value) = value {
self.insert(key, value);
}
self
}
pub fn insert(&mut self, key: impl Into<String>, value: impl std::fmt::Display) {
self.0.insert(key.into(), value.to_string());
}
pub fn get(&self, key: &str) -> Option<&str> {
self.0.get(key).map(String::as_str)
}
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
self.0
.iter()
.map(|(key, value)| (key.as_str(), value.as_str()))
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DeliveryClass {
#[default]
BusinessEvent,
TaskCommand,
Broadcast,
}
const fn schema_version() -> u16 {
SCHEMA_VERSION
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Envelope<T> {
pub id: Uuid,
#[serde(default = "schema_version")]
pub schema_version: u16,
#[serde(default)]
pub delivery: DeliveryClass,
pub kind: String,
#[serde(default)]
pub source: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub idempotency_key: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub trace_id: Option<String>,
#[serde(default, skip_serializing_if = "Labels::is_empty")]
pub labels: Labels,
pub occurred_at: DateTime<Utc>,
#[serde(default)]
pub attempt: u32,
pub payload: T,
}
impl<T> Envelope<T> {
pub fn new(kind: impl Into<String>, payload: T) -> Self {
Self {
id: Uuid::now_v7(),
schema_version: SCHEMA_VERSION,
delivery: DeliveryClass::BusinessEvent,
kind: kind.into(),
source: String::new(),
idempotency_key: None,
trace_id: None,
labels: Labels::new(),
occurred_at: Utc::now(),
attempt: 0,
payload,
}
}
pub fn with_delivery(mut self, delivery: DeliveryClass) -> Self {
self.delivery = delivery;
self
}
pub fn with_source(mut self, source: impl Into<String>) -> Self {
self.source = source.into();
self
}
pub fn with_idempotency_key(mut self, key: impl Into<String>) -> Self {
self.idempotency_key = Some(key.into());
self
}
pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
self.trace_id = Some(trace_id.into());
self
}
pub fn with_labels(mut self, labels: Labels) -> Self {
self.labels = labels;
self
}
pub fn with_label(mut self, key: impl Into<String>, value: impl std::fmt::Display) -> Self {
self.labels.insert(key, value);
self
}
pub fn with_attempt(mut self, attempt: u32) -> Self {
self.attempt = attempt;
self
}
pub fn message_id(&self) -> String {
if self.delivery == DeliveryClass::TaskCommand {
return self
.idempotency_key
.clone()
.unwrap_or_else(|| self.id.to_string());
}
self.id.to_string()
}
pub fn subject(&self) -> QueueResult<String> {
self.bound_subject(EVENT_PREFIX)
}
pub fn bound_subject(&self, prefix: &str) -> QueueResult<String> {
let kind = subject::normalize(&self.kind);
if !subject::is_specific(&kind) {
return Err(QueueError::payload(format!(
"envelope kind `{}` is not a concrete subject",
self.kind
)));
}
Ok(format!("{prefix}{kind}"))
}
pub fn verify_subject(&self, prefix: &str, subject_name: &str) -> QueueResult<()> {
let expected = self.bound_subject(prefix)?;
let actual = subject_name.trim().trim_matches(subject::SEPARATOR);
if expected != actual {
return Err(QueueError::payload(format!(
"subject `{subject_name}` does not carry kind `{}`",
self.kind
)));
}
Ok(())
}
pub fn verify_event_subject(&self, subject_name: &str) -> QueueResult<()> {
self.verify_subject(EVENT_PREFIX, subject_name)
}
pub fn properties(
&self,
payload: Option<&Value>,
request_id: Option<&str>,
) -> Vec<(String, String)> {
let mut properties = Vec::with_capacity(3);
let user = self
.labels
.get("user_id")
.map(str::to_owned)
.or_else(|| payload.and_then(|value| field_text(value, "user_id")))
.or_else(|| payload.and_then(|value| field_text(value, "target_user_id")))
.or_else(|| self.labels.get("actor_user_id").map(str::to_owned));
if let Some(user) = user {
properties.push((PROPERTY_USER.to_string(), user));
}
let org = self
.labels
.get("organization_id")
.map(str::to_owned)
.or_else(|| payload.and_then(|value| field_text(value, "organization_id")));
if let Some(org) = org {
properties.push((PROPERTY_ORG.to_string(), org));
}
let request_id = self
.labels
.get("request_id")
.map(str::to_owned)
.or_else(|| payload.and_then(|value| field_text(value, "request_id")))
.or_else(|| request_id.map(str::to_owned));
if let Some(request_id) = request_id {
properties.push((PROPERTY_REQUEST_ID.to_string(), request_id));
}
properties
}
}
impl<T: Serialize> Envelope<T> {
pub fn encode(&self) -> QueueResult<Bytes> {
serde_json::to_vec(self)
.map(Bytes::from)
.map_err(QueueError::from)
}
pub fn summary_body(&self, summary: &str) -> QueueResult<Bytes> {
serde_json::to_vec(&Summary {
id: self.id,
kind: &self.kind,
occurred_at: &self.occurred_at,
summary,
})
.map(Bytes::from)
.map_err(QueueError::from)
}
}
impl<T: DeserializeOwned> Envelope<T> {
pub fn decode(payload: &[u8]) -> QueueResult<Self> {
serde_json::from_slice(payload).map_err(QueueError::from)
}
}
#[derive(Serialize)]
struct Summary<'a> {
id: Uuid,
kind: &'a str,
occurred_at: &'a DateTime<Utc>,
summary: &'a str,
}
fn field_text(value: &Value, key: &str) -> Option<String> {
match value.get(key)? {
Value::String(text) => Some(text.clone()),
Value::Number(number) => Some(number.to_string()),
_ => None,
}
}