import * as NodeNet from "node:net"; import { it as effectIt } from "@effect/vitest";import { CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS, PREVIEW_URL_MAX_LENGTH, type DiscoveredLocalServer,} from "@t3tools/contracts";import { HostProcessPlatform } from "@t3tools/shared/hostProcess";import * as Net from "@t3tools/shared/Net";import * as Cause from "effect/Cause";import * as Clock from "effect/Clock";import * as Duration from "effect/Duration";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 PlatformError from "effect/PlatformError";import * as Scope from "effect/Scope";import * as TestClock from "effect/testing/TestClock";import { expect } from "vite-plus/test";import { FetchHttpClient } from "effect/unstable/http"; import * as ProcessRunner from "../processRunner.ts";import * as PortScanner from "./PortScanner.ts";const processProbeFailure: ProcessRunner.ProcessRunner["Service"]["run"] = (input) => Effect.fail( new ProcessRunner.ProcessSpawnError({ command: input.command, argumentCount: input.args.length, cwd: input.cwd, cause: PlatformError.systemError({ _tag: "NotFound", module: "ChildProcess", method: "spawn", description: "PowerShell is not installed in the test environment", }), }), ); const TestProcessRunner = Layer.succeed(ProcessRunner.ProcessRunner, { run: processProbeFailure,}); let integrationListeningPort: number | null = null; const TestIntegrationNet = Layer.succeed(Net.NetService, { canListenOnHost: () => Effect.succeed(true), isPortAvailableOnLoopback: (port) => Effect.sync(() => port !== integrationListeningPort), hasListenerOnHost: (port) => Effect.sync(() => port === integrationListeningPort), reserveLoopbackPort: () => Effect.succeed(40_000), findAvailablePort: (preferred) => Effect.succeed(preferred),}); const makeProbeFailureLayer = ( run: ProcessRunner.ProcessRunner["Service"]["run"], fetch: typeof globalThis.fetch = globalThis.fetch,) => PortScanner.layer.pipe( Layer.provide( Layer.mergeAll( Layer.succeed(ProcessRunner.ProcessRunner, { run }), Layer.succeed(Net.NetService, { canListenOnHost: () => Effect.succeed(true), isPortAvailableOnLoopback: () => Effect.succeed(true), hasListenerOnHost: () => Effect.succeed(false), reserveLoopbackPort: () => Effect.succeed(40_000), findAvailablePort: (preferred) => Effect.succeed(preferred), }), Layer.succeed(HostProcessPlatform, "linux"), FetchHttpClient.layer.pipe(Layer.provide(Layer.succeed(FetchHttpClient.Fetch, fetch))), ), ), ); const TestPortDiscoveryLive = PortScanner.layer.pipe( Layer.provide( Layer.mergeAll( TestProcessRunner, TestIntegrationNet, Layer.succeed(HostProcessPlatform, "win32"), FetchHttpClient.layer, ), ),); const LSOF_TEST_PORT = 43_123; const makeLsofScannerLayer = (input: { readonly pid: () => number; readonly fetch: typeof globalThis.fetch;}) => PortScanner.layer.pipe( Layer.provide( Layer.mergeAll( Layer.succeed(ProcessRunner.ProcessRunner, { run: () => Effect.succeed({ stdout: `p${input.pid()}\ncnode\nn*:${LSOF_TEST_PORT}\n`, stderr: "", code: null, timedOut: false, stdoutTruncated: false, stderrTruncated: false, stdoutInvalidUtf8: false, stderrInvalidUtf8: false, }), }), Layer.succeed(Net.NetService, { canListenOnHost: () => Effect.succeed(true), isPortAvailableOnLoopback: () => Effect.succeed(true), hasListenerOnHost: () => Effect.succeed(false), reserveLoopbackPort: () => Effect.succeed(40_000), findAvailablePort: (preferred) => Effect.succeed(preferred), }), Layer.succeed(HostProcessPlatform, "linux"), FetchHttpClient.layer.pipe( Layer.provide(Layer.succeed(FetchHttpClient.Fetch, input.fetch)), ), ), ), ); const openServer = ( port: number, onConnection: (socket: NodeNet.Socket) => void,): Effect.Effect<NodeNet.Server | null> => Effect.callback((resume) => { const server = NodeNet.createServer(onConnection); server.once("error", () => { resume(Effect.succeed(null)); }); server.listen(port, "127.0.0.1", () => { resume(Effect.succeed(server)); }); return Effect.sync(() => { server.close(); }); }); const closeServer = (server: NodeNet.Server): Effect.Effect<void> => Effect.callback((resume) => { server.close(() => resume(Effect.void)); }); const openCommonDevServer = Effect.fn("PortScannerTest.openCommonDevServer")(function* ( ports: ReadonlyArray<number>, onConnection: (socket: NodeNet.Socket) => void,) { for (const port of ports) { const server = yield* openServer(port, onConnection); if (server !== null) return { port, server }; } return yield* Effect.die( new Error("No common development port was available for the preview scanner test"), );}); const commonDevServer = Effect.acquireRelease( openCommonDevServer(PortScanner.COMMON_DEV_PORTS, (socket) => { socket.once("data", () => { socket.end("HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: 5\r\n\r\nhello"); }); }).pipe( Effect.tap(({ port }) => Effect.sync(() => { integrationListeningPort = port; }), ), ), ({ server }) => closeServer(server).pipe( Effect.ensuring( Effect.sync(() => { integrationListeningPort = null; }), ), ),); const commonNonHttpServer = Effect.acquireRelease( openCommonDevServer(PortScanner.COMMON_DEV_PORTS.toReversed(), (socket) => { socket.on("error", () => undefined); socket.once("data", () => socket.end("MYSQL\r\n\r\n")); }).pipe( Effect.tap(({ port }) => Effect.sync(() => { integrationListeningPort = port; }), ), ), ({ server }) => closeServer(server).pipe( Effect.ensuring( Effect.sync(() => { integrationListeningPort = null; }), ), ),); /** * Integration tests against a real TCP listener. We provide the Windows host * platform so the tests exercise the TCP-probe fallback without depending on * `lsof` being installed. */effectIt.layer(TestPortDiscoveryLive)("PortDiscovery integration (TCP probe fallback)", (it) => { it.effect( "scan() returns an HTTP server we just opened on a curated dev port", Effect.fn("PortScannerTest.scanFindsCommonDevServer")(function* () { const { port } = yield* commonDevServer; const scanner = yield* PortScanner.PortDiscovery; const result = yield* scanner.scan(); const found = result.find((server) => server.port === port); expect(found).toBeDefined(); expect(found?.host).toBe("localhost"); }), ); it.effect( "scan() excludes a listening port that does not speak HTTP", Effect.fn("PortScannerTest.scanExcludesNonHttpServer")(function* () { const { port } = yield* commonNonHttpServer; const scanner = yield* PortScanner.PortDiscovery; const result = yield* scanner.scan(); expect(result.some((server) => server.port === port)).toBe(false); }), ); it.effect( "retain drives an immediate broadcast to subscribers", Effect.fn("PortScannerTest.retainBroadcastsImmediately")(function* () { const { port } = yield* commonDevServer; const received: number[] = []; const scanner = yield* PortScanner.PortDiscovery; yield* scanner.subscribe({ configuredUrls: [], initialSnapshot: [] }, (servers) => Effect.sync(() => { for (const server of servers) received.push(server.port); }), ); yield* scanner.retain; expect(received).toContain(port); }), );}); effectIt.effect("revalidates a successful HTML probe after its cache entry expires", () => { let responds = true; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return responds ? Promise.resolve(new Response("hello", { headers: { "content-type": "text/html" } })) : Promise.reject(new TypeError("not HTTP")); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; expect(yield* scanner.scan()).toHaveLength(1); expect(yield* scanner.scan()).toHaveLength(1); expect(requests).toEqual([`http://localhost:${LSOF_TEST_PORT}/`]); responds = false; yield* TestClock.adjust(Duration.seconds(15)); expect(yield* scanner.scan()).toHaveLength(0); expect(requests).toEqual([ `http://localhost:${LSOF_TEST_PORT}/`, `http://localhost:${LSOF_TEST_PORT}/`, `https://localhost:${LSOF_TEST_PORT}/`, ]); }).pipe(Effect.provide(layer));}); effectIt.effect("keeps a full configured URL when the discovered server root fails", () => { const requests: string[] = []; const configuredUrl = `http://localhost:${LSOF_TEST_PORT}/docs`; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { const url = String(input); requests.push(url); return Promise.resolve( url === configuredUrl ? new Response("docs", { headers: { "content-type": "text/html" } }) : new Response("not found", { status: 404, headers: { "content-type": "text/html" }, }), ); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const servers = yield* scanner.scan([configuredUrl]); expect(servers).toHaveLength(1); expect(servers[0]?.url).toBe(configuredUrl); expect(requests).toContain(configuredUrl); }).pipe(Effect.provide(layer));}); effectIt.effect("probes configured custom ports through a canonical loopback host", () => { const customPort = 43_124; const configuredUrl = `http://0.0.0.0:${customPort}/docs`; const expectedUrl = `http://localhost:${customPort}/docs`; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeProbeFailureLayer(processProbeFailure, fetchFn); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const servers = yield* scanner.scan([configuredUrl]); expect(servers).toHaveLength(1); expect(servers[0]?.host).toBe("localhost"); expect(servers[0]?.port).toBe(customPort); expect(servers[0]?.url).toBe(expectedUrl); expect(requests).toEqual([expectedUrl]); }).pipe(Effect.provide(layer));}); effectIt.effect("preserves explicit loopback hosts and bounds wildcard rewrites", () => { const ipv4Url = "https://127.0.0.1:43125/docs"; const ipv6Url = "http://[::1]:43126/docs"; const wildcardPrefix = "http://0.0.0.0/"; const maximumWildcardUrl = `${wildcardPrefix}${"a".repeat( PREVIEW_URL_MAX_LENGTH - wildcardPrefix.length, )}`; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeProbeFailureLayer(processProbeFailure, fetchFn); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const servers = yield* scanner.scan([ipv4Url, ipv6Url, maximumWildcardUrl]); expect(servers.map((server) => server.url)).toEqual([ipv4Url, ipv6Url]); expect(requests).toEqual([ipv4Url, ipv6Url]); }).pipe(Effect.provide(layer));}); effectIt.effect("projects configured paths independently for simultaneous subscribers", () => { const docsUrl = `http://localhost:${LSOF_TEST_PORT}/docs`; const adminUrl = `http://localhost:${LSOF_TEST_PORT}/admin`; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { const url = String(input); return Promise.resolve( url === docsUrl || url === adminUrl ? new Response("app", { headers: { "content-type": "text/html" } }) : new Response("not found", { status: 404 }), ); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const docsSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = []; const adminSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = []; yield* scanner.subscribe({ configuredUrls: [docsUrl], initialSnapshot: [] }, (servers) => Effect.sync(() => docsSnapshots.push(servers)), ); yield* scanner.subscribe({ configuredUrls: [adminUrl], initialSnapshot: [] }, (servers) => Effect.sync(() => adminSnapshots.push(servers)), ); yield* scanner.retain; expect(docsSnapshots.at(-1)?.[0]?.url).toBe(docsUrl); expect(adminSnapshots.at(-1)?.[0]?.url).toBe(adminUrl); }).pipe(Effect.scoped, Effect.provide(layer));}); effectIt.effect( "keeps each subscriber's candidates when their combined union exceeds the per-client cap", () => { const firstSubscriberUrls = Array.from( { length: CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS }, (_, index) => `http://localhost:${LSOF_TEST_PORT}/app-${index}`, ); const secondSubscriberUrl = `http://localhost:${LSOF_TEST_PORT}/app-${CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS}`; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => Promise.resolve( String(input) === secondSubscriberUrl ? new Response("app", { headers: { "content-type": "text/html" } }) : new Response("not found", { status: 404 }), )) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const secondSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = []; yield* scanner.subscribe( { configuredUrls: firstSubscriberUrls, initialSnapshot: [] }, () => Effect.void, ); yield* scanner.subscribe( { configuredUrls: [secondSubscriberUrl], initialSnapshot: [] }, (servers) => Effect.sync(() => secondSnapshots.push(servers)), ); yield* scanner.retain; expect(secondSnapshots.at(-1)?.[0]?.url).toBe(secondSubscriberUrl); }).pipe(Effect.scoped, Effect.provide(layer)); },); effectIt.effect("stops probing a subscriber's configured paths after its scope closes", () => { const docsUrl = `http://localhost:${LSOF_TEST_PORT}/docs`; const adminUrl = `http://localhost:${LSOF_TEST_PORT}/admin`; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { const url = String(input); requests.push(url); return Promise.resolve( url === docsUrl || url === adminUrl ? new Response("app", { headers: { "content-type": "text/html" } }) : new Response("not found", { status: 404 }), ); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const docsScope = yield* Scope.make(); yield* scanner .subscribe({ configuredUrls: [docsUrl], initialSnapshot: [] }, () => Effect.void) .pipe(Effect.provideService(Scope.Scope, docsScope)); yield* scanner.subscribe( { configuredUrls: [adminUrl], initialSnapshot: [] }, () => Effect.void, ); yield* scanner.retain; yield* Scope.close(docsScope, Exit.void); requests.length = 0; yield* TestClock.adjust(Duration.seconds(15)); expect(requests).toContain(adminUrl); expect(requests).not.toContain(docsUrl); }).pipe(Effect.scoped, Effect.provide(layer));}); effectIt.effect("uses the current configured fragment when readiness comes from cache", () => { const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); const oldUrl = `http://localhost:${LSOF_TEST_PORT}/docs#old`; const newUrl = `http://localhost:${LSOF_TEST_PORT}/docs#new`; return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; expect((yield* scanner.scan([oldUrl]))[0]?.url).toBe(oldUrl); const requestCount = requests.length; expect((yield* scanner.scan([newUrl]))[0]?.url).toBe(newUrl); expect(requests).toHaveLength(requestCount); }).pipe(Effect.provide(layer));}); effectIt.effect("shares a configured root probe with discovered-root classification", () => { const requests: string[] = []; const rootUrl = `http://localhost:${LSOF_TEST_PORT}/`; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; expect(yield* scanner.scan([rootUrl])).toHaveLength(1); expect(requests).toEqual([rootUrl]); yield* TestClock.adjust(Duration.seconds(15)); expect(yield* scanner.scan([rootUrl])).toHaveLength(1); expect(requests).toEqual([rootUrl, rootUrl]); }).pipe(Effect.provide(layer));}); effectIt.effect("starts fresh cache entries after the probing batch completes", () => Effect.gen(function* () { const baseClock = yield* Clock.Clock; const times = [0, 20_000, 20_000, 20_000]; let timeIndex = 0; const currentTimeMillis = () => times[Math.min(timeIndex++, times.length - 1)]!; const clock: Clock.Clock = { ...baseClock, currentTimeMillisUnsafe: currentTimeMillis, currentTimeMillis: Effect.sync(currentTimeMillis), }; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); Lines 1–500 of 667. Load more
import * as NodeNet from "node:net"; import { it as effectIt } from "@effect/vitest"; import { CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS, PREVIEW_URL_MAX_LENGTH, type DiscoveredLocalServer, } from "@t3tools/contracts"; import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import * as Net from "@t3tools/shared/Net"; import * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; import * as Duration from "effect/Duration"; 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 PlatformError from "effect/PlatformError"; import * as Scope from "effect/Scope"; import * as TestClock from "effect/testing/TestClock"; import { expect } from "vite-plus/test"; import { FetchHttpClient } from "effect/unstable/http"; import * as ProcessRunner from "../processRunner.ts"; import * as PortScanner from "./PortScanner.ts"; const processProbeFailure: ProcessRunner.ProcessRunner["Service"]["run"] = (input) => Effect.fail( new ProcessRunner.ProcessSpawnError({ command: input.command, argumentCount: input.args.length, cwd: input.cwd, cause: PlatformError.systemError({ _tag: "NotFound", module: "ChildProcess", method: "spawn", description: "PowerShell is not installed in the test environment", }), }), ); const TestProcessRunner = Layer.succeed(ProcessRunner.ProcessRunner, { run: processProbeFailure, }); let integrationListeningPort: number | null = null; const TestIntegrationNet = Layer.succeed(Net.NetService, { canListenOnHost: () => Effect.succeed(true), isPortAvailableOnLoopback: (port) => Effect.sync(() => port !== integrationListeningPort), hasListenerOnHost: (port) => Effect.sync(() => port === integrationListeningPort), reserveLoopbackPort: () => Effect.succeed(40_000), findAvailablePort: (preferred) => Effect.succeed(preferred), }); const makeProbeFailureLayer = ( run: ProcessRunner.ProcessRunner["Service"]["run"], fetch: typeof globalThis.fetch = globalThis.fetch, ) => PortScanner.layer.pipe( Layer.provide( Layer.mergeAll( Layer.succeed(ProcessRunner.ProcessRunner, { run }), Layer.succeed(Net.NetService, { canListenOnHost: () => Effect.succeed(true), isPortAvailableOnLoopback: () => Effect.succeed(true), hasListenerOnHost: () => Effect.succeed(false), reserveLoopbackPort: () => Effect.succeed(40_000), findAvailablePort: (preferred) => Effect.succeed(preferred), }), Layer.succeed(HostProcessPlatform, "linux"), FetchHttpClient.layer.pipe(Layer.provide(Layer.succeed(FetchHttpClient.Fetch, fetch))), ), ), ); const TestPortDiscoveryLive = PortScanner.layer.pipe( Layer.provide( Layer.mergeAll( TestProcessRunner, TestIntegrationNet, Layer.succeed(HostProcessPlatform, "win32"), FetchHttpClient.layer, ), ), ); const LSOF_TEST_PORT = 43_123; const makeLsofScannerLayer = (input: { readonly pid: () => number; readonly fetch: typeof globalThis.fetch; }) => PortScanner.layer.pipe( Layer.provide( Layer.mergeAll( Layer.succeed(ProcessRunner.ProcessRunner, { run: () => Effect.succeed({ stdout: `p${input.pid()}\ncnode\nn*:${LSOF_TEST_PORT}\n`, stderr: "", code: null, timedOut: false, stdoutTruncated: false, stderrTruncated: false, stdoutInvalidUtf8: false, stderrInvalidUtf8: false, }), }), Layer.succeed(Net.NetService, { canListenOnHost: () => Effect.succeed(true), isPortAvailableOnLoopback: () => Effect.succeed(true), hasListenerOnHost: () => Effect.succeed(false), reserveLoopbackPort: () => Effect.succeed(40_000), findAvailablePort: (preferred) => Effect.succeed(preferred), }), Layer.succeed(HostProcessPlatform, "linux"), FetchHttpClient.layer.pipe( Layer.provide(Layer.succeed(FetchHttpClient.Fetch, input.fetch)), ), ), ), ); const openServer = ( port: number, onConnection: (socket: NodeNet.Socket) => void, ): Effect.Effect<NodeNet.Server | null> => Effect.callback((resume) => { const server = NodeNet.createServer(onConnection); server.once("error", () => { resume(Effect.succeed(null)); }); server.listen(port, "127.0.0.1", () => { resume(Effect.succeed(server)); }); return Effect.sync(() => { server.close(); }); }); const closeServer = (server: NodeNet.Server): Effect.Effect<void> => Effect.callback((resume) => { server.close(() => resume(Effect.void)); }); const openCommonDevServer = Effect.fn("PortScannerTest.openCommonDevServer")(function* ( ports: ReadonlyArray<number>, onConnection: (socket: NodeNet.Socket) => void, ) { for (const port of ports) { const server = yield* openServer(port, onConnection); if (server !== null) return { port, server }; } return yield* Effect.die( new Error("No common development port was available for the preview scanner test"), ); }); const commonDevServer = Effect.acquireRelease( openCommonDevServer(PortScanner.COMMON_DEV_PORTS, (socket) => { socket.once("data", () => { socket.end("HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: 5\r\n\r\nhello"); }); }).pipe( Effect.tap(({ port }) => Effect.sync(() => { integrationListeningPort = port; }), ), ), ({ server }) => closeServer(server).pipe( Effect.ensuring( Effect.sync(() => { integrationListeningPort = null; }), ), ), ); const commonNonHttpServer = Effect.acquireRelease( openCommonDevServer(PortScanner.COMMON_DEV_PORTS.toReversed(), (socket) => { socket.on("error", () => undefined); socket.once("data", () => socket.end("MYSQL\r\n\r\n")); }).pipe( Effect.tap(({ port }) => Effect.sync(() => { integrationListeningPort = port; }), ), ), ({ server }) => closeServer(server).pipe( Effect.ensuring( Effect.sync(() => { integrationListeningPort = null; }), ), ), ); /** * Integration tests against a real TCP listener. We provide the Windows host * platform so the tests exercise the TCP-probe fallback without depending on * `lsof` being installed. */ effectIt.layer(TestPortDiscoveryLive)("PortDiscovery integration (TCP probe fallback)", (it) => { it.effect( "scan() returns an HTTP server we just opened on a curated dev port", Effect.fn("PortScannerTest.scanFindsCommonDevServer")(function* () { const { port } = yield* commonDevServer; const scanner = yield* PortScanner.PortDiscovery; const result = yield* scanner.scan(); const found = result.find((server) => server.port === port); expect(found).toBeDefined(); expect(found?.host).toBe("localhost"); }), ); it.effect( "scan() excludes a listening port that does not speak HTTP", Effect.fn("PortScannerTest.scanExcludesNonHttpServer")(function* () { const { port } = yield* commonNonHttpServer; const scanner = yield* PortScanner.PortDiscovery; const result = yield* scanner.scan(); expect(result.some((server) => server.port === port)).toBe(false); }), ); it.effect( "retain drives an immediate broadcast to subscribers", Effect.fn("PortScannerTest.retainBroadcastsImmediately")(function* () { const { port } = yield* commonDevServer; const received: number[] = []; const scanner = yield* PortScanner.PortDiscovery; yield* scanner.subscribe({ configuredUrls: [], initialSnapshot: [] }, (servers) => Effect.sync(() => { for (const server of servers) received.push(server.port); }), ); yield* scanner.retain; expect(received).toContain(port); }), ); }); effectIt.effect("revalidates a successful HTML probe after its cache entry expires", () => { let responds = true; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return responds ? Promise.resolve(new Response("hello", { headers: { "content-type": "text/html" } })) : Promise.reject(new TypeError("not HTTP")); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; expect(yield* scanner.scan()).toHaveLength(1); expect(yield* scanner.scan()).toHaveLength(1); expect(requests).toEqual([`http://localhost:${LSOF_TEST_PORT}/`]); responds = false; yield* TestClock.adjust(Duration.seconds(15)); expect(yield* scanner.scan()).toHaveLength(0); expect(requests).toEqual([ `http://localhost:${LSOF_TEST_PORT}/`, `http://localhost:${LSOF_TEST_PORT}/`, `https://localhost:${LSOF_TEST_PORT}/`, ]); }).pipe(Effect.provide(layer)); }); effectIt.effect("keeps a full configured URL when the discovered server root fails", () => { const requests: string[] = []; const configuredUrl = `http://localhost:${LSOF_TEST_PORT}/docs`; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { const url = String(input); requests.push(url); return Promise.resolve( url === configuredUrl ? new Response("docs", { headers: { "content-type": "text/html" } }) : new Response("not found", { status: 404, headers: { "content-type": "text/html" }, }), ); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const servers = yield* scanner.scan([configuredUrl]); expect(servers).toHaveLength(1); expect(servers[0]?.url).toBe(configuredUrl); expect(requests).toContain(configuredUrl); }).pipe(Effect.provide(layer)); }); effectIt.effect("probes configured custom ports through a canonical loopback host", () => { const customPort = 43_124; const configuredUrl = `http://0.0.0.0:${customPort}/docs`; const expectedUrl = `http://localhost:${customPort}/docs`; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeProbeFailureLayer(processProbeFailure, fetchFn); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const servers = yield* scanner.scan([configuredUrl]); expect(servers).toHaveLength(1); expect(servers[0]?.host).toBe("localhost"); expect(servers[0]?.port).toBe(customPort); expect(servers[0]?.url).toBe(expectedUrl); expect(requests).toEqual([expectedUrl]); }).pipe(Effect.provide(layer)); }); effectIt.effect("preserves explicit loopback hosts and bounds wildcard rewrites", () => { const ipv4Url = "https://127.0.0.1:43125/docs"; const ipv6Url = "http://[::1]:43126/docs"; const wildcardPrefix = "http://0.0.0.0/"; const maximumWildcardUrl = `${wildcardPrefix}${"a".repeat( PREVIEW_URL_MAX_LENGTH - wildcardPrefix.length, )}`; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeProbeFailureLayer(processProbeFailure, fetchFn); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const servers = yield* scanner.scan([ipv4Url, ipv6Url, maximumWildcardUrl]); expect(servers.map((server) => server.url)).toEqual([ipv4Url, ipv6Url]); expect(requests).toEqual([ipv4Url, ipv6Url]); }).pipe(Effect.provide(layer)); }); effectIt.effect("projects configured paths independently for simultaneous subscribers", () => { const docsUrl = `http://localhost:${LSOF_TEST_PORT}/docs`; const adminUrl = `http://localhost:${LSOF_TEST_PORT}/admin`; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { const url = String(input); return Promise.resolve( url === docsUrl || url === adminUrl ? new Response("app", { headers: { "content-type": "text/html" } }) : new Response("not found", { status: 404 }), ); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const docsSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = []; const adminSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = []; yield* scanner.subscribe({ configuredUrls: [docsUrl], initialSnapshot: [] }, (servers) => Effect.sync(() => docsSnapshots.push(servers)), ); yield* scanner.subscribe({ configuredUrls: [adminUrl], initialSnapshot: [] }, (servers) => Effect.sync(() => adminSnapshots.push(servers)), ); yield* scanner.retain; expect(docsSnapshots.at(-1)?.[0]?.url).toBe(docsUrl); expect(adminSnapshots.at(-1)?.[0]?.url).toBe(adminUrl); }).pipe(Effect.scoped, Effect.provide(layer)); }); effectIt.effect( "keeps each subscriber's candidates when their combined union exceeds the per-client cap", () => { const firstSubscriberUrls = Array.from( { length: CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS }, (_, index) => `http://localhost:${LSOF_TEST_PORT}/app-${index}`, ); const secondSubscriberUrl = `http://localhost:${LSOF_TEST_PORT}/app-${CONFIGURED_LOCAL_SERVER_URLS_MAX_ITEMS}`; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => Promise.resolve( String(input) === secondSubscriberUrl ? new Response("app", { headers: { "content-type": "text/html" } }) : new Response("not found", { status: 404 }), )) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const secondSnapshots: ReadonlyArray<DiscoveredLocalServer>[] = []; yield* scanner.subscribe( { configuredUrls: firstSubscriberUrls, initialSnapshot: [] }, () => Effect.void, ); yield* scanner.subscribe( { configuredUrls: [secondSubscriberUrl], initialSnapshot: [] }, (servers) => Effect.sync(() => secondSnapshots.push(servers)), ); yield* scanner.retain; expect(secondSnapshots.at(-1)?.[0]?.url).toBe(secondSubscriberUrl); }).pipe(Effect.scoped, Effect.provide(layer)); }, ); effectIt.effect("stops probing a subscriber's configured paths after its scope closes", () => { const docsUrl = `http://localhost:${LSOF_TEST_PORT}/docs`; const adminUrl = `http://localhost:${LSOF_TEST_PORT}/admin`; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { const url = String(input); requests.push(url); return Promise.resolve( url === docsUrl || url === adminUrl ? new Response("app", { headers: { "content-type": "text/html" } }) : new Response("not found", { status: 404 }), ); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; const docsScope = yield* Scope.make(); yield* scanner .subscribe({ configuredUrls: [docsUrl], initialSnapshot: [] }, () => Effect.void) .pipe(Effect.provideService(Scope.Scope, docsScope)); yield* scanner.subscribe( { configuredUrls: [adminUrl], initialSnapshot: [] }, () => Effect.void, ); yield* scanner.retain; yield* Scope.close(docsScope, Exit.void); requests.length = 0; yield* TestClock.adjust(Duration.seconds(15)); expect(requests).toContain(adminUrl); expect(requests).not.toContain(docsUrl); }).pipe(Effect.scoped, Effect.provide(layer)); }); effectIt.effect("uses the current configured fragment when readiness comes from cache", () => { const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("docs", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); const oldUrl = `http://localhost:${LSOF_TEST_PORT}/docs#old`; const newUrl = `http://localhost:${LSOF_TEST_PORT}/docs#new`; return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; expect((yield* scanner.scan([oldUrl]))[0]?.url).toBe(oldUrl); const requestCount = requests.length; expect((yield* scanner.scan([newUrl]))[0]?.url).toBe(newUrl); expect(requests).toHaveLength(requestCount); }).pipe(Effect.provide(layer)); }); effectIt.effect("shares a configured root probe with discovered-root classification", () => { const requests: string[] = []; const rootUrl = `http://localhost:${LSOF_TEST_PORT}/`; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn }); return Effect.gen(function* () { const scanner = yield* PortScanner.PortDiscovery; expect(yield* scanner.scan([rootUrl])).toHaveLength(1); expect(requests).toEqual([rootUrl]); yield* TestClock.adjust(Duration.seconds(15)); expect(yield* scanner.scan([rootUrl])).toHaveLength(1); expect(requests).toEqual([rootUrl, rootUrl]); }).pipe(Effect.provide(layer)); }); effectIt.effect("starts fresh cache entries after the probing batch completes", () => Effect.gen(function* () { const baseClock = yield* Clock.Clock; const times = [0, 20_000, 20_000, 20_000]; let timeIndex = 0; const currentTimeMillis = () => times[Math.min(timeIndex++, times.length - 1)]!; const clock: Clock.Clock = { ...baseClock, currentTimeMillisUnsafe: currentTimeMillis, currentTimeMillis: Effect.sync(currentTimeMillis), }; const requests: string[] = []; const fetchFn = ((input: Parameters<typeof globalThis.fetch>[0]) => { requests.push(String(input)); return Promise.resolve(new Response("app", { headers: { "content-type": "text/html" } })); }) as typeof globalThis.fetch; const layer = makeLsofScannerLayer({ pid: () => 1234, fetch: fetchFn });