| … | … | @@ -193,8 +193,11 @@ export async function projectBsky (node, hpp, { send = false, limit = Infinity, |
| 193 | 193 | * daemon was the only thing that sent (hyper-social #77 §5, change 3). |
| 194 | 194 | * |
| 195 | 195 | * ⛔ EVENT-DRIVEN, NOT POLLED. The bridge polled every 15 s; this runs one pass at boot and one per |
| 196 | * change to the node's own feed (autobee emits 'update' when its view moves). Coalesced: never two | |
| 197 | * passes at once, and any number of changes during a pass become exactly one more. | |
| 196 | * append to any WRITER core of the node's own feed — ap-bridge FeedHolder's trigger for held favela | |
| 197 | * feeds. ⛔ NOT autobee's 'update': in 2.12.1 it fires only on an interrupt (index.js:676), never when the | |
| 198 | * view moves, and the 0.1.1 watch that listened for it projected once at boot and never again | |
| 199 | * (tabertron, 2026-10-13). Writers are re-hooked on every fire, so a device admitted later is covered. | |
| 200 | * Coalesced: never two passes at once, and any number of changes during a pass become exactly one more. | |
| 198 | 201 | * |
| 199 | 202 | * ⚠ IT SENDS, as the deployed `--watch` did. Consent is not decided here: projectCycle reads the |
| 200 | 203 | * node's reach and withholds anything a network is not switched on for. |
| … | … | @@ -208,6 +211,7 @@ export async function watchProjection (node, atproto, { log = () => {}, project |
| 208 | 211 | |
| 209 | 212 | let running = false |
| 210 | 213 | let again = false |
| 214 | let lastRefusal = null | |
| 211 | 215 | let stopped = false |
| 212 | 216 | const pass = async () => { |
| 213 | 217 | if (stopped) return |
| … | … | @@ -217,18 +221,43 @@ export async function watchProjection (node, atproto, { log = () => {}, project |
| 217 | 221 | do { |
| 218 | 222 | again = false |
| 219 | 223 | // ⛔ One pass NEVER ends the watch — a sink respawning on OTA, a peer restarting, are ordinary. |
| 220 | try { await project(node, hpp, { send: true, log }) } catch (err) { log('projection pass failed: ' + ((err && err.message) || err)) } | |
| 224 | try { | |
| 225 | const r = await project(node, hpp, { send: true, log }) | |
| 226 | // ⛔ projectBsky refuses WITHOUT logging when it cannot pick an identity, and on a node with | |
| 227 | // bsky on that silence means nothing reaches bsky. Said once per distinct reason. | |
| 228 | if (r && r.refused && r.refused !== lastRefusal) log('projection refused: ' + r.refused) | |
| 229 | lastRefusal = r && r.refused ? r.refused : null | |
| 230 | } catch (err) { log('projection pass failed: ' + ((err && err.message) || err)) } | |
| 221 | 231 | } while (again && !stopped) |
| 222 | 232 | } finally { running = false } |
| 223 | 233 | } |
| 224 | 234 | |
| 225 | 235 | const base = node.peer && node.peer.feed && node.peer.feed.base |
| 226 | const onUpdate = () => { pass() } | |
| 227 | if (base && base.on) base.on('update', onUpdate) | |
| 236 | const hooked = new Map() // writer core -> listener | |
| 237 | const rehook = () => { | |
| 238 | let writers | |
| 239 | try { writers = base && base.writers } catch { return } | |
| 240 | if (!writers) return | |
| 241 | try { | |
| 242 | for (const w of writers) { | |
| 243 | const core = w && w.core | |
| 244 | if (!core || typeof core.on !== 'function' || hooked.has(core)) continue | |
| 245 | const onAppend = () => { rehook(); pass() } | |
| 246 | core.on('append', onAppend) | |
| 247 | hooked.set(core, onAppend) | |
| 248 | } | |
| 249 | } catch { /* a writer set mid-change is not a failure; the next fire re-hooks */ } | |
| 250 | } | |
| 251 | rehook() | |
| 228 | 252 | pass() |
| 229 | 253 | return { |
| 230 | 254 | watching: true, |
| 231 | stop () { stopped = true; if (base && base.off) base.off('update', onUpdate) } | |
| 255 | hooked: () => hooked.size, | |
| 256 | stop () { | |
| 257 | stopped = true | |
| 258 | for (const [core, fn] of hooked) { try { core.off('append', fn) } catch {} } | |
| 259 | hooked.clear() | |
| 260 | } | |
| 232 | 261 | } |
| 233 | 262 | } |
| 234 | 263 |
| … | … | @@ -1,12 +1,12 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "favela-host", |
| 3 | "version": "0.1.1", | |
| 3 | "version": "0.1.2", | |
| 4 | 4 | "lockfileVersion": 3, |
| 5 | 5 | "requires": true, |
| 6 | 6 | "packages": { |
| 7 | 7 | "": { |
| 8 | 8 | "name": "favela-host", |
| 9 | "version": "0.1.1", | |
| 9 | "version": "0.1.2", | |
| 10 | 10 | "dependencies": { |
| 11 | 11 | "anproto": "file:../anproto", |
| 12 | 12 | "atproto-identity": "file:../atproto-identity", |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "favela-host", |
| 3 | "version": "0.1.1", | |
| 3 | "version": "0.1.2", | |
| 4 | 4 | "description": "THE favela app on a desktop, home node or server: a favela-peer node, favela-surface rendering it, the atproto signer and the bsky projection, in one process.", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "index.js", |
| … | … | @@ -13,8 +13,10 @@ import { watchProjection } from '../lib/project.js' |
| 13 | 13 | |
| 14 | 14 | const tick = () => new Promise((resolve) => setTimeout(resolve, 20)) |
| 15 | 15 | |
| 16 | // A fake feed whose single writer core emits 'append' — the event a real post produces. | |
| 16 | 17 | function fakeNode () { |
| 17 | const base = new EventEmitter() | |
| 18 | const core = new EventEmitter() | |
| 19 | const base = { writers: [{ core }], emit: (ev) => { if (ev === 'update') core.emit('append') } } | |
| 18 | 20 | return { node: { peer: { feed: { base } } }, base } |
| 19 | 21 | } |
| 20 | 22 | |
| … | … | @@ -81,3 +83,48 @@ test('stop() detaches from the feed', async (t) => { |
| 81 | 83 | base.emit('update'); await tick() |
| 82 | 84 | t.is(runs, 1, 'only the boot pass') |
| 83 | 85 | }) |
| 86 | ||
| 87 | // ⛔ AGAINST A REAL FEED. The tests above drive a fake emitter, and they passed while the watch was dead: | |
| 88 | // autobee 2.12.1 emits 'update' ONLY on an interrupt (index.js:676), never when the view moves, so on | |
| 89 | // tabertron the projection ran once at boot and never again. A post lands as an 'append' on a WRITER | |
| 90 | // core — the trigger ap-bridge's FeedHolder already uses for held favela feeds. | |
| 91 | test('a real post on a real favela feed triggers a projection pass', async (t) => { | |
| 92 | const { FavelaNode } = await import('favela-peer/lib/node.js') | |
| 93 | const { dispatch } = await import('favela-peer/lib/rpc.js') | |
| 94 | const { identity } = await import('favela-peer/test/helpers.js') | |
| 95 | const fs = (await import('bare-fs')).default | |
| 96 | const os = (await import('bare-os')).default | |
| 97 | const { join } = await import('bare-path') | |
| 98 | const storage = join(os.tmpdir(), 'fh-watch-real-' + Date.now() + '-' + Math.random().toString(16).slice(2)) | |
| 99 | fs.mkdirSync(storage, { recursive: true }) | |
| 100 | const n = new FavelaNode(storage, { transport: { async join () {}, async leave () {}, async destroy () {} } }) | |
| 101 | await n.ready() | |
| 102 | t.teardown(() => n.close()) | |
| 103 | const id = await identity() | |
| 104 | await n.link(await id.device()) | |
| 105 | ||
| 106 | let runs = 0 | |
| 107 | const w = await watchProjection(n, Promise.resolve({ nodes: [{}] }), { project: async () => { runs++ } }) | |
| 108 | t.teardown(() => w.stop()) | |
| 109 | await tick() | |
| 110 | t.is(runs, 1, 'the boot pass') | |
| 111 | ||
| 112 | await dispatch(n, 'post', { body: 'hello' }) | |
| 113 | for (let i = 0; i < 50 && runs < 2; i++) await tick() | |
| 114 | t.ok(runs >= 2, 'the post triggered a pass (runs=' + runs + ')') | |
| 115 | }) | |
| 116 | ||
| 117 | // ⛔ A REFUSAL IS SAID, ONCE. projectBsky returns { refused } without logging when it cannot choose an | |
| 118 | // identity — on tabertron (two DIDs, bsky on, none designated) every pass refused and the log said | |
| 119 | // nothing at all. Said once per distinct reason, so a refusing node does not spam every pass. | |
| 120 | test('a refused pass is logged once per reason', async (t) => { | |
| 121 | const { node, base } = fakeNode() | |
| 122 | const said = [] | |
| 123 | const w = await watchProjection(node, Promise.resolve({ nodes: [{}, {}] }), { project: async () => ({ refused: 'several identities, none designated' }), log: (m) => said.push(m) }) | |
| 124 | t.teardown(() => w.stop()) | |
| 125 | await tick() | |
| 126 | base.emit('update'); await tick() | |
| 127 | base.emit('update'); await tick() | |
| 128 | const lines = said.filter((m) => /several identities/.test(m)) | |
| 129 | t.is(lines.length, 1, 'said once, not per pass') | |
| 130 | }) |