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/base.bend as SBase
8	import ../swlive.bend as L
9	import ../policy/types.bend as T
10	import ./events.bend as Ev
11	import ./config.bend as Cfg
12	import ./dec.bend as Dc
13	import ./tick.bend as Tk
14	import ./gather.bend as G
15	import ./pidlock.bend as Pid
16	
17	# `cswap auto --shadow --shadow-log PATH`: the decider run for comparison only.
18	# It reads the usage cache the other poller keeps (no HTTP, no fetch), decides
19	# with the same policy, and appends what it WOULD have done as one JSONL line
20	# per tick to PATH. It never switches, quarantines, releases or touches state,
21	# status, cache or log under the backup root; its only writes are PATH and a
22	# pid/lock pair beside it.
23	
24	# The headroom (slot -> percent) the tick used: the poll event's headroomPct.
25	
26	type Step is Data:
27	  Step{line: Json.Json, mem: T.Mem, checked: Bool}
28	
29	type Sh is Data:
30	  Sh{ctx: SBase.Ctx, cfg: Cfg.Cfg, opts: L.Opts, log: String}
31	
32	def field2(+m: Maybe<&2, Json.Json>) -> Json.Json:
33	  match m:
34	    case None{}:
35	      Json.JNull{}
36	    case Some{j}:
37	      j
38	
39	def field(+k: String, +e: Json.Json) -> Json.Json:
40	  field2(Json.get(k, e))
41	
42	def headroom_of(+evs: List<&2, Json.Json>) -> Json.Json:
43	  match evs:
44	    case Nil{}:
45	      Json.JObj{Json.JNil{}}
46	    case Con{e, t}:
47	      field("headroomPct", e)
48	
49	def ActM(-A: Type) -> Type:
50	  Act.of(A)
51	
52	def ActM.pure(-A: Type, x: A) -> Act.of(A):
53	  Act.pure(A, x)
54	
55	def ActM.bind(-A: Type, -B: Type, m: Act.of(A), f: A -> Act.of(B)) -> Act.of(B):
56	  Act.bind(A, B, m, f)
57	
58	def last_event(+es: List<&2, Json.Json>, +d: Json.Json) -> Json.Json:
59	  match es:
60	    case Nil{}:
61	      d
62	    case Con{e, t}:
63	      last_event(t, e)
64	
65	def ctx_active(+c: Dc.Ctx) -> Json.Json:
66	  match c:
67	    case Dc.Ctx{x, d, cf, n, ts, o, a, np}:
68	      a
69	
70	def out_events(+o: Tk.Out) -> List<&2, Json.Json>:
71	  match o:
72	    case Tk.Out{Dc.TickRes{i, evs}, m, a, es, ac, ck}:
73	      evs
74	
75	def out_active(+o: Tk.Out) -> Maybe<&2, String>:
76	  match o:
77	    case Tk.Out{r, m, a, es, ac, ck}:
78	      a
79	
80	def out_accts(+o: Tk.Out) -> Json.Json:
81	  match o:
82	    case Tk.Out{r, m, a, es, ac, ck}:
83	      ac
84	
85	def out_mem(+o: Tk.Out) -> T.Mem:
86	  match o:
87	    case Tk.Out{r, m, a, es, ac, ck}:
88	      m
89	
90	def out_checked(+o: Tk.Out) -> Bool:
91	  match o:
92	    case Tk.Out{r, m, a, es, ac, ck}:
93	      ck
94	
95	# The decision as the event the live daemon would have logged.
96	def decision_event(+ts: String, +c: Dc.Ctx, +d: Dc.Dec) -> Json.Json:
97	  match d:
98	    case Dc.DNone{reason, detail}:
99	      Ev.no_switch(ts, reason, detail)
100	    case Dc.DExhausted{e}:
101	      Ev.no_switch(ts, "all-exhausted", "")
102	    case Dc.DGo{to, email, trig, lh, lr}:
103	      Ev.switch(ts, trig, ctx_active(c), Ev.ref(to, email), [], True{})
104	
105	def line(+ts: String, +o: Tk.Out, +dec: Json.Json, +evs: List<&2, Json.Json>) -> Json.Json:
106	  Json.JObj{Json.JField{"ts", Json.JStr{ts}, Json.JField{"active", Ev.opt_str(out_active(o)),
107	    Json.JField{"event", field("event", dec), Json.JField{"reason", field("reason", dec), Json.JField{"trigger", field("trigger", dec),
108	    Json.JField{"from", field("from", dec), Json.JField{"to", field("to", dec), Json.JField{"accounts", headroom_of(evs),
109	    Json.JField{"sources", out_accts(o), Json.JField{"events", Json.arr(evs), Json.JNil{}}}}}}}}}}}}
110	
111	def step_of(+ts: String, +o: Tk.Out, +dec: Json.Json, +evs: List<&2, Json.Json>) -> Step:
112	  Step{line(ts, o, dec, evs), out_mem(o), out_checked(o)}
113	
114	def shadow2(+ts: String, +nx: Tk.Next) -> Step:
115	  match nx:
116	    case Tk.NDone{o}:
117	      step_of(ts, o, last_event(out_events(o), Json.JNull{}), out_events(o))
118	    case Tk.NAct{o, c, d}:
119	      step_of(ts, o, decision_event(ts, c, d), List.append(&2, Json.Json, out_events(o), [decision_event(ts, c, d)]))
120	
121	# What this tick would do: the finished outcome's own last event, or the
122	# decision it stopped at. Pure: nothing is fetched, applied or written.
123	def shadow_step(+env: Tk.Env, +t: U32, +m: Maybe<&2, G.Snap>) -> Step:
124	  shadow2(Clock.iso8601_utc(t), Tk.tick(env, t, m))
125	
126	def step_line(+s: Step) -> String:
127	  match s:
128	    case Step{l, m, c}:
129	      Json.emit_py(l) ++ "\n"
130	
131	def step_mem(+s: Step) -> T.Mem:
132	  match s:
133	    case Step{l, m, c}:
134	      m
135	
136	def step_checked(+s: Step) -> Bool:
137	  match s:
138	    case Step{l, m, c}:
139	      c
140	
141	def swallow(r: Fs.Res(Unit)) -> Act.of(Unit):
142	  Act.pure(Unit, Unit{})
143	
144	# ---- the loop ------------------------------------------------------------------------------------------------
145	
146	def sh_env(+r: Sh, +mem: T.Mem, +checked: Bool) -> Tk.Env:
147	  match r:
148	    case Sh{c, cf, o, l}:
149	      Tk.Env{c, cf, True{}, o, mem, checked, []}
150	
151	def sh_ctx(+r: Sh) -> SBase.Ctx:
152	  match r:
153	    case Sh{c, cf, o, l}:
154	      c
155	
156	def sh_log(+r: Sh) -> String:
157	  match r:
158	    case Sh{c, cf, o, l}:
159	      l
160	
161	def sh_interval(+r: Sh) -> U32:
162	  match r:
163	    case Sh{c, cf, o, l}:
164	      Cfg.cfg_interval(cf)
165	
166	def emit(+r: Sh, +s: Step) -> Act.of(Step):
167	  IO.bind(Fs.Res(Unit), Fs.Res(Step), FsW.append(sh_log(r), step_line(s)), u => IO.pure(Fs.Res(Step), Done{s}))
168	
169	def one_tick(+r: Sh, +mem: T.Mem, +checked: Bool) -> Act.of(Step):
170	  do ActM<Step>:
171	    +t : U32 <- Act.lift(U32, Clock.now_s())
172	    m : Maybe<&2, G.Snap> <- G.snapshot(sh_ctx(r))
173	    emit(r, shadow_step(sh_env(r, mem, checked), t, m))
174	
175	def once(+r: Sh) -> Act.of(U32):
176	  do ActM<U32>:
177	    s : Step <- one_tick(r, T.fresh_mem(), False{})
178	    return 0
179	
180	def nap(+delay: U32) -> Act.of(Unit):
181	  Act.lift(Unit, IO.sleep((delay * 1000 : U32)))
182	
183	@unsafe def forever(+r: Sh, mem: T.Mem, checked: Bool) -> Act.of(U32):
184	  do ActM<U32>:
185	    +s : Step <- one_tick(r, mem, checked)
186	    nap(sh_interval(r))
187	    forever(r, step_mem(s), step_checked(s))
188	
189	def body(+r: Sh, +once_only: Bool) -> Act.of(U32):
190	  Act.cond(U32, once_only, once(r), forever(r, T.fresh_mem(), False{}))
191	
192	# Run as the one shadow decider of the log's directory.
193	def run(+r: Sh, +once_only: Bool) -> Act.of(U32):
194	  Pid.with_pidlock(U32, Fs.dirname(sh_log(r)), body(r, once_only))