1	import Base
2	import ./pollrow.bend as Pollrow
3	import ./fs/write.bend as FWrite
4	import ./json.bend as Json
5	import ./fs.bend as Fs
6	import ./act.bend as Act
7	import ./lock.bend as Lock
8	import ./poll.bend as Poll
9	
10	# API (also the daemon's usage store; do not copy it):
11	#   read_rows(root) -> IO(Json)          the accounts object; empty for a missing/torn/other-version file
12	#   row_of(rows, Id) -> Maybe<Json>      the row, only when it belongs to this (email, org) identity
13	#   reserve(root, ids, now, respect, repair, tick) -> Act(List<Claim>)   win the right to fetch (90 s lease)
14	#   record(root, recs, now) -> Act(List<slot>)   merge outcomes fenced by their lease (success, failure, sentinel)
15	#   adopt(root, readings, hold, now) -> Act(AR)  merge another machine's readings, with an optional hold
16	#   clear_dead(root, ids) -> Act(Unit)           reset a slot's row after its credential was replaced
17	# Every write is read-modify-write under <root>/cache/.usage.lock; no network call runs under it.
18	#
19	# cache/usage.json (usage_store.py): rows keyed by slot, guarded by the
20	# slot's (email, organizationUuid), read lock-free and written as
21	# read-modify-write under cache/.usage.lock. No network call runs under the
22	# lock: reserve() leases the rows, the caller fetches, record() merges what
23	# the lease still covers.
24	
25	def ActM(-A: Type) -> Type:
26	  Act.of(A)
27	
28	def ActM.pure(-A: Type, x: A) -> Act.of(A):
29	  Act.pure(A, x)
30	
31	def ActM.bind(-A: Type, -B: Type, m: Act.of(A), f: A -> Act.of(B)) -> Act.of(B):
32	  Act.bind(A, B, m, f)
33	
34	def usage_path(+root: String) -> String:
35	  root ++ "/cache/usage.json"
36	
37	def usage_lock(+root: String) -> String:
38	  root ++ "/cache/.usage.lock"
39	
40	# The identity a slot number currently maps to.
41	type Id is Data:
42	  Id{slot: String, email: String, org: String}
43	
44	# A fetch lease: the slot and its fencing id.
45	type Claim is Data:
46	  Claim{slot: String, id: String}
47	
48	def jv(+k: String, +j: Json.Json) -> Json.Json:
49	  Maybe.default(&2, Json.Json, Json.get(k, j), Json.JNull{})
50	
51	def jstr(s: String) -> Json.Json:
52	  Json.JStr{s}
53	
54	# An epoch as the reference stores it: a float.
55	def fl(+t: U32) -> Json.Json:
56	  Json.JNum{U32.show(t) ++ ".0"}
57	
58	def empty_rows() -> Json.Json:
59	  Json.of_pairs([])
60	
61	def obj_or_empty(j: Json.Json) -> Json.Json:
62	  match j:
63	    case Json.JObj{f}:
64	      Json.JObj{f}
65	    case _:
66	      empty_rows()
67	
68	def rows.pick(ok: Bool, +j: Json.Json) -> Json.Json:
69	  match ok:
70	    case True{}:
71	      obj_or_empty(jv("accounts", j))
72	    case False{}:
73	      empty_rows()
74	
75	def is_v2(j: Json.Json) -> Bool:
76	  match j:
77	    case Json.JNum{t}:
78	      String.eq(t, "2")
79	    case _:
80	      False{}
81	
82	# The accounts object of a parsed usage.json; empty for any other schema.
83	def rows_of(+j: Json.Json) -> Json.Json:
84	  rows.pick(is_v2(jv("schemaVersion", j)), j)
85	
86	def rows.parsed(r: Result<&2, &2, String, Json.Json>) -> Json.Json:
87	  match r:
88	    case Fail{e}:
89	      empty_rows()
90	    case Done{j}:
91	      rows_of(j)
92	
93	def rows.text(r: Fs.Res(String)) -> Json.Json:
94	  match r:
95	    case Fail{e}:
96	      empty_rows()
97	    case Done{t}:
98	      rows.parsed(Json.parse(t))
99	
100	# A missing, torn or other-version file reads as empty.
101	def read_rows(+root: String) -> IO(Json.Json):
102	  IO.bind(Fs.Res(String), Json.Json, Fs.read_file(usage_path(root)), r => IO.pure(Json.Json, rows.text(r)))
103	
104	def matches(+row: Json.Json, +id: Id) -> Bool:
105	  match id:
106	    case Id{slot, email, org}:
107	      String.eq(Maybe.default(&2, String, Json.as_string(jv("email", row)), "\u{1}"), email)
108	        && String.eq(Maybe.default(&2, String, Json.as_string(Maybe.default(&2, Json.Json, Json.get("organizationUuid", row), Json.JStr{""})), "\u{1}"), org)
109	
110	def fresh_row(+id: Id) -> Json.Json:
111	  match id:
112	    case Id{slot, email, org}:
113	      Json.of_pairs([("email", jstr(email)), ("organizationUuid", jstr(org))])
114	
115	def row.if(ok: Bool, +row: Json.Json) -> Maybe<&2, Json.Json>:
116	  match ok:
117	    case True{}:
118	      Some{row}
119	    case False{}:
120	      None{}
121	
122	def row.pick(m: Maybe<&2, Json.Json>, +id: Id) -> Maybe<&2, Json.Json>:
123	  match m:
124	    case None{}:
125	      None{}
126	    case Some{+row}:
127	      row.if(matches(row, id), row)
128	
129	# The stored row of a slot, only when it belongs to this identity.
130	def row_of(+rows: Json.Json, +id: Id) -> Maybe<&2, Json.Json>:
131	  match id:
132	    case Id{slot, email, org}:
133	      row.pick(Json.get(slot, rows), id)
134	
135	def write_rows(+root: String, +rows: Json.Json) -> Act.of(Unit):
136	  do ActM<Unit>:
137	    FWrite.mkdir_p(root ++ "/cache")
138	    FWrite.write_atomic(usage_path(root), Json.emit_pretty(Json.of_pairs([("schemaVersion", Json.JNum{"2"}), ("accounts", rows)])), "600")
139	
140	def with_usage_lock(-A: Type, +root: String, act: Act.of(A)) -> Act.of(A):
141	  Lock.with_lock(A, usage_lock(root), 10000, act)
142	
143	def read_act(+root: String) -> Act.of(Json.Json):
144	  Act.lift(Json.Json, read_rows(root))
145	
146	type RR is Data:
147	  RR{rows: Json.Json, won: List<&2, Claim>}
148	
149	def claim_row(+row: Json.Json, +claim: String, +now: U32) -> Json.Json:
150	  Json.set("claimUntil", fl((now + Poll.claim_ttl() : U32)), Json.set("claimId", jstr(claim), Json.set("lastAttemptAt", fl(now), row)))
151	
152	def reserve.take(+rows: Json.Json, +id: Id, +row: Json.Json, +claim: String, +now: U32, won: List<&2, Claim>) -> RR:
153	  match id:
154	    case Id{slot, email, org}:
155	      RR{Json.set(slot, claim_row(row, claim, now), rows), Claim{slot, claim} <> won}
156	
157	def reserve.fresh(+rows: Json.Json, +id: Id, +claim: String, +now: U32, won: List<&2, Claim>) -> RR:
158	  reserve.take(rows, id, fresh_row(id), claim, now, won)
159	
160	def reserve.stored(ok: Bool, +rows: Json.Json, +id: Id, +row: Json.Json, +claim: String, +now: U32, won: List<&2, Claim>) -> RR:
161	  match ok:
162	    case True{}:
163	      reserve.take(rows, id, row, claim, now, won)
164	    case False{}:
165	      RR{rows, won}
166	
167	def reserve.row(m: Maybe<&2, Json.Json>, +rows: Json.Json, +id: Id, +claim: String, +now: U32, +respect: Bool, +repair: Bool, won: List<&2, Claim>) -> RR:
168	  match m:
169	    case None{}:
170	      reserve.fresh(rows, id, claim, now, won)
171	    case Some{+row}:
172	      reserve.stored(Pollrow.row_eligible(Pollrow.row.of(row), now, respect, repair), rows, id, row, claim, now, won)
173	
174	def claim_id(+slot: String, +now: U32, +tick: Nat) -> String:
175	  "c" ++ U32.show(now) ++ "-" ++ slot ++ "-" ++ Nat.show(tick)
176	
177	def slot_of(+id: Id) -> String:
178	  match id:
179	    case Id{slot, email, org}:
180	      slot
181	
182	def reserve.go(ids: List<&2, Id>, rr: RR, +now: U32, +respect: Bool, +repair: Bool, +tick: Nat) -> RR:
183	  match ids rr:
184	    case Nil{} _:
185	      rr
186	    case Con{+id, t} RR{+rows, won}:
187	      reserve.go(t, reserve.row(row_of(rows, id), rows, id, claim_id(slot_of(id), now, tick), now, respect, repair, won), now, respect, repair, 1n+tick)
188	
189	def reserve.write(won: List<&2, Claim>, +root: String, +rows: Json.Json) -> Act.of(List<&2, Claim>):
190	  match won:
191	    case Nil{}:
192	      Act.pure(List<&2, Claim>, [])
193	    case Con{h, t}:
194	      Act.bind(Unit, List<&2, Claim>, write_rows(root, rows), u => Act.pure(List<&2, Claim>, h <> t))
195	
196	def reserve.commit(+root: String, rr: RR) -> Act.of(List<&2, Claim>):
197	  match rr:
198	    case RR{rows, won}:
199	      reserve.write(won, root, rows)
200	
201	def reserve.act(+root: String, ids: List<&2, Id>, +now: U32, +respect: Bool, +repair: Bool, +tick: Nat) -> Act.of(List<&2, Claim>):
202	  Act.bind(Json.Json, List<&2, Claim>, read_act(root), rows => reserve.commit(root, reserve.go(ids, RR{rows, []}, now, respect, repair, tick)))
203	
204	# Atomically win the right to fetch: re-check each slot's eligibility and
205	# stamp a 90 s lease under the lock (UsageStore.reserve). `respect` is the
206	# on-demand mode (stale AND poll-due), `repair` also frees a reset-parked plan.
207	def reserve(+root: String, ids: List<&2, Id>, +now: U32, +respect: Bool, +repair: Bool, +tick: Nat) -> Act.of(List<&2, Claim>):
208	  with_usage_lock(List<&2, Claim>, root, reserve.act(root, ids, now, respect, repair, tick))
209	
210	type RAd is Data:
211	  RAd{row: Json.Json, adopted: Bool}