1import { describe, expect, it } from "@effect/vitest";
2import { EnvironmentId, type AuthSessionState } from "@t3tools/contracts";
3import * as Clock from "effect/Clock";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as Queue from "effect/Queue";
8import * as Stream from "effect/Stream";
9import * as SubscriptionRef from "effect/SubscriptionRef";
10import * as TestClock from "effect/testing/TestClock";
11import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
12
13import {
14 AVAILABLE_CONNECTION_STATE,
15 PrimaryConnectionTarget,
16 type PreparedConnection,
17} from "../connection/model.ts";
18import * as EnvironmentRegistry from "../connection/registry.ts";
19import * as EnvironmentSupervisor from "../connection/supervisor.ts";
20import * as RpcHttp from "../rpc/http.ts";
21import type { RpcSession } from "../rpc/session.ts";
22import { createEnvironmentSessionAtoms } from "./session.ts";
23
24const TARGET = new PrimaryConnectionTarget({
25 environmentId: EnvironmentId.make("environment-1"),
26 label: "Test environment",
27 httpBaseUrl: "https://environment.example.test",
28 wsBaseUrl: "wss://environment.example.test",
29});
30const PREPARED: PreparedConnection = {
31 environmentId: TARGET.environmentId,
32 label: TARGET.label,
33 httpBaseUrl: "https://environment.example.test",
34 socketUrl: "wss://environment.example.test/ws",
35 httpAuthorization: { _tag: "Bearer", token: "bearer-token" },
36 target: TARGET,
37};
38const SESSION = {
39 authenticated: true,
40 auth: {
41 policy: "remote-reachable",
42 bootstrapMethods: [],
43 sessionMethods: [],
44 sessionCookieName: "t3_session",
45 },
46 scopes: ["orchestration:read"],
47} satisfies AuthSessionState;
48
49const rpcSession = (): RpcSession => ({ closed: Effect.never }) as unknown as RpcSession;
50
51const makeHarness = Effect.fn("TestSessionState.makeHarness")(function* (
52 reply: (requestNumber: number) => Promise<Response>,
53) {
54 let requestCount = 0;
55 const transport = yield* SubscriptionRef.make(Option.some(rpcSession()));
56 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
57 target: TARGET,
58 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
59 session: transport,
60 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
61 connect: Effect.void,
62 disconnect: Effect.void,
63 retryNow: Effect.void,
64 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
65 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
66 _environmentId,
67 stream,
68 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
69 const runtime = Atom.runtime(
70 Layer.mergeAll(
71 Layer.succeed(
72 EnvironmentRegistry.EnvironmentRegistry,
73 EnvironmentRegistry.EnvironmentRegistry.of({
74 followStream,
75 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]),
76 ),
77 RpcHttp.layerRemoteHttpClient(async () => reply(++requestCount)),
78 // The test's clock, so the retry wait moves with `TestClock.adjust`.
79 Layer.succeed(Clock.Clock, yield* Clock.Clock),
80 ),
81 );
82 const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
83 Effect.sync(() => registry.dispose()),
84 );
85 const results = yield* Queue.unbounded<AsyncResult.AsyncResult<AuthSessionState, unknown>>();
86 const atom = createEnvironmentSessionAtoms(runtime).sessionStateAtom(TARGET.environmentId);
87 const unsubscribe = registry.subscribe(atom, (result) => Queue.offerUnsafe(results, result), {
88 immediate: true,
89 });
90 yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));
91 const nextSettled = Effect.fn("TestSessionState.nextSettled")(function* () {
92 for (;;) {
93 const result = yield* Queue.take(results);
94 if (!result.waiting && result._tag !== "Initial") return result;
95 }
96 });
97 return { transport, nextSettled };
98});
99
100const unreachable = () => Promise.reject(new TypeError("Network request failed"));
101
102describe("environment session state", () => {
103 it.effect("fetches permissions again when the connection is re-established", () =>
104 Effect.gen(function* () {
105 const harness = yield* makeHarness((requestNumber) =>
106 requestNumber === 1 ? unreachable() : Promise.resolve(Response.json(SESSION)),
107 );
108 expect((yield* harness.nextSettled())._tag).toBe("Failure");
109
110 yield* SubscriptionRef.set(harness.transport, Option.some(rpcSession()));
111
112 expect(yield* harness.nextSettled()).toMatchObject({ _tag: "Success", value: SESSION });
113 }),
114 );
115
116 it.effect("retries a failed permission check with backoff that restarts per connection", () =>
117 Effect.gen(function* () {
118 const harness = yield* makeHarness((requestNumber) =>
119 requestNumber <= 3 ? unreachable() : Promise.resolve(Response.json(SESSION)),
120 );
121 expect((yield* harness.nextSettled())._tag).toBe("Failure");
122
123 yield* TestClock.adjust("2 seconds");
124 expect((yield* harness.nextSettled())._tag).toBe("Failure");
125
126 yield* SubscriptionRef.set(harness.transport, Option.some(rpcSession()));
127 expect((yield* harness.nextSettled())._tag).toBe("Failure");
128
129 yield* TestClock.adjust("2 seconds");
130 expect(yield* harness.nextSettled()).toMatchObject({ _tag: "Success", value: SESSION });
131 }),
132 );
133});