// Labels — loser-club's signed labels, served to bsky by the gateway (member labels, hyper-social #80 step 4).
//
// hpp signs a labeler DID's labels and serves them as a label core beside the firehose (0.3.128); the
// membrane's fronts replicate it (atproto-did-peer 0.2.15). The gateway serves them on bsky's two labeler
// surfaces with favela-web's semantics (labeler-export.js, label-firehose.js):
// queryLabels uriPatterns with a `*` suffix wildcard, sources default to every labeler we serve,
// `sig` as { $bytes }, numeric cursor, limit ≤ 250
// subscribeLabels dag-cbor { t:'#labels', op:1 } ++ { seq, labels:[label] }, one label per frame,
// replay from ?cursor then live tail; FutureCursor error frame past the head
//
// ⛔ Every label is VERIFIED against its src DID's key before the gateway keeps it — the gateway is keyless
// and must not relay a label it cannot prove. And the seq bsky's cursor counts is the GATEWAY's own log
// (a file in its state dir), not a front's core: co-owners order their cores differently and the served
// front can switch, so a front's index would mean different labels on different days.
import 'atproto-bridge/pds/lib/bare-globals.js' // FIRST: TextEncoder for dag-cbor under Bare
import test from 'brittle'
import fs from 'bare-fs'
import b4a from 'b4a'
import * as dagCbor from '@ipld/dag-cbor'
import { keyFromPrivate } from 'atproto-identity/lib/atproto-identity.js'
import { buildLabel, signLabel, encodeLabel } from 'atproto-identity/lib/labels.js'
import { createLabels } from '../lib/labels.js'
const LC = 'did:plc:wqnwulql7ndcz6xmtvaovl6u'
const WJ = 'did:plc:wj62ebbusln6gxo5azo6ogzo'
const E2 = 'did:plc:e2jevhuurrhlm36ufk7fklpg'
const POST = 'at://' + WJ + '/app.bsky.feed.post/3mrhxnkbhv622'
const lcKey = keyFromPrivate(new Uint8Array(32).fill(44))
const otherKey = keyFromPrivate(new Uint8Array(32).fill(45))
const tmp = () => { const d = '/tmp/gw-labels-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d }
const signed = (uri, val, { neg, cts = '2026-10-14T00:00:00.000Z', key = lcKey, src = LC } = {}) =>
signLabel(buildLabel({ src, uri, val, cts, neg }), key.privateKey)
// A front's label core: what atproto-did-peer exposes as `front.labels`.
function fakeCore (labels = []) {
const blocks = labels.map(encodeLabel)
const fns = new Set()
return {
get length () { return blocks.length },
get: async (i) => blocks[i],
on: (ev, fn) => { if (ev === 'append') fns.add(fn) },
off: (ev, fn) => fns.delete(fn),
push: (l) => { blocks.push(encodeLabel(l)); for (const fn of fns) fn() }
}
}
function fakePeers () {
const fns = new Set()
return {
onLabels: (fn) => { fns.add(fn); return () => fns.delete(fn) },
tell: (did, front) => { for (const fn of fns) fn(did, front) }
}
}
const front = (core, key = lcKey) => ({ labels: core, verifyKey: key.publicKey })
const settle = () => new Promise((resolve) => setTimeout(resolve, 20))
const query = (lab, q) => lab.xrpc['com.atproto.label.queryLabels']({ query: new URLSearchParams(q) })
async function started (t, dir = tmp()) {
const peers = fakePeers()
const lab = createLabels({ peers, stateDir: dir, log: () => {} })
await lab.start()
t.teardown(() => lab.stop())
return { peers, lab, dir }
}
test('★ a front\'s verified labels are served by queryLabels, in favela-web\'s wire shape', async (t) => {
const { peers, lab } = await started(t)
peers.tell(LC, front(fakeCore([signed(WJ, 'loser-club-member'), signed(POST, 'spam')])))
await settle()
const r = await query(lab, [['uriPatterns', '*']])
t.is(r.status, 200)
t.is(r.body.labels.length, 2)
const m = r.body.labels.find((l) => l.val === 'loser-club-member')
t.is(m.src, LC)
t.is(m.uri, WJ)
t.ok(m.sig && typeof m.sig.$bytes === 'string', 'sig as { $bytes } (base64)')
t.absent('cursor' in r.body, 'no cursor when the page is the whole set')
})
test('uriPatterns: exact and `*` suffix; sources filter; numeric cursor', async (t) => {
const { peers, lab } = await started(t)
peers.tell(LC, front(fakeCore([signed(WJ, 'loser-club-member'), signed(E2, 'loser-club-member'), signed(POST, 'spam')])))
await settle()
t.alike((await query(lab, [['uriPatterns', WJ]])).body.labels.map((l) => l.uri), [WJ], 'exact')
t.is((await query(lab, [['uriPatterns', 'at://' + WJ + '/*']])).body.labels.length, 1, 'suffix wildcard')
t.is((await query(lab, [['uriPatterns', '*'], ['sources', 'did:plc:someoneelse']])).body.labels.length, 0, 'sources filter')
const p1 = await query(lab, [['uriPatterns', '*'], ['limit', '2']])
t.is(p1.body.labels.length, 2)
t.is(p1.body.cursor, '2')
const p2 = await query(lab, [['uriPatterns', '*'], ['limit', '2'], ['cursor', p1.body.cursor]])
t.is(p2.body.labels.length, 1)
t.absent('cursor' in p2.body)
})
test('⛔ a label that does not verify against its src DID\'s key is never served', async (t) => {
const { peers, lab } = await started(t)
peers.tell(LC, front(fakeCore([
signed(WJ, 'loser-club-member', { key: otherKey }), // forged: signed by another key
signed(WJ, 'spam', { src: WJ }), // a label naming a different src than the DID that served it
signed(E2, 'loser-club-member')
])))
await settle()
const r = await query(lab, [['uriPatterns', '*']])
t.alike(r.body.labels.map((l) => l.uri), [E2], 'only the verified one')
t.is(lab.metrics().rejected, 2, 'and the two rejects are counted')
})
test('a new label on the front is picked up live; the same label from two co-owners is kept once', async (t) => {
const { peers, lab } = await started(t)
const a = fakeCore([signed(WJ, 'loser-club-member')])
const b = fakeCore([signed(WJ, 'loser-club-member')]) // a co-owner's core holds the same signed label
peers.tell(LC, front(a))
peers.tell(LC, front(b))
await settle()
t.is((await query(lab, [['uriPatterns', '*']])).body.labels.length, 1, 'deduplicated by bytes')
a.push(signed(E2, 'loser-club-member'))
await settle()
t.is((await query(lab, [['uriPatterns', '*']])).body.labels.length, 2, 'the append arrived')
})
test('★ the log persists: the same seq after a restart, whatever order the fronts arrive in', async (t) => {
const dir = tmp()
{
const peers = fakePeers()
const lab = createLabels({ peers, stateDir: dir, log: () => {} })
await lab.start()
peers.tell(LC, front(fakeCore([signed(WJ, 'loser-club-member'), signed(E2, 'loser-club-member')])))
await settle()
await lab.stop()
}
const { peers, lab } = await started(t, dir)
t.is((await query(lab, [['uriPatterns', '*']])).body.labels.length, 2, 'loaded from the state dir before any front')
peers.tell(LC, front(fakeCore([signed(E2, 'loser-club-member'), signed(WJ, 'loser-club-member')])))
await settle()
const uris = (await query(lab, [['uriPatterns', '*']])).body.labels.map((l) => l.uri)
t.alike(uris, [WJ, E2], 'same order as before; nothing duplicated')
})
// subscribeLabels: a fake bare-ws socket.
function fakeWs () {
const frames = []
const fns = {}
return { frames, write: (b) => frames.push(b), on: (ev, fn) => { fns[ev] = fn }, end () {}, destroy () {}, close: () => fns.close && fns.close() }
}
const HEADER = dagCbor.encode({ t: '#labels', op: 1 })
const ERR = dagCbor.encode({ op: -1 })
const body = (frame, head) => dagCbor.decode(frame.subarray(head.length))
test('★ subscribeLabels: replay from the cursor, then the live tail, one label per frame', async (t) => {
const { peers, lab } = await started(t)
const core = fakeCore([signed(WJ, 'loser-club-member'), signed(E2, 'loser-club-member')])
peers.tell(LC, front(core))
await settle()
const ws = fakeWs()
lab.ws['/xrpc/com.atproto.label.subscribeLabels'](ws, new URL('http://x/xrpc/com.atproto.label.subscribeLabels?cursor=1'))
await settle()
t.is(ws.frames.length, 1, 'cursor=1: only seq 2 is replayed')
t.ok(b4a.equals(ws.frames[0].subarray(0, HEADER.length), HEADER), '#labels header')
const f = body(ws.frames[0], HEADER)
t.is(f.seq, 2)
t.is(f.labels.length, 1)
t.is(f.labels[0].uri, E2)
t.ok(f.labels[0].sig instanceof Uint8Array, 'sig as raw bytes on the CBOR wire')
core.push(signed(POST, 'spam'))
await settle()
t.is(ws.frames.length, 2, 'live tail')
t.is(body(ws.frames[1], HEADER).seq, 3)
ws.close()
core.push(signed(E2, 'spam'))
await settle()
t.is(ws.frames.length, 2, 'nothing after the client left')
})
test('subscribeLabels: a cursor past the head gets a FutureCursor error frame first', async (t) => {
const { lab } = await started(t)
const ws = fakeWs()
lab.ws['/xrpc/com.atproto.label.subscribeLabels'](ws, new URL('http://x/xrpc/com.atproto.label.subscribeLabels?cursor=9'))
await settle()
t.ok(b4a.equals(ws.frames[0].subarray(0, ERR.length), ERR))
t.is(body(ws.frames[0], ERR).error, 'FutureCursor')
})