@ atcute/ xrpc- 服务器
xRPC 服务器的网络框架.
npm install @atcute/xrpc-server
先决条件
此框架依赖于由 @atcute/lex-cli,你会需要遵循它的
快速启动引导 关于如何设置它.
对于这些例子,我们将使用简单的查询操作来问候一个名字:
// file: lexicons/com/example/greet.json
{
"lexicon": 1,
"id": "com.example.greet",
"defs": {
"main": {
"type": "query",
"parameters": {
"type": "params",
"required": ["name"],
"properties": {
"name": { "type": "string" }
}
},
"output": {
"encoding": "application/json",
"schema": {
"type": "object",
"required": ["message"],
"properties": {
"message": { "type": "string" }
}
}
}
}
}
}
使用量
处理请求
使用情况 addQuery() 查询(GET)和 addProcedure() 程序(POST)。 处理者收到
类型 params 财务报告和审定财务报表 input,并使用 json() 帮助者 :
import { XRPCRouter, json } from '@atcute/xrpc-server';
import { cors } from '@atcute/xrpc-server/middlewares/cors';
import { ComExampleGreet, ComExampleCreatePost } from './lexicons/index.js';
const router = new XRPCRouter({ middlewares: [cors()] });
router.addQuery(ComExampleGreet, {
async handler({ params }) {
return json({ message: `hello ${params.name}!` });
},
});
router.addProcedure(ComExampleCreatePost, {
async handler({ input }) {
const post = await db.createPost(input);
return json(post);
},
});
export default router;
服务路由器
在Deno、Bun或Cloudflare Workers上,你可以直接输出路由器,并期望它能成功 那个盒子.
对于Node.js, 你需要一个接取到节点的适配器像
@remix-run/node-fetch-server 由于路由器使用标准Web
请求/答复:
import * as http from 'node:http';
import { createRequestListener } from '@remix-run/node-fetch-server';
import { XRPCRouter } from '@atcute/xrpc-server';
const router = new XRPCRouter();
// ... add handlers ...
const server = http.createServer(createRequestListener(router.fetch));
server.listen(3000, () => {
console.log('listening on port 3000');
});
独立处理器
如果您只需要一个 XRPC 操作, 您可以跳过创建路由器并导出一个处理器 直接 :
import { createXrpcHandler, json } from '@atcute/xrpc-server';
import { AppBskyFeedGetFeedSkeleton } from '@atcute/bluesky';
export default createXrpcHandler({
lxm: AppBskyFeedGetFeedSkeleton,
async handler({ params }) {
return json({ feed: [] });
},
});
请求应运至 /xrpc/<nsid>。 。 。 .
错误处理
掷出 XRPCError 在处理器中返回错误回复:
import { XRPCError } from '@atcute/xrpc-server';
router.addQuery(ComExampleGetPost, {
async handler({ params, request }) {
const session = await getSession(request);
if (!session) {
throw new XRPCError({ status: 401, error: 'AuthenticationRequired' });
}
const post = await db.getPost(params.uri);
if (!post) {
throw new XRPCError({ status: 400, error: 'InvalidRequest', message: `post not found` });
}
return json(post);
},
});
也提供方便分类: InvalidRequestError, (中文(简体) ), AuthRequiredError, (中文(简体) ),
ForbiddenError, (中文(简体) ), RateLimitExceededError, (中文(简体) ), InternalServerError, (中文(简体) ), UpstreamFailureError, (中文(简体) ),
NotEnoughResourcesError, (中文(简体) ), UpstreamTimeoutError。 。 。 .
AuthRequiredError 接受一个 wwwAuthenticate 选项,自动格式化 RFC 7235
WWW-Authenticate 回复(和附件)的标题 access-control-expose-headers 所以浏览器
可从遥感中心的答复中读取:
import { AuthRequiredError } from '@atcute/xrpc-server';
throw new AuthRequiredError({
message: 'invalid token',
wwwAuthenticate: { scheme: 'Bearer', params: { error: 'BadJwtSignature' } },
});
观察错误
对日志或衡量标准,请使用 onError 编号 : onSocketError 路由器选项 。 这是火和遗忘
与响应生成同时引用的钩子, 不因客户端引起的错误而被调用
(已撤销的请求, XRPCError, (中文(简体) ), XRPCSubscriptionError)——只针对值得报告的错误:
const router = new XRPCRouter({
onError({ error, request }) {
reportToSentry(error, { url: request.url });
},
onSocketError({ error, request }) {
reportToSentry(error, { url: request.url, subscription: true });
},
});
健康检查
路由器可以选择回答 /xrpc/_health 如果你通过了 handleHealthCheck。这个终点是
非标准——消费者决定反应形状和状态.
const router = new XRPCRouter({
async handleHealthCheck() {
const healthy = await pingDatabase();
return Response.json({ status: healthy ? 'ok' : 'degraded' }, { status: healthy ? 200 : 503 });
},
});
订阅
订阅提供WebSocket上的实时流线. 它们需要一个运行时间特定适配器:
| 运行时间 | 适配软件包 |
|---|---|
| 阿宝 | @atcute/xrpc-server-bun |
| 节点.js | @atcute/xrpc-server-node |
| 德诺 | @atcute/xrpc-server-deno |
| 云花岗工人 | @atcute/xrpc-server-cloudflare |
以下是使用 Bun 的示例:
import { XRPCRouter } from '@atcute/xrpc-server';
import { createBunWebSocket } from '@atcute/xrpc-server-bun';
import { ComExampleSubscribe } from './lexicons/index.js';
const ws = createBunWebSocket();
const router = new XRPCRouter({ websocket: ws.adapter });
router.addSubscription(ComExampleSubscribe, {
async *handler({ params, signal }) {
// yield messages until the client disconnects
while (!signal.aborted) {
const events = await getNewEvents(params.cursor);
for (const event of events) {
yield event;
}
await sleep(1000);
}
},
});
export default ws.wrap(router);
处理器是生成消息的同步生成器。 每个生成的值被编码为 CBOR
框并发送给客户端。 页:1 signal 当客户端断开时中止.
用于订阅错误, 使用 XRPCSubscriptionError数字 :
import { XRPCSubscriptionError } from '@atcute/xrpc-server';
router.addSubscription(ComExampleSubscribe, {
async *handler({ params }) {
if (params.cursor && isCursorTooOld(params.cursor)) {
throw new XRPCSubscriptionError({
error: 'FutureCursor',
message: `cursor is too old`,
});
}
// ...
},
});
每个适配器处理后压:每个适配器 create*WebSocket() 工厂接受
highWaterMark 编号 : lowWaterMark 选项( 默认 250 KB / 50 KB ),当
向外缓冲器会增大。 Cloudflare 工人适配器不应用回压——运行时间
不披露即将退出的 WebSocket 缓冲器.
服务认证
页:1 @atcute/xrpc-server/auth 子包装为服务对服务认证提供公用事业
使用 JWTs.
核查收到的JWT:
import { ServiceJwtVerifier } from '@atcute/xrpc-server/auth';
import {
CompositeDidDocumentResolver,
PlcDidDocumentResolver,
WebDidDocumentResolver,
} from '@atcute/identity-resolver';
const jwtVerifier = new ServiceJwtVerifier({
// list of audience values this service accepts. during the transition to proposal 0014
// (atproto service auth audience), configure both the bare DID and the DID-with-service-ref
// form so that tokens from older issuers keep working alongside new ones.
acceptAudiences: ['did:web:my-service.example.com', 'did:web:my-service.example.com#atproto_pds'],
resolver: new CompositeDidDocumentResolver({
methods: {
plc: new PlcDidDocumentResolver(),
web: new WebDidDocumentResolver(),
},
}),
});
router.addQuery(ComExampleProtectedEndpoint, {
async handler({ request }) {
const auth = await jwtVerifier.verifyRequest(request, { lxm: 'com.example.protectedEndpoint' });
return json({ caller: auth.issuer });
},
});
verifyRequest 描述 Authorization: Bearer 信头, 验证符号, 并扔一个
AuthRequiredError 人口众多 WWW-Authenticate: Bearer error="…" 对每次失败都提出挑战
路径。 前进 request.signal 已输入解析度, 因此中止的请求不会保持网络调用
活着.
附加选项调校校正 :
const jwtVerifier = new ServiceJwtVerifier({
acceptAudiences: [...],
resolver: ...,
maxAge: 300, // max token lifetime window in seconds (default 300)
clockLeeway: 5, // leeway applied to nbf/exp comparisons (default 5)
didResolveTimeout: 5000, // issuer resolution timeout in ms (default 5000; null to disable)
replayStore: { // optional replay protection; requires jti in tokens
async check({ iss, jti }, ttlSeconds) {
const key = `${iss}:${jti}`;
const isNew = await redis.setnx(key, '1');
if (isNew === 1) await redis.expire(key, ttlSeconds);
return isNew === 1;
},
},
});
did resolution 在签名验证之前, 所以伪造的符号可以触发请求
did:web 主机在 iss。 。 。 . didResolveTimeout 限制每项决议的期限; 它不限制
目的地。 目标。 配置解析器 fetch 拒绝非公开地址(见
服务器侧分辨率和 SSRF) (中文(简体) ).
缓存可以减少重复的取景,尽管签名失败触发了未标记的重试.
创建出局 JWT :
import { createServiceJwt } from '@atcute/xrpc-server/auth';
const jwt = await createServiceJwt({
keypair: myServiceKeypair,
issuer: 'did:web:my-service.example.com',
audience: 'did:plc:targetservice',
lxm: 'com.example.someEndpoint',
});
// use jwt in Authorization header when calling other services
内部电话
您可以使用 @atcute/client数字 :
import { Client, ok } from '@atcute/client';
const client = new Client({
handler(pathname, init) {
return router.fetch(new Request(new URL(pathname, 'http://localhost'), init));
},
});
const data = await ok(
client.get('com.example.greet', {
params: { name: 'world' },
}),
);
console.log(data.message); // fully typed!