Scala GraphQL 服务实战：Caliban、Schema 设计与订阅

- 单一端点：所有读写走同一路径，靠 operation 区分语义
- 按需取数：一次请求拿到页面所需的全部字段，避免串行往返
- 强类型契约：Schema 即文档，客户端代码可由 SDL 生成
- 代价一：HTTP 层缓存天然失效，需要持久化查询与 GET 化
- 代价二：查询深度与成本由客户端决定，必须自建上界
- 代价三：N+1 从数据库层上升到图遍历层，批处理成为刚需
- 适合多端字段诉求差异大的 BFF，不适合简单 CRUD 与文件下载

query HomePage {
  viewer {
    id
    nickname
    unreadCount
  }
  recommended(first: 10) {
    edges { node { id title coverUrl } }
    pageInfo { hasNextPage endCursor }
  }
}

- 依赖：caliban 核心，加 caliban-zio-http 或 caliban-http4s 适配层
- 派生：case class 与 sealed trait 自动派生，无需手写 SDL
- 注解：GQLDescription 补充文档，GQLName 重命名字段
- 组装：RootResolver 接收 Queries 与 Mutations 与 Subscriptions
- 执行：graphQL 返回 GraphQLInterpreter，可直接 execute 做测试
- 可选：caliban-codegen 从既有 SDL 反向生成 Scala 类型

// build.sbt
libraryDependencies ++= Seq(
  "com.github.ghostdogpr" %% "caliban"          % "2.6.0",
  "com.github.ghostdogpr" %% "caliban-zio-http" % "2.6.0"
)

import caliban._
import caliban.schema.Annotations.GQLDescription
import zio._

final case class User(id: Long, name: String, email: Option[String])

final case class Queries(
  @GQLDescription("按 id 查询单个用户")
  user: Long => ZIO[UserRepo, Throwable, Option[User]],
  users: ZIO[UserRepo, Throwable, List[User]]
)

final case class Mutations(
  createUser: CreateUserInput => ZIO[UserRepo, AppError, User]
)

val api: GraphQL[UserRepo] =
  graphQL(RootResolver(Queries(UserRepo.find, UserRepo.all), Mutations(UserRepo.create)))

val sdl: String = api.render // 导出 SDL 供客户端代码生成与契约比对

- 类型：type 描述对象，字段默认可空，加感叹号表示非空
- 接口：interface 提供共享字段，实现方必须声明 implements
- 联合：union 表示互斥的多态结果，配合 inline fragment 取值
- 枚举：enum 限制取值集合，比 String 更利于客户端穷举
- 输入：input 单独描述写入载荷，不可与输出类型复用
- null 语义：非空字段一旦为 null 会向上冒泡，整棵子树变 null
- 分页：Relay Connection 以 edges 加 pageInfo 表达游标分页

interface Node { id: ID! }

type User implements Node { id: ID! name: String! email: String role: Role! }
type Post implements Node { id: ID! title: String! author: User! }

union SearchResult = User | Post

enum Role { ADMIN MEMBER GUEST }

input PostFilter { keyword: String authorId: ID }

type PageInfo { hasNextPage: Boolean! endCursor: String }
type PostEdge { node: Post! cursor: String! }
type PostConnection { edges: [PostEdge!]! pageInfo: PageInfo! total: Int! }

sealed trait SearchResult
object SearchResult {
  final case class UserHit(user: User) extends SearchResult
  final case class PostHit(post: Post) extends SearchResult
  implicit val schema: Schema[Any, SearchResult] = Schema.unionType[SearchResult]
}

final case class PageInfo(hasNextPage: Boolean, endCursor: Option[String])
final case class PostEdge(node: Post, cursor: String)
final case class PostConnection(edges: List[PostEdge], pageInfo: PageInfo, total: Int)

- 症状：查 100 条 Post 触发 100 次 author 查询
- 根因：每个字段独立解析，效果系统无法自动识别可合并请求
- 方案一：字段返回 ZQuery，配置 DataSource 按 key 批量取
- 方案二：在 service 层手动批量，按 id 集合一次捞回再映射
- 方案三：请求级缓存，同一 key 在一次执行内只取一次
- 注意：批处理边界是「一次执行」，跨请求不共享缓存
- 验证：打开 SQL 日志，观察一次列表查询的实际语句条数

import zio.query._

final case class User(id: Long, name: String)

def fetchUsers(ids: Set[Long]): ZIO[UserRepo, Nothing, Map[Long, User]] =
  UserRepo.findMany(ids).orDie

val userById: DataSource[UserRepo, Long] =
  DataSource.fromFunctionBatchedZIO("UserById")(fetchUsers)

def user(id: Long): ZQuery[UserRepo, Nothing, Option[User]] =
  ZQuery.fromDataSource(id)(userById)

final case class Post(id: Long, title: String, authorId: Long) {
  def author: ZQuery[UserRepo, Nothing, Option[User]] = user(authorId)
}

final case class Queries(posts: ZQuery[UserRepo, Nothing, List[Post]])

- 输入类型：每个变更定义独立的 input，便于独立演进
- 事务位置：放在 service 方法内，不放在解析器里
- 幂等：客户端传入 requestId，服务端做去重表
- 副作用：事件发布在事务提交之后，避免回滚后消息已发出
- 返回类型：变更返回受影响对象本身，便于客户端更新缓存
- 错误：业务失败走类型化错误，基础设施失败走缺陷

final case class CreatePostInput(title: String, body: String, authorId: Long)

final class PostService(repo: PostRepo, events: EventBus) {
  def create(input: CreatePostInput): ZIO[Any, AppError, Post] =
    repo.transact {
      for {
        author <- repo.findUser(input.authorId).someOrFail(AppError.NotFound("author"))
        post   <- repo.insert(Post(0L, input.title, author.id))
      } yield post
    }.tap(post => events.publish(PostCreated(post.id)))
}

final case class Mutations(createPost: CreatePostInput => ZIO[PostService, AppError, Post])

- 协议：graphql-ws 子协议，握手后通过 message 帧交换
- 类型：订阅字段返回 ZStream，而非 ZIO
- 广播：用 Hub 做一对多分发，每个连接订阅同一个流
- 背压：有界 Hub 会阻塞慢消费者，无界 Hub 有 OOM 风险
- 清理：连接断开时 ZStream 被中断，资源随作用域释放
- 过滤：客户端传入参数，在流上做 filter 而非建多个主题
- 心跳：配置 keepAlive 避免中间代理掐断空闲连接

import caliban.schema.Subscription
import zio.stream.ZStream

final case class Subscriptions(postAdded: ZStream[Any, Nothing, Post])

object Broadcast {
  // 生产建议用有界 Hub 加丢弃策略，避免慢消费者拖垮发布者
  val hub: UIO[Hub[Post]] = Hub.bounded[Post](1024, Strategy.DropOldest)
  def publish(post: Post): UIO[Boolean] = hub.flatMap(_.publish(post))
  def stream: ZStream[Any, Nothing, Post] = ZStream.fromHub(hub)
}

val apiWithSubs =
  graphQL(RootResolver(Queries(...), Mutations(...), Subscriptions(Broadcast.stream)))

- 类型化错误：错误类型有 Schema 时自动渲染进 errors 数组
- 缺陷：ZIO 的 Throwable 缺陷不暴露给客户端，只记日志
- 鉴权位置：wrapExecution 包住整个执行，或自定义指令挂到字段
- 指令：用 GQLDirective 声明注解，再在 Schema 包装中拦截
- 错误扩展：在错误类型里带上 code 字段，客户端按 code 分支
- 脱敏：不要把底层异常消息直接透出，防止泄露表结构
- 日志：为每次执行打上 operationName 与 requestId 标注

sealed trait AppError
object AppError {
  final case class NotFound(what: String) extends AppError
  final case class Invalid(reason: String) extends AppError
  final case class Forbidden(reason: String) extends AppError
  implicit val schema: Schema[Any, AppError] = Schema.gen[AppError]
}

// 整个执行包一层鉴权，未通过直接短路
val secured: GraphQLInterpreter[AuthService with UserRepo] =
  api.wrapExecution { (effect, _) => ZIO.serviceWithZIO[AuthService](_.ensure) *> effect }

- zio-http：原生 ZIO，订阅与路由一体，最省心
- http4s：Cats Effect 生态，用 caliban-http4s 适配
- Play：通过 Tapir 端点接入，保留既有中间件
- Federation：caliban-federation 加 GQLKey 注解声明实体
- 网关：Apollo Router 或 GraphQL Mesh 负责子图编排
- 性能：复杂度限制、深度限制、持久化查询三件套
- 观测：ApolloTracing 或 OpenTelemetry 记录每个解析器耗时

import caliban.federation._
import caliban.schema.Annotations.GQLKey

@GQLKey("id")
final case class User(id: Long, name: String)

// 声明本子图能通过 id 解析出的实体，供网关跨子图拼接
val entityResolver: EntityResolver[UserRepo] =
  EntityResolver.from[User](args => UserRepo.find(args("id").toLong))

val federatedApi = federated(graphQL(RootResolver(Queries(...))), entityResolver)

- 观测：按 operationName 与字段路径打点，记录解析器耗时与错误率
- 缓存：响应缓存以查询哈希加变量哈希为键，注意用户隔离
- 测试：SDL 快照、解析器单测、端到端契约测试三层
- 演进：只增不删，字段先标 deprecated，观察调用量归零再移除
- 兼容：给参数加默认值可安全新增，改类型与改可空性均破坏兼容
- 压测：用真实客户端查询集而非手写简单查询，才暴露 N+1
- 灰度：新字段先对内部用户开放，再逐步放量

// 契约测试：把 SDL 快照作为断言对象
object SchemaSpec extends ZIOSpecDefault {
  def spec = suite("schema")(
    test("SDL 关键类型保持稳定") { assertTrue(api.render.contains("type Query")) }
  )
}