1	import Base
2	import ./store/base.bend as SBase
3	import ./urec.bend as Urec
4	import ./pollrow.bend as Pollrow
5	import ./usage/cred.bend as UCred
6	import ./json.bend as Json
7	import ./clock.bend as Clock
8	import ./act.bend as Act
9	import ./poll.bend as Poll
10	import ./ustore.bend as Ustore
11	import ./usage.bend as Usage
12	import ./view.bend as View
13	import ./collect.bend as Collect
14	
15	def claim.pick(hit: Bool, id: String, rest: Maybe<&2, String>) -> Maybe<&2, String>:
16	  match hit:
17	    case True{}:
18	      Some{id}
19	    case False{}:
20	      rest
21	
22	def claim_of(cs: List<&2, Ustore.Claim>, +slot: String) -> Maybe<&2, String>:
23	  match cs:
24	    case Nil{}:
25	      None{}
26	    case Con{Ustore.Claim{s, id}, t}:
27	      claim.pick(String.eq(s, slot), id, claim_of(t, slot))
28	
29	def struck_json(m: Maybe<&2, String>) -> IO(Json.Json):
30	  match m:
31	    case None{}:
32	      IO.pure(Json.Json, Json.JNull{})
33	    case Some{cred}:
34	      IO.bind(String, Json.Json, Collect.fingerprint(cred), f => IO.pure(Json.Json, Json.JStr{f}))
35	
36	def run_fetch(active: Bool, +creds: String, +slot: String, +email: String, +root: String) -> IO(Usage.Fetched):
37	  match active:
38	    case True{}:
39	      Usage.fetch_usage(creds, True{})
40	    case False{}:
41	      Usage.fetch_usage_gated(creds, False{}, slot, email, root)
42	
43	def rec.ok2(+id: Ustore.Id, +claim: String, +usage: Json.Json, active: Bool, +plan: Poll.Sched) -> Urec.Rec:
44	  Urec.Rec{id, claim, 0, usage, "", None{}, Json.JNull{}, Some{plan}}
45	
46	def rec.ok(+r: Pollrow.Row, +id: Ustore.Id, +claim: String, +usage: Json.Json, +active: Bool, +models: List<&2, String>, +threshold: U32, +now: U32) -> Urec.Rec:
47	  match r:
48	    case Pollrow.Row{fetched, lg, la, fl, le, bu, np, iv, l429, st, sfp, cu, hu}:
49	      rec.ok2(id, claim, usage, active, Poll.plan_after_fetch(iv, lg, usage, active, threshold, models, Pollrow.recent_429(r, now), now))
50	
51	def rec.err(error: Maybe<&2, String>, retry: Maybe<&2, U32>, struck: Json.Json, ok: Urec.Rec, +id: Ustore.Id, +claim: String) -> Urec.Rec:
52	  match error:
53	    case None{}:
54	      ok
55	    case Some{e}:
56	      Urec.Rec{id, claim, 1, Json.JNull{}, e, retry, struck, None{}}
57	
58	def rec.of(o: Usage.Outcome, +id: Ustore.Id, +claim: String, +row: Json.Json, +active: Bool, +models: List<&2, String>, +threshold: U32, +now: U32, struck: Json.Json) -> Urec.Rec:
59	  match o:
60	    case Usage.Outcome{usage, error, retry, st}:
61	      rec.err(error, retry, struck, rec.ok(Pollrow.row.of(row), id, claim, usage, active, models, threshold, now), id, claim)
62	
63	def outcome_struck(o: Usage.Outcome) -> Maybe<&2, String>:
64	  match o:
65	    case Usage.Outcome{usage, error, retry, st}:
66	      st
67	
68	def rec.fetched(f: Usage.Fetched, +id: Ustore.Id, +claim: String, +row: Json.Json, +active: Bool, +models: List<&2, String>, +threshold: U32, +now: U32) -> IO(Urec.Rec):
69	  match f:
70	    case Usage.Fetched{+o, cred}:
71	      IO.bind(Json.Json, Urec.Rec, struck_json(outcome_struck(o)), sj => IO.pure(Urec.Rec, rec.of(o, id, claim, row, active, models, threshold, now, sj)))
72	
73	def fetch_info(+info: Collect.Info, +row: Json.Json, +claim: String, +root: String, +models: List<&2, String>, +threshold: U32, +now: U32) -> IO(Urec.Rec):
74	  match info:
75	    case Collect.Info{num, email, org_name, org_uuid, alias, disabled, active, creds}:
76	      IO.bind(Usage.Fetched, Urec.Rec, run_fetch(active, creds, U32.show(num), email, root),
77	        f => rec.fetched(f, Ustore.Id{U32.show(num), email, org_uuid}, claim, row, active, models, threshold, now))
78	
79	def fetch_st(+st: Collect.St, +claim: String, +root: String, +models: List<&2, String>, +threshold: U32, +now: U32) -> IO(Urec.Rec):
80	  match st:
81	    case Collect.St{info, row, sentinel, healed}:
82	      fetch_info(info, row, claim, root, models, threshold, now)
83	
84	# A state with the lease it won, if any.
85	type Job is Data:
86	  Job{st: Collect.St, claim: Maybe<&2, String>}
87	
88	def jobs(ss: List<&2, Collect.St>, +cs: List<&2, Ustore.Claim>) -> List<&2, Job>:
89	  match ss:
90	    case Nil{}:
91	      []
92	    case Con{+h, t}:
93	      Job{h, claim_of(cs, Collect.st_slot(h))} <> jobs(t, cs)
94	
95	def fetch_all(js: List<&2, Job>, +root: String, +models: List<&2, String>, +threshold: U32, +now: U32) -> IO(List<&2, Urec.Rec>):
96	  match js:
97	    case Nil{}:
98	      IO.pure(List<&2, Urec.Rec>, [])
99	    case Con{Job{st, None{}}, t}:
100	      fetch_all(t, root, models, threshold, now)
101	    case Con{Job{st, Some{claim}}, t}:
102	      IO.bind(Urec.Rec, List<&2, Urec.Rec>, fetch_st(st, claim, root, models, threshold, now),
103	        r => IO.bind(List<&2, Urec.Rec>, List<&2, Urec.Rec>, fetch_all(t, root, models, threshold, now), rest => IO.pure(List<&2, Urec.Rec>, r <> rest)))
104	
105	def healed.cons(h: Bool, +id: Ustore.Id, rest: List<&2, Ustore.Id>) -> List<&2, Ustore.Id>:
106	  match h:
107	    case True{}:
108	      id <> rest
109	    case False{}:
110	      rest
111	
112	def fetchable.cons(ok: Bool, +id: Ustore.Id, rest: List<&2, Ustore.Id>) -> List<&2, Ustore.Id>:
113	  match ok:
114	    case True{}:
115	      id <> rest
116	    case False{}:
117	      rest
118	
119	# Slots with nothing that rules out a fetch.
120	def fetchable(ss: List<&2, Collect.St>) -> List<&2, Ustore.Id>:
121	  match ss:
122	    case Nil{}:
123	      []
124	    case Con{+s, t}:
125	      fetchable.cons(Maybe.is_none(&2, String, Collect.st_sentinel(s)), Collect.st_id(s), fetchable(t))
126	
127	def clear_if(+root: String, ids: List<&2, Ustore.Id>) -> Act.of(Unit):
128	  match ids:
129	    case Nil{}:
130	      Act.pure(Unit, Unit{})
131	    case Con{h, t}:
132	      Urec.clear_dead(root, h <> t)
133	
134	def login_secs(+d: String) -> Maybe<&2, String>:
135	  Maybe.bind(&2, U32, String, U32.read(String.take(d, Nat.sub(String.length(d), 3n))), t => Some{Clock.iso8601_utc(t)})
136	
137	def login.pick(ok: Bool, +i: String) -> Maybe<&2, String>:
138	  match ok:
139	    case True{}:
140	      login_secs(i)
141	    case False{}:
142	      None{}
143	
144	def login.int(ps: List<&2, String>) -> Maybe<&2, String>:
145	  match ps:
146	    case Con{+i, t}:
147	      login.pick(Clock.all_digits(i) && Nat.is_gt(String.length(i), 3n), i)
148	    case Nil{}:
149	      None{}
150	
151	def login.num(j: Json.Json) -> Maybe<&2, String>:
152	  match j:
153	    case Json.JNum{t}:
154	      login.int(String.split(t, '.'))
155	    case _:
156	      None{}
157	
158	# oauth.login_expires_at_iso: refreshTokenExpiresAt (epoch ms) as ISO, when the
159	# stored login records one.
160	def login_expires(+creds: String) -> Maybe<&2, String>:
161	  login.num(Pollrow.jget("refreshTokenExpiresAt", UCred.oauth.of(creds)))
162	
163	def acct.info(+info: Collect.Info, +row: Json.Json, sentinel: Maybe<&2, String>) -> View.Acct:
164	  match info:
165	    case Collect.Info{num, email, org_name, org_uuid, alias, disabled, active, creds}:
166	      View.Acct{num, email, org_name, org_uuid, active, alias, disabled, login_expires(creds), sentinel, Pollrow.row.of(row)}
167	
168	def acct.of(+s: Collect.St) -> View.Acct:
169	  match s:
170	    case Collect.St{info, row, sentinel, healed}:
171	      acct.info(info, row, sentinel)
172	
173	def accts(ss: List<&2, Collect.St>) -> List<&2, View.Acct>:
174	  match ss:
175	    case Nil{}:
176	      []
177	    case Con{h, t}:
178	      acct.of(h) <> accts(t)
179	
180	def pass.final(+root: String, xs: List<&2, Collect.Info>, +now: U32) -> Act.of(List<&2, View.Acct>):
181	  Act.bind(Json.Json, List<&2, View.Acct>, Ustore.read_act(root),
182	    rows => Act.bind(List<&2, Collect.St>, List<&2, View.Acct>, Act.lift(List<&2, Collect.St>, Collect.states(xs, rows, now)), ss => Act.pure(List<&2, View.Acct>, accts(ss))))
183	
184	def pass.fetch(+root: String, +ss: List<&2, Collect.St>, +models: List<&2, String>, +threshold: U32, +now: U32, +tick: Nat, xs: List<&2, Collect.Info>) -> Act.of(List<&2, View.Acct>):
185	  Act.bind(List<&2, Ustore.Claim>, List<&2, View.Acct>, Ustore.reserve(root, fetchable(ss), now, True{}, True{}, tick),
186	    cs => Act.bind(List<&2, Urec.Rec>, List<&2, View.Acct>, Act.lift(List<&2, Urec.Rec>, fetch_all(jobs(ss, cs), root, models, threshold, now)),
187	      rs => Act.bind(List<&2, String>, List<&2, View.Acct>, Urec.record(root, rs, now),
188	        done => pass.final(root, xs, now))))
189	
190	def healed_ids(ss: List<&2, Collect.St>) -> List<&2, Ustore.Id>:
191	  match ss:
192	    case Nil{}:
193	      []
194	    case Con{+s, t}:
195	      healed.cons(Collect.st_healed(s), Collect.st_id(s), healed_ids(t))
196	
197	def pass.heal(+root: String, ss: List<&2, Collect.St>, +models: List<&2, String>, +threshold: U32, +now: U32, +tick: Nat, +xs: List<&2, Collect.Info>) -> Act.of(List<&2, View.Acct>):
198	  Act.bind(Unit, List<&2, View.Acct>, clear_if(root, healed_ids(ss)),
199	    u => Act.bind(Json.Json, List<&2, View.Acct>, Ustore.read_act(root),
200	      rows => Act.bind(List<&2, Collect.St>, List<&2, View.Acct>, Act.lift(List<&2, Collect.St>, Collect.states(xs, rows, now)),
201	        s2 => pass.fetch(root, s2, models, threshold, now, tick, xs))))
202	
203	# One usage pass over `xs`: lease what is eligible, fetch it, merge the
204	# outcomes, and answer an Acct per account with the state read after the merge.
205	def collect(+ctx: SBase.Ctx, +xs: List<&2, Collect.Info>, +now: U32, +models: List<&2, String>, +threshold: U32, +tick: Nat) -> Act.of(List<&2, View.Acct>):
206	  Act.bind(Json.Json, List<&2, View.Acct>, Ustore.read_act(SBase.root_of(ctx)),
207	    rows => Act.bind(List<&2, Collect.St>, List<&2, View.Acct>, Act.lift(List<&2, Collect.St>, Collect.states(xs, rows, now)),
208	      ss => pass.heal(SBase.root_of(ctx), ss, models, threshold, now, tick, xs)))