Will Anderson
491 lines
15 KiB
Rust
491 lines
15 KiB
Rust
//! Built-in aspects for el-ui.
|
|
|
|
use std::{
|
|
collections::HashMap,
|
|
sync::{
|
|
atomic::{AtomicU64, Ordering},
|
|
Mutex,
|
|
},
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
use crate::{AopError, AopResult, Aspect, InvocationContext, InvocationResult, ProceedFn};
|
|
|
|
// ── @authenticate ─────────────────────────────────────────────────────────────
|
|
|
|
/// `@authenticate` — Requires a valid session before the method executes.
|
|
///
|
|
/// Checks `ctx.metadata["session_token"]` or `ctx.metadata["user_id"]`.
|
|
/// If absent, rejects with `AopError::Unauthenticated`.
|
|
pub struct AuthenticateAspect;
|
|
|
|
impl Aspect for AuthenticateAspect {
|
|
fn name(&self) -> &'static str {
|
|
"authenticate"
|
|
}
|
|
|
|
fn before(&self, ctx: &mut InvocationContext) -> AopResult<()> {
|
|
// Look for a session token or user ID in metadata.
|
|
// In production, the auth middleware populates these from the JWT/session.
|
|
let has_user = ctx.metadata.contains_key("user_id")
|
|
|| ctx.metadata.contains_key("session_token");
|
|
if !has_user {
|
|
return Err(AopError::Unauthenticated);
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
// ── @authorize ────────────────────────────────────────────────────────────────
|
|
|
|
/// `@authorize(role: "admin")` — Requires the caller to have a specific role.
|
|
pub struct AuthorizeAspect {
|
|
pub required_role: String,
|
|
}
|
|
|
|
impl AuthorizeAspect {
|
|
pub fn new(role: impl Into<String>) -> Self {
|
|
Self { required_role: role.into() }
|
|
}
|
|
}
|
|
|
|
impl Aspect for AuthorizeAspect {
|
|
fn name(&self) -> &'static str {
|
|
"authorize"
|
|
}
|
|
|
|
fn before(&self, ctx: &mut InvocationContext) -> AopResult<()> {
|
|
let user_roles = ctx
|
|
.metadata
|
|
.get("roles")
|
|
.map(|s| s.as_str())
|
|
.unwrap_or("");
|
|
// Roles are stored as comma-separated string: "admin,user"
|
|
let has_role = user_roles
|
|
.split(',')
|
|
.any(|r| r.trim() == self.required_role.as_str());
|
|
if !has_role {
|
|
return Err(AopError::Forbidden {
|
|
role: self.required_role.clone(),
|
|
});
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
// ── @cache ────────────────────────────────────────────────────────────────────
|
|
|
|
struct CacheEntry {
|
|
value: InvocationResult,
|
|
inserted_at: Instant,
|
|
ttl: Duration,
|
|
}
|
|
|
|
impl CacheEntry {
|
|
fn is_expired(&self) -> bool {
|
|
self.inserted_at.elapsed() > self.ttl
|
|
}
|
|
}
|
|
|
|
/// `@cache(ttl: 300)` — Cache method responses for `ttl` seconds.
|
|
///
|
|
/// Cache key is `"target::method::{args_sorted_json}"`.
|
|
pub struct CacheAspect {
|
|
pub ttl: Duration,
|
|
cache: Mutex<HashMap<String, CacheEntry>>,
|
|
}
|
|
|
|
impl CacheAspect {
|
|
pub fn new(ttl_seconds: u64) -> Self {
|
|
Self {
|
|
ttl: Duration::from_secs(ttl_seconds),
|
|
cache: Mutex::new(HashMap::new()),
|
|
}
|
|
}
|
|
|
|
fn cache_key(ctx: &InvocationContext) -> String {
|
|
let mut pairs: Vec<(&String, &String)> = ctx.args.iter().collect();
|
|
pairs.sort_by_key(|(k, _)| k.as_str());
|
|
let args = pairs
|
|
.iter()
|
|
.map(|(k, v)| format!("{}={}", k, v))
|
|
.collect::<Vec<_>>()
|
|
.join(",");
|
|
format!("{}::{}::{}", ctx.target, ctx.method, args)
|
|
}
|
|
}
|
|
|
|
impl Aspect for CacheAspect {
|
|
fn name(&self) -> &'static str {
|
|
"cache"
|
|
}
|
|
|
|
fn around(
|
|
&self,
|
|
ctx: InvocationContext,
|
|
proceed: &ProceedFn,
|
|
) -> AopResult<InvocationResult> {
|
|
let key = Self::cache_key(&ctx);
|
|
|
|
// Check cache
|
|
{
|
|
let cache = self.cache.lock().expect("cache lock poisoned");
|
|
if let Some(entry) = cache.get(&key) {
|
|
if !entry.is_expired() {
|
|
return Ok(entry.value.clone());
|
|
}
|
|
}
|
|
}
|
|
|
|
// Cache miss — proceed and store result
|
|
let result = proceed(ctx)?;
|
|
{
|
|
let mut cache = self.cache.lock().expect("cache lock poisoned");
|
|
// Evict expired entries while we're here
|
|
cache.retain(|_, v| !v.is_expired());
|
|
cache.insert(
|
|
key,
|
|
CacheEntry {
|
|
value: result.clone(),
|
|
inserted_at: Instant::now(),
|
|
ttl: self.ttl,
|
|
},
|
|
);
|
|
}
|
|
Ok(result)
|
|
}
|
|
}
|
|
|
|
// ── @rate_limit ───────────────────────────────────────────────────────────────
|
|
|
|
struct RateWindow {
|
|
count: u32,
|
|
window_start: Instant,
|
|
window_duration: Duration,
|
|
}
|
|
|
|
/// `@rate_limit(requests: 100, per: 60)` — Allow at most `requests` calls per `per` seconds.
|
|
pub struct RateLimitAspect {
|
|
pub max_requests: u32,
|
|
pub window: Duration,
|
|
state: Mutex<HashMap<String, RateWindow>>,
|
|
}
|
|
|
|
impl RateLimitAspect {
|
|
pub fn new(max_requests: u32, per_seconds: u64) -> Self {
|
|
Self {
|
|
max_requests,
|
|
window: Duration::from_secs(per_seconds),
|
|
state: Mutex::new(HashMap::new()),
|
|
}
|
|
}
|
|
|
|
/// The rate-limit key for a caller. Uses `user_id` or "anonymous".
|
|
fn caller_key(ctx: &InvocationContext) -> String {
|
|
ctx.metadata
|
|
.get("user_id")
|
|
.cloned()
|
|
.unwrap_or_else(|| "anonymous".to_string())
|
|
}
|
|
}
|
|
|
|
impl Aspect for RateLimitAspect {
|
|
fn name(&self) -> &'static str {
|
|
"rate_limit"
|
|
}
|
|
|
|
fn before(&self, ctx: &mut InvocationContext) -> AopResult<()> {
|
|
let key = Self::caller_key(ctx);
|
|
let mut state = self.state.lock().expect("rate limit lock poisoned");
|
|
let now = Instant::now();
|
|
let window = state.entry(key).or_insert(RateWindow {
|
|
count: 0,
|
|
window_start: now,
|
|
window_duration: self.window,
|
|
});
|
|
// Reset window if expired
|
|
if now.duration_since(window.window_start) >= window.window_duration {
|
|
window.count = 0;
|
|
window.window_start = now;
|
|
}
|
|
if window.count >= self.max_requests {
|
|
return Err(AopError::RateLimited {
|
|
requests: self.max_requests,
|
|
per: self.window.as_secs(),
|
|
});
|
|
}
|
|
window.count += 1;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
// ── @log ──────────────────────────────────────────────────────────────────────
|
|
|
|
/// `@log(level: "info")` — Structured logging for every method call.
|
|
pub struct LogAspect {
|
|
pub level: String,
|
|
}
|
|
|
|
impl LogAspect {
|
|
pub fn new(level: impl Into<String>) -> Self {
|
|
Self { level: level.into() }
|
|
}
|
|
}
|
|
|
|
impl Aspect for LogAspect {
|
|
fn name(&self) -> &'static str {
|
|
"log"
|
|
}
|
|
|
|
fn around(
|
|
&self,
|
|
ctx: InvocationContext,
|
|
proceed: &ProceedFn,
|
|
) -> AopResult<InvocationResult> {
|
|
// In production: use the `tracing` crate with the appropriate level macro.
|
|
let _log_entry = format!(
|
|
"[{}] {}.{}({:?})",
|
|
self.level.to_uppercase(),
|
|
ctx.target,
|
|
ctx.method,
|
|
ctx.args
|
|
);
|
|
let result = proceed(ctx.clone());
|
|
let _log_result = match &result {
|
|
Ok(r) => format!("[{}] {}.{} → ok: {}", self.level.to_uppercase(), ctx.target, ctx.method, r.value),
|
|
Err(e) => format!("[ERROR] {}.{} → err: {}", ctx.target, ctx.method, e),
|
|
};
|
|
result
|
|
}
|
|
}
|
|
|
|
// ── @validate ─────────────────────────────────────────────────────────────────
|
|
|
|
/// `@validate` — Run input validation before the method executes.
|
|
///
|
|
/// Validation rules are registered per method. If no rules are registered,
|
|
/// the aspect passes through (fail-open for ease of adoption).
|
|
pub struct ValidateAspect {
|
|
/// `"target::method"` → list of validation rules (field, rule_name)
|
|
rules: Mutex<HashMap<String, Vec<(String, String)>>>,
|
|
}
|
|
|
|
impl ValidateAspect {
|
|
pub fn new() -> Self {
|
|
Self {
|
|
rules: Mutex::new(HashMap::new()),
|
|
}
|
|
}
|
|
|
|
/// Add a validation rule. `rule` is one of: "required", "email", "min:N", "max:N".
|
|
pub fn add_rule(
|
|
&self,
|
|
target: &str,
|
|
method: &str,
|
|
field: impl Into<String>,
|
|
rule: impl Into<String>,
|
|
) {
|
|
let key = format!("{}::{}", target, method);
|
|
self.rules
|
|
.lock()
|
|
.expect("validate lock poisoned")
|
|
.entry(key)
|
|
.or_default()
|
|
.push((field.into(), rule.into()));
|
|
}
|
|
|
|
fn validate_field(value: &str, rule: &str) -> AopResult<()> {
|
|
if rule == "required" && value.trim().is_empty() {
|
|
return Err(AopError::ValidationFailed("field is required".into()));
|
|
}
|
|
if rule == "email" && !value.contains('@') {
|
|
return Err(AopError::ValidationFailed(format!(
|
|
"'{}' is not a valid email",
|
|
value
|
|
)));
|
|
}
|
|
if let Some(min_str) = rule.strip_prefix("min:") {
|
|
let min: usize = min_str.parse().unwrap_or(0);
|
|
if value.len() < min {
|
|
return Err(AopError::ValidationFailed(format!(
|
|
"minimum length is {}",
|
|
min
|
|
)));
|
|
}
|
|
}
|
|
if let Some(max_str) = rule.strip_prefix("max:") {
|
|
let max: usize = max_str.parse().unwrap_or(usize::MAX);
|
|
if value.len() > max {
|
|
return Err(AopError::ValidationFailed(format!(
|
|
"maximum length is {}",
|
|
max
|
|
)));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
impl Default for ValidateAspect {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
impl Aspect for ValidateAspect {
|
|
fn name(&self) -> &'static str {
|
|
"validate"
|
|
}
|
|
|
|
fn before(&self, ctx: &mut InvocationContext) -> AopResult<()> {
|
|
let key = format!("{}::{}", ctx.target, ctx.method);
|
|
let rules = self.rules.lock().expect("validate lock poisoned");
|
|
if let Some(field_rules) = rules.get(&key) {
|
|
for (field, rule) in field_rules {
|
|
let value = ctx.args.get(field).map(|s| s.as_str()).unwrap_or("");
|
|
Self::validate_field(value, rule)?;
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
// ── @retry ────────────────────────────────────────────────────────────────────
|
|
|
|
/// `@retry(attempts: 3, backoff: "exponential")` — Retry on failure.
|
|
pub struct RetryAspect {
|
|
pub attempts: u32,
|
|
pub backoff: BackoffStrategy,
|
|
}
|
|
|
|
#[derive(Debug, Clone, PartialEq)]
|
|
pub enum BackoffStrategy {
|
|
None,
|
|
Fixed(Duration),
|
|
Exponential { base: Duration },
|
|
}
|
|
|
|
impl RetryAspect {
|
|
pub fn new(attempts: u32) -> Self {
|
|
Self { attempts, backoff: BackoffStrategy::None }
|
|
}
|
|
|
|
pub fn with_exponential_backoff(mut self, base_ms: u64) -> Self {
|
|
self.backoff = BackoffStrategy::Exponential {
|
|
base: Duration::from_millis(base_ms),
|
|
};
|
|
self
|
|
}
|
|
|
|
pub fn with_fixed_backoff(mut self, ms: u64) -> Self {
|
|
self.backoff = BackoffStrategy::Fixed(Duration::from_millis(ms));
|
|
self
|
|
}
|
|
|
|
fn sleep_duration(&self, attempt: u32) -> Duration {
|
|
match &self.backoff {
|
|
BackoffStrategy::None => Duration::ZERO,
|
|
BackoffStrategy::Fixed(d) => *d,
|
|
BackoffStrategy::Exponential { base } => {
|
|
// base * 2^attempt, capped at 30s
|
|
let factor = 1u64 << attempt.min(10);
|
|
std::cmp::min(*base * factor as u32, Duration::from_secs(30))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Aspect for RetryAspect {
|
|
fn name(&self) -> &'static str {
|
|
"retry"
|
|
}
|
|
|
|
fn around(
|
|
&self,
|
|
ctx: InvocationContext,
|
|
proceed: &ProceedFn,
|
|
) -> AopResult<InvocationResult> {
|
|
let mut last_error = String::new();
|
|
for attempt in 0..self.attempts {
|
|
match proceed(ctx.clone()) {
|
|
Ok(result) => return Ok(result),
|
|
Err(e) => {
|
|
last_error = e.to_string();
|
|
let sleep_for = self.sleep_duration(attempt);
|
|
if sleep_for > Duration::ZERO && attempt + 1 < self.attempts {
|
|
std::thread::sleep(sleep_for);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Err(AopError::RetriesExhausted {
|
|
attempts: self.attempts,
|
|
last_error,
|
|
})
|
|
}
|
|
}
|
|
|
|
// ── @trace ────────────────────────────────────────────────────────────────────
|
|
|
|
static TRACE_COUNTER: AtomicU64 = AtomicU64::new(1);
|
|
|
|
/// `@trace` — Add a distributed tracing span to every method call.
|
|
///
|
|
/// Injects a `trace_id` and `span_id` into context metadata.
|
|
/// In production, emit the span to an OpenTelemetry collector.
|
|
pub struct TraceAspect {
|
|
pub service_name: String,
|
|
}
|
|
|
|
impl TraceAspect {
|
|
pub fn new(service_name: impl Into<String>) -> Self {
|
|
Self { service_name: service_name.into() }
|
|
}
|
|
|
|
fn new_span_id() -> String {
|
|
let id = TRACE_COUNTER.fetch_add(1, Ordering::Relaxed);
|
|
format!("span-{:016x}", id)
|
|
}
|
|
}
|
|
|
|
impl Aspect for TraceAspect {
|
|
fn name(&self) -> &'static str {
|
|
"trace"
|
|
}
|
|
|
|
fn around(
|
|
&self,
|
|
mut ctx: InvocationContext,
|
|
proceed: &ProceedFn,
|
|
) -> AopResult<InvocationResult> {
|
|
// Create or inherit trace ID
|
|
let trace_id = ctx
|
|
.metadata
|
|
.get("trace_id")
|
|
.cloned()
|
|
.unwrap_or_else(|| format!("trace-{:016x}", TRACE_COUNTER.load(Ordering::Relaxed)));
|
|
let span_id = Self::new_span_id();
|
|
|
|
ctx.metadata.insert("trace_id".into(), trace_id.clone());
|
|
ctx.metadata.insert("span_id".into(), span_id.clone());
|
|
|
|
let start = Instant::now();
|
|
let result = proceed(ctx.clone());
|
|
let duration_us = start.elapsed().as_micros();
|
|
|
|
// In production: emit span to OpenTelemetry:
|
|
// tracer.start_with_context("method_call", parent_cx)
|
|
// .set_attribute(KeyValue::new("service", self.service_name.clone()))
|
|
// .set_attribute(KeyValue::new("method", ctx.method.clone()))
|
|
// .set_attribute(KeyValue::new("duration_us", duration_us as i64))
|
|
// .end();
|
|
let _ = duration_us;
|
|
|
|
result.map(|mut r| {
|
|
r.metadata.insert("trace_id".into(), trace_id);
|
|
r.metadata.insert("span_id".into(), span_id);
|
|
r
|
|
})
|
|
}
|
|
}
|