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	import ./ustore.bend as Ustore
10	
11	# One finished fetch. kind: 0 success, 1 failure, 2 sentinel (nothing to store
12	# but the lease). `struck` is the fingerprint of the refresh token the server
13	# rejected for good (JNull otherwise).
14	type Rec is Data:
15	  Rec{id: Ustore.Id, claim: String, kind: U32, usage: Json.Json, error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>}
16	
17	def release(+row: Json.Json) -> Json.Json:
18	  Json.set("claimUntil", Json.JNum{"0.0"}, Json.set("claimId", Json.JNull{}, row))
19	
20	def with_plan(m: Maybe<&2, Poll.Sched>, +row: Json.Json) -> Json.Json:
21	  match m:
22	    case None{}:
23	      row
24	    case Some{Poll.Sched{next, interval}}:
25	      Json.set("pollIntervalS", Ustore.fl(interval), Json.set("nextPollAt", Ustore.fl(next), row))
26	
27	def apply_ok(+row: Json.Json, +usage: Json.Json, plan: Maybe<&2, Poll.Sched>, +now: U32) -> Json.Json:
28	  Json.set("authDeadStrikes", Json.JNum{"0"},
29	    Json.set("rejectedFingerprint", Json.JNull{},
30	      Json.set("backoffUntil", Json.JNull{},
31	        Json.set("lastError", Json.JNull{},
32	          Json.set("consecutiveFailures", Json.JNum{"0"},
33	            with_plan(plan, Json.set("fetchedAt", Ustore.fl(now), Json.set("lastGood", usage, Json.set("lastAttemptAt", Ustore.fl(now), release(row))))))))))
34	
35	def is_permanent(+e: String) -> Bool:
36	  String.eq(e, "invalid_grant") || String.eq(e, "no_refresh_token")
37	
38	def with_strike(perm: Bool, +strikes: U32, struck: Json.Json, +row: Json.Json) -> Json.Json:
39	  match perm:
40	    case True{}:
41	      Json.set("struckFingerprint", struck, Json.set("authDeadStrikes", Json.JNum{U32.show((strikes + 1 : U32))}, row))
42	    case False{}:
43	      row
44	
45	def with_429(rl: Bool, +now: U32, +row: Json.Json) -> Json.Json:
46	  match rl:
47	    case True{}:
48	      Json.set("last429At", Ustore.fl(now), row)
49	    case False{}:
50	      row
51	
52	def apply_fail2(+row: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, +failures: U32, +strikes: U32, +now: U32) -> Json.Json:
53	  with_strike(is_permanent(error), strikes, struck,
54	    Json.set("backoffUntil", Ustore.fl((now + Pollrow.failure_backoff(failures, retry, String.eq(error, "http-429")) : U32)),
55	      with_429(String.eq(error, "http-429"), now,
56	        Json.set("lastError", Ustore.jstr(error), Json.set("consecutiveFailures", Json.JNum{U32.show(failures)},
57	          Json.set("lastAttemptAt", Ustore.fl(now), release(row)))))))
58	
59	def apply_fail(+row: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, +now: U32) -> Json.Json:
60	  apply_fail2(row, error, retry, struck, (Maybe.default(&2, U32, Pollrow.num.field("consecutiveFailures", row), 0) + 1 : U32),
61	    Maybe.default(&2, U32, Pollrow.num.field("authDeadStrikes", row), 0), now)
62	
63	def apply_kind2(fail: Bool, +row: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, +now: U32) -> Json.Json:
64	  match fail:
65	    case True{}:
66	      apply_fail(row, error, retry, struck, now)
67	    case False{}:
68	      release(row)
69	
70	def apply_kind.pick(ok: Bool, +kind: U32, +row: Json.Json, +usage: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>, +now: U32) -> Json.Json:
71	  match ok:
72	    case True{}:
73	      apply_ok(row, usage, plan, now)
74	    case False{}:
75	      apply_kind2(U32.is_eq(kind, 1), row, error, retry, struck, now)
76	
77	def apply_kind(+kind: U32, +row: Json.Json, +usage: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>, +now: U32) -> Json.Json:
78	  apply_kind.pick(U32.is_eq(kind, 0), kind, row, usage, error, retry, struck, plan, now)
79	
80	type RA is Data:
81	  RA{rows: Json.Json, done: List<&2, String>}
82	
83	def record.fenced(ok: Bool, +rows: Json.Json, +slot: String, +row: Json.Json, +kind: U32, +usage: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>, done: List<&2, String>, +now: U32) -> RA:
84	  match ok:
85	    case True{}:
86	      RA{Json.set(slot, apply_kind(kind, row, usage, error, retry, struck, plan, now), rows), slot <> done}
87	    case False{}:
88	      RA{rows, done}
89	
90	def record.apply(m: Maybe<&2, Json.Json>, +rows: Json.Json, +slot: String, +claim: String, +kind: U32, +usage: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>, done: List<&2, String>, +now: U32) -> RA:
91	  match m:
92	    case None{}:
93	      RA{rows, done}
94	    case Some{+row}:
95	      record.fenced(String.eq(Maybe.default(&2, String, Json.as_string(Ustore.jv("claimId", row)), ""), claim), rows, slot, row, kind, usage, error, retry, struck, plan, done, now)
96	
97	def record.one(+rows: Json.Json, +r: Rec, done: List<&2, String>, +now: U32) -> RA:
98	  match r:
99	    case Rec{id, claim, kind, usage, error, retry, struck, plan}:
100	      record.apply(Ustore.row_of(rows, id), rows, Ustore.slot_of(id), claim, kind, usage, error, retry, struck, plan, done, now)
101	
102	def record.go(rs: List<&2, Rec>, ra: RA, +now: U32) -> RA:
103	  match rs ra:
104	    case Nil{} _:
105	      ra
106	    case Con{h, t} RA{+rows, done}:
107	      record.go(t, record.one(rows, h, done, now), now)
108	
109	def record.write(done: List<&2, String>, +root: String, +rows: Json.Json) -> Act.of(List<&2, String>):
110	  match done:
111	    case Nil{}:
112	      Act.pure(List<&2, String>, [])
113	    case Con{h, t}:
114	      Act.bind(Unit, List<&2, String>, Ustore.write_rows(root, rows), u => Act.pure(List<&2, String>, h <> t))
115	
116	def record.commit(+root: String, ra: RA) -> Act.of(List<&2, String>):
117	  match ra:
118	    case RA{rows, done}:
119	      record.write(done, root, rows)
120	
121	def record.act(+root: String, rs: List<&2, Rec>, +now: U32) -> Act.of(List<&2, String>):
122	  Act.bind(Json.Json, List<&2, String>, Ustore.read_act(root), rows => record.commit(root, record.go(rs, RA{rows, []}, now)))
123	
124	# Merge outcomes fenced by the leases that produced them: a late writer whose
125	# lease or identity was replaced is ignored. Success resets the failure
126	# fields; a failure never touches lastGood and fetchedAt. Returns the slots
127	# accepted.
128	def record(+root: String, rs: List<&2, Rec>, +now: U32) -> Act.of(List<&2, String>):
129	  Ustore.with_usage_lock(List<&2, String>, root, record.act(root, rs, now))
130	
131	type Reading is Data:
132	  Reading{id: Ustore.Id, usage: Json.Json, age: U32}
133	
134	def hold.pick(later: Bool, +held: U32, +row: Json.Json) -> Json.Json:
135	  match later:
136	    case True{}:
137	      Json.set("heldUntil", Ustore.fl(held), row)
138	    case False{}:
139	      Json.del("heldUntil", row)
140	
141	def hold.apply(+held: U32, +now: U32, +row: Json.Json) -> Json.Json:
142	  hold.pick(U32.is_gt(held, now), held, row)
143	
144	def hold.reset(m: Maybe<&2, U32>, +held: U32) -> U32:
145	  match m:
146	    case Some{r}:
147	      Poll.min_u(held, r)
148	    case None{}:
149	      held
150	
151	def hold.row(+h: U32, +stored: U32, +now: U32, +row: Json.Json) -> Json.Json:
152	  hold.apply(hold.reset(Poll.earliest_reset(Poll.relevant(Pollrow.last_good.of(row), ["all"]), None{}), Poll.min_u((now + h : U32), (stored + Poll.trust_ceiling() : U32))), now, row)
153	
154	def adopt.hold(hold: Maybe<&2, U32>, +stored: U32, +now: U32, +row: Json.Json) -> Json.Json:
155	  match hold:
156	    case None{}:
157	      row
158	    case Some{h}:
159	      hold.row(h, stored, now, row)
160	
161	def adopt.newer(replace: Bool, +usage: Json.Json, +at: U32, stored: Maybe<&2, U32>, hold: Maybe<&2, U32>, +now: U32, +row: Json.Json) -> Ustore.RAd:
162	  match replace:
163	    case True{}:
164	      Ustore.RAd{adopt.hold(hold, at, now, Json.set("fetchedAt", Ustore.fl(at), Json.set("lastGood", usage, row))), True{}}
165	    case False{}:
166	      Ustore.RAd{adopt.hold(hold, Maybe.default(&2, U32, stored, at), now, row), False{}}
167	
168	def is_newer(stored: Maybe<&2, U32>, +at: U32) -> Bool:
169	  match stored:
170	    case None{}:
171	      True{}
172	    case Some{s}:
173	      U32.is_gt(at, s)
174	
175	def adopt.row(+row: Json.Json, +usage: Json.Json, +age: U32, hold: Maybe<&2, U32>, +now: U32) -> Ustore.RAd:
176	  adopt.newer(is_newer(Pollrow.num.field("fetchedAt", row), Poll.sub0(now, age)), usage, Poll.sub0(now, age), Pollrow.num.field("fetchedAt", row), hold, now, row)
177	
178	def adopt.stored(m: Maybe<&2, Json.Json>, +id: Ustore.Id) -> Json.Json:
179	  match m:
180	    case Some{row}:
181	      row
182	    case None{}:
183	      Ustore.fresh_row(id)
184	
185	type AR is Data:
186	  AR{rows: Json.Json, adopted: List<&2, String>, kept: List<&2, String>}
187	
188	def adopt.sort(yes: Bool, +slot: String, +rows: Json.Json, adopted: List<&2, String>, kept: List<&2, String>) -> AR:
189	  match yes:
190	    case True{}:
191	      AR{rows, slot <> adopted, kept}
192	    case False{}:
193	      AR{rows, adopted, slot <> kept}
194	
195	def adopt.put(+rd: Ustore.RAd, +slot: String, +rows: Json.Json, adopted: List<&2, String>, kept: List<&2, String>) -> AR:
196	  match rd:
197	    case Ustore.RAd{row, yes}:
198	      adopt.sort(yes, slot, Json.set(slot, row, rows), adopted, kept)
199	
200	def adopt.go(rs: List<&2, Reading>, ar: AR, +hold: Maybe<&2, U32>, +now: U32) -> AR:
201	  match rs ar:
202	    case Nil{} _:
203	      ar
204	    case Con{Reading{+id, usage, age}, t} AR{+rows, adopted, kept}:
205	      adopt.go(t, adopt.put(adopt.row(adopt.stored(Ustore.row_of(rows, id), id), usage, age, hold, now), Ustore.slot_of(id), rows, adopted, kept), hold, now)
206	
207	def adopt.finish(+root: String, ar: AR) -> Act.of(AR):
208	  match ar:
209	    case AR{+rows, adopted, kept}:
210	      Act.bind(Unit, AR, Ustore.write_rows(root, rows), u => Act.pure(AR, AR{rows, adopted, kept}))
211	
212	def adopt.act(+root: String, rs: List<&2, Reading>, hold: Maybe<&2, U32>, +now: U32) -> Act.of(AR):
213	  Act.bind(Json.Json, AR, Ustore.read_act(root), rows => adopt.finish(root, adopt.go(rs, AR{rows, [], []}, hold, now)))
214	
215	# Merge measurements another machine took (UsageStore.adopt): a reading
216	# replaces lastGood only when newer than the stored one; `hold` (seconds)
217	# keeps collectors off the slot, never past the reading's earliest window
218	# reset nor an hour past the stored measurement; Some{0} lifts a hold.
219	def adopt(+root: String, rs: List<&2, Reading>, hold: Maybe<&2, U32>, +now: U32) -> Act.of(AR):
220	  Ustore.with_usage_lock(AR, root, adopt.act(root, rs, hold, now))
221	
222	def cleared(+row: Json.Json) -> Json.Json:
223	  Json.set("backoffUntil", Json.JNull{}, Json.set("lastError", Json.JNull{}, Json.set("consecutiveFailures", Json.JNum{"0"},
224	    Json.set("struckFingerprint", Json.JNull{}, Json.set("authDeadStrikes", Json.JNum{"0"}, release(row))))))
225	
226	def clear.row(m: Maybe<&2, Json.Json>, +rows: Json.Json, +id: Ustore.Id) -> Json.Json:
227	  match m:
228	    case Some{row}:
229	      Json.set(Ustore.slot_of(id), cleared(row), rows)
230	    case None{}:
231	      Json.set(Ustore.slot_of(id), cleared(Ustore.fresh_row(id)), rows)
232	
233	def clear.go(ids: List<&2, Ustore.Id>, +rows: Json.Json) -> Json.Json:
234	  match ids:
235	    case Nil{}:
236	      rows
237	    case Con{+id, t}:
238	      clear.go(t, clear.row(Ustore.row_of(rows, id), rows, id))
239	
240	def clear.act(+root: String, ids: List<&2, Ustore.Id>) -> Act.of(Unit):
241	  Act.bind(Json.Json, Unit, Ustore.read_act(root), rows => Ustore.write_rows(root, clear.go(ids, rows)))
242	
243	# Lift the dead-token quarantine of slots whose credential was replaced.
244	def clear_dead(+root: String, ids: List<&2, Ustore.Id>) -> Act.of(Unit):
245	  Ustore.with_usage_lock(Unit, root, clear.act(root, ids))
246