import { DEFAULT_SERVER_SETTINGS, EnvironmentId, ProviderDriverKind, ProviderInstanceId, ServerConfig, type ServerConfig as ServerConfigType, ServerConfigStreamEvent, type ServerConfigStreamEvent as ServerConfigStreamEventType, WS_METHODS, UsageLimitSourceId,} from "@t3tools/contracts";import { describe, expect, it } from "@effect/vitest";import * as Cause from "effect/Cause";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import * as TestClock from "effect/testing/TestClock";import * as Socket from "effect/unstable/socket/Socket"; import { AVAILABLE_CONNECTION_STATE, ConnectionBlockedError, ConnectionTransientError, PrimaryConnectionTarget, RelayConnectionTarget, type PreparedConnection,} from "../connection/model.ts";import * as EnvironmentSupervisor from "../connection/supervisor.ts";import * as Persistence from "../platform/persistence.ts";import * as RpcSession from "./session.ts";import { makeEnvironmentServerConfigState } from "../state/server.ts";import { applyServerConfigProjection } from "../state/serverConfigProjection.ts";import { NETWORK_BLOCKING_HINT } from "../errors/network.ts"; type SocketEventType = "open" | "message" | "close" | "error";type SocketEvent = { readonly code?: number; readonly data?: unknown; readonly reason?: string; readonly type: SocketEventType;};type SocketListener = (event: SocketEvent) => void; class TestWebSocket { static readonly CONNECTING = 0; static readonly OPEN = 1; static readonly CLOSING = 2; static readonly CLOSED = 3; readyState = TestWebSocket.CONNECTING; readonly sent: string[] = []; readonly url: string; private readonly listeners = new Map<SocketEventType, Set<SocketListener>>(); constructor(url: string) { this.url = url; } addEventListener(type: SocketEventType, listener: SocketListener) { const listeners = this.listeners.get(type) ?? new Set<SocketListener>(); listeners.add(listener); this.listeners.set(type, listeners); } removeEventListener(type: SocketEventType, listener: SocketListener) { this.listeners.get(type)?.delete(listener); } send(data: string) { this.sent.push(data); } close(code = 1000, reason = "") { if (this.readyState === TestWebSocket.CLOSED) { return; } this.readyState = TestWebSocket.CLOSED; this.emit("close", { code, reason, type: "close" }); } open() { this.readyState = TestWebSocket.OPEN; this.emit("open", { type: "open" }); } serverMessage(data: string) { this.emit("message", { data, type: "message" }); } private emit(type: SocketEventType, event: SocketEvent) { for (const listener of this.listeners.get(type) ?? []) { listener(event); } }} const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), label: "Test environment", httpBaseUrl: "https://environment.example.test", wsBaseUrl: "wss://environment.example.test",}); const PREPARED: PreparedConnection = { environmentId: TARGET.environmentId, label: TARGET.label, httpBaseUrl: TARGET.httpBaseUrl, socketUrl: "wss://environment.example.test/ws?wsTicket=test", httpAuthorization: null, target: TARGET,}; const SERVER_CONFIG: ServerConfigType = { environment: { environmentId: TARGET.environmentId, label: TARGET.label, platform: { os: "darwin", arch: "arm64", }, serverVersion: "0.0.0-test", capabilities: { repositoryIdentity: true, connectionProbe: true, }, }, auth: { policy: "loopback-browser", bootstrapMethods: ["one-time-token"], sessionMethods: ["browser-session-cookie", "bearer-access-token"], sessionCookieName: "t3_session", }, cwd: "/tmp/workspace", keybindingsConfigPath: "/tmp/workspace/keybindings.json", keybindings: [], issues: [], providers: [], availableEditors: [], observability: { logsDirectoryPath: "/tmp/logs", localTracingEnabled: false, otlpTracesEnabled: false, otlpMetricsEnabled: false, }, settings: DEFAULT_SERVER_SETTINGS,}; const RpcRequest = Schema.TaggedStruct("Request", { id: Schema.Union([Schema.String, Schema.Number]), payload: Schema.Unknown, tag: Schema.String,});const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));const isRpcRequest = Schema.is(RpcRequest);const isPing = Schema.is(Schema.Struct({ _tag: Schema.Literal("Ping") }));const encodeJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));const encodeServerConfig = Schema.encodeSync(ServerConfig);const encodeServerConfigStreamEvent = Schema.encodeSync(ServerConfigStreamEvent);const encodeDefect = Schema.encodeSync(Schema.Defect());const ENCODED_SERVER_CONFIG = encodeServerConfig(SERVER_CONFIG);const THEME_SERVER_CONFIG: ServerConfigType = { ...SERVER_CONFIG, environment: { ...SERVER_CONFIG.environment, capabilities: { ...SERVER_CONFIG.environment.capabilities, environmentThemes: true, }, },};const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG);const SOURCE_SERVER_CONFIG: ServerConfigType = { ...THEME_SERVER_CONFIG, environment: { ...THEME_SERVER_CONFIG.environment, capabilities: { ...THEME_SERVER_CONFIG.environment.capabilities, usageLimitSources: true }, },};const SOURCE_EVENT: ServerConfigStreamEventType = { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [ { id: UsageLimitSourceId.make("proxy"), kind: "cliproxy", label: "Proxy", checkedAt: "2026-09-04T00:00:00Z", accounts: [], }, ], },}; const LEGACY_SERVER_CONFIG = { ...ENCODED_SERVER_CONFIG, environment: { ...ENCODED_SERVER_CONFIG.environment, capabilities: { repositoryIdentity: true, }, },}; const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* ( options: RpcSession.RpcSessionOptions = {},) { const sockets: TestWebSocket[] = []; const constructorLayer = Layer.succeed(Socket.WebSocketConstructor, (url) => { const socket = new TestWebSocket(url); sockets.push(socket); return socket as unknown as globalThis.WebSocket; }); const layer = RpcSession.layerWithOptions(options).pipe(Layer.provide(constructorLayer)); const factory = yield* RpcSession.RpcSessionFactory.pipe(Effect.provide(layer)); return { factory, sockets };}); const awaitSocket = Effect.fn("TestRpcSessionFactory.awaitSocket")(function* ( sockets: ReadonlyArray<TestWebSocket>, index = 0,) { for (let attempt = 0; attempt < 100; attempt += 1) { const socket = sockets[index]; if (socket) { return socket; } yield* Effect.yieldNow; } return yield* Effect.die(new Error("Expected the RPC protocol to create a websocket."));}); const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* ( socket: TestWebSocket, index = 0,) { for (let attempt = 0; attempt < 100; attempt += 1) { const request = socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)[index]; if (request) { return request; } yield* Effect.yieldNow; } return yield* Effect.die(new Error("Expected the RPC protocol to send a request."));}); const completeInitialConfig = Effect.fn("TestRpcSessionFactory.completeInitialConfig")(function* ( socket: TestWebSocket, config: unknown = ENCODED_SERVER_CONFIG, payload: unknown = {},) { const request = yield* awaitRequest(socket); expect(request).toMatchObject({ _tag: "Request", tag: WS_METHODS.subscribeServerConfig, payload, }); socket.serverMessage( encodeJson({ _tag: "Chunk", requestId: request.id, values: [{ version: 1, type: "snapshot", config }], }), );}); const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents")(function* ( socket: TestWebSocket, events: ReadonlyArray<ServerConfigStreamEventType>,) { const request = yield* awaitRequest(socket); socket.serverMessage( encodeJson({ _tag: "Chunk", requestId: request.id, values: events.map((event) => encodeServerConfigStreamEvent(event)), }), );}); describe("RpcSessionFactory", () => { it.effect("owns one scoped websocket attempt and exposes readiness and closure", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); expect(socket.url).toBe(PREPARED.socketUrl); socket.open(); yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); const config = yield* session.initialConfig; expect(config).toEqual(SERVER_CONFIG); expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength( 1, ); const probeFiber = yield* Effect.forkChild(session.probe); const probeRequest = yield* awaitRequest(socket, 1); expect(probeRequest).toMatchObject({ _tag: "Request", tag: WS_METHODS.serverProbe, payload: {}, }); socket.serverMessage( encodeJson({ _tag: "Exit", requestId: probeRequest.id, exit: { _tag: "Success", value: {}, }, }), ); yield* Fiber.join(probeFiber); expect( socket.sent .map((message) => decodeJson(message)) .filter(isRpcRequest) .map((request) => request.tag), ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverProbe]); socket.close(1012, "service restart"); const error = yield* Effect.flip(session.closed); const configStreamError = yield* session .subscribeServerConfig({}) .pipe(Stream.runDrain, Effect.flip); expect(error).toBeInstanceOf(ConnectionTransientError); expect(error).toMatchObject({ reason: "transport", message: "Test environment disconnected.", }); expect(configStreamError).toMatchObject({ _tag: "RpcClientError" }); yield* Effect.yieldNow; expect(sockets).toHaveLength(1); }), ); it.effect("closes the websocket when the session scope is released", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); yield* Effect.scoped( Effect.gen(function* () { const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); }), ); expect(sockets[0]?.readyState).toBe(TestWebSocket.CLOSED); }), ); it.effect("replays current config and broadcasts updates to every subscriber", () => Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); const collectTwo = session .subscribeServerConfig({}) .pipe(Stream.take(2), Stream.runCollect); const firstSubscriber = yield* Effect.forkChild(collectTwo); const secondSubscriber = yield* Effect.forkChild(collectTwo); yield* Effect.yieldNow; const shortcut = { key: "k", metaKey: false, ctrlKey: false, shiftKey: false, altKey: false, modKey: true, }; const request = yield* awaitRequest(socket); socket.serverMessage( encodeJson({ _tag: "Chunk", requestId: request.id, values: [ { version: 1, type: "keybindingsUpdated", payload: { keybindings: [{ command: "terminal.toggle", shortcut }], issues: [], }, }, ], }), ); const firstEvents = Array.from(yield* Fiber.join(firstSubscriber)); const secondEvents = Array.from(yield* Fiber.join(secondSubscriber)); expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]); expect(secondEvents).toEqual(firstEvents); const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead); expect(replay).toMatchObject({ _tag: "Some", value: { type: "snapshot", config: { keybindings: [{ command: "terminal.toggle", shortcut }] }, }, }); }), ), ); for (const options of [ { environmentThemes: true }, { usageLimitSources: true }, { environmentThemes: true, usageLimitSources: true }, ]) { it.effect( `shares only a config subscription with the same opt-ins: ${JSON.stringify(options)}`, () => Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(options); const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, options); yield* Fiber.join(readyFiber); const shared = yield* session.subscribeServerConfig(options).pipe(Stream.runHead); expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } }); expect( socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest), ).toHaveLength(1); const fallbackFiber = yield* session .subscribeServerConfig({}) .pipe(Stream.runHead, Effect.forkChild); const fallbackRequest = yield* awaitRequest(socket, 1); expect(fallbackRequest).toMatchObject({ tag: WS_METHODS.subscribeServerConfig, payload: {}, }); socket.serverMessage( encodeJson({ _tag: "Chunk", requestId: fallbackRequest.id, values: [ { version: 1, type: "snapshot", config: ENCODED_THEME_SERVER_CONFIG, }, ], }), ); expect(yield* Fiber.join(fallbackFiber)).toMatchObject({ _tag: "Some", value: { type: "snapshot" }, }); }), ), ); } it.effect.each([ { usageLimitSources: true }, { environmentThemes: true, usageLimitSources: true }, ])("replays usage sources, removal, and capability downgrade with %j", (options) => Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(options); const session = yield* factory.connect(PREPARED); const ready = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), options); yield* Fiber.join(ready); const observed = yield* Queue.unbounded<ServerConfigStreamEventType>(); yield* session.subscribeServerConfig(options).pipe( Stream.runForEach((event) => Queue.offer(observed, event)),Lines 1–500 of 1195. Load more
import { DEFAULT_SERVER_SETTINGS, EnvironmentId, ProviderDriverKind, ProviderInstanceId, ServerConfig, type ServerConfig as ServerConfigType, ServerConfigStreamEvent, type ServerConfigStreamEvent as ServerConfigStreamEventType, WS_METHODS, UsageLimitSourceId, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; import * as Socket from "effect/unstable/socket/Socket"; import { AVAILABLE_CONNECTION_STATE, ConnectionBlockedError, ConnectionTransientError, PrimaryConnectionTarget, RelayConnectionTarget, type PreparedConnection, } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "./session.ts"; import { makeEnvironmentServerConfigState } from "../state/server.ts"; import { applyServerConfigProjection } from "../state/serverConfigProjection.ts"; import { NETWORK_BLOCKING_HINT } from "../errors/network.ts"; type SocketEventType = "open" | "message" | "close" | "error"; type SocketEvent = { readonly code?: number; readonly data?: unknown; readonly reason?: string; readonly type: SocketEventType; }; type SocketListener = (event: SocketEvent) => void; class TestWebSocket { static readonly CONNECTING = 0; static readonly OPEN = 1; static readonly CLOSING = 2; static readonly CLOSED = 3; readyState = TestWebSocket.CONNECTING; readonly sent: string[] = []; readonly url: string; private readonly listeners = new Map<SocketEventType, Set<SocketListener>>(); constructor(url: string) { this.url = url; } addEventListener(type: SocketEventType, listener: SocketListener) { const listeners = this.listeners.get(type) ?? new Set<SocketListener>(); listeners.add(listener); this.listeners.set(type, listeners); } removeEventListener(type: SocketEventType, listener: SocketListener) { this.listeners.get(type)?.delete(listener); } send(data: string) { this.sent.push(data); } close(code = 1000, reason = "") { if (this.readyState === TestWebSocket.CLOSED) { return; } this.readyState = TestWebSocket.CLOSED; this.emit("close", { code, reason, type: "close" }); } open() { this.readyState = TestWebSocket.OPEN; this.emit("open", { type: "open" }); } serverMessage(data: string) { this.emit("message", { data, type: "message" }); } private emit(type: SocketEventType, event: SocketEvent) { for (const listener of this.listeners.get(type) ?? []) { listener(event); } } } const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), label: "Test environment", httpBaseUrl: "https://environment.example.test", wsBaseUrl: "wss://environment.example.test", }); const PREPARED: PreparedConnection = { environmentId: TARGET.environmentId, label: TARGET.label, httpBaseUrl: TARGET.httpBaseUrl, socketUrl: "wss://environment.example.test/ws?wsTicket=test", httpAuthorization: null, target: TARGET, }; const SERVER_CONFIG: ServerConfigType = { environment: { environmentId: TARGET.environmentId, label: TARGET.label, platform: { os: "darwin", arch: "arm64", }, serverVersion: "0.0.0-test", capabilities: { repositoryIdentity: true, connectionProbe: true, }, }, auth: { policy: "loopback-browser", bootstrapMethods: ["one-time-token"], sessionMethods: ["browser-session-cookie", "bearer-access-token"], sessionCookieName: "t3_session", }, cwd: "/tmp/workspace", keybindingsConfigPath: "/tmp/workspace/keybindings.json", keybindings: [], issues: [], providers: [], availableEditors: [], observability: { logsDirectoryPath: "/tmp/logs", localTracingEnabled: false, otlpTracesEnabled: false, otlpMetricsEnabled: false, }, settings: DEFAULT_SERVER_SETTINGS, }; const RpcRequest = Schema.TaggedStruct("Request", { id: Schema.Union([Schema.String, Schema.Number]), payload: Schema.Unknown, tag: Schema.String, }); const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const isRpcRequest = Schema.is(RpcRequest); const isPing = Schema.is(Schema.Struct({ _tag: Schema.Literal("Ping") })); const encodeJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const encodeServerConfig = Schema.encodeSync(ServerConfig); const encodeServerConfigStreamEvent = Schema.encodeSync(ServerConfigStreamEvent); const encodeDefect = Schema.encodeSync(Schema.Defect()); const ENCODED_SERVER_CONFIG = encodeServerConfig(SERVER_CONFIG); const THEME_SERVER_CONFIG: ServerConfigType = { ...SERVER_CONFIG, environment: { ...SERVER_CONFIG.environment, capabilities: { ...SERVER_CONFIG.environment.capabilities, environmentThemes: true, }, }, }; const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG); const SOURCE_SERVER_CONFIG: ServerConfigType = { ...THEME_SERVER_CONFIG, environment: { ...THEME_SERVER_CONFIG.environment, capabilities: { ...THEME_SERVER_CONFIG.environment.capabilities, usageLimitSources: true }, }, }; const SOURCE_EVENT: ServerConfigStreamEventType = { version: 1, type: "usageLimitSourcesUpdated", payload: { sources: [ { id: UsageLimitSourceId.make("proxy"), kind: "cliproxy", label: "Proxy", checkedAt: "2026-09-04T00:00:00Z", accounts: [], }, ], }, }; const LEGACY_SERVER_CONFIG = { ...ENCODED_SERVER_CONFIG, environment: { ...ENCODED_SERVER_CONFIG.environment, capabilities: { repositoryIdentity: true, }, }, }; const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* ( options: RpcSession.RpcSessionOptions = {}, ) { const sockets: TestWebSocket[] = []; const constructorLayer = Layer.succeed(Socket.WebSocketConstructor, (url) => { const socket = new TestWebSocket(url); sockets.push(socket); return socket as unknown as globalThis.WebSocket; }); const layer = RpcSession.layerWithOptions(options).pipe(Layer.provide(constructorLayer)); const factory = yield* RpcSession.RpcSessionFactory.pipe(Effect.provide(layer)); return { factory, sockets }; }); const awaitSocket = Effect.fn("TestRpcSessionFactory.awaitSocket")(function* ( sockets: ReadonlyArray<TestWebSocket>, index = 0, ) { for (let attempt = 0; attempt < 100; attempt += 1) { const socket = sockets[index]; if (socket) { return socket; } yield* Effect.yieldNow; } return yield* Effect.die(new Error("Expected the RPC protocol to create a websocket.")); }); const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* ( socket: TestWebSocket, index = 0, ) { for (let attempt = 0; attempt < 100; attempt += 1) { const request = socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)[index]; if (request) { return request; } yield* Effect.yieldNow; } return yield* Effect.die(new Error("Expected the RPC protocol to send a request.")); }); const completeInitialConfig = Effect.fn("TestRpcSessionFactory.completeInitialConfig")(function* ( socket: TestWebSocket, config: unknown = ENCODED_SERVER_CONFIG, payload: unknown = {}, ) { const request = yield* awaitRequest(socket); expect(request).toMatchObject({ _tag: "Request", tag: WS_METHODS.subscribeServerConfig, payload, }); socket.serverMessage( encodeJson({ _tag: "Chunk", requestId: request.id, values: [{ version: 1, type: "snapshot", config }], }), ); }); const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents")(function* ( socket: TestWebSocket, events: ReadonlyArray<ServerConfigStreamEventType>, ) { const request = yield* awaitRequest(socket); socket.serverMessage( encodeJson({ _tag: "Chunk", requestId: request.id, values: events.map((event) => encodeServerConfigStreamEvent(event)), }), ); }); describe("RpcSessionFactory", () => { it.effect("owns one scoped websocket attempt and exposes readiness and closure", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); expect(socket.url).toBe(PREPARED.socketUrl); socket.open(); yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); const config = yield* session.initialConfig; expect(config).toEqual(SERVER_CONFIG); expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength( 1, ); const probeFiber = yield* Effect.forkChild(session.probe); const probeRequest = yield* awaitRequest(socket, 1); expect(probeRequest).toMatchObject({ _tag: "Request", tag: WS_METHODS.serverProbe, payload: {}, }); socket.serverMessage( encodeJson({ _tag: "Exit", requestId: probeRequest.id, exit: { _tag: "Success", value: {}, }, }), ); yield* Fiber.join(probeFiber); expect( socket.sent .map((message) => decodeJson(message)) .filter(isRpcRequest) .map((request) => request.tag), ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverProbe]); socket.close(1012, "service restart"); const error = yield* Effect.flip(session.closed); const configStreamError = yield* session .subscribeServerConfig({}) .pipe(Stream.runDrain, Effect.flip); expect(error).toBeInstanceOf(ConnectionTransientError); expect(error).toMatchObject({ reason: "transport", message: "Test environment disconnected.", }); expect(configStreamError).toMatchObject({ _tag: "RpcClientError" }); yield* Effect.yieldNow; expect(sockets).toHaveLength(1); }), ); it.effect("closes the websocket when the session scope is released", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); yield* Effect.scoped( Effect.gen(function* () { const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); }), ); expect(sockets[0]?.readyState).toBe(TestWebSocket.CLOSED); }), ); it.effect("replays current config and broadcasts updates to every subscriber", () => Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); const collectTwo = session .subscribeServerConfig({}) .pipe(Stream.take(2), Stream.runCollect); const firstSubscriber = yield* Effect.forkChild(collectTwo); const secondSubscriber = yield* Effect.forkChild(collectTwo); yield* Effect.yieldNow; const shortcut = { key: "k", metaKey: false, ctrlKey: false, shiftKey: false, altKey: false, modKey: true, }; const request = yield* awaitRequest(socket); socket.serverMessage( encodeJson({ _tag: "Chunk", requestId: request.id, values: [ { version: 1, type: "keybindingsUpdated", payload: { keybindings: [{ command: "terminal.toggle", shortcut }], issues: [], }, }, ], }), ); const firstEvents = Array.from(yield* Fiber.join(firstSubscriber)); const secondEvents = Array.from(yield* Fiber.join(secondSubscriber)); expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]); expect(secondEvents).toEqual(firstEvents); const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead); expect(replay).toMatchObject({ _tag: "Some", value: { type: "snapshot", config: { keybindings: [{ command: "terminal.toggle", shortcut }] }, }, }); }), ), ); for (const options of [ { environmentThemes: true }, { usageLimitSources: true }, { environmentThemes: true, usageLimitSources: true }, ]) { it.effect( `shares only a config subscription with the same opt-ins: ${JSON.stringify(options)}`, () => Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(options); const session = yield* factory.connect(PREPARED); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, options); yield* Fiber.join(readyFiber); const shared = yield* session.subscribeServerConfig(options).pipe(Stream.runHead); expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } }); expect( socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest), ).toHaveLength(1); const fallbackFiber = yield* session .subscribeServerConfig({}) .pipe(Stream.runHead, Effect.forkChild); const fallbackRequest = yield* awaitRequest(socket, 1); expect(fallbackRequest).toMatchObject({ tag: WS_METHODS.subscribeServerConfig, payload: {}, }); socket.serverMessage( encodeJson({ _tag: "Chunk", requestId: fallbackRequest.id, values: [ { version: 1, type: "snapshot", config: ENCODED_THEME_SERVER_CONFIG, }, ], }), ); expect(yield* Fiber.join(fallbackFiber)).toMatchObject({ _tag: "Some", value: { type: "snapshot" }, }); }), ), ); } it.effect.each([ { usageLimitSources: true }, { environmentThemes: true, usageLimitSources: true }, ])("replays usage sources, removal, and capability downgrade with %j", (options) => Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(options); const session = yield* factory.connect(PREPARED); const ready = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); yield* completeInitialConfig(socket, encodeServerConfig(SOURCE_SERVER_CONFIG), options); yield* Fiber.join(ready); const observed = yield* Queue.unbounded<ServerConfigStreamEventType>(); yield* session.subscribeServerConfig(options).pipe( Stream.runForEach((event) => Queue.offer(observed, event)),