| … | … | @@ -87,7 +87,10 @@ if (roles.includes('lens')) { |
| 87 | 87 | const [id, roomKey] = pair.split('=') |
| 88 | 88 | return { id: (id || '').trim(), roomKey: (roomKey || '').trim() } |
| 89 | 89 | }).filter((f) => f.id && /^[0-9a-f]{64}$/i.test(f.roomKey)) |
| 90 | configs.lens = feeds.length ? { feeds } : new Error('lens config: LENS_FEEDS has no id=roomKey pair') | |
| 90 | // LENS_LABELER_SOCK: the labeler DID's hpp control socket on this host. Set → the roster is projected | |
| 91 | // into `<feed>-member` labels there (hpp refuses unless that DID has `labeler: true`). Unset → none. | |
| 92 | const labelerSock = String(process.env.LENS_LABELER_SOCK || '').trim() || null | |
| 93 | configs.lens = feeds.length ? { feeds, labelerSock } : new Error('lens config: LENS_FEEDS has no id=roomKey pair') | |
| 91 | 94 | } |
| 92 | 95 | |
| 93 | 96 | // Alert state lives beside the control socket unless told otherwise. |
| … | … | @@ -21,6 +21,9 @@ import { startApBridge } from 'ap-bridge/lib/bridge.js' |
| 21 | 21 | import os from 'bare-os' |
| 22 | 22 | import { createProbes } from './probes.js' |
| 23 | 23 | import { createLens } from './lens.js' |
| 24 | import { createLabels } from './labels.js' | |
| 25 | import { createMemberLabels } from './member-labels.js' | |
| 26 | import { Client } from 'hyper-svc' | |
| 24 | 27 | import { createWatcher } from './watch.js' |
| 25 | 28 | import { renderStatusPage } from './panel.js' |
| 26 | 29 | import { renderPublicPage } from './public.js' |
| … | … | @@ -33,7 +36,11 @@ const ROLES = { |
| 33 | 36 | // needs the membrane role up and started first. |
| 34 | 37 | lens: async (cfg, opts, running) => { |
| 35 | 38 | if (!running.membrane) throw new Error('the lens needs the membrane role (start it before the lens)') |
| 36 | const lens = createLens({ membrane: running.membrane, feeds: (cfg && cfg.feeds) || [], log: opts.log }) | |
| 39 | // Labels need the state dir (the gateway's own label log, whose seq bsky's cursor counts). The member | |
| 40 | // projector needs the labeler DID's hpp socket on this host (LENS_LABELER_SOCK) — hpp signs, we state. | |
| 41 | const labels = opts.stateDir ? createLabels({ peers: running.membrane.peers, stateDir: opts.stateDir, log: opts.log }) : null | |
| 42 | const memberLabels = cfg && cfg.labelerSock ? createMemberLabels({ connect: () => Client.connectUds(cfg.labelerSock), log: opts.log }) : null | |
| 43 | const lens = createLens({ membrane: running.membrane, feeds: (cfg && cfg.feeds) || [], log: opts.log, labels, memberLabels }) | |
| 37 | 44 | await lens.start() |
| 38 | 45 | return lens |
| 39 | 46 | } |
| … | … | @@ -63,7 +70,7 @@ export async function startGateway ({ roles, log = defaultLog, bootstrap = null, |
| 63 | 70 | try { |
| 64 | 71 | // A config the entry could not load arrives as an Error: that role is down, for that reason. |
| 65 | 72 | if (configs[name] instanceof Error) throw configs[name] |
| 66 | running[name] = await ROLES[name](configs[name], { log: roleLog, bootstrap, version, adminPages }, running) | |
| 73 | running[name] = await ROLES[name](configs[name], { log: roleLog, bootstrap, version, adminPages, stateDir }, running) | |
| 67 | 74 | log({ msg: 'role up', role: name }) |
| 68 | 75 | } catch (err) { |
| 69 | 76 | down[name] = (err && err.message) || String(err) |
| … | … | @@ -0,0 +1,180 @@ |
| 1 | // Labels — a labeler DID's signed labels served to bsky by the gateway (member labels, hyper-social #80). | |
| 2 | // | |
| 3 | // hpp signs a labeler DID's labels with the DID key and serves them as a label core beside its firehose | |
| 4 | // (hyper-proto-peer 0.3.128); the membrane's fronts replicate that core (atproto-did-peer 0.2.15) and the | |
| 5 | // router tells us when a front gains one (onLabels). This module serves bsky's two labeler surfaces with | |
| 6 | // favela-web's semantics (lib/labeler-export.js, lib/label-firehose.js): | |
| 7 | // | |
| 8 | // com.atproto.label.queryLabels uriPatterns (`*` suffix wildcard), sources (default: every labeler | |
| 9 | // served here), numeric cursor, limit ≤ 250, `sig` as { $bytes } | |
| 10 | // com.atproto.label.subscribeLabels dag-cbor {t:'#labels',op:1} ++ {seq, labels:[label]}, one label per | |
| 11 | // frame; replay from ?cursor, then live; FutureCursor past the head | |
| 12 | // | |
| 13 | // ⛔ VERIFIED before kept: a label must name the DID whose front served it as `src` and verify against that | |
| 14 | // DID's key — the gateway is keyless and relays nothing it cannot prove. | |
| 15 | // | |
| 16 | // ⛔ THE SEQ IS OURS. bsky's AppView resumes subscribeLabels from the last seq it saw. A front's core is a | |
| 17 | // bad source of seq: co-owners append the same labels in different orders and the served front can switch | |
| 18 | // between them. So every verified label is appended once (deduplicated by its bytes) to the gateway's own | |
| 19 | // log, `labels.log` in the state dir — one hex label per line, append-only — and seq is a 1-based index | |
| 20 | // into that. It survives restarts in the same order, whichever front arrives first. | |
| 21 | ||
| 22 | // FIRST: dag-cbor and @noble use TextEncoder at module load, which Bare lacks. Imported here rather than | |
| 23 | // relied on through gateway.js's import order, so this module loads on its own (tests, the CLI). | |
| 24 | import 'atproto-bridge/pds/lib/bare-globals.js' | |
| 25 | import fs from 'bare-fs' | |
| 26 | import { join } from 'bare-path' | |
| 27 | import b4a from 'b4a' | |
| 28 | import * as dagCbor from '@ipld/dag-cbor' | |
| 29 | import { decodeLabel, encodeLabel, verifyLabel } from 'atproto-identity/lib/labels.js' | |
| 30 | import { encodeDidKey, CURVE_SECP256K1 } from 'atproto-identity/lib/did-key.js' | |
| 31 | ||
| 32 | const LABELS_HEADER = dagCbor.encode({ t: '#labels', op: 1 }) | |
| 33 | const ERROR_HEADER = dagCbor.encode({ op: -1 }) | |
| 34 | const concat = (a, b) => { const out = new Uint8Array(a.length + b.length); out.set(a, 0); out.set(b, a.length); return out } | |
| 35 | const labelFrame = (seq, labels) => concat(LABELS_HEADER, dagCbor.encode({ seq, labels })) | |
| 36 | const errorFrame = (error, message = '') => concat(ERROR_HEADER, dagCbor.encode({ error, message })) | |
| 37 | ||
| 38 | // favela-web toWireLabel: JSON XRPC form; undefined optional fields omitted. | |
| 39 | export function toWireLabel (label) { | |
| 40 | const w = { src: label.src, uri: label.uri, val: label.val, cts: label.cts } | |
| 41 | if (label.cid !== undefined) w.cid = label.cid | |
| 42 | if (label.neg !== undefined) w.neg = label.neg | |
| 43 | if (label.exp !== undefined) w.exp = label.exp | |
| 44 | if (label.ver !== undefined) w.ver = label.ver | |
| 45 | if (label.sig) w.sig = { $bytes: b4a.toString(b4a.from(label.sig), 'base64') } | |
| 46 | return w | |
| 47 | } | |
| 48 | ||
| 49 | export function uriMatches (uri, pattern) { | |
| 50 | if (pattern === '*') return true | |
| 51 | if (pattern.endsWith('*')) return String(uri).startsWith(pattern.slice(0, -1)) | |
| 52 | return uri === pattern | |
| 53 | } | |
| 54 | ||
| 55 | export function createLabels ({ peers, stateDir, log = () => {} } = {}) { | |
| 56 | if (!stateDir) throw new Error('createLabels: stateDir required (the label log lives there)') | |
| 57 | const file = join(stateDir, 'labels.log') | |
| 58 | const labels = [] // decoded, in log order; seq = index + 1 | |
| 59 | const seen = new Set() // hex of each kept label | |
| 60 | const subscribers = new Set() // () => void, called after an append | |
| 61 | const wired = new WeakSet() // label cores we listen to | |
| 62 | const progress = new WeakMap() // label core -> next index to read | |
| 63 | let rejected = 0 | |
| 64 | let unsubscribe = null | |
| 65 | let queue = Promise.resolve() | |
| 66 | let stopped = false | |
| 67 | ||
| 68 | const keep = (hex) => { | |
| 69 | const l = decodeLabel(b4a.from(hex, 'hex')) | |
| 70 | labels.push(l) | |
| 71 | seen.add(hex) | |
| 72 | } | |
| 73 | ||
| 74 | // Read a front's label core from where we left off. Serialized, so the log order is one order. | |
| 75 | function pull (did, front) { | |
| 76 | queue = queue.then(async () => { | |
| 77 | if (stopped) return | |
| 78 | const core = front.labels | |
| 79 | const didKey = encodeDidKey(CURVE_SECP256K1, front.verifyKey) | |
| 80 | let i = progress.get(core) || 0 | |
| 81 | const added = [] | |
| 82 | for (; i < core.length; i++) { | |
| 83 | let l = null | |
| 84 | try { l = decodeLabel(await core.get(i)) } catch {} | |
| 85 | if (!l || l.src !== did || !verifyLabel(l, didKey)) { | |
| 86 | rejected++ | |
| 87 | log({ msg: 'labels: rejected a label that does not verify', did, index: i }) | |
| 88 | continue | |
| 89 | } | |
| 90 | const hex = b4a.toString(encodeLabel(l), 'hex') | |
| 91 | if (seen.has(hex)) continue | |
| 92 | added.push(hex) | |
| 93 | keep(hex) | |
| 94 | } | |
| 95 | progress.set(core, i) | |
| 96 | if (!added.length) return | |
| 97 | fs.appendFileSync(file, added.join('\n') + '\n') | |
| 98 | log({ msg: 'labels: kept', did, added: added.length, total: labels.length }) | |
| 99 | for (const fn of subscribers) { try { fn() } catch {} } | |
| 100 | }).catch((err) => log({ msg: 'labels: pull failed', did, error: (err && err.message) || String(err) })) | |
| 101 | return queue | |
| 102 | } | |
| 103 | ||
| 104 | function watch (did, front) { | |
| 105 | if (!front || !front.labels || !front.verifyKey) return | |
| 106 | if (!wired.has(front.labels)) { | |
| 107 | wired.add(front.labels) | |
| 108 | front.labels.on('append', () => { pull(did, front) }) | |
| 109 | } | |
| 110 | return pull(did, front) | |
| 111 | } | |
| 112 | ||
| 113 | async function queryLabels ({ query }) { | |
| 114 | const patterns = query.getAll('uriPatterns') | |
| 115 | const pats = patterns.length ? patterns : ['*'] | |
| 116 | const srcs = query.getAll('sources') | |
| 117 | const allow = srcs.length ? new Set(srcs) : null // default: every labeler served here | |
| 118 | const limit = Math.max(1, Math.min(Number(query.get('limit')) || 50, 250)) | |
| 119 | const start = query.get('cursor') ? Math.max(0, Number(query.get('cursor')) || 0) : 0 | |
| 120 | const matched = labels.filter((l) => (!allow || allow.has(l.src)) && pats.some((p) => uriMatches(l.uri, p))) | |
| 121 | const page = matched.slice(start, start + limit) | |
| 122 | const body = { labels: page.map(toWireLabel) } | |
| 123 | if (start + limit < matched.length) body.cursor = String(start + limit) | |
| 124 | return { status: 200, body } | |
| 125 | } | |
| 126 | ||
| 127 | // One subscriber: replay from ?cursor (the last seq it holds), then live. Serialized + coalesced, so no | |
| 128 | // frame is skipped or sent twice when an append lands during a replay (favela-web's LabelFirehose). | |
| 129 | function subscribeLabels (ws, url) { | |
| 130 | let i = Math.max(0, Number(url && url.searchParams.get('cursor')) || 0) | |
| 131 | if (i > labels.length) { try { ws.write(Buffer.from(errorFrame('FutureCursor', 'cursor is past the labeler head'))) } catch {} ; i = labels.length } | |
| 132 | let running = false | |
| 133 | let pending = false | |
| 134 | let closed = false | |
| 135 | const pump = () => { | |
| 136 | if (running) { pending = true; return } | |
| 137 | running = true | |
| 138 | try { | |
| 139 | while (!closed && i < labels.length) { | |
| 140 | const idx = i++ | |
| 141 | try { ws.write(Buffer.from(labelFrame(idx + 1, [labels[idx]]))) } catch {} | |
| 142 | } | |
| 143 | } finally { running = false } | |
| 144 | if (!closed && pending) { pending = false; pump() } | |
| 145 | } | |
| 146 | const done = () => { closed = true; subscribers.delete(pump) } | |
| 147 | ws.on('close', done) | |
| 148 | ws.on('error', done) | |
| 149 | subscribers.add(pump) | |
| 150 | pump() | |
| 151 | } | |
| 152 | ||
| 153 | async function start () { | |
| 154 | try { | |
| 155 | for (const line of fs.readFileSync(file, 'utf8').split('\n')) if (line) keep(line) | |
| 156 | } catch (err) { if (err && err.code !== 'ENOENT') throw err } | |
| 157 | if (peers && typeof peers.onLabels === 'function') unsubscribe = peers.onLabels((did, front) => { watch(did, front) }) | |
| 158 | log({ msg: 'labels: started', kept: labels.length }) | |
| 159 | } | |
| 160 | ||
| 161 | async function stop () { | |
| 162 | stopped = true | |
| 163 | if (unsubscribe) { try { unsubscribe() } catch {} } | |
| 164 | subscribers.clear() | |
| 165 | await queue | |
| 166 | } | |
| 167 | ||
| 168 | const bySrc = () => { const o = {}; for (const l of labels) o[l.src] = (o[l.src] || 0) + 1; return o } | |
| 169 | ||
| 170 | return { | |
| 171 | start, | |
| 172 | stop, | |
| 173 | watch, // for fronts that already held a label core before start() | |
| 174 | metrics: () => ({ kept: labels.length, rejected, bySrc: bySrc(), subscribers: subscribers.size }), | |
| 175 | xrpc: { 'com.atproto.label.queryLabels': queryLabels }, | |
| 176 | ws: { '/xrpc/com.atproto.label.subscribeLabels': subscribeLabels } | |
| 177 | } | |
| 178 | } | |
| 179 | ||
| 180 | export default createLabels |
| … | … | @@ -14,8 +14,10 @@ |
| 14 | 14 | // Served through the membrane's XRPC extensions, so did.json advertises #bsky_fg and bsky's AppView reaches |
| 15 | 15 | // it at pds.7rnx.net through the community's PLC #bsky_fg — nothing outside this process changes. |
| 16 | 16 | // |
| 17 | // Labels: nothing produces labels yet (the labeler design is open). queryLabels answers a valid empty set | |
| 18 | // and subscribeLabels holds an empty stream, so the community's advertised labeler is valid but silent. | |
| 17 | // Labels (member labels, #80): with `labels` (lib/labels.js) the lens serves queryLabels/subscribeLabels from | |
| 18 | // the verified label log and hands each community DID's front to it; with `memberLabels` | |
| 19 | // (lib/member-labels.js) every roster change is projected into `<feed>-member` labels. Without them the | |
| 20 | // label surfaces answer a valid empty set, so the community's advertised labeler is valid but silent. | |
| 19 | 21 | |
| 20 | 22 | const GENERATOR = 'app.bsky.feed.generator' |
| 21 | 23 | const POST = 'app.bsky.feed.post' |
| … | … | @@ -49,7 +51,7 @@ export function slugRkey (id) { |
| 49 | 51 | * metrics(), communityRooms() |
| 50 | 52 | * @param {Array<{id, roomKey}>} opts.feeds e.g. [{ id: 'loser-club', roomKey: '91ce4dbc…' }] |
| 51 | 53 | */ |
| 52 | export function createLens ({ membrane, feeds = [], log = () => {}, debounceMs = 1000, frontTimeoutMs = 20000 } = {}) { | |
| 54 | export function createLens ({ membrane, feeds = [], log = () => {}, debounceMs = 1000, frontTimeoutMs = 20000, labels = null, memberLabels = null } = {}) { | |
| 53 | 55 | const index = new Map() // did -> { rev, rows: [{ key, did, rkey, createdAt }], at } |
| 54 | 56 | const state = feeds.map((f) => ({ ...f, rkey: slugRkey(f.id), communityDid: null, curation: 'roster', members: [], unwatch: null, error: null })) |
| 55 | 57 | let rooms = null |
| … | … | @@ -151,9 +153,8 @@ export function createLens ({ membrane, feeds = [], log = () => {}, debounceMs = |
| 151 | 153 | return { status: 200, body: { did, feeds: served.filter((f) => f.communityDid === did).map((f) => ({ uri: feedUri(f) })) } } |
| 152 | 154 | } |
| 153 | 155 | |
| 156 | // No labels module: a valid empty set, and an empty stream held open until the client goes. | |
| 154 | 157 | async function queryLabels () { return { status: 200, body: { labels: [] } } } |
| 155 | ||
| 156 | // An empty label stream: hold the socket open until the client goes; nothing to send yet. | |
| 157 | 158 | function subscribeLabels (ws) { |
| 158 | 159 | ws.on('error', () => {}) |
| 159 | 160 | ws.on('close', () => {}) |
| … | … | @@ -162,11 +163,20 @@ export function createLens ({ membrane, feeds = [], log = () => {}, debounceMs = |
| 162 | 163 | const XRPC = { |
| 163 | 164 | 'app.bsky.feed.getFeedSkeleton': getFeedSkeleton, |
| 164 | 165 | 'app.bsky.feed.describeFeedGenerator': describeFeedGenerator, |
| 165 | 'com.atproto.label.queryLabels': queryLabels | |
| 166 | 'com.atproto.label.queryLabels': queryLabels, | |
| 167 | ...(labels ? labels.xrpc : {}) | |
| 168 | } | |
| 169 | const WS = labels ? { ...labels.ws } : { '/xrpc/com.atproto.label.subscribeLabels': subscribeLabels } | |
| 170 | ||
| 171 | // The community DID is the labeler. A front that already held its label core before the labels module | |
| 172 | // started is never announced to it by the router (onLabels fires on OPENING one), so hand it over here. | |
| 173 | const watchLabeler = (did) => { | |
| 174 | if (!labels || !did) return | |
| 175 | membrane.peers.front(did, { timeout: frontTimeoutMs }).then((fr) => labels.watch(did, fr)).catch(() => {}) | |
| 166 | 176 | } |
| 167 | const WS = { '/xrpc/com.atproto.label.subscribeLabels': subscribeLabels } | |
| 168 | 177 | |
| 169 | 178 | async function start () { |
| 179 | if (labels) await labels.start() | |
| 170 | 180 | rooms = await membrane.communityRooms() |
| 171 | 181 | for (const f of state) { |
| 172 | 182 | f.unwatch = await rooms.watch(f.roomKey, async (members) => { |
| … | … | @@ -176,6 +186,9 @@ export function createLens ({ membrane, feeds = [], log = () => {}, debounceMs = |
| 176 | 186 | try { const c = await rooms.curation(f.roomKey); f.curation = (c && c.curator) || 'roster' } catch {} |
| 177 | 187 | if (f.curation !== 'roster') log({ msg: 'lens: curation is not roster — serving roster (only roster is implemented)', feed: f.id, curation: f.curation }) |
| 178 | 188 | log({ msg: 'lens: roster', feed: f.id, members: f.members.length, communityDid: f.communityDid }) |
| 189 | watchLabeler(f.communityDid) | |
| 190 | // Not awaited: a slow or absent labeler socket must not hold up the feed. It reports its own outcome. | |
| 191 | if (memberLabels) memberLabels.apply({ id: f.id, communityDid: f.communityDid, members: [...f.members] }).catch(() => {}) | |
| 179 | 192 | await reconcile() |
| 180 | 193 | }) |
| 181 | 194 | } |
| … | … | @@ -192,10 +205,14 @@ export function createLens ({ membrane, feeds = [], log = () => {}, debounceMs = |
| 192 | 205 | for (const path of Object.keys(WS)) membrane.extensions.ws.delete(path) |
| 193 | 206 | if (membrane.relay && membrane.relay.seq && membrane.relay.seq.core && membrane.relay.seq.core.off) membrane.relay.seq.core.off('append', onAppend) |
| 194 | 207 | for (const f of state) { if (f.unwatch) { try { await f.unwatch() } catch {} } } |
| 208 | if (memberLabels) { try { await memberLabels.close() } catch {} } | |
| 209 | if (labels) { try { await labels.stop() } catch {} } | |
| 195 | 210 | } |
| 196 | 211 | |
| 197 | 212 | function metrics () { |
| 198 | 213 | return { |
| 214 | labels: labels ? labels.metrics() : null, | |
| 215 | memberLabels: memberLabels ? memberLabels.metrics() : null, | |
| 199 | 216 | feeds: state.map((f) => { |
| 200 | 217 | const rows = rowsFor(f) |
| 201 | 218 | return { |
| … | … | @@ -0,0 +1,71 @@ |
| 1 | // The member projector — a community's roster → `<feed id>-member` labels (hyper-social #80, member labels). | |
| 2 | // | |
| 3 | // The gateway is keyless: it STATES intents on the co-located labeler DID's hpp control socket and hpp | |
| 4 | // signs them (hyper-proto-peer 0.3.128 label-intent, idempotent). On each roster change: assert every | |
| 5 | // member, negate every labelled DID that is no longer one. Negated, never deleted — a retraction is a | |
| 6 | // signed negation in atproto, so a subscriber that saw the label also sees it withdrawn. | |
| 7 | // | |
| 8 | // ⛔ An EMPTY roster negates nothing: at boot the room reads empty until it replicates, and "everyone | |
| 9 | // left" would strip every badge. ⛔ The socket must serve the community's DID (get-state), or a socket | |
| 10 | // configured for the wrong DID would write another account's labels. | |
| 11 | // | |
| 12 | // `connect` is injected: () => Promise<{ call(method, params), close() }> — the gateway passes a hyper-svc | |
| 13 | // UDS client to LENS_LABELER_SOCK. Runs are serialized per process; one connection, reopened on failure. | |
| 14 | ||
| 15 | export function createMemberLabels ({ connect, log = () => {} } = {}) { | |
| 16 | let client = null | |
| 17 | let queue = Promise.resolve() | |
| 18 | const feeds = new Map() // id -> { at, added, negated, members, error, skipped } | |
| 19 | ||
| 20 | const getClient = async () => { | |
| 21 | if (!client) client = await connect() | |
| 22 | return client | |
| 23 | } | |
| 24 | const drop = async () => { const c = client; client = null; if (c) { try { await c.close() } catch {} } } | |
| 25 | ||
| 26 | async function run (f) { | |
| 27 | const val = f.id + '-member' | |
| 28 | const out = { feed: f.id, val, added: 0, negated: 0 } | |
| 29 | if (!f.communityDid) return { ...out, skipped: 'no community DID yet' } | |
| 30 | if (!f.members || !f.members.length) return { ...out, skipped: 'empty roster' } | |
| 31 | try { | |
| 32 | const c = await getClient() | |
| 33 | const st = await c.call('get-state', {}) | |
| 34 | if (!st || st.did !== f.communityDid) return { ...out, skipped: `the labeler socket serves ${st && st.did}, not ${f.communityDid}` } | |
| 35 | for (const did of f.members) { | |
| 36 | const r = await c.call('label-intent', { uri: did, val }) | |
| 37 | if (r && r.changed) out.added++ | |
| 38 | } | |
| 39 | const members = new Set(f.members) | |
| 40 | const { intents = [] } = (await c.call('label-list', {})) || {} | |
| 41 | for (const i of intents) { | |
| 42 | if (i.val !== val || i.neg || members.has(i.uri)) continue | |
| 43 | const r = await c.call('label-intent', { uri: i.uri, val, neg: true }) | |
| 44 | if (r && r.changed) out.negated++ | |
| 45 | } | |
| 46 | return out | |
| 47 | } catch (err) { | |
| 48 | await drop() | |
| 49 | return { ...out, error: (err && err.message) || String(err) } | |
| 50 | } | |
| 51 | } | |
| 52 | ||
| 53 | // → the run's outcome; never throws (a down socket is a metric, not a crash). | |
| 54 | function apply (f) { | |
| 55 | const p = queue.then(() => run(f)) | |
| 56 | queue = p.catch(() => {}) | |
| 57 | return p.then((r) => { | |
| 58 | feeds.set(f.id, { at: Date.now(), members: (f.members || []).length, added: r.added, negated: r.negated, skipped: r.skipped || null, error: r.error || (r.skipped && r.skipped.startsWith('the labeler socket') ? r.skipped : null) }) | |
| 59 | if (r.error || r.added || r.negated || r.skipped) log({ msg: 'member labels', ...r }) | |
| 60 | return r | |
| 61 | }) | |
| 62 | } | |
| 63 | ||
| 64 | return { | |
| 65 | apply, | |
| 66 | metrics: () => ({ feeds: Object.fromEntries(feeds) }), | |
| 67 | close: async () => { await queue; await drop() } | |
| 68 | } | |
| 69 | } | |
| 70 | ||
| 71 | export default createMemberLabels |
| … | … | @@ -1,15 +1,18 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "favela-gateway", |
| 3 | "version": "0.5.0", | |
| 3 | "version": "0.6.0", | |
| 4 | 4 | "lockfileVersion": 3, |
| 5 | 5 | "requires": true, |
| 6 | 6 | "packages": { |
| 7 | 7 | "": { |
| 8 | 8 | "name": "favela-gateway", |
| 9 | "version": "0.5.0", | |
| 9 | "version": "0.6.0", | |
| 10 | 10 | "dependencies": { |
| 11 | "@ipld/dag-cbor": "^9.2.1", | |
| 11 | 12 | "ap-bridge": "file:../ap-bridge", |
| 12 | 13 | "atproto-bridge": "file:../atproto-bridge", |
| 14 | "atproto-identity": "file:../atproto-identity", | |
| 15 | "b4a": "^1.6.7", | |
| 13 | 16 | "bare-fs": "^4.8.1", |
| 14 | 17 | "bare-net": "^2.3.3", |
| 15 | 18 | "bare-os": "^3.9.3", |
| … | … | @@ -27,7 +30,7 @@ |
| 27 | 30 | } |
| 28 | 31 | }, |
| 29 | 32 | "../ap-bridge": { |
| 30 | "version": "0.10.16", | |
| 33 | "version": "0.10.17", | |
| 31 | 34 | "license": "MIT", |
| 32 | 35 | "dependencies": { |
| 33 | 36 | "atproto-did-peer": "file:../atproto-did-peer", |
| … | … | @@ -62,7 +65,7 @@ |
| 62 | 65 | } |
| 63 | 66 | }, |
| 64 | 67 | "../atproto-bridge": { |
| 65 | "version": "0.9.54", | |
| 68 | "version": "0.9.56", | |
| 66 | 69 | "hasInstallScript": true, |
| 67 | 70 | "dependencies": { |
| 68 | 71 | "@ipld/car": "^5.4.0", |
| … | … | @@ -109,6 +112,25 @@ |
| 109 | 112 | "@hyperswarm/testnet": "^3.0.0" |
| 110 | 113 | } |
| 111 | 114 | }, |
| 115 | "../atproto-identity": { | |
| 116 | "version": "0.0.4", | |
| 117 | "hasInstallScript": true, | |
| 118 | "dependencies": { | |
| 119 | "@ipld/dag-cbor": "^9.2.1", | |
| 120 | "@noble/curves": "^1.6.0", | |
| 121 | "@noble/hashes": "^1.5.0", | |
| 122 | "@scure/bip32": "^1.5.0", | |
| 123 | "@scure/bip39": "^1.4.0", | |
| 124 | "b4a": "^1.6.7", | |
| 125 | "hypercore-crypto": "^3.0.0", | |
| 126 | "keet-identity-key": "^3.2.0", | |
| 127 | "multiformats": "^13.3.0", | |
| 128 | "sodium-native": "^5.1.0" | |
| 129 | }, | |
| 130 | "devDependencies": { | |
| 131 | "brittle": "^4.0.2" | |
| 132 | } | |
| 133 | }, | |
| 112 | 134 | "../hyper-svc": { |
| 113 | 135 | "version": "0.1.9", |
| 114 | 136 | "dependencies": { |
| … | … | @@ -140,6 +162,16 @@ |
| 140 | 162 | "node": ">=18" |
| 141 | 163 | } |
| 142 | 164 | }, |
| 165 | "node_modules/@ipld/dag-cbor": { | |
| 166 | "version": "9.2.7", | |
| 167 | "resolved": "https://registry.npmjs.org/@ipld/dag-cbor/-/dag-cbor-9.2.7.tgz", | |
| 168 | "integrity": "sha512-ZmfXmElRWATr+hoUTSAOr6HUcjVhOcNHDqgczc76qte2DHHFEK0ZhNzUcdTDQhF/VSIvf2ioaRTRLWwLc83sNw==", | |
| 169 | "license": "Apache-2.0 OR MIT", | |
| 170 | "dependencies": { | |
| 171 | "cborg": "^5.0.1", | |
| 172 | "multiformats": "^13.1.0" | |
| 173 | } | |
| 174 | }, | |
| 143 | 175 | "node_modules/@jridgewell/resolve-uri": { |
| 144 | 176 | "version": "3.1.2", |
| 145 | 177 | "resolved": "https://registry.npmjs.org/@jridgewell/resolve-uri/-/resolve-uri-3.1.2.tgz", |
| … | … | @@ -183,6 +215,10 @@ |
| 183 | 215 | "resolved": "../atproto-bridge", |
| 184 | 216 | "link": true |
| 185 | 217 | }, |
| 218 | "node_modules/atproto-identity": { | |
| 219 | "resolved": "../atproto-identity", | |
| 220 | "link": true | |
| 221 | }, | |
| 186 | 222 | "node_modules/b4a": { |
| 187 | 223 | "version": "1.9.0", |
| 188 | 224 | "resolved": "https://registry.npmjs.org/b4a/-/b4a-1.9.0.tgz", |
| … | … | @@ -917,6 +953,15 @@ |
| 917 | 953 | "brittle-pear": "bin/pear.js" |
| 918 | 954 | } |
| 919 | 955 | }, |
| 956 | "node_modules/cborg": { | |
| 957 | "version": "5.1.11", | |
| 958 | "resolved": "https://registry.npmjs.org/cborg/-/cborg-5.1.11.tgz", | |
| 959 | "integrity": "sha512-oc6Pzg/gkTobxHZNgMmny+G99dOeBMbAmnGHcZWMKtolxZBIVwfi0Pj0khxEtNU8HMFdbT5sK0HmtgecDUPP0A==", | |
| 960 | "license": "Apache-2.0", | |
| 961 | "bin": { | |
| 962 | "cborg": "lib/bin.js" | |
| 963 | } | |
| 964 | }, | |
| 920 | 965 | "node_modules/compact-encoding": { |
| 921 | 966 | "version": "3.5.2", |
| 922 | 967 | "resolved": "https://registry.npmjs.org/compact-encoding/-/compact-encoding-3.5.2.tgz", |
| … | … | @@ -991,6 +1036,12 @@ |
| 991 | 1036 | "sodium-universal": "^5.0.1" |
| 992 | 1037 | } |
| 993 | 1038 | }, |
| 1039 | "node_modules/multiformats": { | |
| 1040 | "version": "13.4.2", | |
| 1041 | "resolved": "https://registry.npmjs.org/multiformats/-/multiformats-13.4.2.tgz", | |
| 1042 | "integrity": "sha512-eh6eHCrRi1+POZ3dA+Dq1C6jhP1GNtr9CRINMb67OKzqW9I5DUuZM/3jLPlzhgpGeiNUlEGEbkCYChXMCc/8DQ==", | |
| 1043 | "license": "Apache-2.0 OR MIT" | |
| 1044 | }, | |
| 994 | 1045 | "node_modules/nanoassert": { |
| 995 | 1046 | "version": "2.0.0", |
| 996 | 1047 | "resolved": "https://registry.npmjs.org/nanoassert/-/nanoassert-2.0.0.tgz", |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "favela-gateway", |
| 3 | "version": "0.5.0", | |
| 3 | "version": "0.6.0", | |
| 4 | 4 | "description": "The favela gateway: the bsky membrane (atproto-bridge) and the fediverse bridge (ap-bridge) in one process, keyless towards atproto. Later roles: the bsky lens and a blind seeder (hyper-social #80).", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "lib/gateway.js", |
| … | … | @@ -11,8 +11,11 @@ |
| 11 | 11 | "test": "brittle-bare test/*.test.js" |
| 12 | 12 | }, |
| 13 | 13 | "dependencies": { |
| 14 | "@ipld/dag-cbor": "^9.2.1", | |
| 14 | 15 | "ap-bridge": "file:../ap-bridge", |
| 15 | 16 | "atproto-bridge": "file:../atproto-bridge", |
| 17 | "atproto-identity": "file:../atproto-identity", | |
| 18 | "b4a": "^1.6.7", | |
| 16 | 19 | "bare-fs": "^4.8.1", |
| 17 | 20 | "bare-net": "^2.3.3", |
| 18 | 21 | "bare-os": "^3.9.3", |
| … | … | @@ -39,6 +42,7 @@ |
| 39 | 42 | "env": { |
| 40 | 43 | "GATEWAY_ROLES": "membrane,fediverse,lens", |
| 41 | 44 | "LENS_FEEDS": "loser-club=91ce4dbcb6b4fe0cfa489582cb9931923c05e0d14cf0a5c939d8334edefe21ed", |
| 45 | "LENS_LABELER_SOCK": "${HOME}/.config/hyper-proto-peer/data/dids/a63702bcab2089f6/control.sock", | |
| 42 | 46 | "GATEWAY_SOCK": "${HOME}/.config/favela-gateway/svc.sock", |
| 43 | 47 | "PDS_BIND": "0.0.0.0", |
| 44 | 48 | "PDS_PORT": "3399", |
| … | … | @@ -177,3 +177,20 @@ test('★ the public page is served on the membrane\'s own host only', async (t) |
| 177 | 177 | t.is(r.status, 200) |
| 178 | 178 | t.ok(r.body.startsWith('<!doctype html>') && r.body.includes('pds.test')) |
| 179 | 179 | }) |
| 180 | ||
| 181 | test('★ the lens role gets the labels module (with a state dir) and the member projector (with a labeler socket)', async (t) => { | |
| 182 | const g = await startGateway({ roles: ['membrane', 'lens'], membrane: membraneConfig(), lens: { feeds: [{ id: 'x', roomKey: 'a'.repeat(64) }], labelerSock: '/tmp/gw-no-such-labeler.sock' }, fetch: offlineFetch, stateDir: tmp('gwl-st-'), ...quiet }) | |
| 183 | t.teardown(() => stop(g)) | |
| 184 | t.ok(g.roles.lens, g.down.lens || 'lens up') | |
| 185 | const m = await g.handlers.metrics({}) | |
| 186 | t.ok(m.lens.labels, 'labels served from the gateway\'s own log') | |
| 187 | t.ok(m.lens.memberLabels, 'the member projector runs') | |
| 188 | }) | |
| 189 | ||
| 190 | test('without a state dir or a labeler socket, the lens runs as before', async (t) => { | |
| 191 | const g = await startGateway({ roles: ['membrane', 'lens'], membrane: membraneConfig(), lens: { feeds: [{ id: 'x', roomKey: 'a'.repeat(64) }] }, fetch: offlineFetch, ...quiet }) | |
| 192 | t.teardown(() => stop(g)) | |
| 193 | const m = await g.handlers.metrics({}) | |
| 194 | t.is(m.lens.labels, null) | |
| 195 | t.is(m.lens.memberLabels, null) | |
| 196 | }) |
| … | … | @@ -0,0 +1,183 @@ |
| 1 | // Labels — loser-club's signed labels, served to bsky by the gateway (member labels, hyper-social #80 step 4). | |
| 2 | // | |
| 3 | // hpp signs a labeler DID's labels and serves them as a label core beside the firehose (0.3.128); the | |
| 4 | // membrane's fronts replicate it (atproto-did-peer 0.2.15). The gateway serves them on bsky's two labeler | |
| 5 | // surfaces with favela-web's semantics (labeler-export.js, label-firehose.js): | |
| 6 | // queryLabels uriPatterns with a `*` suffix wildcard, sources default to every labeler we serve, | |
| 7 | // `sig` as { $bytes }, numeric cursor, limit ≤ 250 | |
| 8 | // subscribeLabels dag-cbor { t:'#labels', op:1 } ++ { seq, labels:[label] }, one label per frame, | |
| 9 | // replay from ?cursor then live tail; FutureCursor error frame past the head | |
| 10 | // | |
| 11 | // ⛔ Every label is VERIFIED against its src DID's key before the gateway keeps it — the gateway is keyless | |
| 12 | // and must not relay a label it cannot prove. And the seq bsky's cursor counts is the GATEWAY's own log | |
| 13 | // (a file in its state dir), not a front's core: co-owners order their cores differently and the served | |
| 14 | // front can switch, so a front's index would mean different labels on different days. | |
| 15 | ||
| 16 | import 'atproto-bridge/pds/lib/bare-globals.js' // FIRST: TextEncoder for dag-cbor under Bare | |
| 17 | import test from 'brittle' | |
| 18 | import fs from 'bare-fs' | |
| 19 | import b4a from 'b4a' | |
| 20 | import * as dagCbor from '@ipld/dag-cbor' | |
| 21 | import { keyFromPrivate } from 'atproto-identity/lib/atproto-identity.js' | |
| 22 | import { buildLabel, signLabel, encodeLabel } from 'atproto-identity/lib/labels.js' | |
| 23 | import { createLabels } from '../lib/labels.js' | |
| 24 | ||
| 25 | const LC = 'did:plc:wqnwulql7ndcz6xmtvaovl6u' | |
| 26 | const WJ = 'did:plc:wj62ebbusln6gxo5azo6ogzo' | |
| 27 | const E2 = 'did:plc:e2jevhuurrhlm36ufk7fklpg' | |
| 28 | const POST = 'at://' + WJ + '/app.bsky.feed.post/3mrhxnkbhv622' | |
| 29 | const lcKey = keyFromPrivate(new Uint8Array(32).fill(44)) | |
| 30 | const otherKey = keyFromPrivate(new Uint8Array(32).fill(45)) | |
| 31 | const tmp = () => { const d = '/tmp/gw-labels-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d } | |
| 32 | ||
| 33 | const signed = (uri, val, { neg, cts = '2026-10-14T00:00:00.000Z', key = lcKey, src = LC } = {}) => | |
| 34 | signLabel(buildLabel({ src, uri, val, cts, neg }), key.privateKey) | |
| 35 | ||
| 36 | // A front's label core: what atproto-did-peer exposes as `front.labels`. | |
| 37 | function fakeCore (labels = []) { | |
| 38 | const blocks = labels.map(encodeLabel) | |
| 39 | const fns = new Set() | |
| 40 | return { | |
| 41 | get length () { return blocks.length }, | |
| 42 | get: async (i) => blocks[i], | |
| 43 | on: (ev, fn) => { if (ev === 'append') fns.add(fn) }, | |
| 44 | off: (ev, fn) => fns.delete(fn), | |
| 45 | push: (l) => { blocks.push(encodeLabel(l)); for (const fn of fns) fn() } | |
| 46 | } | |
| 47 | } | |
| 48 | ||
| 49 | function fakePeers () { | |
| 50 | const fns = new Set() | |
| 51 | return { | |
| 52 | onLabels: (fn) => { fns.add(fn); return () => fns.delete(fn) }, | |
| 53 | tell: (did, front) => { for (const fn of fns) fn(did, front) } | |
| 54 | } | |
| 55 | } | |
| 56 | const front = (core, key = lcKey) => ({ labels: core, verifyKey: key.publicKey }) | |
| 57 | const settle = () => new Promise((resolve) => setTimeout(resolve, 20)) | |
| 58 | const query = (lab, q) => lab.xrpc['com.atproto.label.queryLabels']({ query: new URLSearchParams(q) }) | |
| 59 | ||
| 60 | async function started (t, dir = tmp()) { | |
| 61 | const peers = fakePeers() | |
| 62 | const lab = createLabels({ peers, stateDir: dir, log: () => {} }) | |
| 63 | await lab.start() | |
| 64 | t.teardown(() => lab.stop()) | |
| 65 | return { peers, lab, dir } | |
| 66 | } | |
| 67 | ||
| 68 | test('★ a front\'s verified labels are served by queryLabels, in favela-web\'s wire shape', async (t) => { | |
| 69 | const { peers, lab } = await started(t) | |
| 70 | peers.tell(LC, front(fakeCore([signed(WJ, 'loser-club-member'), signed(POST, 'spam')]))) | |
| 71 | await settle() | |
| 72 | const r = await query(lab, [['uriPatterns', '*']]) | |
| 73 | t.is(r.status, 200) | |
| 74 | t.is(r.body.labels.length, 2) | |
| 75 | const m = r.body.labels.find((l) => l.val === 'loser-club-member') | |
| 76 | t.is(m.src, LC) | |
| 77 | t.is(m.uri, WJ) | |
| 78 | t.ok(m.sig && typeof m.sig.$bytes === 'string', 'sig as { $bytes } (base64)') | |
| 79 | t.absent('cursor' in r.body, 'no cursor when the page is the whole set') | |
| 80 | }) | |
| 81 | ||
| 82 | test('uriPatterns: exact and `*` suffix; sources filter; numeric cursor', async (t) => { | |
| 83 | const { peers, lab } = await started(t) | |
| 84 | peers.tell(LC, front(fakeCore([signed(WJ, 'loser-club-member'), signed(E2, 'loser-club-member'), signed(POST, 'spam')]))) | |
| 85 | await settle() | |
| 86 | t.alike((await query(lab, [['uriPatterns', WJ]])).body.labels.map((l) => l.uri), [WJ], 'exact') | |
| 87 | t.is((await query(lab, [['uriPatterns', 'at://' + WJ + '/*']])).body.labels.length, 1, 'suffix wildcard') | |
| 88 | t.is((await query(lab, [['uriPatterns', '*'], ['sources', 'did:plc:someoneelse']])).body.labels.length, 0, 'sources filter') | |
| 89 | const p1 = await query(lab, [['uriPatterns', '*'], ['limit', '2']]) | |
| 90 | t.is(p1.body.labels.length, 2) | |
| 91 | t.is(p1.body.cursor, '2') | |
| 92 | const p2 = await query(lab, [['uriPatterns', '*'], ['limit', '2'], ['cursor', p1.body.cursor]]) | |
| 93 | t.is(p2.body.labels.length, 1) | |
| 94 | t.absent('cursor' in p2.body) | |
| 95 | }) | |
| 96 | ||
| 97 | test('⛔ a label that does not verify against its src DID\'s key is never served', async (t) => { | |
| 98 | const { peers, lab } = await started(t) | |
| 99 | peers.tell(LC, front(fakeCore([ | |
| 100 | signed(WJ, 'loser-club-member', { key: otherKey }), // forged: signed by another key | |
| 101 | signed(WJ, 'spam', { src: WJ }), // a label naming a different src than the DID that served it | |
| 102 | signed(E2, 'loser-club-member') | |
| 103 | ]))) | |
| 104 | await settle() | |
| 105 | const r = await query(lab, [['uriPatterns', '*']]) | |
| 106 | t.alike(r.body.labels.map((l) => l.uri), [E2], 'only the verified one') | |
| 107 | t.is(lab.metrics().rejected, 2, 'and the two rejects are counted') | |
| 108 | }) | |
| 109 | ||
| 110 | test('a new label on the front is picked up live; the same label from two co-owners is kept once', async (t) => { | |
| 111 | const { peers, lab } = await started(t) | |
| 112 | const a = fakeCore([signed(WJ, 'loser-club-member')]) | |
| 113 | const b = fakeCore([signed(WJ, 'loser-club-member')]) // a co-owner's core holds the same signed label | |
| 114 | peers.tell(LC, front(a)) | |
| 115 | peers.tell(LC, front(b)) | |
| 116 | await settle() | |
| 117 | t.is((await query(lab, [['uriPatterns', '*']])).body.labels.length, 1, 'deduplicated by bytes') | |
| 118 | a.push(signed(E2, 'loser-club-member')) | |
| 119 | await settle() | |
| 120 | t.is((await query(lab, [['uriPatterns', '*']])).body.labels.length, 2, 'the append arrived') | |
| 121 | }) | |
| 122 | ||
| 123 | test('★ the log persists: the same seq after a restart, whatever order the fronts arrive in', async (t) => { | |
| 124 | const dir = tmp() | |
| 125 | { | |
| 126 | const peers = fakePeers() | |
| 127 | const lab = createLabels({ peers, stateDir: dir, log: () => {} }) | |
| 128 | await lab.start() | |
| 129 | peers.tell(LC, front(fakeCore([signed(WJ, 'loser-club-member'), signed(E2, 'loser-club-member')]))) | |
| 130 | await settle() | |
| 131 | await lab.stop() | |
| 132 | } | |
| 133 | const { peers, lab } = await started(t, dir) | |
| 134 | t.is((await query(lab, [['uriPatterns', '*']])).body.labels.length, 2, 'loaded from the state dir before any front') | |
| 135 | peers.tell(LC, front(fakeCore([signed(E2, 'loser-club-member'), signed(WJ, 'loser-club-member')]))) | |
| 136 | await settle() | |
| 137 | const uris = (await query(lab, [['uriPatterns', '*']])).body.labels.map((l) => l.uri) | |
| 138 | t.alike(uris, [WJ, E2], 'same order as before; nothing duplicated') | |
| 139 | }) | |
| 140 | ||
| 141 | // subscribeLabels: a fake bare-ws socket. | |
| 142 | function fakeWs () { | |
| 143 | const frames = [] | |
| 144 | const fns = {} | |
| 145 | return { frames, write: (b) => frames.push(b), on: (ev, fn) => { fns[ev] = fn }, end () {}, destroy () {}, close: () => fns.close && fns.close() } | |
| 146 | } | |
| 147 | const HEADER = dagCbor.encode({ t: '#labels', op: 1 }) | |
| 148 | const ERR = dagCbor.encode({ op: -1 }) | |
| 149 | const body = (frame, head) => dagCbor.decode(frame.subarray(head.length)) | |
| 150 | ||
| 151 | test('★ subscribeLabels: replay from the cursor, then the live tail, one label per frame', async (t) => { | |
| 152 | const { peers, lab } = await started(t) | |
| 153 | const core = fakeCore([signed(WJ, 'loser-club-member'), signed(E2, 'loser-club-member')]) | |
| 154 | peers.tell(LC, front(core)) | |
| 155 | await settle() | |
| 156 | const ws = fakeWs() | |
| 157 | lab.ws['/xrpc/com.atproto.label.subscribeLabels'](ws, new URL('http://x/xrpc/com.atproto.label.subscribeLabels?cursor=1')) | |
| 158 | await settle() | |
| 159 | t.is(ws.frames.length, 1, 'cursor=1: only seq 2 is replayed') | |
| 160 | t.ok(b4a.equals(ws.frames[0].subarray(0, HEADER.length), HEADER), '#labels header') | |
| 161 | const f = body(ws.frames[0], HEADER) | |
| 162 | t.is(f.seq, 2) | |
| 163 | t.is(f.labels.length, 1) | |
| 164 | t.is(f.labels[0].uri, E2) | |
| 165 | t.ok(f.labels[0].sig instanceof Uint8Array, 'sig as raw bytes on the CBOR wire') | |
| 166 | core.push(signed(POST, 'spam')) | |
| 167 | await settle() | |
| 168 | t.is(ws.frames.length, 2, 'live tail') | |
| 169 | t.is(body(ws.frames[1], HEADER).seq, 3) | |
| 170 | ws.close() | |
| 171 | core.push(signed(E2, 'spam')) | |
| 172 | await settle() | |
| 173 | t.is(ws.frames.length, 2, 'nothing after the client left') | |
| 174 | }) | |
| 175 | ||
| 176 | test('subscribeLabels: a cursor past the head gets a FutureCursor error frame first', async (t) => { | |
| 177 | const { lab } = await started(t) | |
| 178 | const ws = fakeWs() | |
| 179 | lab.ws['/xrpc/com.atproto.label.subscribeLabels'](ws, new URL('http://x/xrpc/com.atproto.label.subscribeLabels?cursor=9')) | |
| 180 | await settle() | |
| 181 | t.ok(b4a.equals(ws.frames[0].subarray(0, ERR.length), ERR)) | |
| 182 | t.is(body(ws.frames[0], ERR).error, 'FutureCursor') | |
| 183 | }) |
| … | … | @@ -163,3 +163,47 @@ test('stop removes every registered method — a stopped lens serves nothing', a |
| 163 | 163 | t.is(m.extensions.xrpc.size, 0) |
| 164 | 164 | t.is(m.extensions.ws.size, 0) |
| 165 | 165 | }) |
| 166 | ||
| 167 | // ── labels (member labels, #80 step 4): the lens hosts the label surfaces and drives the projector ─────── | |
| 168 | function fakeLabels () { | |
| 169 | const s = { watched: [], started: 0, stopped: 0 } | |
| 170 | return { | |
| 171 | s, | |
| 172 | start: async () => { s.started++ }, | |
| 173 | stop: async () => { s.stopped++ }, | |
| 174 | watch: (did, front) => { s.watched.push({ did, front }) }, | |
| 175 | metrics: () => ({ kept: 3 }), | |
| 176 | xrpc: { 'com.atproto.label.queryLabels': async () => ({ status: 200, body: { labels: ['real'] } }) }, | |
| 177 | ws: { '/xrpc/com.atproto.label.subscribeLabels': () => {} } | |
| 178 | } | |
| 179 | } | |
| 180 | function fakeProjector () { | |
| 181 | const applied = [] | |
| 182 | return { applied, apply: async (f) => { applied.push(f); return {} }, metrics: () => ({ feeds: { 'loser-club': { added: 2 } } }), close: async () => {} } | |
| 183 | } | |
| 184 | ||
| 185 | test('★ with labels injected: the real queryLabels is served, the community DID\'s front is watched', async (t) => { | |
| 186 | const m = fakeMembrane({ records: baseRecords(), revs: { [WJ]: 'r1', [E2]: 'r1', [LC]: 'r1' } }) | |
| 187 | const labels = fakeLabels() | |
| 188 | const lens = createLens({ membrane: m, feeds: [{ id: 'loser-club', roomKey: ROOM }], log: () => {}, debounceMs: 5, labels }) | |
| 189 | await lens.start() | |
| 190 | t.teardown(() => lens.stop()) | |
| 191 | t.is(labels.s.started, 1) | |
| 192 | const r = await m.extensions.xrpc.get('com.atproto.label.queryLabels')({ query: new URLSearchParams() }) | |
| 193 | t.alike(r.body.labels, ['real'], 'the labels module answers, not the empty placeholder') | |
| 194 | await new Promise((resolve) => setTimeout(resolve, 20)) | |
| 195 | t.ok(labels.s.watched.some((w) => w.did === LC && w.front), 'the community DID\'s front is handed to the labels module') | |
| 196 | t.alike(lens.metrics().labels, { kept: 3 }) | |
| 197 | }) | |
| 198 | ||
| 199 | test('★ with a projector: every roster change is applied, with the community DID', async (t) => { | |
| 200 | const m = fakeMembrane({ records: baseRecords(), revs: { [WJ]: 'r1', [E2]: 'r1', [LC]: 'r1' } }) | |
| 201 | const memberLabels = fakeProjector() | |
| 202 | const lens = createLens({ membrane: m, feeds: [{ id: 'loser-club', roomKey: ROOM }], log: () => {}, debounceMs: 5, memberLabels }) | |
| 203 | await lens.start() | |
| 204 | t.teardown(() => lens.stop()) | |
| 205 | t.alike(memberLabels.applied[0], { id: 'loser-club', communityDid: LC, members: [WJ, E2] }) | |
| 206 | await m.state.watchers[0]([WJ]) | |
| 207 | t.alike(memberLabels.applied[memberLabels.applied.length - 1].members, [WJ], 'a member leaving is applied too') | |
| 208 | t.alike(lens.metrics().memberLabels, { feeds: { 'loser-club': { added: 2 } } }) | |
| 209 | }) |
| … | … | @@ -0,0 +1,99 @@ |
| 1 | // The member projector: a community's roster → `<feed>-member` labels (hyper-social #80 step 4). | |
| 2 | // | |
| 3 | // The gateway stays keyless: it STATES intents on the co-located labeler DID's hpp socket (label-intent, | |
| 4 | // idempotent) and hpp signs them. On every roster change it asserts each member and negates each labelled | |
| 5 | // DID that is no longer a member. | |
| 6 | // | |
| 7 | // ⛔ Two guards, each against a real way to strip every badge at once: | |
| 8 | // - an EMPTY roster negates nothing. At boot the room has not replicated yet and reads empty; treating | |
| 9 | // that as "everyone left" would negate every member. | |
| 10 | // - the socket must serve the community's DID (get-state). A socket configured for the wrong DID would | |
| 11 | // write another account's labels. | |
| 12 | ||
| 13 | import test from 'brittle' | |
| 14 | import { createMemberLabels } from '../lib/member-labels.js' | |
| 15 | ||
| 16 | const LC = 'did:plc:wqnwulql7ndcz6xmtvaovl6u' | |
| 17 | const WJ = 'did:plc:wj62ebbusln6gxo5azo6ogzo' | |
| 18 | const E2 = 'did:plc:e2jevhuurrhlm36ufk7fklpg' | |
| 19 | const OLD = 'did:plc:oldmemberoldmemberoldmem' | |
| 20 | ||
| 21 | // A fake hpp socket holding label intents, as label-intent/label-list answer them. | |
| 22 | function fakeHpp ({ did = LC, intents = [] } = {}) { | |
| 23 | const map = new Map(intents.map((i) => [i.uri + '|' + i.val, i])) | |
| 24 | const calls = [] | |
| 25 | return { | |
| 26 | calls, | |
| 27 | map, | |
| 28 | connect: async () => ({ | |
| 29 | call: async (method, p = {}) => { | |
| 30 | calls.push([method, p]) | |
| 31 | if (method === 'get-state') return { did } | |
| 32 | if (method === 'label-list') return { intents: [...map.values()], signed: map.size } | |
| 33 | if (method === 'label-intent') { | |
| 34 | const k = p.uri + '|' + p.val | |
| 35 | const cur = map.get(k) | |
| 36 | if (cur ? !!cur.neg === !!p.neg : p.neg) return { intent: cur || null, changed: false } | |
| 37 | map.set(k, { uri: p.uri, val: p.val, neg: !!p.neg, cts: 'now' }) | |
| 38 | return { changed: true } | |
| 39 | } | |
| 40 | throw new Error('unknown method: ' + method) | |
| 41 | }, | |
| 42 | close: async () => {} | |
| 43 | }) | |
| 44 | } | |
| 45 | } | |
| 46 | ||
| 47 | const feed = (members, communityDid = LC) => ({ id: 'loser-club', communityDid, members }) | |
| 48 | const active = (hpp) => [...hpp.map.values()].filter((i) => !i.neg && i.val === 'loser-club-member').map((i) => i.uri).sort() | |
| 49 | ||
| 50 | test('★ each member gets `<feed>-member`; a member who left is negated', async (t) => { | |
| 51 | const hpp = fakeHpp({ intents: [{ uri: OLD, val: 'loser-club-member', neg: false }, { uri: WJ, val: 'spam', neg: false }] }) | |
| 52 | const p = createMemberLabels({ connect: hpp.connect, log: () => {} }) | |
| 53 | const r = await p.apply(feed([WJ, E2])) | |
| 54 | t.alike(active(hpp), [E2, WJ].sort(), 'members labelled, the old member not') | |
| 55 | t.is(hpp.map.get(OLD + '|loser-club-member').neg, true, 'the old member is negated, not deleted') | |
| 56 | t.is(hpp.map.get(WJ + '|spam').neg, false, 'labels of other values are not touched') | |
| 57 | t.alike({ added: r.added, negated: r.negated }, { added: 2, negated: 1 }) | |
| 58 | }) | |
| 59 | ||
| 60 | test('idempotent: the same roster again writes nothing', async (t) => { | |
| 61 | const hpp = fakeHpp() | |
| 62 | const p = createMemberLabels({ connect: hpp.connect, log: () => {} }) | |
| 63 | await p.apply(feed([WJ, E2])) | |
| 64 | const r = await p.apply(feed([WJ, E2])) | |
| 65 | t.alike({ added: r.added, negated: r.negated }, { added: 0, negated: 0 }) | |
| 66 | }) | |
| 67 | ||
| 68 | test('⛔ an empty roster negates nothing (the room has not replicated yet)', async (t) => { | |
| 69 | const hpp = fakeHpp({ intents: [{ uri: WJ, val: 'loser-club-member', neg: false }] }) | |
| 70 | const p = createMemberLabels({ connect: hpp.connect, log: () => {} }) | |
| 71 | const r = await p.apply(feed([])) | |
| 72 | t.is(r.skipped, 'empty roster') | |
| 73 | t.alike(active(hpp), [WJ]) | |
| 74 | t.absent(hpp.calls.some(([m]) => m === 'label-intent'), 'no write at all') | |
| 75 | }) | |
| 76 | ||
| 77 | test('⛔ a socket serving a different DID is refused — no write', async (t) => { | |
| 78 | const hpp = fakeHpp({ did: 'did:plc:someoneelse' }) | |
| 79 | const p = createMemberLabels({ connect: hpp.connect, log: () => {} }) | |
| 80 | const r = await p.apply(feed([WJ])) | |
| 81 | t.ok(/serves did:plc:someoneelse/.test(r.skipped), 'skipped, naming the DID it found') | |
| 82 | t.absent(hpp.calls.some(([m]) => m === 'label-intent')) | |
| 83 | t.ok(/someoneelse/.test(p.metrics().feeds['loser-club'].error), 'and the metrics say why') | |
| 84 | }) | |
| 85 | ||
| 86 | test('no community DID yet: skipped, nothing called', async (t) => { | |
| 87 | const hpp = fakeHpp() | |
| 88 | const p = createMemberLabels({ connect: hpp.connect, log: () => {} }) | |
| 89 | const r = await p.apply(feed([WJ], null)) | |
| 90 | t.ok(r.skipped) | |
| 91 | t.is(hpp.calls.length, 0) | |
| 92 | }) | |
| 93 | ||
| 94 | test('the socket being down is reported, not thrown', async (t) => { | |
| 95 | const p = createMemberLabels({ connect: async () => { throw new Error('connect ENOENT') }, log: () => {} }) | |
| 96 | const r = await p.apply(feed([WJ])) | |
| 97 | t.ok(/ENOENT/.test(r.error)) | |
| 98 | t.ok(/ENOENT/.test(p.metrics().feeds['loser-club'].error)) | |
| 99 | }) |