/SKILL.md

name, description, risk, source, version, author, tags

name	description	risk	source	version	author	tags
fp-backend	使用 fp-ts、ReaderTaskEither 和函数式依赖注入的 Node.js/Deno 后端开发函数式编程模式	unknown	community	1.0.0	kadu	fp-ts typescript backend functional-programming node deno dependency-injection reader-task-either

fp-ts 后端模式

使用 fp-ts 构建类型安全、可测试的后端服务的函数式编程模式。

何时使用

• 你正在使用 fp-ts 构建或重构 Node.js 或 Deno 后端。
• 任务涉及依赖注入、服务组合，或使用 ReaderTaskEither 进行类型化后端错误处理。
• 你需要的是函数式后端架构模式，而非零散的工具代码片段。

核心概念

ReaderTaskEither（RTE）

ReaderTaskEither<R, E, A> 类型是函数式后端开发的骨架：

• R（Reader）：依赖/环境（数据库、配置、日志器）
• E（Either left）：错误类型
• A（Either right）：成功值

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

// 定义你的依赖
type Deps = {
  db: DatabaseClient
  logger: Logger
  config: Config
}

// 定义领域错误
type AppError =
  | { _tag: 'NotFound'; resource: string; id: string }
  | { _tag: 'ValidationError'; message: string }
  | { _tag: 'DatabaseError'; cause: unknown }
  | { _tag: 'Unauthorized'; reason: string }

// 一个服务函数
const getUser = (id: string): RTE.ReaderTaskEither<Deps, AppError, User> =>
  pipe(
    RTE.ask<Deps>(),
    RTE.flatMap(({ db, logger }) =>
      pipe(
        RTE.fromTaskEither(db.users.findById(id)),
        RTE.mapLeft((e): AppError => ({ _tag: 'DatabaseError', cause: e })),
        RTE.flatMap(user =>
          user
            ? RTE.right(user)
            : RTE.left({ _tag: 'NotFound', resource: 'User', id })
        ),
        RTE.tap(user => RTE.fromIO(() => logger.info(`Found user: ${user.id}`)))
      )
    )
  )

服务层模式

定义服务模块

将服务组织为导出 RTE 函数的模块：

// src/services/user.service.ts
import * as RTE from 'fp-ts/ReaderTaskEither'
import * as TE from 'fp-ts/TaskEither'
import * as A from 'fp-ts/Array'
import { pipe } from 'fp-ts/function'

type UserDeps = {
  db: DatabaseClient
  hasher: PasswordHasher
  mailer: EmailService
}

type UserError =
  | { _tag: 'UserNotFound'; id: string }
  | { _tag: 'EmailExists'; email: string }
  | { _tag: 'InvalidPassword' }

// 创建用户
export const create = (
  input: CreateUserInput
): RTE.ReaderTaskEither<UserDeps, UserError, User> =>
  pipe(
    RTE.ask<UserDeps>(),
    RTE.flatMap(({ db, hasher }) =>
      pipe(
        // 检查邮箱唯一性
        checkEmailUnique(input.email),
        RTE.flatMap(() =>
          RTE.fromTaskEither(hasher.hash(input.password))
        ),
        RTE.flatMap(hashedPassword =>
          RTE.fromTaskEither(
            db.users.create({
              ...input,
              password: hashedPassword,
            })
          )
        )
      )
    )
  )

// 按 ID 查找
export const findById = (
  id: string
): RTE.ReaderTaskEither<UserDeps, UserError, User> =>
  pipe(
    RTE.ask<UserDeps>(),
    RTE.flatMap(({ db }) =>
      pipe(
        RTE.fromTaskEither(db.users.findUnique({ where: { id } })),
        RTE.flatMap(user =>
          user
            ? RTE.right(user)
            : RTE.left({ _tag: 'UserNotFound' as const, id })
        )
      )
    )
  )

// 批量查找（带分页）
export const findMany = (
  params: PaginationParams
): RTE.ReaderTaskEither<UserDeps, UserError, PaginatedResult<User>> =>
  pipe(
    RTE.ask<UserDeps>(),
    RTE.flatMap(({ db }) =>
      RTE.fromTaskEither(
        pipe(
          TE.Do,
          TE.bind('users', () => db.users.findMany({
            skip: params.offset,
            take: params.limit,
          })),
          TE.bind('total', () => db.users.count()),
          TE.map(({ users, total }) => ({
            data: users,
            total,
            ...params,
          }))
        )
      )
    )
  )

const checkEmailUnique = (
  email: string
): RTE.ReaderTaskEither<UserDeps, UserError, void> =>
  pipe(
    RTE.ask<UserDeps>(),
    RTE.flatMap(({ db }) =>
      pipe(
        RTE.fromTaskEither(db.users.findUnique({ where: { email } })),
        RTE.flatMap(existing =>
          existing
            ? RTE.left({ _tag: 'EmailExists' as const, email })
            : RTE.right(undefined)
        )
      )
    )
  )

组合服务

// src/services/order.service.ts
import * as UserService from './user.service'
import * as ProductService from './product.service'
import * as PaymentService from './payment.service'

type OrderDeps = UserService.UserDeps &
  ProductService.ProductDeps &
  PaymentService.PaymentDeps & {
    db: DatabaseClient
  }

export const createOrder = (
  userId: string,
  items: OrderItem[]
): RTE.ReaderTaskEither<OrderDeps, OrderError, Order> =>
  pipe(
    RTE.Do,
    // 验证用户存在
    RTE.bind('user', () =>
      pipe(
        UserService.findById(userId),
        RTE.mapLeft(toOrderError)
      )
    ),
    // 验证并获取产品
    RTE.bind('products', () =>
      pipe(
        items,
        A.traverse(RTE.ApplicativePar)(item =>
          ProductService.findById(item.productId)
        ),
        RTE.mapLeft(toOrderError)
      )
    ),
    // 计算总额
    RTE.bind('total', ({ products }) =>
      RTE.right(calculateTotal(products, items))
    ),
    // 处理支付
    RTE.bind('payment', ({ user, total }) =>
      pipe(
        PaymentService.charge(user, total),
        RTE.mapLeft(toOrderError)
      )
    ),
    // 创建订单
    RTE.flatMap(({ user, products, total, payment }) =>
      createOrderRecord(user, products, items, total, payment)
    )
  )

函数式依赖注入

构建依赖容器

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

// 第 0 层：配置（无依赖）
type Config = {
  database: { url: string; poolSize: number }
  redis: { url: string }
  jwt: { secret: string; expiresIn: string }
}

const loadConfig = (): TE.TaskEither<Error, Config> =>
  TE.tryCatch(
    async () => ({
      database: {
        url: process.env.DATABASE_URL!,
        poolSize: parseInt(process.env.DB_POOL_SIZE || '10'),
      },
      redis: { url: process.env.REDIS_URL! },
      jwt: {
        secret: process.env.JWT_SECRET!,
        expiresIn: process.env.JWT_EXPIRES || '1d',
      },
    }),
    (e) => new Error(`Config error: ${e}`)
  )

// 第 1 层：基础设施（依赖配置）
type Infrastructure = {
  config: Config
  db: PrismaClient
  redis: RedisClient
  logger: Logger
}

const buildInfrastructure = (
  config: Config
): TE.TaskEither<Error, Infrastructure> =>
  pipe(
    TE.Do,
    TE.bind('db', () =>
      TE.tryCatch(
        async () => {
          const prisma = new PrismaClient({
            datasources: { db: { url: config.database.url } },
          })
          await prisma.$connect()
          return prisma
        },
        (e) => new Error(`Database error: ${e}`)
      )
    ),
    TE.bind('redis', () =>
      TE.tryCatch(
        async () => createRedisClient(config.redis.url),
        (e) => new Error(`Redis error: ${e}`)
      )
    ),
    TE.bind('logger', () => TE.right(createLogger())),
    TE.map(({ db, redis, logger }) => ({
      config,
      db,
      redis,
      logger,
    }))
  )

// 第 2 层：服务（依赖基础设施）
type Services = {
  hasher: PasswordHasher
  jwt: JwtService
  mailer: EmailService
}

const buildServices = (infra: Infrastructure): Services => ({
  hasher: createBcryptHasher(),
  jwt: createJwtService(infra.config.jwt),
  mailer: createEmailService(infra.config),
})

// 完整的应用依赖
export type AppDeps = Infrastructure & Services

export const buildDeps = (): TE.TaskEither<Error, AppDeps> =>
  pipe(
    loadConfig(),
    TE.flatMap(buildInfrastructure),
    TE.map(infra => ({
      ...infra,
      ...buildServices(infra),
    }))
  )

// 清理
export const destroyDeps = (deps: AppDeps): TE.TaskEither<Error, void> =>
  pipe(
    TE.tryCatch(
      async () => {
        await deps.db.$disconnect()
        await deps.redis.quit()
      },
      (e) => new Error(`Cleanup error: ${e}`)
    )
  )

使用依赖运行程序

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

const program: RTE.ReaderTaskEither<AppDeps, AppError, void> = pipe(
  RTE.ask<AppDeps>(),
  RTE.flatMap(deps =>
    pipe(
      startServer(deps),
      RTE.fromTaskEither
    )
  )
)

const main = async () => {
  const result = await pipe(
    buildDeps(),
    TE.mapLeft((e): AppError => ({ _tag: 'StartupError', cause: e })),
    TE.flatMap(deps =>
      pipe(
        program(deps),
        TE.tap(() => TE.fromIO(() => console.log('Server running'))),
        // 退出时清理
        TE.tapError(() => destroyDeps(deps))
      )
    )
  )()

  if (result._tag === 'Left') {
    console.error('Failed to start:', result.left)
    process.exit(1)
  }
}

main()

数据库操作

Prisma 封装

// src/lib/db.ts
import * as TE from 'fp-ts/TaskEither'
import * as O from 'fp-ts/Option'
import { PrismaClient, Prisma } from '@prisma/client'

type DbError =
  | { _tag: 'RecordNotFound'; model: string; id: string }
  | { _tag: 'UniqueViolation'; field: string }
  | { _tag: 'ForeignKeyViolation'; field: string }
  | { _tag: 'UnknownDbError'; cause: unknown }

// 封装 Prisma 操作
const wrapPrisma = <A>(
  operation: () => Promise<A>
): TE.TaskEither<DbError, A> =>
  TE.tryCatch(operation, (error): DbError => {
    if (error instanceof Prisma.PrismaClientKnownRequestError) {
      switch (error.code) {
        case 'P2002':
          return {
            _tag: 'UniqueViolation',
            field: (error.meta?.target as string[])?.join(', ') || 'unknown',
          }
        case 'P2003':
          return {
            _tag: 'ForeignKeyViolation',
            field: error.meta?.field_name as string || 'unknown',
          }
        case 'P2025':
          return {
            _tag: 'RecordNotFound',
            model: error.meta?.modelName as string || 'unknown',
            id: 'unknown',
          }
      }
    }
    return { _tag: 'UnknownDbError', cause: error }
  })

// 仓库工厂
export const createRepository = <
  Model,
  CreateInput,
  UpdateInput,
  WhereUnique,
  WhereMany
>(
  db: PrismaClient,
  delegate: {
    findUnique: (args: { where: WhereUnique }) => Promise<Model | null>
    findMany: (args: { where?: WhereMany; skip?: number; take?: number }) => Promise<Model[]>
    create: (args: { data: CreateInput }) => Promise<Model>
    update: (args: { where: WhereUnique; data: UpdateInput }) => Promise<Model>
    delete: (args: { where: WhereUnique }) => Promise<Model>
    count: (args?: { where?: WhereMany }) => Promise<number>
  }
) => ({
  findUnique: (where: WhereUnique): TE.TaskEither<DbError, O.Option<Model>> =>
    pipe(
      wrapPrisma(() => delegate.findUnique({ where })),
      TE.map(O.fromNullable)
    ),

  findMany: (
    where?: WhereMany,
    pagination?: { skip: number; take: number }
  ): TE.TaskEither<DbError, Model[]> =>
    wrapPrisma(() => delegate.findMany({ where, ...pagination })),

  create: (data: CreateInput): TE.TaskEither<DbError, Model> =>
    wrapPrisma(() => delegate.create({ data })),

  update: (
    where: WhereUnique,
    data: UpdateInput
  ): TE.TaskEither<DbError, Model> =>
    wrapPrisma(() => delegate.update({ where, data })),

  delete: (where: WhereUnique): TE.TaskEither<DbError, Model> =>
    wrapPrisma(() => delegate.delete({ where })),

  count: (where?: WhereMany): TE.TaskEither<DbError, number> =>
    wrapPrisma(() => delegate.count({ where })),
})

// 使用示例
const userRepo = createRepository(prisma, prisma.user)

事务处理

// src/lib/transaction.ts
import * as TE from 'fp-ts/TaskEither'
import * as RTE from 'fp-ts/ReaderTaskEither'
import { PrismaClient } from '@prisma/client'
import { pipe } from 'fp-ts/function'

type TxClient = Omit<
  PrismaClient,
  '$connect' | '$disconnect' | '$on' | '$transaction' | '$use'
>

type TxDeps = { tx: TxClient }

// 事务包装器
export const withTransaction = <R extends { db: PrismaClient }, E, A>(
  program: RTE.ReaderTaskEither<R & TxDeps, E, A>
): RTE.ReaderTaskEither<R, E | DbError, A> =>
  pipe(
    RTE.ask<R>(),
    RTE.flatMap(deps =>
      RTE.fromTaskEither(
        TE.tryCatch(
          () =>
            deps.db.$transaction(async tx => {
              const result = await program({ ...deps, tx })()
              if (result._tag === 'Left') {
                throw result.left // 回滚
              }
              return result.right
            }),
          (error): E | DbError => {
            // 重新抛出领域错误
            if (typeof error === 'object' && error !== null && '_tag' in error) {
              return error as E
            }
            return { _tag: 'UnknownDbError', cause: error }
          }
        )
      )
    )
  )

// 在服务中使用
export const transferFunds = (
  fromId: string,
  toId: string,
  amount: number
): RTE.ReaderTaskEither<AppDeps, TransferError, Transfer> =>
  withTransaction(
    pipe(
      RTE.Do,
      RTE.bind('from', () => debitAccount(fromId, amount)),
      RTE.bind('to', () => creditAccount(toId, amount)),
      RTE.bind('transfer', ({ from, to }) =>
        createTransferRecord(from, to, amount)
      ),
      RTE.map(({ transfer }) => transfer)
    )
  )

// 在事务内部，使用 tx 代替 db
const debitAccount = (
  accountId: string,
  amount: number
): RTE.ReaderTaskEither<TxDeps, TransferError, Account> =>
  pipe(
    RTE.ask<TxDeps>(),
    RTE.flatMap(({ tx }) =>
      RTE.fromTaskEither(
        pipe(
          TE.tryCatch(
            () =>
              tx.account.update({
                where: { id: accountId },
                data: { balance: { decrement: amount } },
              }),
            toDbError
          ),
          TE.flatMap(account =>
            account.balance < 0
              ? TE.left({ _tag: 'InsufficientFunds' as const, accountId })
              : TE.right(account)
          )
        )
      )
    )
  )

中间件模式

Express 中间件

// src/middleware/fp-express.ts
import { Request, Response, NextFunction, RequestHandler } from 'express'
import * as TE from 'fp-ts/TaskEither'
import * as RTE from 'fp-ts/ReaderTaskEither'
import * as E from 'fp-ts/Either'
import { pipe } from 'fp-ts/function'

// 将 RTE handler 转换为 Express 中间件
export const toHandler =
  <R, E, A>(
    getDeps: (req: Request) => R,
    handler: (req: Request) => RTE.ReaderTaskEither<R, E, A>,
    onError: (error: E, res: Response) => void
  ): RequestHandler =>
  async (req, res, next) => {
    const deps = getDeps(req)
    const result = await handler(req)(deps)()

    pipe(
      result,
      E.fold(
        error => onError(error, res),
        data => res.json(data)
      )
    )
  }

// 错误处理器
const handleError = (error: AppError, res: Response): void => {
  switch (error._tag) {
    case 'NotFound':
      res.status(404).json({ error: error.resource + ' not found' })
      break
    case 'ValidationError':
      res.status(400).json({ error: error.message })
      break
    case 'Unauthorized':
      res.status(401).json({ error: error.reason })
      break
    default:
      res.status(500).json({ error: 'Internal server error' })
  }
}

// 使用示例
const getUserHandler = toHandler(
  req => req.app.locals.deps as AppDeps,
  req => UserService.findById(req.params.id),
  handleError
)

app.get('/users/:id', getUserHandler)

Hono 中间件

// src/middleware/fp-hono.ts
import { Hono, Context, MiddlewareHandler } from 'hono'
import * as RTE from 'fp-ts/ReaderTaskEither'
import * as E from 'fp-ts/Either'
import { pipe } from 'fp-ts/function'

// 在上下文中存储依赖
declare module 'hono' {
  interface ContextVariableMap {
    deps: AppDeps
  }
}

// 依赖注入中间件
export const withDeps = (deps: AppDeps): MiddlewareHandler =>
  async (c, next) => {
    c.set('deps', deps)
    await next()
  }

// 将 RTE 转换为 Hono handler
export const toHonoHandler =
  <E, A>(
    handler: (c: Context) => RTE.ReaderTaskEither<AppDeps, E, A>,
    onError: (error: E, c: Context) => Response
  ) =>
  async (c: Context): Promise<Response> => {
    const deps = c.get('deps')
    const result = await handler(c)(deps)()

    return pipe(
      result,
      E.fold(
        error => onError(error, c),
        data => c.json(data)
      )
    )
  }

// 校验中间件
export const validate =
  <T>(schema: z.ZodSchema<T>): MiddlewareHandler =>
  async (c, next) => {
    const body = await c.req.json()
    const result = schema.safeParse(body)

    if (!result.success) {
      return c.json(
        { error: 'Validation failed', details: result.error.flatten() },
        400
      )
    }

    c.set('validatedBody', result.data)
    await next()
  }

// 使用 RTE 的认证中间件
export const requireAuth: MiddlewareHandler = async (c, next) => {
  const deps = c.get('deps')
  const token = c.req.header('Authorization')?.replace('Bearer ', '')

  if (!token) {
    return c.json({ error: 'No token provided' }, 401)
  }

  const result = await pipe(
    deps.jwt.verify(token),
    TE.mapLeft(() => ({ _tag: 'Unauthorized' as const, reason: 'Invalid token' }))
  )()

  if (E.isLeft(result)) {
    return c.json({ error: result.left.reason }, 401)
  }

  c.set('user', result.right)
  await next()
}

// 使用示例
const app = new Hono()

app.use('*', withDeps(deps))
app.use('/api/*', requireAuth)

app.get(
  '/api/users/:id',
  toHonoHandler(
    c => UserService.findById(c.req.param('id')),
    (error, c) => {
      if (error._tag === 'UserNotFound') {
        return c.json({ error: 'User not found' }, 404)
      }
      return c.json({ error: 'Internal error' }, 500)
    }
  )
)

请求上下文模式

// src/context.ts
import * as RTE from 'fp-ts/ReaderTaskEither'
import { pipe } from 'fp-ts/function'

// 请求范围的上下文
type RequestContext = {
  requestId: string
  userId: O.Option<string>
  startTime: number
}

type ContextDeps = AppDeps & { ctx: RequestContext }

// 带上下文的日志记录
const logWithContext =
  (level: 'info' | 'warn' | 'error') =>
  (message: string, meta?: object): RTE.ReaderTaskEither<ContextDeps, never, void> =>
    pipe(
      RTE.ask<ContextDeps>(),
      RTE.flatMap(({ logger, ctx }) =>
        RTE.fromIO(() =>
          logger.log(level, message, {
            requestId: ctx.requestId,
            userId: O.toUndefined(ctx.userId),
            ...meta,
            elapsed: Date.now() - ctx.startTime,
          })
        )
      )
    )

export const log = {
  info: logWithContext('info'),
  warn: logWithContext('warn'),
  error: logWithContext('error'),
}

// 创建上下文的中间件
export const withContext: MiddlewareHandler = async (c, next) => {
  const deps = c.get('deps')
  const ctx: RequestContext = {
    requestId: crypto.randomUUID(),
    userId: O.fromNullable(c.get('user')?.id),
    startTime: Date.now(),
  }

  c.set('deps', { ...deps, ctx })

  // 记录请求开始
  deps.logger.info('Request started', {
    requestId: ctx.requestId,
    method: c.req.method,
    path: c.req.path,
  })

  await next()

  // 记录请求结束
  deps.logger.info('Request completed', {
    requestId: ctx.requestId,
    status: c.res.status,
    elapsed: Date.now() - ctx.startTime,
  })
}

错误处理模式

类型化错误层次

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

// 基础错误类型
type DomainError =
  | NotFoundError
  | ValidationError
  | ConflictError
  | AuthError
  | InfrastructureError

type NotFoundError = {
  _tag: 'NotFoundError'
  resource: string
  id: string
}

type ValidationError = {
  _tag: 'ValidationError'
  field: string
  message: string
  value?: unknown
}

type ConflictError = {
  _tag: 'ConflictError'
  resource: string
  field: string
  value: string
}

type AuthError =
  | { _tag: 'Unauthenticated' }
  | { _tag: 'Unauthorized'; required: string }
  | { _tag: 'TokenExpired' }

type InfrastructureError = {
  _tag: 'InfrastructureError'
  service: string
  cause: unknown
}

// 智能构造函数
export const notFound = (resource: string, id: string): NotFoundError => ({
  _tag: 'NotFoundError',
  resource,
  id,
})

export const validation = (
  field: string,
  message: string,
  value?: unknown
): ValidationError => ({
  _tag: 'ValidationError',
  field,
  message,
  value,
})

export const conflict = (
  resource: string,
  field: string,
  value: string
): ConflictError => ({
  _tag: 'ConflictError',
  resource,
  field,
  value,
})

// 错误到 HTTP 状态码映射
export const toHttpStatus = (error: DomainError): number => {
  switch (error._tag) {
    case 'NotFoundError':
      return 404
    case 'ValidationError':
      return 400
    case 'ConflictError':
      return 409
    case 'Unauthenticated':
      return 401
    case 'Unauthorized':
      return 403
    case 'TokenExpired':
      return 401
    case 'InfrastructureError':
      return 503
    default:
      return 500
  }
}

// 错误到响应体
export const toResponseBody = (
  error: DomainError
): { error: string; details?: unknown } => {
  switch (error._tag) {
    case 'NotFoundError':
      return { error: `${error.resource} not found` }
    case 'ValidationError':
      return {
        error: 'Validation failed',
        details: { field: error.field, message: error.message },
      }
    case 'ConflictError':
      return {
        error: `${error.resource} with ${error.field} already exists`,
      }
    case 'Unauthenticated':
      return { error: 'Authentication required' }
    case 'Unauthorized':
      return { error: `Permission denied: ${error.required}` }
    case 'TokenExpired':
      return { error: 'Token expired' }
    case 'InfrastructureError':
      return { error: 'Service temporarily unavailable' }
  }
}

错误恢复

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

// 带指数退避的重试
export const withRetry =
  <R, E, A>(
    maxAttempts: number,
    baseDelayMs: number,
    shouldRetry: (error: E) => boolean
  ) =>
  (
    operation: RTE.ReaderTaskEither<R, E, A>
  ): RTE.ReaderTaskEither<R, E, A> =>
    pipe(
      RTE.ask<R>(),
      RTE.flatMap(deps => {
        const attempt = (
          remaining: number,
          delay: number
        ): TE.TaskEither<E, A> =>
          pipe(
            operation(deps),
            TE.orElse(error => {
              if (remaining <= 0 || !shouldRetry(error)) {
                return TE.left(error)
              }
              return pipe(
                TE.fromTask(() => new Promise(r => setTimeout(r, delay))),
                TE.flatMap(() => attempt(remaining - 1, delay * 2))
              )
            })
          )

        return RTE.fromTaskEither(attempt(maxAttempts - 1, baseDelayMs))
      })
    )

// 回退到缓存值
export const withFallback =
  <R extends { cache: CacheClient }, E, A>(
    cacheKey: string,
    ttlSeconds: number
  ) =>
  (
    operation: RTE.ReaderTaskEither<R, E, A>
  ): RTE.ReaderTaskEither<R, E, A> =>
    pipe(
      RTE.ask<R>(),
      RTE.flatMap(({ cache, ...rest }) =>
        pipe(
          operation,
          // 成功后缓存结果
          RTE.tap(result =>
            RTE.fromTaskEither(cache.set(cacheKey, result, ttlSeconds))
          ),
          // 失败后尝试获取缓存值
          RTE.orElse(error =>
            pipe(
              RTE.fromTaskEither(cache.get<A>(cacheKey)),
              RTE.flatMap(cached =>
                cached ? RTE.right(cached) : RTE.left(error)
              )
            )
          )
        )
      )
    )

// 断路器
type CircuitState = 'closed' | 'open' | 'half-open'

export const createCircuitBreaker = <E>(
  failureThreshold: number,
  resetTimeoutMs: number,
  isFailure: (error: E) => boolean
) => {
  let state: CircuitState = 'closed'
  let failures = 0
  let lastFailure = 0

  return <R, A>(
    operation: RTE.ReaderTaskEither<R, E, A>
  ): RTE.ReaderTaskEither<R, E | { _tag: 'CircuitOpen' }, A> =>
    pipe(
      RTE.ask<R>(),
      RTE.flatMap(deps => {
        // 检查断路器是否应复位
        if (
          state === 'open' &&
          Date.now() - lastFailure > resetTimeoutMs
        ) {
          state = 'half-open'
        }

        if (state === 'open') {
          return RTE.left({ _tag: 'CircuitOpen' as const })
        }

        return pipe(
          operation,
          RTE.tap(() => {
            if (state === 'half-open') {
              state = 'closed'
              failures = 0
            }
            return RTE.right(undefined)
          }),
          RTE.tapError(error => {
            if (isFailure(error)) {
              failures++
              lastFailure = Date.now()
              if (failures >= failureThreshold) {
                state = 'open'
              }
            }
            return RTE.right(undefined)
          })
        )
      })
    )
}

测试策略

模拟依赖

// src/services/__tests__/user.service.test.ts
import * as TE from 'fp-ts/TaskEither'
import * as E from 'fp-ts/Either'
import * as O from 'fp-ts/Option'
import { describe, it, expect, vi } from 'vitest'
import * as UserService from '../user.service'

// 创建模拟依赖
const createMockDeps = (overrides: Partial<UserDeps> = {}): UserDeps => ({
  db: {
    users: {
      findUnique: vi.fn(() => Promise.resolve(null)),
      create: vi.fn(data => Promise.resolve({ id: '1', ...data })),
      update: vi.fn((where, data) => Promise.resolve({ id: where.id, ...data })),
    },
  },
  hasher: {
    hash: vi.fn(password => TE.right(`hashed_${password}`)),
    verify: vi.fn(() => TE.right(true)),
  },
  mailer: {
    send: vi.fn(() => TE.right(undefined)),
  },
  ...overrides,
})

describe('UserService', () => {
  describe('create', () => {
    it('should create a user with hashed password', async () => {
      const deps = createMockDeps()
      const input = {
        email: 'test@example.com',
        password: 'secret123',
        name: 'Test User',
      }

      const result = await UserService.create(input)(deps)()

      expect(E.isRight(result)).toBe(true)
      if (E.isRight(result)) {
        expect(result.right.email).toBe(input.email)
      }
      expect(deps.hasher.hash).toHaveBeenCalledWith('secret123')
    })

    it('should fail when email already exists', async () => {
      const existingUser = { id: '1', email: 'test@example.com' }
      const deps = createMockDeps({
        db: {
          users: {
            findUnique: vi.fn(() => Promise.resolve(existingUser)),
            create: vi.fn(),
          },
        },
      })

      const result = await UserService.create({
        email: 'test@example.com',
        password: 'secret',
        name: 'Test',
      })(deps)()

      expect(E.isLeft(result)).toBe(true)
      if (E.isLeft(result)) {
        expect(result.left._tag).toBe('EmailExists')
      }
    })
  })

  describe('findById', () => {
    it('should return user when found', async () => {
      const user = { id: '1', email: 'test@example.com', name: 'Test' }
      const deps = createMockDeps({
        db: {
          users: {
            findUnique: vi.fn(() => Promise.resolve(user)),
          },
        },
      })

      const result = await UserService.findById('1')(deps)()

      expect(E.isRight(result)).toBe(true)
      if (E.isRight(result)) {
        expect(result.right).toEqual(user)
      }
    })

    it('should return NotFound when user does not exist', async () => {
      const deps = createMockDeps()

      const result = await UserService.findById('nonexistent')(deps)()

      expect(E.isLeft(result)).toBe(true)
      if (E.isLeft(result)) {
        expect(result.left._tag).toBe('UserNotFound')
        expect(result.left.id).toBe('nonexistent')
      }
    })
  })
})

使用测试容器的集成测试

// src/__tests__/integration/user.integration.test.ts
import { PostgreSqlContainer } from '@testcontainers/postgresql'
import { PrismaClient } from '@prisma/client'
import * as TE from 'fp-ts/TaskEither'
import * as E from 'fp-ts/Either'
import { pipe } from 'fp-ts/function'
import { describe, it, expect, beforeAll, afterAll } from 'vitest'
import { buildDeps, destroyDeps, AppDeps } from '../../deps'
import * as UserService from '../../services/user.service'

describe('UserService Integration', () => {
  let container: PostgreSqlContainer
  let deps: AppDeps

  beforeAll(async () => {
    // 启动 PostgreSQL 容器
    container = await new PostgreSqlContainer().start()

    // 使用测试数据库构建真实依赖
    process.env.DATABASE_URL = container.getConnectionUri()

    const depsResult = await buildDeps()()
    if (E.isLeft(depsResult)) {
      throw new Error(`Failed to build deps: ${depsResult.left}`)
    }
    deps = depsResult.right

    // 运行迁移
    await deps.db.$executeRaw`CREATE EXTENSION IF NOT EXISTS "uuid-ossp"`
    // ... 运行 Prisma 迁移
  }, 60000)

  afterAll(async () => {
    await destroyDeps(deps)()
    await container.stop()
  })

  it('should create and retrieve a user', async () => {
    // 创建用户
    const createResult = await UserService.create({
      email: 'integration@test.com',
      password: 'password123',
      name: 'Integration Test',
    })(deps)()

    expect(E.isRight(createResult)).toBe(true)
    if (E.isLeft(createResult)) return

    const user = createResult.right

    // 检索用户
    const findResult = await UserService.findById(user.id)(deps)()

    expect(E.isRight(findResult)).toBe(true)
    if (E.isRight(findResult)) {
      expect(findResult.right.email).toBe('integration@test.com')
    }
  })
})

基于属性的测试

// src/__tests__/property/user.property.test.ts
import * as fc from 'fast-check'
import * as E from 'fp-ts/Either'
import { describe, it, expect } from 'vitest'
import { validateEmail, validatePassword } from '../../validation'

describe('Validation Properties', () => {
  it('valid emails should pass validation', () => {
    fc.assert(
      fc.property(fc.emailAddress(), email => {
        const result = validateEmail(email)
        return E.isRight(result)
      })
    )
  })

  it('passwords meeting requirements should pass', () => {
    const validPassword = fc
      .tuple(
        fc.stringOf(fc.constantFrom(...'abcdefghijklmnopqrstuvwxyz'), {
          minLength: 4,
        }),
        fc.stringOf(fc.constantFrom(...'ABCDEFGHIJKLMNOPQRSTUVWXYZ'), {
          minLength: 1,
        }),
        fc.stringOf(fc.constantFrom(...'0123456789'), { minLength: 1 }),
        fc.stringOf(fc.constantFrom(...'!@#$%^&*'), { minLength: 1 })
      )
      .map(parts => parts.join(''))

    fc.assert(
      fc.property(validPassword, password => {
        const result = validatePassword(password)
        return E.isRight(result)
      })
    )
  })

  it('empty strings should fail email validation', () => {
    const result = validateEmail('')
    expect(E.isLeft(result)).toBe(true)
  })
})

快速参考

常用导入

import * as RTE from 'fp-ts/ReaderTaskEither'
import * as TE from 'fp-ts/TaskEither'
import * as E from 'fp-ts/Either'
import * as O from 'fp-ts/Option'
import * as A from 'fp-ts/Array'
import * as T from 'fp-ts/Task'
import { pipe, flow } from 'fp-ts/function'

RTE 速查表

操作	描述
RTE.right(a)	将值提升为成功
RTE.left(e)	创建错误
RTE.ask<R>()	获取依赖
RTE.fromTaskEither(te)	提升 TaskEither
RTE.fromEither(e)	提升 Either
RTE.fromOption(onNone)(o)	提升 Option
RTE.flatMap(f)	链式操作
RTE.map(f)	转换成功值
RTE.mapLeft(f)	转换错误
RTE.tap(f)	成功时的副作用
RTE.tapError(f)	错误时的副作用
RTE.orElse(f)	从错误中恢复
RTE.getOrElse(f)	提取值并回退

服务模板

// 新服务的模板
import * as RTE from 'fp-ts/ReaderTaskEither'
import { pipe } from 'fp-ts/function'

type MyServiceDeps = {
  db: DatabaseClient
  // ... 其他依赖
}

type MyServiceError =
  | { _tag: 'NotFound'; id: string }
  | { _tag: 'ValidationFailed'; reason: string }

export const myOperation = (
  input: Input
): RTE.ReaderTaskEither<MyServiceDeps, MyServiceError, Output> =>
  pipe(
    RTE.ask<MyServiceDeps>(),
    RTE.flatMap(deps =>
      // 在此处实现你的逻辑
      RTE.right(output)
    )
  )

限制

• 仅当任务明确符合上述范围时使用此技能。
• 请勿将输出视为环境特定验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请停下并请求澄清。

@质检员-陈晨

翻译说明：

1. 术语一致性：全文保持统一的术语翻译，如"dependency injection"译为"依赖注入"，"functional programming"译为"函数式编程"，"ReaderTaskEither"保留英文原名（技术库名）。
2. 技术标识符保护：所有代码块内的内容（TypeScript 代码、类型定义、函数名、变量名、路径等）全部保持原样，未做任何翻译。
3. YAML frontmatter：键名保持英文（name, description, risk 等），description 等散文部分译为中文，技术性字段如 version、author 等保持原样。
4. 表格翻译：RTE 速查表中的操作名（如 RTE.right(a)）保持原样，对应的描述列译为中文。
5. 专有名词处理：fp-ts、Node.js、Deno、Prisma、Express、Hono、PostgreSQL 等技术名词保留原文，不音译。
6. 注释翻译：代码块内的行内注释译为中文，便于中文读者理解。
7. 修复原文 bug：原文中有 loggerlevel 一处明显拼写错误（应为 logger.log(level, ...)），翻译时已按正确的 logger.log(level, message, ...) 形式呈现，与代码语义一致。