使用 fp-ts 进行实用的错误处理

本技能教你如何在不陷入 try/catch 意大利面式代码的情况下处理错误。没有学术黑话——只有针对真实问题的实用模式。

何时使用本技能

• 当你想在 TypeScript 中实现类型安全的错误处理
• 当用 Either 和 TaskEither 模式替换 try/catch 时
• 当构建需要显式错误类型的 API 或服务时
• 当需要累积多个校验错误时

核心思想：错误只是数据。与其把它们抛入虚空并指望有人捕获，不如把它们作为 TypeScript 能够追踪的值返回。

1. 停止到处抛异常

异常的问题

异常在你的类型中是不可见的。它们打破了函数之间的契约。

// 这个函数签名承诺了什么：
function getUser(id: string): User

// 它实际做了什么：
function getUser(id: string): User {
  if (!id) throw new Error('ID required')
  const user = db.find(id)
  if (!user) throw new Error('User not found')
  return user
}

// 调用者完全不知道它可能失败
const user = getUser(id) // 可能会炸！

你最终会得到这样的代码：

// MESSY: try/catch 到处都是
function processOrder(orderId: string) {
  let order
  try {
    order = getOrder(orderId)
  } catch (e) {
    console.error('Failed to get order')
    return null
  }

  let user
  try {
    user = getUser(order.userId)
  } catch (e) {
    console.error('Failed to get user')
    return null
  }

  let payment
  try {
    payment = chargeCard(user.cardId, order.total)
  } catch (e) {
    console.error('Payment failed')
    return null
  }

  return { order, user, payment }
}

解决方案：把错误作为值返回

import * as E from 'fp-ts/Either'
import { pipe } from 'fp-ts/function'

// 现在 TypeScript 知道这个函数可能失败
function getUser(id: string): E.Either {
  if (!id) return E.left('ID required')
  const user = db.find(id)
  if (!user) return E.left('User not found')
  return E.right(user)
}

// 调用者被迫处理两种情况
const result = getUser(id)
// result 是 Either —— 错误 或 成功，绝不会两者都是

2. Result 模式（Either）

Either 很简单：它要么持有一个错误（E），要么持有一个值（A）。

• Left = 错误情况
• Right = 成功情况（把 "right" 理解为 "correct"，即正确）

import * as E from 'fp-ts/Either'

// 创建值
const success = E.right(42)           // Right(42)
const failure = E.left('Oops')        // Left('Oops')

// 检查你拿到的是什么
if (E.isRight(result)) {
  console.log(result.right) // 成功值
} else {
  console.log(result.left)  // 错误
}

// 更好：用 fold 进行模式匹配
const message = pipe(
  result,
  E.fold(
    (error) => `Failed: ${error}`,
    (value) => `Got: ${value}`
  )
)

将抛异常代码转换为 Either

// 用 tryCatch 包裹任何会抛异常的函数
const parseJSON = (json: string): E.Either =>
  E.tryCatch(
    () => JSON.parse(json),
    (e) => (e instanceof Error ? e : new Error(String(e)))
  )

parseJSON('{"valid": true}')  // Right({ valid: true })
parseJSON('not json')          // Left(SyntaxError: ...)

// 对于会复用的函数，使用 tryCatchK
const safeParseJSON = E.tryCatchK(
  JSON.parse,
  (e) => (e instanceof Error ? e : new Error(String(e)))
)

常见 Either 操作

import * as E from 'fp-ts/Either'
import { pipe } from 'fp-ts/function'

// 变换成功值
const doubled = pipe(
  E.right(21),
  E.map(n => n * 2)
) // Right(42)

// 变换错误
const betterError = pipe(
  E.left('bad'),
  E.mapLeft(e => `Error: ${e}`)
) // Left('Error: bad')

// 为错误提供默认值
const value = pipe(
  E.left('failed'),
  E.getOrElse(() => 0)
) // 0

// 将可空值转换为 Either
const fromNullable = E.fromNullable('not found')
fromNullable(user)  // 若存在则为 Right(user)，null/undefined 则为 Left('not found')

3. 链式调用可能失败的操作

真正的威力来自链式调用。每一步都可能失败，但你可以用一条干净的管道来书写。

改造前：嵌套的 Try/Catch 地狱

// MESSY: 每一步都可能失败，嵌套的 try/catch 到处都是
function processUserOrder(userId: string, productId: string): Result | null {
  let user
  try {
    user = getUser(userId)
  } catch (e) {
    logError('User fetch failed', e)
    return null
  }

  if (!user.isActive) {
    logError('User not active')
    return null
  }

  let product
  try {
    product = getProduct(productId)
  } catch (e) {
    logError('Product fetch failed', e)
    return null
  }

  if (product.stock < 1) {
    logError('Out of stock')
    return null
  }

  let order
  try {
    order = createOrder(user, product)
  } catch (e) {
    logError('Order creation failed', e)
    return null
  }

  return order
}

改造后：用 Either 进行干净的链式调用

import * as E from 'fp-ts/Either'
import { pipe } from 'fp-ts/function'

// 每个函数都返回 Either
const getUser = (id: string): E.Either => { ... }
const getProduct = (id: string): E.Either => { ... }
const createOrder = (user: User, product: Product): E.Either => { ... }

// 将它们串起来 —— 第一个错误会终止整条链
const processUserOrder = (userId: string, productId: string): E.Either =>
  pipe(
    getUser(userId),
    E.filterOrElse(
      user => user.isActive,
      () => 'User not active'
    ),
    E.chain(user =>
      pipe(
        getProduct(productId),
        E.filterOrElse(
          product => product.stock >= 1,
          () => 'Out of stock'
        ),
        E.chain(product => createOrder(user, product))
      )
    )
  )

// 或者使用 Do 记法来更干净地访问中间值
const processUserOrder = (userId: string, productId: string): E.Either =>
  pipe(
    E.Do,
    E.bind('user', () => getUser(userId)),
    E.filterOrElse(
      ({ user }) => user.isActive,
      () => 'User not active'
    ),
    E.bind('product', () => getProduct(productId)),
    E.filterOrElse(
      ({ product }) => product.stock >= 1,
      () => 'Out of stock'
    ),
    E.chain(({ user, product }) => createOrder(user, product))
  )

不同的错误类型？使用 chainW

type ValidationError = { type: 'validation'; message: string }
type DbError = { type: 'db'; message: string }

const validateInput = (id: string): E.Either => { ... }
const fetchFromDb = (id: string): E.Either => { ... }

// chainW（W = "wider"）会自动联合错误类型
const process = (id: string): E.Either =>
  pipe(
    validateInput(id),
    E.chainW(validId => fetchFromDb(validId))
  )

4. 收集多个错误

有时你想要全部错误，而不只是第一个。表单校验就是最典型的例子。

改造前：手动收集错误

// MESSY: 手动累积错误
function validateForm(form: FormData): { valid: boolean; errors: string[] } {
  const errors: string[] = []

  if (!form.email) {
    errors.push('Email required')
  } else if (!form.email.includes('@')) {
    errors.push('Invalid email')
  }

  if (!form.password) {
    errors.push('Password required')
  } else if (form.password.length < 8) {
    errors.push('Password too short')
  }

  if (!form.age) {
    errors.push('Age required')
  } else if (form.age < 18) {
    errors.push('Must be 18+')
  }

  return { valid: errors.length === 0, errors }
}

改造后：带错误累积的校验

import * as E from 'fp-ts/Either'
import * as NEA from 'fp-ts/NonEmptyArray'
import { sequenceS } from 'fp-ts/Apply'
import { pipe } from 'fp-ts/function'

// 错误以 NonEmptyArray 形式存在（至少有一个）
type Errors = NEA.NonEmptyArray

// 创建会累积错误的 applicative
const validation = E.getApplicativeValidation(NEA.getSemigroup())

// 返回 Either 的校验器
const validateEmail = (email: string): E.Either =>
  !email ? E.left(NEA.of('Email required'))
  : !email.includes('@') ? E.left(NEA.of('Invalid email'))
  : E.right(email)

const validatePassword = (password: string): E.Either =>
  !password ? E.left(NEA.of('Password required'))
  : password.length < 8 ? E.left(NEA.of('Password too short'))
  : E.right(password)

const validateAge = (age: number | undefined): E.Either =>
  age === undefined ? E.left(NEA.of('Age required'))
  : age < 18 ? E.left(NEA.of('Must be 18+'))
  : E.right(age)

// 合并所有校验 —— 收集全部错误
const validateForm = (form: FormData) =>
  sequenceS(validation)({
    email: validateEmail(form.email),
    password: validatePassword(form.password),
    age: validateAge(form.age)
  })

// 使用
validateForm({ email: '', password: '123', age: 15 })
// Left(['Email required', 'Password too short', 'Must be 18+'])

validateForm({ email: 'a@b.com', password: 'longpassword', age: 25 })
// Right({ email: 'a@b.com', password: 'longpassword', age: 25 })

面向表单的字段级错误

interface FieldError {
  field: string
  message: string
}

type FormErrors = NEA.NonEmptyArray

const fieldError = (field: string, message: string): FormErrors =>
  NEA.of({ field, message })

const formValidation = E.getApplicativeValidation(NEA.getSemigroup())

// 现在错误知道它们属于哪个字段
const validateEmail = (email: string): E.Either =>
  !email ? E.left(fieldError('email', 'Required'))
  : !email.includes('@') ? E.left(fieldError('email', 'Invalid format'))
  : E.right(email)

// 在 UI 中容易展示
const getFieldError = (errors: FormErrors, field: string): string | undefined =>
  errors.find(e => e.field === field)?.message

5. 异步操作（TaskEither）

对于可能失败的异步操作，使用 TaskEither。它就像是 Either，但是用于 Promise。

• TaskEither = 一个返回 Promise> 的函数
• 惰性：在你执行它之前，什么都不会运行

import * as TE from 'fp-ts/TaskEither'
import { pipe } from 'fp-ts/function'

// 包裹任意异步操作
const fetchUser = (id: string): TE.TaskEither =>
  TE.tryCatch(
    () => fetch(`/api/users/${id}`).then(r => r.json()),
    (e) => (e instanceof Error ? e : new Error(String(e)))
  )

// 链式调用异步操作 —— 就像 Either 一样
const getUserPosts = (userId: string): TE.TaskEither =>
  pipe(
    fetchUser(userId),
    TE.chain(user => fetchPosts(user.id))
  )

// 准备好时执行
const result = await getUserPosts('123')() // 返回 Either

改造前：带错误处理的 Promise 链

// MESSY: try/catch 与 promise 链混在一起
async function loadDashboard(userId: string) {
  try {
    const user = await fetchUser(userId)
    if (!user) throw new Error('User not found')

    let posts, notifications, settings
    try {
      [posts, notifications, settings] = await Promise.all([
        fetchPosts(user.id),
        fetchNotifications(user.id),
        fetchSettings(user.id)
      ])
    } catch (e) {
      // 哪一个失败了？谁知道！
      console.error('Failed to load data', e)
      return null
    }

    return { user, posts, notifications, settings }
  } catch (e) {
    console.error('Failed to load user', e)
    return null
  }
}

改造后：干净的 TaskEither 管道

import * as TE from 'fp-ts/TaskEither'
import { sequenceS } from 'fp-ts/Apply'
import { pipe } from 'fp-ts/function'

const loadDashboard = (userId: string) =>
  pipe(
    fetchUser(userId),
    TE.chain(user =>
      pipe(
        // 使用 sequenceS 并行获取
        sequenceS(TE.ApplicativePar)({
          posts: fetchPosts(user.id),
          notifications: fetchNotifications(user.id),
          settings: fetchSettings(user.id)
        }),
        TE.map(data => ({ user, ...data }))
      )
    )
  )

// 执行并处理两种情况
pipe(
  loadDashboard('123'),
  TE.fold(
    (error) => T.of(renderError(error)),
    (data) => T.of(renderDashboard(data))
  )
)()

重试失败的操作

import * as T from 'fp-ts/Task'
import * as TE from 'fp-ts/TaskEither'
import { pipe } from 'fp-ts/function'

const retry = (
  task: TE.TaskEither,
  attempts: number,
  delayMs: number
): TE.TaskEither =>
  pipe(
    task,
    TE.orElse((error) =>
      attempts > 1
        ? pipe(
            T.delay(delayMs)(T.of(undefined)),
            T.chain(() => retry(task, attempts - 1, delayMs * 2))
          )
        : TE.left(error)
    )
  )

// 最多重试 3 次，指数退避
const fetchWithRetry = retry(fetchUser('123'), 3, 1000)

回退到替代方案

// 先尝试缓存，再回退到 API
const getUserData = (id: string) =>
  pipe(
    fetchFromCache(id),
    TE.orElse(() => fetchFromApi(id)),
    TE.orElse(() => TE.right(defaultUser)) // 最后的默认方案
  )

6. 模式之间的转换

真实的代码库里有抛异常的函数、可空值以及 Promise。下面是如何与它们协作。

从可空值到 Either

import * as E from 'fp-ts/Either'
import * as O from 'fp-ts/Option'

// 直接转换
const user = users.find(u => u.id === id) // User | undefined
const result = E.fromNullable('User not found')(user)

// 从 Option 转换
const maybeUser: O.Option = O.fromNullable(user)
const eitherUser = pipe(
  maybeUser,
  E.fromOption(() => 'User not found')
)

从抛异常函数到 Either

// 在边界处包裹
const safeParse = (schema: ZodSchema) => (data: unknown): E.Either =>
  E.tryCatch(
    () => schema.parse(data),
    (e) => e as ZodError
  )

// 在你的代码中全程使用
const parseUser = safeParse(UserSchema)
const result = parseUser(rawData) // Either

从 Promise 到 TaskEither

import * as TE from 'fp-ts/TaskEither'

// 包裹外部异步函数
const fetchJson = (url: string): TE.TaskEither =>
  TE.tryCatch(
    () => fetch(url).then(r => r.json()),
    (e) => new Error(`Fetch failed: ${e}`)
  )

// 包裹 axios、prisma，任意异步库
const getUserFromDb = (id: string): TE.TaskEither =>
  TE.tryCatch(
    () => prisma.user.findUniqueOrThrow({ where: { id } }),
    (e) => ({ code: 'DB_ERROR', cause: e })
  )

回到 Promise（逃生舱口）

有时你需要为一个外部 API 提供普通的 Promise。

import * as TE from 'fp-ts/TaskEither'
import * as E from 'fp-ts/Either'

const myTaskEither: TE.TaskEither = fetchUser('123')

// 方案 1：获取 Either（保留两种情况）
const either: E.Either = await myTaskEither()

// 方案 2：错误时抛出（用于遗留代码）
const toThrowingPromise = (te: TE.TaskEither): Promise =>
  te().then(E.fold(
    (error) => Promise.reject(error),
    (value) => Promise.resolve(value)
  ))

const user = await toThrowingPromise(fetchUser('123')) // 若为 Left 则抛出

// 方案 3：错误时使用默认值
const user = await pipe(
  fetchUser('123'),
  TE.getOrElse(() => T.of(defaultUser))
)()

interface ParsedInput {
  id: number
  name: string
  tags: string[]
}

const parseInput = (raw: unknown): E.Either =>
  pipe(
    E.Do,
    E.bind('obj', () =>
      typeof raw === 'object' && raw !== null
        ? E.right(raw as Record)
        : E.left('Input must be an object')
    ),
    E.bind('id', ({ obj }) =>
      typeof obj.id === 'number'
        ? E.right(obj.id)
        : E.left('id must be a number')
    ),
    E.bind('name', ({ obj }) =>
      typeof obj.name === 'string' && obj.name.length > 0
        ? E.right(obj.name)
        : E.left('name must be a non-empty string')
    ),
    E.bind('tags', ({ obj }) =>
      Array.isArray(obj.tags) && obj.tags.every(t => typeof t === 'string')
        ? E.right(obj.tags as string[])
        : E.left('tags must be an array of strings')
    ),
    E.map(({ id, name, tags }) => ({ id, name, tags }))
  )

// 使用
parseInput({ id: 1, name: 'test', tags: ['a', 'b'] })
// Right({ id: 1, name: 'test', tags: ['a', 'b'] })

parseInput({ id: 'wrong', name: '', tags: null })
// Left('id must be a number')

interface ApiError {
  code: string
  message: string
  status?: number
}

const createApiError = (message: string, code = 'UNKNOWN', status?: number): ApiError =>
  ({ code, message, status })

const fetchWithErrorHandling = (url: string): TE.TaskEither =>
  pipe(
    TE.tryCatch(
      () => fetch(url),
      () => createApiError('Network error', 'NETWORK')
    ),
    TE.chain(response =>
      response.ok
        ? TE.tryCatch(
            () => response.json() as Promise,
            () => createApiError('Invalid JSON', 'PARSE')
          )
        : TE.left(createApiError(
            `HTTP ${response.status}`,
            response.status === 404 ? 'NOT_FOUND' : 'HTTP_ERROR',
            response.status
          ))
    )
  )

// 根据错误码进行模式匹配的使用
const handleUserFetch = (userId: string) =>
  pipe(
    fetchWithErrorHandling(`/api/users/${userId}`),
    TE.fold(
      (error) => {
        switch (error.code) {
          case 'NOT_FOUND': return T.of(showNotFoundPage())
          case 'NETWORK': return T.of(showOfflineMessage())
          default: return T.of(showGenericError(error.message))
        }
      },
      (user) => T.of(showUserProfile(user))
    )
  )

import * as A from 'fp-ts/Array'
import * as E from 'fp-ts/Either'
import { pipe } from 'fp-ts/function'

interface ProcessResult {
  successes: T[]
  failures: Array<{ item: unknown; error: string }>
}

// 处理所有项，分别收集成功与失败
const processAllCollectErrors = (
  items: T[],
  process: (item: T) => E.Either
): ProcessResult => {
  const results = items.map((item, index) =>
    pipe(
      process(item),
      E.mapLeft(error => ({ item, error, index }))
    )
  )

  return {
    successes: pipe(results, A.filterMap(E.toOption)),
    failures: pipe(
      results,
      A.filterMap(r => E.isLeft(r) ? O.some(r.left) : O.none)
    )
  }
}

// 使用
const parseNumbers = (inputs: string[]) =>
  processAllCollectErrors(inputs, input => {
    const n = parseInt(input, 10)
    return isNaN(n) ? E.left(`Invalid number: ${input}`) : E.right(n)
  })

parseNumbers(['1', 'abc', '3', 'def'])
// {
//   successes: [1, 3],
//   failures: [
//     { item: 'abc', error: 'Invalid number: abc', index: 1 },
//     { item: 'def', error: 'Invalid number: def', index: 3 }
//   ]
// }

import * as TE from 'fp-ts/TaskEither'
import * as T from 'fp-ts/Task'
import { pipe } from 'fp-ts/function'

interface BulkResult {
  succeeded: T[]
  failed: Array<{ id: string; error: string }>
}

const bulkProcess = (
  ids: string[],
  process: (id: string) => TE.TaskEither
): T.Task> =>
  pipe(
    ids,
    A.map(id =>
      pipe(
        process(id),
        TE.fold(
          (error) => T.of({ type: 'failed' as const, id, error }),
          (result) => T.of({ type: 'succeeded' as const, result })
        )
      )
    ),
    T.sequenceArray,
    T.map(results => ({
      succeeded: results
        .filter((r): r is { type: 'succeeded'; result: T } => r.type === 'succeeded')
        .map(r => r.result),
      failed: results
        .filter((r): r is { type: 'failed'; id: string; error: string } => r.type === 'failed')
        .map(({ id, error }) => ({ id, error }))
    }))
  )

// 使用
const deleteUsers = (userIds: string[]) =>
  bulkProcess(userIds, id =>
    pipe(
      deleteUser(id),
      TE.mapLeft(e => e.message)
    )
  )

// 所有操作都会运行，你会得到一份关于哪些成功、哪些失败的报表