| … | … | @@ -0,0 +1,161 @@ |
| 1 | // bsky-ingest.js — bsky interactions on BOUND DIDs' posts → each identity's observation feed (hyper-social #82, I2b). | |
| 2 | // | |
| 3 | // The wiring around three pure pieces: bsky-bind.js (is this fronted DID a favela identity's, by both sides' word), | |
| 4 | // spacedust.js (the filtered link subscription), bsky-inbound.js (a link → an observation). The observation feed is | |
| 5 | // the SAME one the fediverse writes to (I1), reached through the fediverse role's `observationsForFeed(feedKey)`, so | |
| 6 | // the node renders bsky arrivals with no change at all. | |
| 7 | // | |
| 8 | // ⛔ ONLY BOUND DIDs ARE SUBSCRIBED, and never an empty filter (that is the whole network; spacedust.js refuses too). | |
| 9 | // ⛔ EVENT-DRIVEN: drained on arrival (onEnqueue), re-bound when the caller says a DID's repo moved — no poll. | |
| 10 | // ⚠ A reply's text is fetched from the public AppView for DISPLAY only; it is never a canonical record here. | |
| 11 | ||
| 12 | import { SpacedustClient } from './spacedust.js' | |
| 13 | import { bindDid } from './bsky-bind.js' | |
| 14 | import { linkToObservation, WANTED } from './bsky-inbound.js' | |
| 15 | import { tidFor } from 'favela-bridge/lib/tid.js' | |
| 16 | ||
| 17 | const DEFAULT_APPVIEW = 'https://public.api.bsky.app' | |
| 18 | ||
| 19 | // Which DID a link concerns: a follow's subject IS the DID; anything else names one of its records. | |
| 20 | function subjectDid (link) { | |
| 21 | const s = link && link.subject | |
| 22 | if (typeof s !== 'string') return null | |
| 23 | if (s.startsWith('did:')) return s | |
| 24 | if (!s.startsWith('at://')) return null | |
| 25 | return s.slice(5).split('/')[0] || null | |
| 26 | } | |
| 27 | ||
| 28 | export class BskyIngest { | |
| 29 | constructor ({ membrane, fediverse, fetch = null, createSocket = null, log = () => {}, appview = DEFAULT_APPVIEW, frontTimeoutMs = 20000 } = {}) { | |
| 30 | this.membrane = membrane | |
| 31 | this.fediverse = fediverse | |
| 32 | this.fetch = fetch | |
| 33 | this.createSocket = createSocket | |
| 34 | this.log = log | |
| 35 | this.appview = appview | |
| 36 | this.frontTimeoutMs = frontTimeoutMs | |
| 37 | this._bound = new Map() // did -> feed | |
| 38 | this._refused = {} | |
| 39 | this._names = new Map() // feed -> Map(tid -> favela name) | |
| 40 | this._client = null | |
| 41 | this._draining = false | |
| 42 | this._again = false | |
| 43 | this._rebinding = null | |
| 44 | this._stopped = false | |
| 45 | } | |
| 46 | ||
| 47 | async start () { await this.rebind(); return this } | |
| 48 | ||
| 49 | /** Re-evaluate every fronted DID's binding and (re)subscribe for exactly the bound ones. Coalesced. */ | |
| 50 | rebind () { | |
| 51 | if (this._rebinding) return this._rebinding | |
| 52 | this._rebinding = this._rebind().finally(() => { this._rebinding = null }) | |
| 53 | return this._rebinding | |
| 54 | } | |
| 55 | ||
| 56 | async _rebind () { | |
| 57 | if (this._stopped) return | |
| 58 | let dids = [] | |
| 59 | try { dids = (this.membrane.metrics().dids || []).map((d) => d.did).filter(Boolean) } catch {} | |
| 60 | const bound = new Map() | |
| 61 | const refused = {} | |
| 62 | for (const did of dids) { | |
| 63 | let front = null | |
| 64 | try { front = await this.membrane.peers.front(did, { timeout: this.frontTimeoutMs }) } catch { refused[did] = 'not fronted right now'; continue } | |
| 65 | const r = await bindDid(did, { front, feedOf: async (f) => this.fediverse.heldFeed(f) }) | |
| 66 | if (!r.feed) { refused[did] = r.refused; continue } | |
| 67 | if (!this.fediverse.observationsForFeed(r.feed)) { refused[did] = 'no observation feed for that identity on this gateway'; continue } | |
| 68 | bound.set(did, r.feed) | |
| 69 | } | |
| 70 | const changed = [...bound.keys()].sort().join(',') !== [...this._bound.keys()].sort().join(',') | |
| 71 | this._bound = bound | |
| 72 | this._refused = refused | |
| 73 | if (changed) this.log({ msg: 'bsky ingest: bound', dids: [...bound.keys()], refused }) | |
| 74 | this._subscribe() | |
| 75 | } | |
| 76 | ||
| 77 | _subscribe () { | |
| 78 | const subjectDids = [...this._bound.keys()] | |
| 79 | if (!subjectDids.length) { | |
| 80 | if (this._client) { try { this._client.close() } catch {} ; this._client = null } | |
| 81 | return | |
| 82 | } | |
| 83 | const filter = { subjectDids, sources: WANTED } | |
| 84 | if (this._client) { this._client.updateFilter(filter); return } | |
| 85 | if (!this.createSocket) { this.log({ msg: 'bsky ingest: no socket factory — not subscribing' }); return } | |
| 86 | this._client = new SpacedustClient(filter, { createSocket: this.createSocket, onEnqueue: () => { this._drain() } }).start() | |
| 87 | } | |
| 88 | ||
| 89 | async _postName (feed, rkey, { refresh = false } = {}) { | |
| 90 | let idx = this._names.get(feed) | |
| 91 | if (!idx || refresh) { | |
| 92 | idx = new Map() | |
| 93 | try { | |
| 94 | const held = this.fediverse.heldFeed(feed) | |
| 95 | const { rows } = held ? await held.ownPosts() : { rows: [] } | |
| 96 | for (const row of rows || []) { if (row && row.name) idx.set(tidFor(row.name), row.name) } | |
| 97 | } catch {} | |
| 98 | this._names.set(feed, idx) | |
| 99 | } | |
| 100 | return idx.get(rkey) || null | |
| 101 | } | |
| 102 | ||
| 103 | async _replyText (uri) { | |
| 104 | if (!this.fetch) return '' | |
| 105 | try { | |
| 106 | const res = await this.fetch(this.appview + '/xrpc/app.bsky.feed.getPosts?uris=' + encodeURIComponent(uri)) | |
| 107 | if (!res || !res.ok) return '' | |
| 108 | const body = await res.json() | |
| 109 | const p = body && Array.isArray(body.posts) ? body.posts[0] : null | |
| 110 | return (p && p.record && typeof p.record.text === 'string') ? p.record.text.slice(0, 4000) : '' | |
| 111 | } catch { return '' } | |
| 112 | } | |
| 113 | ||
| 114 | async _drain () { | |
| 115 | if (this._draining) { this._again = true; return } | |
| 116 | this._draining = true | |
| 117 | try { | |
| 118 | do { | |
| 119 | this._again = false | |
| 120 | const links = this._client ? this._client.drain() : [] | |
| 121 | for (const link of links) { | |
| 122 | try { await this._one(link) } catch (err) { this.log({ msg: 'bsky ingest: link failed', error: (err && err.message) || String(err) }) } | |
| 123 | } | |
| 124 | } while (this._again && !this._stopped) | |
| 125 | } finally { this._draining = false } | |
| 126 | } | |
| 127 | ||
| 128 | async _one (link) { | |
| 129 | const did = subjectDid(link) | |
| 130 | const feed = did ? this._bound.get(did) : null | |
| 131 | if (!feed) return | |
| 132 | const sink = this.fediverse.observationsForFeed(feed) | |
| 133 | if (!sink) return | |
| 134 | // A post made since the index was built is unknown once; rebuild once before giving up on it. | |
| 135 | const rkey = typeof link.subject === 'string' && link.subject.startsWith('at://') ? link.subject.split('/').pop() : null | |
| 136 | if (rkey && !(await this._postName(feed, rkey))) await this._postName(feed, rkey, { refresh: true }) | |
| 137 | const names = this._names.get(feed) || new Map() | |
| 138 | const r = linkToObservation(link, { did, feed, postName: (k) => names.get(k) || null }) | |
| 139 | if (!r) return | |
| 140 | if (r.op === 'remove') { await sink.remove(r.activity); return } | |
| 141 | const obs = { ...r.obs, at: Date.now() } | |
| 142 | if (obs.kind === 'reply') obs.body = await this._replyText(obs.activity) | |
| 143 | await sink.note(obs) | |
| 144 | } | |
| 145 | ||
| 146 | metrics () { | |
| 147 | return { | |
| 148 | bound: [...this._bound.entries()].map(([did, feed]) => ({ did, feed })), | |
| 149 | refused: { ...this._refused }, | |
| 150 | subscribed: !!this._client, | |
| 151 | stats: this._client ? { ...this._client.stats, pending: this._client.pending } : null | |
| 152 | } | |
| 153 | } | |
| 154 | ||
| 155 | stop () { | |
| 156 | this._stopped = true | |
| 157 | if (this._client) { try { this._client.close() } catch {} ; this._client = null } | |
| 158 | } | |
| 159 | } | |
| 160 | ||
| 161 | export default { BskyIngest } |
| … | … | @@ -42,7 +42,33 @@ const ROLES = { |
| 42 | 42 | const memberLabels = cfg && cfg.labelerSock ? createMemberLabels({ connect: () => Client.connectUds(cfg.labelerSock), log: opts.log }) : null |
| 43 | 43 | const lens = createLens({ membrane: running.membrane, feeds: (cfg && cfg.feeds) || [], log: opts.log, labels, memberLabels }) |
| 44 | 44 | await lens.start() |
| 45 | return lens | |
| 45 | // ⛔ bsky → favela notifications (hyper-social #82, I2b): interactions on BOUND DIDs' posts go into the identity's | |
| 46 | // observation feed — the one the fediverse role already writes for it. Needs that role; never blocks the lens. | |
| 47 | let ingest = null | |
| 48 | let rebindTimer = null | |
| 49 | let onAppend = null | |
| 50 | const relayCore = running.membrane.relay && running.membrane.relay.seq && running.membrane.relay.seq.core | |
| 51 | if (running.fediverse && typeof running.fediverse.observationsForFeed === 'function' && !(cfg && cfg.bskyInbound === false)) { | |
| 52 | const { BskyIngest } = await import('./bsky-ingest.js') | |
| 53 | const { createBareSocket } = await import('./spacedust-socket.js') | |
| 54 | ingest = new BskyIngest({ membrane: running.membrane, fediverse: running.fediverse, fetch: opts.fetch || globalThis.fetch || null, createSocket: createBareSocket, log: opts.log }) | |
| 55 | ingest.start().catch((err) => opts.log({ msg: 'bsky ingest did not start', error: (err && err.message) || String(err) })) | |
| 56 | // Re-bind when any fronted repo commits: a new net.favela.feed/self, or a new DID. Debounced, no poll. | |
| 57 | if (relayCore) { | |
| 58 | onAppend = () => { clearTimeout(rebindTimer); rebindTimer = setTimeout(() => { ingest.rebind().catch(() => {}) }, 5000) } | |
| 59 | relayCore.on('append', onAppend) | |
| 60 | } | |
| 61 | } | |
| 62 | return { | |
| 63 | ...lens, | |
| 64 | metrics: () => ({ ...lens.metrics(), bskyIngest: ingest ? ingest.metrics() : null }), | |
| 65 | stop: async () => { | |
| 66 | clearTimeout(rebindTimer) | |
| 67 | if (relayCore && onAppend && relayCore.off) { try { relayCore.off('append', onAppend) } catch {} } | |
| 68 | if (ingest) { try { ingest.stop() } catch {} } | |
| 69 | await lens.stop() | |
| 70 | } | |
| 71 | } | |
| 46 | 72 | } |
| 47 | 73 | } |
| 48 | 74 | |
| … | … | @@ -70,7 +96,7 @@ export async function startGateway ({ roles, log = defaultLog, bootstrap = null, |
| 70 | 96 | try { |
| 71 | 97 | // A config the entry could not load arrives as an Error: that role is down, for that reason. |
| 72 | 98 | if (configs[name] instanceof Error) throw configs[name] |
| 73 | running[name] = await ROLES[name](configs[name], { log: roleLog, bootstrap, version, adminPages, stateDir }, running) | |
| 99 | running[name] = await ROLES[name](configs[name], { log: roleLog, bootstrap, version, adminPages, stateDir, fetch }, running) | |
| 74 | 100 | log({ msg: 'role up', role: name }) |
| 75 | 101 | } catch (err) { |
| 76 | 102 | down[name] = (err && err.message) || String(err) |
| … | … | @@ -0,0 +1,22 @@ |
| 1 | // spacedust-socket.js — the Bare-only half of the spacedust client. | |
| 2 | // | |
| 3 | // This file exists so `spacedust.js` can stay runtime-agnostic and unit-testable under plain Node. | |
| 4 | // `bare-ws` transitively pulls native addons (bare-dns), so any module that imports it is Bare-only. | |
| 5 | // Keeping that import here means the protocol logic — URL building, event parsing, filter framing, | |
| 6 | // queue behaviour — is testable with `node`, and only the socket itself needs `bare`. | |
| 7 | // | |
| 8 | // bare-ws details that matter and were verified against its source: | |
| 9 | // - `new Socket(url)` accepts a URL string and derives host/path/port/secure from it, so `wss://` | |
| 10 | // with a query string works (lib/socket.js:18-21, :77 routes through bare-https). | |
| 11 | // - It auto-replies PONG with the ECHOED payload on PING (lib/socket.js:163-165), which is exactly | |
| 12 | // what spacedust's 30s keepalive verifies. Nothing to implement. | |
| 13 | // - It is a non-objectMode Duplex that pushes one Buffer per message (:195), so messages arrive as | |
| 14 | // 'data' events — see the framing note in spacedust.js. | |
| 15 | ||
| 16 | import bareWs from 'bare-ws' | |
| 17 | import { USER_AGENT } from './spacedust.js' | |
| 18 | ||
| 19 | // The production socket factory. Pass as `createSocket` to SpacedustClient. | |
| 20 | export function createBareSocket (url) { | |
| 21 | return new bareWs.Socket(url, { headers: { 'User-Agent': USER_AGENT } }) | |
| 22 | } |
| … | … | @@ -0,0 +1,366 @@ |
| 1 | // spacedust.js — live inbound link subscription. Copied from favela-web (being retired, hyper-social #77) for the | |
| 2 | // gateway's bsky ingest (#82 I2b); the one change is `onEnqueue`, so the drain is driven by arrivals, not a clock. | |
| 3 | // | |
| 4 | // WHAT IT IS. Spacedust (microcosm.blue) watches the global firehose, extracts every LINK it sees, and | |
| 5 | // re-emits them over a filtered WebSocket. We subscribe with our roster's DIDs as the subjects, so we | |
| 6 | // receive exactly the interactions that touch our own content — replies, quotes, likes, follows — and | |
| 7 | // nothing else. This is NOT the firehose: the filtering happens server-side, and the caps (10k subject | |
| 8 | // DIDs, 1k sources) are far above what a community needs. | |
| 9 | // | |
| 10 | // WHAT IT IS NOT. It does not fetch, verify, or store anything. Spacedust deliberately ships only a | |
| 11 | // LOCATOR — `{did, collection, rkey}` reconstructed from `source_record`, plus the subject it points at. | |
| 12 | // Their own source says so ("include the record too? ... no, not for now"), and that is the right call: | |
| 13 | // an unsigned record from a third-party index is not something we retain. This module emits pointers; | |
| 14 | // §7.6's pipeline (resolveMiniDoc -> getRecord -> verify -> OverlayRepoStore) turns them into data. | |
| 15 | // | |
| 16 | // THE 21-SECOND BUFFER IS LOAD-BEARING — LEAVE IT ON. By default spacedust holds every link for 21s | |
| 17 | // before emitting, so interactions that are undone almost immediately never reach us (~<1% of links are | |
| 18 | // deleted within 21s). That is precisely what Tier B wants: we should never spend a fetch+verify+write | |
| 19 | // on a reply that was deleted two seconds later. `instant: true` bypasses it and is for UIs that want | |
| 20 | // noise, not for a retention pipeline. | |
| 21 | // | |
| 22 | // BACKPRESSURE IS THE REAL HAZARD, AND IT IS ARCHITECTURAL. Spacedust DROPS a subscriber that lags | |
| 23 | // (`RecvError::Lagged` -> cancel). If we did fetch/verify/store work inside the socket handler we would | |
| 24 | // block the Bare event loop and be disconnected — the same failure class as pgv3's `reconcileAll` and | |
| 25 | // pgh-sidecar's `spawnSync`. So the handler does one thing: parse and enqueue. All work happens in a | |
| 26 | // drain loop. If the queue fills we drop the OLDEST pointer and count it, because a dropped pointer is | |
| 27 | // recoverable (Constellation can backfill the same link by subject+source) while a lag-disconnect loses | |
| 28 | // the entire stream until reconnect. Trading a recoverable loss for an unrecoverable one is the whole | |
| 29 | // point of the bound. | |
| 30 | // | |
| 31 | // PURE + INJECTED. `createSocket` and `now` are parameters, so the URL building, event parsing, filter | |
| 32 | // framing and queue behaviour are all unit-testable with no network and no clock. | |
| 33 | // | |
| 34 | // NO RUNTIME-SPECIFIC IMPORTS HERE — deliberately. `bare-ws` pulls native addons (bare-dns) that cannot | |
| 35 | // load under Node, so importing it at the top of this file would make the whole module Bare-only and the | |
| 36 | // "pure + injected" claim above a fiction. The real socket factory lives in `spacedust-socket.js`; | |
| 37 | // `createSocket` is therefore REQUIRED, and tests inject a fake. | |
| 38 | ||
| 39 | export const DEFAULT_ENDPOINT = 'wss://spacedust.microcosm.blue/subscribe' | |
| 40 | ||
| 41 | // They instrument by user-agent and ask consumers to identify themselves. | |
| 42 | export const USER_AGENT = 'favela-gateway (+https://ap.7rnx.net)' | |
| 43 | ||
| 44 | // Link sources are `<collection NSID>:<record property path>` with NO leading dot — spacedust strips it | |
| 45 | // (`link.path.strip_prefix('.')`). These are the inbound interactions Tier B/C care about. | |
| 46 | // | |
| 47 | // VERIFIED against live traffic 2026-07-31 via Constellation's link enumeration on a post with 426 | |
| 48 | // quotes: `.embed.record.uri` 376 + `.embed.record.record.uri` 50 = 426 exactly. BOTH quote paths are | |
| 49 | // real — plain quote vs recordWithMedia — so subscribing to only the first silently missed ~12% of | |
| 50 | // quotes. Reply paths confirmed the same way (.reply.parent.uri / .reply.root.uri). | |
| 51 | export const SOURCES = { | |
| 52 | REPLY_PARENT: 'app.bsky.feed.post:reply.parent.uri', | |
| 53 | REPLY_ROOT: 'app.bsky.feed.post:reply.root.uri', | |
| 54 | QUOTE: 'app.bsky.feed.post:embed.record.uri', | |
| 55 | QUOTE_MEDIA: 'app.bsky.feed.post:embed.record.record.uri', // recordWithMedia — a DIFFERENT path | |
| 56 | LIKE: 'app.bsky.feed.like:subject.uri', | |
| 57 | REPOST: 'app.bsky.feed.repost:subject.uri', | |
| 58 | FOLLOW: 'app.bsky.graph.follow:subject' | |
| 59 | } | |
| 60 | ||
| 61 | // Content-bearing sources (Tier B — worth a fetch+verify+retain) vs count-bearing (Tier C — reference | |
| 62 | // only, answered on demand by Constellation's getBacklinksCount). §7.6's pressure valve lives here: a | |
| 63 | // post with 50k likes must not become 50k retained records. | |
| 64 | export const CONTENT_BEARING = new Set([ | |
| 65 | SOURCES.REPLY_PARENT, SOURCES.REPLY_ROOT, SOURCES.QUOTE, SOURCES.QUOTE_MEDIA | |
| 66 | ]) | |
| 67 | ||
| 68 | const DEFAULTS = { | |
| 69 | endpoint: DEFAULT_ENDPOINT, | |
| 70 | instant: false, // see the 21-second note above — do not flip this for a retention pipeline | |
| 71 | maxQueue: 1000, | |
| 72 | reconnectMinMs: 1000, | |
| 73 | reconnectMaxMs: 60_000 | |
| 74 | } | |
| 75 | ||
| 76 | // Build the subscribe URL. Params are camelCase and REPEAT rather than comma-separating: | |
| 77 | // `wantedSubjectDids=a&wantedSubjectDids=b`. Server-side caps are 50k subjects / 100 prefixes / | |
| 78 | // 10k subject DIDs / 1k sources; we do not enforce them here (the server rejects, loudly). | |
| 79 | export function buildSubscribeUrl (filter = {}, { endpoint = DEFAULT_ENDPOINT, instant = false } = {}) { | |
| 80 | const url = new URL(endpoint) | |
| 81 | const add = (name, values) => { | |
| 82 | for (const v of values || []) url.searchParams.append(name, v) | |
| 83 | } | |
| 84 | add('wantedSubjectDids', filter.subjectDids) | |
| 85 | add('wantedSubjects', filter.subjects) | |
| 86 | add('wantedSubjectPrefixes', filter.subjectPrefixes) | |
| 87 | add('wantedSources', filter.sources) | |
| 88 | if (instant) url.searchParams.set('instant', 'true') | |
| 89 | return url.toString() | |
| 90 | } | |
| 91 | ||
| 92 | // The runtime filter-update frame. Sent as a TEXT frame over the SAME socket — no reconnect, which is | |
| 93 | // what lets a mutating roster stay live. Note the shape asymmetry: the envelope is snake_case-tagged | |
| 94 | // (serde `tag="type", content="payload"`) while the payload is camelCase. It REPLACES the whole query | |
| 95 | // (`*self = opts`), so always send the complete filter, never a delta. | |
| 96 | export function optionsUpdateFrame (filter = {}) { | |
| 97 | return JSON.stringify({ | |
| 98 | type: 'options_update', | |
| 99 | payload: { | |
| 100 | wantedSubjectDids: filter.subjectDids || [], | |
| 101 | wantedSubjects: filter.subjects || [], | |
| 102 | wantedSubjectPrefixes: filter.subjectPrefixes || [], | |
| 103 | wantedSources: filter.sources || [] | |
| 104 | } | |
| 105 | }) | |
| 106 | } | |
| 107 | ||
| 108 | // Split an at-uri into a record locator. Returns null on anything that is not `at://<did>/<nsid>/<rkey>` | |
| 109 | // — a plain URL subject, or a bare DID, is legitimate and simply is not a record. | |
| 110 | export function parseAtUri (uri) { | |
| 111 | if (typeof uri !== 'string' || !uri.startsWith('at://')) return null | |
| 112 | const parts = uri.slice(5).split('/') | |
| 113 | if (parts.length !== 3) return null | |
| 114 | const [did, collection, rkey] = parts | |
| 115 | if (!did || !collection || !rkey) return null | |
| 116 | return { did, collection, rkey } | |
| 117 | } | |
| 118 | ||
| 119 | // Parse one link event into a normalized POINTER. Returns null for anything unexpected rather than | |
| 120 | // throwing — a malformed frame must never take down the subscription. | |
| 121 | // | |
| 122 | // Wire shape (spacedust/src/lib.rs ClientEvent/ClientLinkEvent): | |
| 123 | // { kind:"link", origin:"live"|"replay"|"backfill", | |
| 124 | // link:{ operation:"create"|"delete", source, source_record, source_rev, subject } } | |
| 125 | export function parseLinkEvent (raw) { | |
| 126 | let body | |
| 127 | try { | |
| 128 | body = typeof raw === 'string' ? JSON.parse(raw) : JSON.parse(raw.toString()) | |
| 129 | } catch { | |
| 130 | return null // see the framing note in the class below | |
| 131 | } | |
| 132 | if (!body || body.kind !== 'link' || !body.link) return null | |
| 133 | ||
| 134 | const l = body.link | |
| 135 | if (typeof l.source !== 'string' || typeof l.subject !== 'string') return null | |
| 136 | if (l.operation !== 'create' && l.operation !== 'delete') return null | |
| 137 | ||
| 138 | const source = parseAtUri(l.source_record) | |
| 139 | if (source === null) return null // the linking record must be addressable, or we cannot fetch it | |
| 140 | ||
| 141 | return { | |
| 142 | operation: l.operation, | |
| 143 | source: l.source, // the `<nsid>:<path>` spec, e.g. app.bsky.feed.like:subject.uri | |
| 144 | subject: l.subject, // what was linked TO (one of ours) | |
| 145 | // The linking record — this is the thing to fetch+verify+retain if content-bearing. | |
| 146 | did: source.did, | |
| 147 | collection: source.collection, | |
| 148 | rkey: source.rkey, | |
| 149 | // Repo revision of the commit that produced this link. The backfilling guide's skip test: if this | |
| 150 | // rev is <= the rev we already hold for this DID, we already have this state. | |
| 151 | rev: typeof l.source_rev === 'string' ? l.source_rev : null, | |
| 152 | contentBearing: CONTENT_BEARING.has(l.source), | |
| 153 | origin: body.origin || null | |
| 154 | } | |
| 155 | } | |
| 156 | ||
| 157 | // Exponential backoff with full jitter. Jitter matters: if the service restarts, every subscriber | |
| 158 | // reconnecting on the same schedule is a thundering herd (DESIGN §7.5 makes the same point about ghost | |
| 159 | // sweeps). | |
| 160 | export function backoffDelay (attempt, { min, max }, random = Math.random) { | |
| 161 | const ceiling = Math.min(max, min * 2 ** Math.max(0, attempt - 1)) | |
| 162 | return Math.floor(random() * ceiling) | |
| 163 | } | |
| 164 | ||
| 165 | // Does this filter narrow the stream BY SUBJECT? `sources` does not count: it defaults to every | |
| 166 | // source, so a sources-only filter looks configured and narrows nothing. Without a subject axis the | |
| 167 | // server sends the whole network (see buildSubscribeUrl — an empty array appends no param at all). | |
| 168 | export function hasSubjectAxis (filter = {}) { | |
| 169 | return (filter.subjectDids || []).length > 0 || | |
| 170 | (filter.subjects || []).length > 0 || | |
| 171 | (filter.subjectPrefixes || []).length > 0 | |
| 172 | } | |
| 173 | ||
| 174 | export class SpacedustClient { | |
| 175 | // FAIL CLOSED on an empty roster (hyper-social #47). 0.3.58 made the roster empty-by-default so an | |
| 176 | // unconfigured node would not silently index someone else's timeline — but this path turned the same | |
| 177 | // condition into a subscription to EVERYONE'S. Measured on ashatron: 98 retained posts from one | |
| 178 | // unrelated DID, zero from the roster. Unconfigured has to mean inert, not indiscriminate. | |
| 179 | // | |
| 180 | // `allowUnfiltered: true` preserves the URL builder's documented contract for a caller that really | |
| 181 | // does want the firehose — it just has to say so, rather than arrive there via an empty array. | |
| 182 | constructor (filter = {}, opts = {}) { | |
| 183 | this.opts = { ...DEFAULTS, ...opts } | |
| 184 | this._allowUnfiltered = !!opts.allowUnfiltered | |
| 185 | this.filter = { | |
| 186 | subjectDids: filter.subjectDids || [], | |
| 187 | subjects: filter.subjects || [], | |
| 188 | subjectPrefixes: filter.subjectPrefixes || [], | |
| 189 | sources: filter.sources || Object.values(SOURCES) | |
| 190 | } | |
| 191 | ||
| 192 | // REQUIRED — see the no-runtime-imports note above. Production wiring passes the factory from | |
| 193 | // `spacedust-socket.js`; tests pass a fake. | |
| 194 | if (typeof opts.createSocket !== 'function') { | |
| 195 | throw new Error('SpacedustClient requires opts.createSocket (see lib/spacedust-socket.js)') | |
| 196 | } | |
| 197 | this._createSocket = opts.createSocket | |
| 198 | // Called after every enqueue, so a caller drains on arrival rather than on a timer (event-driven, #82 I2b). | |
| 199 | this._onEnqueue = typeof opts.onEnqueue === 'function' ? opts.onEnqueue : null | |
| 200 | this._now = opts.now || (() => Date.now()) | |
| 201 | ||
| 202 | this._socket = null | |
| 203 | this._queue = [] | |
| 204 | this._closed = false | |
| 205 | this._attempt = 0 | |
| 206 | this._reconnectTimer = null | |
| 207 | ||
| 208 | // Observability — §7.6 budgets are meaningless if drops are silent. | |
| 209 | this.stats = { received: 0, parsed: 0, dropped: 0, unparseable: 0, connects: 0, disconnects: 0, refusedUnfiltered: 0 } | |
| 210 | } | |
| 211 | ||
| 212 | // Pointers waiting to be drained. The caller owns the drain loop, so it can do fetch/verify/retain at | |
| 213 | // whatever rate it can sustain WITHOUT ever blocking the socket. | |
| 214 | get pending () { | |
| 215 | return this._queue.length | |
| 216 | } | |
| 217 | ||
| 218 | start () { | |
| 219 | if (this._closed) throw new Error('client is closed') | |
| 220 | this._connect() | |
| 221 | return this | |
| 222 | } | |
| 223 | ||
| 224 | _connect () { | |
| 225 | // Refuse rather than widen. Guarded here, not in start(), so a RECONNECT cannot quietly restore an | |
| 226 | // unfiltered subscription after the roster was emptied at runtime. | |
| 227 | if (!hasSubjectAxis(this.filter) && !this._allowUnfiltered) { | |
| 228 | this.stats.refusedUnfiltered++ | |
| 229 | console.error(JSON.stringify({ | |
| 230 | msg: 'spacedust: refusing to subscribe with no subject filter', | |
| 231 | reason: 'an empty roster would subscribe to the ENTIRE network, not to nothing', | |
| 232 | fix: 'set FAVELA_ROSTER (or FAVELA_LIST), or pass allowUnfiltered:true to mean it' | |
| 233 | })) | |
| 234 | return | |
| 235 | } | |
| 236 | const url = buildSubscribeUrl(this.filter, this.opts) | |
| 237 | let socket | |
| 238 | try { | |
| 239 | socket = this._createSocket(url) | |
| 240 | } catch { | |
| 241 | return this._scheduleReconnect() // fail soft — a bad connect is a retry, not a crash | |
| 242 | } | |
| 243 | ||
| 244 | this._socket = socket | |
| 245 | this.stats.connects++ | |
| 246 | ||
| 247 | // bare-ws auto-replies PONG with the echoed payload (lib/socket.js:163-165), which is exactly what | |
| 248 | // spacedust's 30s keepalive checks (it verifies the 8 random bytes come back). Nothing to do here. | |
| 249 | ||
| 250 | socket.on('data', (chunk) => this._onMessage(chunk)) | |
| 251 | socket.on('error', () => this._onDisconnect()) | |
| 252 | socket.on('close', () => this._onDisconnect()) | |
| 253 | socket.on('end', () => this._onDisconnect()) | |
| 254 | } | |
| 255 | ||
| 256 | // THE HOT PATH. Parse and enqueue. Nothing else. Any await here risks a lag-disconnect. | |
| 257 | // | |
| 258 | // Framing note: bare-ws is a non-objectMode Duplex that pushes one Buffer per WebSocket message | |
| 259 | // (lib/socket.js:195), and flowing-mode 'data' emits per push — so one chunk is one message. We do not | |
| 260 | // rely on that silently: an unparseable chunk is counted and dropped rather than throwing, and a lost | |
| 261 | // pointer is recoverable via Constellation backfill by (subject, source). | |
| 262 | _onMessage (chunk) { | |
| 263 | this.stats.received++ | |
| 264 | const pointer = parseLinkEvent(chunk) | |
| 265 | if (pointer === null) { | |
| 266 | this.stats.unparseable++ | |
| 267 | return | |
| 268 | } | |
| 269 | this.stats.parsed++ | |
| 270 | this.enqueue(pointer) | |
| 271 | } | |
| 272 | ||
| 273 | // Enqueue an already-parsed pointer. PUBLIC because the live socket is not the only source: the | |
| 274 | // Constellation backfill pushes historical pointers through this same queue, so both share one bound, | |
| 275 | // one drain and one dedup surface rather than growing a second path. | |
| 276 | // | |
| 277 | // Bounded: drop the OLDEST. Losing one recoverable pointer beats being disconnected and losing the | |
| 278 | // stream — Constellation can re-offer a dropped pointer, nothing can re-offer a missed subscription. | |
| 279 | enqueue (pointer) { | |
| 280 | if (!pointer) return false | |
| 281 | if (this._queue.length >= this.opts.maxQueue) { | |
| 282 | this._queue.shift() | |
| 283 | this.stats.dropped++ | |
| 284 | } | |
| 285 | this._queue.push({ ...pointer, receivedAt: this._now() }) | |
| 286 | if (this._onEnqueue) { try { this._onEnqueue() } catch {} } | |
| 287 | return true | |
| 288 | } | |
| 289 | ||
| 290 | // Drain up to `max` pointers. Returns them; the caller runs the §7.6 pipeline. | |
| 291 | drain (max = Infinity) { | |
| 292 | if (max >= this._queue.length) { | |
| 293 | const all = this._queue | |
| 294 | this._queue = [] | |
| 295 | return all | |
| 296 | } | |
| 297 | return this._queue.splice(0, max) | |
| 298 | } | |
| 299 | ||
| 300 | // Change the filter WITHOUT reconnecting. Send the complete filter — the server replaces wholesale. | |
| 301 | // Falls back to a reconnect if the socket is not writable, so a roster change is never silently lost. | |
| 302 | updateFilter (filter = {}) { | |
| 303 | // The server REPLACES the whole query, so an empty update is the same unfiltered subscription | |
| 304 | // reached by a different door. Refuse WITHOUT mutating this.filter: the previous (narrower) query | |
| 305 | // stays in force, which is stale but bounded — never wider than what was already asked for. | |
| 306 | const next = { ...this.filter, ...filter } | |
| 307 | if (!hasSubjectAxis(next) && !this._allowUnfiltered) { | |
| 308 | this.stats.refusedUnfiltered++ | |
| 309 | console.error(JSON.stringify({ | |
| 310 | msg: 'spacedust: refusing a filter update that would remove every subject', | |
| 311 | reason: 'the server replaces the whole query — an empty one matches the ENTIRE network' | |
| 312 | })) | |
| 313 | return false | |
| 314 | } | |
| 315 | this.filter = next | |
| 316 | if (this._socket === null) return false | |
| 317 | try { | |
| 318 | this._socket.write(optionsUpdateFrame(this.filter)) | |
| 319 | return true | |
| 320 | } catch { | |
| 321 | this._onDisconnect() | |
| 322 | return false | |
| 323 | } | |
| 324 | } | |
| 325 | ||
| 326 | _onDisconnect () { | |
| 327 | if (this._socket === null) return // already handled; close/end/error can all fire | |
| 328 | this._socket = null | |
| 329 | this.stats.disconnects++ | |
| 330 | if (!this._closed) this._scheduleReconnect() | |
| 331 | } | |
| 332 | ||
| 333 | _scheduleReconnect () { | |
| 334 | if (this._closed || this._reconnectTimer !== null) return | |
| 335 | this._attempt++ | |
| 336 | const delay = backoffDelay(this._attempt, { | |
| 337 | min: this.opts.reconnectMinMs, | |
| 338 | max: this.opts.reconnectMaxMs | |
| 339 | }) | |
| 340 | this._reconnectTimer = setTimeout(() => { | |
| 341 | this._reconnectTimer = null | |
| 342 | if (!this._closed) this._connect() | |
| 343 | }, delay) | |
| 344 | } | |
| 345 | ||
| 346 | // NOTE for the caller: a reconnect leaves a GAP. Spacedust replays nothing — there is no cursor in the | |
| 347 | // subscribe contract. Gap-filling is Constellation's job (getBacklinks by subject+source, paginated), | |
| 348 | // which is why §7.6 pairs the live subscription with the historical index rather than treating either | |
| 349 | // as sufficient alone. | |
| 350 | close () { | |
| 351 | this._closed = true | |
| 352 | if (this._reconnectTimer !== null) { | |
| 353 | clearTimeout(this._reconnectTimer) | |
| 354 | this._reconnectTimer = null | |
| 355 | } | |
| 356 | const socket = this._socket | |
| 357 | this._socket = null | |
| 358 | if (socket !== null) { | |
| 359 | try { | |
| 360 | socket.end() | |
| 361 | } catch { | |
| 362 | // already gone | |
| 363 | } | |
| 364 | } | |
| 365 | } | |
| 366 | } |
| … | … | @@ -1,12 +1,12 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "favela-gateway", |
| 3 | "version": "0.6.4", | |
| 3 | "version": "0.6.5", | |
| 4 | 4 | "lockfileVersion": 3, |
| 5 | 5 | "requires": true, |
| 6 | 6 | "packages": { |
| 7 | 7 | "": { |
| 8 | 8 | "name": "favela-gateway", |
| 9 | "version": "0.6.4", | |
| 9 | "version": "0.6.5", | |
| 10 | 10 | "dependencies": { |
| 11 | 11 | "@ipld/dag-cbor": "^9.2.1", |
| 12 | 12 | "ap-bridge": "file:../ap-bridge", |
| … | … | @@ -18,6 +18,8 @@ |
| 18 | 18 | "bare-os": "^3.9.3", |
| 19 | 19 | "bare-path": "^3.1.2", |
| 20 | 20 | "bare-process": "^4.5.1", |
| 21 | "bare-ws": "^3.2.0", | |
| 22 | "favela-bridge": "file:../favela-bridge", | |
| 21 | 23 | "hyper-svc": "file:../hyper-svc" |
| 22 | 24 | }, |
| 23 | 25 | "bin": { |
| … | … | @@ -1212,6 +1214,23 @@ |
| 1212 | 1214 | "integrity": "sha512-0ugbP4CJW4e2D20jvEcC4973dCgIaHI4Rw1PT+26U9zEve7FyYdWAIwUnoeOYvoCfn+wXHoHTKb1KhkYlb60Pw==", |
| 1213 | 1215 | "dev": true, |
| 1214 | 1216 | "license": "Apache-2.0" |
| 1217 | }, | |
| 1218 | "../favela-bridge": { | |
| 1219 | "version": "0.0.13", | |
| 1220 | "dependencies": { | |
| 1221 | "bare-net": "^1.3.0", | |
| 1222 | "bare-process": "^4.0.0", | |
| 1223 | "favela-identity": "file:../favela-identity", | |
| 1224 | "favela-markdown": "file:../favela-markdown", | |
| 1225 | "hyper-svc": "file:../hyper-svc" | |
| 1226 | }, | |
| 1227 | "devDependencies": { | |
| 1228 | "brittle": "^3.5.0" | |
| 1229 | } | |
| 1230 | }, | |
| 1231 | "node_modules/favela-bridge": { | |
| 1232 | "resolved": "../favela-bridge", | |
| 1233 | "link": true | |
| 1215 | 1234 | } |
| 1216 | 1235 | } |
| 1217 | 1236 | } |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "favela-gateway", |
| 3 | "version": "0.6.4", | |
| 3 | "version": "0.6.5", | |
| 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", |
| … | … | @@ -21,6 +21,8 @@ |
| 21 | 21 | "bare-os": "^3.9.3", |
| 22 | 22 | "bare-path": "^3.1.2", |
| 23 | 23 | "bare-process": "^4.5.1", |
| 24 | "bare-ws": "^3.2.0", | |
| 25 | "favela-bridge": "file:../favela-bridge", | |
| 24 | 26 | "hyper-svc": "file:../hyper-svc" |
| 25 | 27 | }, |
| 26 | 28 | "devDependencies": { |
| … | … | @@ -0,0 +1,94 @@ |
| 1 | // bsky-ingest.test.js — the wiring: bound DIDs → a Spacedust subscription → the identity's observation feed | |
| 2 | // (hyper-social #82, I2b). | |
| 3 | // | |
| 4 | // ⛔ ONLY BOUND DIDs ARE SUBSCRIBED. A fronted DID that does not name a held favela feed whose reach names it back | |
| 5 | // is not anyone's here (bsky-bind.js); subscribing for it would collect interactions nobody may see. | |
| 6 | // ⛔ AND NEVER AN EMPTY FILTER — that is a subscription to the whole network (spacedust.js refuses it too). | |
| 7 | ||
| 8 | import test from 'brittle' | |
| 9 | import { BskyIngest } from '../lib/bsky-ingest.js' | |
| 10 | import { tidFor } from 'favela-bridge/lib/tid.js' | |
| 11 | ||
| 12 | const DID = 'did:plc:fleekyfleekyfleeky' | |
| 13 | const LC = 'did:plc:loserclubloserclub' | |
| 14 | const FEED = 'd2'.repeat(32) | |
| 15 | const ALICE = 'did:plc:alicealicealice' | |
| 16 | const NAME = '01k4qspvquliq9js01m6c5658b' | |
| 17 | const POST = `at://${DID}/app.bsky.feed.post/${tidFor(NAME)}` | |
| 18 | ||
| 19 | function harness () { | |
| 20 | const notes = [] | |
| 21 | const removes = [] | |
| 22 | const sockets = [] | |
| 23 | const fetched = [] | |
| 24 | const membrane = { | |
| 25 | metrics: () => ({ dids: [{ did: DID }, { did: LC }] }), | |
| 26 | peers: { | |
| 27 | front: async (did) => ({ getRecord: async () => (did === DID ? { value: { $type: 'net.favela.feed', feed: FEED } } : null) }) | |
| 28 | } | |
| 29 | } | |
| 30 | const fediverse = { | |
| 31 | heldFeed: (feed) => (feed === FEED | |
| 32 | ? { reach: async () => ({ networks: ['bsky'], credential: DID }), ownPosts: async () => ({ feed: FEED, rows: [{ name: NAME }] }) } | |
| 33 | : null), | |
| 34 | observationsForFeed: (feed) => (feed === FEED | |
| 35 | ? { note: async (obs) => { notes.push(obs) }, remove: async (activity) => { removes.push(activity) } } | |
| 36 | : null) | |
| 37 | } | |
| 38 | const createSocket = (url) => { | |
| 39 | const handlers = {} | |
| 40 | const s = { on: (ev, fn) => { (handlers[ev] = handlers[ev] || []).push(fn); return s }, emit: (ev, ...a) => { for (const fn of handlers[ev] || []) fn(...a) } } | |
| 41 | s.url = url; s.write = () => {}; s.destroy = () => {}; sockets.push(s); return s } | |
| 42 | const fetch = async (url) => { | |
| 43 | fetched.push(url) | |
| 44 | return { ok: true, json: async () => ({ posts: [{ uri: `at://${ALICE}/app.bsky.feed.post/3rep`, record: { text: 'megusta from bsky' } }] }) } | |
| 45 | } | |
| 46 | return { membrane, fediverse, createSocket, fetch, notes, removes, sockets, fetched } | |
| 47 | } | |
| 48 | ||
| 49 | const frame = (operation, source, subject, collection, rkey) => Buffer.from(JSON.stringify({ | |
| 50 | kind: 'link', origin: 'live', link: { operation, source, subject, source_record: `at://${ALICE}/${collection}/${rkey}`, source_rev: '3x' } | |
| 51 | })) | |
| 52 | const settle = () => new Promise((resolve) => setTimeout(resolve, 30)) | |
| 53 | ||
| 54 | test('★★ only BOUND DIDs are subscribed, and a like lands as an observation aimed at the favela post', async (t) => { | |
| 55 | const h = harness() | |
| 56 | const ingest = new BskyIngest({ membrane: h.membrane, fediverse: h.fediverse, fetch: h.fetch, createSocket: h.createSocket, log: () => {} }) | |
| 57 | await ingest.start() | |
| 58 | t.teardown(() => ingest.stop()) | |
| 59 | t.is(h.sockets.length, 1, 'one subscription') | |
| 60 | t.ok(h.sockets[0].url.includes(encodeURIComponent(DID)) || h.sockets[0].url.includes(DID), 'for the bound DID') | |
| 61 | t.absent(h.sockets[0].url.includes(LC), 'never for an unbound one') | |
| 62 | t.alike(ingest.metrics().bound, [{ did: DID, feed: FEED }]) | |
| 63 | ||
| 64 | h.sockets[0].emit('data', frame('create', 'app.bsky.feed.like:subject.uri', POST, 'app.bsky.feed.like', '3lk')) | |
| 65 | await settle() | |
| 66 | t.is(h.notes.length, 1) | |
| 67 | t.alike({ kind: h.notes[0].kind, actor: h.notes[0].actor, target: h.notes[0].target, activity: h.notes[0].activity }, | |
| 68 | { kind: 'like', actor: ALICE, target: { feed: FEED, collection: 'post', name: NAME }, activity: `at://${ALICE}/app.bsky.feed.like/3lk` }) | |
| 69 | }) | |
| 70 | ||
| 71 | test('★ a reply carries its text, fetched for display; a delete removes', async (t) => { | |
| 72 | const h = harness() | |
| 73 | const ingest = new BskyIngest({ membrane: h.membrane, fediverse: h.fediverse, fetch: h.fetch, createSocket: h.createSocket, log: () => {} }) | |
| 74 | await ingest.start() | |
| 75 | t.teardown(() => ingest.stop()) | |
| 76 | h.sockets[0].emit('data', frame('create', 'app.bsky.feed.post:reply.parent.uri', POST, 'app.bsky.feed.post', '3rep')) | |
| 77 | await settle() | |
| 78 | t.is(h.notes.length, 1) | |
| 79 | t.is(h.notes[0].kind, 'reply') | |
| 80 | t.is(h.notes[0].body, 'megusta from bsky', 'the reply text') | |
| 81 | t.ok(h.fetched.some((u) => u.includes('app.bsky.feed.getPosts')), 'from the public AppView') | |
| 82 | h.sockets[0].emit('data', frame('delete', 'app.bsky.feed.like:subject.uri', POST, 'app.bsky.feed.like', '3lk')) | |
| 83 | await settle() | |
| 84 | t.alike(h.removes, [`at://${ALICE}/app.bsky.feed.like/3lk`]) | |
| 85 | }) | |
| 86 | ||
| 87 | test('⛔ nothing bound → no subscription at all (an empty filter would be the whole network)', async (t) => { | |
| 88 | const h = harness() | |
| 89 | h.membrane.metrics = () => ({ dids: [{ did: LC }] }) | |
| 90 | const ingest = new BskyIngest({ membrane: h.membrane, fediverse: h.fediverse, fetch: h.fetch, createSocket: h.createSocket, log: () => {} }) | |
| 91 | await ingest.start() | |
| 92 | t.teardown(() => ingest.stop()) | |
| 93 | t.is(h.sockets.length, 0) | |
| 94 | }) |