The Universal Event Model

use std::str::FromStr; use thiserror::Error; use serde::{Serialize, Deserialize}; /// The Universal Event Model identifier #[derive(Debug, Clone, PartialEq, Eq, Hash)] #[derive(Serialize, Deserialize)] pub struct EventId { pub domain: String, pub entity: String, pub action: String, pub version: u32, } #[derive(Error, Debug)] pub enum EventIdError { #[error("Invalid format")] InvalidFormat, #[error("Invalid version: {0}")] InvalidVersion(#[from] std::num::ParseIntError), } impl FromStr for EventId { type Err = EventIdError; fn from_str(s: &str) -> Result<Self, Self::Err> { let (body, ver) = s.rsplit_once(':') .ok_or(EventIdError::InvalidFormat)?; let parts: Vec<&str> = body.split('.').collect(); if parts.len() != 3 { return Err(EventIdError::InvalidFormat); } Ok(EventId { domain: parts[0].to_string(), entity: parts[1].to_string(), action: parts[2].to_string(), version: ver.parse()?, }) } }

/// emit! macro for type-safe event creation macro_rules! emit { ($d:ident.$e:ident.$a:ident:$v:literal, $($key:ident: $val:expr),* $(,)?) => {{ let event_id = EventId { domain: stringify!($d).to_string(), entity: stringify!($e).to_string(), action: stringify!($a).to_string(), version: $v, }; let mut payload = serde_json::Map::new(); $( payload.insert( stringify!($key).to_string(), serde_json::to_value(&$val).unwrap() ); )* Event::new(event_id, payload) }}; } // Usage - reads like a sentence let event = emit!(ai.code.generated:1, file_path: "src/auth/login.rs", lines: 47, model: "claude-3.5-sonnet", confidence: 0.94, ); collector.send(event).await?;

use tokio::sync::mpsc; use sqlx::PgPool; pub struct EventCollector { sender: mpsc::Sender<Event>, } impl EventCollector { pub fn new(pool: PgPool) -> Self { let (tx, mut rx) = mpsc::channel::<Event>(10_000); tokio::spawn(async move { while let Some(event) = rx.recv().await { sqlx::query(" INSERT INTO events (id, domain, entity, action, version, payload) VALUES ($1, $2, $3, $4, $5, $6) ") .bind(&event.id) .bind(&event.event_id.domain) .bind(&event.event_id.entity) .bind(&event.event_id.action) .bind(event.event_id.version as i32) .bind(&event.payload) .execute(&pool).await.ok(); } }); Self { sender: tx } } pub async fn send(&self, e: Event) -> Result<()> { self.sender.send(e).await?; Ok(()) } }

// Query all events in a domain pub async fn query_by_domain( pool: &PgPool, domain: &str, ) -> Result<Vec<Event>> { sqlx::query_as(" SELECT * FROM events WHERE domain = $1 ORDER BY timestamp DESC ") .bind(domain) .fetch_all(pool).await } // Pattern match with LIKE pub async fn query_pattern( pool: &PgPool, pattern: &str, // e.g. "ai.%.generated" ) -> Result<Vec<Event>> { let sql_pattern = pattern.replace('%', "%"); sqlx::query_as(" SELECT * FROM events WHERE CONCAT(domain,'.',entity,'.',action) LIKE $1 ") .bind(&sql_pattern) .fetch_all(pool).await } // Count by domain for analytics pub async fn count_by_domain( pool: &PgPool, ) -> Result<Vec<(String, i64)>> { sqlx::query_as(" SELECT domain, COUNT(*) as count FROM events GROUP BY domain ORDER BY count DESC ") .fetch_all(pool).await }