// bsky-ingest.js — bsky interactions on BOUND DIDs' posts → each identity's observation feed (hyper-social #82, I2b).
//
// The wiring around three pure pieces: bsky-bind.js (is this fronted DID a favela identity's, by both sides' word),
// spacedust.js (the filtered link subscription), bsky-inbound.js (a link → an observation). The observation feed is
// the SAME one the fediverse writes to (I1), reached through the fediverse role's `observationsForFeed(feedKey)`, so
// the node renders bsky arrivals with no change at all.
//
// ⛔ ONLY BOUND DIDs ARE SUBSCRIBED, and never an empty filter (that is the whole network; spacedust.js refuses too).
// ⛔ EVENT-DRIVEN: drained on arrival (onEnqueue), re-bound when the caller says a DID's repo moved — no poll.
// ⚠ A reply's text is fetched from the public AppView for DISPLAY only; it is never a canonical record here.
import { SpacedustClient } from './spacedust.js'
import { bindDid } from './bsky-bind.js'
import { linkToObservation, WANTED } from './bsky-inbound.js'
import { tidFor } from 'favela-bridge/lib/tid.js'
const DEFAULT_APPVIEW = 'https://public.api.bsky.app'
// Which DID a link concerns: a follow's subject IS the DID; anything else names one of its records.
function subjectDid (link) {
const s = link && link.subject
if (typeof s !== 'string') return null
if (s.startsWith('did:')) return s
if (!s.startsWith('at://')) return null
return s.slice(5).split('/')[0] || null
}
export class BskyIngest {
constructor ({ membrane, fediverse, fetch = null, createSocket = null, log = () => {}, appview = DEFAULT_APPVIEW, frontTimeoutMs = 20000 } = {}) {
this.membrane = membrane
this.fediverse = fediverse
this.fetch = fetch
this.createSocket = createSocket
this.log = log
this.appview = appview
this.frontTimeoutMs = frontTimeoutMs
this._bound = new Map() // did -> feed
this._refused = {}
this._names = new Map() // feed -> Map(tid -> favela name)
this._client = null
this._draining = false
this._again = false
this._rebinding = null
this._stopped = false
}
async start () { await this.rebind(); return this }
/** Re-evaluate every fronted DID's binding and (re)subscribe for exactly the bound ones. Coalesced. */
rebind () {
if (this._rebinding) return this._rebinding
this._rebinding = this._rebind().finally(() => { this._rebinding = null })
return this._rebinding
}
async _rebind () {
if (this._stopped) return
let dids = []
try { dids = (this.membrane.metrics().dids || []).map((d) => d.did).filter(Boolean) } catch {}
const bound = new Map()
const refused = {}
for (const did of dids) {
let front = null
try { front = await this.membrane.peers.front(did, { timeout: this.frontTimeoutMs }) } catch { refused[did] = 'not fronted right now'; continue }
const r = await bindDid(did, { front, feedOf: async (f) => this.fediverse.heldFeed(f) })
if (!r.feed) { refused[did] = r.refused; continue }
if (!this.fediverse.observationsForFeed(r.feed)) { refused[did] = 'no observation feed for that identity on this gateway'; continue }
bound.set(did, r.feed)
}
const changed = [...bound.keys()].sort().join(',') !== [...this._bound.keys()].sort().join(',')
this._bound = bound
this._refused = refused
if (changed) this.log({ msg: 'bsky ingest: bound', dids: [...bound.keys()], refused })
this._subscribe()
}
_subscribe () {
const subjectDids = [...this._bound.keys()]
if (!subjectDids.length) {
if (this._client) { try { this._client.close() } catch {} ; this._client = null }
return
}
const filter = { subjectDids, sources: WANTED }
if (this._client) { this._client.updateFilter(filter); return }
if (!this.createSocket) { this.log({ msg: 'bsky ingest: no socket factory — not subscribing' }); return }
this._client = new SpacedustClient(filter, { createSocket: this.createSocket, onEnqueue: () => { this._drain() } }).start()
}
async _postName (feed, rkey, { refresh = false } = {}) {
let idx = this._names.get(feed)
if (!idx || refresh) {
idx = new Map()
try {
const held = this.fediverse.heldFeed(feed)
const { rows } = held ? await held.ownPosts() : { rows: [] }
for (const row of rows || []) { if (row && row.name) idx.set(tidFor(row.name), row.name) }
} catch {}
this._names.set(feed, idx)
}
return idx.get(rkey) || null
}
async _replyText (uri) {
if (!this.fetch) return ''
try {
const res = await this.fetch(this.appview + '/xrpc/app.bsky.feed.getPosts?uris=' + encodeURIComponent(uri))
if (!res || !res.ok) return ''
const body = await res.json()
const p = body && Array.isArray(body.posts) ? body.posts[0] : null
return (p && p.record && typeof p.record.text === 'string') ? p.record.text.slice(0, 4000) : ''
} catch { return '' }
}
async _drain () {
if (this._draining) { this._again = true; return }
this._draining = true
try {
do {
this._again = false
const links = this._client ? this._client.drain() : []
for (const link of links) {
try { await this._one(link) } catch (err) { this.log({ msg: 'bsky ingest: link failed', error: (err && err.message) || String(err) }) }
}
} while (this._again && !this._stopped)
} finally { this._draining = false }
}
async _one (link) {
const did = subjectDid(link)
const feed = did ? this._bound.get(did) : null
if (!feed) return
const sink = this.fediverse.observationsForFeed(feed)
if (!sink) return
// A post made since the index was built is unknown once; rebuild once before giving up on it.
const rkey = typeof link.subject === 'string' && link.subject.startsWith('at://') ? link.subject.split('/').pop() : null
if (rkey && !(await this._postName(feed, rkey))) await this._postName(feed, rkey, { refresh: true })
const names = this._names.get(feed) || new Map()
const r = linkToObservation(link, { did, feed, postName: (k) => names.get(k) || null })
if (!r) return
if (r.op === 'remove') { await sink.remove(r.activity); return }
const obs = { ...r.obs, at: Date.now() }
if (obs.kind === 'reply') obs.body = await this._replyText(obs.activity)
await sink.note(obs)
}
metrics () {
return {
bound: [...this._bound.entries()].map(([did, feed]) => ({ did, feed })),
refused: { ...this._refused },
subscribed: !!this._client,
stats: this._client ? { ...this._client.stats, pending: this._client.pending } : null
}
}
stop () {
this._stopped = true
if (this._client) { try { this._client.close() } catch {} ; this._client = null }
}
}
export default { BskyIngest }