owned_event_storage.bend
bendvy-ecs-core@0.1.0.1 · 78 lines · 0 definitions
1import Base
2import bendvy-ecs-core@0.1.0.1/world.bend as W
3import bendvy-ecs-core@0.1.0.1/world_resource_owner.bend as Resource
4import bendvy-ecs-core@0.1.0.1/event_runtime.bend as Ev
5import bendvy-ecs-core@0.1.0.1/capabilities.bend as Cap
6import bendvy-ecs-core@0.1.0.1/owned_event_log.bend as Log
7
8# Trusted schema storage: user resources and each affine payload log have one owner.
9# Data IDs use the existing event runtime; its cursor/lag/retention policy is unchanged.
10type Resources<R:Type,P:Type> is Type:
11 Resources{resource:R,log:Log.Log<P>}
12def World(~S:Data,~C:Type,~R:Type,~P:Type) -> Type:W.World<S,C,Resources<R,P>,Nat>
13def Runtime(~S:Data,~C:Type,~R:Type,~P:Type) -> Type:Ev.Runtime<S,C,Resources<R,P>,Nat>
14
15# Canonical World resource splitting preserves every non-resource field.
16def resources_returned(~S:Data,~C:Type,~R:Type,~P:Type,~Out:Type,rest:Resource.Rest<S,C,Resources<R,P>,Nat>,returned:Resources<R,P> & Out) -> World(~S,~C,~R,~P) & Out:
17 match returned:
18 case (resources,output):(Resource.put(~S,~C,~Resources<R,P>,~Nat,rest,resources),output)
19def resources_taken(~S:Data,~C:Type,~R:Type,~P:Type,~Out:Type,effect:Resources<R,P> -> Resources<R,P> & Out,taken:Resource.Rest<S,C,Resources<R,P>,Nat> & Resources<R,P>) -> World(~S,~C,~R,~P) & Out:
20 match taken:
21 case (rest,resources):resources_returned(~S,~C,~R,~P,~Out,rest,effect(resources))
22def with_resources(~S:Data,~C:Type,~R:Type,~P:Type,~Out:Type,effect:Resources<R,P> -> Resources<R,P> & Out,world:World(~S,~C,~R,~P)) -> World(~S,~C,~R,~P) & Out:
23 resources_taken(~S,~C,~R,~P,~Out,effect,Resource.take(~S,~C,~Resources<R,P>,~Nat,world))
24
25def appended(~R:Type,~P:Type,resource:R,result:Log.Log<P> & List<&2,Nat>) -> Resources<R,P> & List<&2,Nat>:
26 match result:
27 case (log,ids):(Resources{resource,log},ids)
28def append_resources(~R:Type,~P:Type,values:List<P>,resources:Resources<R,P>) -> Resources<R,P> & List<&2,Nat>:
29 match resources:
30 case Resources{resource,log}:appended(~R,~P,resource,Log.append_many(~P,log,values))
31def publish_ids(~S:Data,~C:Type,~R:Type,~P:Type,result:World(~S,~C,~R,~P) & List<&2,Nat>) -> World(~S,~C,~R,~P):
32 match result:
33 case (world,ids):W.event_publish(~S,~C,~Resources<R,P>,~Nat,world,ids)
34# Called only after successful canonical transaction finish: owners and IDs publish
35# in the same FIFO order. Failure/abort transport remains the ordinary adapter's job.
36def commit_owned(~S:Data,~C:Type,~R:Type,~P:Type,world:World(~S,~C,~R,~P),values:List<P>) -> World(~S,~C,~R,~P):
37 publish_ids(~S,~C,~R,~P,with_resources(~S,~C,~R,~P,~List<&2,Nat>,resources => append_resources(~R,~P,values,resources),world))
38
39def retained(~R:Type,~P:Type,resource:R,result:Log.Log<P> & List<Log.Row<P>>) -> Resources<R,P> & List<Log.Row<P>>:
40 match result:
41 case (log,retired):(Resources{resource,log},retired)
42def retain_resources(~R:Type,~P:Type,ids:List<&2,Nat>,resources:Resources<R,P>) -> Resources<R,P> & List<Log.Row<P>>:
43 match resources:
44 case Resources{resource,log}:retained(~R,~P,resource,Log.retain(~P,log,ids))
45def drain_resources(~R:Type,~P:Type,resources:Resources<R,P>) -> Resources<R,P> & List<Log.Row<P>>:
46 match resources:
47 case Resources{resource,log}:retained(~R,~P,resource,Log.drain(~P,log))
48def runtime_returned(~S:Data,~C:Type,~R:Type,~P:Type,~Out:Type,namespace:U32,batches:List<&2,Ev.Batch<Nat>>,positions:List<&2,Ev.Position>,nextReader:U32,tick:Nat,frameStart:Nat,boundary:Nat,capacity:Nat,dropped:Nat,result:World(~S,~C,~R,~P) & Out) -> Runtime(~S,~C,~R,~P) & Out:
49 match result:
50 case (world,output):(Ev.Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped},output)
51def framed(~S:Data,~C:Type,~R:Type,~P:Type,runtime:Runtime(~S,~C,~R,~P)) -> Runtime(~S,~C,~R,~P) & List<Log.Row<P>>:
52 match runtime:
53 case Ev.Runtime{world,namespace,+batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped}:
54 runtime_returned(~S,~C,~R,~P,~List<Log.Row<P>>,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped,with_resources(~S,~C,~R,~P,~List<Log.Row<P>>,resources => retain_resources(~R,~P,Ev.values(~Nat,batches),resources),world))
55# Existing Ev.frame chooses surviving complete batches. Return every other payload
56# owner once; the caller chooses reuse or release, with no automatic IO finalizer.
57def frame(~S:Data,~C:Type,~R:Type,~P:Type,runtime:Runtime(~S,~C,~R,~P)) -> Runtime(~S,~C,~R,~P) & List<Log.Row<P>>:
58 framed(~S,~C,~R,~P,Ev.frame(~S,~C,~Resources<R,P>,~Nat,runtime))
59def cleared_events(~S:Data,~C:Type,~R:Type,~P:Type,result:World(~S,~C,~R,~P) & List<Log.Row<P>>) -> World(~S,~C,~R,~P) & List<Log.Row<P>>:
60 match result:
61 case (world,rows):(W.events_replace(~S,~C,~Resources<R,P>,~Nat,world,[]),rows)
62# Shutdown clears only queued IDs/batches and log rows. Preserve reader/runtime
63# metadata and Log.next so old IDs cannot alias a later appended payload.
64def drain(~S:Data,~C:Type,~R:Type,~P:Type,runtime:Runtime(~S,~C,~R,~P)) -> Runtime(~S,~C,~R,~P) & List<Log.Row<P>>:
65 match runtime:
66 case Ev.Runtime{world,namespace,_,positions,nextReader,tick,frameStart,boundary,capacity,dropped}:
67 runtime_returned(~S,~C,~R,~P,~List<Log.Row<P>>,namespace,[],positions,nextReader,tick,frameStart,boundary,capacity,dropped,cleared_events(~S,~C,~R,~P,with_resources(~S,~C,~R,~P,~List<Log.Row<P>>,drain_resources(~R,~P),world)))
68
69def read_returned(~R:Type,~P:Type,~Args:Type,~Observation:Type,resource:R,result:Log.Log<P> & Log.ReadResult<Args,Observation>) -> Resources<R,P> & Log.ReadResult<Args,Observation>:
70 match result:
71 case (log,reading):(Resources{resource,log},reading)
72def read_resources(~R:Type,~P:Type,~Request:Type,~Projection:Data,~Args:Type,~Observation:Type,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & Observation,resources:Resources<R,P>,id:Nat,args:Args) -> Resources<R,P> & Log.ReadResult<Args,Observation>:
73 match resources:
74 case Resources{resource,log}:read_returned(~R,~P,~Args,~Observation,resource,Log.read(~P,~Request,~Projection,~Args,~Observation,~at,~body,log,id,args))
75# The projection is trusted schema provisioning. The closed body receives only an
76# abstract payload handle and declared scoped read request; it never gets World.
77def read(~S:Data,~C:Type,~R:Type,~P:Type,~Request:Type,~Projection:Data,~Args:Type,~Observation:Type,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & Observation,world:World(~S,~C,~R,~P),id:Nat,args:Args) -> World(~S,~C,~R,~P) & Log.ReadResult<Args,Observation>:
78 with_resources(~S,~C,~R,~P,~Log.ReadResult<Args,Observation>,resources => read_resources(~R,~P,~Request,~Projection,~Args,~Observation,~at,~body,resources,id,args),world)