repos / starfx

supervisor.ts

 1import {
 2  type Callable,
 3  type Operation,
 4  type Task,
 5  call,
 6  race,
 7  sleep,
 8  spawn,
 9} from "effection";
10import { createAction, take } from "./action.js";
11import { getIdFromAction } from "./action.js";
12import type { CreateActionPayload } from "./query/index.js";
13import type { ActionWithPayload, AnyAction } from "./types.js";
14
15const MS = 1000;
16const SECONDS = 1 * MS;
17const MINUTES = 60 * SECONDS;
18
19export function poll(parentTimer: number = 5 * SECONDS, cancelType?: string) {
20  return function* poller<T>(
21    actionType: string,
22    op: (action: AnyAction) => Operation<T>,
23  ): Operation<T> {
24    const cancel = cancelType || actionType;
25    function* fire(action: { type: string }, timer: number) {
26      while (true) {
27        yield* op(action);
28        yield* sleep(timer);
29      }
30    }
31
32    while (true) {
33      const action = yield* take<{ timer?: number }>(actionType);
34      const timer = action.payload?.timer || parentTimer;
35      yield* race([fire(action, timer), take(`${cancel}`) as Operation<void>]);
36    }
37  };
38}
39
40type ClearTimerPayload = string | { type: string; payload: { key: string } };
41
42export const clearTimers = createAction<
43  ClearTimerPayload | ClearTimerPayload[]
44>("clear-timers");
45
46/**
47 * timer() will create a cache timer for each `key` inside
48 * of a starfx api endpoint.  `key` is a hash of the action type and payload.
49 *
50 * Why do we want this?  If we have an api endpoint to fetch a single app: `fetchApp({ id: 1 })`
51 * if we don't set a timer per key then all calls to `fetchApp` will be on a timer.
52 * So if we call `fetchApp({ id: 1 })` and then `fetchApp({ id: 2 })` if we use a normal
53 * cache timer then the second call will not send an http request.
54 */
55export function timer(timer: number = 5 * MINUTES) {
56  return function* onTimer(
57    actionType: string,
58    op: (action: AnyAction) => Callable<unknown>,
59  ) {
60    const map: { [key: string]: Task<unknown> } = {};
61
62    function* activate(action: ActionWithPayload<CreateActionPayload>) {
63      yield* call(() => op(action));
64      const idA = getIdFromAction(action);
65
66      const matchFn = (
67        act: ActionWithPayload<ClearTimerPayload | ClearTimerPayload[]>,
68      ) => {
69        if (act.type !== `${clearTimers}`) return false;
70        if (!act.payload) return false;
71        const ids = Array.isArray(act.payload) ? act.payload : [act.payload];
72        return ids.some((id) => {
73          if (id === "*") {
74            return true;
75          }
76          if (typeof id === "string") {
77            return idA === id;
78          }
79          return idA === getIdFromAction(id);
80        });
81      };
82      yield* race([sleep(timer), take(matchFn as any) as Operation<void>]);
83
84      delete map[action.payload.key];
85    }
86
87    while (true) {
88      const action = yield* take<CreateActionPayload>(`${actionType}`);
89      const key = action.payload.key;
90      if (!map[key]) {
91        const task = yield* spawn(function* () {
92          yield* activate(action);
93        });
94        map[key] = task;
95      }
96    }
97  };
98}