1	import Base
2	import world.bend as W
3	import event_runtime.bend as Ev
4	import transaction.bend as Tx
5	import ordinary_event_declaration.bend as Event
6	import event_stream.bend as Leaf
7	
8	# Each ordinary emit stages one singleton product record. Tx.finish preserves
9	# the record list's emission order; products are never merged across emissions.
10	type Publication<A:Data,B:Data> is Data:
11	  Publication{first:List<&2,A>,second:List<&2,B>}
12	type Streams<S:Data,A:Data,B:Data,H:Type> is Type:
13	  Streams{first:Leaf.Stream<S,A>,second:Leaf.Stream<S,B>,owners:H}
14	type Application<S:Data,C:Type,R:Type,A:Data,B:Data,H:Type> is Type:
15	  Application{world:W.World<S,C,R,Publication<A,B>>,streams:Streams<S,A,B,H>,journal:List<&2,Ev.Batch<Publication<A,B>>>,tick:Nat,frameStart:Nat}
16	def setup(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,world:W.World<S,C,R,Publication<A,B>>,first:Event.Domain<S,A>,second:Event.Domain<S,B>,owners:H,+capacity:Nat) -> Application<S,C,R,A,B,H>:
17	  Application{world,Streams{Leaf.provision(~S,~A,first,capacity),Leaf.provision(~S,~B,second,capacity),owners},[],0n,0n}
18	def first_take(~S:Data,~A:Data,~B:Data,~H:Type,streams:Streams<S,A,B,H>) -> Streams<S,A,B,H> & Leaf.Stream<S,A>:
19	  match streams:
20	    case Streams{+first,second,owners}:(Streams{first,second,owners},first)
21	def first_put(~S:Data,~A:Data,~B:Data,~H:Type,streams:Streams<S,A,B,H>,first:Leaf.Stream<S,A>) -> Streams<S,A,B,H>:
22	  match streams:
23	    case Streams{_,second,owners}:Streams{first,second,owners}
24	def second_take(~S:Data,~A:Data,~B:Data,~H:Type,streams:Streams<S,A,B,H>) -> Streams<S,A,B,H> & Leaf.Stream<S,B>:
25	  match streams:
26	    case Streams{first,+second,owners}:(Streams{first,second,owners},second)
27	def second_put(~S:Data,~A:Data,~B:Data,~H:Type,streams:Streams<S,A,B,H>,second:Leaf.Stream<S,B>) -> Streams<S,A,B,H>:
28	  match streams:
29	    case Streams{first,_,owners}:Streams{first,second,owners}
30	def emit_first(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,tx:Tx.Tx<W.World<S,C,R,Publication<A,B>>,Publication<A,B>>,event:A) -> Tx.Tx<W.World<S,C,R,Publication<A,B>>,Publication<A,B>>:
31	  Tx.emit(~W.World<S,C,R,Publication<A,B>>,~Publication<A,B>,tx,Publication{[event],[]})
32	def emit_second(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,tx:Tx.Tx<W.World<S,C,R,Publication<A,B>>,Publication<A,B>>,event:B) -> Tx.Tx<W.World<S,C,R,Publication<A,B>>,Publication<A,B>>:
33	  Tx.emit(~W.World<S,C,R,Publication<A,B>>,~Publication<A,B>,tx,Publication{[],[event]})
34	# Reverse-accumulate each record's values, then reverse once. Both traversals
35	# decrease before the accumulator and preserve order within/across records.
36	def prepend_values(~E:Data,values:List<&2,E>,reversed:List<&2,E>) -> List<&2,E>:
37	  match values:
38	    case []:reversed
39	    case head <> rest:prepend_values(~E,rest,head <> reversed)
40	def first_values_loop(~A:Data,~B:Data,items:List<&2,Publication<A,B>>,reversed:List<&2,A>) -> List<&2,A>:
41	  match items:
42	    case []:List.reverse(&2,A,reversed)
43	    case Publication{first,_} <> rest:first_values_loop(~A,~B,rest,prepend_values(~A,first,reversed))
44	def first_values(~A:Data,~B:Data,items:List<&2,Publication<A,B>>) -> List<&2,A>:first_values_loop(~A,~B,items,[])
45	def second_values_loop(~A:Data,~B:Data,items:List<&2,Publication<A,B>>,reversed:List<&2,B>) -> List<&2,B>:
46	  match items:
47	    case []:List.reverse(&2,B,reversed)
48	    case Publication{_,second} <> rest:second_values_loop(~A,~B,rest,prepend_values(~B,second,reversed))
49	def second_values(~A:Data,~B:Data,items:List<&2,Publication<A,B>>) -> List<&2,B>:second_values_loop(~A,~B,items,[])
50	def synced(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,observed:W.World<S,C,R,Publication<A,B>> & List<&2,Publication<A,B>>,streams:Streams<S,A,B,H>,+journal:List<&2,Ev.Batch<Publication<A,B>>>,+tick:Nat,frameStart:Nat) -> Application<S,C,R,A,B,H>:
51	  match observed streams:
52	    case (world,items) Streams{first,second,owners}:
53	      +published=List.drop(&2,Publication<A,B>,items,Ev.size(~Publication<A,B>,journal))
54	      Application{world,Streams{Leaf.append(~S,~A,first,tick,first_values(~A,~B,published)),Leaf.append(~S,~B,second,tick,second_values(~A,~B,published)),owners},Ev.sync_checked(~Publication<A,B>,Nat.is_lt(0n,Ev.count(~Publication<A,B>,published)),journal,tick,published),tick,frameStart}
55	def sync(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,app:Application<S,C,R,A,B,H>) -> Application<S,C,R,A,B,H>:
56	  match app:
57	    case Application{world,streams,journal,+tick,frameStart}:synced(~S,~C,~R,~A,~B,~H,W.events_view(~S,~C,~R,~Publication<A,B>,world),streams,journal,1n+tick,frameStart)
58	
59	# Arbitrary finite products can use this same closed selection operation. The
60	# generated take/put retain every other leaf and arbitrary affine owner H.
61	def observed_read(~S:Data,~E:Data,~F:Type,~put:F -> Leaf.Stream<S,E> -> F,family:F,returned:Leaf.Stream<S,E> & Ev.Reading<E>) -> F & Ev.Reading<E>:
62	  match returned:
63	    case (stream,reading):(put(family,stream),reading)
64	def observed_leaf(~S:Data,~E:Data,~F:Type,~put:F -> Leaf.Stream<S,E> -> F,returned:F & Leaf.Stream<S,E>,+tick:Nat,id:U32) -> F & Ev.Reading<E>:
65	  match returned:
66	    case (family,+stream):observed_read(~S,~E,~F,~put,family,Leaf.reading(~S,~E,stream,id,tick))
67	def observe(~S:Data,~E:Data,~F:Type,~take:F -> F & Leaf.Stream<S,E>,~put:F -> Leaf.Stream<S,E> -> F,family:F,id:U32,tick:Nat) -> F & Ev.Reading<E>:
68	  observed_leaf(~S,~E,~F,~put,take(family),tick,id)
69	
70	def retained_values(~E:Data,remove:Bool,values:List<&2,E>) -> List<&2,E>:
71	  match remove:
72	    case True{}:[]
73	    case False{}:values
74	def retained_record(~A:Data,~B:Data,record:Publication<A,B>,+tick:Nat,+firstDropped:Nat,+secondDropped:Nat) -> Publication<A,B>:
75	  match record:
76	    case Publication{first,second}:Publication{retained_values(~A,Nat.is_le(tick,firstDropped),first),retained_values(~B,Nat.is_le(tick,secondDropped),second)}
77	def nonempty_record(~A:Data,~B:Data,record:Publication<A,B>) -> Bool:
78	  match record:
79	    case Publication{first,second}:Nat.is_lt(0n,Nat.add(Ev.count(~A,first),Ev.count(~B,second)))
80	def retained_head(~A:Data,~B:Data,keep:Bool,record:Publication<A,B>,rest:List<&2,Publication<A,B>>) -> List<&2,Publication<A,B>>:
81	  match keep:
82	    case True{}:record <> rest
83	    case False{}:rest
84	def retained_join(~A:Data,~B:Data,+record:Publication<A,B>,rest:List<&2,Publication<A,B>>) -> List<&2,Publication<A,B>>:
85	  retained_head(~A,~B,nonempty_record(~A,~B,record),record,rest)
86	def retained_records(~A:Data,~B:Data,records:List<&2,Publication<A,B>>,+tick:Nat,+firstDropped:Nat,+secondDropped:Nat) -> List<&2,Publication<A,B>>:
87	  match records:
88	    case []:[]
89	    case record <> rest:retained_join(~A,~B,retained_record(~A,~B,record,tick,firstDropped,secondDropped),retained_records(~A,~B,rest,tick,firstDropped,secondDropped))
90	def retained_batch(~A:Data,~B:Data,keep:Bool,tick:Nat,records:List<&2,Publication<A,B>>,rest:List<&2,Ev.Batch<Publication<A,B>>>) -> List<&2,Ev.Batch<Publication<A,B>>>:
91	  match keep:
92	    case True{}:Ev.Batch{tick,records} <> rest
93	    case False{}:rest
94	def retained_batch_join(~A:Data,~B:Data,tick:Nat,+records:List<&2,Publication<A,B>>,rest:List<&2,Ev.Batch<Publication<A,B>>>) -> List<&2,Ev.Batch<Publication<A,B>>>:
95	  retained_batch(~A,~B,Nat.is_lt(0n,Ev.count(~Publication<A,B>,records)),tick,records,rest)
96	def retained_journal(~A:Data,~B:Data,journal:List<&2,Ev.Batch<Publication<A,B>>>,+firstDropped:Nat,+secondDropped:Nat) -> List<&2,Ev.Batch<Publication<A,B>>>:
97	  match journal:
98	    case []:[]
99	    case Ev.Batch{+tick,records} <> rest:retained_batch_join(~A,~B,tick,retained_records(~A,~B,records,tick,firstDropped,secondDropped),retained_journal(~A,~B,rest,firstDropped,secondDropped))
100	def framed_journal(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,world:W.World<S,C,R,Publication<A,B>>,streams:Streams<S,A,B,H>,+tick:Nat,+journal:List<&2,Ev.Batch<Publication<A,B>>>) -> Application<S,C,R,A,B,H>:
101	  Application{W.events_replace(~S,~C,~R,~Publication<A,B>,world,Ev.values(~Publication<A,B>,journal)),streams,journal,tick,tick}
102	def framed_streams(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,world:W.World<S,C,R,Publication<A,B>>,streams:Streams<S,A,B,H>,tick:Nat,journal:List<&2,Ev.Batch<Publication<A,B>>>) -> Application<S,C,R,A,B,H>:
103	  match streams:
104	    case Streams{+first,+second,owners}:
105	      framed_journal(~S,~C,~R,~A,~B,~H,world,Streams{first,second,owners},tick,retained_journal(~A,~B,journal,Leaf.dropped(~S,~A,first),Leaf.dropped(~S,~B,second)))
106	def frame(~S:Data,~C:Type,~R:Type,~A:Data,~B:Data,~H:Type,app:Application<S,C,R,A,B,H>) -> Application<S,C,R,A,B,H>:
107	  match app:
108	    case Application{world,Streams{first,second,owners},journal,+tick,_}:framed_streams(~S,~C,~R,~A,~B,~H,world,Streams{Leaf.frame(~S,~A,first,tick),Leaf.frame(~S,~B,second,tick),owners},tick,journal)