1	import Base
2	
3	# Owned descriptor-local FIFO metadata. Back is newest-first, front oldest-first.
4	# Cached event size and batch count retain empty-batch compatibility without scans.
5	type Batch<E: Data> is Data:
6	  Batch{tick: Nat,count: Nat,values: List<&2,E>}
7	type Queue<E: Data> is Data:
8	  Queue{front: List<&2,Batch<E>>,back: List<&2,Batch<E>>,size: Nat,dropped: Nat,batchCount: Nat}
9	
10	def empty(~E: Data) -> Queue<E>: Queue{[],[],0n,0n,0n}
11	
12	def count_loop(~E: Data,items: List<&2,E>,acc: Nat) -> Nat:
13	  match items:
14	    case []: acc
15	    case _ <> rest: count_loop(~E,rest,Nat.add(1n,acc))
16	
17	# The compatibility batch-list lowering also uses count zero for empty batches.
18	def publish_nonempty(~E: Data,queue: Queue<E>,+tick: Nat,+items: List<&2,E>,+count: Nat) -> Queue<E>:
19	  match queue:
20	    case Queue{front,back,size,dropped,batches}: Queue{front,Batch{tick,count,items} <> back,Nat.add(size,count),dropped,1n+batches}
21	
22	def publish_checked(~E: Data,nonempty: Bool,queue: Queue<E>,tick: Nat,items: List<&2,E>,count: Nat) -> Queue<E>:
23	  match nonempty:
24	    case False{}: queue
25	    case True{}: publish_nonempty(~E,queue,tick,items,count)
26	
27	def publish(~E: Data,queue: Queue<E>,tick: Nat,+items: List<&2,E>) -> Queue<E>:
28	  publish_checked(~E,Nat.is_lt(0n,count_loop(~E,items,0n)),queue,tick,items,count_loop(~E,items,0n))
29	
30	def normalize(~E: Data,queue: Queue<E>) -> Queue<E>:
31	  match queue:
32	    case Queue{Nil{},back,size,dropped,batches}: Queue{List.reverse(&2,Batch<E>,back),[],size,dropped,batches}
33	    case Queue{front,back,size,dropped,batches}: Queue{front,back,size,dropped,batches}
34	
35	def trim_step(~E: Data,remove: Bool,+tick: Nat,+count: Nat,+items: List<&2,E>,+rest: List<&2,Batch<E>>,+back: List<&2,Batch<E>>,+size: Nat,+dropped: Nat,+batches:Nat,ih: Queue<E> -> Queue<E>) -> Queue<E>:
36	  match remove:
37	    case False{}: Queue{Batch{tick,count,items} <> rest,back,size,dropped,batches}
38	    case True{}: ih(Queue{rest,back,Nat.sub(size,count),Nat.max(dropped,tick),Nat.sub(batches,1n)})
39	
40	# Match the canonical two-pass policy even for non-monotonic supplied ticks:
41	# first remove the expired prefix, then evict whole batches for capacity only.
42	def past_normalized(~E:Data,queue:Queue<E>,+boundary:Nat,ih:Queue<E> -> Queue<E>) -> Queue<E>:
43	  match queue:
44	    case Queue{Nil{},back,size,dropped,batches}:Queue{[],back,size,dropped,batches}
45	    case Queue{Batch{+tick,+count,+items} <> +rest,+back,+size,+dropped,+batches}:trim_step(~E,Nat.is_le(tick,boundary),tick,count,items,rest,back,size,dropped,batches,ih)
46	def past_loop(~E:Data,fuel:Nat,queue:Queue<E>,+boundary:Nat) -> Queue<E>:
47	  match fuel:
48	    case Zero{}:queue
49	    case Succ{rest}:past_normalized(~E,normalize(~E,queue),boundary,next => past_loop(~E,rest,next,boundary))
50	def capacity_normalized(~E:Data,queue:Queue<E>,+capacity:Nat,ih:Queue<E> -> Queue<E>) -> Queue<E>:
51	  match queue:
52	    case Queue{Nil{},back,size,dropped,batches}:Queue{[],back,size,dropped,batches}
53	    case Queue{Batch{+tick,+count,+items} <> +rest,+back,+size,+dropped,+batches}:trim_step(~E,Nat.is_lt(capacity,size),tick,count,items,rest,back,size,dropped,batches,ih)
54	def capacity_loop(~E:Data,fuel:Nat,queue:Queue<E>,+capacity:Nat) -> Queue<E>:
55	  match fuel:
56	    case Zero{}:queue
57	    case Succ{rest}:capacity_normalized(~E,normalize(~E,queue),capacity,next => capacity_loop(~E,rest,next,capacity))
58	def capacity_pass(~E:Data,queue:Queue<E>,capacity:Nat) -> Queue<E>:
59	  match queue:
60	    case Queue{front,back,size,dropped,+batches}:capacity_loop(~E,Nat.add(batches,1n),Queue{front,back,size,dropped,batches},capacity)
61	def trim(~E:Data,queue:Queue<E>,boundary:Nat,capacity:Nat) -> Queue<E>:
62	  match queue:
63	    case Queue{front,back,size,dropped,+batches}:capacity_pass(~E,past_loop(~E,Nat.add(batches,1n),Queue{front,back,size,dropped,batches},boundary),capacity)
64	
65	# Materialization is tail-safe; no recursive append of a growing retained prefix.
66	def prepend_items(~E: Data,items: List<&2,E>,acc: List<&2,E>) -> List<&2,E>:
67	  match items:
68	    case []: acc
69	    case item <> rest: prepend_items(~E,rest,item <> acc)
70	
71	def visible_items(~E: Data,visible: Bool,items: List<&2,E>,acc: List<&2,E>) -> List<&2,E>:
72	  match visible:
73	    case False{}: acc
74	    case True{}: prepend_items(~E,items,acc)
75	
76	def read_loop(~E: Data,batches: List<&2,Batch<E>>,+cursor: Nat,acc: List<&2,E>) -> List<&2,E>:
77	  match batches:
78	    case []: acc
79	    case Batch{tick,_,items} <> rest: read_loop(~E,rest,cursor,visible_items(~E,Nat.is_lt(cursor,tick),items,acc))
80	
81	def read_back(~E: Data,+front: List<&2,Batch<E>>,+back: List<&2,Batch<E>>,+cursor: Nat) -> List<&2,E>:
82	  List.reverse(&2,E,read_loop(~E,List.reverse(&2,Batch<E>,back),cursor,read_loop(~E,front,cursor,[])))
83	
84	def read(~E: Data,queue: Queue<E>,cursor: Nat) -> List<&2,E>:
85	  match queue:
86	    case Queue{front,back,_,_,_}: read_back(~E,front,back,cursor)