1	import Base
2	import owner_product.bend as Product
3	import ordinary_event_declaration.bend as Event
4	import event_runtime.bend as Ev
5	import event_stream.bend as Leaf
6	import event_publication.bend as Publication
7	
8	# Builder domains are operational declarations, not a second metadata list.
9	# Detached left/right domain declarations for a heterogeneous event family.
10	# Their nested shape matches the publication and storage product, not live readers.
11	type Domains<A:Data,B:Data> is Data:Domains{first:A,second:B}
12	# Typed family setup containing ordinary domains and closed storage/routing
13	# operations. Runtime streams are constructed separately; this declaration
14	# supplies no live cursor or affine payload. Raw constructors are trusted setup.
15	# Family construction and reader lifecycle: docs/parity/typed-data-events.md.
16	type Provision<S:Data,P:Data,D:Data,F:Type,-make:D -> Nat -> F,-sync:F -> Nat -> List<&2,P> -> F,-frame:F -> Nat -> F,-retain:F -> Nat -> List<&2,P> -> F & List<&2,P>,-inventory:F -> F & List<&2,String>> is Data:
17	  Provision{domains:D}
18	def leaf_make(~S:Data,~E:Data,domain:Event.Domain<S,E>,capacity:Nat) -> Leaf.Stream<S,E>:Leaf.provision(~S,~E,domain,capacity)
19	def leaf_sync(~S:Data,~E:Data,stream:Leaf.Stream<S,E>,tick:Nat,values:List<&2,E>) -> Leaf.Stream<S,E>:Leaf.append(~S,~E,stream,tick,values)
20	def leaf_frame(~S:Data,~E:Data,stream:Leaf.Stream<S,E>,tick:Nat) -> Leaf.Stream<S,E>:Leaf.frame(~S,~E,stream,tick)
21	def leaf_retained(~S:Data,~E:Data,stream:Leaf.Stream<S,E>,remove:Bool,values:List<&2,E>) -> Leaf.Stream<S,E> & List<&2,E>:
22	  match remove:
23	    case False{}:(stream,values)
24	    case True{}:(stream,[])
25	def leaf_retain(~S:Data,~E:Data,stream:Leaf.Stream<S,E>,tick:Nat,values:List<&2,E>) -> Leaf.Stream<S,E> & List<&2,E>:
26	  +retained=stream
27	  leaf_retained(~S,~E,retained,Nat.is_le(tick,Leaf.dropped(~S,~E,retained)),values)
28	def leaf_inventory(~S:Data,~E:Data,stream:Leaf.Stream<S,E>) -> Leaf.Stream<S,E> & List<&2,String>:
29	  +retained=stream
30	  (retained,[Leaf.inventory(~S,~E,retained)])
31	# Declare one Data event family leaf from its schema-scoped domain. The
32	# resulting provision supplies stream creation, synchronization and retention
33	# operations, but does not create a stream until runtime setup invokes make.
34	# Example: Family.leaf(~S,~E,Event.domain(~S,~E,"hit")).
35	def leaf(~S:Data,~E:Data,domain:Event.Domain<S,E>) -> Provision<S,E,Event.Domain<S,E>,Leaf.Stream<S,E>,leaf_make(~S,~E),leaf_sync(~S,~E),leaf_frame(~S,~E),leaf_retain(~S,~E),leaf_inventory(~S,~E)>:Provision{domain}
36	def pair_make(~DA:Data,~DB:Data,~FA:Type,~FB:Type,~makeA:DA -> Nat -> FA,~makeB:DB -> Nat -> FB,domains:Domains<DA,DB>,+capacity:Nat) -> Product.Owner<FA,FB>:
37	  match domains:
38	    case Domains{first,second}:Product.Owner{makeA(first,capacity),makeB(second,capacity)}
39	def pair_sync(~A:Data,~B:Data,~FA:Type,~FB:Type,~syncA:FA -> Nat -> List<&2,A> -> FA,~syncB:FB -> Nat -> List<&2,B> -> FB,family:Product.Owner<FA,FB>,+tick:Nat,+records:List<&2,Publication.Publication<A,B>>) -> Product.Owner<FA,FB>:
40	  match family:
41	    case Product.Owner{first,second}:Product.Owner{syncA(first,tick,Publication.first_values(~A,~B,records)),syncB(second,tick,Publication.second_values(~A,~B,records))}
42	def pair_frame(~FA:Type,~FB:Type,~frameA:FA -> Nat -> FA,~frameB:FB -> Nat -> FB,family:Product.Owner<FA,FB>,+tick:Nat) -> Product.Owner<FA,FB>:
43	  match family:
44	    case Product.Owner{first,second}:Product.Owner{frameA(first,tick),frameB(second,tick)}
45	def pair_inventory_join(~FA:Type,~FB:Type,first:FA & List<&2,String>,second:FB & List<&2,String>) -> Product.Owner<FA,FB> & List<&2,String>:
46	  (first,firstNames)=first
47	  (second,secondNames)=second
48	  (Product.Owner{first,second},Ev.append(~String,firstNames,secondNames))
49	def pair_inventory(~FA:Type,~FB:Type,~inventoryA:FA -> FA & List<&2,String>,~inventoryB:FB -> FB & List<&2,String>,family:Product.Owner<FA,FB>) -> Product.Owner<FA,FB> & List<&2,String>:
50	  match family:
51	    case Product.Owner{first,second}:pair_inventory_join(~FA,~FB,inventoryA(first),inventoryB(second))
52	def retained_tail(~A:Data,~B:Data,~F:Type,record:Publication.Publication<A,B>,returned:F & List<&2,Publication.Publication<A,B>>) -> F & List<&2,Publication.Publication<A,B>>:
53	  match returned:
54	    case (family,rest):(family,Publication.retained_join(~A,~B,record,rest))
55	def retained_children(~A:Data,~B:Data,~FA:Type,~FB:Type,rest:Product.Owner<FA,FB> -> Product.Owner<FA,FB> & List<&2,Publication.Publication<A,B>>,first:FA & List<&2,A>,second:FB & List<&2,B>) -> Product.Owner<FA,FB> & List<&2,Publication.Publication<A,B>>:
56	  (first,firstValues)=first
57	  (second,secondValues)=second
58	  retained_tail(~A,~B,~Product.Owner<FA,FB>,Publication.Publication{firstValues,secondValues},rest(Product.Owner{first,second}))
59	def retained_record(~A:Data,~B:Data,~FA:Type,~FB:Type,~retainA:FA -> Nat -> List<&2,A> -> FA & List<&2,A>,~retainB:FB -> Nat -> List<&2,B> -> FB & List<&2,B>,rest:Product.Owner<FA,FB> -> Product.Owner<FA,FB> & List<&2,Publication.Publication<A,B>>,+tick:Nat,record:Publication.Publication<A,B>,family:Product.Owner<FA,FB>) -> Product.Owner<FA,FB> & List<&2,Publication.Publication<A,B>>:
60	  match record family:
61	    case Publication.Publication{a,b} Product.Owner{first,second}:retained_children(~A,~B,~FA,~FB,rest,retainA(first,tick,a),retainB(second,tick,b))
62	def retained_one_children(~A:Data,~B:Data,~FA:Type,~FB:Type,first:FA & List<&2,A>,second:FB & List<&2,B>) -> Product.Owner<FA,FB> & Publication.Publication<A,B>:
63	  (first,firstValues)=first
64	  (second,secondValues)=second
65	  (Product.Owner{first,second},Publication.Publication{firstValues,secondValues})
66	def retained_one(~A:Data,~B:Data,~FA:Type,~FB:Type,~retainA:FA -> Nat -> List<&2,A> -> FA & List<&2,A>,~retainB:FB -> Nat -> List<&2,B> -> FB & List<&2,B>,+tick:Nat,record:Publication.Publication<A,B>,family:Product.Owner<FA,FB>) -> Product.Owner<FA,FB> & Publication.Publication<A,B>:
67	  match record family:
68	    case Publication.Publication{a,b} Product.Owner{first,second}:retained_one_children(~A,~B,~FA,~FB,retainA(first,tick,a),retainB(second,tick,b))
69	def pair_retain_acc(~A:Data,~B:Data,~FA:Type,~FB:Type,~retainA:FA -> Nat -> List<&2,A> -> FA & List<&2,A>,~retainB:FB -> Nat -> List<&2,B> -> FB & List<&2,B>,records:List<&2,Publication.Publication<A,B>>,returned:Product.Owner<FA,FB> & Publication.Publication<A,B>,+tick:Nat,reversed:List<&2,Publication.Publication<A,B>>) -> Product.Owner<FA,FB> & List<&2,Publication.Publication<A,B>>:
70	  match records:
71	    case []:
72	      (family,record)=returned
73	      (family,List.reverse(&2,Publication.Publication<A,B>,Publication.retained_join(~A,~B,record,reversed)))
74	    case head <> rest:
75	      (family,record)=returned
76	      pair_retain_acc(~A,~B,~FA,~FB,~retainA,~retainB,rest,retained_one(~A,~B,~FA,~FB,~retainA,~retainB,tick,head,family),tick,Publication.retained_join(~A,~B,record,reversed))
77	def pair_retain_loop(~A:Data,~B:Data,~FA:Type,~FB:Type,~retainA:FA -> Nat -> List<&2,A> -> FA & List<&2,A>,~retainB:FB -> Nat -> List<&2,B> -> FB & List<&2,B>,records:List<&2,Publication.Publication<A,B>>,family:Product.Owner<FA,FB>,+tick:Nat) -> Product.Owner<FA,FB> & List<&2,Publication.Publication<A,B>>:
78	  match records:
79	    case []:(family,[])
80	    case record <> rest:pair_retain_acc(~A,~B,~FA,~FB,~retainA,~retainB,rest,retained_one(~A,~B,~FA,~FB,~retainA,~retainB,tick,record,family),tick,[])
81	def pair_retain(~A:Data,~B:Data,~FA:Type,~FB:Type,~retainA:FA -> Nat -> List<&2,A> -> FA & List<&2,A>,~retainB:FB -> Nat -> List<&2,B> -> FB & List<&2,B>,family:Product.Owner<FA,FB>,tick:Nat,records:List<&2,Publication.Publication<A,B>>) -> Product.Owner<FA,FB> & List<&2,Publication.Publication<A,B>>:
82	  pair_retain_loop(~A,~B,~FA,~FB,~retainA,~retainB,records,family,tick)
83	
84	def node_make(~DA:Data,~DB:Data,~FA:Type,~FB:Type,~makeA:DA -> Nat -> FA,~makeB:DB -> Nat -> FB,domains:Domains<DA,DB>,capacity:Nat) -> Product.Owner<FA,FB>:pair_make(~DA,~DB,~FA,~FB,~makeA,~makeB,domains,capacity)
85	def node_sync(~A:Data,~B:Data,~FA:Type,~FB:Type,~syncA:FA -> Nat -> List<&2,A> -> FA,~syncB:FB -> Nat -> List<&2,B> -> FB,family:Product.Owner<FA,FB>,tick:Nat,records:List<&2,Publication.Publication<A,B>>) -> Product.Owner<FA,FB>:pair_sync(~A,~B,~FA,~FB,~syncA,~syncB,family,tick,records)
86	def node_frame(~FA:Type,~FB:Type,~frameA:FA -> Nat -> FA,~frameB:FB -> Nat -> FB,family:Product.Owner<FA,FB>,tick:Nat) -> Product.Owner<FA,FB>:pair_frame(~FA,~FB,~frameA,~frameB,family,tick)
87	# Compose two event provisions under the same schema in left/right order.
88	# Publication wrappers and family storage become matching nested products;
89	# each leaf retains its own typed stream. Composition neither publishes records
90	# nor combines independent reader cursors into one. Nest branches for more types.
91	def branch(~S:Data,~A:Data,~B:Data,~DA:Data,~DB:Data,~FA:Type,~FB:Type,~makeA:DA -> Nat -> FA,~makeB:DB -> Nat -> FB,~syncA:FA -> Nat -> List<&2,A> -> FA,~syncB:FB -> Nat -> List<&2,B> -> FB,~frameA:FA -> Nat -> FA,~frameB:FB -> Nat -> FB,~retainA:FA -> Nat -> List<&2,A> -> FA & List<&2,A>,~retainB:FB -> Nat -> List<&2,B> -> FB & List<&2,B>,~inventoryA:FA -> FA & List<&2,String>,~inventoryB:FB -> FB & List<&2,String>,first:Provision<S,A,DA,FA,makeA,syncA,frameA,retainA,inventoryA>,second:Provision<S,B,DB,FB,makeB,syncB,frameB,retainB,inventoryB>) -> Provision<S,Publication.Publication<A,B>,Domains<DA,DB>,Product.Owner<FA,FB>,node_make(~DA,~DB,~FA,~FB,~makeA,~makeB),node_sync(~A,~B,~FA,~FB,~syncA,~syncB),node_frame(~FA,~FB,~frameA,~frameB),pair_retain(~A,~B,~FA,~FB,~retainA,~retainB),pair_inventory(~FA,~FB,~inventoryA,~inventoryB)>:
92	  match first second:
93	    case Provision{first} Provision{second}:Provision{Domains{first,second}}
94	
95	# Typed path to one event stream in a family, with owner-threading take/put
96	# operations. This Data path owns no stream or reader; raw indices are trusted
97	# setup. Equal payload types do not justify substituting unrelated family paths.
98	type Selection<S:Data,E:Data,F:Type,-take:F -> F & Leaf.Stream<S,E>,-put:F -> Leaf.Stream<S,E> -> F> is Data:Selection{}
99	def selected_leaf(~S:Data,~E:Data,stream:Leaf.Stream<S,E>) -> Leaf.Stream<S,E> & Leaf.Stream<S,E>:
100	  +retained=stream
101	  (retained,retained)
102	def replaced_leaf(~S:Data,~E:Data,_:Leaf.Stream<S,E>,stream:Leaf.Stream<S,E>) -> Leaf.Stream<S,E>:stream
103	# Select a stream that is itself the complete family owner. The selection
104	# records closed extraction/replacement operations and performs neither here.
105	def select_leaf(~S:Data,~E:Data) -> Selection<S,E,Leaf.Stream<S,E>,selected_leaf(~S,~E),replaced_leaf(~S,~E)>:Selection{}
106	def selected_left(~S:Data,~E:Data,~FA:Type,~FB:Type,second:FB,returned:FA & Leaf.Stream<S,E>) -> Product.Owner<FA,FB> & Leaf.Stream<S,E>:
107	  match returned:
108	    case (first,stream):(Product.Owner{first,second},stream)
109	def left_take(~S:Data,~E:Data,~FA:Type,~FB:Type,~take:FA -> FA & Leaf.Stream<S,E>,family:Product.Owner<FA,FB>) -> Product.Owner<FA,FB> & Leaf.Stream<S,E>:
110	  match family:
111	    case Product.Owner{first,second}:selected_left(~S,~E,~FA,~FB,second,take(first))
112	def left_put(~S:Data,~E:Data,~FA:Type,~FB:Type,~put:FA -> Leaf.Stream<S,E> -> FA,family:Product.Owner<FA,FB>,stream:Leaf.Stream<S,E>) -> Product.Owner<FA,FB>:
113	  match family:
114	    case Product.Owner{first,second}:Product.Owner{put(first,stream),second}
115	def selected_right(~S:Data,~E:Data,~FA:Type,~FB:Type,first:FA,returned:FB & Leaf.Stream<S,E>) -> Product.Owner<FA,FB> & Leaf.Stream<S,E>:
116	  match returned:
117	    case (second,stream):(Product.Owner{first,second},stream)
118	def right_take(~S:Data,~E:Data,~FA:Type,~FB:Type,~take:FB -> FB & Leaf.Stream<S,E>,family:Product.Owner<FA,FB>) -> Product.Owner<FA,FB> & Leaf.Stream<S,E>:
119	  match family:
120	    case Product.Owner{first,second}:selected_right(~S,~E,~FA,~FB,first,take(second))
121	def right_put(~S:Data,~E:Data,~FA:Type,~FB:Type,~put:FB -> Leaf.Stream<S,E> -> FB,family:Product.Owner<FA,FB>,stream:Leaf.Stream<S,E>) -> Product.Owner<FA,FB>:
122	  match family:
123	    case Product.Owner{first,second}:Product.Owner{first,put(second,stream)}
124	# Lift a leaf selection through the left side of a family product.
125	# The generated path preserves the right sibling when accessing/replacing
126	# the selected stream; it does not allocate or advance a reader.
127	def select_left(~S:Data,~E:Data,~FA:Type,~FB:Type,~take:FA -> FA & Leaf.Stream<S,E>,~put:FA -> Leaf.Stream<S,E> -> FA,_:Selection<S,E,FA,take,put>) -> Selection<S,E,Product.Owner<FA,FB>,left_take(~S,~E,~FA,~FB,~take),left_put(~S,~E,~FA,~FB,~put)>:Selection{}
128	# Lift a leaf selection through the right side of a family product,
129	# preserving the left sibling. Compose paths to address nested families.
130	def select_right(~S:Data,~E:Data,~FA:Type,~FB:Type,~take:FB -> FB & Leaf.Stream<S,E>,~put:FB -> Leaf.Stream<S,E> -> FB,_:Selection<S,E,FB,take,put>) -> Selection<S,E,Product.Owner<FA,FB>,right_take(~S,~E,~FA,~FB,~take),right_put(~S,~E,~FA,~FB,~put)>:Selection{}
131	# Identity injection for a Data event in a leaf publication. Does not send
132	# it or establish a publication tick; runtime publication does that separately.
133	def inject_leaf(~E:Data,event:E) -> E:event
134	# Wrap one event on the left side of a heterogeneous publication with an
135	# empty right list. The closed inject callback embeds nested leaf shapes;
136	# this creates a Data publication value, without mutating any runtime stream.
137	def inject_left(~E:Data,~A:Data,~B:Data,~inject:E -> A,event:E) -> Publication.Publication<A,B>:Publication.Publication{[inject(event)],[]}
138	# Wrap one event on the right side of a heterogeneous publication with an
139	# empty left list. Retains the nested injection shape; does not publish or
140	# change a cursor until the runtime processes the resulting value.
141	def inject_right(~E:Data,~A:Data,~B:Data,~inject:E -> B,event:E) -> Publication.Publication<A,B>:Publication.Publication{[],[inject(event)]}