// observation-feeds.js — what the fediverse did to a registered user's records, written into a feed the GATEWAY
// owns and the user's nodes HOLD (hyper-social #82, I1, decision D2).
//
// ⛔ AN OBSERVATION IS NOT AN EDGE (OBSERVED-INTERACTIONS §2). A like from a Mastodon account is not ours to
// author, and the user never authored it either — so it is not written into the user's feed. It goes into a
// feed the gateway founded under ITS OWN identity, one per user, which the user's nodes hold read-only like
// any other feed. Delivery is replication: an append reaches every node holding the feed, with no polling and
// no pickup verb — this replaces the drained buffer + `collect` design (OBSERVED §9) by operator decision D2.
//
// ⛔ THE GATEWAY HOLDS A FAVELA IDENTITY (operator decision O-1). Minted ONCE into `<storage>/observer/`: the
// vault (the recovery root) and the device credential, both 0600, like the AP actor keys beside them. It is
// never re-minted — a new root would make every node's pinned observer root wrong at once.
//
// ⚠ ONE CORESTORE NAMESPACE PER FEED, and never one a held feed uses. Two Autobees on one namespace derive the
// same local writer and DEADLOCK (favela-peer node.js). Held feeds are `feed:<key>`; these are
// `obs:<username>:<generation>`, so a released and re-claimed name gets a NEW feed and never inherits its
// predecessor's observations (the generation rule registrations.js already applies to actor ids).
//
// ⛔ PUBLIC MATERIAL ONLY. A feed replicates in the clear to anyone holding its key. Whatever is written here is
// readable by them; DMs and followers-only replies are kept OUT by the caller until an encrypted channel exists
// (operator decision, #82 D2).
import fs from 'bare-fs'
import { join } from 'bare-path'
import crypto from 'bare-crypto'
import b4a from 'b4a'
import { Feed } from 'favela-peer/lib/feed-base.js'
import { RecordStore } from 'favela-peer/lib/record-store.js'
import { loadCredential, saveCredential, bindWriter } from 'favela-peer/lib/device.js'
import { mintIdentity } from 'favela-peer/lib/mint.js'
import { readProof } from 'favela-peer/lib/attestation.js'
export const COLLECTION = 'observation'
export const KINDS = ['like', 'repost', 'follow', 'reply', 'mention']
const MAX_BODY = 4000
// The record name for an activity: a content key, so a remote server re-delivering the same activity lands on
// the same record instead of a second one, and an Undo can find what it undoes from the activity id alone.
//
// ⚠ WHICH id the caller passes as `activity` is the identity of the observation. Like / Announce / Follow: the
// ACTIVITY id (the Undo names it). Reply / mention: the OBJECT id (the Note) — an Update arrives with a new
// activity id for the same object and must overwrite, and a Delete names only the object.
export function observationName (activityId) {
return crypto.createHash('sha256').update(String(activityId)).digest('hex').slice(0, 32)
}
export class ObservationFeeds {
constructor ({ corestore, storage, overlay = null, swarm = null, log = () => {} } = {}) {
if (!corestore) throw new Error('ObservationFeeds: a corestore is required')
if (!storage) throw new Error('ObservationFeeds: a storage directory is required')
this.corestore = corestore
this.storage = storage
this.overlay = overlay
// ⛔ THE SAME SWARM as FeedHolder (the bridge's persistent keypair), never a second one: every connection
// replicates every observation feed, exactly as FeedHolder replicates every held feed.
this.swarm = swarm
if (swarm) swarm.on('connection', (socket) => this.replicate(socket))
this._log = log
this._dir = join(storage, 'observer')
this._mapPath = join(this._dir, 'feeds.json')
this._feeds = new Map() // "<username>:<generation>" -> { key, feed }
this._opening = new Map()
this.cred = null
this.root = null
}
async ready () {
fs.mkdirSync(this._dir, { recursive: true })
let cred = loadCredential(this._dir)
if (!cred) {
const vaultPath = join(this._dir, 'vault.json')
// ⛔ mintIdentity REFUSES to overwrite a vault. A vault without a credential means a half-finished mint
// or a lost device.json: re-derive the credential from it rather than minting a different identity.
if (fs.existsSync(vaultPath)) {
const { mintFromVault } = await import('favela-peer/lib/device.js')
cred = await mintFromVault(vaultPath)
} else {
const minted = await mintIdentity({ vaultPath })
cred = {
keyPair: { publicKey: b4a.from(minted.credential.publicKey, 'hex'), secretKey: b4a.from(minted.credential.secretKey, 'hex') },
proof: minted.credential.proof
}
this._log({ msg: 'observer: minted the gateway identity (once)' })
}
fs.chmodSync(vaultPath, 0o600)
saveCredential(this._dir, cred)
}
fs.chmodSync(join(this._dir, 'device.json'), 0o600)
const proven = readProof(cred.proof)
if (!proven || !proven.root) throw new Error('ObservationFeeds: the observer credential does not verify')
this.cred = cred
this.root = typeof proven.root === 'string' ? proven.root : b4a.toString(proven.root, 'hex')
return this
}
_map () {
try { return JSON.parse(fs.readFileSync(this._mapPath, 'utf8')) || {} } catch { return {} }
}
_saveMap (map) {
const tmp = this._mapPath + '.tmp'
fs.writeFileSync(tmp, JSON.stringify(map, null, 2) + '\n')
fs.renameSync(tmp, this._mapPath)
}
/** Open (founding on first use) the observation feed for one registered user. Idempotent. → { key, feed } */
async open (username, generation) {
if (!this.cred) throw new Error('ObservationFeeds: ready() first')
const user = String(username || '').toLowerCase()
if (!/^[a-z0-9_]+$/.test(user)) throw new Error('ObservationFeeds: a username is required')
const gen = Number(generation) | 0
const id = user + ':' + gen
if (this._feeds.has(id)) return this._feeds.get(id)
if (this._opening.has(id)) return this._opening.get(id)
const opening = (async () => {
const map = this._map()
const known = map[id] || null
const dir = join(this.storage, 'records', 'obs-' + user + '-' + gen)
const store = new RecordStore(dir, { overlay: this.overlay })
const feed = new Feed(this.corestore.namespace('obs:' + id), known ? b4a.from(known, 'hex') : null, {
store,
// The same rule FeedHolder states: every favela node opens feeds with the doorbell ON, so a holder
// computes the same view as this writer does.
optimistic: true
})
await feed.ready()
// Founding is a device admitting itself; on a reopen it is the same idempotent call favela-peer's node
// makes at every boot.
await feed.addDevice(bindWriter(this.cred, feed.deviceKey))
await feed.update()
const key = b4a.toString(feed.key, 'hex')
if (known && known !== key) throw new Error('ObservationFeeds: ' + id + ' reopened as a different feed')
if (!known) { map[id] = key; this._saveMap(map) }
// Bytes written before a restart must be servable again (favela-peer node.js _seedOverlay).
store.reregisterLocal().catch(() => {})
// Announce the feed and replicate it to whoever is already connected — a feed opened after boot would
// otherwise wait for a new connection that may never come (FeedHolder.hold, the same rule).
if (this.swarm) {
this.swarm.join(feed.discoveryKey, { server: true, client: true })
for (const socket of this.swarm.connections) { try { feed.base.replicate(socket) } catch {} }
}
const entry = { key, feed }
this._feeds.set(id, entry)
return entry
})()
this._opening.set(id, opening)
try { return await opening } finally { this._opening.delete(id) }
}
/** The feed key for a user, or null if it was never opened. */
key (username, generation) {
return this._map()[String(username).toLowerCase() + ':' + (Number(generation) | 0)] || null
}
/** Every open feed — what the swarm replicates on each connection. */
feeds () { return [...this._feeds.values()].map((e) => e.feed) }
/**
* Record one observation. → its record name.
*
* ⛔ REFUSED rather than stored when it names no actor, no activity, or a kind outside the set: a row nobody
* can be attributed to, or that nothing could ever undo, is noise that reads later as data (observations.js).
*/
async note (username, generation, { kind, actor, subject = null, target = null, at = Date.now(), activity, body = '' } = {}) {
if (!KINDS.includes(kind)) throw new Error('ObservationFeeds.note: unknown kind "' + kind + '"')
if (typeof actor !== 'string' || !actor) throw new Error('ObservationFeeds.note: an actor is required')
if (typeof activity !== 'string' || !activity) throw new Error('ObservationFeeds.note: an activity id is required')
const { feed } = await this.open(username, generation)
const name = observationName(activity)
const frontmatter = { kind, actor, subject: subject || null, target: target || null, at: Number(at) || Date.now(), activity }
await feed.put(COLLECTION, name, { frontmatter, body: String(body || '').slice(0, MAX_BODY) })
return name
}
/** Remove the observation an activity produced (an Undo or a Delete). → true if there was one. */
async remove (username, generation, activity) {
const { feed } = await this.open(username, generation)
const name = observationName(activity)
await feed.update()
if (!(await feed.get(COLLECTION, name))) return false
await feed.del(COLLECTION, name)
return true
}
replicate (socket) {
for (const f of this.feeds()) { try { f.base.replicate(socket) } catch {} }
}
async close () {
for (const { feed } of this._feeds.values()) { try { await feed.close() } catch {} }
this._feeds.clear()
}
}
export default { ObservationFeeds, observationName, COLLECTION, KINDS }