gwk_domain/
envelope.rs

1//! Event and command envelopes — the two records every party agrees on.
2//!
3//! The envelope is the unit of append and of transport. Payloads above the
4//! inline bound live outside the log as content-addressed blobs referenced by
5//! [`PayloadRef`]; the inline `payload` field carries bounded JSON metadata only.
6//!
7//! Wire discipline (see `docs/contract/NAMING.md`): every field snake_case, all
8//! optional fields omitted when absent, every 64-bit counter a decimal string.
9
10use crate::ids::{
11    AggregateId, ByteCount, CommandId, CorrelationId, EventId, IdempotencyKey, ProjectId, Seq,
12    Timestamp,
13};
14
15/// The envelope schema version this crate reads and writes.
16pub const ENVELOPE_SCHEMA_VERSION: u32 = 1;
17
18/// Inline `payload` bound, in serialized bytes. Anything larger belongs in a
19/// content-addressed blob behind [`PayloadRef`]. Enforced by the kernel at
20/// append and re-checked by `gwk-cert`.
21pub const INLINE_PAYLOAD_MAX_BYTES: usize = 64 * 1024;
22
23/// The EXPORT-side shape of JSON metadata payloads. At runtime payloads are
24/// `serde_json::Value` (exact round-trips, arbitrary integer precision); this
25/// type exists so the generated TypeScript says "JSON tree", not `any` — and
26/// it encodes the discipline that a payload NUMBER is IEEE-754 double
27/// territory: anything 64-bit crosses the wire as a decimal string.
28#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize, specta::Type)]
29#[serde(untagged)]
30pub enum JsonValue {
31    Null(()),
32    Bool(bool),
33    Number(f64),
34    String(String),
35    Array(Vec<JsonValue>),
36    Object(std::collections::BTreeMap<String, JsonValue>),
37}
38
39/// Who caused an event or issued a command. `kind` is an OPEN bounded string
40/// (e.g. `operator`, `kernel`, `engine`, `liveness_producer`).
41#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, specta::Type)]
42#[serde(deny_unknown_fields)]
43pub struct Actor {
44    pub kind: String,
45    #[serde(default, skip_serializing_if = "Option::is_none")]
46    #[specta(optional)]
47    pub id: Option<String>,
48}
49
50/// The producing system, opaque to the contract (e.g. a kernel node name, an
51/// adapter). `ref` optionally pins a system-local origin detail.
52#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, specta::Type)]
53#[serde(deny_unknown_fields)]
54pub struct Origin {
55    pub system: String,
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    #[specta(optional)]
58    pub r#ref: Option<String>,
59}
60
61/// Reference to an out-of-line, content-addressed payload blob.
62#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, specta::Type)]
63#[serde(deny_unknown_fields)]
64pub struct PayloadRef {
65    /// Content address, `<scheme>:<hex>` (e.g. `sha256:…`). The scheme prefix is
66    /// part of the address; consumers must not assume a single scheme forever.
67    pub digest: String,
68    pub media_type: String,
69    pub byte_size: ByteCount,
70    #[serde(default, skip_serializing_if = "Option::is_none")]
71    #[specta(optional)]
72    pub retention_class: Option<String>,
73    /// Pinned as evidence: excluded from retention sweeps until unpinned.
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    #[specta(optional)]
76    pub evidence_pin: Option<bool>,
77}
78
79/// One appended event. Immutable once written; `global_sequence` is assigned by
80/// the kernel append actor in commit order.
81///
82/// The inline-payload byte bound ([`INLINE_PAYLOAD_MAX_BYTES`]) is enforced by
83/// the certifier; consuming kernels must re-check it — this type itself is not
84/// size-bounded.
85#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize, specta::Type)]
86#[serde(deny_unknown_fields)]
87pub struct EventEnvelope {
88    pub event_id: EventId,
89    pub project_id: ProjectId,
90    /// OPEN bounded string naming the aggregate family (e.g. `task`, `attempt`).
91    pub aggregate_type: String,
92    pub aggregate_id: AggregateId,
93    /// Per-aggregate version this event produced: contiguous from 1.
94    pub aggregate_version: u32,
95    /// OPEN bounded string naming the event (e.g. `task_state_changed`).
96    pub event_type: String,
97    pub schema_version: u32,
98    pub global_sequence: Seq,
99    /// When the change happened in the producing system.
100    pub occurred_at: Timestamp,
101    /// When the kernel append actor committed it (assigned, never client-set).
102    pub appended_at: Timestamp,
103    pub actor: Actor,
104    pub origin: Origin,
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    #[specta(optional)]
107    pub causation_id: Option<EventId>,
108    #[serde(default, skip_serializing_if = "Option::is_none")]
109    #[specta(optional)]
110    pub correlation_id: Option<CorrelationId>,
111    #[serde(default, skip_serializing_if = "Option::is_none")]
112    #[specta(optional)]
113    pub idempotency_key: Option<IdempotencyKey>,
114    /// Bounded inline JSON metadata (≤ [`INLINE_PAYLOAD_MAX_BYTES`] serialized).
115    #[specta(type = JsonValue)]
116    pub payload: serde_json::Value,
117    #[serde(default, skip_serializing_if = "Option::is_none")]
118    #[specta(optional)]
119    pub payload_ref: Option<PayloadRef>,
120}
121
122/// One issued command. Commands are requests — only the events they cause are
123/// truth. `idempotency_key` is required: reissuing the same key is a no-op.
124#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize, specta::Type)]
125#[serde(deny_unknown_fields)]
126pub struct CommandEnvelope {
127    pub command_id: CommandId,
128    pub project_id: ProjectId,
129    /// OPEN bounded string naming the command (e.g. `cancel_attempt`).
130    pub command_type: String,
131    pub schema_version: u32,
132    pub issued_at: Timestamp,
133    pub actor: Actor,
134    pub origin: Origin,
135    #[serde(default, skip_serializing_if = "Option::is_none")]
136    #[specta(optional)]
137    pub target_aggregate_type: Option<String>,
138    #[serde(default, skip_serializing_if = "Option::is_none")]
139    #[specta(optional)]
140    pub target_aggregate_id: Option<AggregateId>,
141    /// CAS precondition: the target aggregate's current version.
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    #[specta(optional)]
144    pub expected_version: Option<u32>,
145    pub idempotency_key: IdempotencyKey,
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    #[specta(optional)]
148    pub causation_id: Option<EventId>,
149    #[serde(default, skip_serializing_if = "Option::is_none")]
150    #[specta(optional)]
151    pub correlation_id: Option<CorrelationId>,
152    #[specta(type = JsonValue)]
153    pub payload: serde_json::Value,
154}
155
156/// A consumer met a `schema_version` it cannot read and has no upcaster for.
157#[derive(Debug, Clone, Copy, PartialEq, Eq)]
158pub struct UnknownSchemaVersion {
159    pub found: u32,
160    pub supported: u32,
161}
162
163impl std::fmt::Display for UnknownSchemaVersion {
164    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
165        write!(
166            f,
167            "unknown schema_version {} (supported: {}, no upcaster registered)",
168            self.found, self.supported
169        )
170    }
171}
172
173impl std::error::Error for UnknownSchemaVersion {}
174
175/// Accept `found` iff it is the supported version or an upcaster covers it.
176/// Unknown versions are a typed error, never a silent best-effort read.
177pub fn accept_schema_version(
178    found: u32,
179    supported: u32,
180    upcast_from: &[u32],
181) -> Result<(), UnknownSchemaVersion> {
182    if found == supported || upcast_from.contains(&found) {
183        Ok(())
184    } else {
185        Err(UnknownSchemaVersion { found, supported })
186    }
187}
188
189#[cfg(test)]
190mod tests {
191    use super::*;
192
193    fn fixture() -> EventEnvelope {
194        EventEnvelope {
195            event_id: EventId::new("evt-1"),
196            project_id: ProjectId::new("proj-1"),
197            aggregate_type: "task".into(),
198            aggregate_id: AggregateId::new("task-1"),
199            aggregate_version: 1,
200            event_type: "task_state_changed".into(),
201            schema_version: ENVELOPE_SCHEMA_VERSION,
202            global_sequence: Seq::new(42),
203            occurred_at: Timestamp::new("2026-07-27T00:00:00Z"),
204            appended_at: Timestamp::new("2026-07-27T00:00:01Z"),
205            actor: Actor {
206                kind: "kernel".into(),
207                id: None,
208            },
209            origin: Origin {
210                system: "kernel".into(),
211                r#ref: None,
212            },
213            causation_id: None,
214            correlation_id: None,
215            idempotency_key: None,
216            payload: serde_json::json!({ "to": "working" }),
217            payload_ref: None,
218        }
219    }
220
221    #[test]
222    fn envelope_round_trips_and_omits_absent_optionals() {
223        let env = fixture();
224        let json = serde_json::to_string(&env).expect("serialize");
225        // Absent options are OMITTED, not null — the tri-state discipline.
226        assert!(!json.contains("causation_id"));
227        assert!(!json.contains("payload_ref"));
228        // `r#ref` serializes as plain `ref`; Seq as a decimal string.
229        assert!(json.contains("\"system\":\"kernel\""));
230        assert!(json.contains("\"global_sequence\":\"42\""));
231        let back: EventEnvelope = serde_json::from_str(&json).expect("deserialize");
232        assert_eq!(back, env);
233    }
234
235    #[test]
236    fn envelope_rejects_unknown_fields() {
237        let mut value = serde_json::to_value(fixture()).expect("to_value");
238        value["surprise"] = serde_json::json!(true);
239        assert!(serde_json::from_value::<EventEnvelope>(value).is_err());
240    }
241
242    #[test]
243    fn schema_version_gate_is_typed() {
244        assert!(accept_schema_version(1, 1, &[]).is_ok());
245        assert!(accept_schema_version(0, 1, &[0]).is_ok());
246        assert_eq!(
247            accept_schema_version(9, 1, &[0]),
248            Err(UnknownSchemaVersion {
249                found: 9,
250                supported: 1
251            })
252        );
253    }
254}
gwk_domain/envelope.rs

gwk_domain/
envelope.rs
