Something went wrong. Try again.
A Space host for truly public data
Something went wrong. Try again.
2.5 kB · 74 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475import { type Keypair } from '@atproto/crypto'import { createServiceAuthJwt } from './credential'import { type SpaceHostDb } from './db'import { lexBytes } from './xrpc'
const DELIVERY_TIMEOUT_MS = 10_000export const NOTIFY_WRITE_LXM = 'com.atproto.space.notifyWrite'
export type WriteEvent = { space: string repo: string rev: string hash: Uint8Array}
/** * Forward a write notification to every syncer registered for the space. * Best-effort per proposal 0016: failures are logged and dropped; a missed * notification is healed by a later write or a periodic sweep. * * Each registration names a service identifier (DID + optional fragment). * The call is addressed to that identifier — service auth `aud` is the * identifier as registered — and delivered to the endpoint it resolves to * now, falling back to the endpoint recorded at registration time. Signed * with the authority's key when the pool can sign for it, unsigned * otherwise (a spec-conformant recipient will reject the latter, but there * is nothing better to do without the key). */export async function fanOutWrite( db: SpaceHostDb, authorityDid: string, keypair: Keypair | null, event: WriteEvent, resolveEndpoint: (service: string) => Promise<string | null>,): Promise<void> { const body = { space: event.space, repo: event.repo, rev: event.rev, hash: lexBytes(event.hash), }
const registrations = db.listRegistrations(event.space) await Promise.all( registrations.map(async (reg) => { try { const endpoint = (await resolveEndpoint(reg.service).catch(() => null)) ?? reg.endpoint const headers: Record<string, string> = { 'content-type': 'application/json', } if (keypair) { const jwt = await createServiceAuthJwt( { iss: authorityDid, aud: reg.service, lxm: NOTIFY_WRITE_LXM }, keypair, ) headers.authorization = `Bearer ${jwt}` } const url = new URL(`/xrpc/${NOTIFY_WRITE_LXM}`, endpoint) const res = await fetch(url, { method: 'POST', headers, body: JSON.stringify(body), signal: AbortSignal.timeout(DELIVERY_TIMEOUT_MS), }) if (!res.ok) { console.warn(`notify delivery to ${reg.service} (${endpoint}) failed: ${res.status}`) } } catch (err) { console.warn(`notify delivery to ${reg.service} failed:`, err) } }), )}