// One projection pass, with no transport in it.
//
// ⛔ THIS RUNS IN TWO PLACES AND MUST BE ONE IMPLEMENTATION. On a desktop the peer and the bsky sink
// are unix sockets; on a handset both are in the same worklet, because bsky records are authored and
// SIGNED next to the DID key and that key is on the phone. A projection written twice drifts, and this
// monorepo already carries six copies of one signing format, three of which eat the first character of
// a hash. So the loop takes `peer` and `bsky` as objects and never learns which it got — the same
// shape favela-host's inProcessClient uses to hand favela-surface a daemon without a socket.
//
// ⚠ THE FEDIVERSE NEEDS NO EQUIVALENT, and that asymmetry is the reason this file exists. ap-bridge
// HOLDS the feed and projects server-side, so a registered name federates without anything running on
// the device. bsky cannot work that way: the membrane is keyless and re-serves only what the device
// signed. A handset with a correct DID, a resolving handle and nine fediverse posts therefore had an
// atproto repo containing one bootstrap record — nothing was broken, nothing was ever running.
//
// ⛔ NO process.exit AND NO console. A refusal is returned, not shouted: this is called from a CLI that
// owns the exit code and from a worklet where exiting kills the app.
import { planRecord, MAX_IMAGE_BYTES, selectImages } from './plan.js'
import { planProfile, sameProfile } from './profile.js'
import { samePost } from './plan.js'
import { tidFor } from './tid.js'
import { POST_COLLECTION } from './bsky.js'
import { planEdge } from './plan-edge.js'
import { maxReach } from 'favela-markdown/lib/reach.js'
// oid -> blob ref. Content-addressed, so this can never go stale.
const blobCache = new Map()
/**
* Turn the oids planBsky named into an `app.bsky.embed.images`, uploading what is missing.
*
* ⚠ A post whose images ALL fail still goes, as text. Withholding a post because a picture would not
* upload loses more than it protects.
*/
export async function buildEmbed (peer, bsky, imgs, { maxImageBytes = MAX_IMAGE_BYTES } = {}) {
if (!imgs || !imgs.length) return { embed: null, losses: [] }
const losses0 = []
// ⛔ SELECT, NEVER DOWNSAMPLE (SOCIAL-MODEL §6.6.1). The pyramid was built once at authoring; a
// bridge picks the level nearest its ceiling. Re-encoding here would do the work per delivery on
// every bridge AND produce bytes the author never made.
const described = []
for (const im of imgs) {
let chosen = null
try { chosen = await peer.call('blob-level', { oid: im.oid, maxBytes: maxImageBytes }) } catch { chosen = null }
if (!chosen || !chosen.level) {
losses0.push('image omitted — ' + ((chosen && chosen.reason) || 'no level could be chosen'))
continue
}
// ★ THE HONESTY REQUIREMENT. A bridge that quietly serves a smaller picture has altered what was
// published, so the level it chose is reported even when everything worked.
if (chosen.reason && chosen.reason !== 'full size' && chosen.reason !== 'single level') {
losses0.push('image: ' + chosen.reason)
}
let f = null
try { f = await peer.call('file-get', { oid: chosen.level.oid }) } catch { f = null }
described.push({ ...im, path: (f && f.path) || null, mimeType: f && f.mimeType, size: f && f.size })
}
const { images, losses: sizeLosses } = selectImages(described)
const losses = losses0.concat(sizeLosses)
const out = []
for (const d of images) {
let ref = blobCache.get(d.oid)
if (!ref) {
ref = await bsky.uploadBlob(d.path, d.mimeType)
if (ref) blobCache.set(d.oid, ref)
}
// ⛔ Alt text travels. It is an accessibility obligation and the author already wrote it.
if (ref) out.push({ alt: d.alt || '', image: ref })
else losses.push('an image could not be uploaded — omitted from this post')
}
if (!out.length) return { embed: null, losses }
return { embed: { $type: 'app.bsky.embed.images', images: out }, losses }
}
/**
* Bring the bsky profile into line with favela's.
*
* ⛔ RECONCILIATION, not an event. Comparing against the REPO rather than a local note of what was
* last sent is the part that matters: a projector that trusts its own memory drifts from the thing it
* claims to describe.
*/
export async function reconcileProfile (peer, bsky, reach, send, log) {
if (!bsky.configured) return { done: false, reason: 'no sink' }
// ⛔ CONSENT FIRST, exactly as for a post. A profile is published content too.
if (!reach.networks.includes('bsky')) { log(' profile bsky: not enabled for this identity'); return { done: false, reason: 'not enabled' } }
let prof = null
try { prof = await peer.call('profile') } catch { prof = null }
if (!prof) { log(' profile could not be read from the peer — leaving bsky alone'); return { done: false, reason: 'unreadable' } }
const describe = async (oid) => {
if (!oid) return null
try {
const r = await peer.call('avatar-get', { oid })
return (r && r.path) ? { oid, path: r.path, mimeType: r.mimeType, size: r.size } : null
} catch { return null }
}
const plan = planProfile({
name: prof.name,
bio: prof.bio,
avatar: await describe(prof.avatar),
banner: await describe(prof.banner)
})
for (const l of plan.losses) log(' profile ' + l)
if (!plan.project) { log(' profile nothing published on favela — leaving bsky alone'); return { done: false, reason: 'nothing to publish' } }
const actual = await bsky.getProfile()
// ⛔ "COULD NOT ASK" IS NOT "ABSENT". Writing because a read failed republishes on every transient
// error, and each write is a repo version.
if (actual && actual.unknown) { log(' profile could not read what bsky holds — not writing'); return { done: false, reason: 'unknown' } }
const record = { ...plan.record }
for (const [field, img] of Object.entries(plan.images)) {
let ref = blobCache.get(img.oid)
if (!ref) {
ref = await bsky.uploadBlob(img.path, img.mimeType)
if (ref) blobCache.set(img.oid, ref)
}
if (ref) record[field] = ref
else log(' profile ' + field + ' could not be uploaded — omitted from this write')
}
const fields = () => Object.keys(record).filter((k) => k !== '$type').join(', ')
if (sameProfile(record, actual)) { log(' profile bsky: already matches'); return { done: false, reason: 'already matches' } }
if (!send) { log(' profile bsky: would update (' + fields() + ')'); return { done: false, reason: 'dry run' } }
const r = await bsky.putProfile(record)
log(' profile bsky: ' + (r.sent ? 'UPDATED (' + fields() + ')' : 'FAILED — ' + r.error))
return { done: !!r.sent, error: r.error || null }
}
/**
* The DID names its favela feed back: `net.favela.feed/self = { feed }` (hyper-social #82, I2a / D4a).
*
* ⛔ TWO-SIDED, OR NOT AT ALL. The identity's reach already names its DID (`credential`); this is the other half,
* signed by the DID key. A gateway binds a DID to an identity only when both agree, so nobody can attach somebody
* else's DID to their feed — or their feed to somebody else's DID.
*
* ⛔ PUBLISHED CONTENT: waits for bsky consent like the profile, and RECONCILES against the repo — written only when
* absent or different, and never after a failed read ("could not ask" is not "absent").
*/
export const FEED_CLAIM = 'net.favela.feed'
export async function reconcileFeedClaim (feed, bsky, reach, send, log, roomType = 'public') {
if (!bsky.configured) return { done: false, reason: 'no sink' }
if (!reach.networks.includes('bsky')) return { done: false, reason: 'not enabled' }
// ⛔ THE REACH LATTICE TOO, not only the switch. The claim names the feed key, and a favela feed replicates in the
// clear to anyone holding its key — so it may leave only from where a post may (a members room reaches nothing).
if (!maxReach({ roomType, verb: 'post' }).has('bsky')) return { done: false, reason: 'room type does not reach bsky' }
if (!/^[0-9a-f]{64}$/i.test(String(feed || ''))) return { done: false, reason: 'no feed key' }
const actual = await bsky.getRecord(FEED_CLAIM, 'self')
if (actual && actual.unknown) { log(' feed link could not read what bsky holds — not writing'); return { done: false, reason: 'unknown' } }
const want = String(feed).toLowerCase()
if (actual && actual.feed === want) return { done: false, reason: 'already matches' }
if (!send) { log(' feed link bsky: would write net.favela.feed/self'); return { done: false, reason: 'dry run' } }
const r = await bsky.putRecord(FEED_CLAIM, 'self', { $type: FEED_CLAIM, feed: want, createdAt: new Date().toISOString() })
log(' feed link bsky: ' + (r.sent ? 'WRITTEN (this DID names feed ' + want.slice(0, 12) + ')' : 'FAILED — ' + r.error))
return { done: !!r.sent, error: r.error || null }
}
/**
* One full pass: read consent, reconcile the profile, then project the posts that reach bsky.
*
* ⛔ REFUSES RATHER THAN GUESSES. `have()` answering null means we could not find out what the repo
* holds, which is NOT an empty repo — and the difference is a duplicate of every post. A send run in
* that state returns `{ refused }` and writes nothing.
*/
export async function projectCycle ({
peer, bsky, roomType = 'public', send = false, limit = Infinity,
reproject = false, maxImageBytes = MAX_IMAGE_BYTES, log = () => {},
// Two header details the CLI knows and this function cannot: whether the room type was CONFIGURED
// or assumed, and which socket the sink is. Reported because "public (assumed)" is the input the
// reach lattice cannot infer, and if it is ever wrong it is wrong in the direction of a leak.
roomTypeNote = '', sinkLabel = null,
// ⛔ INJECTED, because resolving a subject means computing an atproto record CID and this package
// does not depend on the atproto stack. favela-host supplies both; the desktop CLI supplies neither
// and simply projects no edges, exactly as before.
resolveSubject = null, resolveRoot = null
} = {}) {
const reach = await peer.reach()
const { feed, rows } = await peer.ownPosts()
log('favela-bridge · feed ' + String(feed).slice(0, 12) + ' · ' + rows.length + ' own post(s)')
log(' room type ' + roomType + roomTypeNote)
log(' bsky sink ' + (bsky.configured ? (sinkLabel || 'configured') : 'not configured'))
log(' enabled ' + (reach.unknown
? 'UNKNOWN — the peer could not be asked, so nothing will be published'
: (reach.networks.length ? reach.networks.join(', ') : 'nothing (turn a network on in settings)')))
log(' mode ' + (send ? 'SEND' : 'dry run — nothing will be published'))
const found = bsky.configured ? await bsky.have() : new Set()
const have = found === null ? new Set() : found
if (send && found === null) {
log(' refusing to send: could not read what the bsky sink already holds')
return { refused: 'could not read what the bsky sink already holds', feed, rows: rows.length, sent: 0, skipped: 0, withheld: 0, held: 0 }
}
if (found === null) log(' ⚠ could not read the sink — "already there" is unknown below')
const profile = await reconcileProfile(peer, bsky, reach, send, log)
// ⛔ AND THE OTHER HALF OF THE IDENTITY LINK (I2a): the DID names this feed. Never fatal to the pass.
try { await reconcileFeedClaim(feed, bsky, reach, send, log, roomType) } catch (err) { log(' feed link failed: ' + ((err && err.message) || err)) }
let sent = 0; let skipped = 0; let withheld = 0; let held = 0
for (const row of rows) {
const plan = planRecord(row, { roomType, verb: 'post' })
const line = ' ' + row.name + ' [' + plan.networks.join(',') + ']'
// ⛔ CONSENT FIRST, then capability. The reach lattice says what a room MAY reach; this says what
// its owner has ASKED for. A post may be publishable and still must not be published.
if (!reach.networks.includes('bsky')) {
withheld++
log(line + ' bsky: not enabled for this identity')
} else if (!plan.bsky.project) {
withheld++
log(line + ' bsky: withheld — ' + plan.bsky.losses.join('; '))
} else if (have.has(tidFor(row.name))) {
if (!reproject) {
skipped++
log(line + ' bsky: already there')
} else {
const { embed, losses: imgLosses } = await buildEmbed(peer, bsky, plan.bsky.images, { maxImageBytes })
for (const l of imgLosses) log(line + ' bsky: ' + l)
const desired = embed ? { ...plan.bsky.record, embed } : plan.bsky.record
const rkey = tidFor(row.name)
const actual = await bsky.getRecord(POST_COLLECTION, rkey)
if (actual && actual.unknown) {
log(line + ' bsky: could not read what is there — not touching it')
} else if (samePost(desired, actual)) {
skipped++
log(line + ' bsky: already matches')
} else if (!send) {
log(line + ' bsky: would REPROJECT ' + rkey + (embed ? ' (+' + embed.images.length + ' image)' : ''))
} else {
const r = await bsky.putRecord(POST_COLLECTION, rkey, desired)
if (r.sent) { sent++; log(line + ' bsky: REPROJECTED ' + rkey + (embed ? ' (+' + embed.images.length + ' image)' : '')) }
else log(line + ' bsky: reproject FAILED — ' + r.error)
}
}
} else if (send && sent >= limit) {
held++
log(line + ' bsky: held — limit ' + limit + ' reached this cycle')
} else if (!send) {
log(line + ' bsky: would send ' + JSON.stringify(plan.bsky.record.text.slice(0, 48)) +
(plan.bsky.images.length ? ' + ' + plan.bsky.images.length + ' image(s)' : '') +
(plan.bsky.losses.length ? ' — losing: ' + plan.bsky.losses.join('; ') : ''))
} else {
const { embed, losses: imgLosses } = await buildEmbed(peer, bsky, plan.bsky.images, { maxImageBytes })
for (const l of imgLosses) log(line + ' bsky: ' + l)
const record = embed ? { ...plan.bsky.record, embed } : plan.bsky.record
const r = await bsky.send(row.name, record)
if (r.sent) { sent++; log(line + ' bsky: SENT ' + r.rkey + (plan.bsky.losses.length ? ' — lost: ' + plan.bsky.losses.join('; ') : '')) }
else log(line + ' bsky: FAILED — ' + r.error)
}
// ⚠ Reported, never delivered, from this side. ap-bridge projects the fediverse from the feed it
// HOLDS, so silence here would read as "the fediverse got nothing" when it is simply not our job.
if (plan.fediverse.project) log(' fediverse: planned (' + plan.fediverse.content.length + ' bytes of html) — delivered by ap-bridge from the held feed')
else log(' fediverse: withheld — ' + plan.fediverse.losses.join('; '))
}
// ── deletes ──────────────────────────────────────────────────────────────────────────────────
//
// ⛔ ON POSITIVE EVIDENCE ONLY. A name absent from ownPosts() may be deleted or may simply not have
// replicated, so absence never deletes anything; a TOMBSTONE the peer reports can only mean deleted.
// Posts and replies both live at app.bsky.feed.post/<tidFor(name)>.
//
// ⚠ NOT gated on consent. Deleting publishes nothing, and a person who switched bsky off and then
// deleted a post still wants it gone from there. It does need a readable repo (`found` — a send with
// an unreadable one was refused above) and SEND. A peer adapter without `ownDeleted` (the desktop
// CLI's socket client) retracts nothing, as before.
let deleted = 0
if (typeof peer.ownDeleted === 'function' && found !== null) {
let gone = null
try { gone = await peer.ownDeleted() } catch (err) { log(' deletes: could not read — ' + ((err && err.message) || err)) }
const names = gone ? [...(gone.post || []), ...(gone.reply || [])] : []
for (const name of names) {
const rkey = tidFor(name)
if (!have.has(rkey)) continue
if (!send) { log(' ' + name + ' bsky: would DELETE ' + rkey); continue }
const d = await bsky.deleteRecord(POST_COLLECTION, rkey)
if (d.sent) { deleted++; log(' ' + name + ' bsky: DELETED ' + rkey) }
else log(' ' + name + ' bsky: delete FAILED — ' + d.error)
}
}
// ── edges ────────────────────────────────────────────────────────────────────────────────────
//
// ⛔ AFTER THE POSTS, ALWAYS. An edge names a subject, and the subject resolves only once the post
// it points at is in the repo — projecting edges first would withhold every one of them on the first
// pass with "not in the atproto repo yet" and fix itself silently on the second.
//
// ⚠ Skipped entirely when the caller supplies no resolver: favela-bridge's CLI has no way to compute
// a subject CID (that needs the atproto stack, which this package deliberately does not depend on),
// so a desktop run keeps doing exactly what it did before.
const edgeStats = { sent: 0, skipped: 0, withheld: 0 }
if (resolveSubject && reach.networks.includes('bsky')) {
for (const collection of ['reaction', 'repost', 'reply']) {
let list = []
try { list = await peer.call('edges-by', { collection }) } catch { list = [] }
for (const e of Array.isArray(list) ? list : []) {
// ⛔ RESOLVED BEFORE PLANNING. planEdge is pure and calls `subject()` synchronously, so handing
// it an async resolver puts a Promise where the lexicon wants { uri, cid } — the record would
// be built, pass every check, and publish an object with no subject.
const subj = await resolveSubject(e.target)
const rootSubj = resolveRoot ? await resolveRoot(e.target) : null
const plan = planEdge(e, { ownFeed: feed, roomType, subject: () => subj, root: rootSubj ? () => rootSubj : null })
const line = ' ' + collection + '/' + e.name
if (!plan.project) {
edgeStats.withheld++
log(line + ' ' + plan.collection + ': withheld — ' + plan.losses.join('; '))
continue
}
// ⛔ READ BEFORE WRITE, per collection. `have()` lists posts only, and an edge written every
// pass would append a repo version for a like nobody changed.
const existing = await bsky.getRecord(plan.collection, plan.rkey)
if (existing && existing.unknown) {
log(line + ' ' + plan.collection + ': could not read what is there — not touching it')
continue
}
if (existing) {
edgeStats.skipped++
log(line + ' ' + plan.collection + ': already there')
continue
}
if (!send) {
log(line + ' ' + plan.collection + ': would send → ' + plan.record.subject.uri)
continue
}
const w = await bsky.putRecord(plan.collection, plan.rkey, plan.record)
if (w.sent) { edgeStats.sent++; log(line + ' ' + plan.collection + ': SENT ' + plan.rkey) }
else log(line + ' ' + plan.collection + ': FAILED — ' + w.error)
}
}
log(' edges: sent ' + edgeStats.sent + ' · already there ' + edgeStats.skipped +
' · withheld ' + edgeStats.withheld)
}
log(' ' + (send ? 'sent ' + sent : 'would send ' + (rows.length - withheld - skipped)) +
' · already there ' + skipped + ' · withheld ' + withheld +
(held ? ' · held by limit ' + held : ''))
return { refused: null, feed, rows: rows.length, sent, skipped, withheld, held, deleted, profile, reach, edges: edgeStats }
}
export default { projectCycle, reconcileProfile, buildEmbed }