1use std::collections::BTreeMap;
37
38use bynk_syntax::ast::{EventDecl, TypeRef};
39use bynk_syntax::error::CompileError;
40use bynk_syntax::span::Span;
41
42use bynk_project::schema_registry::{EventEntry, FieldShape, SchemaRegistry};
43
44use crate::symbols::UnitTable;
45
46pub fn parse_or_diagnose(
52 existing: Option<&str>,
53 project_root: &std::path::Path,
54) -> Result<SchemaRegistry, CompileError> {
55 bynk_project::schema_registry::parse(existing, project_root).map_err(|msg| {
56 CompileError::new("bynk.project.schema_registry_corrupt", Span::default(), msg)
57 })
58}
59
60fn snapshot(event: &EventDecl) -> Vec<FieldShape> {
67 let mut fields: Vec<FieldShape> = event
68 .body
69 .fields
70 .iter()
71 .map(|f| FieldShape {
72 name: f.name.name.clone(),
73 ty: canon_type(&f.type_ref),
74 default: f.init.is_some(),
75 })
76 .collect();
77 fields.sort_by(|a, b| a.name.cmp(&b.name));
78 fields
79}
80
81fn canon_type(t: &TypeRef) -> String {
90 match t {
91 TypeRef::Base(b, _) => b.name().to_string(),
92 TypeRef::Named(id) => id.name.clone(),
93 TypeRef::Result(a, b, _) => format!("Result[{}, {}]", canon_type(a), canon_type(b)),
94 TypeRef::Option(t, _) => format!("Option[{}]", canon_type(t)),
95 TypeRef::Effect(t, _) => format!("Effect[{}]", canon_type(t)),
96 TypeRef::HttpResult(t, _) => format!("HttpResult[{}]", canon_type(t)),
97 TypeRef::QueueResult(_) => "QueueResult".to_string(),
98 TypeRef::List(t, _) => format!("List[{}]", canon_type(t)),
99 TypeRef::Map(k, v, _) => format!("Map[{}, {}]", canon_type(k), canon_type(v)),
100 TypeRef::Query(t, _) => format!("Query[{}]", canon_type(t)),
101 TypeRef::Stream(t, _) => format!("Stream[{}]", canon_type(t)),
102 TypeRef::Connection(t, _) => format!("Connection[{}]", canon_type(t)),
103 TypeRef::History(t, _) => format!("History[{}]", canon_type(t)),
104 TypeRef::ValidationError(_) => "ValidationError".to_string(),
105 TypeRef::JsonError(_) => "JsonError".to_string(),
106 TypeRef::Unit(_) => "()".to_string(),
107 TypeRef::Fn(params, ret, _) => format!(
108 "({}) -> {}",
109 params.iter().map(canon_type).collect::<Vec<_>>().join(", "),
110 canon_type(ret)
111 ),
112 TypeRef::App { name, args, .. } => format!(
113 "{}[{}]",
114 name.name,
115 args.iter().map(canon_type).collect::<Vec<_>>().join(", ")
116 ),
117 }
118}
119
120enum Reconciled {
123 Baseline { fields: Vec<FieldShape> },
126 Unchanged {
128 fields: Vec<FieldShape>,
129 stored: i64,
130 },
131 Additive {
133 fields: Vec<FieldShape>,
134 bumped: i64,
135 },
136 NonAdditive {
139 removed: Vec<String>,
140 retyped: Vec<String>,
141 added_without_default: Vec<String>,
142 lost_default: Vec<String>,
143 },
144}
145
146fn reconcile_one(current: &[FieldShape], stored: Option<&EventEntry>) -> Reconciled {
147 let Some(stored) = stored else {
148 return Reconciled::Baseline {
149 fields: current.to_vec(),
150 };
151 };
152 if current == stored.fields.as_slice() {
153 return Reconciled::Unchanged {
154 fields: current.to_vec(),
155 stored: stored.schema,
156 };
157 }
158
159 fn by_name(fields: &[FieldShape]) -> BTreeMap<&str, &FieldShape> {
160 fields.iter().map(|f| (f.name.as_str(), f)).collect()
161 }
162 let old = by_name(&stored.fields);
163 let new = by_name(current);
164
165 let removed: Vec<String> = old
166 .keys()
167 .filter(|n| !new.contains_key(*n))
168 .map(|n| n.to_string())
169 .collect();
170 let retyped: Vec<String> = old
171 .iter()
172 .filter_map(|(n, old_field)| {
173 new.get(n)
174 .filter(|new_field| new_field.ty != old_field.ty)
175 .map(|_| n.to_string())
176 })
177 .collect();
178 let added_without_default: Vec<String> = new
179 .iter()
180 .filter(|(n, f)| !old.contains_key(*n) && !f.default)
181 .map(|(n, _)| n.to_string())
182 .collect();
183 let lost_default: Vec<String> = old
188 .iter()
189 .filter_map(|(n, old_field)| {
190 new.get(n).filter(|new_field| {
191 new_field.ty == old_field.ty && old_field.default && !new_field.default
192 })
193 })
194 .map(|f| f.name.clone())
195 .collect();
196
197 if removed.is_empty()
198 && retyped.is_empty()
199 && added_without_default.is_empty()
200 && lost_default.is_empty()
201 {
202 Reconciled::Additive {
203 fields: current.to_vec(),
204 bumped: stored.schema + 1,
205 }
206 } else {
207 Reconciled::NonAdditive {
208 removed,
209 retyped,
210 lost_default,
211 added_without_default,
212 }
213 }
214}
215
216pub fn reconcile(
222 existing: &SchemaRegistry,
223 unit_tables: &std::collections::HashMap<String, UnitTable>,
224 errors: &mut Vec<CompileError>,
225) -> (SchemaRegistry, std::collections::HashMap<String, i64>) {
226 let mut updated = SchemaRegistry::new();
227 let mut effective = std::collections::HashMap::new();
228
229 let mut units: Vec<_> = unit_tables.iter().collect();
230 units.sort_by_key(|(name, _)| *name);
231
232 for (unit_name, table) in units {
233 let mut events: Vec<_> = table.events.iter().collect();
234 events.sort_by_key(|(name, _)| *name);
235 for (event_name, event) in events {
236 let key = format!("{unit_name}.{event_name}");
237 let fields = snapshot(event);
238 let declared = event.schema_version();
239 let annotation_span = event
240 .annotations
241 .iter()
242 .find(|a| a.name.name == "schema")
243 .map(|a| a.span);
244 let has_annotation = annotation_span.is_some();
245
246 let (effective_version, entry) = match reconcile_one(&fields, existing.get(&key)) {
247 Reconciled::Baseline { fields } => (
248 declared,
249 EventEntry {
250 schema: declared,
251 fields,
252 },
253 ),
254 Reconciled::Unchanged { fields, stored } => {
255 if has_annotation && declared != stored {
256 errors.push(mismatch_error(
257 event_name,
258 annotation_span.unwrap(),
259 declared,
260 stored,
261 ));
262 }
263 (
264 stored,
265 EventEntry {
266 schema: stored,
267 fields,
268 },
269 )
270 }
271 Reconciled::Additive { fields, bumped } => {
272 if has_annotation && declared != bumped {
273 errors.push(mismatch_error(
274 event_name,
275 annotation_span.unwrap(),
276 declared,
277 bumped,
278 ));
279 }
280 (
281 bumped,
282 EventEntry {
283 schema: bumped,
284 fields,
285 },
286 )
287 }
288 Reconciled::NonAdditive {
289 removed,
290 retyped,
291 added_without_default,
292 lost_default,
293 } => {
294 errors.push(non_additive_error(
295 event,
296 event_name,
297 &removed,
298 &retyped,
299 &added_without_default,
300 &lost_default,
301 ));
302 let old = existing
307 .get(&key)
308 .expect("a NonAdditive verdict only fires against a stored entry")
309 .clone();
310 (old.schema, old)
311 }
312 };
313
314 effective.insert(key.clone(), effective_version);
315 updated.insert(key, entry);
316 }
317 }
318
319 (updated, effective)
320}
321
322fn mismatch_error(
323 event_name: &str,
324 span: bynk_syntax::span::Span,
325 declared: i64,
326 computed: i64,
327) -> CompileError {
328 CompileError::new(
329 "bynk.event.schema_version_mismatch",
330 span,
331 format!(
332 "`{event_name}`'s `@schema({declared})` disagrees with the schema \
333 registry, which computes version {computed} from the event's \
334 build history"
335 ),
336 )
337 .with_note(format!(
338 "update the annotation to `@schema({computed})`, or remove it to let \
339 the compiler track the version automatically"
340 ))
341}
342
343fn non_additive_error(
344 event: &EventDecl,
345 event_name: &str,
346 removed: &[String],
347 retyped: &[String],
348 added_without_default: &[String],
349 lost_default: &[String],
350) -> CompileError {
351 let mut parts = Vec::new();
352 if !removed.is_empty() {
353 parts.push(format!("field(s) removed: {}", removed.join(", ")));
354 }
355 if !retyped.is_empty() {
356 parts.push(format!("field(s) retyped: {}", retyped.join(", ")));
357 }
358 if !added_without_default.is_empty() {
359 parts.push(format!(
360 "field(s) added without a default: {}",
361 added_without_default.join(", ")
362 ));
363 }
364 if !lost_default.is_empty() {
365 parts.push(format!(
366 "field(s) lost their default: {}",
367 lost_default.join(", ")
368 ));
369 }
370 CompileError::new(
371 "bynk.event.non_additive_schema_change",
372 event.span,
373 format!(
374 "`{event_name}` changed in a way the schema registry cannot \
375 evolve additively — {}",
376 parts.join("; ")
377 ),
378 )
379 .with_note(
380 "an additive change adds only fields that carry a default; give a \
381 breaking change a new event type name instead",
382 )
383}
384
385#[cfg(test)]
386mod tests {
387 use super::*;
388 use bynk_syntax::ast::{
389 Annotation, AnnotationArg, BaseType, Expr, ExprId, ExprKind, Ident, RecordBody, Trivia,
390 };
391 use bynk_syntax::span::Span;
392 use std::collections::HashMap as StdHashMap;
393
394 fn ident(name: &str) -> Ident {
395 Ident {
396 name: name.to_string(),
397 span: Span::default(),
398 }
399 }
400
401 fn int_lit(value: i64) -> Expr {
402 Expr {
403 id: ExprId::SYNTHETIC,
404 kind: ExprKind::IntLit {
405 value,
406 lexeme: value.to_string(),
407 },
408 span: Span::default(),
409 }
410 }
411
412 fn field(name: &str, ty: TypeRef, has_default: bool) -> bynk_syntax::ast::RecordField {
413 bynk_syntax::ast::RecordField {
414 name: ident(name),
415 type_ref: ty,
416 refinement: None,
417 init: has_default.then(|| int_lit(0)),
418 span: Span::default(),
419 }
420 }
421
422 fn schema_annotation(n: i64) -> Annotation {
423 Annotation {
424 name: ident("schema"),
425 args: vec![AnnotationArg {
426 label: None,
427 value: int_lit(n),
428 span: Span::default(),
429 }],
430 span: Span::default(),
431 }
432 }
433
434 fn event(
435 name: &str,
436 annotations: Vec<Annotation>,
437 fields: Vec<bynk_syntax::ast::RecordField>,
438 ) -> EventDecl {
439 EventDecl {
440 name: ident(name),
441 annotations,
442 body: RecordBody {
443 fields,
444 span: Span::default(),
445 },
446 documentation: None,
447 span: Span::default(),
448 trivia: Trivia::default(),
449 }
450 }
451
452 fn int_ty() -> TypeRef {
453 TypeRef::Base(BaseType::Int, Span::default())
454 }
455 fn string_ty() -> TypeRef {
456 TypeRef::Base(BaseType::String, Span::default())
457 }
458
459 fn shape(name: &str, ty: &str, default: bool) -> FieldShape {
460 FieldShape {
461 name: name.to_string(),
462 ty: ty.to_string(),
463 default,
464 }
465 }
466
467 #[test]
470 fn canon_type_renders_base_and_generic_shapes() {
471 assert_eq!(canon_type(&int_ty()), "Int");
472 assert_eq!(
473 canon_type(&TypeRef::Option(Box::new(string_ty()), Span::default())),
474 "Option[String]"
475 );
476 assert_eq!(
477 canon_type(&TypeRef::List(Box::new(int_ty()), Span::default())),
478 "List[Int]"
479 );
480 }
481
482 #[test]
485 fn no_entry_baselines_silently() {
486 let current = vec![shape("orderId", "String", false)];
487 match reconcile_one(¤t, None) {
488 Reconciled::Baseline { fields } => assert_eq!(fields, current),
489 _ => panic!("expected Baseline"),
490 }
491 }
492
493 #[test]
494 fn unchanged_shape_keeps_stored_version() {
495 let current = vec![shape("orderId", "String", false)];
496 let stored = EventEntry {
497 schema: 2,
498 fields: current.clone(),
499 };
500 match reconcile_one(¤t, Some(&stored)) {
501 Reconciled::Unchanged { stored: v, .. } => assert_eq!(v, 2),
502 _ => panic!("expected Unchanged"),
503 }
504 }
505
506 #[test]
507 fn additive_field_with_default_bumps_version() {
508 let old = vec![shape("orderId", "String", false)];
509 let new = vec![
510 shape("orderId", "String", false),
511 shape("region", "Region", true),
512 ];
513 let stored = EventEntry {
514 schema: 1,
515 fields: old,
516 };
517 match reconcile_one(&new, Some(&stored)) {
518 Reconciled::Additive { bumped, .. } => assert_eq!(bumped, 2),
519 _ => panic!("expected Additive"),
520 }
521 }
522
523 #[test]
524 fn field_removed_is_non_additive() {
525 let old = vec![
526 shape("orderId", "String", false),
527 shape("region", "Region", false),
528 ];
529 let new = vec![shape("orderId", "String", false)];
530 let stored = EventEntry {
531 schema: 1,
532 fields: old,
533 };
534 match reconcile_one(&new, Some(&stored)) {
535 Reconciled::NonAdditive { removed, .. } => {
536 assert_eq!(removed, vec!["region".to_string()])
537 }
538 _ => panic!("expected NonAdditive"),
539 }
540 }
541
542 #[test]
543 fn field_retyped_is_non_additive() {
544 let old = vec![shape("orderId", "String", false)];
545 let new = vec![shape("orderId", "Int", false)];
546 let stored = EventEntry {
547 schema: 1,
548 fields: old,
549 };
550 match reconcile_one(&new, Some(&stored)) {
551 Reconciled::NonAdditive { retyped, .. } => {
552 assert_eq!(retyped, vec!["orderId".to_string()])
553 }
554 _ => panic!("expected NonAdditive"),
555 }
556 }
557
558 #[test]
559 fn field_added_without_default_is_non_additive() {
560 let old = vec![shape("orderId", "String", false)];
561 let new = vec![
562 shape("orderId", "String", false),
563 shape("region", "Region", false),
564 ];
565 let stored = EventEntry {
566 schema: 1,
567 fields: old,
568 };
569 match reconcile_one(&new, Some(&stored)) {
570 Reconciled::NonAdditive {
571 added_without_default,
572 ..
573 } => assert_eq!(added_without_default, vec!["region".to_string()]),
574 _ => panic!("expected NonAdditive"),
575 }
576 }
577
578 #[test]
579 fn a_field_losing_its_default_is_non_additive() {
580 let old = vec![
585 shape("orderId", "String", false),
586 shape("region", "String", true),
587 ];
588 let new = vec![
589 shape("orderId", "String", false),
590 shape("region", "String", false),
591 ];
592 let stored = EventEntry {
593 schema: 1,
594 fields: old,
595 };
596 match reconcile_one(&new, Some(&stored)) {
597 Reconciled::NonAdditive { lost_default, .. } => {
598 assert_eq!(lost_default, vec!["region".to_string()])
599 }
600 _ => panic!("expected NonAdditive"),
601 }
602 }
603
604 fn table_with(events: Vec<(&str, EventDecl)>) -> UnitTable {
607 UnitTable {
608 kind: None,
609 types: StdHashMap::new(),
610 fns: StdHashMap::new(),
611 methods: StdHashMap::new(),
612 capabilities: StdHashMap::new(),
613 providers: StdHashMap::new(),
614 services: StdHashMap::new(),
615 agents: StdHashMap::new(),
616 actors: StdHashMap::new(),
617 exported_capabilities: Default::default(),
618 events: events
619 .into_iter()
620 .map(|(n, e)| (n.to_string(), e))
621 .collect(),
622 }
623 }
624
625 #[test]
626 fn reconcile_baselines_a_brand_new_event_at_its_declared_annotation() {
627 let e = event(
628 "PaymentConfirmed",
629 vec![schema_annotation(3)],
630 vec![field("orderId", string_ty(), false)],
631 );
632 let mut units = StdHashMap::new();
633 units.insert(
634 "commerce.order".to_string(),
635 table_with(vec![("PaymentConfirmed", e)]),
636 );
637 let existing = SchemaRegistry::new();
638 let mut errors = Vec::new();
639 let (updated, effective) = reconcile(&existing, &units, &mut errors);
640 assert!(errors.is_empty(), "a first-ever compile must not error");
641 assert_eq!(effective.get("commerce.order.PaymentConfirmed"), Some(&3));
642 assert_eq!(
643 updated
644 .get("commerce.order.PaymentConfirmed")
645 .unwrap()
646 .schema,
647 3
648 );
649 }
650
651 #[test]
652 fn reconcile_rejects_a_mismatched_schema_annotation_on_an_unchanged_event() {
653 let e = event(
654 "PaymentConfirmed",
655 vec![schema_annotation(5)],
656 vec![field("orderId", string_ty(), false)],
657 );
658 let mut units = StdHashMap::new();
659 units.insert(
660 "commerce.order".to_string(),
661 table_with(vec![("PaymentConfirmed", e)]),
662 );
663 let mut existing = SchemaRegistry::new();
664 existing.insert(
665 "commerce.order.PaymentConfirmed".to_string(),
666 EventEntry {
667 schema: 2,
668 fields: vec![shape("orderId", "String", false)],
669 },
670 );
671 let mut errors = Vec::new();
672 let (_, effective) = reconcile(&existing, &units, &mut errors);
673 assert_eq!(errors.len(), 1);
674 assert_eq!(errors[0].category, "bynk.event.schema_version_mismatch");
675 assert_eq!(effective.get("commerce.order.PaymentConfirmed"), Some(&2));
678 }
679
680 #[test]
681 fn reconcile_auto_bumps_an_unannotated_additive_change() {
682 let e = event(
683 "OrderCancelled",
684 vec![],
685 vec![
686 field("orderId", string_ty(), false),
687 field("reason", string_ty(), true),
688 ],
689 );
690 let mut units = StdHashMap::new();
691 units.insert(
692 "commerce.order".to_string(),
693 table_with(vec![("OrderCancelled", e)]),
694 );
695 let mut existing = SchemaRegistry::new();
696 existing.insert(
697 "commerce.order.OrderCancelled".to_string(),
698 EventEntry {
699 schema: 1,
700 fields: vec![shape("orderId", "String", false)],
701 },
702 );
703 let mut errors = Vec::new();
704 let (updated, effective) = reconcile(&existing, &units, &mut errors);
705 assert!(errors.is_empty());
706 assert_eq!(effective.get("commerce.order.OrderCancelled"), Some(&2));
707 assert_eq!(
708 updated.get("commerce.order.OrderCancelled").unwrap().schema,
709 2
710 );
711 }
712
713 #[test]
714 fn reconcile_rejects_a_non_additive_change_and_keeps_the_old_entry() {
715 let e = event(
716 "OrderCancelled",
717 vec![],
718 vec![field("orderId", string_ty(), false)],
719 );
720 let mut units = StdHashMap::new();
721 units.insert(
722 "commerce.order".to_string(),
723 table_with(vec![("OrderCancelled", e)]),
724 );
725 let mut existing = SchemaRegistry::new();
726 existing.insert(
727 "commerce.order.OrderCancelled".to_string(),
728 EventEntry {
729 schema: 4,
730 fields: vec![
731 shape("orderId", "String", false),
732 shape("reason", "String", false),
733 ],
734 },
735 );
736 let mut errors = Vec::new();
737 let (updated, effective) = reconcile(&existing, &units, &mut errors);
738 assert_eq!(errors.len(), 1);
739 assert_eq!(errors[0].category, "bynk.event.non_additive_schema_change");
740 assert_eq!(effective.get("commerce.order.OrderCancelled"), Some(&4));
741 assert_eq!(
742 updated.get("commerce.order.OrderCancelled").unwrap().schema,
743 4
744 );
745 }
746
747 #[test]
748 fn a_stale_key_for_a_renamed_event_is_dropped_silently() {
749 let e = event(
754 "PaymentConfirmedV2",
755 vec![],
756 vec![field("orderId", string_ty(), false)],
757 );
758 let mut units = StdHashMap::new();
759 units.insert(
760 "commerce.order".to_string(),
761 table_with(vec![("PaymentConfirmedV2", e)]),
762 );
763 let mut existing = SchemaRegistry::new();
764 existing.insert(
765 "commerce.order.PaymentConfirmed".to_string(),
766 EventEntry {
767 schema: 3,
768 fields: vec![shape("orderId", "String", false)],
769 },
770 );
771 let mut errors = Vec::new();
772 let (updated, _) = reconcile(&existing, &units, &mut errors);
773 assert!(errors.is_empty());
774 assert!(updated.get("commerce.order.PaymentConfirmed").is_none());
775 assert!(updated.get("commerce.order.PaymentConfirmedV2").is_some());
776 }
777
778 #[test]
785 fn parse_or_diagnose_passes_through_a_valid_registry() {
786 let mut reg = SchemaRegistry::new();
787 reg.insert(
788 "commerce.order.PaymentConfirmed".to_string(),
789 EventEntry {
790 schema: 1,
791 fields: vec![shape("orderId", "String", false)],
792 },
793 );
794 let text = bynk_project::schema_registry::serialize(®);
795 let parsed = parse_or_diagnose(Some(&text), std::path::Path::new("/tmp"))
796 .expect("a freshly serialized registry must parse");
797 assert_eq!(
798 parsed
799 .get("commerce.order.PaymentConfirmed")
800 .map(|e| e.schema),
801 Some(1)
802 );
803 }
804
805 #[test]
806 fn parse_or_diagnose_reports_a_corrupt_registry_under_its_own_code() {
807 let err = parse_or_diagnose(Some("not valid toml {{{"), std::path::Path::new("/tmp"))
808 .expect_err("garbage content must not parse");
809 assert_eq!(err.category, "bynk.project.schema_registry_corrupt");
810 }
811
812 #[test]
813 fn parse_or_diagnose_with_no_existing_content_baselines_empty() {
814 let parsed = parse_or_diagnose(None, std::path::Path::new("/tmp"))
815 .expect("no lock file yet is not corruption");
816 assert!(parsed.get("anything.at_all").is_none());
817 }
818}