1	import Base
2	import ./event_runtime.bend as Ev
3	import ./ordinary_event_declaration.bend as Event
4	
5	# Detached Data metadata; the containing application retains the sole affine World.
6	# Detached typed stream metadata: ordered Data batches and activated reader
7	# positions with retention accounting. It owns no World or affine payload log;
8	# raw construction/selection belongs to trusted family provisioning.
9	type Stream<S:Data,E:Data> is Data:
10	  Stream{domain:Event.Domain<S,E>,batches:List<&2,Ev.Batch<E>>,positions:List<&2,Ev.Position>,nextReader:U32,frameStart:Nat,boundary:Nat,capacity:Nat,dropped:Nat}
11	# Initialize an empty stream for a declared domain/capacity. No reader is active.
12	def provision(~S:Data,~E:Data,domain:Event.Domain<S,E>,capacity:Nat) -> Stream<S,E>:
13	  Stream{domain,[],[],1,0n,0n,capacity,0n}
14	# Check remaining U32 reader-id space, without allocating or activating a reader.
15	def can_register(~S:Data,~E:Data,stream:Stream<S,E>) -> Bool:
16	  match stream:
17	    case Stream{_,_,_,nextReader,_,_,_,_}:U32.is_lt(nextReader,4294967295)
18	def increment_reader(~S:Data,~E:Data,stream:Stream<S,E>,+tick:Nat) -> Stream<S,E> & U32:
19	  match stream:
20	    case Stream{domain,batches,positions,+nextReader,frameStart,boundary,capacity,dropped}:(Stream{domain,batches,positions,(nextReader+1:U32),frameStart,boundary,capacity,dropped},nextReader)
21	def registered_checked(~S:Data,~E:Data,available:Bool,stream:Stream<S,E>,+tick:Nat) -> Stream<S,E> & U32:
22	  match available:
23	    case False{}:(stream,0)
24	    case True{}:increment_reader(~S,~E,stream,tick)
25	# Append nonempty Data publications at the supplied tick using runtime batch rules.
26	# Does not advance readers or trim capacity; caller supplies ordered publication data.
27	def append(~S:Data,~E:Data,stream:Stream<S,E>,tick:Nat,+values:List<&2,E>) -> Stream<S,E>:
28	  match stream:
29	    case Stream{domain,+batches,positions,nextReader,frameStart,boundary,capacity,dropped}:Stream{domain,Ev.sync_checked(~E,Nat.is_lt(0n,Ev.count(~E,values)),batches,tick,values),positions,nextReader,frameStart,boundary,capacity,dropped}
30	# Allocate the next reader id, returning zero on exhaustion with stream unchanged.
31	# Does not activate its position or deliver unread data; routing checks admission
32	# before using this low-level result.
33	def registered(~S:Data,~E:Data,+stream:Stream<S,E>,+tick:Nat) -> Stream<S,E> & U32:
34	  registered_checked(~S,~E,can_register(~S,~E,stream),stream,tick)
35	def read_position(~S:Data,~E:Data,stream:Stream<S,E>,+id:U32,+tick:Nat,position:Maybe<Ev.Position>) -> Stream<S,E> & Ev.Reading<E>:
36	  match stream position:
37	    case Stream{domain,+batches,positions,nextReader,frameStart,boundary,capacity,+dropped} Some{position}:(Stream{domain,batches,positions,nextReader,frameStart,boundary,capacity,dropped},Ev.reading(~E,position,batches,dropped))
38	    case Stream{domain,+batches,positions,nextReader,frameStart,boundary,capacity,+dropped} None{}:(Stream{domain,batches,positions,nextReader,frameStart,boundary,capacity,dropped},Ev.reading(~E,Ev.Position{id,0n,tick},batches,dropped))
39	# Read ordered unread Data plus lag status using an existing position, or a
40	# not-yet-activated position for id/tick. Returns stream metadata without advancing
41	# its cursor. This helper does not validate an external reader token or World.
42	def reading(~S:Data,~E:Data,+stream:Stream<S,E>,+id:U32,+tick:Nat) -> Stream<S,E> & Ev.Reading<E>:
43	  match stream:
44	    case Stream{+domain,+batches,+positions,+nextReader,+frameStart,+boundary,+capacity,+dropped}:read_position(~S,~E,Stream{domain,batches,positions,nextReader,frameStart,boundary,capacity,dropped},id,tick,Ev.lookup(positions,id))
45	# Activate id if needed then advance its cursor to tick. Trusted completion
46	# operation, not an independent registration/provenance validation.
47	def advance(~S:Data,~E:Data,stream:Stream<S,E>,+id:U32,+tick:Nat) -> Stream<S,E>:
48	  match stream:
49	    case Stream{domain,batches,positions,nextReader,frameStart,boundary,capacity,dropped}:Stream{domain,batches,Ev.update(Ev.activate(positions,id,Nat.sub(tick,1n)),id,tick,False{}),nextReader,frameStart,boundary,capacity,dropped}
50	def trimmed(~S:Data,~E:Data,domain:Event.Domain<S,E>,positions:List<&2,Ev.Position>,nextReader:U32,frameStart:Nat,boundary:Nat,capacity:Nat,trimmed:Ev.Trimmed<E>) -> Stream<S,E>:
51	  match trimmed:
52	    case Ev.Trimmed{batches,dropped}:Stream{domain,batches,positions,nextReader,frameStart,boundary,capacity,dropped}
53	# Apply frame-window and capacity trimming with current active-reader holds.
54	# Return metadata with updated boundaries/dropped history; hard capacity can cause
55	# lag, so active readers do not promise indefinite retention.
56	def frame(~S:Data,~E:Data,stream:Stream<S,E>,+tick:Nat) -> Stream<S,E>:
57	  match stream:
58	    case Stream{domain,batches,+positions,nextReader,+frameStart,_,+capacity,dropped}:trimmed(~S,~E,domain,positions,nextReader,tick,frameStart,capacity,Ev.trim_capacity(~E,Ev.past_window(~E,batches,Ev.hold_boundary(positions,frameStart),dropped),capacity))
59	# Return the detached authored domain key without reading message payloads.
60	def inventory(~S:Data,~E:Data,stream:Stream<S,E>) -> String:
61	  match stream:
62	    case Stream{Event.Domain{key},_,_,_,_,_,_,_}:key
63	# Return the dropped-through history marker, not a message count.
64	def dropped(~S:Data,~E:Data,stream:Stream<S,E>) -> Nat:
65	  match stream:
66	    case Stream{_,_,_,_,_,_,_,value}:value
67	def completed_positions(success:Bool,+positions:List<&2,Ev.Position>,+id:U32,+tick:Nat) -> List<&2,Ev.Position>:
68	  match success:
69	    case False{}:Ev.activate(positions,id,Nat.sub(tick,1n))
70	    case True{}:Ev.update(Ev.activate(positions,id,Nat.sub(tick,1n)),id,tick,False{})
71	# Update one reader after execution: success advances to tick; failure preserves
72	# unread position while activating absent position under failed-reader rules.
73	# Caller supplies actual id/outcome; this does not execute or validate a callback.
74	def completed(~S:Data,~E:Data,stream:Stream<S,E>,id:U32,tick:Nat,success:Bool) -> Stream<S,E>:
75	  match stream:
76	    case Stream{domain,batches,positions,nextReader,frameStart,boundary,capacity,dropped}:Stream{domain,batches,completed_positions(success,positions,id,tick),nextReader,frameStart,boundary,capacity,dropped}
77	# Derive read-mode declaration from the retained stream domain, not a new writer grant.
78	def read_declaration(~S:Data,~E:Data,stream:Stream<S,E>) -> Event.Declaration<S,E>:
79	  match stream:
80	    case Stream{domain,_,_,_,_,_,_,_}:Event.read(~S,~E,domain)
81	# Project the retained typed domain; metadata alone is not reader ownership.
82	def domain(~S:Data,~E:Data,stream:Stream<S,E>) -> Event.Domain<S,E>:
83	  match stream:
84	    case Stream{domain,_,_,_,_,_,_,_}:domain