1import {
2 DEFAULT_PROVIDER_HEALTH_REFRESH_INTERVAL,
3 type ServerProvider,
4 ServerSettingsError,
5} from "@t3tools/contracts";
6import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";
7import * as Duration from "effect/Duration";
8import * as Effect from "effect/Effect";
9import * as Equal from "effect/Equal";
10import * as Fiber from "effect/Fiber";
11import * as PubSub from "effect/PubSub";
12import * as Queue from "effect/Queue";
13import * as Ref from "effect/Ref";
14import * as Scope from "effect/Scope";
15import * as Stream from "effect/Stream";
16import * as Semaphore from "effect/Semaphore";
17
18import * as ProviderHost from "./ProviderHost.ts";
19import { applyUsageLimitsUpdate, resolveUsageLimitsAfterProbe } from "./usageLimits.ts";
20import type { ManagedServerProvider } from "./snapshot.ts";
21
22interface ProviderSnapshotState {
23 readonly snapshot: ServerProvider;
24 readonly enrichmentGeneration: number;
25}
26
27function withUsageLimits(
28 snapshot: ServerProvider,
29 usageLimits: ServerProvider["usageLimits"],
30): ServerProvider {
31 if (snapshot.usageLimits === usageLimits) {
32 return snapshot;
33 }
34 const { usageLimits: _previous, ...rest } = snapshot;
35 return usageLimits ? { ...rest, usageLimits } : rest;
36}
37
38export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")(function* <
39 Settings,
40>(input: {
41 readonly resolveMaintenance: ManagedServerProvider["resolveMaintenance"];
42 readonly getSettings: Effect.Effect<Settings, ServerSettingsError>;
43 readonly streamSettings: Stream.Stream<Settings>;
44 readonly haveSettingsChanged: (previous: Settings, next: Settings) => boolean;
45 readonly initialSnapshot: (settings: Settings) => Effect.Effect<ServerProvider>;
46 readonly checkProvider: Effect.Effect<ServerProvider, ServerSettingsError>;
47 readonly enrichSnapshot?: (input: {
48 readonly settings: Settings;
49 readonly snapshot: ServerProvider;
50 readonly getSnapshot: Effect.Effect<ServerProvider>;
51 readonly publishSnapshot: (snapshot: ServerProvider) => Effect.Effect<void>;
52 }) => Effect.Effect<void>;
53 readonly refreshInterval?: Duration.Input;
54 readonly refreshOnInterval?: boolean;
55 readonly checkProviderOnSettingsChange?: (previous: Settings, next: Settings) => boolean;
56}): Effect.fn.Return<
57 ManagedServerProvider,
58 ServerSettingsError,
59 Scope.Scope | ProviderHost.ProviderHost
60> {
61 const host = yield* ProviderHost.ProviderHost;
62 const refreshSemaphore = yield* Semaphore.make(1);
63 const changesPubSub = yield* Effect.acquireRelease(
64 PubSub.unbounded<ServerProvider>(),
65 PubSub.shutdown,
66 );
67 const initialSettings = yield* input.getSettings;
68 const initialSnapshot = yield* input.initialSnapshot(initialSettings);
69 const snapshotStateRef = yield* Ref.make<ProviderSnapshotState>({
70 snapshot: initialSnapshot,
71 enrichmentGeneration: 0,
72 });
73 const settingsRef = yield* Ref.make(initialSettings);
74 const enrichmentFiberRef = yield* Ref.make<Fiber.Fiber<void, unknown> | null>(null);
75 const scope = yield* Effect.scope;
76
77 const publishEnrichedSnapshot = Effect.fn("publishEnrichedSnapshot")(function* (
78 generation: number,
79 nextSnapshot: ServerProvider,
80 ) {
81 const snapshotToPublish = yield* Ref.modify(snapshotStateRef, (state) => {
82 if (state.enrichmentGeneration !== generation) {
83 return [null, state] as const;
84 }
85 // Enrichment derives from the snapshot it was handed; a runtime usage
86 // update that landed since must not be reverted by it.
87 const merged = withUsageLimits(nextSnapshot, state.snapshot.usageLimits);
88 if (Equal.equals(state.snapshot, merged)) {
89 return [null, state] as const;
90 }
91 return [merged, { ...state, snapshot: merged }] as const;
92 });
93 if (snapshotToPublish === null) {
94 return;
95 }
96 yield* PubSub.publish(changesPubSub, snapshotToPublish);
97 });
98
99 const restartSnapshotEnrichment = Effect.fn("restartSnapshotEnrichment")(function* (
100 settings: Settings,
101 snapshot: ServerProvider,
102 generation: number,
103 ) {
104 const previousFiber = yield* Ref.getAndSet(enrichmentFiberRef, null);
105 if (previousFiber) {
106 yield* Fiber.interrupt(previousFiber).pipe(Effect.ignore);
107 }
108
109 // Enrichment resolves binaries and package-manager state, which a disabled
110 // provider has no use for.
111 if (!input.enrichSnapshot || !snapshot.enabled) {
112 return;
113 }
114
115 const fiber = yield* input
116 .enrichSnapshot({
117 settings,
118 snapshot,
119 getSnapshot: Ref.get(snapshotStateRef).pipe(Effect.map((state) => state.snapshot)),
120 publishSnapshot: (nextSnapshot) => publishEnrichedSnapshot(generation, nextSnapshot),
121 })
122 .pipe(Effect.ignoreCause({ log: true }), Effect.forkIn(scope));
123
124 yield* Ref.set(enrichmentFiberRef, fiber);
125 });
126
127 const applySnapshotBase = Effect.fn("applySnapshot")(function* (
128 nextSettings: Settings,
129 options?: { readonly forceRefresh?: boolean },
130 ) {
131 const forceRefresh = options?.forceRefresh === true;
132 const previousSettings = yield* Ref.get(settingsRef);
133 if (!forceRefresh && !input.haveSettingsChanged(previousSettings, nextSettings)) {
134 yield* Ref.set(settingsRef, nextSettings);
135 return yield* Ref.get(snapshotStateRef).pipe(Effect.map((state) => state.snapshot));
136 }
137
138 if (
139 !forceRefresh &&
140 input.checkProviderOnSettingsChange?.(previousSettings, nextSettings) === false
141 ) {
142 const state = yield* Ref.get(snapshotStateRef);
143 const nextGeneration = state.enrichmentGeneration + 1;
144 yield* Ref.set(snapshotStateRef, {
145 ...state,
146 enrichmentGeneration: nextGeneration,
147 });
148 yield* Ref.set(settingsRef, nextSettings);
149 yield* restartSnapshotEnrichment(nextSettings, state.snapshot, nextGeneration);
150 return state.snapshot;
151 }
152
153 const probedSnapshot = yield* input.checkProvider;
154 const { snapshot: nextSnapshot, generation: nextGeneration } = yield* Ref.modify(
155 snapshotStateRef,
156 (state) => {
157 const generation = input.enrichSnapshot
158 ? state.enrichmentGeneration + 1
159 : state.enrichmentGeneration;
160 // A failed read keeps the last good limits only for the same account,
161 // and they keep the workspace they were read for.
162 const previous = state.snapshot.auth;
163 const differs = (a: string | undefined, b: string | undefined) =>
164 a !== undefined && b !== undefined && a !== b;
165 // Limits read without a workspace cannot be attributed to a new one.
166 const switchedAccount =
167 differs(probedSnapshot.auth.email?.toLowerCase(), previous.email?.toLowerCase()) ||
168 (probedSnapshot.auth.workspaceId !== undefined &&
169 probedSnapshot.auth.workspaceId !== previous.workspaceId);
170 const usageLimits = switchedAccount
171 ? probedSnapshot.usageLimits
172 : resolveUsageLimitsAfterProbe({
173 published: state.snapshot.usageLimits,
174 probed: probedSnapshot.usageLimits,
175 });
176 const workspaceId =
177 usageLimits === probedSnapshot.usageLimits ? undefined : state.snapshot.auth.workspaceId;
178 const snapshot = withUsageLimits(
179 workspaceId && !probedSnapshot.auth.workspaceId
180 ? { ...probedSnapshot, auth: { ...probedSnapshot.auth, workspaceId } }
181 : probedSnapshot,
182 usageLimits,
183 );
184 return [
185 { snapshot, generation },
186 { snapshot, enrichmentGeneration: generation },
187 ] as const;
188 },
189 );
190 yield* Ref.set(settingsRef, nextSettings);
191 yield* PubSub.publish(changesPubSub, nextSnapshot);
192 yield* restartSnapshotEnrichment(nextSettings, nextSnapshot, nextGeneration);
193 return nextSnapshot;
194 });
195 const applySnapshot = (nextSettings: Settings, options?: { readonly forceRefresh?: boolean }) =>
196 refreshSemaphore.withPermits(1)(applySnapshotBase(nextSettings, options));
197
198 /**
199 * Runtime usage updates arrive between probes. They patch only
200 * `usageLimits` on whatever snapshot is published and leave the enrichment
201 * generation alone, so an in-flight enrichment still lands.
202 */
203 const applyUsageLimits: ManagedServerProvider["applyUsageLimits"] = (update) =>
204 Effect.gen(function* () {
205 const snapshotToPublish = yield* Ref.modify(snapshotStateRef, (state) => {
206 const usageLimits = applyUsageLimitsUpdate({
207 previous: state.snapshot.usageLimits,
208 update,
209 checkedAt: update.checkedAt,
210 });
211 // `applyUsageLimitsUpdate` hands back the same object when nothing
212 // moved, which is the common case for Codex's per-tick notification.
213 if (usageLimits === state.snapshot.usageLimits) {
214 return [null, state] as const;
215 }
216 const snapshot = withUsageLimits(state.snapshot, usageLimits);
217 return [snapshot, { ...state, snapshot }] as const;
218 });
219 if (snapshotToPublish !== null) {
220 yield* PubSub.publish(changesPubSub, snapshotToPublish);
221 }
222 });
223
224 const refreshSnapshot = Effect.fn("refreshSnapshot")(function* () {
225 const nextSettings = yield* input.getSettings;
226 return yield* applySnapshot(nextSettings, { forceRefresh: true });
227 });
228
229 const hasProviderStatusDemand = Effect.gen(function* () {
230 const state = yield* Ref.get(snapshotStateRef);
231 const instanceId = state.snapshot.instanceId;
232 const [genericDemand, instanceDemand] = yield* Effect.all([
233 host.shouldRunBackgroundWork({ type: "provider-status" }),
234 host.shouldRunBackgroundWork({ type: "provider-status", instanceId }),
235 ]);
236 return genericDemand || instanceDemand;
237 });
238
239 const getRefreshInterval =
240 input.refreshInterval !== undefined
241 ? Effect.succeed(input.refreshInterval)
242 : host.settings.get.pipe(
243 Effect.map(
244 (settings) =>
245 resolveServerBackgroundActivitySettings(settings).providerHealthRefreshInterval,
246 ),
247 Effect.orElseSucceed(() => DEFAULT_PROVIDER_HEALTH_REFRESH_INTERVAL),
248 );
249
250 const refreshIntervalChanges = yield* Queue.sliding<void>(1);
251 if (input.refreshInterval === undefined) {
252 const serverSettingsChanges = yield* host.settings.subscribe;
253 yield* serverSettingsChanges.pipe(
254 Stream.map((settings) =>
255 Duration.toMillis(
256 resolveServerBackgroundActivitySettings(settings).providerHealthRefreshInterval,
257 ),
258 ),
259 Stream.changes,
260 Stream.runForEach(() => Queue.offer(refreshIntervalChanges, undefined).pipe(Effect.asVoid)),
261 Effect.forkScoped,
262 );
263 }
264
265 yield* Stream.runForEach(input.streamSettings, (nextSettings) =>
266 Effect.asVoid(applySnapshot(nextSettings)),
267 ).pipe(Effect.forkScoped);
268
269 yield* Effect.forever(
270 getRefreshInterval.pipe(
271 Effect.flatMap((refreshInterval) =>
272 // @effect-diagnostics-next-line raceFirstWithSleepToTimeout:off - races the interval against a settings-change signal, not a timeout
273 Effect.raceFirst(
274 Effect.sleep(
275 Duration.toMillis(Duration.fromInputUnsafe(refreshInterval)) <= 0
276 ? "60 seconds"
277 : refreshInterval,
278 ).pipe(Effect.as(true)),
279 Queue.take(refreshIntervalChanges).pipe(Effect.as(false)),
280 ).pipe(
281 Effect.flatMap((intervalElapsed) =>
282 input.refreshOnInterval !== false &&
283 intervalElapsed &&
284 Duration.toMillis(Duration.fromInputUnsafe(refreshInterval)) > 0
285 ? hasProviderStatusDemand.pipe(
286 Effect.flatMap((shouldRefresh) =>
287 shouldRefresh ? refreshSnapshot().pipe(Effect.asVoid) : Effect.void,
288 ),
289 )
290 : Effect.void,
291 ),
292 ),
293 ),
294 Effect.ignoreCause({ log: true }),
295 ),
296 ).pipe(Effect.forkScoped);
297
298 yield* applySnapshot(initialSettings, { forceRefresh: true }).pipe(
299 Effect.ignoreCause({ log: true }),
300 Effect.forkScoped,
301 );
302
303 return {
304 resolveMaintenance: input.resolveMaintenance,
305 getSnapshot: Ref.get(snapshotStateRef).pipe(Effect.map((state) => state.snapshot)),
306 refresh: refreshSnapshot().pipe(Effect.tapError(Effect.logError), Effect.orDie),
307 applyUsageLimits,
308 get streamChanges() {
309 return Stream.fromPubSub(changesPubSub);
310 },
311 } satisfies ManagedServerProvider;
312});