27fffe3
use std::collections::BTreeMap;
use bytes::Bytes;
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::Value;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::error::{QueueError, QueueResult};
use crate::subject;
pub const SCHEMA_VERSION: u16 = 1;
pub const EVENT_PREFIX: &str = "events.";
fn schema_version() -> u16 {
SCHEMA_VERSION
}
#[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()
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Envelope<T> {
pub id: Uuid,
#[serde(default = "schema_version")]
pub schema_version: u16,
pub kind: String,
#[serde(default)]
pub source: String,
#[serde(default, skip_serializing_if = "Labels::is_empty")]
pub labels: Labels,
pub occurred_at: OffsetDateTime,
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,
kind: kind.into(),
source: String::new(),
labels: Labels::new(),
occurred_at: OffsetDateTime::now_utc(),
payload,
}
}
pub fn with_source(mut self, source: impl Into<String>) -> Self {
self.source = source.into();
self
}
pub fn with_labels(mut self, labels: Labels) -> Self {
self.labels = labels;
self
}
pub fn subject(&self) -> 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!("{EVENT_PREFIX}{kind}"))
}
}
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)
}
pub fn properties(&self) -> Vec<(String, String)> {
let payload = serde_json::to_value(&self.payload).ok();
self.properties_of(payload.as_ref())
}
}
impl<T> Envelope<T> {
fn properties_of(&self, payload: Option<&Value>) -> Vec<(String, String)> {
let mut properties = Vec::with_capacity(4);
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(("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(("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")));
if let Some(request_id) = request_id {
properties.push(("request_id".to_string(), request_id));
}
let room = self
.labels
.get("room_id")
.map(str::to_owned)
.or_else(|| payload.and_then(|value| field_text(value, "room_id")));
if let Some(room) = room {
properties.push(("room_id".to_string(), room));
}
for key in ["issue_id", "pull_id"] {
let value = self
.labels
.get(key)
.map(str::to_owned)
.or_else(|| payload.and_then(|value| field_text(value, key)));
if let Some(value) = value {
properties.push((key.to_string(), value));
}
}
properties
}
}
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 OffsetDateTime,
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,
}
}