fp-ts Do Notation

fp-ts Do 记法指南

Do 记法是 fp-ts 对回调地狱的解决方案。它提供了一种编写顺序执行、看起来像命令式的代码的方式，同时保持函数式纯度和类型安全。

问题：函数式代码中的回调地狱

不使用 Do 记法时，链式调用相互依赖的操作会导致代码层层嵌套：

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

// BAD: Nested chain hell
const processOrder = (orderId: string) =>
  pipe(
    fetchOrder(orderId),
    TE.chain((order) =>
      pipe(
        fetchUser(order.userId),
        TE.chain((user) =>
          pipe(
            fetchInventory(order.productId),
            TE.chain((inventory) =>
              pipe(
                validateStock(inventory, order.quantity),
                TE.chain((validated) =>
                  pipe(
                    calculatePrice(order, user.discount),
                    TE.chain((price) =>
                      createInvoice(order, user, price) // Lost context of inventory!
                    )
                  )
                )
              )
            )
          )
        )
      )
    )
  )

嵌套链式调用的问题：

1. 可读性差 - 逻辑被埋在嵌套结构中
2. 上下文丢失 - 内层作用域可能无法访问之前的值
3. 难以重构 - 添加或删除步骤都需要调整结构
4. 难以并行化 - 所有操作看起来都是顺序执行的

解决方案：Do 记法

Do 记法可以扁平化结构，并让所有值都处于作用域内：

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

// GOOD: Flat, readable Do notation
const processOrder = (orderId: string) =>
  pipe(
    TE.Do,
    TE.bind('order', () => fetchOrder(orderId)),
    TE.bind('user', ({ order }) => fetchUser(order.userId)),
    TE.bind('inventory', ({ order }) => fetchInventory(order.productId)),
    TE.bind('validated', ({ inventory, order }) => validateStock(inventory, order.quantity)),
    TE.bind('price', ({ order, user }) => calculatePrice(order, user.discount)),
    TE.bind('invoice', ({ order, user, price }) => createInvoice(order, user, price))
  )

Do 记法核心函数

Do - 起点

Do 会创建一个包装在 Monad 中的空上下文对象 {}：

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

TE.Do  // TaskEither<never, {}>
E.Do   // Either<never, {}>
O.Do   // Option<{}>

bindTo - 使用首个值初始化

如果已有一个值并希望开始使用 Do 记法，请使用 bindTo：

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

// Instead of:
pipe(
  TE.Do,
  TE.bind('user', () => fetchUser(userId))
)

// Use bindTo for cleaner initialization:
pipe(
  fetchUser(userId),
  TE.bindTo('user'),
  TE.bind('orders', ({ user }) => fetchOrders(user.id))
)

在语义上，bindTo 等同于 TE.map(user => ({ user }))，但可读性更好。

bind - 顺序执行相互依赖的操作

当下一个操作依赖之前的值时，请使用 bind：

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

const getUserWithPosts = (userId: string) =>
  pipe(
    TE.Do,
    TE.bind('user', () => fetchUser(userId)),           // First: get user
    TE.bind('posts', ({ user }) => fetchPosts(user.id)), // Then: use user.id
    TE.bind('comments', ({ posts }) =>                   // Then: use posts
      TE.traverseArray(fetchComments)(posts.map(p => p.id))
    )
  )

bind 的主要特点：

• 操作会按顺序执行
• 每一步都可以访问所有先前的值
• 首次出错时即短路（适用于 Either/TaskEither）
• 回调会接收累积的上下文对象

apS - 并行执行相互独立的操作

当操作相互独立且可以并行执行时，请使用 apS：

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

const getDashboardData = (userId: string) =>
  pipe(
    TE.Do,
    TE.bind('user', () => fetchUser(userId)),
    // These three are INDEPENDENT - use apS for parallel execution
    TE.apS('notifications', fetchNotifications(userId)),
    TE.apS('settings', fetchSettings(userId)),
    TE.apS('recentActivity', fetchRecentActivity(userId))
  )

apS 的主要特点：

• 操作可以并行执行（适用于 TaskEither）
• 值会立即计算（不是惰性计算）
• 无法访问先前的上下文值
• 错误会被收集或导致短路，具体取决于 applicative

let - 计算或派生值

对于基于现有值进行的同步计算，请使用 let：

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

const processPayment = (orderId: string) =>
  pipe(
    TE.Do,
    TE.bind('order', () => fetchOrder(orderId)),
    TE.bind('user', ({ order }) => fetchUser(order.userId)),
    // Computed values - no async operation needed
    TE.let('subtotal', ({ order }) => order.items.reduce((sum, i) => sum + i.price, 0)),
    TE.let('discount', ({ user, subtotal }) => subtotal * (user.discountPercent / 100)),
    TE.let('total', ({ subtotal, discount }) => subtotal - discount),
    TE.bind('payment', ({ total, user }) => chargeCard(user.paymentMethod, total))
  )

let 的主要特点：

• 同步纯计算
• 可以访问所有先前的值
• 不会失败（适用于错误类型）
• 用于转换、计算和格式化

bind 与 apS：如何选择

决策指南

情况	使用	原因
下一个操作需要先前的结果	bind	存在顺序依赖
操作相互独立	apS	可以并行执行
需要转换或计算	let	同步执行，始终成功
从现有值开始	bindTo	比 Do + bind 更简洁

性能：顺序执行与并行执行

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

// SLOW: Sequential execution with bind (3 seconds total)
const slowDashboard = pipe(
  TE.Do,
  TE.bind('users', () => fetchUsers()),      // 1 second
  TE.bind('products', () => fetchProducts()), // 1 second (waits for users)
  TE.bind('orders', () => fetchOrders())      // 1 second (waits for products)
)

// FAST: Parallel execution with apS (1 second total)
const fastDashboard = pipe(
  TE.Do,
  TE.apS('users', fetchUsers()),      // 1 second
  TE.apS('products', fetchProducts()), // 1 second (runs in parallel)
  TE.apS('orders', fetchOrders())      // 1 second (runs in parallel)
)

混合模式：先顺序执行，再并行执行

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

const getOrderDetails = (orderId: string) =>
  pipe(
    TE.Do,
    // Sequential: need order first
    TE.bind('order', () => fetchOrder(orderId)),
    // Parallel: these only need order.userId and order.productId
    TE.apS('user', pipe(
      TE.Do,
      TE.bind('order', () => fetchOrder(orderId)), // Need to refetch or...
    )),
    // Better pattern: bind first, then parallel for truly independent operations
  )

// BETTER: Restructure for clarity
const getOrderDetailsBetter = (orderId: string) =>
  pipe(
    fetchOrder(orderId),
    TE.bindTo('order'),
    TE.bind('user', ({ order }) => fetchUser(order.userId)),
    // Now these are independent of each other (but dependent on user/order)
    TE.apS('shippingOptions', fetchShippingOptions(orderId)),
    TE.apS('paymentMethods', fetchPaymentMethods(orderId)),
    TE.let('canCheckout', ({ user, shippingOptions }) =>
      user.verified && shippingOptions.length > 0
    )
  )

实际示例

示例 1：用户注册流程

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

interface RegistrationInput {
  email: string
  password: string
  name: string
}

interface User {
  id: string
  email: string
  name: string
}

const registerUser = (input: RegistrationInput): TE.TaskEither<Error, User> =>
  pipe(
    TE.Do,
    // Validate input (synchronous)
    TE.bind('validated', () => pipe(
      validateEmail(input.email),
      E.chain(() => validatePassword(input.password)),
      E.map(() => input),
      TE.fromEither
    )),
    // Check if email exists (async)
    TE.bind('emailAvailable', ({ validated }) =>
      checkEmailAvailable(validated.email)
    ),
    // Hash password (async, CPU-intensive)
    TE.bind('hashedPassword', ({ validated }) =>
      hashPassword(validated.password)
    ),
    // Create user in database
    TE.bind('user', ({ validated, hashedPassword }) =>
      createUser({
        email: validated.email,
        name: validated.name,
        passwordHash: hashedPassword
      })
    ),
    // Send welcome email (fire and forget, but still in chain)
    TE.chainFirst(({ user }) => sendWelcomeEmail(user.email)),
    // Return just the user
    TE.map(({ user }) => user)
  )

示例 2：电商结账

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

interface CheckoutResult {
  orderId: string
  paymentId: string
  estimatedDelivery: Date
}

const checkout = (
  userId: string,
  cartId: string,
  shippingAddressId: string
): TE.TaskEither<CheckoutError, CheckoutResult> =>
  pipe(
    TE.Do,
    // Fetch required data in parallel where possible
    TE.bind('cart', () => fetchCart(cartId)),
    TE.bind('user', () => fetchUser(userId)),
    TE.apS('shippingAddress', fetchAddress(shippingAddressId)),

    // Validate cart has items
    TE.bind('validatedCart', ({ cart }) =>
      cart.items.length === 0
        ? TE.left(new CheckoutError('Cart is empty'))
        : TE.right(cart)
    ),

    // Check inventory for all items (parallel)
    TE.bind('inventoryCheck', ({ validatedCart }) =>
      pipe(
        validatedCart.items,
        TE.traverseArray((item) => checkInventory(item.productId, item.quantity))
      )
    ),

    // Calculate totals (synchronous)
    TE.let('subtotal', ({ validatedCart }) =>
      validatedCart.items.reduce((sum, item) => sum + item.price * item.quantity, 0)
    ),
    TE.let('tax', ({ subtotal, shippingAddress }) =>
      calculateTax(subtotal, shippingAddress.state)
    ),
    TE.let('shippingCost', ({ shippingAddress, validatedCart }) =>
      calculateShipping(shippingAddress, validatedCart.totalWeight)
    ),
    TE.let('total', ({ subtotal, tax, shippingCost }) =>
      subtotal + tax + shippingCost
    ),

    // Process payment
    TE.bind('payment', ({ user, total }) =>
      processPayment(user.defaultPaymentMethod, total)
    ),

    // Create order
    TE.bind('order', ({ user, validatedCart, shippingAddress, payment, total }) =>
      createOrder({
        userId: user.id,
        items: validatedCart.items,
        shippingAddressId: shippingAddress.id,
        paymentId: payment.id,
        total
      })
    ),

    // Reserve inventory
    TE.chainFirst(({ order, validatedCart }) =>
      pipe(
        validatedCart.items,
        TE.traverseArray((item) => reserveInventory(item.productId, item.quantity, order.id))
      )
    ),

    // Clear cart
    TE.chainFirst(({ cart }) => clearCart(cart.id)),

    // Calculate delivery estimate
    TE.let('estimatedDelivery', ({ shippingAddress }) =>
      calculateDeliveryDate(shippingAddress)
    ),

    // Return result
    TE.map(({ order, payment, estimatedDelivery }) => ({
      orderId: order.id,
      paymentId: payment.id,
      estimatedDelivery
    }))
  )

示例 3：使用依赖注入的 ReaderTaskEither

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

// Define dependencies
interface Deps {
  userRepo: UserRepository
  orderRepo: OrderRepository
  paymentService: PaymentService
  emailService: EmailService
  logger: Logger
}

// Use RTE.Do for dependency-injected workflows
const processRefund = (
  orderId: string,
  reason: string
): RTE.ReaderTaskEither<Deps, RefundError, RefundResult> =>
  pipe(
    RTE.Do,
    // Access dependencies via RTE.asks
    RTE.bind('deps', () => RTE.ask<Deps>()),

    // Fetch order
    RTE.bind('order', ({ deps }) =>
      RTE.fromTaskEither(deps.orderRepo.findById(orderId))
    ),

    // Validate refund is possible
    RTE.bind('validatedOrder', ({ order }) =>
      order.status !== 'completed'
        ? RTE.left(new RefundError('Order not eligible for refund'))
        : RTE.right(order)
    ),

    // Process refund with payment service
    RTE.bind('refund', ({ deps, validatedOrder }) =>
      RTE.fromTaskEither(
        deps.paymentService.refund(validatedOrder.paymentId, validatedOrder.total)
      )
    ),

    // Update order status
    RTE.bind('updatedOrder', ({ deps, validatedOrder, refund }) =>
      RTE.fromTaskEither(
        deps.orderRepo.update(validatedOrder.id, {
          status: 'refunded',
          refundId: refund.id,
          refundReason: reason
        })
      )
    ),

    // Fetch user for email
    RTE.bind('user', ({ deps, validatedOrder }) =>
      RTE.fromTaskEither(deps.userRepo.findById(validatedOrder.userId))
    ),

    // Send notification (fire and forget)
    RTE.chainFirst(({ deps, user, refund }) =>
      RTE.fromTaskEither(
        deps.emailService.sendRefundConfirmation(user.email, refund)
      )
    ),

    // Log the refund
    RTE.chainFirst(({ deps, order, refund }) =>
      RTE.fromTaskEither(
        deps.logger.info('Refund processed', { orderId: order.id, refundId: refund.id })
      )
    ),

    // Return result
    RTE.map(({ refund, updatedOrder }) => ({
      refundId: refund.id,
      orderId: updatedOrder.id,
      amount: refund.amount,
      status: 'completed'
    }))
  )

// Execute with dependencies
const runRefund = (deps: Deps, orderId: string, reason: string) =>
  processRefund(orderId, reason)(deps)()

示例 4：累积错误的验证

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

// For parallel validation that accumulates ALL errors (not short-circuit)
// Use Apply.sequenceS instead of Do notation

interface ValidationError {
  field: string
  message: string
}

type ValidationResult<A> = E.Either<ValidationError[], A>

const validateUserInput = (input: unknown): ValidationResult<ValidUser> => {
  const validateField = <A>(
    field: string,
    value: unknown,
    validator: (v: unknown) => E.Either<string, A>
  ): ValidationResult<A> =>
    pipe(
      validator(value),
      E.mapLeft((message) => [{ field, message }])
    )

  // Use sequenceS with validation applicative to collect ALL errors
  return pipe(
    sequenceS(E.getApplicativeValidation(A.getSemigroup<ValidationError>()))({
      email: validateField('email', input.email, validateEmail),
      password: validateField('password', input.password, validatePassword),
      age: validateField('age', input.age, validateAge),
      name: validateField('name', input.name, validateName)
    })
  )
}

// For TaskEither with parallel execution AND error accumulation:
const validateUserAsync = (input: UserInput): TE.TaskEither<ValidationError[], ValidUser> =>
  pipe(
    sequenceS(TE.ApplicativePar)({
      emailUnique: checkEmailUnique(input.email),
      usernameAvailable: checkUsernameAvailable(input.username),
      phoneValid: validatePhoneNumber(input.phone)
    }),
    TE.map(({ emailUnique, usernameAvailable, phoneValid }) => ({
      ...input,
      emailVerified: emailUnique,
      usernameVerified: usernameAvailable,
      phoneVerified: phoneValid
    }))
  )

示例 5：复杂数据聚合

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

interface DashboardData {
  user: User
  stats: UserStats
  recentOrders: Order[]
  recommendations: Product[]
  notifications: Notification[]
}

const loadDashboard = (userId: string): TE.TaskEither<Error, DashboardData> =>
  pipe(
    TE.Do,
    // First, get user (required for everything)
    TE.bind('user', () => fetchUser(userId)),

    // These are all independent - parallel execution
    TE.apS('stats', fetchUserStats(userId)),
    TE.apS('recentOrders', fetchRecentOrders(userId)),
    TE.apS('notifications', fetchNotifications(userId)),

    // Recommendations depend on user preferences
    TE.bind('recommendations', ({ user }) =>
      fetchRecommendations(user.preferences)
    ),

    // Enhance orders with product details (depends on recentOrders)
    TE.bind('ordersWithProducts', ({ recentOrders }) =>
      pipe(
        recentOrders,
        A.map((order) =>
          pipe(
            fetchProductDetails(order.productId),
            TE.map((product) => ({ ...order, product }))
          )
        ),
        TE.sequenceArray
      )
    ),

    // Compute derived data
    TE.let('unreadCount', ({ notifications }) =>
      notifications.filter((n) => !n.read).length
    ),
    TE.let('totalSpent', ({ recentOrders }) =>
      recentOrders.reduce((sum, o) => sum + o.total, 0)
    ),

    // Return final shape
    TE.map(({ user, stats, ordersWithProducts, recommendations, notifications }) => ({
      user,
      stats,
      recentOrders: ordersWithProducts,
      recommendations,
      notifications
    }))
  )

常见模式和技巧

模式 1：提前返回 / 守卫子句

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

const deleteAccount = (userId: string, confirmationCode: string) =>
  pipe(
    TE.Do,
    TE.bind('user', () => fetchUser(userId)),
    // Guard: Check confirmation code
    TE.bind('confirmed', ({ user }) =>
      confirmationCode === user.deleteConfirmationCode
        ? TE.right(true)
        : TE.left(new Error('Invalid confirmation code'))
    ),
    // Guard: Check no pending orders
    TE.bind('pendingOrders', ({ user }) => fetchPendingOrders(user.id)),
    TE.bind('canDelete', ({ pendingOrders }) =>
      pendingOrders.length === 0
        ? TE.right(true)
        : TE.left(new Error('Cannot delete account with pending orders'))
    ),
    // Proceed with deletion
    TE.bind('deleted', ({ user }) => deleteUserAccount(user.id))
  )

模式 2：使用 chainFirst 执行可选操作

当你想执行副作用但保留原始值时，请使用 chainFirst：

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

const createPost = (input: PostInput) =>
  pipe(
    TE.Do,
    TE.bind('post', () => savePost(input)),
    // Log creation (side effect, ignore result)
    TE.chainFirst(({ post }) => logPostCreation(post.id)),
    // Notify followers (side effect, ignore result)
    TE.chainFirst(({ post }) => notifyFollowers(post.authorId, post.id)),
    // Index for search (side effect, ignore result)
    TE.chainFirst(({ post }) => indexForSearch(post)),
    // Return just the post
    TE.map(({ post }) => post)
  )

模式 3：条件绑定

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

const processOrder = (orderId: string, promoCode?: string) =>
  pipe(
    TE.Do,
    TE.bind('order', () => fetchOrder(orderId)),
    // Conditionally apply promo code
    TE.bind('discount', ({ order }) =>
      promoCode
        ? validatePromoCode(promoCode, order.total)
        : TE.right(0)
    ),
    TE.let('finalTotal', ({ order, discount }) => order.total - discount)
  )

模式 4：在 Do 中处理数组

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

const processOrders = (orderIds: string[]) =>
  pipe(
    TE.Do,
    // Fetch all orders in parallel
    TE.bind('orders', () =>
      pipe(
        orderIds,
        A.map(fetchOrder),
        TE.sequenceArray
      )
    ),
    // Process each order
    TE.bind('processed', ({ orders }) =>
      pipe(
        orders,
        A.map(processOrder),
        TE.sequenceArray
      )
    ),
    // Aggregate results
    TE.let('summary', ({ processed }) => ({
      total: processed.length,
      successful: processed.filter((p) => p.status === 'success').length,
      failed: processed.filter((p) => p.status === 'failed').length
    }))
  )

性能注意事项

1. 对相互独立的操作优先使用 apS

// SLOW: 3 sequential API calls
pipe(
  TE.Do,
  TE.bind('a', () => fetchA()), // 100ms
  TE.bind('b', () => fetchB()), // 100ms
  TE.bind('c', () => fetchC())  // 100ms
) // Total: 300ms

// FAST: 3 parallel API calls
pipe(
  TE.Do,
  TE.apS('a', fetchA()), // 100ms
  TE.apS('b', fetchB()), // 100ms (parallel)
  TE.apS('c', fetchC())  // 100ms (parallel)
) // Total: ~100ms

2. 对纯计算使用 let

// WRONG: Using bind for pure computation
TE.bind('total', ({ items }) => TE.right(items.reduce((s, i) => s + i.price, 0)))

// RIGHT: Using let for pure computation
TE.let('total', ({ items }) => items.reduce((s, i) => s + i.price, 0))

3. 批量执行数据库操作

// SLOW: N+1 queries
pipe(
  TE.Do,
  TE.bind('orders', () => fetchOrders(userId)),
  TE.bind('products', ({ orders }) =>
    pipe(
      orders,
      A.map((o) => fetchProduct(o.productId)), // N queries!
      TE.sequenceArray
    )
  )
)

// FAST: Batch query
pipe(
  TE.Do,
  TE.bind('orders', () => fetchOrders(userId)),
  TE.bind('products', ({ orders }) =>
    fetchProductsByIds(orders.map((o) => o.productId)) // 1 query
  )
)

4. 避免重建上下文

// INEFFICIENT: Rebuilding large context
pipe(
  TE.Do,
  TE.bind('hugeData', () => fetchHugeData()),
  TE.map(({ hugeData }) => ({ hugeData, processed: true })), // Copies hugeData
  TE.bind('more', () => fetchMore()) // hugeData still in context
)

// BETTER: Extract what you need early
pipe(
  TE.Do,
  TE.bind('hugeData', () => fetchHugeData()),
  TE.let('summary', ({ hugeData }) => summarize(hugeData)), // Extract summary
  TE.map(({ summary }) => summary) // Drop hugeData from context
)

总结

Do 记法可将深层嵌套的回调链转换为扁平、易读的流水线：

函数	用途	使用时机
Do	开始一个空上下文	链的开头
bindTo	从一个值开始	已有初始值时
bind	顺序操作	依赖之前的值时
apS	并行操作	与其他值相互独立时
let	纯计算	同步派生值时
chainFirst	副作用	执行后不等待结果的操作

核心原则：

1. 依赖其他值时使用 bind，相互独立时使用 apS
2. 纯计算使用 let，不要使用 bind 搭配 TE.right
3. 保持上下文精简，不要累积不必要的数据
4. 处理集合时搭配 sequenceArray/traverseArray
5. 对不应影响结果的副作用使用 chainFirst

Do 记法是编写易维护 fp-ts 代码的关键。掌握它，就能让函数式编程像命令式代码一样易读，同时保留其全部优势。