1	import Base
2	import world.bend as W
3	import event_runtime.bend as Ev
4	
5	# Affine transport owner for two independently retained Data streams. Gameplay
6	# does not receive this owner. The closed classifier belongs to schema setup.
7	type Events<S: Data,E: Data> is Type:
8	  Events{namespace: U32,batches: List<&2,Ev.Batch<E>>,positions: List<&2,Ev.Position>,nextReader: U32,tick: Nat,frameStart: Nat,boundary: Nat,capacity: Nat,dropped: Nat,values: List<&2,E>}
9	type Detached<S: Data,C: Type,R: Type,E: Data> is Type:
10	  Detached{world: W.World<S,C,R,E>,events: Events<S,E>}
11	def choose(~E: Data,selected: Bool,+value: E,+events: List<&2,E>,+removals: List<&2,E>) -> List<&2,E> & List<&2,E>:
12	  match selected:
13	    case True{}: (value <> events,removals)
14	    case False{}: (events,value <> removals)
15	def split_reverse(~E: Data,pair: List<&2,E> & List<&2,E>) -> List<&2,E> & List<&2,E>:
16	  (events,removals) = pair
17	  (List.reverse(&2,E,events),List.reverse(&2,E,removals))
18	def split_loop(~E: Data,~is_event: E -> Bool,values: List<&2,E>,acc: List<&2,E> & List<&2,E>) -> List<&2,E> & List<&2,E>:
19	  match values:
20	    case []: split_reverse(~E,acc)
21	    case +value <> rest:
22	      (events,removals) = acc
23	      split_loop(~E,~is_event,rest,choose(~E,is_event(value),value,events,removals))
24	def split(~E: Data,~is_event: E -> Bool,values: List<&2,E>) -> List<&2,E> & List<&2,E>:
25	  split_loop(~E,~is_event,values,([],[]))
26	def detached_values(~S: Data,~C: Type,~R: Type,~E: Data,+namespace: U32,+batches: List<&2,Ev.Batch<E>>,+positions: List<&2,Ev.Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat,+removals: List<&2,E>,observed: W.World<S,C,R,E> & List<&2,E>) -> Detached<S,C,R,E>:
27	  (world,+values) = observed
28	  Detached{W.events_replace(~S,~C,~R,~E,world,removals),Events{namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped,values}}
29	def detach(~S: Data,~C: Type,~R: Type,~E: Data,runtime: Ev.Runtime<S,C,R,E>,+removals: List<&2,E>) -> Detached<S,C,R,E>:
30	  match runtime:
31	    case Ev.Runtime{world,+ns,+batches,+positions,+next,+tick,+start,+boundary,+capacity,+dropped}: detached_values(~S,~C,~R,~E,ns,batches,positions,next,tick,start,boundary,capacity,dropped,removals,W.events_view(~S,~C,~R,~E,world))
32	def attach(~S: Data,~C: Type,~R: Type,~E: Data,events: Events<S,E>,world: W.World<S,C,R,E>) -> Ev.Runtime<S,C,R,E>:
33	  match events:
34	    case Events{ns,batches,positions,next,tick,start,boundary,capacity,dropped,+values}: Ev.Runtime{W.events_replace(~S,~C,~R,~E,world,values),ns,batches,positions,next,tick,start,boundary,capacity,dropped}
35	# Absorb newly committed ordinary events from a removal-domain World through
36	# the real Event runtime publication synchronization. Deferred barriers use
37	# this same operation; rollback leaves no publication delta to transport.
38	def collected(~S: Data,~C: Type,~R: Type,~E: Data,world: W.World<S,C,R,E>,events: Events<S,E>,partition: List<&2,E> & List<&2,E>) -> Detached<S,C,R,E>:
39	  match events partition:
40	    case Events{+ns,+batches,+positions,+next,+tick,+start,+boundary,+capacity,+dropped,+old}, (+fresh,+removals):
41	      +all = Ev.append(~E,old,fresh)
42	      detach(~S,~C,~R,~E,Ev.sync_values(~S,~C,~R,~E,(W.events_replace(~S,~C,~R,~E,world,all),all),ns,batches,positions,next,1n+tick,start,boundary,capacity,dropped),removals)
43	def collect_view(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,events: Events<S,E>,observed: W.World<S,C,R,E> & List<&2,E>) -> Detached<S,C,R,E>:
44	  (world,+values) = observed
45	  collected(~S,~C,~R,~E,world,events,split(~E,~is_event,values))
46	def collect(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,events: Events<S,E>,world: W.World<S,C,R,E>) -> Detached<S,C,R,E>:
47	  collect_view(~S,~C,~R,~E,~is_event,events,W.events_view(~S,~C,~R,~E,world))
48	
49	# Reusable pre-dispatch seam. The setup callback receives the actual separated
50	# event metadata and complete World owners; no affine owner is copied.
51	def driven(~S: Data,~C: Type,~R: Type,~E: Data,~H: Type,~Out: Type,~body: Events<S,E> -> H -> W.World<S,C,R,E> -> Out,owners: H,result: Detached<S,C,R,E>) -> Out:
52	  match result:
53	    case Detached{world,events}: body(events,owners,world)
54	def drive(~S: Data,~C: Type,~R: Type,~E: Data,~H: Type,~Out: Type,~is_event: E -> Bool,~body: Events<S,E> -> H -> W.World<S,C,R,E> -> Out,events: Events<S,E>,owners: H,world: W.World<S,C,R,E>) -> Out:
55	  driven(~S,~C,~R,~E,~H,~Out,~body,owners,collect(~S,~C,~R,~E,~is_event,events,world))
56	
57	# Event cursors are publication ticks, not list offsets. Removing notice-only
58	# batches therefore retains every real cursor/tick and the original dropped
59	# boundary. Ev.run does not trim; retention is applied only by Ev.frame.
60	def event_batch(~E: Data,~is_event: E -> Bool,+tick: Nat,filtered: List<&2,E> & List<&2,E>,rest: List<&2,Ev.Batch<E>>) -> List<&2,Ev.Batch<E>>:
61	  match filtered:
62	    case ([],_): rest
63	    case (items,_): Ev.Batch{tick,items} <> rest
64	def event_batches_loop(~E: Data,~is_event: E -> Bool,batches: List<&2,Ev.Batch<E>>,acc: List<&2,Ev.Batch<E>>) -> List<&2,Ev.Batch<E>>:
65	  match batches:
66	    case []: List.reverse(&2,Ev.Batch<E>,acc)
67	    case Ev.Batch{tick,items} <> rest: event_batches_loop(~E,~is_event,rest,event_batch(~E,~is_event,tick,split(~E,~is_event,items),acc))
68	def event_batches(~E: Data,~is_event: E -> Bool,batches: List<&2,Ev.Batch<E>>) -> List<&2,Ev.Batch<E>>:
69	  event_batches_loop(~E,~is_event,batches,[])
70	def recovered(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,world: W.World<S,C,R,E>,+namespace: U32,+batches: List<&2,Ev.Batch<E>>,+positions: List<&2,Ev.Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat,+old_removals: List<&2,E>,partition: List<&2,E> & List<&2,E>) -> Detached<S,C,R,E>:
71	  match partition:
72	    case (+values,+fresh_removals): Detached{W.events_replace(~S,~C,~R,~E,world,Ev.append(~E,old_removals,fresh_removals)),Events{namespace,event_batches(~E,~is_event,batches),positions,nextReader,tick,frameStart,boundary,capacity,dropped,values}}
73	def recovery_view(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,+namespace: U32,+batches: List<&2,Ev.Batch<E>>,+positions: List<&2,Ev.Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat,+old_removals: List<&2,E>,observed: W.World<S,C,R,E> & List<&2,E>) -> Detached<S,C,R,E>:
74	  (world,values) = observed
75	  recovered(~S,~C,~R,~E,~is_event,world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped,old_removals,split(~E,~is_event,values))
76	def recover(~S: Data,~C: Type,~R: Type,~E: Data,~is_event: E -> Bool,runtime: Ev.Runtime<S,C,R,E>,+old_removals: List<&2,E>) -> Detached<S,C,R,E>:
77	  match runtime:
78	    case Ev.Runtime{world,+namespace,+batches,+positions,+nextReader,+tick,+frameStart,+boundary,+capacity,+dropped}: recovery_view(~S,~C,~R,~E,~is_event,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped,old_removals,W.events_view(~S,~C,~R,~E,world))