1import { describe, expect, it } from "@effect/vitest";
2import { EnvironmentId } from "@t3tools/contracts";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Option from "effect/Option";
6import * as Stream from "effect/Stream";
7import * as SubscriptionRef from "effect/SubscriptionRef";
8import { Atom, AtomRegistry } from "effect/reactivity";
9
10import {
11 AVAILABLE_CONNECTION_STATE,
12 ConnectionTransientError,
13 type SupervisorConnectionState,
14} from "../connection/model.ts";
15import * as EnvironmentRegistry from "../connection/registry.ts";
16import * as EnvironmentSupervisor from "../connection/supervisor.ts";
17import type * as RpcSession from "../rpc/session.ts";
18import { createEnvironmentCatalogAtoms } from "./connections.ts";
19
20const ENVIRONMENT_ID = EnvironmentId.make("catalog-environment");
21
22const connectionState = (
23 overrides: Partial<SupervisorConnectionState>,
24): SupervisorConnectionState => ({
25 ...AVAILABLE_CONNECTION_STATE,
26 desired: true,
27 network: "online",
28 ...overrides,
29});
30
31const makeCatalogHarness = Effect.fn("TestCatalog.makeHarness")(function* () {
32 const state = yield* SubscriptionRef.make(
33 connectionState({ phase: "connected", attempt: 1, generation: 1 }),
34 );
35 const session = yield* SubscriptionRef.make(Option.some({} as RpcSession.RpcSession));
36 const supervisor = {
37 state,
38 session,
39 } as unknown as EnvironmentSupervisor.EnvironmentSupervisor["Service"];
40 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
41 _environmentId,
42 stream,
43 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
44 const registry = EnvironmentRegistry.EnvironmentRegistry.of({
45 followStream,
46 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
47 const catalog = createEnvironmentCatalogAtoms(
48 Atom.runtime(Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, registry)),
49 );
50 const atom = catalog.sessionGenerationAtom(ENVIRONMENT_ID);
51 const atoms = AtomRegistry.make();
52 const unmount = atoms.mount(atom);
53 yield* Effect.addFinalizer(() =>
54 Effect.sync(() => {
55 unmount();
56 atoms.dispose();
57 }),
58 );
59 yield* Effect.yieldNow;
60 /** Moves the supervisor to `next`, with a new session object when one is established. */
61 const generationAfter = (
62 next: SupervisorConnectionState,
63 nextSession: Option.Option<RpcSession.RpcSession> = Option.none(),
64 ) =>
65 SubscriptionRef.set(session, nextSession).pipe(
66 Effect.andThen(SubscriptionRef.set(state, next)),
67 Effect.andThen(Effect.yieldNow),
68 Effect.map(() => atoms.get(atom)),
69 );
70 return { generation: () => atoms.get(atom), generationAfter };
71});
72
73const lastFailure = new ConnectionTransientError({ reason: "transport", detail: "Down." });
74const newSession = () => Option.some({} as RpcSession.RpcSession);
75
76describe("environment catalog atoms", () => {
77 it.effect("changes the session generation only when a new session comes up", () =>
78 Effect.gen(function* () {
79 const { generation, generationAfter } = yield* makeCatalogHarness();
80 const first = generation();
81 // A retry reports the generation it is attempting before it is established.
82 expect(
83 yield* generationAfter(
84 connectionState({ phase: "connecting", stage: "opening", generation: 2, lastFailure }),
85 ),
86 ).toBe(first);
87 expect(
88 yield* generationAfter(
89 connectionState({ phase: "backoff", generation: 1, lastFailure, retryAt: 1 }),
90 ),
91 ).toBe(first);
92 expect(
93 yield* generationAfter(
94 connectionState({ phase: "connected", attempt: 2, generation: 2 }),
95 newSession(),
96 ),
97 ).not.toBe(first);
98 }),
99 );
100
101 it.effect("changes when a replaced supervisor's first session reuses an old generation", () =>
102 Effect.gen(function* () {
103 const { generation, generationAfter } = yield* makeCatalogHarness();
104 const first = generation();
105 yield* generationAfter(
106 connectionState({ phase: "backoff", generation: 1, lastFailure, retryAt: 1 }),
107 );
108 // Re-pairing or editing a route installs a new supervisor, whose counter starts over.
109 yield* generationAfter(connectionState({ phase: "available", generation: 0 }));
110 expect(
111 yield* generationAfter(
112 connectionState({ phase: "connected", attempt: 1, generation: 1 }),
113 newSession(),
114 ),
115 ).not.toBe(first);
116 }),
117 );
118});