event_stream.bend

bendvy-ecs-core@0.1.0.4 · 84 lines · 18 definitions

1import Base
2import bendvy-ecs-core@0.1.0.4/event_runtime.bend as Ev
3import bendvy-ecs-core@0.1.0.4/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.
9type 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.
12def 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.
15def can_register(~S:Data,~E:Data,stream:Stream<S,E>) -> Bool:
16  match stream:
17    case Stream{_,_,_,nextReader,_,_,_,_}:U32.is_lt(nextReader,4294967295)
18def 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)
21def 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.
27def 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.
33def 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)
35def 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.
42def 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.
47def 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}
50def 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.
56def 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.
60def 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.
64def dropped(~S:Data,~E:Data,stream:Stream<S,E>) -> Nat:
65  match stream:
66    case Stream{_,_,_,_,_,_,_,value}:value
67def 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.
74def 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.
78def 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.
82def domain(~S:Data,~E:Data,stream:Stream<S,E>) -> Event.Domain<S,E>:
83  match stream:
84    case Stream{domain,_,_,_,_,_,_,_}:domain