// The bsky sink: a co-located hyper-proto-peer, reached over its unix control socket.
//
// ⛔ WE DO NOT SIGN AND WE DO NOT SPEAK TO BSKY. hpp holds the DID signing key and forms the commit;
// the keyless membrane (atproto-bridge) re-serves it over HTTP; bsky's relay crawls that. This hands
// hpp a record and stops. Verified 2026-09-07 on ashatron: apply-write returned ok, records 79→80,
// and get-repo-car rebuilt and verified the resulting commit.
//
// ⚠ hpp's UDS surface is filesystem-gated and trusted WITHOUT the writer allowlist — its own header
// says so, and favela-web was its first co-located client. So the socket path IS the capability:
// whoever can open it can write that DID. That is why it is configuration and never a default.
import { tidFor } from './tid.js'
// ⛔ hyper-svc IS NOT IMPORTED AT MODULE SCOPE, and that is what makes this sink usable on a handset.
// It reaches bare-net and a unix socket; SELinux refuses an unprivileged Android app a `sock_file`, so
// the phone can never use that path — but bare-pack bundles by IMPORT, not by use, so naming it up here
// shipped the whole transport into a worklet that cannot call it. Loaded only on the socket path, which
// is the only path that needs it. Same reason favela-peer's rpc.js defers `./bridge.js`.
export const PROFILE_COLLECTION = 'app.bsky.actor.profile'
export const PROFILE_RKEY = 'self'
export const POST_COLLECTION = 'app.bsky.feed.post'
export class BskySink {
constructor (sock, opts = {}) {
this.sock = sock || null
this._connect = opts.connect || null
this._log = opts.log || (() => {})
}
// ⛔ AN INJECTED CONNECT IS A CONFIGURED SINK. This asked only about `sock`, so every method
// short-circuited to "no bsky sink configured" for a caller that had supplied a working transport and
// no socket path — which is exactly the in-process case on a phone, where hpp is in the same worklet
// and there is no socket to name.
get configured () { return !!this.sock || !!this._connect }
async _open () {
if (this._connect) return this._connect(this.sock)
// ⛔ OPAQUE, AND MEASURED. `await import('hyper-svc')` is a static string, so bare-pack FOLLOWS it
// and bundles the whole transport anyway — deferring it changed nothing: the packed worklet grew a
// third copy of hyper-svc at
// favela-host/node_modules/favela-bridge/node_modules/hyper-svc (610 files → 915). The repo's
// deployment rules already name this trick for bare-sidecar (`'pear-' + 'runtime'`); a computed
// specifier still resolves normally at runtime on a desktop, where this path is the only one used,
// and is unfollowable at pack time for a phone that can never open a unix socket anyway.
const { Client } = await import('hyper-' + 'svc')
return Client.discover({ uds: this.sock, prefer: 'uds' })
}
/**
* Send one planned record. Resolves `{ sent, rkey?, error? }` and NEVER rejects — a bridge that is
* down is an ordinary condition and must not become an exception in the caller's loop.
*/
async send (name, record) {
if (!this.configured) return { sent: false, error: 'no bsky sink configured' }
let client = null
try {
client = await this._open()
// ⛔ A REAL TID. An earlier version used 'favela-' + name because hpp accepted it and the
// rkey then traced back to its source record. MEASURED on the live network: our PDS served
// all 74 posts, and bsky's AppView showed the 71 with TID rkeys and dropped the three without,
// silently. `app.bsky.feed.post` is keyed "tid" and the AppView enforces it.
const rkey = tidFor(name)
const res = await client.call('apply-write', {
action: 'create', collection: 'app.bsky.feed.post', rkey, record
})
this._log('sent', { rkey, ok: !!(res && res.ok) })
// ⚠ NOT verified here. apply-write acknowledges the APPEND; the commit is re-signed a moment
// later. Measured: calling get-repo-car immediately after returned "signed commit data CID !=
// rebuilt MST root", and the same call seconds later verified. Treating that window as failure
// would raise a false alarm on every post.
return { sent: true, rkey }
} catch (err) {
const msg = (err && err.message) || String(err)
this._log('send-failed', { err: msg })
return { sent: false, error: msg }
} finally {
// hpp respawns on OTA; a cached connection across one fails every later post with an error
// about a socket rather than about a bridge.
if (client && client.close) { try { await client.close() } catch {} }
}
}
/** What this DID already holds, so a run does not re-send what it sent last time. */
async have () {
if (!this.configured) return new Set()
let client = null
try {
client = await this._open()
const s = await client.call('get-state', {})
return new Set((s.records || [])
.filter((r) => r.collection === 'app.bsky.feed.post')
.map((r) => r.rkey))
} catch {
// ⛔ An EMPTY set would read as "nothing is there" and re-send everything. Unknown is not
// empty, so the caller is told to do nothing rather than told the repo is bare.
return null
} finally {
if (client && client.close) { try { await client.close() } catch {} }
}
}
// ── the profile ─────────────────────────────────────────────────────────────────────────────
//
// ⛔ READ BEFORE WRITE. The bsky profile is a singleton at rkey `self` and applyWrite upserts it,
// so writing unconditionally on every cycle would append a repo version every fifteen seconds for
// a profile nobody changed. Reconciling against the record actually in the repo — rather than a
// local note of what we last sent — is also the only version that cannot drift from it.
/**
* One record currently in the repo, or null. `{ unknown: true }` when we could not ask.
*
* ⛔ "COULD NOT ASK" IS NOT "ABSENT" and the caller must not treat them alike — writing because a
* read failed republishes on every transient error, and each write is a repo version.
*/
async getRecord (collection, rkey) {
if (!this.configured) return null
let client = null
try {
client = await this._open()
return await client.call('get-record', { collection, rkey })
} catch (err) {
this._log('record-read-failed', { collection, rkey, err: (err && err.message) || String(err) })
return { unknown: true }
} finally {
if (client && client.close) { try { await client.close() } catch {} }
}
}
/** Replace an existing record — reprojection, not a new post. */
async putRecord (collection, rkey, record) {
if (!this.configured) return { sent: false, error: 'no bsky sink configured' }
let client = null
try {
client = await this._open()
await client.call('apply-write', { action: 'put', collection, rkey, record })
return { sent: true }
} catch (err) {
const msg = (err && err.message) || String(err)
this._log('record-write-failed', { collection, rkey, err: msg })
return { sent: false, error: msg }
} finally {
if (client && client.close) { try { await client.close() } catch {} }
}
}
/** Delete one record — the retraction of a favela post that was deleted. `{ sent, error? }`, never rejects. */
async deleteRecord (collection, rkey) {
if (!this.configured) return { sent: false, error: 'no bsky sink configured' }
let client = null
try {
client = await this._open()
await client.call('apply-write', { action: 'delete', collection, rkey })
return { sent: true }
} catch (err) {
const msg = (err && err.message) || String(err)
this._log('record-delete-failed', { collection, rkey, err: msg })
return { sent: false, error: msg }
} finally {
if (client && client.close) { try { await client.close() } catch {} }
}
}
/** The `app.bsky.actor.profile` currently in the repo, or null. Never throws. */
async getProfile () {
if (!this.configured) return null
let client = null
try {
client = await this._open()
return await client.call('get-record', { collection: PROFILE_COLLECTION, rkey: PROFILE_RKEY })
} catch (err) {
// ⛔ null means "absent"; this means "could not ask", and the caller must not treat the two
// alike — writing because a read failed would republish on every transient error.
this._log('profile-read-failed', { err: (err && err.message) || String(err) })
return { unknown: true }
} finally {
if (client && client.close) { try { await client.close() } catch {} }
}
}
/**
* Hand hpp a picture BY PATH and return an atproto blob ref, or null.
*
* ⛔ A PATH, never bytes. Both ends are on one machine — it is a unix socket — and favela-peer
* settled this same question the same way. Hex-encoding a picture to hand it to a process that
* writes it straight back to disk doubles it in transit, and needs a buffer library this package
* would otherwise not depend on: b4a resolves here only from a PARENT node_modules, so declaring
* it would have published a tree that could not find it and crashed the installed app at boot.
*/
async uploadBlob (path, mimeType) {
if (!this.configured || !path) return null
let client = null
try {
client = await this._open()
const r = await client.call('upload-blob', { path, mimeType })
if (!r || !r.cid) return null
// The atproto blob shape. `size` is the byte count hpp actually stored, not what we sent.
return { $type: 'blob', ref: { $link: r.cid }, mimeType: r.mimeType || mimeType, size: r.size }
} catch (err) {
this._log('blob-upload-failed', { err: (err && err.message) || String(err) })
return null
} finally {
if (client && client.close) { try { await client.close() } catch {} }
}
}
/** Write the profile record. `{ sent, error? }`, never rejects. */
async putProfile (record) {
if (!this.configured) return { sent: false, error: 'no bsky sink configured' }
let client = null
try {
client = await this._open()
await client.call('apply-write', {
// ⛔ `put`, not `create`: the profile is a singleton and every write after the first is a
// replacement. applyWrite routes both to putRecord, so this is naming the act correctly
// rather than relying on that.
action: 'put', collection: PROFILE_COLLECTION, rkey: PROFILE_RKEY, record
})
this._log('profile-written', {})
return { sent: true }
} catch (err) {
const msg = (err && err.message) || String(err)
this._log('profile-write-failed', { err: msg })
return { sent: false, error: msg }
} finally {
if (client && client.close) { try { await client.close() } catch {} }
}
}
}
export default { BskySink }