// serveDevice — the swarm control surface for a SwarmDevice, the multi-device analog of
// peer.serve (lib/peer.js). It is SEPARATE from StateBaseV3's own replication swarm (that one
// converges the intent-log between the DEVICES); this one lets a CLIENT/APP submit writes to,
// and observe, a running device.
//
// Mirrors the single-tenant peer-signs-over-RPC channel exactly: a ProtomuxRPC muxed over each
// Hyperswarm connection, protocol 'hyper-proto-peer/device/v1'. The device joins BOTH its own
// serverKey topic (address one specific device) and the DID discovery topic (reach ANY device of
// the DID — any of them accepts a write into the shared log). Ops:
// get-state -> always: observable snapshot { did, self, holdsLease, head, records } (keys only)
// apply-write -> gated by canWrite(remotePubKeyHex): device.applyWrite(op) into the shared log
// canWrite is an allowlist of authorized client Noise pubkeys (config PEER_WRITERS), null = no
// remote writers (secure default), exactly as lib/peer.js. The device NEVER hands over the shared
// signing key; a caller only requests the write, the device owns the intent-log.
import b4a from 'b4a'
import c from 'compact-encoding'
import ProtomuxRPC from 'protomux-rpc'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { conflictDiag } from './conflict-watch.js'
const DEVICE_PROTOCOL = 'hyper-proto-peer/device/v1'
// Methods a service-auth token may be minted for. ONE entry deliberately: this is the narrowest thing
// that makes listNotifications work, and every addition widens what a minted token can do at the AppView.
const SERVICE_AUTH_LXM = ['app.bsky.notification.listNotifications']
const SERVICE_AUTH_TTL_SEC = 120
export function serveDevice (device, swarm, { canWrite = null, blobs = null, sessions = null } = {}) {
const serverKey = b4a.toString(swarm.keyPair.publicKey, 'hex')
const onconn = (conn) => {
const rk = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
console.log(JSON.stringify({ msg: 'device control connection', from: rk, did: device.did }))
conn.on('close', () => console.log(JSON.stringify({ msg: 'device control closed', from: rk })))
const rpc = new ProtomuxRPC(conn, { protocol: DEVICE_PROTOCOL, valueEncoding: c.json })
// get-state: always available — read-only observability (record keys, lease, head). No secret.
rpc.respond('get-state', async () => device.state())
// conflict-diag: hypercore conflicts seen since boot. Same public read surface as get-state — it
// carries discovery keys, lengths and counts, no record content and no key material.
rpc.respond('conflict-diag', async () => conflictDiag())
// members-diag: read-only membership/admission diagnostic. reproject-members: force-re-project
// __members__ (writable creator only). Read/idempotent; same public surface as get-state.
rpc.respond('members-diag', async () => device.membersDiag())
rpc.respond('reproject-members', async () => device.reprojectMembers())
// get-repo-car: export the signed repo as an atproto CAR (getRepo shape) — what a PEER fetches for
// fork recovery (reconcileFromCar into a fresh line). Public read (like get-state/get-blob). Bytes hex.
rpc.respond('get-repo-car', async () => { const r = await device.exportCar(); return { car: b4a.toString(r.carBytes, 'hex'), commitCid: r.commitCid, dataCid: r.dataCid, rev: r.rev, recordCount: r.recordCount } })
// get-blob: PUBLIC read (bsky fetches avatars/embedded images unauthenticated) — the counterpart to the
// writer-gated upload-blob. The keyless bridge relays this (peer-front.getBlob) and streams the bytes to
// bsky; server-bare.js expects { mimeType, bytes(hex) } or null (→ 404). This responder was MISSING:
// blobs could be uploaded but never read back, so every getBlob 404'd regardless of who held the bytes.
if (blobs) {
// list-blobs: the DID's published blob inventory. READ-ONLY and ungated like get-blob — it returns
// content addresses the peer already serves by CID, so it grants no access that get-blob does not.
// This is what makes ORPHAN detection possible: a blob referenced by no record cannot be found by
// walking records, only by asking the authority what it has (spec/convergence.md §6).
rpc.respond('list-blobs', async ({ limit, cursor } = {}) => {
if (typeof blobs.list !== 'function') return { cids: [], cursor: null, skipped: 'not-supported' }
return blobs.list({ limit, cursor })
})
rpc.respond('get-blob', async ({ cid } = {}) => {
if (!cid) throw new Error('get-blob: cid required')
const blob = await blobs.get(String(cid))
if (!blob) return null
return { cid: String(cid), mimeType: blob.mimeType, size: blob.size, bytes: b4a.toString(blob.bytes, 'hex') }
})
}
if (canWrite) {
// pointer-diag: WRITER-GATED, unlike get-state/members-diag above. It is a READ, but it hashes every
// record file, so publishing it to any peer that knows the DID would hand out a free
// CPU-exhaustion trigger — the same reason reindex-overlay stays UDS-only. Gating it on the existing
// trusted-key allowlist is what lets an operator ask a REMOTE node "do your bytes match your
// pointers?" (hyper-proto-peer diag --remote) without an ssh session, while keeping the cost behind
// a key. `limit` bounds the entries returned; the walk itself is bounded by the record set.
rpc.respond('pointer-diag', async (op = {}) => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not run pointer-diag for ${device.did}`)
return device.pointerDiag(op || {})
})
// get-service-auth — mint a short-lived ES256K service-auth JWT so the KEYLESS membrane can make a
// USER-SCOPED AppView read AS this account. This is the missing link in a chain that was otherwise
// built end to end: atproto-did-peer declares the RPC (did-peer.js:379), the bridge's server-bare
// wires a token cache and calls front.getServiceAuth, and proxyToAppView attaches the Bearer — but
// nothing here answered, so every user-scoped read silently fell back to the public proxy. That is
// why app.bsky.notification.listNotifications never worked.
//
// TWO GATES, both required. They answer different questions and neither is sufficient:
// canWrite(remote) — may this PEER ask at all? (PEER_WRITERS Noise allowlist; this
// handler is only REGISTERED when configured, so it fails closed
// by not existing, matching pointer-diag/apply-write above)
// sessions.verifyAccess(token) — is a real USER logged in? The bridge forwards the end user's
// bearer precisely so this can be checked. Without it an
// allowlisted bridge could mint tokens for anyone at any time with
// nobody present.
//
// A minted token authenticates AS the DID, signed by the same #atproto key that signs every repo
// commit (verified against both hosted DIDs' PLC documents). So it is a signing oracle, and beyond
// the two gates it is constrained: lxm REQUIRED and allowlisted (the primitive treats lxm as
// optional, and a token without one is valid for ANY method at that audience), aud must be a DID,
// TTL capped short. Widening the allowlist is a visible edit here, not a caller's choice.
//
// REFUSES when sessions are unconfigured (canLogin false). That is correct rather than a limitation:
// wj62/e2jev are accounts with appPasswordHash+jwtSecret, while a co-owned community feed like
// loser-club has neither because nobody logs in AS the feed — so there is no user whose
// notifications these would be, and it keeps the public unauthenticated proxy, which is right for a
// feed. (A community inbox would want the WRITER SET as its authority, not a bearer — separate
// decision.)
rpc.respond('get-service-auth', async (op = {}) => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not mint service auth for ${device.did}`)
if (!sessions || !sessions.canLogin) throw new Error(`get-service-auth: sessions not configured for ${device.did} — refusing to mint ungated`)
const { aud, lxm, token } = op || {}
if (typeof aud !== 'string' || !aud.startsWith('did:')) throw new Error('get-service-auth: aud must be a service DID')
if (typeof lxm !== 'string' || !lxm) throw new Error('get-service-auth: lxm required (a token without one is valid for ANY method)')
if (!SERVICE_AUTH_LXM.includes(lxm)) throw new Error(`get-service-auth: lxm "${lxm}" is not allowlisted`)
sessions.verifyAccess(token) // throws unless a valid, in-scope access JWT for THIS did
const now = Math.floor(Date.now() / 1000)
// Log the GRANT, never the token.
console.log(JSON.stringify({ msg: 'service-auth minted (device surface)', did: device.did, aud, lxm, from: remote.slice(0, 12), ttlSec: SERVICE_AUTH_TTL_SEC }))
return device.getServiceAuth({ aud, lxm, exp: now + SERVICE_AUTH_TTL_SEC })
})
rpc.respond('apply-write', async (op) => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not write to ${device.did}`)
return device.applyWrite(op)
})
// delegate: writer-gated (same allowlist as apply-write). Authorize a browser sub-key for DID
// web sign-in — the #atproto key signs a delegation the keyless verifier (favela) checks vs PLC.
// Privileged (a new device authorization), so it rides the canWrite allowlist, like apply-write.
rpc.respond('delegate', async (op) => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not delegate for ${device.did}`)
return device.delegate(op)
})
// device registry (Option 3): list / revoke / query the DID's delegated browser sub-keys. Records
// live as reserved StateBase entries (never public); revocation is a replicated tombstone the
// keyless verifier (favela login) enforces. All writer-gated (same allowlist as delegate).
rpc.respond('list-devices', async () => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not list devices for ${device.did}`)
return { devices: await device.listDevices() }
})
rpc.respond('revoke-device', async (op) => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not revoke devices for ${device.did}`)
if (!op || !op.subKey) throw new Error('revoke-device: subKey required')
return op.restore ? device.restoreDevice(op.subKey) : device.revokeDevice(op.subKey, { reason: op.reason || null })
})
rpc.respond('is-device-revoked', async (op) => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not query devices for ${device.did}`)
if (!op || !op.subKey) throw new Error('is-device-revoked: subKey required')
return { revoked: await device.isDeviceRevoked(op.subKey) }
})
// upload-blob: writer-gated (same allowlist as apply-write). Stores content-addressed bytes into
// the peer's shared BlobStore so get-blob (membrane surface) can serve them. bytes are hex in the
// c.json RPC value. Returns { cid, mimeType, size } — the caller builds the blob ref for a record.
if (blobs) {
rpc.respond('upload-blob', async ({ bytes, mimeType } = {}) => {
const remote = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null
if (!remote || !canWrite(remote)) throw new Error(`unauthorized: caller may not write to ${device.did}`)
if (!bytes) throw new Error('upload-blob: bytes (hex) required')
const { cid, size } = await blobs.put(b4a.from(bytes, 'hex'), mimeType || 'application/octet-stream')
return { cid: cid.toString(), mimeType: mimeType || 'application/octet-stream', size }
})
}
}
}
swarm.on('connection', onconn)
const keyTopic = swarm.keyPair.publicKey
const didTopic = discoveryTopic(device.did)
const dKey = swarm.join(keyTopic, { server: true, client: false })
const dDid = swarm.join(didTopic, { server: true, client: false })
return {
serverKey,
discoveryTopic: b4a.toString(didTopic, 'hex'),
ready: () => Promise.all([dKey.flushed(), dDid.flushed()]),
close: () => swarm.off('connection', onconn)
}
}
export { DEVICE_PROTOCOL }