platform/
recording.rs

1//! Unified runtime recording model.
2//!
3//! This module provides a typed `Record<{...}>` model that can be routed to
4//! observability/audit/security/operations channels with one entry point.
5
6use chrono::Utc;
7use once_cell::sync::Lazy;
8use serde::Serialize;
9use serde_json::{Map, Value, json};
10use std::fmt;
11use std::sync::OnceLock;
12use thiserror::Error;
13use uuid::Uuid;
14
15/// Recording channel mask: observability.
16pub const CHANNEL_OBSERVABILITY: u8 = 0b0001;
17/// Recording channel mask: audit.
18pub const CHANNEL_AUDIT: u8 = 0b0010;
19/// Recording channel mask: security.
20pub const CHANNEL_SECURITY: u8 = 0b0100;
21/// Recording channel mask: operations.
22pub const CHANNEL_OPERATIONS: u8 = 0b1000;
23
24const CHANNEL_MASK_ALL: u8 =
25    CHANNEL_OBSERVABILITY | CHANNEL_AUDIT | CHANNEL_SECURITY | CHANNEL_OPERATIONS;
26
27static FALLBACK_SOURCE_NODE: Lazy<Option<String>> = Lazy::new(detect_source_node);
28
29/// Stable identifier for each recording item.
30#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
31pub struct RecordId(pub String);
32
33impl RecordId {
34    /// Create a new random record id.
35    pub fn new() -> Self {
36        Self(Uuid::new_v4().to_string())
37    }
38}
39
40impl Default for RecordId {
41    fn default() -> Self {
42        Self::new()
43    }
44}
45
46impl fmt::Display for RecordId {
47    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
48        self.0.fmt(f)
49    }
50}
51
52/// Unified severity level for runtime recording.
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
54#[serde(rename_all = "lowercase")]
55pub enum RecordLevel {
56    Trace,
57    Debug,
58    #[default]
59    Info,
60    Warn,
61    Error,
62}
63
64/// Result/outcome of one runtime action.
65#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
66#[serde(rename_all = "snake_case")]
67pub enum Outcome {
68    Success,
69    Failure,
70    Denied,
71    Timeout,
72    #[default]
73    Unknown,
74}
75
76/// Common metadata shared by all channel payloads.
77#[derive(Debug, Clone, Serialize)]
78pub struct Common {
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub actor: Option<String>,
81    pub outcome: Outcome,
82    pub level: RecordLevel,
83    #[serde(skip_serializing_if = "Option::is_none")]
84    pub trace_id: Option<String>,
85    #[serde(skip_serializing_if = "Option::is_none")]
86    pub span_id: Option<String>,
87    #[serde(skip_serializing_if = "Option::is_none")]
88    pub request_id: Option<String>,
89    #[serde(skip_serializing_if = "Option::is_none")]
90    pub source_service: Option<String>,
91    #[serde(skip_serializing_if = "Option::is_none")]
92    pub source_node: Option<String>,
93}
94
95impl Default for Common {
96    fn default() -> Self {
97        Self {
98            actor: None,
99            outcome: Outcome::Unknown,
100            level: RecordLevel::Info,
101            trace_id: None,
102            span_id: None,
103            request_id: None,
104            source_service: None,
105            source_node: None,
106        }
107    }
108}
109
110/// Observability-specific payload.
111#[derive(Debug, Clone, Serialize, Default)]
112pub struct ObservabilityPayload {
113    /// Human readable summary for operators.
114    #[serde(skip_serializing_if = "Option::is_none")]
115    pub summary: Option<String>,
116    /// Component/subsystem that emitted the record.
117    #[serde(skip_serializing_if = "Option::is_none")]
118    pub component: Option<String>,
119    /// Logical operation name, such as `http.route.add`.
120    #[serde(skip_serializing_if = "Option::is_none")]
121    pub operation: Option<String>,
122    /// Network/runtime protocol context.
123    #[serde(skip_serializing_if = "Option::is_none")]
124    pub protocol: Option<Protocol>,
125    /// HTTP/gRPC/ws route or endpoint pattern.
126    #[serde(skip_serializing_if = "Option::is_none")]
127    pub route: Option<String>,
128    /// HTTP/gRPC method verb.
129    #[serde(skip_serializing_if = "Option::is_none")]
130    pub method: Option<String>,
131    /// Response status code when applicable.
132    #[serde(skip_serializing_if = "Option::is_none")]
133    pub status_code: Option<u16>,
134    /// End-to-end duration in milliseconds.
135    #[serde(skip_serializing_if = "Option::is_none")]
136    pub duration_ms: Option<u64>,
137}
138
139/// Runtime/request protocol.
140#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
141#[serde(rename_all = "snake_case")]
142pub enum Protocol {
143    #[default]
144    Internal,
145    Http,
146    Grpc,
147    #[serde(rename = "websocket")]
148    WebSocket,
149    Stun,
150    Turn,
151    Tcp,
152    Udp,
153}
154
155impl ObservabilityPayload {
156    fn validate(&self) -> Result<(), RecordingError> {
157        if self.summary.is_none()
158            && self.component.is_none()
159            && self.operation.is_none()
160            && self.protocol.is_none()
161            && self.route.is_none()
162            && self.method.is_none()
163            && self.status_code.is_none()
164            && self.duration_ms.is_none()
165        {
166            return Err(RecordingError::EmptyPayload {
167                channel: "observability",
168            });
169        }
170
171        validate_optional_non_empty("observability", "summary", &self.summary)?;
172        validate_optional_non_empty("observability", "component", &self.component)?;
173        validate_optional_non_empty("observability", "operation", &self.operation)?;
174        validate_optional_non_empty("observability", "route", &self.route)?;
175        validate_optional_non_empty("observability", "method", &self.method)?;
176
177        Ok(())
178    }
179}
180
181/// Audit-specific payload.
182#[derive(Debug, Clone, Serialize)]
183pub struct AuditPayload {
184    /// Action verb, e.g. `config.update`.
185    pub action: String,
186    /// Target resource class, e.g. `service.binding`.
187    pub resource: String,
188    /// Stable resource id when available.
189    #[serde(skip_serializing_if = "Option::is_none")]
190    pub resource_id: Option<String>,
191    /// Optional business/operator reason for the action.
192    #[serde(skip_serializing_if = "Option::is_none")]
193    pub reason: Option<String>,
194    /// Optional source address for the action.
195    #[serde(skip_serializing_if = "Option::is_none")]
196    pub remote_addr: Option<String>,
197    /// Optional session/correlation id.
198    #[serde(skip_serializing_if = "Option::is_none")]
199    pub session_id: Option<String>,
200    /// Before snapshot (JSON) for change actions.
201    #[serde(skip_serializing_if = "Option::is_none")]
202    pub before: Option<Value>,
203    /// After snapshot (JSON) for change actions.
204    #[serde(skip_serializing_if = "Option::is_none")]
205    pub after: Option<Value>,
206}
207
208impl AuditPayload {
209    fn validate(&self) -> Result<(), RecordingError> {
210        validate_required_non_empty("audit", "action", &self.action)?;
211        validate_required_non_empty("audit", "resource", &self.resource)?;
212        validate_optional_non_empty("audit", "resource_id", &self.resource_id)?;
213        validate_optional_non_empty("audit", "reason", &self.reason)?;
214        validate_optional_non_empty("audit", "remote_addr", &self.remote_addr)?;
215        validate_optional_non_empty("audit", "session_id", &self.session_id)?;
216
217        Ok(())
218    }
219}
220
221/// Security impact level.
222#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
223#[serde(rename_all = "snake_case")]
224pub enum SecuritySeverity {
225    Low,
226    #[default]
227    Medium,
228    High,
229    Critical,
230}
231
232// ── Per-channel semantic filters ────────────────────────────────────────
233
234/// Observability channel filter — controls resolution (how much detail).
235#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
236pub enum ObservabilityFilter {
237    Off,
238    #[default]
239    Digest,
240    Detailed,
241    Full,
242}
243
244/// Audit channel filter — controls scope (which actions).
245#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
246pub enum AuditFilter {
247    Off,
248    #[default]
249    Mutations,
250    All,
251}
252
253/// Security channel filter — controls severity threshold.
254#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
255pub enum SecurityFilter {
256    Off,
257    Critical,
258    High,
259    Medium,
260    #[default]
261    All,
262}
263
264/// Operations channel filter — controls detail level.
265#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
266pub enum OperationsFilter {
267    Off,
268    #[default]
269    Lifecycle,
270    Detailed,
271}
272
273/// Holds the active filter for each channel.
274#[derive(Debug, Clone, Default)]
275pub struct ChannelFilters {
276    pub observability: ObservabilityFilter,
277    pub audit: AuditFilter,
278    pub security: SecurityFilter,
279    pub operations: OperationsFilter,
280}
281
282static CHANNEL_FILTERS: OnceLock<ChannelFilters> = OnceLock::new();
283
284/// Install channel filters (call once at pipeline init).
285pub fn set_channel_filters(filters: ChannelFilters) {
286    let _ = CHANNEL_FILTERS.set(filters);
287}
288
289/// Read the active channel filters (returns defaults if never set).
290pub fn get_channel_filters() -> &'static ChannelFilters {
291    static DEFAULT: Lazy<ChannelFilters> = Lazy::new(ChannelFilters::default);
292    CHANNEL_FILTERS.get().unwrap_or(&DEFAULT)
293}
294
295/// Parse a config string into an `ObservabilityFilter`.
296pub fn parse_observability_filter(s: &str) -> ObservabilityFilter {
297    match s {
298        "off" => ObservabilityFilter::Off,
299        "digest" => ObservabilityFilter::Digest,
300        "detailed" => ObservabilityFilter::Detailed,
301        "full" => ObservabilityFilter::Full,
302        _ => ObservabilityFilter::default(),
303    }
304}
305
306/// Parse a config string into an `AuditFilter`.
307pub fn parse_audit_filter(s: &str) -> AuditFilter {
308    match s {
309        "off" => AuditFilter::Off,
310        "mutations" => AuditFilter::Mutations,
311        "all" => AuditFilter::All,
312        _ => AuditFilter::default(),
313    }
314}
315
316/// Parse a config string into a `SecurityFilter`.
317pub fn parse_security_filter(s: &str) -> SecurityFilter {
318    match s {
319        "off" => SecurityFilter::Off,
320        "critical" => SecurityFilter::Critical,
321        "high" => SecurityFilter::High,
322        "medium" => SecurityFilter::Medium,
323        "all" => SecurityFilter::All,
324        _ => SecurityFilter::default(),
325    }
326}
327
328/// Parse a config string into an `OperationsFilter`.
329pub fn parse_operations_filter(s: &str) -> OperationsFilter {
330    match s {
331        "off" => OperationsFilter::Off,
332        "lifecycle" => OperationsFilter::Lifecycle,
333        "detailed" => OperationsFilter::Detailed,
334        _ => OperationsFilter::default(),
335    }
336}
337
338// ── Gate functions ──────────────────────────────────────────────────────
339
340/// Read-only verbs for audit mutation detection.
341const AUDIT_READ_PREFIXES: &[&str] = &["read", "get", "list", "query", "check", "view", "describe"];
342
343fn is_audit_read_action(action: &str) -> bool {
344    let lower = action.to_ascii_lowercase();
345    AUDIT_READ_PREFIXES
346        .iter()
347        .any(|prefix| lower.starts_with(prefix))
348}
349
350fn should_emit_observability(common: &Common, filter: ObservabilityFilter) -> bool {
351    match filter {
352        ObservabilityFilter::Off => false,
353        ObservabilityFilter::Digest => matches!(
354            common.level,
355            RecordLevel::Info | RecordLevel::Warn | RecordLevel::Error
356        ),
357        ObservabilityFilter::Detailed => matches!(
358            common.level,
359            RecordLevel::Debug | RecordLevel::Info | RecordLevel::Warn | RecordLevel::Error
360        ),
361        ObservabilityFilter::Full => true,
362    }
363}
364
365fn should_emit_audit(payload: &AuditPayload, filter: AuditFilter) -> bool {
366    match filter {
367        AuditFilter::Off => false,
368        AuditFilter::Mutations => !is_audit_read_action(&payload.action),
369        AuditFilter::All => true,
370    }
371}
372
373fn should_emit_security(payload: &SecurityPayload, filter: SecurityFilter) -> bool {
374    match filter {
375        SecurityFilter::Off => false,
376        SecurityFilter::Critical => matches!(payload.severity, SecuritySeverity::Critical),
377        SecurityFilter::High => matches!(
378            payload.severity,
379            SecuritySeverity::High | SecuritySeverity::Critical
380        ),
381        SecurityFilter::Medium => matches!(
382            payload.severity,
383            SecuritySeverity::Medium | SecuritySeverity::High | SecuritySeverity::Critical
384        ),
385        SecurityFilter::All => true,
386    }
387}
388
389fn should_emit_operations(common: &Common, filter: OperationsFilter) -> bool {
390    match filter {
391        OperationsFilter::Off => false,
392        OperationsFilter::Lifecycle => matches!(
393            common.level,
394            RecordLevel::Info | RecordLevel::Warn | RecordLevel::Error
395        ),
396        OperationsFilter::Detailed => true,
397    }
398}
399
400/// Security-specific payload.
401#[derive(Debug, Clone, Serialize)]
402pub struct SecurityPayload {
403    /// Security control/policy/rule that evaluated this action.
404    pub control: String,
405    /// Severity for triage/alerting.
406    pub severity: SecuritySeverity,
407    /// Security category, e.g. `authn`, `authz`, `abuse`, `network`.
408    #[serde(skip_serializing_if = "Option::is_none")]
409    pub category: Option<String>,
410    /// Subject under evaluation (actor/user/device/service).
411    #[serde(skip_serializing_if = "Option::is_none")]
412    pub subject: Option<String>,
413    /// Source address when network-related.
414    #[serde(skip_serializing_if = "Option::is_none")]
415    pub source_addr: Option<String>,
416    /// Destination address when network-related.
417    #[serde(skip_serializing_if = "Option::is_none")]
418    pub destination_addr: Option<String>,
419    /// Extra evidence/context (JSON).
420    #[serde(skip_serializing_if = "Option::is_none")]
421    pub evidence: Option<Value>,
422}
423
424impl SecurityPayload {
425    fn validate(&self) -> Result<(), RecordingError> {
426        validate_required_non_empty("security", "control", &self.control)?;
427        validate_optional_non_empty("security", "category", &self.category)?;
428        validate_optional_non_empty("security", "subject", &self.subject)?;
429        validate_optional_non_empty("security", "source_addr", &self.source_addr)?;
430        validate_optional_non_empty("security", "destination_addr", &self.destination_addr)?;
431
432        Ok(())
433    }
434}
435
436/// Operations-specific payload.
437#[derive(Debug, Clone, Serialize)]
438pub struct OperationsPayload {
439    /// Operation name, e.g. `service.restart`.
440    pub operation: String,
441    /// Component/service affected by the operation.
442    #[serde(skip_serializing_if = "Option::is_none")]
443    pub component: Option<String>,
444    /// Runbook name/id.
445    #[serde(skip_serializing_if = "Option::is_none")]
446    pub runbook: Option<String>,
447    /// Ticket/incident id.
448    #[serde(skip_serializing_if = "Option::is_none")]
449    pub ticket: Option<String>,
450    /// Deployment/change id.
451    #[serde(skip_serializing_if = "Option::is_none")]
452    pub change_id: Option<String>,
453    /// Retry/attempt number.
454    #[serde(skip_serializing_if = "Option::is_none")]
455    pub attempt: Option<u32>,
456    /// Execution window in milliseconds.
457    #[serde(skip_serializing_if = "Option::is_none")]
458    pub duration_ms: Option<u64>,
459}
460
461impl OperationsPayload {
462    fn validate(&self) -> Result<(), RecordingError> {
463        validate_required_non_empty("operations", "operation", &self.operation)?;
464        validate_optional_non_empty("operations", "component", &self.component)?;
465        validate_optional_non_empty("operations", "runbook", &self.runbook)?;
466        validate_optional_non_empty("operations", "ticket", &self.ticket)?;
467        validate_optional_non_empty("operations", "change_id", &self.change_id)?;
468
469        Ok(())
470    }
471}
472
473#[derive(Debug, Clone)]
474struct Callsite {
475    file: &'static str,
476    line: u32,
477    module_path: &'static str,
478}
479
480fn enrich_common(mut common: Common, callsite: &Callsite) -> Common {
481    common.actor = sanitize_optional_string(common.actor);
482    common.request_id = sanitize_optional_string(common.request_id);
483
484    let explicit_trace_id = sanitize_optional_string(common.trace_id);
485    let explicit_span_id = sanitize_optional_string(common.span_id);
486    let (auto_trace_id, auto_span_id) = if explicit_trace_id.is_none() || explicit_span_id.is_none()
487    {
488        detect_current_trace_context()
489    } else {
490        (None, None)
491    };
492    common.trace_id = explicit_trace_id.or(auto_trace_id);
493    common.span_id = explicit_span_id.or(auto_span_id);
494
495    let explicit_source_service = sanitize_optional_string(common.source_service);
496    common.source_service = explicit_source_service.or_else(|| detect_source_service(callsite));
497
498    let explicit_source_node = sanitize_optional_string(common.source_node);
499    common.source_node = explicit_source_node.or_else(|| FALLBACK_SOURCE_NODE.as_ref().cloned());
500
501    common
502}
503
504fn sanitize_optional_string(value: Option<String>) -> Option<String> {
505    value.and_then(sanitize_string)
506}
507
508fn sanitize_string(value: String) -> Option<String> {
509    let trimmed = value.trim();
510    if trimmed.is_empty() {
511        return None;
512    }
513
514    Some(trimmed.to_string())
515}
516
517fn detect_source_service(callsite: &Callsite) -> Option<String> {
518    let source_service = std::env::var("ACTRIX_SOURCE_SERVICE")
519        .ok()
520        .and_then(sanitize_string);
521    if source_service.is_some() {
522        return source_service;
523    }
524
525    if callsite.module_path == "unknown" {
526        return None;
527    }
528
529    callsite
530        .module_path
531        .split("::")
532        .next()
533        .and_then(|value| sanitize_string(value.to_string()))
534}
535
536fn detect_source_node() -> Option<String> {
537    std::env::var("ACTRIX_SOURCE_NODE")
538        .ok()
539        .and_then(sanitize_string)
540        .or_else(|| std::env::var("HOSTNAME").ok().and_then(sanitize_string))
541        .or_else(|| std::env::var("COMPUTERNAME").ok().and_then(sanitize_string))
542        .or_else(|| {
543            std::fs::read_to_string("/etc/hostname")
544                .ok()
545                .and_then(sanitize_string)
546        })
547}
548
549#[cfg(feature = "opentelemetry")]
550fn detect_current_trace_context() -> (Option<String>, Option<String>) {
551    use opentelemetry::trace::TraceContextExt;
552    use tracing_opentelemetry::OpenTelemetrySpanExt;
553
554    let context = tracing::Span::current().context();
555    let span_ref = context.span();
556    let span_context = span_ref.span_context();
557    if !span_context.is_valid() {
558        return (None, None);
559    }
560
561    (
562        Some(span_context.trace_id().to_string()),
563        Some(span_context.span_id().to_string()),
564    )
565}
566
567#[cfg(not(feature = "opentelemetry"))]
568fn detect_current_trace_context() -> (Option<String>, Option<String>) {
569    (None, None)
570}
571
572/// Full recording item, with channel mask on type level.
573#[derive(Debug, Clone)]
574pub struct Record<const CHANNEL_MASK: u8> {
575    pub id: RecordId,
576    pub kind: String,
577    pub common: Common,
578    pub observability: Option<ObservabilityPayload>,
579    pub audit: Option<AuditPayload>,
580    pub security: Option<SecurityPayload>,
581    pub operations: Option<OperationsPayload>,
582    pub attributes: Map<String, Value>,
583}
584
585impl<const CHANNEL_MASK: u8> Record<CHANNEL_MASK> {
586    /// Create a new record with defaults.
587    pub fn new(kind: impl Into<String>) -> Self {
588        Self {
589            id: RecordId::new(),
590            kind: kind.into(),
591            common: Common::default(),
592            observability: None,
593            audit: None,
594            security: None,
595            operations: None,
596            attributes: Map::new(),
597        }
598    }
599
600    fn validate(&self) -> Result<(), RecordingError> {
601        validate_channel_mask::<CHANNEL_MASK>()?;
602
603        if self.kind.trim().is_empty() {
604            return Err(RecordingError::EmptyKind);
605        }
606
607        validate_channel_payload(
608            CHANNEL_MASK,
609            CHANNEL_OBSERVABILITY,
610            self.observability.is_some(),
611            "observability",
612        )?;
613        validate_channel_payload(CHANNEL_MASK, CHANNEL_AUDIT, self.audit.is_some(), "audit")?;
614        validate_channel_payload(
615            CHANNEL_MASK,
616            CHANNEL_SECURITY,
617            self.security.is_some(),
618            "security",
619        )?;
620        validate_channel_payload(
621            CHANNEL_MASK,
622            CHANNEL_OPERATIONS,
623            self.operations.is_some(),
624            "operations",
625        )?;
626
627        if let Some(payload) = &self.observability {
628            payload.validate()?;
629        }
630        if let Some(payload) = &self.audit {
631            payload.validate()?;
632        }
633        if let Some(payload) = &self.security {
634            payload.validate()?;
635        }
636        if let Some(payload) = &self.operations {
637            payload.validate()?;
638        }
639
640        Ok(())
641    }
642}
643
644/// Runtime recording validation/emit error.
645#[derive(Debug, Error)]
646pub enum RecordingError {
647    #[error("recording channel mask cannot be empty")]
648    EmptyChannelMask,
649    #[error("recording channel mask has unsupported bits: {mask:#010b}")]
650    UnsupportedChannelMask { mask: u8 },
651    #[error("recording kind cannot be empty")]
652    EmptyKind,
653    #[error("channel '{channel}' was selected but payload is missing")]
654    MissingSelectedChannelPayload { channel: &'static str },
655    #[error("channel '{channel}' payload is provided but channel was not selected")]
656    UnexpectedChannelPayload { channel: &'static str },
657    #[error("channel '{channel}' field '{field}' cannot be empty")]
658    EmptyPayloadField {
659        channel: &'static str,
660        field: &'static str,
661    },
662    #[error("channel '{channel}' payload cannot be empty")]
663    EmptyPayload { channel: &'static str },
664    #[error("failed to serialize '{channel}' payload: {error}")]
665    SerializePayload {
666        channel: &'static str,
667        #[source]
668        error: serde_json::Error,
669    },
670}
671
672/// Emit a typed recording through current tracing backend.
673pub fn emit<const CHANNEL_MASK: u8>(record: Record<CHANNEL_MASK>) -> Result<(), RecordingError> {
674    emit_with_callsite(record, "unknown", 0, "unknown")
675}
676
677/// Emit a typed recording with explicit callsite metadata.
678pub fn emit_with_callsite<const CHANNEL_MASK: u8>(
679    mut record: Record<CHANNEL_MASK>,
680    file: &'static str,
681    line: u32,
682    module_path: &'static str,
683) -> Result<(), RecordingError> {
684    let callsite = Callsite {
685        file,
686        line,
687        module_path,
688    };
689    record.common = enrich_common(record.common, &callsite);
690    record.validate()?;
691
692    let filters = get_channel_filters();
693    let level = record.common.level;
694    let common = record.common;
695    let Record {
696        id,
697        kind,
698        common: _,
699        observability,
700        audit,
701        security,
702        operations,
703        attributes,
704    } = record;
705
706    if let Some(payload) = observability
707        && should_emit_observability(&common, filters.observability)
708    {
709        let value = build_channel_record(
710            "observability",
711            &id,
712            &kind,
713            &common,
714            payload,
715            &attributes,
716            &callsite,
717        )?;
718        emit_channel(level, Channel::Observability, &value);
719    }
720
721    if let Some(payload) = audit
722        && should_emit_audit(&payload, filters.audit)
723    {
724        let value = build_channel_record(
725            "audit",
726            &id,
727            &kind,
728            &common,
729            payload,
730            &attributes,
731            &callsite,
732        )?;
733        emit_channel(level, Channel::Audit, &value);
734    }
735
736    if let Some(payload) = security
737        && should_emit_security(&payload, filters.security)
738    {
739        let value = build_channel_record(
740            "security",
741            &id,
742            &kind,
743            &common,
744            payload,
745            &attributes,
746            &callsite,
747        )?;
748        emit_channel(level, Channel::Security, &value);
749    }
750
751    if let Some(payload) = operations
752        && should_emit_operations(&common, filters.operations)
753    {
754        let value = build_channel_record(
755            "operations",
756            &id,
757            &kind,
758            &common,
759            payload,
760            &attributes,
761            &callsite,
762        )?;
763        emit_channel(level, Channel::Operations, &value);
764    }
765
766    Ok(())
767}
768
769/// Emit one serialized channel record through the unified tracing pipeline.
770#[derive(Debug, Clone, Copy)]
771enum Channel {
772    Observability,
773    Audit,
774    Security,
775    Operations,
776}
777
778/// Emit one serialized channel record through the unified tracing pipeline.
779#[allow(clippy::disallowed_macros)]
780fn emit_channel(level: RecordLevel, channel: Channel, record: &Value) {
781    match channel {
782        Channel::Observability => match level {
783            RecordLevel::Trace => {
784                tracing::event!(target: "actrix::observability", tracing::Level::TRACE, recording = %record)
785            }
786            RecordLevel::Debug => {
787                tracing::event!(target: "actrix::observability", tracing::Level::DEBUG, recording = %record)
788            }
789            RecordLevel::Info => {
790                tracing::event!(target: "actrix::observability", tracing::Level::INFO, recording = %record)
791            }
792            RecordLevel::Warn => {
793                tracing::event!(target: "actrix::observability", tracing::Level::WARN, recording = %record)
794            }
795            RecordLevel::Error => {
796                tracing::event!(target: "actrix::observability", tracing::Level::ERROR, recording = %record)
797            }
798        },
799        Channel::Audit => match level {
800            RecordLevel::Trace => {
801                tracing::event!(target: "actrix::audit", tracing::Level::TRACE, recording = %record)
802            }
803            RecordLevel::Debug => {
804                tracing::event!(target: "actrix::audit", tracing::Level::DEBUG, recording = %record)
805            }
806            RecordLevel::Info => {
807                tracing::event!(target: "actrix::audit", tracing::Level::INFO, recording = %record)
808            }
809            RecordLevel::Warn => {
810                tracing::event!(target: "actrix::audit", tracing::Level::WARN, recording = %record)
811            }
812            RecordLevel::Error => {
813                tracing::event!(target: "actrix::audit", tracing::Level::ERROR, recording = %record)
814            }
815        },
816        Channel::Security => match level {
817            RecordLevel::Trace => {
818                tracing::event!(target: "actrix::security", tracing::Level::TRACE, recording = %record)
819            }
820            RecordLevel::Debug => {
821                tracing::event!(target: "actrix::security", tracing::Level::DEBUG, recording = %record)
822            }
823            RecordLevel::Info => {
824                tracing::event!(target: "actrix::security", tracing::Level::INFO, recording = %record)
825            }
826            RecordLevel::Warn => {
827                tracing::event!(target: "actrix::security", tracing::Level::WARN, recording = %record)
828            }
829            RecordLevel::Error => {
830                tracing::event!(target: "actrix::security", tracing::Level::ERROR, recording = %record)
831            }
832        },
833        Channel::Operations => match level {
834            RecordLevel::Trace => {
835                tracing::event!(target: "actrix::operations", tracing::Level::TRACE, recording = %record)
836            }
837            RecordLevel::Debug => {
838                tracing::event!(target: "actrix::operations", tracing::Level::DEBUG, recording = %record)
839            }
840            RecordLevel::Info => {
841                tracing::event!(target: "actrix::operations", tracing::Level::INFO, recording = %record)
842            }
843            RecordLevel::Warn => {
844                tracing::event!(target: "actrix::operations", tracing::Level::WARN, recording = %record)
845            }
846            RecordLevel::Error => {
847                tracing::event!(target: "actrix::operations", tracing::Level::ERROR, recording = %record)
848            }
849        },
850    }
851}
852
853fn validate_required_non_empty(
854    channel: &'static str,
855    field: &'static str,
856    value: &str,
857) -> Result<(), RecordingError> {
858    if value.trim().is_empty() {
859        return Err(RecordingError::EmptyPayloadField { channel, field });
860    }
861
862    Ok(())
863}
864
865fn validate_optional_non_empty(
866    channel: &'static str,
867    field: &'static str,
868    value: &Option<String>,
869) -> Result<(), RecordingError> {
870    if let Some(v) = value
871        && v.trim().is_empty()
872    {
873        return Err(RecordingError::EmptyPayloadField { channel, field });
874    }
875
876    Ok(())
877}
878
879const fn validate_channel_mask<const CHANNEL_MASK: u8>() -> Result<(), RecordingError> {
880    if CHANNEL_MASK == 0 {
881        return Err(RecordingError::EmptyChannelMask);
882    }
883
884    if CHANNEL_MASK & !CHANNEL_MASK_ALL != 0 {
885        return Err(RecordingError::UnsupportedChannelMask { mask: CHANNEL_MASK });
886    }
887
888    Ok(())
889}
890
891fn validate_channel_payload(
892    selected_mask: u8,
893    channel_mask: u8,
894    has_payload: bool,
895    channel_name: &'static str,
896) -> Result<(), RecordingError> {
897    let selected = selected_mask & channel_mask != 0;
898
899    if selected && !has_payload {
900        return Err(RecordingError::MissingSelectedChannelPayload {
901            channel: channel_name,
902        });
903    }
904
905    if !selected && has_payload {
906        return Err(RecordingError::UnexpectedChannelPayload {
907            channel: channel_name,
908        });
909    }
910
911    Ok(())
912}
913
914fn build_channel_record<P: Serialize>(
915    channel_name: &'static str,
916    record_id: &RecordId,
917    kind: &str,
918    common: &Common,
919    payload: P,
920    attributes: &Map<String, Value>,
921    callsite: &Callsite,
922) -> Result<Value, RecordingError> {
923    let payload_value =
924        serde_json::to_value(payload).map_err(|error| RecordingError::SerializePayload {
925            channel: channel_name,
926            error,
927        })?;
928
929    Ok(json!({
930        "recording_id": record_id,
931        "kind": kind,
932        "channel": channel_name,
933        "timestamp": Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
934        "common": common,
935        "callsite": {
936            "file": callsite.file,
937            "line": callsite.line,
938            "module": callsite.module_path,
939        },
940        "process": {
941            "pid": std::process::id(),
942            "thread": format!("{:?}", std::thread::current().id()),
943        },
944        "attributes": attributes,
945        "payload": payload_value,
946    }))
947}
948
949/// Macro entry-point.
950///
951/// Usage:
952/// `platform::recording::log!(Record::<{ CHANNEL_OBSERVABILITY }> { ... });`
953#[macro_export]
954macro_rules! recording_log {
955    ($record:expr $(,)?) => {{ $crate::recording::emit_with_callsite($record, file!(), line!(), module_path!()) }};
956}
957
958/// Re-export macro as `recording::log!`.
959pub use crate::recording_log as log;
960
961/// Emit a minimal observability record from format-style log arguments.
962///
963/// This is a migration bridge for legacy `info!/warn!/error!/debug!/trace!`
964/// callsites. New code should prefer explicit `recording::log!(Record::<...> { ... })`.
965#[macro_export]
966macro_rules! recording_observability_log {
967    ($level:expr, $($arg:tt)+) => {{
968        let _ = $crate::recording::emit_with_callsite(
969            $crate::recording::Record::<{ $crate::recording::CHANNEL_OBSERVABILITY }> {
970                id: $crate::recording::RecordId::new(),
971                kind: "runtime.log".to_string(),
972                common: $crate::recording::Common {
973                    level: $level,
974                    ..Default::default()
975                },
976                observability: Some($crate::recording::ObservabilityPayload {
977                    summary: Some(format!($($arg)+)),
978                    ..Default::default()
979                }),
980                audit: None,
981                security: None,
982                operations: None,
983                attributes: Default::default(),
984            },
985            file!(),
986            line!(),
987            module_path!(),
988        );
989    }};
990}
991
992#[macro_export]
993macro_rules! recording_trace {
994    ($($arg:tt)+) => {{
995        $crate::recording_observability_log!($crate::recording::RecordLevel::Trace, $($arg)+);
996    }};
997}
998
999#[macro_export]
1000macro_rules! recording_debug {
1001    ($($arg:tt)+) => {{
1002        $crate::recording_observability_log!($crate::recording::RecordLevel::Debug, $($arg)+);
1003    }};
1004}
1005
1006#[macro_export]
1007macro_rules! recording_info {
1008    ($($arg:tt)+) => {{
1009        $crate::recording_observability_log!($crate::recording::RecordLevel::Info, $($arg)+);
1010    }};
1011}
1012
1013#[macro_export]
1014macro_rules! recording_warn {
1015    ($($arg:tt)+) => {{
1016        $crate::recording_observability_log!($crate::recording::RecordLevel::Warn, $($arg)+);
1017    }};
1018}
1019
1020#[macro_export]
1021macro_rules! recording_error {
1022    ($($arg:tt)+) => {{
1023        $crate::recording_observability_log!($crate::recording::RecordLevel::Error, $($arg)+);
1024    }};
1025}
1026
1027/// Re-export migration bridge macros under `recording::*`.
1028pub use crate::recording_debug as debug;
1029pub use crate::recording_error as error;
1030pub use crate::recording_info as info;
1031pub use crate::recording_trace as trace;
1032pub use crate::recording_warn as warn;
1033
1034#[cfg(test)]
1035mod tests {
1036    use super::*;
1037
1038    fn base_record<const CHANNEL_MASK: u8>() -> Record<CHANNEL_MASK> {
1039        Record::new("test.recording")
1040    }
1041
1042    fn make_callsite(module_path: &'static str) -> Callsite {
1043        Callsite {
1044            file: "test.rs",
1045            line: 42,
1046            module_path,
1047        }
1048    }
1049
1050    #[test]
1051    fn enrich_common_should_preserve_explicit_fields() {
1052        let common = Common {
1053            actor: Some("actor:1".to_string()),
1054            outcome: Outcome::Success,
1055            level: RecordLevel::Info,
1056            trace_id: Some("trace-explicit".to_string()),
1057            span_id: Some("span-explicit".to_string()),
1058            request_id: Some("request-1".to_string()),
1059            source_service: Some("service-explicit".to_string()),
1060            source_node: Some("node-explicit".to_string()),
1061        };
1062
1063        let enriched = enrich_common(common, &make_callsite("actrix::service::manager"));
1064
1065        assert_eq!(enriched.actor.as_deref(), Some("actor:1"));
1066        assert_eq!(enriched.trace_id.as_deref(), Some("trace-explicit"));
1067        assert_eq!(enriched.span_id.as_deref(), Some("span-explicit"));
1068        assert_eq!(enriched.request_id.as_deref(), Some("request-1"));
1069        assert_eq!(enriched.source_service.as_deref(), Some("service-explicit"));
1070        assert_eq!(enriched.source_node.as_deref(), Some("node-explicit"));
1071    }
1072
1073    #[test]
1074    fn enrich_common_should_fill_source_service_from_module_path() {
1075        let common = Common::default();
1076        let enriched = enrich_common(common, &make_callsite("actrix::service::manager"));
1077
1078        assert_eq!(enriched.source_service.as_deref(), Some("actrix"));
1079    }
1080
1081    #[test]
1082    fn enrich_common_should_treat_blank_explicit_values_as_missing() {
1083        let common = Common {
1084            source_service: Some("  ".to_string()),
1085            request_id: Some("   ".to_string()),
1086            ..Default::default()
1087        };
1088
1089        let enriched = enrich_common(common, &make_callsite("actrix::service::manager"));
1090
1091        assert_eq!(enriched.source_service.as_deref(), Some("actrix"));
1092        assert!(enriched.request_id.is_none());
1093    }
1094
1095    #[test]
1096    fn emit_observability_record_should_succeed() {
1097        let mut record = base_record::<{ CHANNEL_OBSERVABILITY }>();
1098        record.observability = Some(ObservabilityPayload {
1099            summary: Some("ready".to_string()),
1100            component: Some("health.check".to_string()),
1101            operation: Some("http.health".to_string()),
1102            protocol: Some(Protocol::Http),
1103            route: Some("/health".to_string()),
1104            method: Some("GET".to_string()),
1105            ..Default::default()
1106        });
1107
1108        assert!(emit(record).is_ok());
1109    }
1110
1111    #[test]
1112    fn emit_should_fail_when_selected_channel_payload_is_missing() {
1113        let record = base_record::<{ CHANNEL_AUDIT }>();
1114        let error = emit(record).expect_err("audit payload should be required");
1115
1116        assert!(matches!(
1117            error,
1118            RecordingError::MissingSelectedChannelPayload { channel: "audit" }
1119        ));
1120    }
1121
1122    #[test]
1123    fn emit_should_fail_when_unselected_channel_payload_is_present() {
1124        let mut record = base_record::<{ CHANNEL_OBSERVABILITY }>();
1125        record.observability = Some(ObservabilityPayload {
1126            summary: Some("http route mounted".to_string()),
1127            ..Default::default()
1128        });
1129        record.audit = Some(AuditPayload {
1130            action: "config.update".to_string(),
1131            resource: "bind.http.port".to_string(),
1132            resource_id: None,
1133            reason: None,
1134            remote_addr: None,
1135            session_id: None,
1136            before: Some(json!("8080")),
1137            after: Some(json!("8443")),
1138        });
1139
1140        let error = emit(record).expect_err("audit payload should not be accepted");
1141
1142        assert!(matches!(
1143            error,
1144            RecordingError::UnexpectedChannelPayload { channel: "audit" }
1145        ));
1146    }
1147
1148    #[test]
1149    fn emit_should_fail_on_empty_channel_mask() {
1150        let record = base_record::<0>();
1151        let error = emit(record).expect_err("empty mask should fail");
1152
1153        assert!(matches!(error, RecordingError::EmptyChannelMask));
1154    }
1155
1156    #[test]
1157    fn emit_should_fail_when_required_payload_field_is_empty() {
1158        let mut record = base_record::<{ CHANNEL_AUDIT }>();
1159        record.audit = Some(AuditPayload {
1160            action: String::new(),
1161            resource: "service.binding".to_string(),
1162            resource_id: None,
1163            reason: None,
1164            remote_addr: None,
1165            session_id: None,
1166            before: None,
1167            after: None,
1168        });
1169
1170        let error = emit(record).expect_err("empty action should fail validation");
1171
1172        assert!(matches!(
1173            error,
1174            RecordingError::EmptyPayloadField {
1175                channel: "audit",
1176                field: "action"
1177            }
1178        ));
1179    }
1180
1181    #[test]
1182    fn emit_should_fail_when_optional_payload_field_is_blank() {
1183        let mut record = base_record::<{ CHANNEL_SECURITY }>();
1184        record.security = Some(SecurityPayload {
1185            control: "rate_limit.connection".to_string(),
1186            severity: SecuritySeverity::High,
1187            category: Some(String::new()),
1188            subject: None,
1189            source_addr: None,
1190            destination_addr: None,
1191            evidence: None,
1192        });
1193
1194        let error = emit(record).expect_err("blank optional field should fail validation");
1195
1196        assert!(matches!(
1197            error,
1198            RecordingError::EmptyPayloadField {
1199                channel: "security",
1200                field: "category"
1201            }
1202        ));
1203    }
1204
1205    #[test]
1206    fn emit_should_fail_when_observability_payload_is_empty() {
1207        let mut record = base_record::<{ CHANNEL_OBSERVABILITY }>();
1208        record.observability = Some(ObservabilityPayload::default());
1209
1210        let error = emit(record).expect_err("empty observability payload should fail validation");
1211
1212        assert!(matches!(
1213            error,
1214            RecordingError::EmptyPayload {
1215                channel: "observability"
1216            }
1217        ));
1218    }
1219
1220    // ── Filter parse tests ──
1221
1222    #[test]
1223    fn parse_observability_filter_values() {
1224        assert_eq!(parse_observability_filter("off"), ObservabilityFilter::Off);
1225        assert_eq!(
1226            parse_observability_filter("digest"),
1227            ObservabilityFilter::Digest
1228        );
1229        assert_eq!(
1230            parse_observability_filter("detailed"),
1231            ObservabilityFilter::Detailed
1232        );
1233        assert_eq!(
1234            parse_observability_filter("full"),
1235            ObservabilityFilter::Full
1236        );
1237        assert_eq!(
1238            parse_observability_filter("bogus"),
1239            ObservabilityFilter::Digest
1240        );
1241    }
1242
1243    #[test]
1244    fn parse_audit_filter_values() {
1245        assert_eq!(parse_audit_filter("off"), AuditFilter::Off);
1246        assert_eq!(parse_audit_filter("mutations"), AuditFilter::Mutations);
1247        assert_eq!(parse_audit_filter("all"), AuditFilter::All);
1248        assert_eq!(parse_audit_filter("bogus"), AuditFilter::Mutations);
1249    }
1250
1251    #[test]
1252    fn parse_security_filter_values() {
1253        assert_eq!(parse_security_filter("off"), SecurityFilter::Off);
1254        assert_eq!(parse_security_filter("critical"), SecurityFilter::Critical);
1255        assert_eq!(parse_security_filter("high"), SecurityFilter::High);
1256        assert_eq!(parse_security_filter("medium"), SecurityFilter::Medium);
1257        assert_eq!(parse_security_filter("all"), SecurityFilter::All);
1258        assert_eq!(parse_security_filter("bogus"), SecurityFilter::All);
1259    }
1260
1261    #[test]
1262    fn parse_operations_filter_values() {
1263        assert_eq!(parse_operations_filter("off"), OperationsFilter::Off);
1264        assert_eq!(
1265            parse_operations_filter("lifecycle"),
1266            OperationsFilter::Lifecycle
1267        );
1268        assert_eq!(
1269            parse_operations_filter("detailed"),
1270            OperationsFilter::Detailed
1271        );
1272        assert_eq!(
1273            parse_operations_filter("bogus"),
1274            OperationsFilter::Lifecycle
1275        );
1276    }
1277
1278    // ── Gate logic tests ──
1279
1280    #[test]
1281    fn observability_gate_digest_passes_info_blocks_debug() {
1282        let info = Common {
1283            level: RecordLevel::Info,
1284            ..Default::default()
1285        };
1286        let debug = Common {
1287            level: RecordLevel::Debug,
1288            ..Default::default()
1289        };
1290        let trace = Common {
1291            level: RecordLevel::Trace,
1292            ..Default::default()
1293        };
1294        assert!(should_emit_observability(
1295            &info,
1296            ObservabilityFilter::Digest
1297        ));
1298        assert!(!should_emit_observability(
1299            &debug,
1300            ObservabilityFilter::Digest
1301        ));
1302        assert!(!should_emit_observability(
1303            &trace,
1304            ObservabilityFilter::Digest
1305        ));
1306    }
1307
1308    #[test]
1309    fn observability_gate_detailed_passes_debug_blocks_trace() {
1310        let debug = Common {
1311            level: RecordLevel::Debug,
1312            ..Default::default()
1313        };
1314        let trace = Common {
1315            level: RecordLevel::Trace,
1316            ..Default::default()
1317        };
1318        assert!(should_emit_observability(
1319            &debug,
1320            ObservabilityFilter::Detailed
1321        ));
1322        assert!(!should_emit_observability(
1323            &trace,
1324            ObservabilityFilter::Detailed
1325        ));
1326    }
1327
1328    #[test]
1329    fn observability_gate_full_passes_all() {
1330        let trace = Common {
1331            level: RecordLevel::Trace,
1332            ..Default::default()
1333        };
1334        assert!(should_emit_observability(&trace, ObservabilityFilter::Full));
1335    }
1336
1337    #[test]
1338    fn observability_gate_off_blocks_all() {
1339        let error = Common {
1340            level: RecordLevel::Error,
1341            ..Default::default()
1342        };
1343        assert!(!should_emit_observability(&error, ObservabilityFilter::Off));
1344    }
1345
1346    #[test]
1347    fn audit_gate_mutations_filters_reads() {
1348        let write_payload = AuditPayload {
1349            action: "config.update".to_string(),
1350            resource: "bind".to_string(),
1351            resource_id: None,
1352            reason: None,
1353            remote_addr: None,
1354            session_id: None,
1355            before: None,
1356            after: None,
1357        };
1358        let read_payload = AuditPayload {
1359            action: "read.config".to_string(),
1360            resource: "bind".to_string(),
1361            resource_id: None,
1362            reason: None,
1363            remote_addr: None,
1364            session_id: None,
1365            before: None,
1366            after: None,
1367        };
1368        let list_payload = AuditPayload {
1369            action: "list.realms".to_string(),
1370            resource: "realm".to_string(),
1371            resource_id: None,
1372            reason: None,
1373            remote_addr: None,
1374            session_id: None,
1375            before: None,
1376            after: None,
1377        };
1378        assert!(should_emit_audit(&write_payload, AuditFilter::Mutations));
1379        assert!(!should_emit_audit(&read_payload, AuditFilter::Mutations));
1380        assert!(!should_emit_audit(&list_payload, AuditFilter::Mutations));
1381        assert!(should_emit_audit(&read_payload, AuditFilter::All));
1382    }
1383
1384    #[test]
1385    fn security_gate_severity_threshold() {
1386        let low = SecurityPayload {
1387            control: "test".to_string(),
1388            severity: SecuritySeverity::Low,
1389            category: None,
1390            subject: None,
1391            source_addr: None,
1392            destination_addr: None,
1393            evidence: None,
1394        };
1395        let high = SecurityPayload {
1396            control: "test".to_string(),
1397            severity: SecuritySeverity::High,
1398            category: None,
1399            subject: None,
1400            source_addr: None,
1401            destination_addr: None,
1402            evidence: None,
1403        };
1404        let critical = SecurityPayload {
1405            control: "test".to_string(),
1406            severity: SecuritySeverity::Critical,
1407            category: None,
1408            subject: None,
1409            source_addr: None,
1410            destination_addr: None,
1411            evidence: None,
1412        };
1413        assert!(!should_emit_security(&low, SecurityFilter::High));
1414        assert!(should_emit_security(&high, SecurityFilter::High));
1415        assert!(should_emit_security(&critical, SecurityFilter::High));
1416        assert!(should_emit_security(&critical, SecurityFilter::Critical));
1417        assert!(!should_emit_security(&high, SecurityFilter::Critical));
1418        assert!(should_emit_security(&low, SecurityFilter::All));
1419    }
1420
1421    #[test]
1422    fn operations_gate_lifecycle_passes_info() {
1423        let info = Common {
1424            level: RecordLevel::Info,
1425            ..Default::default()
1426        };
1427        let debug = Common {
1428            level: RecordLevel::Debug,
1429            ..Default::default()
1430        };
1431        assert!(should_emit_operations(&info, OperationsFilter::Lifecycle));
1432        assert!(!should_emit_operations(&debug, OperationsFilter::Lifecycle));
1433        assert!(should_emit_operations(&debug, OperationsFilter::Detailed));
1434    }
1435}
platform/recording.rs

platform/
recording.rs
