Effect-TS 与函数式编程：类型化的副作用、依赖注入与并发

Effect<A, E, R> 三个类型参数：
  A = 成功值类型（Success）
  E = 错误类型（Error channel，never 表示不会失败）
  R = 所需依赖（Requirements，never 表示无依赖）

import { Effect } from "effect"

type FetchUser = Effect.Effect<User, HttpError | ParseError, HttpClient>
//           成功 User ↑   ↑ 可能失败的两类错误        ↑ 需要注入 HttpClient

import { Effect } from "effect"

const ok = Effect.succeed(42)                  // Effect<number, never, never>
const bad = Effect.fail(new Error("boom"))     // Effect<never, Error, never>
const parsed = Effect.try({
  try: () => JSON.parse('{"a":1}') as unknown,
  catch: (e) => new ParseError(String(e)),     // 同步异常映射为 E 通道
})

const program = Effect.succeed(1).pipe(Effect.map((n) => n + 1))
// 此时什么都没发生
await Effect.runPromise(program)   // 2

import { pipe } from "effect"
const program = pipe(Effect.succeed(2), Effect.map((n) => n * 10))

import { Effect } from "effect"

const program = Effect.gen(function* () {
  const user = yield* fetchUser("1")          // 自动 flatMap
  const posts = yield* fetchPosts(user.id)
  return { user, posts }
})

async/await        Effect.gen
await  → throw     yield* → 错误进 E 通道，不抛
try/catch          Effect.catchAll
隐式 Promise       Effect<A, E, R> 显式三通道

import { Context, Effect } from "effect"

class HttpClient extends Context.Tag("HttpClient")<
  HttpClient,
  { get: (url: string) => Effect.Effect<string, HttpError> }
>() {}

const fetchUser = (id: string): Effect.Effect<User, HttpError, HttpClient> =>
  Effect.gen(function* () {
    const http = yield* HttpClient
    const raw = yield* http.get(`/users/${id}`)
    return JSON.parse(raw) as User
  })

import { Layer, Effect } from "effect"

const HttpClientLive = Layer.succeed(HttpClient, {
  get: (url) => Effect.tryPromise({
    try: () => fetch(url).then((r) => r.text()),
    catch: (e) => new HttpError(String(e)),
  }),
})

const UserRepoLive = Layer.effect(UserRepo, Effect.gen(function* () {
  const http = yield* HttpClient
  return { find: (id: string) => fetchUser(id) }
}))

const AppLive = UserRepoLive.pipe(Layer.provide(HttpClientLive))
await Effect.runPromise(program.pipe(Effect.provide(AppLive)))

const recovered = program.pipe(
  Effect.catchAll((err: HttpError | ParseError) => Effect.succeed(defaultUser)),
)

import { Effect, Data } from "effect"

class HttpError extends Data.TaggedError("HttpError")<{ status: number }> {}
class ParseError extends Data.TaggedError("ParseError")<{ raw: string }> {}

const handled = program.pipe(
  Effect.catchTag("HttpError", () => Effect.succeed(fallback)),
  Effect.catchTag("ParseError", (e) => Effect.fail(new FatalError(e.raw))),
)

import { Effect } from "effect"

// 并行执行，收集所有结果
const all = Effect.all([fetchUser("1"), fetchUser("2")], { concurrency: "unbounded" })

// 并发受限（最多 5 个同时）
const limited = Effect.forEach(ids, (id) => fetchUser(id), { concurrency: 5 })

const winner = Effect.race(fetchFromPrimary, fetchFromReplica) // 谁先成功用谁

const program = Effect.gen(function* () {
  const fiber = yield* Effect.fork(longTask)   // 启动子 fiber
  return yield* Fiber.join(fiber)              // 等待汇合
})

const withConn = Effect.acquireRelease(
  openConnection(),                          // acquire
  (conn) => Effect.sync(() => conn.close()), // release（保证执行）
)

const program = Effect.scoped(
  Effect.gen(function* () {
    const conn = yield* withConn
    const tx = yield* beginTransaction(conn)
    yield* tx.commit()
    return "done"
  }),
)

import { Effect, Schedule } from "effect"

const retried = fetchUser("1").pipe(
  Effect.retry(
    Schedule.exponential("100 millis").pipe(Schedule.compose(Schedule.recurs(5))),
  ),
)

const timed = fetchUser("1").pipe(Effect.timeout("2 seconds"))
// 超时后 E 通道加入 TimeoutException

// exponential 指数退避 / recurs(n) 限次 / spaced 固定间隔 / intersect 同时满足
const policy = Schedule.exponential("50 millis").pipe(
  Schedule.intersect(Schedule.recurs(3)),
  Schedule.jittered,        // 加抖动避免惊群
)

import { Effect } from "effect"

// Promise → Effect
const fromPromise = Effect.tryPromise({
  try: () => fetch("/api").then((r) => r.json()),
  catch: (e) => new HttpError(String(e)),
})

// Effect → Promise（在程序边界）
const value = await Effect.runPromise(program)

const program = Effect.gen(function* () {
  const data = yield* Effect.promise(() => someAsyncFn())
  return data
})

// 重构前：隐式依赖、隐式错误、隐式并发
async function getUser(id: string) {
  const conn = await pool.connect()
  try {
    const row = await conn.query("select * from users where id=$1", [id])
    const profile = await fetch(`/profile/${id}`).then((r) => r.json())
    return { ...row, ...profile }
  } finally { conn.release() }
}

// 重构后：依赖/错误/资源全部进类型
const getUser = (id: string): Effect.Effect<Profile, DbError | HttpError, Db | HttpClient> =>
  Effect.gen(function* () {
    const db = yield* Db
    const http = yield* HttpClient
    const row = yield* db.query("select * from users where id=$1", [id])
    const profile = yield* http.getJson(`/profile/${id}`)
    return { ...row, ...profile }
  })

1. 忘记 runPromise：构造了 Effect 却没执行，程序静默无输出
2. 在 Effect.gen 中直接 await：应使用 yield*
3. 在内部层层 runPromise：破坏组合，应只在边界调用
4. 依赖未 provide：编译报 R 通道不满足，别用 as any 绕过

第一步：只在新模块引入 Effect，边界用 runPromise 暴露 Promise API
第二步：错误用 Data.TaggedError 分类，依赖抽成 Context.Tag + Layer
第三步：用 Effect.all / forEach 收敛并发
第四步：把重试、超时、资源管理替换为 Effect 原语