effect-portable-patterns

Effect：可移植的 Promise 工具

将 Effect 用作轻量级工具，以稳健地运行 Promise。每个 Effect 都是自包含的（不需要服务、层或依赖注入），并通过边界处的 Effect.runPromise 解析为普通 Promise。

结构始终为：Effect.fn 或 Effect.gen -> 管道操作符 -> Effect.runPromise。

快速参考

能力	API	避免
封装 Promise	Effect.tryPromise	Effect.promise（会吞掉错误）
定义函数	Effect.fn("name")	匿名生成器
错误	使用 _tag 的 Data.TaggedError	普通 Error 或未标记的对象
按标签捕获	catchTag / catchTags	catchAll（会丢失类型缩窄）
回退	orElse、orElseSucceed	嵌套 try/catch
超时	Effect.timeout / Effect.timeoutFail	手动使用 AbortController + setTimeout
重试	使用 Schedule 的 Effect.retry	手动重试循环
缓存	Effect.cachedWithTTL / Effect.cachedFunction	基于 Map 的手动缓存
并发	带 { concurrency: N } 的 Effect.all	手动对 Promise.all 分块
模式匹配	使用 Match.tag 的 Match.value / Match.type	对 _tag 使用 Switch 语句
追踪	Effect.withSpan / Effect.annotateCurrentSpan	手动使用 console.time
在边界处运行	Effect.runPromise	对异步工作使用 Effect.runSync

核心模式：可移植的 Effect 函数

每个 Effect 函数都遵循以下结构——构建一个 Effect<Success, Error, never>（无依赖项），然后在调用处将其作为 Promise 运行。

import { Effect, Data } from "effect";

class FetchError extends Data.TaggedError("FetchError")<{
  url: string;
  status: number;
  message: string;
}> {}

const fetchUser = Effect.fn("fetchUser")(function* (userId: string) {
  const response = yield* Effect.tryPromise({
    try: () => fetch(`/api/users/${userId}`),
    catch: () =>
      new FetchError({
        url: `/api/users/${userId}`,
        status: 0,
        message: "Network error",
      }),
  });

  if (!response.ok) {
    return yield* Effect.fail(
      new FetchError({
        url: `/api/users/${userId}`,
        status: response.status,
        message: response.statusText,
      }),
    );
  }

  const user = yield* Effect.tryPromise({
    try: () => response.json() as Promise<User>,
    catch: () =>
      new FetchError({
        url: `/api/users/${userId}`,
        status: response.status,
        message: "Invalid JSON",
      }),
  });

  return user;
});

// At the call site - always resolves to a plain Promise
const user: User = await Effect.runPromise(fetchUser("123"));

带标签的错误

使用 Data.TaggedError 定义错误。_tag 字段支持类型安全的错误匹配，无需服务或模式。

import { Data } from "effect";

class TimeoutError extends Data.TaggedError("TimeoutError")<{
  operation: string;
  durationMs: number;
}> {}

class NotFoundError extends Data.TaggedError("NotFoundError")<{
  resource: string;
  id: string;
}> {}

class ValidationError extends Data.TaggedError("ValidationError")<{
  field: string;
  message: string;
}> {}

按标签捕获错误

单个标签使用 catchTag，多个标签使用 catchTags。两者都会保留类型缩窄。

const result =
  yield *
  fetchUser("123").pipe(
    Effect.catchTag("NotFoundError", (error) =>
      Effect.succeed({ id: error.id, name: "Unknown", fallback: true }),
    ),
    Effect.catchTag("TimeoutError", (error) =>
      Effect.fail(
        new ServiceUnavailableError({
          message: `${error.operation} timed out`,
        }),
      ),
    ),
  );

// Or handle multiple tags at once
const result =
  yield *
  fetchUser("123").pipe(
    Effect.catchTags({
      NotFoundError: (error) => Effect.succeed(defaultUser),
      ValidationError: (error) => Effect.fail(new BadRequestError({ message: error.message })),
    }),
  );

超时

基本超时（抛出 TimeoutException）

const result = yield * fetchUser("123").pipe(Effect.timeout("5 seconds"));

使用自定义错误的超时

const result =
  yield *
  fetchUser("123").pipe(
    Effect.timeoutFail({
      duration: "5 seconds",
      onTimeout: () => new TimeoutError({ operation: "fetchUser", durationMs: 5000 }),
    }),
  );

超时并使用回退值

const result =
  yield *
  fetchUser("123").pipe(
    Effect.timeoutTo({
      duration: "5 seconds",
      onSuccess: (user) => user,
      onTimeout: () => defaultUser,
    }),
  );

重试

固定重试次数

import { Schedule } from "effect";

const result = yield * fetchUser("123").pipe(Effect.retry({ times: 3 }));

指数退避

const result =
  yield *
  fetchUser("123").pipe(
    Effect.retry(Schedule.exponential("100 millis").pipe(Schedule.compose(Schedule.recurs(5)))),
  );

仅重试特定错误

const result =
  yield *
  fetchUser("123").pipe(
    Effect.retry({
      times: 3,
      while: (error) => error._tag === "TimeoutError",
    }),
  );

重试耗尽后使用回退值

const result =
  yield *
  Effect.retryOrElse(fetchUser("123"), { times: 3 }, (error, fiberId) =>
    Effect.succeed(defaultUser),
  );

组合超时与重试

const robustFetch = Effect.fn("robustFetch")(function* (userId: string) {
  const user = yield* fetchUser(userId).pipe(
    Effect.timeoutFail({
      duration: "3 seconds",
      onTimeout: () => new TimeoutError({ operation: "fetchUser", durationMs: 3000 }),
    }),
    Effect.retry(Schedule.exponential("200 millis").pipe(Schedule.compose(Schedule.recurs(3)))),
  );
  return user;
});

回退

// Try primary, fall back to secondary on any failure
const result = yield * fetchFromPrimary(id).pipe(Effect.orElse(() => fetchFromSecondary(id)));

// Fall back to a default value
const result = yield * fetchUser(id).pipe(Effect.orElseSucceed(() => defaultUser));

// Remap the error type on failure
const result =
  yield *
  fetchUser(id).pipe(
    Effect.orElseFail(() => new ServiceUnavailableError({ message: "All sources failed" })),
  );

// Try multiple sources, use the first that succeeds
const result =
  yield * Effect.firstSuccessOf([fetchFromCache(id), fetchFromPrimary(id), fetchFromSecondary(id)]);

缓存

使用 TTL 缓存 Effect

import { Effect } from "effect";

const cachedConfig = Effect.cachedWithTTL(
  Effect.tryPromise(() => fetch("/api/config").then((r) => r.json())),
  "5 minutes",
);

// Use it - first call fetches, subsequent calls return cached value within TTL
const program = Effect.gen(function* () {
  const getConfig = yield* cachedConfig;
  const config1 = yield* getConfig;
  const config2 = yield* getConfig; // same value, no re-fetch
});

使用手动失效的缓存

const [getConfig, invalidate] = yield * Effect.cachedInvalidateWithTTL(fetchConfig, "10 minutes");

const config = yield * getConfig;
yield * invalidate; // force re-fetch on next call

按参数记忆化函数

const memoizedFetchUser =
  yield *
  Effect.cachedFunction((userId: string) =>
    Effect.tryPromise(() => fetch(`/api/users/${userId}`).then((r) => r.json())),
  );

const user1 = yield * memoizedFetchUser("123"); // fetches
const user2 = yield * memoizedFetchUser("123"); // returns cached
const user3 = yield * memoizedFetchUser("456"); // fetches (different key)

并发

并行执行

// Run all effects in parallel (unbounded)
const [users, posts, comments] =
  yield *
  Effect.all([fetchUsers, fetchPosts, fetchComments], {
    concurrency: "unbounded",
  });

// Bounded concurrency (e.g., max 5 at a time)
const results = yield * Effect.all(tasks, { concurrency: 5 });

// Parallel forEach
const enrichedUsers = yield * Effect.forEach(userIds, (id) => fetchUser(id), { concurrency: 10 });

竞速（首个成功者）

// Race two effects, take the first to complete
const result = yield * Effect.race(fetchFromEast, fetchFromWest);

// Race many effects
const result = yield * Effect.raceAll([fetchFromCache(id), fetchFromDb(id), fetchFromRemote(id)]);

模式匹配

使用 Match 对带标签的联合类型和错误类型进行穷尽且类型安全的分支处理。

匹配带标签的错误

import { Match } from "effect";

type ApiError = NotFoundError | TimeoutError | ValidationError;

const describeError = (error: ApiError) =>
  Match.value(error).pipe(
    Match.tag("NotFoundError", (e) => `${e.resource} ${e.id} not found`),
    Match.tag("TimeoutError", (e) => `${e.operation} timed out after ${e.durationMs}ms`),
    Match.tag("ValidationError", (e) => `Invalid ${e.field}: ${e.message}`),
    Match.exhaustive,
  );

按类型匹配（可复用匹配器）

const handleResult = Match.type<string | number | boolean>().pipe(
  Match.when(Match.string, (s) => `string: ${s}`),
  Match.when(Match.number, (n) => `number: ${n}`),
  Match.when(Match.boolean, (b) => `boolean: ${b}`),
  Match.exhaustive,
);

handleResult("hello"); // "string: hello"
handleResult(42); // "number: 42"

使用谓词进行匹配

const categorize = Match.type<{ status: number }>().pipe(
  Match.when({ status: (s) => s >= 500 }, () => "server_error"),
  Match.when({ status: (s) => s >= 400 }, () => "client_error"),
  Match.when({ status: (s) => s >= 200 }, () => "success"),
  Match.orElse(() => "unknown"),
);

追踪

添加 Span

Effect.fn 会自动创建一个带指定名称的 Span。若要添加更多 Span 或注解：

const fetchAndProcess = Effect.fn("fetchAndProcess")(function* (userId: string) {
  yield* Effect.annotateCurrentSpan("userId", userId);

  const user = yield* fetchUser(userId).pipe(Effect.withSpan("fetchUser"));

  const processed = yield* processUser(user).pipe(Effect.withSpan("processUser"));

  yield* Effect.annotateCurrentSpan("processed", true);
  return processed;
});

在 Effect 中记录日志

const program = Effect.fn("program")(function* () {
  yield* Effect.log("Starting operation");
  const result = yield* doWork();
  yield* Effect.log("Operation complete", { resultId: result.id });
  return result;
});

完整示例：稳健的 API 调用

将所有模式组合到一个可移植函数中：

import { Effect, Data, Schedule, Match } from "effect";

class ApiError extends Data.TaggedError("ApiError")<{
  url: string;
  status: number;
  message: string;
}> {}

class ApiTimeoutError extends Data.TaggedError("ApiTimeoutError")<{
  url: string;
  durationMs: number;
}> {}

const fetchApi = Effect.fn("fetchApi")(function* <T>(url: string) {
  yield* Effect.annotateCurrentSpan("url", url);
  yield* Effect.log("Fetching", { url });

  const response = yield* Effect.tryPromise({
    try: () => fetch(url),
    catch: () => new ApiError({ url, status: 0, message: "Network error" }),
  }).pipe(
    Effect.timeoutFail({
      duration: "10 seconds",
      onTimeout: () => new ApiTimeoutError({ url, durationMs: 10_000 }),
    }),
    Effect.retry(Schedule.exponential("500 millis").pipe(Schedule.compose(Schedule.recurs(3)))),
  );

  if (!response.ok) {
    return yield* Effect.fail(
      new ApiError({
        url,
        status: response.status,
        message: response.statusText,
      }),
    );
  }

  return yield* Effect.tryPromise({
    try: () => response.json() as Promise<T>,
    catch: () => new ApiError({ url, status: response.status, message: "Invalid JSON" }),
  });
});

// Usage at the boundary - resolves to a plain Promise
const data = await Effect.runPromise(
  fetchApi<User>("/api/users/123").pipe(
    Effect.catchTag("ApiTimeoutError", () =>
      Effect.succeed({ id: "123", name: "Unknown" } as User),
    ),
  ),
);

规则

规则	指引
不使用服务或层	将副作用限制在自身范围内，并使用 R = never。仅在适用 effect-best-practices 技能时使用服务。
始终使用 Effect.fn	为每个函数命名，以便自动创建跟踪跨度。
始终使用 Data.TaggedError	每种不同的失败都应有自己的标记错误类。
始终使用 Effect.tryPromise	切勿使用 Effect.promise——它会将错误吞掉并视为缺陷。
优先使用 catchTag	使用 catchTag / catchTags 而不是 catchAll，以保留类型收窄。
为所有操作设置超时	外部调用始终应使用 Effect.timeout 或 Effect.timeoutFail。
在边界处运行	在最外层的调用位置调用 Effect.runPromise，不要在 Effect 函数内部调用。

参考资料

• 创建 Effect
• 预期错误
• 回退
• 重试
• 超时
• 模式匹配
• 基本并发
• 缓存 Effect
• 跟踪