1import * as Effect from "effect/Effect";
2import * as Option from "effect/Option";
3import * as Stream from "effect/Stream";
4import * as SubscriptionRef from "effect/SubscriptionRef";
5import { AsyncResult, Atom } from "effect/reactivity";
6import {
7 AuthOrchestrationOperateScope,
8 HUB_THREAD_ORGANIZATION_FIELDS,
9 type EnvironmentId,
10 type ExecutionEnvironmentCapabilities,
11 type HubInboxEnvironment,
12 type ServerConfig,
13} from "@t3tools/contracts";
14import * as Equal from "effect/Equal";
15import * as Struct from "effect/Struct";
16
17import { HubConnection, EMPTY_HUB_CONNECTION_STATE } from "../connection/hub.ts";
18import type { PairingConnectionInput } from "../connection/onboarding.ts";
19import { createAtomCommandScheduler, createRuntimeCommand } from "./runtime.ts";
20import type { EnvironmentShellState } from "./shell.ts";
21
22export type { HubConnectionState, HubServerConnection } from "../connection/hub.ts";
23export type { PairingConnectionInput } from "../connection/onboarding.ts";
24
25export function applyHubInboxEnvironment(
26 state: EnvironmentShellState,
27 entry: HubInboxEnvironment | undefined,
28): EnvironmentShellState {
29 if (entry === undefined) return state;
30 const local = Option.getOrNull(state.snapshot);
31 const snapshot =
32 local !== null &&
33 (state.status === "live" || local.snapshotSequence > entry.snapshot.snapshotSequence)
34 ? local
35 : entry.snapshot;
36 const pending = new Set(entry.pendingThreadIds);
37 const organization = new Map(entry.snapshot.threads.map((thread) => [thread.id, thread]));
38 let changed = false;
39 const threads = snapshot.threads.map((thread) => {
40 const central = organization.get(thread.id);
41 if (!pending.has(thread.id) || central === undefined) return thread;
42 const fields = Struct.pick(central, HUB_THREAD_ORGANIZATION_FIELDS);
43 if (Equal.equals(fields, Struct.pick(thread, HUB_THREAD_ORGANIZATION_FIELDS))) return thread;
44 changed = true;
45 return { ...thread, ...fields };
46 });
47 const next = changed ? { ...snapshot, threads } : snapshot;
48 if (local === next) return state;
49 return {
50 ...state,
51 snapshot: Option.some(next),
52 status: state.status === "live" ? "live" : "cached",
53 };
54}
55
56export function createHubConnectionAtoms<R, E>(runtime: Atom.AtomRuntime<HubConnection | R, E>) {
57 const stateAtom = runtime.atom(
58 Stream.unwrap(HubConnection.pipe(Effect.map((hub) => SubscriptionRef.changes(hub.state)))),
59 { initialValue: EMPTY_HUB_CONNECTION_STATE },
60 );
61 const stateValueAtom = Atom.make((get) =>
62 Option.getOrElse(AsyncResult.value(get(stateAtom)), () => EMPTY_HUB_CONNECTION_STATE),
63 );
64 const inboxValueAtom = Atom.make((get) => get(stateValueAtom).inbox);
65 const organizationEnvironmentIdsAtom = Atom.make((get) => {
66 const state = get(stateValueAtom);
67 if (state.connection === null) return new Set<EnvironmentId>();
68 const permitted = new Set(
69 state.directory?.environments.flatMap((entry) =>
70 entry.scopes.includes(AuthOrchestrationOperateScope) ? [entry.environmentId] : [],
71 ),
72 );
73 return new Set(
74 state.inbox.environments.flatMap((entry) =>
75 permitted.has(entry.environmentId) ? [entry.environmentId] : [],
76 ),
77 );
78 }).pipe(
79 Atom.withEquality<Set<EnvironmentId>>(
80 (left, right) => left.size === right.size && [...left].every((id) => right.has(id)),
81 ),
82 );
83 const organizationPermissionAtom = Atom.family((environmentId: EnvironmentId | null) =>
84 Atom.make((get) =>
85 environmentId === null ? false : get(organizationEnvironmentIdsAtom).has(environmentId),
86 ),
87 );
88 const scheduler = createAtomCommandScheduler();
89 const concurrency = { mode: "serial" as const, key: () => "hub" };
90 const connect = createRuntimeCommand(runtime, {
91 label: "hub:connect",
92 scheduler,
93 concurrency,
94 execute: (input: PairingConnectionInput) =>
95 HubConnection.pipe(Effect.flatMap((hub) => hub.connect(input))),
96 });
97 const connectAuthenticated = createRuntimeCommand(runtime, {
98 label: "hub:sign-in",
99 scheduler,
100 concurrency,
101 execute: (input: { readonly httpBaseUrl: string }) =>
102 HubConnection.pipe(Effect.flatMap((hub) => hub.connectAuthenticated(input))),
103 });
104 const refresh = createRuntimeCommand(runtime, {
105 label: "hub:refresh",
106 scheduler,
107 concurrency,
108 execute: (_input: void) => HubConnection.pipe(Effect.flatMap((hub) => hub.refresh)),
109 });
110 const disconnect = createRuntimeCommand(runtime, {
111 label: "hub:disconnect",
112 scheduler,
113 concurrency,
114 execute: (_input: void) => HubConnection.pipe(Effect.flatMap((hub) => hub.disconnect)),
115 });
116 return {
117 stateAtom,
118 stateValueAtom,
119 inboxValueAtom,
120 organizationEnvironmentIdsAtom,
121 organizationPermissionAtom,
122 connect,
123 connectAuthenticated,
124 refresh,
125 disconnect,
126 };
127}
128
129export function createHubEnvironmentCapabilitiesAtom(
130 configsAtom: Atom.Atom<ReadonlyMap<EnvironmentId, ServerConfig>>,
131 hub: Pick<
132 ReturnType<typeof createHubConnectionAtoms>,
133 "stateValueAtom" | "organizationEnvironmentIdsAtom"
134 >,
135) {
136 return Atom.make((get): ReadonlyMap<EnvironmentId, ExecutionEnvironmentCapabilities> => {
137 const capabilities = new Map(
138 get(hub.stateValueAtom).directory?.environments.map((entry) => [
139 entry.environmentId,
140 entry.descriptor.capabilities,
141 ]),
142 );
143 for (const [id, config] of get(configsAtom))
144 capabilities.set(id, config.environment.capabilities);
145 for (const id of get(hub.organizationEnvironmentIdsAtom)) {
146 const current = capabilities.get(id);
147 if (current === undefined) continue;
148 capabilities.set(id, {
149 ...current,
150 threadSettlement: true,
151 threadSnooze: true,
152 threadPinning: true,
153 threadPinReorder: true,
154 threadActiveReorder: true,
155 threadAutoSettleOptOut: true,
156 });
157 }
158 return capabilities;
159 }).pipe(
160 Atom.withEquality<ReadonlyMap<EnvironmentId, ExecutionEnvironmentCapabilities>>(
161 (left, right) =>
162 left.size === right.size &&
163 [...left].every(([id, capabilities]) => {
164 const other = right.get(id);
165 return (
166 other !== undefined &&
167 Object.keys(capabilities).length === Object.keys(other).length &&
168 Object.entries(capabilities).every(([key, value]) => Reflect.get(other, key) === value)
169 );
170 }),
171 ),
172 );
173}