1use crate::{
2 AutonomousServiceContract, CausationContext, CommonContextContract, CommonContextIssueCode,
3 CommonContextRequirement, CompatibilityCategory, CompatibilityReason,
4 ContractCompatibilityKind, ContractCompatibilityResult, ContractContextRequirements,
5 DeadlineContext, DelegatedActorContext, EventArtifactFormat, EventArtifactReference,
6 EventContractArtifact, IdempotencyKeyContext, RegionContext, ServicePrincipal,
7 ServiceTenancyMode, StoryContext, TenantContext, TraceContext, evaluate_event_compatibility,
8 validate_common_context_contract_value,
9};
10use chrono::DateTime;
11use serde::{Deserialize, Serialize};
12use serde_json::{Value, json};
13use std::collections::BTreeSet;
14
15pub const EVENT_CONTRACT_ARTIFACT_PROTOCOL: &str = "lenso.event-contract.v1";
16pub const EVENT_ENVELOPE_PROTOCOL: &str = "lenso.event-envelope.v1";
17const CLOUDEVENTS_SPEC_VERSION: &str = "1.0";
18const EVENT_ENVELOPE_SCHEMA: &str =
19 "https://lenso.dev/contracts/lenso.event-envelope.v1.schema.json";
20
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22#[serde(rename_all = "camelCase", deny_unknown_fields)]
23pub struct GeneratedEventContract {
24 pub protocol: String,
25 pub event_type: String,
26 pub contract_id: String,
27 pub contract_version: String,
28 pub producer_service_id: String,
29 pub module_id: String,
30 pub operating_regions: Vec<String>,
31 pub tenancy_mode: ServiceTenancyMode,
32 pub context: ContractContextRequirements,
33 pub artifact: EventArtifactReference,
34 pub payload_schema: Value,
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38#[serde(rename_all = "snake_case")]
39pub enum EventContractGenerationError {
40 EmptyProducerServiceId,
41 InvalidDeclaration,
42 UnownedModule,
43 UnsupportedArtifactFormat,
44 InvalidArtifactReference,
45 InvalidPayloadSchema,
46}
47
48#[must_use]
49pub fn generate_event_contract(
50 service: &AutonomousServiceContract,
51 declaration: &EventContractArtifact,
52 payload_schema: &Value,
53) -> Result<GeneratedEventContract, EventContractGenerationError> {
54 if service.service_id.trim().is_empty() {
55 return Err(EventContractGenerationError::EmptyProducerServiceId);
56 }
57 if !service.modules.contains(&declaration.module_id) {
58 return Err(EventContractGenerationError::UnownedModule);
59 }
60 if !service.event_contracts.contains(declaration)
61 || !crate::validate_autonomous_service_contract(service).is_empty()
62 || declaration.contract_id.trim().is_empty()
63 || declaration.module_id.trim().is_empty()
64 || declaration.version.trim().is_empty()
65 || declaration.context.protocol != crate::COMMON_CONTEXT_PROTOCOL
66 || (declaration.tenancy_mode == ServiceTenancyMode::Required
67 && !declaration
68 .context
69 .required
70 .contains(&CommonContextRequirement::Tenant))
71 {
72 return Err(EventContractGenerationError::InvalidDeclaration);
73 }
74 if declaration.artifact.format != EventArtifactFormat::JsonSchema {
75 return Err(EventContractGenerationError::UnsupportedArtifactFormat);
76 }
77 let event_type = declaration
78 .artifact
79 .path
80 .rsplit('/')
81 .next()
82 .and_then(|name| name.strip_suffix(".schema.json"))
83 .filter(|name| !name.is_empty())
84 .ok_or(EventContractGenerationError::InvalidArtifactReference)?;
85 if !event_type.ends_with(&format!(
86 "{}.{}",
87 declaration.contract_id, declaration.version
88 )) {
89 return Err(EventContractGenerationError::InvalidArtifactReference);
90 }
91 if !valid_payload_schema_definition(payload_schema, event_type) {
92 return Err(EventContractGenerationError::InvalidPayloadSchema);
93 }
94
95 Ok(GeneratedEventContract {
96 protocol: EVENT_CONTRACT_ARTIFACT_PROTOCOL.to_owned(),
97 event_type: event_type.to_owned(),
98 contract_id: declaration.contract_id.clone(),
99 contract_version: declaration.version.clone(),
100 producer_service_id: service.service_id.clone(),
101 module_id: declaration.module_id.clone(),
102 operating_regions: service.operating_regions.clone(),
103 tenancy_mode: declaration.tenancy_mode.clone(),
104 context: declaration.context.clone(),
105 artifact: declaration.artifact.clone(),
106 payload_schema: payload_schema.clone(),
107 })
108}
109
110#[must_use]
111pub fn evaluate_generated_event_contract_compatibility(
112 before: &GeneratedEventContract,
113 after: &GeneratedEventContract,
114) -> ContractCompatibilityResult {
115 let mut result = ContractCompatibilityResult {
116 category: CompatibilityCategory::Safe,
117 contract_kind: ContractCompatibilityKind::EventContract,
118 contract_id: after.contract_id.clone(),
119 changed_version: after.contract_version.clone(),
120 affected_references: vec![
121 format!("autonomous_service:{}", after.producer_service_id),
122 format!("module:{}", after.module_id),
123 ],
124 reasons: Vec::new(),
125 };
126 if !valid_generated_event_contract(before) || !valid_generated_event_contract(after) {
127 compatibility_issue(
128 &mut result,
129 CompatibilityCategory::Blocked,
130 "event_artifact_unverifiable",
131 "$",
132 "Generated Event Contract artifacts must use supported protocols and internally consistent identities.",
133 "Regenerate both Event Contract artifacts from valid Autonomous Service declarations and payload schemas.",
134 );
135 }
136 if before.contract_version == after.contract_version {
137 compatibility_issue(
138 &mut result,
139 CompatibilityCategory::Blocked,
140 "event_version_unverifiable",
141 "$.contractVersion",
142 "Event Contract evolution must identify a changed Contract Version.",
143 "Generate the candidate artifact with a distinct version.",
144 );
145 }
146 for (field, old, new) in [
147 ("contractId", &before.contract_id, &after.contract_id),
148 (
149 "producerServiceId",
150 &before.producer_service_id,
151 &after.producer_service_id,
152 ),
153 ("moduleId", &before.module_id, &after.module_id),
154 ] {
155 if old != new {
156 compatibility_issue(
157 &mut result,
158 CompatibilityCategory::Breaking,
159 &format!("event_{}_changed", super::camel_to_snake(field)),
160 &format!("$.{field}"),
161 "A stable Event Contract identity changed.",
162 "Keep the existing identity or publish a separately coordinated Event Contract.",
163 );
164 }
165 }
166 for (code, path, old, new) in [
167 (
168 "event_type_identity_changed",
169 "$.eventType",
170 event_type_family(before),
171 event_type_family(after),
172 ),
173 (
174 "event_artifact_identity_changed",
175 "$.artifact.path",
176 artifact_family(before),
177 artifact_family(after),
178 ),
179 ] {
180 if old != new {
181 compatibility_issue(
182 &mut result,
183 CompatibilityCategory::Breaking,
184 code,
185 path,
186 "A stable Event Contract artifact identity changed.",
187 "Keep the existing Event Type and artifact family or publish a separately coordinated Event Contract.",
188 );
189 }
190 }
191 if before.tenancy_mode != after.tenancy_mode {
192 let category = if after.tenancy_mode == ServiceTenancyMode::Required {
193 CompatibilityCategory::Breaking
194 } else {
195 CompatibilityCategory::NeedsAttention
196 };
197 compatibility_issue(
198 &mut result,
199 category,
200 "event_tenancy_changed",
201 "$.tenancyMode",
202 "The Event Contract Tenancy Mode changed.",
203 "Review Producer and Consumer tenant scoping before publishing.",
204 );
205 }
206 let old_context = before
207 .context
208 .required
209 .iter()
210 .copied()
211 .collect::<BTreeSet<_>>();
212 let new_context = after
213 .context
214 .required
215 .iter()
216 .copied()
217 .collect::<BTreeSet<_>>();
218 for requirement in new_context.difference(&old_context) {
219 compatibility_issue(
220 &mut result,
221 CompatibilityCategory::Breaking,
222 "event_required_context_added",
223 "$.context.required",
224 &format!("A new required context field was added: {requirement:?}."),
225 "Keep the context optional or coordinate the requirement with every affected Producer and Consumer.",
226 );
227 }
228 for requirement in old_context.difference(&new_context) {
229 compatibility_issue(
230 &mut result,
231 CompatibilityCategory::NeedsAttention,
232 "event_required_context_removed",
233 "$.context.required",
234 &format!("A required context field was removed: {requirement:?}."),
235 "Review identity, tenancy, causation, and evidence semantics with affected owners.",
236 );
237 }
238 let old_regions = before.operating_regions.iter().collect::<BTreeSet<_>>();
239 let new_regions = after.operating_regions.iter().collect::<BTreeSet<_>>();
240 if old_regions != new_regions {
241 compatibility_issue(
242 &mut result,
243 CompatibilityCategory::NeedsAttention,
244 "event_operating_regions_changed",
245 "$.operatingRegions",
246 "The producing Service Operating Regions changed.",
247 "Review regional publication and consumption expectations before publishing.",
248 );
249 }
250 let payload_result = evaluate_event_compatibility(&json!({
251 "contractId": after.contract_id,
252 "changedVersion": after.contract_version,
253 "affectedReferences": result.affected_references,
254 "before": {
255 "format": "json_schema",
256 "version": before.contract_version,
257 "schema": before.payload_schema
258 },
259 "after": {
260 "format": "json_schema",
261 "version": after.contract_version,
262 "schema": after.payload_schema
263 }
264 }));
265 result.category = result.category.max(payload_result.category);
266 result.reasons.extend(payload_result.reasons);
267 if result.category != CompatibilityCategory::Safe {
268 result
269 .reasons
270 .retain(|reason| reason.code != "event_backward_compatible");
271 }
272 result.affected_references.sort();
273 result.affected_references.dedup();
274 result.reasons.sort();
275 result.reasons.dedup();
276 result
277}
278
279fn valid_generated_event_contract(contract: &GeneratedEventContract) -> bool {
280 contract.protocol == EVENT_CONTRACT_ARTIFACT_PROTOCOL
281 && contract.context.protocol == crate::COMMON_CONTEXT_PROTOCOL
282 && contract.artifact.format == EventArtifactFormat::JsonSchema
283 && artifact_event_type(&contract.artifact.path) == Some(contract.event_type.as_str())
284 && contract.event_type.ends_with(&format!(
285 "{}.{}",
286 contract.contract_id, contract.contract_version
287 ))
288 && valid_payload_schema_definition(&contract.payload_schema, &contract.event_type)
289}
290
291fn artifact_event_type(path: &str) -> Option<&str> {
292 path.rsplit('/')
293 .next()
294 .and_then(|name| name.strip_suffix(".schema.json"))
295}
296
297fn event_type_family(contract: &GeneratedEventContract) -> Option<&str> {
298 contract
299 .event_type
300 .strip_suffix(&format!(".{}", contract.contract_version))
301}
302
303fn artifact_family(contract: &GeneratedEventContract) -> Option<&str> {
304 contract
305 .artifact
306 .path
307 .strip_suffix(&format!(".{}.schema.json", contract.contract_version))
308}
309
310fn compatibility_issue(
311 result: &mut ContractCompatibilityResult,
312 category: CompatibilityCategory,
313 code: &str,
314 path: &str,
315 message: &str,
316 next_action: &str,
317) {
318 result.category = result.category.max(category);
319 result.reasons.push(CompatibilityReason {
320 code: code.to_owned(),
321 path: path.to_owned(),
322 message: message.to_owned(),
323 next_action: next_action.to_owned(),
324 });
325}
326
327#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
328#[serde(rename_all = "camelCase", deny_unknown_fields)]
329pub struct EventContext {
330 pub protocol: String,
331 #[serde(default, skip_serializing_if = "Option::is_none")]
332 pub story: Option<StoryContext>,
333 #[serde(default, skip_serializing_if = "Option::is_none")]
334 pub trace: Option<TraceContext>,
335 #[serde(default, skip_serializing_if = "Option::is_none")]
336 pub service_principal: Option<ServicePrincipal>,
337 #[serde(default, skip_serializing_if = "Option::is_none")]
338 pub delegated_actor: Option<DelegatedActorContext>,
339 #[serde(default, skip_serializing_if = "Option::is_none")]
340 pub tenant: Option<TenantContext>,
341 #[serde(default, skip_serializing_if = "Option::is_none")]
342 pub deadline: Option<DeadlineContext>,
343 #[serde(default, skip_serializing_if = "Option::is_none")]
344 pub idempotency_key: Option<IdempotencyKeyContext>,
345 #[serde(default, skip_serializing_if = "Option::is_none")]
346 pub causation: Option<CausationContext>,
347 #[serde(default, skip_serializing_if = "Option::is_none")]
348 pub region: Option<RegionContext>,
349}
350
351impl From<CommonContextContract> for EventContext {
352 fn from(context: CommonContextContract) -> Self {
353 Self {
354 protocol: context.protocol,
355 story: Some(context.story),
356 trace: Some(context.trace),
357 service_principal: Some(context.service_principal),
358 delegated_actor: Some(context.delegated_actor),
359 tenant: Some(context.tenant),
360 deadline: Some(context.deadline),
361 idempotency_key: Some(context.idempotency_key),
362 causation: Some(context.causation),
363 region: Some(context.region),
364 }
365 }
366}
367
368#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
369#[serde(rename_all = "camelCase", deny_unknown_fields)]
370pub struct EventContent {
371 pub content_type: String,
372 pub schema: String,
373 pub data: Value,
374}
375
376#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
377#[serde(rename_all = "camelCase", deny_unknown_fields)]
378pub struct EventEnvelope {
379 pub protocol: String,
380 pub event_id: String,
381 pub event_type: String,
382 pub contract_id: String,
383 pub contract_version: String,
384 pub producer_service_id: String,
385 pub module_id: String,
386 pub occurred_at: String,
387 pub tenancy_mode: ServiceTenancyMode,
388 pub context: EventContext,
389 pub content: EventContent,
390}
391
392impl EventEnvelope {
393 #[must_use]
394 pub fn new<C>(
395 contract: &GeneratedEventContract,
396 event_id: impl Into<String>,
397 occurred_at: impl Into<String>,
398 context: C,
399 data: Value,
400 ) -> Self
401 where
402 C: Into<EventContext>,
403 {
404 Self {
405 protocol: EVENT_ENVELOPE_PROTOCOL.to_owned(),
406 event_id: event_id.into(),
407 event_type: contract.event_type.clone(),
408 contract_id: contract.contract_id.clone(),
409 contract_version: contract.contract_version.clone(),
410 producer_service_id: contract.producer_service_id.clone(),
411 module_id: contract.module_id.clone(),
412 occurred_at: occurred_at.into(),
413 tenancy_mode: contract.tenancy_mode.clone(),
414 context: context.into(),
415 content: EventContent {
416 content_type: "application/json".to_owned(),
417 schema: contract.artifact.path.clone(),
418 data,
419 },
420 }
421 }
422
423 #[must_use]
424 pub fn to_cloudevent(&self) -> CloudEvent {
425 CloudEvent {
426 specversion: CLOUDEVENTS_SPEC_VERSION.to_owned(),
427 id: self.event_id.clone(),
428 source: format!("urn:lenso:service:{}", self.producer_service_id),
429 event_type: self.event_type.clone(),
430 subject: format!(
431 "{}/{}/{}",
432 self.module_id, self.contract_id, self.contract_version
433 ),
434 time: self.occurred_at.clone(),
435 datacontenttype: "application/json".to_owned(),
436 dataschema: EVENT_ENVELOPE_SCHEMA.to_owned(),
437 data: serde_json::to_value(self).expect("EventEnvelope must serialize"),
438 }
439 }
440}
441
442#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
443pub struct CloudEvent {
444 pub specversion: String,
445 pub id: String,
446 pub source: String,
447 #[serde(rename = "type")]
448 pub event_type: String,
449 pub subject: String,
450 pub time: String,
451 pub datacontenttype: String,
452 pub dataschema: String,
453 pub data: Value,
454}
455
456#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
457#[serde(rename_all = "snake_case")]
458pub enum EventEnvelopeIssueCode {
459 InvalidProtocol,
460 InvalidEventIdentity,
461 IncompatibleContractIdentity,
462 IncompatibleContext,
463 IncompatibleProducerIdentity,
464 IncompatibleModuleIdentity,
465 IncompatibleRegion,
466 IncompatibleTenancy,
467 InvalidOccurrenceTime,
468 InvalidContentMetadata,
469 InvalidContent,
470 InvalidCloudEventsRepresentation,
471 MissingRequiredContext,
472 MalformedContext,
473 UntrustedContext,
474}
475
476pub fn event_envelope_from_cloudevent(
477 contract: &GeneratedEventContract,
478 cloud_event: &CloudEvent,
479) -> Result<EventEnvelope, Vec<EventEnvelopeIssue>> {
480 let mut issues = Vec::new();
481 for (path, actual, expected) in [
482 (
483 "$.specversion",
484 cloud_event.specversion.as_str(),
485 CLOUDEVENTS_SPEC_VERSION,
486 ),
487 (
488 "$.id",
489 cloud_event.id.as_str(),
490 string_at(&cloud_event.data, "eventId"),
491 ),
492 (
493 "$.source",
494 cloud_event.source.as_str(),
495 &format!("urn:lenso:service:{}", contract.producer_service_id),
496 ),
497 (
498 "$.type",
499 cloud_event.event_type.as_str(),
500 contract.event_type.as_str(),
501 ),
502 (
503 "$.subject",
504 cloud_event.subject.as_str(),
505 &format!(
506 "{}/{}/{}",
507 contract.module_id, contract.contract_id, contract.contract_version
508 ),
509 ),
510 (
511 "$.time",
512 cloud_event.time.as_str(),
513 string_at(&cloud_event.data, "occurredAt"),
514 ),
515 (
516 "$.datacontenttype",
517 cloud_event.datacontenttype.as_str(),
518 "application/json",
519 ),
520 (
521 "$.dataschema",
522 cloud_event.dataschema.as_str(),
523 EVENT_ENVELOPE_SCHEMA,
524 ),
525 ] {
526 if actual != expected {
527 push_issue(
528 &mut issues,
529 EventEnvelopeIssueCode::InvalidCloudEventsRepresentation,
530 path,
531 format!("CloudEvents attribute must match `{expected}`"),
532 "Restore the authoritative Lenso Event Envelope attribute before decoding.",
533 );
534 }
535 }
536 issues.extend(validate_event_envelope_value(contract, &cloud_event.data));
537 if !issues.is_empty() {
538 return Err(issues);
539 }
540 serde_json::from_value(cloud_event.data.clone()).map_err(|error| {
541 vec![EventEnvelopeIssue {
542 code: EventEnvelopeIssueCode::InvalidCloudEventsRepresentation,
543 path: "$.data".to_owned(),
544 message: format!("CloudEvents data is not a Lenso Event Envelope: {error}"),
545 next_action:
546 "Encode the complete validated Lenso Event Envelope as structured CloudEvents data."
547 .to_owned(),
548 }]
549 })
550}
551
552fn string_at<'a>(value: &'a Value, field: &str) -> &'a str {
553 value.get(field).and_then(Value::as_str).unwrap_or_default()
554}
555
556#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
557#[serde(rename_all = "camelCase")]
558pub struct EventEnvelopeIssue {
559 pub code: EventEnvelopeIssueCode,
560 pub path: String,
561 pub message: String,
562 pub next_action: String,
563}
564
565#[must_use]
566pub fn validate_event_envelope(
567 contract: &GeneratedEventContract,
568 envelope: &EventEnvelope,
569) -> Vec<EventEnvelopeIssue> {
570 validate_event_envelope_value(
571 contract,
572 &serde_json::to_value(envelope).expect("EventEnvelope must serialize"),
573 )
574}
575
576#[must_use]
577pub fn validate_event_envelope_value(
578 contract: &GeneratedEventContract,
579 value: &Value,
580) -> Vec<EventEnvelopeIssue> {
581 let mut issues = Vec::new();
582 validate_exact_string(
583 value,
584 "protocol",
585 EVENT_ENVELOPE_PROTOCOL,
586 EventEnvelopeIssueCode::InvalidProtocol,
587 "Use the supported Event Envelope protocol.",
588 &mut issues,
589 );
590 validate_non_empty_string(
591 value,
592 "eventId",
593 EventEnvelopeIssueCode::InvalidEventIdentity,
594 "Assign a stable event identity before publication.",
595 &mut issues,
596 );
597 if value
598 .get("occurredAt")
599 .and_then(Value::as_str)
600 .is_some_and(|time| !time.is_empty() && DateTime::parse_from_rfc3339(time).is_err())
601 {
602 push_issue(
603 &mut issues,
604 EventEnvelopeIssueCode::InvalidOccurrenceTime,
605 "$.occurredAt",
606 "occurredAt must be an RFC 3339 timestamp",
607 "Set the authoritative event occurrence time in RFC 3339 form.",
608 );
609 }
610 for (field, expected, code, action) in [
611 (
612 "eventType",
613 contract.event_type.as_str(),
614 EventEnvelopeIssueCode::IncompatibleContractIdentity,
615 "Regenerate the envelope from the authoritative Event Contract.",
616 ),
617 (
618 "contractId",
619 contract.contract_id.as_str(),
620 EventEnvelopeIssueCode::IncompatibleContractIdentity,
621 "Use the declared Event Contract identity.",
622 ),
623 (
624 "contractVersion",
625 contract.contract_version.as_str(),
626 EventEnvelopeIssueCode::IncompatibleContractIdentity,
627 "Use the declared Event Contract version.",
628 ),
629 (
630 "producerServiceId",
631 contract.producer_service_id.as_str(),
632 EventEnvelopeIssueCode::IncompatibleProducerIdentity,
633 "Use the Service identity that generated the Event Contract artifact.",
634 ),
635 (
636 "moduleId",
637 contract.module_id.as_str(),
638 EventEnvelopeIssueCode::IncompatibleModuleIdentity,
639 "Use the Module identity declared by the Event Contract.",
640 ),
641 (
642 "content/schema",
643 contract.artifact.path.as_str(),
644 EventEnvelopeIssueCode::InvalidContentMetadata,
645 "Use the authoritative payload artifact reference.",
646 ),
647 (
648 "content/contentType",
649 "application/json",
650 EventEnvelopeIssueCode::InvalidContentMetadata,
651 "Use `application/json` for the declared JSON Schema payload.",
652 ),
653 ] {
654 validate_exact_string(value, field, expected, code, action, &mut issues);
655 }
656 validate_non_empty_string(
657 value,
658 "occurredAt",
659 EventEnvelopeIssueCode::InvalidOccurrenceTime,
660 "Set an RFC 3339 occurrence time.",
661 &mut issues,
662 );
663
664 let expected_tenancy =
665 serde_json::to_value(&contract.tenancy_mode).expect("ServiceTenancyMode must serialize");
666 if value.get("tenancyMode") != Some(&expected_tenancy) {
667 push_issue(
668 &mut issues,
669 EventEnvelopeIssueCode::IncompatibleTenancy,
670 "$.tenancyMode",
671 "tenancyMode does not match the Event Contract",
672 "Use the Tenancy Mode declared by the Event Contract.",
673 );
674 }
675
676 validate_payload_value(
677 &contract.payload_schema,
678 value.pointer("/content/data").unwrap_or(&Value::Null),
679 "$.content.data",
680 &mut issues,
681 );
682
683 let context = value.get("context").unwrap_or(&Value::Null);
684 for issue in validate_common_context_contract_value(context) {
685 let requirement = requirement_for_common_issue(issue.code);
686 let required =
687 requirement.is_some_and(|required| contract.context.required.contains(&required));
688 let present = requirement
689 .and_then(context_field_for_requirement)
690 .is_some_and(|field| context.get(field).is_some());
691 let untrusted = matches!(
692 issue.code,
693 CommonContextIssueCode::UntrustedActorClaim
694 | CommonContextIssueCode::UntrustedTenantClaim
695 );
696 let incompatible = matches!(
697 issue.code,
698 CommonContextIssueCode::InvalidProtocol | CommonContextIssueCode::AudienceMismatch
699 );
700 if !required && !present && !untrusted && !incompatible {
701 continue;
702 }
703 let path = format!(
704 "$.context{}",
705 issue.path.strip_prefix('$').unwrap_or(&issue.path)
706 );
707 let missing = common_issue_pointer(&issue.path)
708 .is_none_or(|pointer| context.pointer(&pointer).is_none());
709 let code = if incompatible {
710 EventEnvelopeIssueCode::IncompatibleContext
711 } else if untrusted {
712 EventEnvelopeIssueCode::UntrustedContext
713 } else if missing {
714 EventEnvelopeIssueCode::MissingRequiredContext
715 } else {
716 EventEnvelopeIssueCode::MalformedContext
717 };
718 push_issue(&mut issues, code, path, issue.message, issue.next_action);
719 }
720 if contract
721 .context
722 .required
723 .contains(&CommonContextRequirement::Region)
724 && context
725 .pointer("/region/operatingRegion")
726 .and_then(Value::as_str)
727 .is_some_and(|region| {
728 !contract
729 .operating_regions
730 .iter()
731 .any(|known| known == region)
732 })
733 {
734 push_issue(
735 &mut issues,
736 EventEnvelopeIssueCode::IncompatibleRegion,
737 "$.context.region.operatingRegion",
738 "Operating Region is not declared by the producing Service",
739 "Use an Operating Region declared by the authoritative Autonomous Service contract.",
740 );
741 }
742 issues
743}
744
745fn context_field_for_requirement(requirement: CommonContextRequirement) -> Option<&'static str> {
746 match requirement {
747 CommonContextRequirement::Story => Some("story"),
748 CommonContextRequirement::Trace => Some("trace"),
749 CommonContextRequirement::ServicePrincipal => Some("servicePrincipal"),
750 CommonContextRequirement::DelegatedActor => Some("delegatedActor"),
751 CommonContextRequirement::Tenant => Some("tenant"),
752 CommonContextRequirement::Deadline => Some("deadline"),
753 CommonContextRequirement::IdempotencyKey => Some("idempotencyKey"),
754 CommonContextRequirement::Causation => Some("causation"),
755 CommonContextRequirement::Region => Some("region"),
756 }
757}
758
759fn valid_payload_schema_definition(schema: &Value, event_type: &str) -> bool {
760 schema.get("title").and_then(Value::as_str) == Some(event_type)
761 && jsonschema::draft202012::meta::validate(schema).is_ok()
762 && jsonschema::draft202012::options()
763 .should_validate_formats(true)
764 .build(schema)
765 .is_ok()
766}
767
768fn validate_payload_value(
769 schema: &Value,
770 data: &Value,
771 path: &str,
772 issues: &mut Vec<EventEnvelopeIssue>,
773) {
774 let Ok(validator) = jsonschema::draft202012::options()
775 .should_validate_formats(true)
776 .build(schema)
777 else {
778 push_issue(
779 issues,
780 EventEnvelopeIssueCode::InvalidContent,
781 path,
782 "authoritative Event Contract payload schema is invalid",
783 "Regenerate or replace the Event Contract artifact before validating event content.",
784 );
785 return;
786 };
787 let mut errors = validator
788 .iter_errors(data)
789 .map(|error| {
790 let suffix = error
791 .instance_path()
792 .to_string()
793 .split('/')
794 .filter(|segment| !segment.is_empty())
795 .map(|segment| format!(".{segment}"))
796 .collect::<String>();
797 (format!("{path}{suffix}"), error.to_string())
798 })
799 .collect::<Vec<_>>();
800 errors.sort();
801 errors.dedup();
802 for (error_path, message) in errors {
803 push_issue(
804 issues,
805 EventEnvelopeIssueCode::InvalidContent,
806 error_path,
807 message,
808 "Provide content that satisfies the authoritative generated payload schema.",
809 );
810 }
811}
812
813fn requirement_for_common_issue(code: CommonContextIssueCode) -> Option<CommonContextRequirement> {
814 match code {
815 CommonContextIssueCode::InvalidStoryContext => Some(CommonContextRequirement::Story),
816 CommonContextIssueCode::InvalidTraceContext => Some(CommonContextRequirement::Trace),
817 CommonContextIssueCode::InvalidServicePrincipal => {
818 Some(CommonContextRequirement::ServicePrincipal)
819 }
820 CommonContextIssueCode::InvalidDelegatedActorContext => {
821 Some(CommonContextRequirement::DelegatedActor)
822 }
823 CommonContextIssueCode::InvalidTenantContext => Some(CommonContextRequirement::Tenant),
824 CommonContextIssueCode::InvalidDeadline => Some(CommonContextRequirement::Deadline),
825 CommonContextIssueCode::InvalidIdempotencyKey => {
826 Some(CommonContextRequirement::IdempotencyKey)
827 }
828 CommonContextIssueCode::InvalidCausation => Some(CommonContextRequirement::Causation),
829 CommonContextIssueCode::InvalidRegion => Some(CommonContextRequirement::Region),
830 CommonContextIssueCode::InvalidProtocol
831 | CommonContextIssueCode::UntrustedActorClaim
832 | CommonContextIssueCode::UntrustedTenantClaim
833 | CommonContextIssueCode::AudienceMismatch => None,
834 }
835}
836
837fn common_issue_pointer(path: &str) -> Option<String> {
838 path.strip_prefix("$.")
839 .map(|suffix| format!("/{}", suffix.replace('.', "/")))
840}
841
842fn validate_exact_string(
843 value: &Value,
844 field: &str,
845 expected: &str,
846 code: EventEnvelopeIssueCode,
847 next_action: &str,
848 issues: &mut Vec<EventEnvelopeIssue>,
849) {
850 let pointer = format!("/{}", field.replace('/', "/"));
851 if value.pointer(&pointer).and_then(Value::as_str) != Some(expected) {
852 push_issue(
853 issues,
854 code,
855 format!("$.{}", field.replace('/', ".")),
856 format!("value must match `{expected}`"),
857 next_action,
858 );
859 }
860}
861
862fn validate_non_empty_string(
863 value: &Value,
864 field: &str,
865 code: EventEnvelopeIssueCode,
866 next_action: &str,
867 issues: &mut Vec<EventEnvelopeIssue>,
868) {
869 if value
870 .get(field)
871 .and_then(Value::as_str)
872 .is_none_or(|text| text.trim().is_empty())
873 {
874 push_issue(
875 issues,
876 code,
877 format!("$.{field}"),
878 format!("{field} must be a non-empty string"),
879 next_action,
880 );
881 }
882}
883
884fn push_issue(
885 issues: &mut Vec<EventEnvelopeIssue>,
886 code: EventEnvelopeIssueCode,
887 path: impl Into<String>,
888 message: impl Into<String>,
889 next_action: impl Into<String>,
890) {
891 issues.push(EventEnvelopeIssue {
892 code,
893 path: path.into(),
894 message: message.into(),
895 next_action: next_action.into(),
896 });
897}