1import * as Effect from "effect/Effect";
2import * as Stream from "effect/Stream";
3import * as SubscriptionRef from "effect/SubscriptionRef";
4import { AsyncResult, Atom } from "effect/reactivity";
5
6import * as RelayEnvironmentDiscovery from "../relay/discovery.ts";
7import { createRuntimeCommand } from "./runtime.ts";
8
9export function createRelayEnvironmentDiscoveryAtoms<R, E>(
10 runtime: Atom.AtomRuntime<RelayEnvironmentDiscovery.RelayEnvironmentDiscovery | R, E>,
11) {
12 const stateAtom = runtime.atom(
13 Stream.unwrap(
14 RelayEnvironmentDiscovery.RelayEnvironmentDiscovery.pipe(
15 Effect.map((discovery) => SubscriptionRef.changes(discovery.state)),
16 ),
17 ),
18 { initialValue: RelayEnvironmentDiscovery.EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE },
19 );
20 const stateValueAtom = Atom.make((get) =>
21 AsyncResult.getOrElse(
22 get(stateAtom),
23 () => RelayEnvironmentDiscovery.EMPTY_RELAY_ENVIRONMENT_DISCOVERY_STATE,
24 ),
25 ).pipe(Atom.withLabel("relay-environment-discovery-value"));
26 const refresh = createRuntimeCommand(runtime, {
27 label: "relay-environment-discovery:refresh",
28 concurrency: { mode: "singleFlight", key: () => "refresh" },
29 execute: (_input: void) =>
30 RelayEnvironmentDiscovery.RelayEnvironmentDiscovery.pipe(
31 Effect.flatMap((discovery) => discovery.refresh),
32 ),
33 });
34
35 const refreshAfterMutation = createRuntimeCommand(runtime, {
36 label: "relay-environment-discovery:refresh-after-mutation",
37 concurrency: { mode: "serial", key: () => "refresh-after-mutation" },
38 execute: (_input: void) =>
39 RelayEnvironmentDiscovery.RelayEnvironmentDiscovery.pipe(
40 Effect.flatMap((discovery) => discovery.refresh),
41 ),
42 });
43
44 return {
45 stateAtom,
46 stateValueAtom,
47 refresh,
48 refreshAfterMutation,
49 };
50}