1	import Base
2	import ../json.bend as Json
3	import ../fs.bend as Fs
4	import ../fs/write.bend as FsW
5	import ../act.bend as Act
6	import ../clock.bend as Clock
7	import ../store.bend as Store
8	import ../store/base.bend as SBase
9	import ../swlive.bend as L
10	import ../policy/types.bend as T
11	import ./events.bend as Ev
12	import ../fmt/out.bend as O
13	import ./cadence.bend as Cad
14	import ./config.bend as Cfg
15	import ./live.bend as Lv
16	import ../store/io.bend as SIo
17	import ./walk.bend as Wk
18	import ./quarantine.bend as Q
19	import ./dec.bend as Dc
20	import ./tick.bend as Tk
21	import ./gather.bend as G
22	import ./collect.bend as Col
23	import ./status.bend as Stat
24	import ./pidlock.bend as Pid
25	
26	# `cswap auto`: --once is one tick (exit 0 switched, 1 error, 2 no action, 3
27	# blocked); otherwise tick, sleep, tick, forever. One decider per backup root
28	# (a pid-file flock), the status file rewritten after every tick, and a JSONL
29	# log line per tick with where each account's reading came from. A SIGTERM
30	# ends the process (the kernel drops the lock); the pid file is then stale
31	# but harmless, as the lock is what counts.
32	
33	# What a run is: the roster context, the state dir, the settings in force,
34	# --json, --dry-run, the switch options, and when the process started, the
35	# pid, and what is needed to read the settings again each tick (the flags and
36	# the profile in force: src/daemon/live.bend).
37	
38	type Run is Data:
39	  Run{ctx: SBase.Ctx, state_dir: String, cfg: Cfg.Cfg, fmt: O.Fmt, dry: Bool, opts: L.Opts, started: String, version: String, pid: U32, live: Lv.Live}
40	
41	def ActM(-A: Type) -> Type:
42	  Act.of(A)
43	
44	def ActM.pure(-A: Type, x: A) -> Act.of(A):
45	  Act.pure(A, x)
46	
47	def ActM.bind(-A: Type, -B: Type, m: Act.of(A), f: A -> Act.of(B)) -> Act.of(B):
48	  Act.bind(A, B, m, f)
49	
50	def run_ctx(r: Run) -> SBase.Ctx:
51	  match r:
52	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
53	      c
54	
55	def run_dir(r: Run) -> String:
56	  match r:
57	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
58	      sd
59	
60	def run_cfg(r: Run) -> Cfg.Cfg:
61	  match r:
62	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
63	      cf
64	
65	def run_fmt(r: Run) -> O.Fmt:
66	  match r:
67	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
68	      j
69	
70	def run_dry(r: Run) -> Bool:
71	  match r:
72	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
73	      d
74	
75	def run_opts(r: Run) -> L.Opts:
76	  match r:
77	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
78	      o
79	
80	def run_started(r: Run) -> String:
81	  match r:
82	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
83	      st
84	
85	def run_version(r: Run) -> String:
86	  match r:
87	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
88	      v
89	
90	def run_pid(r: Run) -> U32:
91	  match r:
92	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
93	      p
94	
95	def run_live(r: Run) -> Lv.Live:
96	  match r:
97	    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
98	      lv
99	
100	def with_live(+r: Run, +cfg: Cfg.Cfg, +lv: Lv.Live) -> Run:
101	  match r:
102	    case Run{c, sd, cf, j, d, o, st, v, p, l0}:
103	      Run{c, sd, cfg, j, d, o, st, v, p, lv}
104	
105	# The run with the settings of this tick: settings.json read again, the
106	# profile laid over it, the flags over both.
107	def refreshed(+r: Run, +text: String) -> Run:
108	  with_live(r, Lv.cfg_for(text, Lv.live_flags(run_live(r)), Lv.live_over(run_live(r))), Lv.live_for(text, Lv.live_flags(run_live(r)), Lv.live_over(run_live(r))))
109	
110	def refresh(+r: Run) -> Act.of(Run):
111	  do ActM<Run>:
112	    text : String <- Act.lift(String, Lv.read_text(SBase.root_of(run_ctx(r))))
113	    return refreshed(r, text)
114	
115	def tagged(+r: Run, +o: Tk.Out) -> Tk.Out:
116	  match o:
117	    case Tk.Out{Dc.TickRes{i, evs}, m, a, es, ac, ck}:
118	      Tk.Out{Dc.TickRes{i, Lv.tag_events(Lv.live_tag(run_live(r)), evs)}, m, a, es, ac, ck}
119	
120	def interval(r: Run) -> U32:
121	  Cfg.cfg_interval(run_cfg(r))
122	
123	# ---- printing --------------------------------------------------------------------------------------------
124	
125	def show2(human: Bool, +f: O.Fmt, +e: Json.Json) -> String:
126	  match human:
127	    case True{}:
128	      Ev.human(e)
129	    case False{}:
130	      O.event_text(f, e)
131	
132	def show(+f: O.Fmt, +e: Json.Json) -> String:
133	  show2(O.is_text(f), f, e)
134	
135	def print_all(+f: O.Fmt, +evs: List<&2, Json.Json>) -> Act.of(Unit):
136	  match evs:
137	    case Nil{}:
138	      Act.pure(Unit, Unit{})
139	    case Con{e, t}:
140	      do ActM<Unit>:
141	        Act.lift(Unit, IO.print(show(f, e)))
142	        print_all(f, t)
143	
144	# ---- one tick, never raising ------------------------------------------------------------------------------
145	
146	def failed_out(+env: Tk.Env, +ts: String, +msg: String) -> Tk.Out:
147	  Tk.Out{Dc.TickRes{Cad.Info{Cad.Failed{}, None{}, False{}, False{}, None{}}, [Ev.error(ts, msg, True{})]}, Tk.env_mem(env), None{}, [], Json.arr([]), Tk.env_checked(env)}
148	
149	def prepend(+extra: List<&2, Json.Json>, +o: Tk.Out) -> Tk.Out:
150	  match o:
151	    case Tk.Out{Dc.TickRes{i, evs}, m, a, es, ac, ck}:
152	      Tk.Out{Dc.TickRes{i, List.append(&2, Json.Json, extra, evs)}, m, a, es, ac, ck}
153	
154	def tick_out(+env: Tk.Env, +ts: String, r: Fs.Res(Tk.Out)) -> Tk.Out:
155	  match r:
156	    case Done{o}:
157	      o
158	    case Fail{(code, msg)}:
159	      failed_out(env, ts, msg)
160	
161	def release_with(+env: Tk.Env, +t: U32, +sq: Maybe<&2, Json.Json>) -> Act.of(List<&2, Json.Json>):
162	  match sq:
163	    case None{}:
164	      Act.pure(List<&2, Json.Json>, [])
165	    case Some{seq}:
166	      Q.release_recovered(SBase.root_of(Tk.env_ctx(env)), seq, Clock.iso8601_utc(t))
167	
168	def release_now(+env: Tk.Env, +t: U32) -> Act.of(List<&2, Json.Json>):
169	  do ActM<List<&2, Json.Json>>:
170	    sq : Maybe<&2, Json.Json> <- SIo.read_seq(SBase.root_of(Tk.env_ctx(env)))
171	    release_with(env, t, sq)
172	
173	# Quarantines that no longer hold (the account was replaced, or its login was
174	# renewed) are released before deciding; a dry run changes nothing.
175	def released(+env: Tk.Env, +t: U32) -> Act.of(List<&2, Json.Json>):
176	  Act.cond(List<&2, Json.Json>, Tk.env_dry(env), Act.pure(List<&2, Json.Json>, []), release_now(env, t))
177	
178	def fetched(+env: Tk.Env, +t: U32) -> Act.of(Maybe<&2, G.Snap>):
179	  do ActM<Maybe<&2, G.Snap>>:
180	    m0 : Maybe<&2, G.Snap> <- G.snapshot(Tk.env_ctx(env))
181	    Col.refresh_opt(Tk.env_ctx(env), Tk.env_cfg(env), t, m0)
182	
183	# Release what no longer holds, snapshot, fetch what is due, decide, and walk the ranking.
184	def look(+env: Tk.Env) -> Act.of(Tk.Out):
185	  do ActM<Tk.Out>:
186	    +t : U32 <- Act.lift(U32, Clock.now_s())
187	    rel : List<&2, Json.Json> <- released(env, t)
188	    m : Maybe<&2, G.Snap> <- fetched(env, t)
189	    o : Tk.Out <- Wk.run(env, t, Clock.iso8601_utc(t), m)
190	    return prepend(rel, o)
191	
192	def tick_safe(+env: Tk.Env) -> Act.of(Tk.Out):
193	  do ActM<Tk.Out>:
194	    t : U32 <- Act.lift(U32, Clock.now_s())
195	    r : Fs.Res(Tk.Out) <- Act.lift(Fs.Res(Tk.Out), look(env))
196	    return tick_out(env, Clock.iso8601_utc(t), r)
197	
198	# ---- after a tick: status file and log --------------------------------------------------------------------
199	
200	def out_info(+o: Tk.Out) -> Cad.Info:
201	  match o:
202	    case Tk.Out{Dc.TickRes{info, evs}, m, a, es, ac, ck}:
203	      info
204	
205	def out_events(+o: Tk.Out) -> List<&2, Json.Json>:
206	  match o:
207	    case Tk.Out{Dc.TickRes{info, evs}, m, a, es, ac, ck}:
208	      evs
209	
210	def out_mem(+o: Tk.Out) -> T.Mem:
211	  match o:
212	    case Tk.Out{r, m, a, es, ac, ck}:
213	      m
214	
215	def out_active(+o: Tk.Out) -> Maybe<&2, String>:
216	  match o:
217	    case Tk.Out{r, m, a, es, ac, ck}:
218	      a
219	
220	def out_entries(+o: Tk.Out) -> List<&2, Stat.Entry>:
221	  match o:
222	    case Tk.Out{r, m, a, es, ac, ck}:
223	      es
224	
225	def out_checked(+o: Tk.Out) -> Bool:
226	  match o:
227	    case Tk.Out{r, m, a, es, ac, ck}:
228	      ck
229	
230	def out_accts(+o: Tk.Out) -> Json.Json:
231	  match o:
232	    case Tk.Out{r, m, a, es, ac, ck}:
233	      ac
234	
235	def info_outcome(i: Cad.Info) -> Cad.Outcome:
236	  match i:
237	    case Cad.Info{o, u, l, idle, np}:
238	      o
239	
240	def outcome_name(o: Cad.Outcome) -> String:
241	  match o:
242	    case Cad.Switched{}:
243	      "switched"
244	    case Cad.NoAction{}:
245	      "no-action"
246	    case Cad.Blocked{}:
247	      "blocked"
248	    case Cad.Failed{}:
249	      "failed"
250	
251	def status_of(+r: Run, +o: Tk.Out, +ts: String, +next_at: String) -> Stat.Status:
252	  Stat.Status{run_pid(r), run_version(r), run_started(r), Some{ts}, Some{next_at}, out_active(o), out_entries(o), run_cfg(r)}
253	
254	# A blocked tick is a warning, a failed one an error.
255	def level_of(o: Cad.Outcome) -> String:
256	  match o:
257	    case Cad.Blocked{}:
258	      "warn"
259	    case Cad.Failed{}:
260	      "error"
261	    case _:
262	      "info"
263	
264	# One JSONL log line: the tick, what came of it, and per account where its
265	# reading came from (usage-endpoint, header-probe or cache).
266	def log_json(+ts: String, +o: Tk.Out) -> Json.Json:
267	  Json.JObj{Json.JField{"ts", Json.JStr{ts}, Json.JField{"level", Json.JStr{level_of(info_outcome(out_info(o)))}, Json.JField{"msg", Json.JStr{"tick"},
268	    Json.JField{"outcome", Json.JStr{outcome_name(info_outcome(out_info(o)))}, Json.JField{"active", Ev.opt_str(out_active(o)), Json.JField{"accounts", out_accts(o), Json.JNil{}}}}}}}}
269	
270	def swallow(r: Fs.Res(Unit)) -> Act.of(Unit):
271	  Act.pure(Unit, Unit{})
272	
273	def best_effort(m: Act.of(Unit)) -> Act.of(Unit):
274	  IO.bind(Fs.Res(Unit), Fs.Res(Unit), m, r => swallow(r))
275	
276	def write_status(+r: Run, +o: Tk.Out, +ts: String, +delay: U32, +now: U32) -> Act.of(Unit):
277	  best_effort(Stat.write(run_dir(r), status_of(r, o, ts, Clock.iso8601_utc((now + delay : U32)))))
278	
279	def write_log(+r: Run, +o: Tk.Out, +ts: String) -> Act.of(Unit):
280	  best_effort(FsW.append(run_dir(r) ++ "/daemon.log", Json.emit_py(Lv.tag_log(Lv.live_tag(run_live(r)), log_json(ts, o))) ++ "\n"))
281	
282	def record_tick(+r: Run, +o: Tk.Out, +ts: String, +delay: U32, +now: U32) -> Act.of(Unit):
283	  do ActM<Unit>:
284	    write_status(r, o, ts, delay, now)
285	    write_log(r, o, ts)
286	
287	def sleep_if(+c: Bool, +e: Json.Json) -> List<&2, Json.Json>:
288	  match c:
289	    case True{}:
290	      [e]
291	    case False{}:
292	      []
293	
294	def sleep_event(r: Run, +delay: U32, +now: U32) -> List<&2, Json.Json>:
295	  sleep_if(Cad.announce(delay, interval(r)), Ev.sleep(Clock.iso8601_utc(now), (delay * 10 : U32), Clock.iso8601_utc((now + delay : U32))))
296	
297	def env_of(+r: Run, +mem: T.Mem, +checked: Bool) -> Tk.Env:
298	  Tk.Env{run_ctx(r), run_cfg(r), run_dry(r), run_opts(r), mem, checked, []}
299	
300	# ---- --once --------------------------------------------------------------------------------------------------
301	
302	def finish_tick(+r: Run, +o: Tk.Out, +delay: U32, +now: U32, +extra: List<&2, Json.Json>) -> Act.of(Unit):
303	  do ActM<Unit>:
304	    print_all(run_fmt(r), List.append(&2, Json.Json, out_events(o), extra))
305	    record_tick(r, o, Clock.iso8601_utc(now), delay, now)
306	
307	def first_tick(+r: Run) -> Act.of(Tk.Out):
308	  do ActM<Tk.Out>:
309	    +o : Tk.Out <- tick_safe(env_of(r, T.fresh_mem(), False{}))
310	    return tagged(r, o)
311	
312	def out_code(+o: Tk.Out) -> U32:
313	  Cad.exit_code(info_outcome(out_info(o)))
314	
315	def once(+r: Run) -> Act.of(U32):
316	  do ActM<U32>:
317	    +r2 : Run <- refresh(r)
318	    +o : Tk.Out <- first_tick(r2)
319	    now : U32 <- Act.lift(U32, Clock.now_s())
320	    finish_tick(r2, o, interval(r2), now, [])
321	    return out_code(o)
322	
323	# ---- the loop ------------------------------------------------------------------------------------------------
324	
325	# Tick, record, sleep, repeat. Not terminating by construction: kept in this
326	# module, which no proof imports.
327	def nap(+delay: U32) -> Act.of(Unit):
328	  Act.lift(Unit, IO.sleep((delay * 1000 : U32)))
329	
330	def pace(+r: Run, +o: Tk.Out, +now: U32) -> Act.of(Unit):
331	  do ActM<Unit>:
332	    +delay : U32 = Cad.next_delay(out_info(o), now, interval(r))
333	    finish_tick(r, o, delay, now, sleep_event(r, delay, now))
334	    nap(delay)
335	
336	def settle(r: Run, +o: Tk.Out) -> Act.of(Unit):
337	  do ActM<Unit>:
338	    now : U32 <- Act.lift(U32, Clock.now_s())
339	    pace(r, o, now)
340	
341	def tick_tagged(+r: Run, +mem: T.Mem, +checked: Bool) -> Act.of(Tk.Out):
342	  do ActM<Tk.Out>:
343	    +o : Tk.Out <- tick_safe(env_of(r, mem, checked))
344	    return tagged(r, o)
345	
346	def round(+r: Run, +mem: T.Mem, +checked: Bool) -> Act.of(Tk.Out):
347	  do ActM<Tk.Out>:
348	    +r2 : Run <- refresh(r)
349	    +o : Tk.Out <- tick_tagged(r2, mem, checked)
350	    settle(r2, o)
351	    return o
352	
353	@unsafe def forever(+r: Run, mem: T.Mem, checked: Bool) -> Act.of(U32):
354	  do ActM<U32>:
355	    +o : Tk.Out <- round(r, mem, checked)
356	    forever(r, out_mem(o), out_checked(o))