@ atproto/ tap (中文(简体) )
类型脚本客户端库 磁带,用于AT协议网络的同步工具.
磁盘处理火花连接、密码验证、回填和过滤。 这个客户端库允许您连接到一个 Tap 实例, 并接收简单的 JSON 事件, 用于您所关心的重置 .
快速启动
npm install @atproto/tap
import { Tap, SimpleIndexer } from '@atproto/tap'
const tap = new Tap('http://localhost:2480', { adminPassword: 'secret' })
const indexer = new SimpleIndexer()
indexer.identity(async (evt) => {
console.log(`${evt.did} updated identity: ${evt.handle} (${evt.status})`)
})
indexer.record(async (evt) => {
const uri = `at://${evt.did}/${evt.collection}/${evt.rkey}`
if (evt.action === 'create' || evt.action === 'update') {
console.log(`${evt.action}: ${uri}`)
} else {
console.log(`deleted: ${uri}`)
}
})
indexer.error((err) => console.error(err))
const channel = tap.channel(indexer)
channel.start()
await tap.addRepos(['did:plc:ewvi7nxzyoun6zhxrhs64oiz'])
// On shutdown
await channel.destroy()
运行磁带
见 磁带读取 详细信息 踢出和运行。 您的应用程序可以在当地或通过互联网与它通信.
此库旨在与 Tap 运行在默认模式"WebScoket with acks"中使用. 在这种模式下,Tap提供:
- 至少一次交货: 如果接通器在接通前下降, 事件可以重新交付
- 逐项命令: 同一雷波事件按顺序交付
- 后补:当你添加回波时,历史事件会在直播事件之前交付
API 密码
Tap
与 Tap 服务器互动的主要客户端.
const tap = new Tap(url: string, config?: TapConfig)
配置选项 :
adminPassword?: string- Basic 认证的密码( 如果 Tap 服务器启用了认证, 则需要)
方法:
channel(handler: TapHandler, opts?: TapWebsocketOptions): TapChannel- 创建接收事件的WebSocket频道addRepos(dids: string[]): Promise<void>- 在音轨上添加再投放( 触发回填)removeRepos(dids: string[]): Promise<void>- 停止跟踪重置resolveDid(did: string): Promise<DidDocument | null>- 解决文件问题getRepoInfo(did: string): Promise<RepoInfo>- 获取有关追踪回波的信息
TapChannel
接收事件的 WebSocket 连接 。 通过 tap.channel()。 。 。 .
const channel = tap.channel(handler, opts?)
方法:
start(): Promise<void>- 开始接收事件 返回在连接被破坏或出错时解析的保证 .destroy(): Promise<void>- 关闭连接
该频道会自动处理重联并保持功能. 处理器成功完成后自动确认事件 .
SimpleIndexer
按类型处理事件的便利类. 传入 tap.channel() 当与 Tap 打开频道时.
const indexer = new SimpleIndexer()
indexer.identity(async (evt: IdentityEvent) => { ... })
indexer.record(async (evt: RecordEvent) => { ... })
indexer.error((err: Error) => { ... })
如果没有错误处理器被注册,错误会作为未处理的例外抛出.
LexIndexer
使用键入索引器 @atproto/lex 类型安全事件处理计划。 特定记录类型和行动的登记处理员:
import { LexIndexer } from '@atproto/tap'
import * as com from './lexicons/com'
const indexer = new LexIndexer()
// Handle creates for a specific record type
indexer.create(com.example.post, async (evt) => {
// evt.record is fully typed as com.example.post.Main
console.log(`New post: ${evt.record.text}`)
})
// Handle updates
indexer.update(com.example.post, async (evt) => {
console.log(`Updated post: ${evt.record.text}`)
})
// Handle deletes (no record on delete events)
indexer.delete(com.example.post, async (evt) => {
console.log(`Deleted: at://${evt.did}/${evt.collection}/${evt.rkey}`)
})
// Handle both creates and updates with put()
indexer.put(com.example.like, async (evt) => {
console.log(`Like ${evt.action}: ${evt.record.subject.uri}`)
})
// Fallback for unhandled record types/actions
indexer.other(async (evt) => {
console.log(`Unhandled: ${evt.action}, ${evt.collection}`)
})
// Identity and error handlers
indexer.identity(async (evt) => { ... })
indexer.error((err) => { ... })
const channel = tap.channel(indexer)
记录在调用处理人员之前,先根据他们的计划加以验证。 如果校验失败, 就会出现错误, 错误会通过 error 管理者...
重复处理器注册会丢出一个错误,包括 put() 财务报告和审定财务报表 create()编号 :update() 同样的计划.
TapHandler
您可以通过创建一个执行 TapHandler 接口 :
interface TapHandler {
onEvent: (evt: TapEvent, opts: HandlerOpts) => void | Promise<void>
onError: (err: Error) => void
}
interface HandlerOpts {
signal: AbortSignal
ack: () => Promise<void>
}
执行自定义处理器时,请务必打电话 ack() 当你处理完事件后.
事件类型
RecordEvent
type RecordEvent = {
id: number
type: 'record'
action: 'create' | 'update' | 'delete'
did: string
rev: string
collection: string
rkey: string
record?: Record<string, unknown> // present for create/update
cid?: string // present for create/update
live: boolean // true if from firehose, false if from backfill
}
IdentityEvent
type IdentityEvent = {
id: number
type: 'identity'
did: string
handle: string
isActive: boolean
status: 'active' | 'takendown' | 'suspended' | 'deactivated' | 'deleted'
}
Webhook 模式
如果您的 Tap 服务器配置为 Webhook 发送, 您可以使用 parseTapEvent 用于验证所运入的webhook有效载荷:
import express from 'express'
import { parseTapEvent, assureAdminAuth } from '@atproto/tap'
const ADMIN_PASSWORD = process.env.ADMIN_PASSWORD
const app = express()
app.use(express.json())
app.post('/webhook', async (req, res) => {
try {
assureAdminAuth(ADMIN_PASSWORD, req.headers.authorization)
} catch {
return res.status(401).json({ error: 'Unauthorized' })
}
try {
const evt = parseTapEvent(req.body)
// handle event...
res.sendStatus(200)
} catch (err) {
console.error('Failed to process event:', err)
res.status(500).json({ error: 'Failed to process event' })
}
})
公用事业
认证助手
import {
formatAdminAuthHeader,
parseAdminAuthHeader,
assureAdminAuth,
} from '@atproto/tap'
// Format a password into a Basic auth header value
const header = formatAdminAuthHeader('secret')
// => 'Basic YWRtaW46c2VjcmV0'
// Parse an auth header to extract the password (throws if invalid)
const password = parseAdminAuthHeader(header)
// Verify auth header matches expected password (timing-safe, throws if invalid)
assureAdminAuth('secret', req.headers.authorization)
事件解析
import { parseTapEvent } from '@atproto/tap'
const evt = parseTapEvent(jsonData) // validates and returns typed TapEvent
许可证
麻省理工学院