| … | … | @@ -0,0 +1,186 @@ |
| 1 | // observation-feeds.js — what the fediverse did to a registered user's records, written into a feed the GATEWAY |
| 2 | // owns and the user's nodes HOLD (hyper-social #82, I1, decision D2). |
| 3 | // |
| 4 | // ⛔ AN OBSERVATION IS NOT AN EDGE (OBSERVED-INTERACTIONS §2). A like from a Mastodon account is not ours to |
| 5 | // author, and the user never authored it either — so it is not written into the user's feed. It goes into a |
| 6 | // feed the gateway founded under ITS OWN identity, one per user, which the user's nodes hold read-only like |
| 7 | // any other feed. Delivery is replication: an append reaches every node holding the feed, with no polling and |
| 8 | // no pickup verb — this replaces the drained buffer + `collect` design (OBSERVED §9) by operator decision D2. |
| 9 | // |
| 10 | // ⛔ THE GATEWAY HOLDS A FAVELA IDENTITY (operator decision O-1). Minted ONCE into `<storage>/observer/`: the |
| 11 | // vault (the recovery root) and the device credential, both 0600, like the AP actor keys beside them. It is |
| 12 | // never re-minted — a new root would make every node's pinned observer root wrong at once. |
| 13 | // |
| 14 | // ⚠ ONE CORESTORE NAMESPACE PER FEED, and never one a held feed uses. Two Autobees on one namespace derive the |
| 15 | // same local writer and DEADLOCK (favela-peer node.js). Held feeds are `feed:<key>`; these are |
| 16 | // `obs:<username>:<generation>`, so a released and re-claimed name gets a NEW feed and never inherits its |
| 17 | // predecessor's observations (the generation rule registrations.js already applies to actor ids). |
| 18 | // |
| 19 | // ⛔ PUBLIC MATERIAL ONLY. A feed replicates in the clear to anyone holding its key. Whatever is written here is |
| 20 | // readable by them; DMs and followers-only replies are kept OUT by the caller until an encrypted channel exists |
| 21 | // (operator decision, #82 D2). |
| 22 | |
| 23 | import fs from 'bare-fs' |
| 24 | import { join } from 'bare-path' |
| 25 | import crypto from 'bare-crypto' |
| 26 | import b4a from 'b4a' |
| 27 | import { Feed } from 'favela-peer/lib/feed-base.js' |
| 28 | import { RecordStore } from 'favela-peer/lib/record-store.js' |
| 29 | import { loadCredential, saveCredential, bindWriter } from 'favela-peer/lib/device.js' |
| 30 | import { mintIdentity } from 'favela-peer/lib/mint.js' |
| 31 | import { readProof } from 'favela-peer/lib/attestation.js' |
| 32 | |
| 33 | export const COLLECTION = 'observation' |
| 34 | export const KINDS = ['like', 'repost', 'follow', 'reply', 'mention'] |
| 35 | const MAX_BODY = 4000 |
| 36 | |
| 37 | // The record name for an activity: a content key, so a remote server re-delivering the same activity lands on |
| 38 | // the same record instead of a second one, and an Undo can find what it undoes from the activity id alone. |
| 39 | // |
| 40 | // ⚠ WHICH id the caller passes as `activity` is the identity of the observation. Like / Announce / Follow: the |
| 41 | // ACTIVITY id (the Undo names it). Reply / mention: the OBJECT id (the Note) — an Update arrives with a new |
| 42 | // activity id for the same object and must overwrite, and a Delete names only the object. |
| 43 | export function observationName (activityId) { |
| 44 | return crypto.createHash('sha256').update(String(activityId)).digest('hex').slice(0, 32) |
| 45 | } |
| 46 | |
| 47 | export class ObservationFeeds { |
| 48 | constructor ({ corestore, storage, overlay = null, log = () => {} } = {}) { |
| 49 | if (!corestore) throw new Error('ObservationFeeds: a corestore is required') |
| 50 | if (!storage) throw new Error('ObservationFeeds: a storage directory is required') |
| 51 | this.corestore = corestore |
| 52 | this.storage = storage |
| 53 | this.overlay = overlay |
| 54 | this._log = log |
| 55 | this._dir = join(storage, 'observer') |
| 56 | this._mapPath = join(this._dir, 'feeds.json') |
| 57 | this._feeds = new Map() // "<username>:<generation>" -> { key, feed } |
| 58 | this._opening = new Map() |
| 59 | this.cred = null |
| 60 | this.root = null |
| 61 | } |
| 62 | |
| 63 | async ready () { |
| 64 | fs.mkdirSync(this._dir, { recursive: true }) |
| 65 | let cred = loadCredential(this._dir) |
| 66 | if (!cred) { |
| 67 | const vaultPath = join(this._dir, 'vault.json') |
| 68 | // ⛔ mintIdentity REFUSES to overwrite a vault. A vault without a credential means a half-finished mint |
| 69 | // or a lost device.json: re-derive the credential from it rather than minting a different identity. |
| 70 | if (fs.existsSync(vaultPath)) { |
| 71 | const { mintFromVault } = await import('favela-peer/lib/device.js') |
| 72 | cred = await mintFromVault(vaultPath) |
| 73 | } else { |
| 74 | const minted = await mintIdentity({ vaultPath }) |
| 75 | cred = { |
| 76 | keyPair: { publicKey: b4a.from(minted.credential.publicKey, 'hex'), secretKey: b4a.from(minted.credential.secretKey, 'hex') }, |
| 77 | proof: minted.credential.proof |
| 78 | } |
| 79 | this._log({ msg: 'observer: minted the gateway identity (once)' }) |
| 80 | } |
| 81 | fs.chmodSync(vaultPath, 0o600) |
| 82 | saveCredential(this._dir, cred) |
| 83 | } |
| 84 | fs.chmodSync(join(this._dir, 'device.json'), 0o600) |
| 85 | const proven = readProof(cred.proof) |
| 86 | if (!proven || !proven.root) throw new Error('ObservationFeeds: the observer credential does not verify') |
| 87 | this.cred = cred |
| 88 | this.root = typeof proven.root === 'string' ? proven.root : b4a.toString(proven.root, 'hex') |
| 89 | return this |
| 90 | } |
| 91 | |
| 92 | _map () { |
| 93 | try { return JSON.parse(fs.readFileSync(this._mapPath, 'utf8')) || {} } catch { return {} } |
| 94 | } |
| 95 | |
| 96 | _saveMap (map) { |
| 97 | const tmp = this._mapPath + '.tmp' |
| 98 | fs.writeFileSync(tmp, JSON.stringify(map, null, 2) + '\n') |
| 99 | fs.renameSync(tmp, this._mapPath) |
| 100 | } |
| 101 | |
| 102 | /** Open (founding on first use) the observation feed for one registered user. Idempotent. → { key, feed } */ |
| 103 | async open (username, generation) { |
| 104 | if (!this.cred) throw new Error('ObservationFeeds: ready() first') |
| 105 | const user = String(username || '').toLowerCase() |
| 106 | if (!/^[a-z0-9_]+$/.test(user)) throw new Error('ObservationFeeds: a username is required') |
| 107 | const gen = Number(generation) | 0 |
| 108 | const id = user + ':' + gen |
| 109 | if (this._feeds.has(id)) return this._feeds.get(id) |
| 110 | if (this._opening.has(id)) return this._opening.get(id) |
| 111 | |
| 112 | const opening = (async () => { |
| 113 | const map = this._map() |
| 114 | const known = map[id] || null |
| 115 | const dir = join(this.storage, 'records', 'obs-' + user + '-' + gen) |
| 116 | const store = new RecordStore(dir, { overlay: this.overlay }) |
| 117 | const feed = new Feed(this.corestore.namespace('obs:' + id), known ? b4a.from(known, 'hex') : null, { |
| 118 | store, |
| 119 | // The same rule FeedHolder states: every favela node opens feeds with the doorbell ON, so a holder |
| 120 | // computes the same view as this writer does. |
| 121 | optimistic: true |
| 122 | }) |
| 123 | await feed.ready() |
| 124 | // Founding is a device admitting itself; on a reopen it is the same idempotent call favela-peer's node |
| 125 | // makes at every boot. |
| 126 | await feed.addDevice(bindWriter(this.cred, feed.deviceKey)) |
| 127 | await feed.update() |
| 128 | const key = b4a.toString(feed.key, 'hex') |
| 129 | if (known && known !== key) throw new Error('ObservationFeeds: ' + id + ' reopened as a different feed') |
| 130 | if (!known) { map[id] = key; this._saveMap(map) } |
| 131 | // Bytes written before a restart must be servable again (favela-peer node.js _seedOverlay). |
| 132 | store.reregisterLocal().catch(() => {}) |
| 133 | const entry = { key, feed } |
| 134 | this._feeds.set(id, entry) |
| 135 | return entry |
| 136 | })() |
| 137 | this._opening.set(id, opening) |
| 138 | try { return await opening } finally { this._opening.delete(id) } |
| 139 | } |
| 140 | |
| 141 | /** The feed key for a user, or null if it was never opened. */ |
| 142 | key (username, generation) { |
| 143 | return this._map()[String(username).toLowerCase() + ':' + (Number(generation) | 0)] || null |
| 144 | } |
| 145 | |
| 146 | /** Every open feed — what the swarm replicates on each connection. */ |
| 147 | feeds () { return [...this._feeds.values()].map((e) => e.feed) } |
| 148 | |
| 149 | /** |
| 150 | * Record one observation. → its record name. |
| 151 | * |
| 152 | * ⛔ REFUSED rather than stored when it names no actor, no activity, or a kind outside the set: a row nobody |
| 153 | * can be attributed to, or that nothing could ever undo, is noise that reads later as data (observations.js). |
| 154 | */ |
| 155 | async note (username, generation, { kind, actor, subject = null, target = null, at = Date.now(), activity, body = '' } = {}) { |
| 156 | if (!KINDS.includes(kind)) throw new Error('ObservationFeeds.note: unknown kind "' + kind + '"') |
| 157 | if (typeof actor !== 'string' || !actor) throw new Error('ObservationFeeds.note: an actor is required') |
| 158 | if (typeof activity !== 'string' || !activity) throw new Error('ObservationFeeds.note: an activity id is required') |
| 159 | const { feed } = await this.open(username, generation) |
| 160 | const name = observationName(activity) |
| 161 | const frontmatter = { kind, actor, subject: subject || null, target: target || null, at: Number(at) || Date.now(), activity } |
| 162 | await feed.put(COLLECTION, name, { frontmatter, body: String(body || '').slice(0, MAX_BODY) }) |
| 163 | return name |
| 164 | } |
| 165 | |
| 166 | /** Remove the observation an activity produced (an Undo or a Delete). → true if there was one. */ |
| 167 | async remove (username, generation, activity) { |
| 168 | const { feed } = await this.open(username, generation) |
| 169 | const name = observationName(activity) |
| 170 | await feed.update() |
| 171 | if (!(await feed.get(COLLECTION, name))) return false |
| 172 | await feed.del(COLLECTION, name) |
| 173 | return true |
| 174 | } |
| 175 | |
| 176 | replicate (socket) { |
| 177 | for (const f of this.feeds()) { try { f.base.replicate(socket) } catch {} } |
| 178 | } |
| 179 | |
| 180 | async close () { |
| 181 | for (const { feed } of this._feeds.values()) { try { await feed.close() } catch {} } |
| 182 | this._feeds.clear() |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | export default { ObservationFeeds, observationName, COLLECTION, KINDS } |