// Labels — a labeler DID's signed labels served to bsky by the gateway (member labels, hyper-social #80).
//
// hpp signs a labeler DID's labels with the DID key and serves them as a label core beside its firehose
// (hyper-proto-peer 0.3.128); the membrane's fronts replicate that core (atproto-did-peer 0.2.15) and the
// router tells us when a front gains one (onLabels). This module serves bsky's two labeler surfaces with
// favela-web's semantics (lib/labeler-export.js, lib/label-firehose.js):
//
// com.atproto.label.queryLabels uriPatterns (`*` suffix wildcard), sources (default: every labeler
// served here), numeric cursor, limit ≤ 250, `sig` as { $bytes }
// com.atproto.label.subscribeLabels dag-cbor {t:'#labels',op:1} ++ {seq, labels:[label]}, one label per
// frame; replay from ?cursor, then live; FutureCursor past the head
//
// ⛔ VERIFIED before kept: a label must name the DID whose front served it as `src` and verify against that
// DID's key — the gateway is keyless and relays nothing it cannot prove.
//
// ⛔ THE SEQ IS OURS. bsky's AppView resumes subscribeLabels from the last seq it saw. A front's core is a
// bad source of seq: co-owners append the same labels in different orders and the served front can switch
// between them. So every verified label is appended once (deduplicated by its bytes) to the gateway's own
// log, `labels.log` in the state dir — one hex label per line, append-only — and seq is a 1-based index
// into that. It survives restarts in the same order, whichever front arrives first.
// FIRST: dag-cbor and @noble use TextEncoder at module load, which Bare lacks. Imported here rather than
// relied on through gateway.js's import order, so this module loads on its own (tests, the CLI).
import 'atproto-bridge/pds/lib/bare-globals.js'
import fs from 'bare-fs'
import { join } from 'bare-path'
import b4a from 'b4a'
import * as dagCbor from '@ipld/dag-cbor'
import { decodeLabel, encodeLabel, verifyLabel } from 'atproto-identity/lib/labels.js'
import { encodeDidKey, CURVE_SECP256K1 } from 'atproto-identity/lib/did-key.js'
const LABELS_HEADER = dagCbor.encode({ t: '#labels', op: 1 })
const ERROR_HEADER = dagCbor.encode({ op: -1 })
const concat = (a, b) => { const out = new Uint8Array(a.length + b.length); out.set(a, 0); out.set(b, a.length); return out }
const labelFrame = (seq, labels) => concat(LABELS_HEADER, dagCbor.encode({ seq, labels }))
const errorFrame = (error, message = '') => concat(ERROR_HEADER, dagCbor.encode({ error, message }))
// favela-web toWireLabel: JSON XRPC form; undefined optional fields omitted.
export function toWireLabel (label) {
const w = { src: label.src, uri: label.uri, val: label.val, cts: label.cts }
if (label.cid !== undefined) w.cid = label.cid
if (label.neg !== undefined) w.neg = label.neg
if (label.exp !== undefined) w.exp = label.exp
if (label.ver !== undefined) w.ver = label.ver
if (label.sig) w.sig = { $bytes: b4a.toString(b4a.from(label.sig), 'base64') }
return w
}
export function uriMatches (uri, pattern) {
if (pattern === '*') return true
if (pattern.endsWith('*')) return String(uri).startsWith(pattern.slice(0, -1))
return uri === pattern
}
export function createLabels ({ peers, stateDir, log = () => {} } = {}) {
if (!stateDir) throw new Error('createLabels: stateDir required (the label log lives there)')
const file = join(stateDir, 'labels.log')
const labels = [] // decoded, in log order; seq = index + 1
const seen = new Set() // hex of each kept label
const subscribers = new Set() // () => void, called after an append
const wired = new WeakSet() // label cores we listen to
const progress = new WeakMap() // label core -> next index to read
let rejected = 0
let unsubscribe = null
let queue = Promise.resolve()
let stopped = false
const keep = (hex) => {
const l = decodeLabel(b4a.from(hex, 'hex'))
labels.push(l)
seen.add(hex)
}
// Read a front's label core from where we left off. Serialized, so the log order is one order.
function pull (did, front) {
queue = queue.then(async () => {
if (stopped) return
const core = front.labels
const didKey = encodeDidKey(CURVE_SECP256K1, front.verifyKey)
let i = progress.get(core) || 0
const added = []
for (; i < core.length; i++) {
let l = null
try { l = decodeLabel(await core.get(i)) } catch {}
if (!l || l.src !== did || !verifyLabel(l, didKey)) {
rejected++
log({ msg: 'labels: rejected a label that does not verify', did, index: i })
continue
}
const hex = b4a.toString(encodeLabel(l), 'hex')
if (seen.has(hex)) continue
added.push(hex)
keep(hex)
}
progress.set(core, i)
if (!added.length) return
fs.appendFileSync(file, added.join('\n') + '\n')
log({ msg: 'labels: kept', did, added: added.length, total: labels.length })
for (const fn of subscribers) { try { fn() } catch {} }
}).catch((err) => log({ msg: 'labels: pull failed', did, error: (err && err.message) || String(err) }))
return queue
}
function watch (did, front) {
if (!front || !front.labels || !front.verifyKey) return
if (!wired.has(front.labels)) {
wired.add(front.labels)
front.labels.on('append', () => { pull(did, front) })
}
return pull(did, front)
}
async function queryLabels ({ query }) {
const patterns = query.getAll('uriPatterns')
const pats = patterns.length ? patterns : ['*']
const srcs = query.getAll('sources')
const allow = srcs.length ? new Set(srcs) : null // default: every labeler served here
const limit = Math.max(1, Math.min(Number(query.get('limit')) || 50, 250))
const start = query.get('cursor') ? Math.max(0, Number(query.get('cursor')) || 0) : 0
const matched = labels.filter((l) => (!allow || allow.has(l.src)) && pats.some((p) => uriMatches(l.uri, p)))
const page = matched.slice(start, start + limit)
const body = { labels: page.map(toWireLabel) }
if (start + limit < matched.length) body.cursor = String(start + limit)
return { status: 200, body }
}
// One subscriber: replay from ?cursor (the last seq it holds), then live. Serialized + coalesced, so no
// frame is skipped or sent twice when an append lands during a replay (favela-web's LabelFirehose).
function subscribeLabels (ws, url) {
let i = Math.max(0, Number(url && url.searchParams.get('cursor')) || 0)
if (i > labels.length) { try { ws.write(Buffer.from(errorFrame('FutureCursor', 'cursor is past the labeler head'))) } catch {} ; i = labels.length }
let running = false
let pending = false
let closed = false
const pump = () => {
if (running) { pending = true; return }
running = true
try {
while (!closed && i < labels.length) {
const idx = i++
try { ws.write(Buffer.from(labelFrame(idx + 1, [labels[idx]]))) } catch {}
}
} finally { running = false }
if (!closed && pending) { pending = false; pump() }
}
const done = () => { closed = true; subscribers.delete(pump) }
ws.on('close', done)
ws.on('error', done)
subscribers.add(pump)
pump()
}
async function start () {
try {
for (const line of fs.readFileSync(file, 'utf8').split('\n')) if (line) keep(line)
} catch (err) { if (err && err.code !== 'ENOENT') throw err }
if (peers && typeof peers.onLabels === 'function') unsubscribe = peers.onLabels((did, front) => { watch(did, front) })
log({ msg: 'labels: started', kept: labels.length })
}
async function stop () {
stopped = true
if (unsubscribe) { try { unsubscribe() } catch {} }
subscribers.clear()
await queue
}
const bySrc = () => { const o = {}; for (const l of labels) o[l.src] = (o[l.src] || 0) + 1; return o }
return {
start,
stop,
watch, // for fronts that already held a label core before start()
metrics: () => ({ kept: labels.length, rejected, bySrc: bySrc(), subscribers: subscribers.size }),
xrpc: { 'com.atproto.label.queryLabels': queryLabels },
ws: { '/xrpc/com.atproto.label.subscribeLabels': subscribeLabels }
}
}
export default createLabels