1	import Base
2	import queue.bend as Q
3	import ../../event_runtime.bend as Ev
4	
5	# Schema-owned descriptor IDs; a key's readers never hold another key's log.
6	type DescriptorKey<S: Data> is Data:
7	  DeclaredKey{id: U32}
8	type Domain<S: Data,E: Data> is Data:
9	  Domain{key: DescriptorKey<S>,queue: Q.Queue<E>,readers: List<&2,U32>}
10	
11	def key_equal(~S: Data,left: DescriptorKey<S>,right: DescriptorKey<S>) -> Bool:
12	  match left right:
13	    case DeclaredKey{a} DeclaredKey{b}: U32.is_eq(a,b)
14	
15	def member(ids: List<&2,U32>,+wanted: U32) -> Bool:
16	  match ids:
17	    case []: False{}
18	    case id <> rest: Bool.or(U32.is_eq(id,wanted),member(rest,wanted))
19	
20	def add_checked(found: Bool,ids: List<&2,U32>,id: U32) -> List<&2,U32>:
21	  match found:
22	    case True{}: ids
23	    case False{}: id <> ids
24	
25	def add(+ids: List<&2,U32>,+id: U32) -> List<&2,U32>:
26	  add_checked(member(ids,id),ids,id)
27	
28	def select_boundary(registered: Bool,+cursor: Nat,boundary: Nat) -> Nat:
29	  match registered:
30	    case False{}: boundary
31	    case True{}: Nat.min(cursor,boundary)
32	
33	def boundary(positions: List<&2,Ev.Position>,+ids: List<&2,U32>,acc: Nat) -> Nat:
34	  match positions:
35	    case []: acc
36	    case Ev.Position{id,cursor,_} <> rest: boundary(rest,ids,select_boundary(member(ids,id),cursor,acc))
37	
38	def trim_domain(~S: Data,~E: Data,domain: Domain<S,E>,+positions: List<&2,Ev.Position>,fallback: Nat,capacity: Nat) -> Domain<S,E>:
39	  match domain:
40	    case Domain{key,queue,+ids}: Domain{key,Q.trim(~E,queue,boundary(positions,ids,fallback),capacity),ids}
41	
42	def trim(~S: Data,~E: Data,domains: List<&2,Domain<S,E>>,+positions: List<&2,Ev.Position>,+fallback: Nat,+capacity: Nat) -> List<&2,Domain<S,E>>:
43	  match domains:
44	    case []: []
45	    case domain <> rest: trim_domain(~S,~E,domain,positions,fallback,capacity) <> trim(~S,~E,rest,positions,fallback,capacity)
46	
47	def activate_domain(~S: Data,~E: Data,same: Bool,domain: Domain<S,E>,id: U32) -> Domain<S,E>:
48	  match same domain:
49	    case False{} domain: domain
50	    case True{} Domain{key,queue,ids}: Domain{key,queue,add(ids,id)}
51	
52	def activate_walk(~S: Data,~E: Data,same: Bool,domain: Domain<S,E>,+rest: List<&2,Domain<S,E>>,id: U32,ih: Unit -> List<&2,Domain<S,E>>) -> List<&2,Domain<S,E>>:
53	  match same:
54	    case True{}: activate_domain(~S,~E,True{},domain,id) <> rest
55	    case False{}: domain <> ih(Unit{})
56	
57	def activate_one(~S: Data,~E: Data,domains: List<&2,Domain<S,E>>,+wanted: DescriptorKey<S>,+id: U32) -> List<&2,Domain<S,E>>:
58	  match domains:
59	    case []: Domain{wanted,Q.empty(~E),[id]} <> []
60	    case Domain{+key,+queue,+ids} <> +rest: activate_walk(~S,~E,key_equal(~S,key,wanted),Domain{key,queue,ids},rest,id,_ => activate_one(~S,~E,rest,wanted,id))