1/**
2 * ProviderInstanceRegistry — the single Effect service in the new model.
3 *
4 * Owns a `Map<ProviderInstanceId, ProviderInstance>` produced by running
5 * registered driver factories against `ServerSettings.providerInstances`.
6 * The registry watches settings; when an instance's config changes (or
7 * the entry disappears), the registry tears down the affected instance's
8 * scope and rebuilds — that's the entire hot-reload story.
9 *
10 * What rest-of-server reads from here:
11 * - `getInstance(instanceId)` — for routing turn/session calls.
12 * - `listInstances` — for snapshot aggregation in `ProviderRegistry`.
13 * - `listUnavailable` — `ServerProvider` shadows for instances whose
14 * driver is not registered in this build (rollback / fork tolerance).
15 * - `streamChanges` — coalesced "registry mutated" pings so consumers
16 * can re-pull lists or re-broadcast.
17 *
18 * The live implementation (`makeProviderInstanceRegistry`, `layer`) also
19 * provides its sibling mutator.
20 *
21 * Materializes every entry in a `ProviderInstanceConfigMap`:
22 *
23 * - When the entry's `driver` matches a registered driver, the registry
24 * decodes the opaque `config` envelope through `driver.configSchema`
25 * and calls `driver.create()` inside a fresh child scope. The
26 * resulting `ProviderInstance` is stored keyed by instance id,
27 * alongside its scope so the entry can be torn down independently.
28 * - When the entry's `driver` is unknown to this build (fork, rollback,
29 * in-flight PR branch), the registry emits an `"unavailable"` shadow
30 * `ServerProvider` snapshot instead of failing. This is what makes
31 * downgrades and fork-hopping safe per the
32 * `forward/backward compatibility invariant` in
33 * `packages/contracts/src/providerInstance.ts`.
34 * - When the entry's config fails schema decode, the registry logs and
35 * emits a shadow snapshot with the schema detail — same bucket as an
36 * unknown driver.
37 *
38 * Unlike the pre-Slice-D layer, the registry now holds mutable state
39 * (`Ref`s + `PubSub`) and exposes an internal mutator
40 * (`ProviderInstanceRegistryMutator`) whose `reconcile` method diffs a
41 * fresh config map against the live state, tearing down removed instances
42 * and building new ones without disturbing unaffected instances.
43 *
44 * Every live instance runs inside its own child `Scope`. The registry's
45 * own scope owns all child scopes via finalizers, so closing the registry
46 * tears every instance down in reverse order; closing a single instance
47 * (via `reconcile` removing it) leaves the rest untouched.
48 *
49 * @module provider/ProviderInstanceRegistry
50 */
51import {
52 providerInstanceConfigEnabledFlag,
53 ProviderInstanceId,
54 type ProviderInstanceConfig,
55 type ProviderInstanceConfigMap,
56 type ProviderDriverKind,
57 type ServerProvider,
58} from "@t3tools/contracts";
59import * as Context from "effect/Context";
60import * as Effect from "effect/Effect";
61import * as Equal from "effect/Equal";
62import * as Exit from "effect/Exit";
63import * as Layer from "effect/Layer";
64import * as PubSub from "effect/PubSub";
65import * as Ref from "effect/Ref";
66import * as Schema from "effect/Schema";
67import * as Scope from "effect/Scope";
68import * as Stream from "effect/Stream";
69
70import { buildUnavailableProviderSnapshot } from "./unavailableProviderSnapshot.ts";
71import * as ProviderInstanceRegistryMutator from "./ProviderInstanceRegistryMutator.ts";
72import type { AnyProviderDriver, ProviderInstance } from "@t3tools/provider-core/server/driver";
73
74export class ProviderInstanceRegistry extends Context.Service<
75 ProviderInstanceRegistry,
76 {
77 /**
78 * Look up one instance by id. Returns `undefined` (not Option) when the
79 * id is unknown — callers branch on falsy and emit
80 * `ProviderInstanceNotFoundError`.
81 */
82 readonly getInstance: (
83 instanceId: ProviderInstanceId,
84 ) => Effect.Effect<ProviderInstance | undefined>;
85 /**
86 * Every available (driver-registered, successfully created) instance,
87 * in stable settings-author order.
88 */
89 readonly listInstances: Effect.Effect<ReadonlyArray<ProviderInstance>>;
90 /**
91 * Wire-shape shadow snapshots for instances whose driver is unknown to
92 * this build (or whose config failed to decode). Suitable for merging
93 * directly into `ProviderRegistry` output.
94 */
95 readonly listUnavailable: Effect.Effect<ReadonlyArray<ServerProvider>>;
96 /**
97 * Push notification stream emitted whenever the registry's contents
98 * change — instance added, removed, or rebuilt. The payload is `void`
99 * because consumers always want to re-pull `listInstances` /
100 * `listUnavailable` together.
101 *
102 * NOTE: because `Stream.fromPubSub` defers `PubSub.subscribe` until the
103 * stream starts running, forking a consumer via
104 * `Stream.runForEach(...).pipe(Effect.forkScoped)` races the next
105 * publish — the forked fiber may not have subscribed yet when the
106 * publish lands. Hot-reload consumers that must not miss a publish
107 * should use `subscribeChanges` below instead, which acquires the
108 * subscription synchronously in the caller's fiber before the consumer
109 * loop is forked.
110 */
111 readonly streamChanges: Stream.Stream<void>;
112 /**
113 * Acquire a subscription to the registry's change channel synchronously
114 * in the caller's fiber. Returns a `PubSub.Subscription<void>` whose
115 * lifetime is scoped to the provided `Scope` (the subscription is
116 * released when the scope closes). Consumers typically `yield*` this
117 * in the same fiber that forks their consumer loop, then drain with
118 * `PubSub.take(subscription)` inside `Effect.forever`. Because the
119 * subscription is registered with the PubSub before this `yield*`
120 * returns, no subsequent publish can land in a gap.
121 *
122 * This exists because the `ProviderInstanceRegistry` publishes on a
123 * PubSub and `Stream.fromPubSub` defers subscription until the stream
124 * starts executing — a consumer that `forkScoped`s the stream
125 * consumption can miss a publish that lands in the narrow window
126 * between "fiber scheduled" and "fiber starts running".
127 */
128 readonly subscribeChanges: Effect.Effect<PubSub.Subscription<void>, never, Scope.Scope>;
129 }
130>()("t3/provider/ProviderInstanceRegistry") {}
131
132/**
133 * Live registry entry: the materialized `ProviderInstance` + the fresh
134 * child scope its `create` effect ran in + the original `entry` envelope
135 * so `reconcile` can cheaply detect "no-op" updates.
136 */
137interface LiveEntry {
138 readonly instance: ProviderInstance;
139 readonly scope: Scope.Closeable;
140 readonly entry: ProviderInstanceConfig;
141}
142
143/**
144 * Internal state shared between the public registry service and the
145 * mutator service. Both services are thin shells around these refs.
146 */
147interface RegistryState {
148 readonly entries: Ref.Ref<ReadonlyMap<ProviderInstanceId, LiveEntry>>;
149 readonly unavailable: Ref.Ref<ReadonlyMap<ProviderInstanceId, ServerProvider>>;
150 readonly changes: PubSub.PubSub<void>;
151}
152
153/**
154 * Structural equality on `ProviderInstanceConfig` envelopes. Used by
155 * `reconcile` to skip rebuilds when settings arrive unchanged. Config
156 * payloads are opaque `unknown` at the envelope layer; `Equal.equals`
157 * falls back to structural equality for plain records, which matches how
158 * the schema decode output is constructed.
159 */
160const entryEqual = (a: ProviderInstanceConfig, b: ProviderInstanceConfig): boolean =>
161 Equal.equals(a, b);
162
163/**
164 * Resolve an entry's enabled state. An explicit false on either the
165 * envelope or the raw config blob wins (most restrictive) — old settings
166 * files can carry both flags with conflicting values, and a user's disable
167 * must never be silently undone. Otherwise the envelope flag wins, then the
168 * decoded config's flag (which carries the driver schema's default for
169 * built-ins and forks alike), then enabled by default.
170 */
171const resolveEntryEnabled = (entry: ProviderInstanceConfig, typedConfig: unknown): boolean => {
172 const rawConfigEnabled = providerInstanceConfigEnabledFlag(entry.config);
173 if (entry.enabled === false || rawConfigEnabled === false) {
174 return false;
175 }
176 return entry.enabled ?? providerInstanceConfigEnabledFlag(typedConfig) ?? true;
177};
178
179/**
180 * Build one live entry from a raw config envelope. Returns either a
181 * `LiveEntry` plus undefined unavailable shadow, or a shadow snapshot and
182 * undefined entry — callers dispatch to the appropriate Ref bucket.
183 */
184const buildEntry = <R>(input: {
185 readonly driversById: ReadonlyMap<ProviderDriverKind, AnyProviderDriver<R>>;
186 readonly parentScope: Scope.Scope;
187 readonly instanceId: ProviderInstanceId;
188 readonly rawInstanceId: string;
189 readonly entry: ProviderInstanceConfig;
190}): Effect.Effect<
191 | { readonly kind: "live"; readonly live: LiveEntry }
192 | { readonly kind: "unavailable"; readonly snapshot: ServerProvider },
193 never,
194 R
195> =>
196 Effect.gen(function* () {
197 const { driversById, parentScope, instanceId, rawInstanceId, entry } = input;
198 const driver = driversById.get(entry.driver);
199 if (!driver) {
200 return {
201 kind: "unavailable" as const,
202 snapshot: yield* buildUnavailableProviderSnapshot({
203 driverKind: entry.driver,
204 instanceId,
205 displayName: entry.displayName,
206 accentColor: entry.accentColor,
207 reason: `Driver '${entry.driver}' is not registered in this build.`,
208 }),
209 };
210 }
211
212 const decoder = Schema.decodeUnknownEffect(driver.configSchema);
213 const decodeResult = yield* decoder(entry.config ?? driver.defaultConfig()).pipe(Effect.result);
214 if (decodeResult._tag === "Failure") {
215 const issue = decodeResult.failure;
216 const detail = issue.message ?? String(issue);
217 yield* Effect.logError("Failed to decode provider instance config", {
218 instanceId: rawInstanceId,
219 driver: entry.driver,
220 detail,
221 });
222 return {
223 kind: "unavailable" as const,
224 snapshot: yield* buildUnavailableProviderSnapshot({
225 driverKind: entry.driver,
226 instanceId,
227 displayName: entry.displayName,
228 accentColor: entry.accentColor,
229 reason: `Invalid config for instance '${rawInstanceId}': ${detail}`,
230 }),
231 };
232 }
233
234 const typedConfig = decodeResult.success;
235 const childScope = yield* Scope.make();
236 // Attach the child scope to the registry's parent scope: if the
237 // registry scope closes, each surviving instance's child scope is
238 // closed through this finalizer. `reconcile` manually closes the
239 // child scope on remove/replace; subsequent close via the parent's
240 // finalizer is a no-op because `Scope.close` is idempotent.
241 yield* Scope.addFinalizer(parentScope, Scope.close(childScope, Exit.void).pipe(Effect.ignore));
242
243 const createResult = yield* driver
244 .create({
245 instanceId,
246 displayName: entry.displayName,
247 accentColor: entry.accentColor,
248 environment: entry.environment ?? [],
249 enabled: resolveEntryEnabled(entry, typedConfig),
250 config: typedConfig,
251 })
252 .pipe(Effect.provideService(Scope.Scope, childScope), Effect.result);
253 if (createResult._tag === "Failure") {
254 yield* Effect.logError("Failed to create provider instance", {
255 instanceId: rawInstanceId,
256 driver: entry.driver,
257 detail: createResult.failure.detail,
258 });
259 yield* Scope.close(childScope, Exit.void).pipe(Effect.ignore);
260 return {
261 kind: "unavailable" as const,
262 snapshot: yield* buildUnavailableProviderSnapshot({
263 driverKind: entry.driver,
264 instanceId,
265 displayName: entry.displayName,
266 accentColor: entry.accentColor,
267 reason: `Driver '${entry.driver}' failed to create instance: ${createResult.failure.detail}`,
268 }),
269 };
270 }
271
272 return {
273 kind: "live" as const,
274 live: {
275 instance: createResult.success,
276 scope: childScope,
277 entry,
278 },
279 };
280 });
281
282/**
283 * Reconcile-only implementation of the mutator. Exposed to the hydration
284 * layer; never called directly by the rest of the server.
285 */
286const makeReconcile = <R>(input: {
287 readonly state: RegistryState;
288 readonly driversById: ReadonlyMap<ProviderDriverKind, AnyProviderDriver<R>>;
289 readonly parentScope: Scope.Scope;
290}): ((configMap: ProviderInstanceConfigMap) => Effect.Effect<void, never, R>) => {
291 const { state, driversById, parentScope } = input;
292 return (configMap: ProviderInstanceConfigMap) =>
293 Effect.gen(function* () {
294 const previousEntries = yield* Ref.get(state.entries);
295 const previousUnavailable = yield* Ref.get(state.unavailable);
296 const nextRaw = Object.entries(configMap);
297 const nextKeys = new Set<ProviderInstanceId>(
298 nextRaw.map(([raw]) => ProviderInstanceId.make(raw)),
299 );
300
301 // 1. Close scopes for instances that disappeared or whose config
302 // changed. Do this BEFORE creating replacements so ids map 1-to-1
303 // to live scopes at all times.
304 const removedIds: Array<ProviderInstanceId> = [];
305 const replacedIds = new Set<ProviderInstanceId>();
306 for (const [instanceId, live] of previousEntries) {
307 if (!nextKeys.has(instanceId)) {
308 removedIds.push(instanceId);
309 continue;
310 }
311 const nextEntry = configMap[instanceId];
312 if (nextEntry !== undefined && !entryEqual(live.entry, nextEntry)) {
313 replacedIds.add(instanceId);
314 }
315 }
316 for (const id of [...removedIds, ...replacedIds]) {
317 const live = previousEntries.get(id);
318 if (live) {
319 yield* Scope.close(live.scope, Exit.void).pipe(Effect.ignore);
320 }
321 }
322
323 // 2. Build additions and replacements. Walk `nextRaw` so the final
324 // entry order follows settings-author order.
325 const builtEntries = new Map<ProviderInstanceId, LiveEntry>();
326 const builtUnavailable = new Map<ProviderInstanceId, ServerProvider>();
327 let orderChanged = false;
328 const previousOrder = [...previousEntries.keys()];
329 const nextOrder: Array<ProviderInstanceId> = [];
330
331 for (const [rawInstanceId, entry] of nextRaw) {
332 const instanceId = ProviderInstanceId.make(rawInstanceId);
333 nextOrder.push(instanceId);
334
335 const existing = previousEntries.get(instanceId);
336 if (existing !== undefined && !replacedIds.has(instanceId)) {
337 // No-op update: keep the existing live entry and scope.
338 builtEntries.set(instanceId, existing);
339 continue;
340 }
341
342 const result = yield* buildEntry({
343 driversById,
344 parentScope,
345 instanceId,
346 rawInstanceId,
347 entry,
348 });
349 if (result.kind === "live") {
350 builtEntries.set(instanceId, result.live);
351 } else {
352 builtUnavailable.set(instanceId, result.snapshot);
353 }
354 }
355
356 if (previousOrder.length === nextOrder.length) {
357 for (let i = 0; i < previousOrder.length; i++) {
358 if (previousOrder[i] !== nextOrder[i]) {
359 orderChanged = true;
360 break;
361 }
362 }
363 } else {
364 orderChanged = true;
365 }
366
367 const entriesChanged =
368 orderChanged ||
369 removedIds.length > 0 ||
370 replacedIds.size > 0 ||
371 builtEntries.size !== previousEntries.size;
372 const unavailableChanged =
373 builtUnavailable.size !== previousUnavailable.size ||
374 [...builtUnavailable].some(([id, snapshot]) => {
375 const prev = previousUnavailable.get(id);
376 return prev === undefined || !Equal.equals(prev, snapshot);
377 }) ||
378 [...previousUnavailable].some(([id]) => !builtUnavailable.has(id));
379
380 yield* Ref.set(state.entries, builtEntries);
381 yield* Ref.set(state.unavailable, builtUnavailable);
382
383 if (entriesChanged || unavailableChanged) {
384 yield* PubSub.publish(state.changes, undefined);
385 }
386 });
387};
388
389/**
390 * Build the registry's runtime state from a concrete configMap. Returns a
391 * record containing:
392 *
393 * - `registry`: the read-only `ProviderInstanceRegistry["Service"]` to expose
394 * under `ProviderInstanceRegistry`.
395 * - `mutator`: the `ProviderInstanceRegistryMutatorShape` to expose
396 * under `ProviderInstanceRegistryMutator`.
397 * - `reconcile`: the raw reconcile function, provided for convenience so
398 * boot-time layers can hydrate an initial map before publishing the
399 * services.
400 *
401 * The scope that this effect runs in owns every per-instance child scope
402 * created during `reconcile`. Closing that scope closes every live
403 * instance.
404 */
405export const makeProviderInstanceRegistry = <R>(input: {
406 readonly drivers: ReadonlyArray<AnyProviderDriver<R>>;
407 readonly configMap: ProviderInstanceConfigMap;
408}): Effect.Effect<
409 {
410 readonly registry: ProviderInstanceRegistry["Service"];
411 readonly mutator: ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutatorShape;
412 },
413 never,
414 R | Scope.Scope
415> =>
416 Effect.gen(function* () {
417 const driversById = new Map<ProviderDriverKind, AnyProviderDriver<R>>(
418 input.drivers.map((driver) => [driver.driverKind, driver]),
419 );
420
421 // Capture the enclosing scope so per-instance child scopes can be
422 // attached to it at `reconcile` time. Without this, `reconcile`
423 // called later (e.g. from the hydration layer) would attach child
424 // scopes to the *caller's* scope instead of the registry's.
425 const parentScope = yield* Scope.Scope;
426
427 // Capture the driver R context at construction time so `reconcile`
428 // can be invoked later without re-providing driver dependencies.
429 // The service tag's declared `reconcile: Effect<void>` hides R from
430 // consumers — we materialize that here.
431 const driverContext = yield* Effect.context<R>();
432
433 const entries = yield* Ref.make<ReadonlyMap<ProviderInstanceId, LiveEntry>>(new Map());
434 const unavailable = yield* Ref.make<ReadonlyMap<ProviderInstanceId, ServerProvider>>(new Map());
435 const changes = yield* PubSub.unbounded<void>();
436 yield* Effect.addFinalizer(() => PubSub.shutdown(changes));
437
438 const state: RegistryState = { entries, unavailable, changes };
439 const reconcileWithR = makeReconcile({ state, driversById, parentScope });
440 const reconcile: ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutatorShape["reconcile"] =
441 (configMap) => reconcileWithR(configMap).pipe(Effect.provideContext(driverContext));
442
443 // Hydrate the initial configMap synchronously so callers can read
444 // `listInstances` immediately after this effect completes.
445 yield* reconcile(input.configMap);
446
447 const registry: ProviderInstanceRegistry["Service"] = {
448 getInstance: (id) => Ref.get(entries).pipe(Effect.map((map) => map.get(id)?.instance)),
449 listInstances: Ref.get(entries).pipe(
450 Effect.map(
451 (map) =>
452 Array.from(map.values(), (live) => live.instance) as ReadonlyArray<ProviderInstance>,
453 ),
454 ),
455 listUnavailable: Ref.get(unavailable).pipe(
456 Effect.map((map) => Array.from(map.values()) as ReadonlyArray<ServerProvider>),
457 ),
458 // Getters: each read constructs a fresh Stream / Effect descriptor
459 // so multiple consumers don't share a single already-started
460 // Channel or subscription. Matches the pattern `ProviderRegistry`
461 // uses for its own `streamChanges`.
462 get streamChanges() {
463 return Stream.fromPubSub(changes);
464 },
465 // Synchronous subscribe — callers that need to consume changes
466 // from a forked fibre must acquire the subscription in their own
467 // fibre first (via `yield* registry.subscribeChanges`) and only
468 // then fork a consumer loop on `Stream.fromSubscription(...)` /
469 // `PubSub.take(...)`. See the shape docs for the race this avoids.
470 get subscribeChanges() {
471 return PubSub.subscribe(changes);
472 },
473 };
474
475 const mutator: ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutatorShape = {
476 reconcile,
477 };
478
479 return { registry, mutator };
480 });
481
482/**
483 * Layer variant that also exposes the mutator tag. Consumed by
484 * `ProviderInstanceRegistryHydrationLive` to reconcile on settings
485 * changes. Tests that exercise the mutator directly can pair this Layer
486 * with a test-local `ServerSettingsService`.
487 */
488export const layer = <R>(input: {
489 readonly drivers: ReadonlyArray<AnyProviderDriver<R>>;
490 readonly configMap: ProviderInstanceConfigMap;
491}): Layer.Layer<
492 ProviderInstanceRegistry | ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator,
493 never,
494 R
495> =>
496 Layer.effectContext(
497 makeProviderInstanceRegistry(input).pipe(
498 Effect.map(({ registry, mutator }) =>
499 Context.make(ProviderInstanceRegistry, registry).pipe(
500 Context.add(ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator, mutator),
501 ),
502 ),
503 ),
504 ) as Layer.Layer<
505 ProviderInstanceRegistry | ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator,
506 never,
507 R
508 >;