@ atproto/ lex- server

Atproto Lexicon 协议和计划请求路由器. 见 更改日志 用于版本历史.

npm install @atproto/lex-server

• 基于 Lexicon 方案的安全请求路由
• 为查询、程序和WebSocket订阅提供支持
• 建于网络标准 Request编号 :Response API(跨跑道便携式)
• 自定义认证及证书通过
• 优雅的关闭 AsyncDisposable 模式

[大

此软件包目前处于 预览. API 和特性在稳定发布前会有所改变.

这是什么?

构建AT协议服务器需要处理XRPC请求,对照Lexicon schemas验证输入,管理认证,并支持实时订阅. @atproto/lex-server 以 :

1. 基于 Lexicon 计程器对类型安全处理器的运行请求
2. 自动验证请求参数和正文
3. 提供具有定制战略的灵活认证系统
4. 支持使用后压处理的 WebSocket 订阅

import { LexRouter } from '@atproto/lex-server'
import { serve, upgradeWebSocket } from '@atproto/lex-server/nodejs'
import * as app from './lexicons/app.js'

const router = new LexRouter({ upgradeWebSocket })
  .add(app.bsky.actor.getProfile, async ({ params }) => {
    const profile = await db.getProfile(params.actor)
    return { body: profile }
  })
  .add(app.bsky.feed.post.create, {
    auth: requireAuth,
    handler: async ({ credentials, input }) => {
      const result = await db.createPost(credentials.did, input.body)
      return { body: result }
    },
  })

await serve(router, { port: 3000 })

• 快速启动
• 莱克萨斯路特
  • 创建路由器
  • 添加路由
  • 处理器上下文
  • 处理器输出
• 查询和程序
  • 查询处理器
  • 程序处理器
  • 二进制有效载荷
• 订阅
• 认证
  • 自定义认证
  • WWW- 认证信头
• 处理错误
  • LexServer 错误
  • LexServer Auth 错误
  • 处理器召回错误
• 节点.js 服务器
  • 服务( )
  • 创建 Server ()
  • 请求收听器 ()
  • 升级WebSocket ()
• 高级使用
  • 自定义响应对象
  • 答复信头
  • 连接信息
• 许可证

快速启动

1. 安装软件包

npm install @atproto/lex-server

2. 产生Lexicon计划

使用 @atproto/lex 从 Lexicon 定义生成 TypeScript 计划:

lex install app.bsky.actor.getProfile
lex build

3. 创建路由器并添加处理器

import { LexRouter, LexServerError } from '@atproto/lex-server'
import { serve, upgradeWebSocket } from '@atproto/lex-server/nodejs'
import * as app from './lexicons/app.js'

const router = new LexRouter({ upgradeWebSocket })

// Add a query handler
router.add(app.bsky.actor.getProfile, async ({ params }) => {
  const profile = await db.getProfile(params.actor)
  if (!profile) {
    throw new LexServerError(404, {
      error: 'NotFound',
      message: 'Profile not found',
    })
  }
  return { body: profile }
})

// Start the server
const server = await serve(router, { port: 3000 })
console.log('Server listening on port 3000')

莱克萨斯路特

这个 LexRouter 类是核心 @atproto/lex-server它向基于Lexicon schemas的打字安全处理器传递XRPC请求.

创建路由器

import { LexRouter } from '@atproto/lex-server'
import { upgradeWebSocket } from '@atproto/lex-server/nodejs'

const router = new LexRouter({
  // Required for WebSocket subscriptions (Node.js)
  upgradeWebSocket,

  // Optional: Add logging or error reporting for unexpected errors in handlers
  onHandlerError: ({ error, request, method }) => {
    console.error(`Error in ${method.nsid}:`, error)
  },

  // Optional: WebSocket backpressure settings
  highWaterMark: 250_000, // bytes (default: 250KB)
  lowWaterMark: 50_000, // bytes (default: 50KB)
})

添加路由

使用 .add() 方法,它接受Lexicon的图案和处理器:

// Simple handler (no authentication)
router.add(schema, async ({ params, input, request }) => {
  return { body: result }
})

// Handler with authentication
router.add(schema, {
  auth: async ({ request, params }) => credentials,
  handler: async ({ params, input, credentials, request }) => {
    return { body: result }
  },
})

路由器支持方法链:

const router = new LexRouter()
  .add(app.bsky.actor.getProfile, profileHandler)
  .add(app.bsky.feed.getTimeline, timelineHandler)
  .add(app.bsky.feed.post.create, postHandler)

处理器上下文

处理器接收具有以下属性的上下文对象:

type LexRouterHandlerContext<Method, Credentials> = {
  credentials: Credentials // Result of auth function (undefined if no auth)
  input: InferMethodInput<Method> // Parsed request body (procedures only)
  params: InferMethodParams<Method> // Parsed URL query parameters
  request: Request // Original Web Request object
  connection?: ConnectionInfo // Network connection info
}

处理器输出

处理器可以返回各种输出格式 :

// JSON response (encoding inferred from schema)
return { body: { key: 'value' } }

// With custom encoding
return { encoding: 'text/plain', body: 'Hello, world!' }

// With response headers
return { body: data, headers: { 'x-custom': 'value' } }

// Empty response (200 OK with no body)
return {}

// Custom Response object (full control)
return new Response(body, { status: 201, headers })

// Proxy Response
return fetch('https://example.com/data')

查询和程序

查询处理器

查询手柄 GET 请求和接收 URL 查询字符串中的参数 :

import * as app from './lexicons/app.js'

router.add(app.bsky.actor.getProfile, async ({ params }) => {
  // params.actor is typed and validated
  const profile = await db.getProfile(params.actor)
  return { body: profile }
})

程序处理器

程序处理 POST 请求和接收请求机构:

router.add(app.bsky.feed.post.create, async ({ input }) => {
  // input.body contains the parsed and validated request body
  const post = await db.createPost(input.body)
  return { body: { uri: post.uri, cid: post.cid } }
})

二进制有效载荷

对于接受或返回二进制数据的终点:

// Binary input
router.add(app.example.uploadBlob, async ({ input }) => {
  // input.body is a Request object for streaming
  // input.encoding contains the content-type
  const blob = await input.body.arrayBuffer()
  return { body: { cid: await store(blob) } }
})

// Binary output
router.add(app.example.getBlob, async ({ params }) => {
  const stream = await getBlob(params.cid)
  return {
    encoding: 'application/octet-stream',
    body: stream,
  }
})

订阅

订阅通过WebSocket连接提供实时数据. 处理器是生成消息的同步生成器:

import { LexRouter, LexError } from '@atproto/lex-server'
import { serve, upgradeWebSocket } from '@atproto/lex-server/nodejs'
import { scheduler } from 'node:timers/promises'

const router = new LexRouter({
  upgradeWebSocket, // Required for WebSocket support in nodejs
})

router.add(com.example.stream, async function* ({ params, request }) {
  const { cursor = 0, limit = 10 } = params
  const { signal } = request

  for (let i = 0; i < limit; i++) {
    // Yield messages to the client
    yield com.example.stream.message.$build({
      data: `Message ${cursor + i}`,
      cursor: cursor + i,
    })

    // Wait between messages (respects abort signal)
    await scheduler.wait(1000, { signal })
  }

  // Throwing a LexError closes the connection with an error frame
  throw new LexError('LimitReached', `Limit of ${limit} messages reached`)
})

信件为 CBOR- 编码后作为 WebSocket 二进制框发送 。 路由器处理:

• WebSocket 升级谈判
• 后压管理
• 优雅的连接清理
• 错误框架编码

认证

自定义认证

认证通过 auth 函数在处理器配置中:

import { LexError, LexServerAuthError } from '@atproto/lex-server'

type Credentials = { did: string; scope: string[] }

const requireAuth = async ({
  request,
}: {
  request: Request
}): Promise<Credentials> => {
  const header = request.headers.get('authorization')
  if (!header?.startsWith('Bearer ')) {
    throw new LexServerAuthError(
      'AuthenticationRequired',
      'Bearer token required',
      {
        Bearer: { realm: 'api' },
      },
    )
  }

  const token = header.slice(7)
  const session = await verifyToken(token)
  if (!session) {
    throw new LexServerAuthError(
      'InvalidToken',
      'Token is invalid or expired',
      {
        Bearer: { realm: 'api', error: 'invalid_token' },
      },
    )
  }

  return { did: session.did, scope: session.scope }
}

// Use with handlers
router.add(app.bsky.feed.post.create, {
  auth: requireAuth,
  handler: async ({ credentials, input }) => {
    // credentials.did is available here
    const post = await db.createPost(credentials.did, input.body)
    return { body: post }
  },
})

认证函数 :

1. 名为 在此之前 解析请求正文
2. 接收 params, (中文(简体) ), request,以及 connection 资讯
3. 应投出 LexServerError (或 LexServerAuthError关于失败
4. 返回交给处理者的证书

WWW- 认证信头

使用 LexServerAuthError 包括 WWW-Authenticate 错误回复中的标题 :

import { LexServerAuthError } from '@atproto/lex-server'

// Simple Bearer challenge
throw new LexServerAuthError('AuthenticationRequired', 'Login required', {
  Bearer: { realm: 'api' },
})
// WWW-Authenticate: Bearer realm="api"

// Multiple schemes
throw new LexServerAuthError('AuthenticationRequired', 'Auth required', {
  Bearer: { realm: 'api', scope: 'read write' },
  Basic: { realm: 'api' },
})
// WWW-Authenticate: Bearer realm="api", scope="read write", Basic realm="api"

// Token68 format
throw new LexServerAuthError('AuthenticationRequired', 'Auth required', {
  Bearer: 'token68value',
})
// WWW-Authenticate: Bearer token68value

处理错误

LexServer 错误

掷出 LexServerError 返回结构化的 XRPC 错误响应 :

import { LexServerError } from '@atproto/lex-server'

router.add(app.bsky.actor.getProfile, async ({ params }) => {
  try {
    const profile = await db.getProfile(params.actor)
    return { body: profile }
  } catch (cause) {
    throw new LexServerError(
      404,
      {
        error: 'NotFound',
        message: 'Profile not found',
      },
      {
        'cache-control': 'max-age=600',
      },
      { cause },
    )
  }
})

使用 XRPC 格式的错误反应 :

{
  "error": "NotFound",
  "message": "Profile not found"
}

LexServer Auth 错误

LexServerAuthError 扩展 LexServerError 与 WWW-Authenticate 页眉支持 :

import { LexServerAuthError } from '@atproto/lex-server'

throw new LexServerAuthError('AuthenticationRequired', 'Invalid credentials', {
  Bearer: { realm: 'api' },
})

返回与 $401 的响应 WWW-Authenticate 头曰.

处理器召回错误

使用 onHandlerError 以登录或报告出乎意料的错误 :

const router = new LexRouter({
  onHandlerError: async ({ error, request, method }) => {
    // Log errors (excluding expected abort signals)
    console.error(`Error in ${method.nsid}:`, error)
    await reportToSentry(error)
  },
})

[注]

回调只用于意外错误,而不是用于 LexError 实例或请求中止.

节点.js 服务器

这个 @atproto/lex-server/nodejs 子路径提供Node.js特定公用事业.

服务( )

启动服务器并开始监听 :

import { serve } from '@atproto/lex-server/nodejs'

const server = await serve(router, { port: 3000 })
console.log('Server listening on port 3000')

// Graceful shutdown
await server.terminate()

服务器支持 AsyncDisposable数字 :

await using server = await serve(router, { port: 3000 })
// Server is automatically terminated when scope exits

选项 :

type StartServerOptions = {
  port?: number
  host?: string
  gracefulTerminationTimeout?: number // ms to wait for connections to close
}

创建 Server ()

不启动而创建服务器 :

import { createServer } from '@atproto/lex-server/nodejs'

const server = createServer(router, {
  gracefulTerminationTimeout: 5000,
})

server.listen(3000, () => {
  console.log('Server listening')
})

请求收听器 ()

将处理器转换为快递/连接兼容的中间软件 :

import express from 'express'
import { toRequestListener } from '@atproto/lex-server/nodejs'

const app = express()

// Mount the XRPC router
app.use(toRequestListener(router.fetch))

app.listen(3000)

升级WebSocket ()

在节点.js中需要 WebSocket 订阅支持 :

import { LexRouter } from '@atproto/lex-server'
import { upgradeWebSocket } from '@atproto/lex-server/nodejs'

const router = new LexRouter({ upgradeWebSocket })

高级使用

自定义响应对象

还原 a Response 用于完全控制响应的对象(任何类型的功能) 该机构的审定:

router.add(schema, async ({ params }) => {
  if (params.redirect) {
    return Response.redirect('https://example.com', 302)
  }

  return Response.json(
    { custom: true },
    {
      status: 201,
      headers: {
        'X-Custom-Header': 'value',
      },
    },
  )
})

答复信头

在回复中添加标题 :

router.add(schema, async ({ params }) => {
  return {
    body: { data: 'value' },
    headers: {
      'Cache-Control': 'public, max-age=3600',
      'X-Request-Id': crypto.randomUUID(),
    },
  }
})

连接信息

访问网络连接信息 :

router.add(schema, async ({ connection }) => {
  console.log('Remote address:', connection?.remoteAddr?.hostname)
  console.log('Local address:', connection?.localAddr?.hostname)
  return { body: { status: 'ok' } }
})

连接信息结构 :

type ConnectionInfo = {
  localAddr?: {
    hostname: string
    port: number
    transport: 'tcp' | 'udp'
  }
  remoteAddr?: {
    hostname: string
    port: number
    transport: 'tcp' | 'udp'
  }
}

许可证

麻省理工学院或Apachi2