1	import Base
2	import machine.bend as M
3	import event_runtime.bend as Ev
4	import machine_stream_queue.bend as Q
5	
6	# FIFO metadata: front oldest-first, back newest-first; size counts events.
7	type Stream<V: Data> is Data:
8	  Stream{queue: Q.Queue<M.Transition<V>>,positions: List<&2,Ev.Position>,frameStart: Nat}
9	type Reader<S: Data,M: Data> is Type:
10	  Reader{namespace: U32,id: U32}
11	
12	def empty(~V: Data) -> Stream<V>: Stream{Q.empty(~M.Transition<V>),[],0n}
13	def publish(~V: Data,stream: Stream<V>,tick: Nat,event: M.Transition<V>) -> Stream<V>:
14	  match stream:
15	    case Stream{queue,positions,frameStart}: Stream{Q.publish(~M.Transition<V>,queue,tick,[event]),positions,frameStart}
16	def activate(~V: Data,stream: Stream<V>,id: U32,tick: Nat) -> Stream<V>:
17	  match stream:
18	    case Stream{queue,positions,frameStart}: Stream{queue,Ev.activate(positions,id,tick),frameStart}
19	def complete(~V: Data,stream: Stream<V>,id: U32,tick: Nat,success: Bool) -> Stream<V>:
20	  match stream success:
21	    case Stream{queue,positions,frameStart} True{}: Stream{queue,Ev.update(positions,id,tick,False{}),frameStart}
22	    case stream False{}: stream
23	# A condition-false system never executes its reader parameter.
24	def skip(~V: Data,stream: Stream<V>,id: U32,tick: Nat) -> Stream<V>:
25	  stream
26	
27	# Bevy insert_state replaces messages while preserving reader positions and loss.
28	def initial_queue(~V:Data,queue:Q.Queue<M.Transition<V>>,tick:Nat,value:V) -> Q.Queue<M.Transition<V>>:
29	  match queue:
30	    case Q.Queue{_,_,_,dropped,_}:Q.Queue{[],[Q.Batch{tick,1n,[M.Initial{value}]}],1n,dropped,1n}
31	def initial(~V:Data,stream:Stream<V>,tick:Nat,value:V) -> Stream<V>:
32	  match stream:
33	    case Stream{queue,positions,frameStart}:Stream{initial_queue(~V,queue,tick,value),positions,frameStart}
34	
35	def observe_position(~V: Data,position: Maybe<Ev.Position>,batches: List<&2,Ev.Batch<M.Transition<V>>>,dropped: Nat,id: U32,tick: Nat) -> Ev.Reading<M.Transition<V>>:
36	  match position:
37	    case None{}: Ev.reading(~M.Transition<V>,Ev.Position{id,0n,tick},batches,dropped)
38	    case Some{position}: Ev.reading(~M.Transition<V>,position,batches,dropped)
39	
40	def observe_registered(~V: Data,position: Ev.Position,+queue: Q.Queue<M.Transition<V>>) -> Ev.Reading<M.Transition<V>>:
41	  match position queue:
42	    case Ev.Position{_,+cursor,registered} Q.Queue{+front,+back,+size,+dropped,+batches}: Ev.Reading{Q.read(~M.Transition<V>,Q.Queue{front,back,size,dropped,batches},cursor),Nat.is_lt(Nat.max(cursor,registered),dropped)}
43	def observe_queue_position(~V: Data,position: Maybe<Ev.Position>,queue: Q.Queue<M.Transition<V>>,id: U32,tick: Nat) -> Ev.Reading<M.Transition<V>>:
44	  match position:
45	    case None{}: observe_registered(~V,Ev.Position{id,0n,tick},queue)
46	    case Some{position}: observe_registered(~V,position,queue)
47	def observe(~V: Data,stream: Stream<V>,+id: U32,tick: Nat) -> Ev.Reading<M.Transition<V>>:
48	  match stream:
49	    case Stream{queue,positions,_}: observe_queue_position(~V,Ev.lookup(positions,id),queue,id,tick)
50	def batches_into_queue(~V:Data,batches:List<&2,Ev.Batch<M.Transition<V>>>,queue:Q.Queue<M.Transition<V>>) -> Q.Queue<M.Transition<V>>:
51	  match batches:
52	    case []:queue
53	    case Ev.Batch{tick,+items} <> rest:batches_into_queue(~V,rest,Q.publish_nonempty(~M.Transition<V>,queue,tick,items,Q.count_loop(~M.Transition<V>,items,0n)))
54	def trimmed(~V:Data,frameStart:Nat,positions:List<&2,Ev.Position>,result:Ev.Trimmed<M.Transition<V>>) -> Stream<V>:
55	  match result:
56	    case Ev.Trimmed{batches,dropped}:Stream{batches_into_queue(~V,batches,Q.Queue{[],[],0n,dropped,0n}),positions,frameStart}
57	
58	def frame(~V: Data,stream: Stream<V>,tick: Nat) -> Stream<V>:
59	  match stream:
60	    case Stream{queue,+positions,frameStart}: Stream{Q.trim(~M.Transition<V>,queue,Ev.hold_boundary(positions,frameStart),65536n),positions,tick}
61	def dispose(~V: Data,stream: Stream<V>,id: U32) -> Stream<V>:
62	  match stream:
63	    case Stream{queue,positions,frameStart}: Stream{queue,Ev.update(positions,id,0n,True{}),frameStart}
64	
65	# Canonical snapshot includes every retained batch/payload, without changing cursor.
66	def batches_loop(~V: Data,batches: List<&2,Q.Batch<M.Transition<V>>>,acc: List<&2,Ev.Batch<M.Transition<V>>>) -> List<&2,Ev.Batch<M.Transition<V>>>:
67	  match batches:
68	    case []: List.reverse(&2,Ev.Batch<M.Transition<V>>,acc)
69	    case Q.Batch{tick,_,items} <> rest: batches_loop(~V,rest,Ev.Batch{tick,items} <> acc)
70	def snapshot_queue(~V: Data,queue: Q.Queue<M.Transition<V>>,positions: List<&2,Ev.Position>,frameStart: Nat) -> List<&2,Ev.Batch<M.Transition<V>>> & (List<&2,Ev.Position> & (Nat & Nat)):
71	  match queue:
72	    case Q.Queue{front,back,_,dropped,_}: (batches_loop(~V,Ev.append(~Q.Batch<M.Transition<V>>,front,List.reverse(&2,Q.Batch<M.Transition<V>>,back)),[]),(positions,(dropped,frameStart)))
73	def snapshot(~V: Data,stream: Stream<V>) -> List<&2,Ev.Batch<M.Transition<V>>> & (List<&2,Ev.Position> & (Nat & Nat)):
74	  match stream:
75	    case Stream{queue,positions,frameStart}: snapshot_queue(~V,queue,positions,frameStart)
76	
77	# Engine entry points consume and return the real affine reader capability.
78	# A foreign-world refusal preserves both owners and never activates a cursor.
79	type OpenResult<S: Data,M: Data,V: Data> is Type:
80	  Opened{reader: Reader<S,M>,stream: Stream<V>,reading: Ev.Reading<M.Transition<V>>}
81	  ReaderRefused{reader: Reader<S,M>,stream: Stream<V>}
82	
83	def open_checked(~S: Data,~M: Data,~V: Data,allowed: Bool,reader: Reader<S,M>,+stream: Stream<V>,+tick: Nat) -> OpenResult<S,M,V>:
84	  match allowed reader:
85	    case False{}, reader: ReaderRefused{reader,stream}
86	    case True{}, Reader{namespace,+id}:
87	      Opened{Reader{namespace,id},activate(~V,stream,id,tick),observe(~V,stream,id,tick)}
88	
89	def open(~S: Data,~M: Data,~V: Data,reader: Reader<S,M>,stream: Stream<V>,namespace: U32,tick: Nat) -> OpenResult<S,M,V>:
90	  match reader:
91	    case Reader{+owner,id}: open_checked(~S,~M,~V,U32.is_eq(owner,namespace),Reader{owner,id},stream,tick)
92	
93	def finish(~S: Data,~M: Data,~V: Data,reader: Reader<S,M>,stream: Stream<V>,tick: Nat,success: Bool) -> Reader<S,M> & Stream<V>:
94	  match reader:
95	    case Reader{namespace,+id}: (Reader{namespace,id},complete(~V,stream,id,tick,success))
96	
97	def skip_owned(~S: Data,~M: Data,~V: Data,reader: Reader<S,M>,stream: Stream<V>,tick: Nat) -> Reader<S,M> & Stream<V>:
98	  match reader:
99	    case Reader{namespace,+id}: (Reader{namespace,id},skip(~V,stream,id,tick))