test(cli-e2e): cover the e2e harness with effect lint (CLI-2878)

.oxlintrc.effect.json

@@ -9,12 +9,17 @@
     "!packages/api/**",
     "!packages/config/**",
     "!packages/stack/**",
+    "!apps/cli-e2e/src/server/**",
+    "!apps/cli-e2e/src/tests/env.ts",
+    "!apps/cli-e2e/tests/**",
+    "!apps/cli-e2e/vitest.config.ts",
     "!apps/cli/**",
     "!apps/docs/**",
     "!packages/cli-test-helpers/**",
     "!tools/**",
     // Last match wins across the whole list: keep bare re-exclusions after every
     // `!` entry that would otherwise re-include them.
+    "apps/cli-e2e/src/server/replay-server.ts",
     "apps/cli/src/shared/cli/bin.ts",
     "apps/cli/src/shared/compute/stacks/**",
     "apps/cli/src/shared/feedback/database.types.ts",

apps/cli-e2e/package.json

@@ -12,7 +12,9 @@
     "types:check": "tsc --noEmit"
   },
   "dependencies": {
-    "@supabase/cli-test-helpers": "workspace:*"
+    "@effect/platform-bun": "catalog:",
+    "@supabase/cli-test-helpers": "workspace:*",
+    "effect": "catalog:"
   },
   "devDependencies": {
     "@tsconfig/bun": "catalog:",

apps/cli-e2e/scripts/replay-serve.ts

@@ -21,6 +21,8 @@ const server = await startReplayServer({
   fixturesDir: FIXTURES_DIR,
   hostname: "127.0.0.1",
   port: parsePort(process.argv.slice(2)),
+  record: false,
+  stagingUrl: undefined,
 });
 
 let stopping = false;

apps/cli-e2e/src/server/fixture-loader.ts

@@ -1,59 +1,86 @@
-import { existsSync, readdirSync, readFileSync } from "node:fs";
-import { join } from "node:path";
-
-export interface FixtureRequest {
-  method: string;
-  path: string;
-  query: Record<string, string>;
-  headers: Record<string, string>;
-  body: unknown;
-}
+import { Data, Effect, FileSystem, Formatter, Path, Schema } from "effect";
 
-export interface FixtureResponse {
-  status: number;
-  headers: Record<string, string>;
-  body: unknown;
-}
+/* Hand-authored fixtures may omit headers (served as none) and the response body (served empty). */
+const FixtureHeaders = Schema.Record(Schema.String, Schema.String).pipe(
+  Schema.withDecodingDefaultKey(Effect.sync(() => ({}))),
+);
 
-export interface FixtureEntry {
-  request: FixtureRequest;
-  response: FixtureResponse;
-}
+const FixtureRequest = Schema.Struct({
+  method: Schema.String,
+  path: Schema.mutableKey(Schema.String),
+  query: Schema.Record(Schema.String, Schema.String),
+  headers: FixtureHeaders,
+  body: Schema.Unknown,
+});
+
+const FixtureResponse = Schema.Struct({
+  status: Schema.Finite,
+  headers: FixtureHeaders,
+  body: Schema.optionalKey(Schema.Unknown),
+});
+
+const FixtureEntry = Schema.Struct({ request: FixtureRequest, response: FixtureResponse });
+
+export type FixtureRequest = typeof FixtureRequest.Type;
+export type FixtureResponse = typeof FixtureResponse.Type;
+export type FixtureEntry = typeof FixtureEntry.Type;
 
 /** All fixtures indexed by their key (e.g. "GET_v1_projects"), each holding
  *  an ordered queue of entries (for sequential calls to the same endpoint). */
 export type FixtureStore = Map<string, FixtureEntry[]>;
 
+/* Mirrors `existsSync`: any failure to reach the path counts as absent. */
+const pathExists = (fs: FileSystem.FileSystem, path: string) =>
+  fs.exists(path).pipe(Effect.orElseSucceed(() => false));
+
+/** A fixture file is missing or does not hold valid JSON. */
+class FixtureFileError extends Data.Error<{ readonly message: string; readonly cause?: unknown }> {}
+
 /** Load an ordered scenario from scenarios/<name>/interactions.json.
  *  Returns null if the scenario file does not exist — caller decides whether
  *  to fail loudly or fall back to per-endpoint fixtures. */
-export function loadScenario(scenariosDir: string, name: string): FixtureEntry[] | null {
-  const scenarioFile = join(scenariosDir, name, "interactions.json");
-  if (!existsSync(scenarioFile)) return null;
-  return parseFixtureFile<FixtureEntry[]>(scenarioFile);
-}
+export const loadScenario = Effect.fn("FixtureLoader.loadScenario")(function* (
+  scenariosDir: string,
+  name: string,
+) {
+  const fs = yield* FileSystem.FileSystem;
+  const path = yield* Path.Path;
+  const scenarioFile = path.join(scenariosDir, name, "interactions.json");
+  if (!(yield* pathExists(fs, scenarioFile))) return null;
+  const entries: FixtureEntry[] = yield* parseFixtureFile(
+    scenarioFile,
+    Schema.mutable(Schema.Array(FixtureEntry)),
+  );
+  return entries;
+});
 
 /** Load all fixture pairs from the recorded/ directory into a FixtureStore.
  *  Fails fast with a descriptive error if any fixture file is malformed. */
-export function loadFixtures(fixturesDir: string): FixtureStore {
-  const recordedDir = join(fixturesDir, "recorded");
+export const loadFixtures = Effect.fn("FixtureLoader.loadFixtures")(function* (
+  fixturesDir: string,
+) {
+  const fs = yield* FileSystem.FileSystem;
+  const path = yield* Path.Path;
+  const recordedDir = path.join(fixturesDir, "recorded");
   const store: FixtureStore = new Map();
 
-  if (!existsSync(recordedDir)) {
+  if (!(yield* pathExists(fs, recordedDir))) {
     return store;
   }
 
-  const keys = readdirSync(recordedDir, { withFileTypes: true })
-    .filter((d) => d.isDirectory())
-    .map((d) => d.name);
+  const keys: string[] = [];
+  for (const name of yield* fs.readDirectory(recordedDir)) {
+    const info = yield* fs.stat(path.join(recordedDir, name));
+    if (info.type === "Directory") keys.push(name);
+  }
 
   for (const key of keys) {
-    const keyDir = join(recordedDir, key);
+    const keyDir = path.join(recordedDir, key);
     const entries: FixtureEntry[] = [];
 
     // Collect numbered pairs: 1.request.json/1.response.json, 2.request.json, ...
     // Also accept "default" as an alias for "1".
-    const files = readdirSync(keyDir).sort();
+    const files = (yield* fs.readDirectory(keyDir)).sort();
     const indices = new Set<string>();
 
     for (const file of files) {
@@ -62,11 +89,11 @@ export function loadFixtures(fixturesDir: string): FixtureStore {
     }
 
     for (const index of [...indices].sort(compareIndices)) {
-      const reqFile = join(keyDir, `${index}.request.json`);
-      const resFile = join(keyDir, `${index}.response.json`);
+      const reqFile = path.join(keyDir, `${index}.request.json`);
+      const resFile = path.join(keyDir, `${index}.response.json`);
 
-      const request = parseFixtureFile<FixtureRequest>(reqFile);
-      const response = parseFixtureFile<FixtureResponse>(resFile);
+      const request = yield* parseFixtureFile(reqFile, FixtureRequest);
+      const response = yield* parseFixtureFile(resFile, FixtureResponse);
       entries.push({ request, response });
     }
 
@@ -75,19 +102,31 @@ export function loadFixtures(fixturesDir: string): FixtureStore {
     }
   }
 
+  yield* Effect.annotateCurrentSpan("fixture.key_count", store.size);
   return store;
-}
-
-function parseFixtureFile<T>(filePath: string): T {
-  if (!existsSync(filePath)) {
-    throw new Error(`Missing fixture file: ${filePath}`);
-  }
-  try {
-    return JSON.parse(readFileSync(filePath, "utf8")) as T;
-  } catch (cause) {
-    throw new Error(`Malformed fixture file: ${filePath}`, { cause });
+});
+
+const parseFixtureFile = Effect.fnUntraced(function* <S extends Schema.Constraint>(
+  filePath: string,
+  schema: S,
+) {
+  const fs = yield* FileSystem.FileSystem;
+  if (!(yield* pathExists(fs, filePath))) {
+    return yield* new FixtureFileError({ message: `Missing fixture file: ${filePath}` });
   }
-}
+  const malformed = (cause: unknown) =>
+    new FixtureFileError({ message: `Malformed fixture file: ${filePath}`, cause });
+  const bytes = yield* fs.readFile(filePath).pipe(Effect.mapError(malformed));
+  const text = new TextDecoder("utf-8", { ignoreBOM: true }).decode(bytes);
+  return yield* Schema.decodeEffect(Schema.fromJsonString(schema))(text).pipe(
+    Effect.mapError(
+      (error) =>
+        new FixtureFileError({
+          message: `Malformed fixture file: ${filePath}\n  ${error.message}: ${Formatter.format(text.slice(0, 200))}`,
+        }),
+    ),
+  );
+});
 
 /** Sort indices so "default" comes first, then numerically. */
 function compareIndices(a: string, b: string): number {

apps/cli-e2e/src/server/pg-mock.ts

@@ -22,6 +22,8 @@
  *   - Terminate ('X') → closes connection gracefully
  */
 
+import { Console, Effect } from "effect";
+
 export interface PgFixture {
   /** Lowercase column names, matching the field names the inspect commands map query results onto. */
   columns: string[];
@@ -440,13 +442,13 @@ export function startPgMock(): PgMockHandle {
         try {
           processMessages(socket, () => state);
         } catch (err) {
-          console.error("[pg-mock] error processing message:", err);
+          Effect.runSync(Console.error("[pg-mock] error processing message:", err));
           socket.end();
         }
       },
       close(_socket) {},
       error(_socket, err) {
-        console.error("[pg-mock] socket error:", err);
+        Effect.runSync(Console.error("[pg-mock] socket error:", err));
       },
     },
   });

apps/cli-e2e/src/server/replay-server.ts

@@ -2,6 +2,8 @@ import { existsSync, mkdirSync, readdirSync, rmSync, unlinkSync, writeFileSync }
 import { request as httpRequest } from "node:http";
 import { join } from "node:path";
 import { URL } from "node:url";
+import { BunFileSystem, BunPath } from "@effect/platform-bun";
+import { Effect, Layer } from "effect";
 import type {
   FixtureEntry,
   FixtureRequest,
@@ -157,11 +159,17 @@ interface ReplayServerOptions {
   hostname?: string;
   /** Optional Postgres mock server to control via /_ctrl/pg-* endpoints. */
   pgMock?: PgMockHandle;
+  /** Proxy to staging and record fixtures instead of replaying them. */
+  record: boolean;
+  /** Staging Management API base URL; required when recording. */
+  stagingUrl: string | undefined;
 }
 
+const fixtureFileLayer = Layer.mergeAll(BunFileSystem.layer, BunPath.layer);
+
 export async function startReplayServer(options: ReplayServerOptions): Promise<ReplayServerHandle> {
-  const isRecord = process.env["RECORD"] === "true";
-  const stagingUrl = process.env["SUPABASE_STAGING_URL"];
+  const isRecord = options.record;
+  const stagingUrl = options.stagingUrl;
 
   if (isRecord && !stagingUrl) {
     throw new Error("RECORD=true requires SUPABASE_STAGING_URL to be set");
@@ -175,7 +183,11 @@ export async function startReplayServer(options: ReplayServerOptions): Promise<R
     rmSync(join(options.fixturesDir, "scenarios"), { recursive: true, force: true });
   }
 
-  const store: FixtureStore = isRecord ? new Map() : loadFixtures(options.fixturesDir);
+  const store: FixtureStore = isRecord
+    ? new Map()
+    : await Effect.runPromise(
+        loadFixtures(options.fixturesDir).pipe(Effect.provide(fixtureFileLayer)),
+      );
 
   const counters: SequenceCounters = new Map();
   const requestLog: RecordedRequest[] = [];
@@ -923,7 +935,11 @@ async function handleControl(req: Request, url: URL, ctx: ControlContext): Promi
       const { name } = (await req.json()) as { name: string };
 
       if (!ctx.isRecord) {
-        const interactions = loadScenario(join(ctx.fixturesDir, "scenarios"), name);
+        const interactions = await Effect.runPromise(
+          loadScenario(join(ctx.fixturesDir, "scenarios"), name).pipe(
+            Effect.provide(fixtureFileLayer),
+          ),
+        );
         if (!interactions) {
           return new Response(
             JSON.stringify({

apps/cli-e2e/src/tests/env.ts

@@ -1,64 +1,60 @@
+import { Config, ConfigProvider, Effect, Option } from "effect";
 import { REPLAY_PROJECT_REF } from "../server/placeholder.ts";
 
-type CliE2eMode = "replay" | "record";
+/** Reads an environment variable through `Config`, keeping empty values as set. */
+const readEnv = (name: string): string | undefined =>
+  Effect.runSync(
+    Config.option(Config.String(name))
+      .parse(ConfigProvider.fromEnv({ preserveEmptyStrings: true }))
+      .pipe(Effect.map(Option.getOrUndefined)),
+  );
 
 // Runtime mode. `replay` (default) serves recorded fixtures; `record` proxies to
 // staging and captures fixtures.
 // Back-compat: RECORD=true still maps to `record`.
-const MODE: CliE2eMode =
-  (process.env["CLI_E2E_MODE"] as CliE2eMode | undefined) ??
-  (process.env["RECORD"] === "true" ? "record" : "replay");
+const MODE = readEnv("CLI_E2E_MODE") ?? (readEnv("RECORD") === "true" ? "record" : "replay");
 
 export const isRecording = MODE === "record";
 
-// The replay server + tests/setup.ts key recording off RECORD directly, so keep
-// it in sync with MODE in both directions: an explicit CLI_E2E_MODE=replay must
-// not record and wipe fixtures just because RECORD=true lingers in the shell.
-if (isRecording) {
-  process.env["RECORD"] = "true";
-} else {
-  delete process.env["RECORD"];
-}
-
-// startReplayServer + tests/setup.ts read SUPABASE_STAGING_URL directly as the
-// record proxy target. Normalise it from CLI_E2E_API_URL so
+// The record proxy target. Normalised from CLI_E2E_API_URL so
 // `CLI_E2E_MODE=record CLI_E2E_API_URL=…` works without also setting the legacy var.
-if (isRecording && !process.env["SUPABASE_STAGING_URL"] && process.env["CLI_E2E_API_URL"]) {
-  process.env["SUPABASE_STAGING_URL"] = process.env["CLI_E2E_API_URL"];
-}
+const LEGACY_STAGING_URL = readEnv("SUPABASE_STAGING_URL");
+const API_URL = readEnv("CLI_E2E_API_URL");
+export const STAGING_URL =
+  isRecording && !LEGACY_STAGING_URL && API_URL ? API_URL : LEGACY_STAGING_URL;
 
 // In replay mode the token never reaches a real API, but the CLI validates the
 // format before making any request (must match sbp_[a-f0-9]{40}). In record mode
 // it must be a valid token for the staging API.
 export const ACCESS_TOKEN =
-  process.env["SUPABASE_ACCESS_TOKEN"] ?? "sbp_0000000000000000000000000000000000000000";
+  readEnv("SUPABASE_ACCESS_TOKEN") ?? "sbp_0000000000000000000000000000000000000000";
 
 // Region for the fresh recording project.
-export const REGION = process.env["CLI_E2E_REGION"] ?? "us-east-1";
+export const REGION = readEnv("CLI_E2E_REGION") ?? "us-east-1";
 
 // Recorded fixtures store every project ref and org slug as REPLAY_PROJECT_REF, so replay
 // requests use it too. In record mode supply a real project ref via env.
-export const PROJECT_REF = process.env["SUPABASE_TEST_PROJECT_REF"] ?? REPLAY_PROJECT_REF;
+export const PROJECT_REF = readEnv("SUPABASE_TEST_PROJECT_REF") ?? REPLAY_PROJECT_REF;
 
 // In record mode supply a real org slug via env, or let the resolver derive it.
-export const ORG_ID = process.env["SUPABASE_TEST_ORG_ID"] ?? REPLAY_PROJECT_REF;
+export const ORG_ID = readEnv("SUPABASE_TEST_ORG_ID") ?? REPLAY_PROJECT_REF;
 
 // UUID of an existing SAML provider on the staging project.
 // In replay mode any UUID normalises to __UUID__ in fixture paths.
 // In record mode supply a real provider ID via env.
 export const PROVIDER_ID =
-  process.env["SUPABASE_TEST_PROVIDER_ID"] ?? "00000000-0000-0000-0000-000000000000";
+  readEnv("SUPABASE_TEST_PROVIDER_ID") ?? "00000000-0000-0000-0000-000000000000";
 
 // UUID of an existing SQL snippet on the staging project.
 // In replay mode any UUID normalises to __UUID__ in fixture paths.
 // In record mode supply a real snippet UUID via env.
 export const SNIPPET_ID =
-  process.env["SUPABASE_TEST_SNIPPET_ID"] ?? "00000000-0000-4000-8000-000000000001";
+  readEnv("SUPABASE_TEST_SNIPPET_ID") ?? "00000000-0000-4000-8000-000000000001";
 
 // Unix epoch seconds for a PITR restore timestamp within the staging project's backup window.
 // In replay mode the replay server serves responses in order regardless of the request body value.
 // In record mode supply a real timestamp (within the backup window) via env.
 export const BACKUP_TIMESTAMP = parseInt(
-  process.env["SUPABASE_TEST_BACKUP_TIMESTAMP"] ?? "1707407047",
+  readEnv("SUPABASE_TEST_BACKUP_TIMESTAMP") ?? "1707407047",
   10,
 );