1	import Base
2	import ./capabilities.bend as Cap
3	import ./owned_event_scope.bend as Scope
4	# Sequential identities belong to the log; payloads are never copied.
5	type Row<P:Type> is Type:Row{id:Nat,payload:P}
6	type Log<P:Type> is Type:Log{next:Nat,rows:List<Row<P>>}
7	type ReadResult<Args:Type,Observation:Type> is Type:
8	  Found{observation:Observation}
9	  Missing{args:Args}
10	def empty(~P:Type) -> Log<P>:Log{1n,[]}
11	def append(~P:Type,log:Log<P>,payload:P) -> Log<P> & Nat:
12	  match log:
13	    case Log{+next,rows}:(Log{1n+next,List.append(&1,Row<P>,rows,[Row{next,payload}])},next)
14	def appended_many(~P:Type,id:Nat,result:Log<P> & List<&2,Nat>) -> Log<P> & List<&2,Nat>:
15	  match result:
16	    case (log,ids):(log,id <> ids)
17	def append_remaining(~P:Type,next:Log<P> -> Log<P> & List<&2,Nat>,result:Log<P> & Nat) -> Log<P> & List<&2,Nat>:
18	  match result:
19	    case (log,id):appended_many(~P,id,next(log))
20	def append_rows(~P:Type,payloads:List<P>,log:Log<P>) -> Log<P> & List<&2,Nat>:
21	  match payloads:
22	    case []:(log,[])
23	    case payload <> rest:append_remaining(~P,next => append_rows(~P,rest,next),append(~P,log,payload))
24	def append_many(~P:Type,log:Log<P>,payloads:List<P>) -> Log<P> & List<&2,Nat>:
25	  append_rows(~P,payloads,log)
26	def contains(+ids:List<&2,Nat>,+id:Nat) -> Bool:
27	  match ids:
28	    case []:False{}
29	    case head <> rest:Bool.or(Nat.is_eq(head,id),contains(rest,id))
30	def partitioned(~P:Type,keep:Bool,row:Row<P>,result:List<Row<P>> & List<Row<P>>) -> List<Row<P>> & List<Row<P>>:
31	  match keep result:
32	    case True{}, (retained,retired):(row <> retained,retired)
33	    case False{}, (retained,retired):(retained,row <> retired)
34	def partition(~P:Type,rows:List<Row<P>>,+ids:List<&2,Nat>) -> List<Row<P>> & List<Row<P>>:
35	  match rows:
36	    case []:([],[])
37	    case Row{+id,payload} <> rest:partitioned(~P,contains(ids,id),Row{id,payload},partition(~P,rest,ids))
38	def retired(~P:Type,next:Nat,result:List<Row<P>> & List<Row<P>>) -> Log<P> & List<Row<P>>:
39	  match result:
40	    case (retained,removed):(Log{next,retained},removed)
41	def retire(~P:Type,log:Log<P>,ids:List<&2,Nat>) -> Log<P> & List<Row<P>>:
42	  match log:
43	    case Log{next,rows}:retired(~P,next,partition(~P,rows,ids))
44	def retain(~P:Type,log:Log<P>,ids:List<&2,Nat>) -> Log<P> & List<Row<P>>:
45	  retire(~P,log,ids)
46	def drain(~P:Type,log:Log<P>) -> Log<P> & List<Row<P>>:
47	  match log:
48	    case Log{next,rows}:(Log{next,[]},rows)
49	def read_returned(~P:Type,~Args:Type,~Observation:Type,id:Nat,tail:List<Row<P>>,observed:P & Observation) -> List<Row<P>> & ReadResult<Args,Observation>:
50	  match observed:
51	    case (payload,value):(Row{id,payload} <> tail,Found{value})
52	def read_rebuilt(~P:Type,~Args:Type,~Observation:Type,row:Row<P>,result:List<Row<P>> & ReadResult<Args,Observation>) -> List<Row<P>> & ReadResult<Args,Observation>:
53	  match result:
54	    case (tail,result):(row <> tail,result)
55	def read_selected(~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,next:List<Row<P>> -> Args -> List<Row<P>> & ReadResult<Args,Observation>,selected:Bool,id:Nat,payload:P,tail:List<Row<P>>,args:Args) -> List<Row<P>> & ReadResult<Args,Observation>:
56	  match selected:
57	    case True{}:read_returned(~P,~Args,~Observation,id,tail,Scope.provide(~P,~Request,~Projection,~Args,~Observation,~at,~body,payload,args))
58	    case False{}:read_rebuilt(~P,~Args,~Observation,Row{id,payload},next(tail,args))
59	def read_rows(~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,fuel:Nat,rows:List<Row<P>>,+wanted:Nat,args:Args) -> List<Row<P>> & ReadResult<Args,Observation>:
60	  match fuel rows:
61	    case _, []:([],Missing{args})
62	    case 0n, rows:(rows,Missing{args})
63	    case 1n+fuel, Row{+id,payload} <> tail:read_selected(~P,~Request,~Projection,~Args,~Observation,~at,~body,rows => args => read_rows(~P,~Request,~Projection,~Args,~Observation,~at,~body,fuel,rows,wanted,args),Nat.is_eq(id,wanted),id,payload,tail,args)
64	def counted_row(~P:Type,row:Row<P>,result:List<Row<P>> & Nat) -> List<Row<P>> & Nat:
65	  match result:
66	    case (rows,count):(row <> rows,1n+count)
67	def count_rows(~P:Type,rows:List<Row<P>>) -> List<Row<P>> & Nat:
68	  match rows:
69	    case []:([],0n)
70	    case row <> rest:counted_row(~P,row,count_rows(~P,rest))
71	def read_finished(~P:Type,~Args:Type,~Observation:Type,next:Nat,result:List<Row<P>> & ReadResult<Args,Observation>) -> Log<P> & ReadResult<Args,Observation>:
72	  match result:
73	    case (rows,result):(Log{next,rows},result)
74	def read_counted(~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,next:Nat,id:Nat,args:Args,counted:List<Row<P>> & Nat) -> Log<P> & ReadResult<Args,Observation>:
75	  match counted:
76	    case (rows,count):read_finished(~P,~Args,~Observation,next,read_rows(~P,~Request,~Projection,~Args,~Observation,~at,~body,count,rows,id,args))
77	def read(~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,log:Log<P>,id:Nat,args:Args) -> Log<P> & ReadResult<Args,Observation>:
78	  match log:
79	    case Log{next,rows}:read_counted(~P,~Request,~Projection,~Args,~Observation,~at,~body,next,id,args,count_rows(~P,rows))