| … | … | @@ -21,11 +21,13 @@ |
| 21 | 21 | import process from 'bare-process' |
| 22 | 22 | import os from 'bare-os' |
| 23 | 23 | import { join } from 'bare-path' |
| 24 | import { planRecord, selectImages, samePost, MAX_IMAGE_BYTES } from './lib/plan.js' | |
| 25 | import { planProfile, sameProfile } from './lib/profile.js' | |
| 26 | 24 | import { FavelaPeer } from './lib/peer.js' |
| 27 | import { BskySink, POST_COLLECTION } from './lib/bsky.js' | |
| 28 | import { tidFor } from './lib/tid.js' | |
| 25 | import { BskySink } from './lib/bsky.js' | |
| 26 | // ⛔ THE PASS ITSELF LIVES IN lib/, WITH NO TRANSPORT IN IT. It has to run in two places — here | |
| 27 | // against unix sockets, and inside the handset worklet, where bsky records are authored and SIGNED | |
| 28 | // next to the DID key because that key is on the phone. It used to live in this file next to argv | |
| 29 | // parsing and process.exit, so the only way to run it on a phone was to write it a second time. | |
| 30 | import { projectCycle } from './lib/project.js' | |
| 29 | 31 | |
| 30 | 32 | const argv = process.argv.slice(2) |
| 31 | 33 | const SEND = argv.includes('--send') |
| … | … | @@ -78,227 +80,29 @@ async function main () { |
| 78 | 80 | await cycle(SEND) |
| 79 | 81 | } |
| 80 | 82 | |
| 83 | ||
| 84 | // ⛔ THIS FILE OWNS THE PROCESS, lib/project.js owns the projection. The split is exactly the exit | |
| 85 | // code: a refusal to send must end a CLI run non-zero and must NOT end a worklet, where exiting kills | |
| 86 | // the app the person is holding. | |
| 81 | 87 | async function cycle (send) { |
| 82 | 88 | const peer = new FavelaPeer(PEER_SOCK) |
| 83 | 89 | const bsky = new BskySink(BSKY_SOCK) |
| 84 | ||
| 85 | // ⛔ The identity's own preference decides, not this process's configuration. Read fresh so a | |
| 86 | // toggle in settings takes effect on the next cycle rather than on the next restart. | |
| 87 | const reach = await peer.reach() | |
| 88 | const { feed, rows } = await peer.ownPosts() | |
| 89 | console.log('favela-bridge · feed ' + feed.slice(0, 12) + ' · ' + rows.length + ' own post(s)') | |
| 90 | console.log(' room type ' + ROOM_TYPE + (process.env.FAVELA_ROOM_TYPE ? '' : ' (assumed)')) | |
| 91 | console.log(' bsky sink ' + (bsky.configured ? BSKY_SOCK : 'not configured')) | |
| 92 | console.log(' enabled ' + (reach.unknown | |
| 93 | ? 'UNKNOWN — the peer could not be asked, so nothing will be published' | |
| 94 | : (reach.networks.length ? reach.networks.join(', ') : 'nothing (turn a network on in settings)'))) | |
| 95 | console.log(' mode ' + (send ? 'SEND' : 'dry run — nothing will be published')) | |
| 96 | ||
| 97 | // ⛔ Ask what is already there ALWAYS, not only when sending. A dry run whose whole job is to say | |
| 98 | // what would happen must not overstate it — the first version skipped this read and reported | |
| 99 | // "would send 6" for a repo that already held two of them. | |
| 100 | // | |
| 101 | // A null answer means we could not find out, which is different from an empty repo, and the | |
| 102 | // difference is a duplicate of every post. | |
| 103 | const found = bsky.configured ? await bsky.have() : new Set() | |
| 104 | const have = found === null ? new Set() : found | |
| 105 | if (send && found === null) { | |
| 106 | console.error(' refusing to send: could not read what the bsky sink already holds') | |
| 107 | process.exit(1) | |
| 108 | } | |
| 109 | if (found === null) console.log(' ⚠ could not read the sink — "already there" is unknown below') | |
| 110 | ||
| 111 | // ⛔ THE PROFILE IS PROJECTED TOO, and it never was. favela-bridge carried posts and nothing else, | |
| 112 | // so a display name, bio, avatar or banner reached the fediverse (ap-bridge has projected all four | |
| 113 | // from the start) and stopped — the bsky profile stayed empty with nothing reporting a failure, | |
| 114 | // because there was no failure. The projection had simply never been written. | |
| 115 | await reconcileProfile(peer, bsky, reach, send) | |
| 116 | ||
| 117 | let sent = 0; let skipped = 0; let withheld = 0; let held = 0 | |
| 118 | for (const row of rows) { | |
| 119 | const plan = planRecord(row, { roomType: ROOM_TYPE, verb: 'post' }) | |
| 120 | const line = ' ' + row.name + ' [' + plan.networks.join(',') + ']' | |
| 121 | ||
| 122 | // ⛔ CONSENT FIRST, then capability. The reach lattice says what a room MAY reach; this says | |
| 123 | // what its owner has ASKED for. A post may be publishable and still must not be published. | |
| 124 | if (!reach.networks.includes('bsky')) { | |
| 125 | withheld++ | |
| 126 | console.log(line + ' bsky: not enabled for this identity') | |
| 127 | } else if (!plan.bsky.project) { | |
| 128 | withheld++ | |
| 129 | console.log(line + ' bsky: withheld — ' + plan.bsky.losses.join('; ')) | |
| 130 | } else if (have.has(tidFor(row.name))) { | |
| 131 | if (!REPROJECT) { | |
| 132 | skipped++ | |
| 133 | console.log(line + ' bsky: already there') | |
| 134 | } else { | |
| 135 | const { embed, losses: imgLosses } = await buildEmbed(peer, bsky, plan.bsky.images) | |
| 136 | for (const l of imgLosses) console.log(line + ' bsky: ' + l) | |
| 137 | const desired = embed ? { ...plan.bsky.record, embed } : plan.bsky.record | |
| 138 | const rkey = tidFor(row.name) | |
| 139 | const actual = await bsky.getRecord(POST_COLLECTION, rkey) | |
| 140 | if (actual && actual.unknown) { | |
| 141 | console.log(line + ' bsky: could not read what is there — not touching it') | |
| 142 | } else if (samePost(desired, actual)) { | |
| 143 | skipped++ | |
| 144 | console.log(line + ' bsky: already matches') | |
| 145 | } else if (!send) { | |
| 146 | console.log(line + ' bsky: would REPROJECT ' + rkey + (embed ? ' (+' + embed.images.length + ' image)' : '')) | |
| 147 | } else { | |
| 148 | const r = await bsky.putRecord(POST_COLLECTION, rkey, desired) | |
| 149 | if (r.sent) { sent++; console.log(line + ' bsky: REPROJECTED ' + rkey + (embed ? ' (+' + embed.images.length + ' image)' : '')) } | |
| 150 | else console.log(line + ' bsky: reproject FAILED — ' + r.error) | |
| 151 | } | |
| 152 | } | |
| 153 | } else if (send && sent >= LIMIT) { | |
| 154 | // Reported per record rather than silently truncated: a run that stopped early and one that | |
| 155 | // had nothing left to do print the same summary otherwise. | |
| 156 | held++ | |
| 157 | console.log(line + ' bsky: held — --limit ' + LIMIT + ' reached this cycle') | |
| 158 | } else if (!send) { | |
| 159 | console.log(line + ' bsky: would send ' + JSON.stringify(plan.bsky.record.text.slice(0, 48)) + | |
| 160 | (plan.bsky.images.length ? ' + ' + plan.bsky.images.length + ' image(s)' : '') + | |
| 161 | (plan.bsky.losses.length ? ' — losing: ' + plan.bsky.losses.join('; ') : '')) | |
| 162 | } else { | |
| 163 | const { embed, losses: imgLosses } = await buildEmbed(peer, bsky, plan.bsky.images) | |
| 164 | for (const l of imgLosses) console.log(line + ' bsky: ' + l) | |
| 165 | const record = embed ? { ...plan.bsky.record, embed } : plan.bsky.record | |
| 166 | const r = await bsky.send(row.name, record) | |
| 167 | if (r.sent) { sent++; console.log(line + ' bsky: SENT ' + r.rkey + (plan.bsky.losses.length ? ' — lost: ' + plan.bsky.losses.join('; ') : '')) } | |
| 168 | else console.log(line + ' bsky: FAILED — ' + r.error) | |
| 169 | } | |
| 170 | ||
| 171 | // Reported, never delivered, in v1. Silence here would read as "the fediverse got it". | |
| 172 | if (plan.fediverse.project) console.log(' fediverse: planned (' + plan.fediverse.content.length + ' bytes of html) — delivery unwired') | |
| 173 | else console.log(' fediverse: withheld — ' + plan.fediverse.losses.join('; ')) | |
| 174 | } | |
| 175 | ||
| 176 | console.log(' ' + (send ? 'sent ' + sent : 'would send ' + (rows.length - withheld - skipped)) + | |
| 177 | ' · already there ' + skipped + ' · withheld ' + withheld + | |
| 178 | (held ? ' · held by --limit ' + held : '')) | |
| 179 | } | |
| 180 | ||
| 181 | /** | |
| 182 | * Bring the bsky profile into line with favela's, and say what it did. | |
| 183 | * | |
| 184 | * ⛔ RECONCILIATION, not an event. There is no "the profile changed" signal to subscribe to here — | |
| 185 | * the cycle already reads the peer every pass, so the honest shape is to compare what SHOULD be | |
| 186 | * there against what IS and act on the difference. Comparing against the repo rather than a local | |
| 187 | * note of what we last sent is the part that matters: a projector that trusts its own memory drifts | |
| 188 | * from the thing it claims to describe, which is how a follower relationship existed in our file and | |
| 189 | * on no server. | |
| 190 | */ | |
| 191 | const blobCache = new Map() // oid -> blob ref. Content-addressed, so this can never go stale. | |
| 192 | ||
| 193 | /** | |
| 194 | * Turn the oids planBsky named into an `app.bsky.embed.images`, uploading what is missing. | |
| 195 | * | |
| 196 | * ⛔ THE PICTURE IS WHY THE POST EXISTS, quite often — and until now every one of them reached the | |
| 197 | * fediverse and none reached bsky, reported as "bsky needs an uploaded blob ref, which is unbuilt". | |
| 198 | * It was unbuilt because nothing here could upload; hpp has always had upload-blob. | |
| 199 | * | |
| 200 | * ⚠ A post whose images ALL fail still goes, as text. The alternative is withholding a post because | |
| 201 | * a picture would not upload, which loses more than it protects. | |
| 202 | */ | |
| 203 | async function buildEmbed (peer, bsky, imgs) { | |
| 204 | if (!imgs || !imgs.length) return { embed: null, losses: [] } | |
| 205 | const losses0 = [] | |
| 206 | ||
| 207 | // ⛔ SELECT, NEVER DOWNSAMPLE (SOCIAL-MODEL §6.6.1). The pyramid was built once at authoring; a | |
| 208 | // bridge picks the level nearest its ceiling. Re-encoding here would do the work per delivery on | |
| 209 | // every bridge AND produce bytes the author never made — §8's rule, applied to pixels. | |
| 210 | // | |
| 211 | // ⚠ The peer does the arithmetic against a ceiling WE supply, so there is one ladder rather than | |
| 212 | // one per bridge. A blob with no pointer answers with its single level, which is every blob | |
| 213 | // written before today. | |
| 214 | const described = [] | |
| 215 | for (const im of imgs) { | |
| 216 | let chosen = null | |
| 217 | try { chosen = await peer.call('blob-level', { oid: im.oid, maxBytes: MAX_IMAGE_BYTES }) } catch { chosen = null } | |
| 218 | if (!chosen || !chosen.level) { | |
| 219 | // ⛔ A CLEAR ABSENCE, not a silent drop: "no level fits" and "the fetch failed" are different | |
| 220 | // facts and the author is entitled to know which happened. | |
| 221 | losses0.push('image omitted — ' + ((chosen && chosen.reason) || 'no level could be chosen')) | |
| 222 | continue | |
| 223 | } | |
| 224 | // ★ THE HONESTY REQUIREMENT. A bridge that quietly serves a smaller picture has altered what | |
| 225 | // was published, so the level it chose is reported even when everything worked. | |
| 226 | if (chosen.reason && chosen.reason !== 'full size' && chosen.reason !== 'single level') { | |
| 227 | losses0.push('image: ' + chosen.reason) | |
| 228 | } | |
| 229 | let f = null | |
| 230 | try { f = await peer.call('file-get', { oid: chosen.level.oid }) } catch { f = null } | |
| 231 | described.push({ ...im, path: (f && f.path) || null, mimeType: f && f.mimeType, size: f && f.size }) | |
| 232 | } | |
| 233 | ||
| 234 | const { images, losses: sizeLosses } = selectImages(described) | |
| 235 | const losses = losses0.concat(sizeLosses) | |
| 236 | const out = [] | |
| 237 | for (const d of images) { | |
| 238 | let ref = blobCache.get(d.oid) | |
| 239 | if (!ref) { | |
| 240 | ref = await bsky.uploadBlob(d.path, d.mimeType) | |
| 241 | if (ref) blobCache.set(d.oid, ref) | |
| 242 | } | |
| 243 | // ⛔ Alt text travels. It is an accessibility obligation and the author already wrote it. | |
| 244 | if (ref) out.push({ alt: d.alt || '', image: ref }) | |
| 245 | else losses.push('an image could not be uploaded — omitted from this post') | |
| 246 | } | |
| 247 | ||
| 248 | if (!out.length) return { embed: null, losses } | |
| 249 | return { embed: { $type: 'app.bsky.embed.images', images: out }, losses } | |
| 250 | } | |
| 251 | ||
| 252 | async function reconcileProfile (peer, bsky, reach, send) { | |
| 253 | if (!bsky.configured) return | |
| 254 | // ⛔ CONSENT FIRST, exactly as for a post. A profile is published content too. | |
| 255 | if (!reach.networks.includes('bsky')) { console.log(' profile bsky: not enabled for this identity'); return } | |
| 256 | ||
| 257 | let prof = null | |
| 258 | try { prof = await peer.call('profile') } catch { prof = null } | |
| 259 | if (!prof) { console.log(' profile could not be read from the peer — leaving bsky alone'); return } | |
| 260 | ||
| 261 | // An oid names a picture the peer holds; this is where it becomes a file on disk. | |
| 262 | const describe = async (oid) => { | |
| 263 | if (!oid) return null | |
| 264 | try { | |
| 265 | const r = await peer.call('avatar-get', { oid }) | |
| 266 | return (r && r.path) ? { oid, path: r.path, mimeType: r.mimeType, size: r.size } : null | |
| 267 | } catch { return null } | |
| 268 | } | |
| 269 | ||
| 270 | const plan = planProfile({ | |
| 271 | name: prof.name, | |
| 272 | bio: prof.bio, | |
| 273 | avatar: await describe(prof.avatar), | |
| 274 | banner: await describe(prof.banner) | |
| 90 | const r = await projectCycle({ | |
| 91 | peer, | |
| 92 | bsky, | |
| 93 | roomType: ROOM_TYPE, | |
| 94 | roomTypeNote: process.env.FAVELA_ROOM_TYPE ? '' : ' (assumed)', | |
| 95 | sinkLabel: BSKY_SOCK, | |
| 96 | send, | |
| 97 | limit: LIMIT, | |
| 98 | reproject: REPROJECT, | |
| 99 | log: (line) => console.log(line) | |
| 275 | 100 | }) |
| 276 | for (const l of plan.losses) console.log(' profile ' + l) | |
| 277 | if (!plan.project) { console.log(' profile nothing published on favela — leaving bsky alone'); return } | |
| 278 | ||
| 279 | const actual = await bsky.getProfile() | |
| 280 | // ⛔ "COULD NOT ASK" IS NOT "ABSENT". Writing because a read failed would republish the profile on | |
| 281 | // every transient socket error, and each write is a repo version. | |
| 282 | if (actual && actual.unknown) { console.log(' profile could not read what bsky holds — not writing'); return } | |
| 283 | ||
| 284 | const record = { ...plan.record } | |
| 285 | for (const [field, img] of Object.entries(plan.images)) { | |
| 286 | let ref = blobCache.get(img.oid) | |
| 287 | if (!ref) { | |
| 288 | // ⚠ Only uploaded when we do not already hold its ref: the bytes are content-addressed, so a | |
| 289 | // re-upload is idempotent, but it is still a 512 KiB round trip per cycle if done blindly. | |
| 290 | ref = await bsky.uploadBlob(img.path, img.mimeType) | |
| 291 | if (ref) blobCache.set(img.oid, ref) | |
| 292 | } | |
| 293 | if (ref) record[field] = ref | |
| 294 | else console.log(' profile ' + field + ' could not be uploaded — omitted from this write') | |
| 101 | if (r.refused && send) { | |
| 102 | console.error(' refusing to send: ' + r.refused) | |
| 103 | process.exit(1) | |
| 295 | 104 | } |
| 296 | ||
| 297 | if (sameProfile(record, actual)) { console.log(' profile bsky: already matches'); return } | |
| 298 | if (!send) { console.log(' profile bsky: would update (' + Object.keys(record).filter((k) => k !== '$type').join(', ') + ')'); return } | |
| 299 | ||
| 300 | const r = await bsky.putProfile(record) | |
| 301 | console.log(' profile bsky: ' + (r.sent ? 'UPDATED (' + Object.keys(record).filter((k) => k !== '$type').join(', ') + ')' : 'FAILED — ' + r.error)) | |
| 105 | return r | |
| 302 | 106 | } |
| 303 | 107 | |
| 304 | 108 | main().catch((err) => { console.error('favela-bridge: ' + ((err && err.message) || err)); process.exit(1) }) |
| … | … | @@ -0,0 +1,236 @@ |
| 1 | // One projection pass, with no transport in it. | |
| 2 | // | |
| 3 | // ⛔ THIS RUNS IN TWO PLACES AND MUST BE ONE IMPLEMENTATION. On a desktop the peer and the bsky sink | |
| 4 | // are unix sockets; on a handset both are in the same worklet, because bsky records are authored and | |
| 5 | // SIGNED next to the DID key and that key is on the phone. A projection written twice drifts, and this | |
| 6 | // monorepo already carries six copies of one signing format, three of which eat the first character of | |
| 7 | // a hash. So the loop takes `peer` and `bsky` as objects and never learns which it got — the same | |
| 8 | // shape favela-host's inProcessClient uses to hand favela-surface a daemon without a socket. | |
| 9 | // | |
| 10 | // ⚠ THE FEDIVERSE NEEDS NO EQUIVALENT, and that asymmetry is the reason this file exists. ap-bridge | |
| 11 | // HOLDS the feed and projects server-side, so a registered name federates without anything running on | |
| 12 | // the device. bsky cannot work that way: the membrane is keyless and re-serves only what the device | |
| 13 | // signed. A handset with a correct DID, a resolving handle and nine fediverse posts therefore had an | |
| 14 | // atproto repo containing one bootstrap record — nothing was broken, nothing was ever running. | |
| 15 | // | |
| 16 | // ⛔ NO process.exit AND NO console. A refusal is returned, not shouted: this is called from a CLI that | |
| 17 | // owns the exit code and from a worklet where exiting kills the app. | |
| 18 | ||
| 19 | import { planRecord, MAX_IMAGE_BYTES, selectImages } from './plan.js' | |
| 20 | import { planProfile, sameProfile } from './profile.js' | |
| 21 | import { samePost } from './plan.js' | |
| 22 | import { tidFor } from './tid.js' | |
| 23 | import { POST_COLLECTION } from './bsky.js' | |
| 24 | ||
| 25 | // oid -> blob ref. Content-addressed, so this can never go stale. | |
| 26 | const blobCache = new Map() | |
| 27 | ||
| 28 | /** | |
| 29 | * Turn the oids planBsky named into an `app.bsky.embed.images`, uploading what is missing. | |
| 30 | * | |
| 31 | * ⚠ A post whose images ALL fail still goes, as text. Withholding a post because a picture would not | |
| 32 | * upload loses more than it protects. | |
| 33 | */ | |
| 34 | export async function buildEmbed (peer, bsky, imgs, { maxImageBytes = MAX_IMAGE_BYTES } = {}) { | |
| 35 | if (!imgs || !imgs.length) return { embed: null, losses: [] } | |
| 36 | const losses0 = [] | |
| 37 | ||
| 38 | // ⛔ SELECT, NEVER DOWNSAMPLE (SOCIAL-MODEL §6.6.1). The pyramid was built once at authoring; a | |
| 39 | // bridge picks the level nearest its ceiling. Re-encoding here would do the work per delivery on | |
| 40 | // every bridge AND produce bytes the author never made. | |
| 41 | const described = [] | |
| 42 | for (const im of imgs) { | |
| 43 | let chosen = null | |
| 44 | try { chosen = await peer.call('blob-level', { oid: im.oid, maxBytes: maxImageBytes }) } catch { chosen = null } | |
| 45 | if (!chosen || !chosen.level) { | |
| 46 | losses0.push('image omitted — ' + ((chosen && chosen.reason) || 'no level could be chosen')) | |
| 47 | continue | |
| 48 | } | |
| 49 | // ★ THE HONESTY REQUIREMENT. A bridge that quietly serves a smaller picture has altered what was | |
| 50 | // published, so the level it chose is reported even when everything worked. | |
| 51 | if (chosen.reason && chosen.reason !== 'full size' && chosen.reason !== 'single level') { | |
| 52 | losses0.push('image: ' + chosen.reason) | |
| 53 | } | |
| 54 | let f = null | |
| 55 | try { f = await peer.call('file-get', { oid: chosen.level.oid }) } catch { f = null } | |
| 56 | described.push({ ...im, path: (f && f.path) || null, mimeType: f && f.mimeType, size: f && f.size }) | |
| 57 | } | |
| 58 | ||
| 59 | const { images, losses: sizeLosses } = selectImages(described) | |
| 60 | const losses = losses0.concat(sizeLosses) | |
| 61 | const out = [] | |
| 62 | for (const d of images) { | |
| 63 | let ref = blobCache.get(d.oid) | |
| 64 | if (!ref) { | |
| 65 | ref = await bsky.uploadBlob(d.path, d.mimeType) | |
| 66 | if (ref) blobCache.set(d.oid, ref) | |
| 67 | } | |
| 68 | // ⛔ Alt text travels. It is an accessibility obligation and the author already wrote it. | |
| 69 | if (ref) out.push({ alt: d.alt || '', image: ref }) | |
| 70 | else losses.push('an image could not be uploaded — omitted from this post') | |
| 71 | } | |
| 72 | ||
| 73 | if (!out.length) return { embed: null, losses } | |
| 74 | return { embed: { $type: 'app.bsky.embed.images', images: out }, losses } | |
| 75 | } | |
| 76 | ||
| 77 | /** | |
| 78 | * Bring the bsky profile into line with favela's. | |
| 79 | * | |
| 80 | * ⛔ RECONCILIATION, not an event. Comparing against the REPO rather than a local note of what was | |
| 81 | * last sent is the part that matters: a projector that trusts its own memory drifts from the thing it | |
| 82 | * claims to describe. | |
| 83 | */ | |
| 84 | export async function reconcileProfile (peer, bsky, reach, send, log) { | |
| 85 | if (!bsky.configured) return { done: false, reason: 'no sink' } | |
| 86 | // ⛔ CONSENT FIRST, exactly as for a post. A profile is published content too. | |
| 87 | if (!reach.networks.includes('bsky')) { log(' profile bsky: not enabled for this identity'); return { done: false, reason: 'not enabled' } } | |
| 88 | ||
| 89 | let prof = null | |
| 90 | try { prof = await peer.call('profile') } catch { prof = null } | |
| 91 | if (!prof) { log(' profile could not be read from the peer — leaving bsky alone'); return { done: false, reason: 'unreadable' } } | |
| 92 | ||
| 93 | const describe = async (oid) => { | |
| 94 | if (!oid) return null | |
| 95 | try { | |
| 96 | const r = await peer.call('avatar-get', { oid }) | |
| 97 | return (r && r.path) ? { oid, path: r.path, mimeType: r.mimeType, size: r.size } : null | |
| 98 | } catch { return null } | |
| 99 | } | |
| 100 | ||
| 101 | const plan = planProfile({ | |
| 102 | name: prof.name, | |
| 103 | bio: prof.bio, | |
| 104 | avatar: await describe(prof.avatar), | |
| 105 | banner: await describe(prof.banner) | |
| 106 | }) | |
| 107 | for (const l of plan.losses) log(' profile ' + l) | |
| 108 | if (!plan.project) { log(' profile nothing published on favela — leaving bsky alone'); return { done: false, reason: 'nothing to publish' } } | |
| 109 | ||
| 110 | const actual = await bsky.getProfile() | |
| 111 | // ⛔ "COULD NOT ASK" IS NOT "ABSENT". Writing because a read failed republishes on every transient | |
| 112 | // error, and each write is a repo version. | |
| 113 | if (actual && actual.unknown) { log(' profile could not read what bsky holds — not writing'); return { done: false, reason: 'unknown' } } | |
| 114 | ||
| 115 | const record = { ...plan.record } | |
| 116 | for (const [field, img] of Object.entries(plan.images)) { | |
| 117 | let ref = blobCache.get(img.oid) | |
| 118 | if (!ref) { | |
| 119 | ref = await bsky.uploadBlob(img.path, img.mimeType) | |
| 120 | if (ref) blobCache.set(img.oid, ref) | |
| 121 | } | |
| 122 | if (ref) record[field] = ref | |
| 123 | else log(' profile ' + field + ' could not be uploaded — omitted from this write') | |
| 124 | } | |
| 125 | ||
| 126 | const fields = () => Object.keys(record).filter((k) => k !== '$type').join(', ') | |
| 127 | if (sameProfile(record, actual)) { log(' profile bsky: already matches'); return { done: false, reason: 'already matches' } } | |
| 128 | if (!send) { log(' profile bsky: would update (' + fields() + ')'); return { done: false, reason: 'dry run' } } | |
| 129 | ||
| 130 | const r = await bsky.putProfile(record) | |
| 131 | log(' profile bsky: ' + (r.sent ? 'UPDATED (' + fields() + ')' : 'FAILED — ' + r.error)) | |
| 132 | return { done: !!r.sent, error: r.error || null } | |
| 133 | } | |
| 134 | ||
| 135 | /** | |
| 136 | * One full pass: read consent, reconcile the profile, then project the posts that reach bsky. | |
| 137 | * | |
| 138 | * ⛔ REFUSES RATHER THAN GUESSES. `have()` answering null means we could not find out what the repo | |
| 139 | * holds, which is NOT an empty repo — and the difference is a duplicate of every post. A send run in | |
| 140 | * that state returns `{ refused }` and writes nothing. | |
| 141 | */ | |
| 142 | export async function projectCycle ({ | |
| 143 | peer, bsky, roomType = 'public', send = false, limit = Infinity, | |
| 144 | reproject = false, maxImageBytes = MAX_IMAGE_BYTES, log = () => {}, | |
| 145 | // Two header details the CLI knows and this function cannot: whether the room type was CONFIGURED | |
| 146 | // or assumed, and which socket the sink is. Reported because "public (assumed)" is the input the | |
| 147 | // reach lattice cannot infer, and if it is ever wrong it is wrong in the direction of a leak. | |
| 148 | roomTypeNote = '', sinkLabel = null | |
| 149 | } = {}) { | |
| 150 | const reach = await peer.reach() | |
| 151 | const { feed, rows } = await peer.ownPosts() | |
| 152 | ||
| 153 | log('favela-bridge · feed ' + String(feed).slice(0, 12) + ' · ' + rows.length + ' own post(s)') | |
| 154 | log(' room type ' + roomType + roomTypeNote) | |
| 155 | log(' bsky sink ' + (bsky.configured ? (sinkLabel || 'configured') : 'not configured')) | |
| 156 | log(' enabled ' + (reach.unknown | |
| 157 | ? 'UNKNOWN — the peer could not be asked, so nothing will be published' | |
| 158 | : (reach.networks.length ? reach.networks.join(', ') : 'nothing (turn a network on in settings)'))) | |
| 159 | log(' mode ' + (send ? 'SEND' : 'dry run — nothing will be published')) | |
| 160 | ||
| 161 | const found = bsky.configured ? await bsky.have() : new Set() | |
| 162 | const have = found === null ? new Set() : found | |
| 163 | if (send && found === null) { | |
| 164 | log(' refusing to send: could not read what the bsky sink already holds') | |
| 165 | return { refused: 'could not read what the bsky sink already holds', feed, rows: rows.length, sent: 0, skipped: 0, withheld: 0, held: 0 } | |
| 166 | } | |
| 167 | if (found === null) log(' ⚠ could not read the sink — "already there" is unknown below') | |
| 168 | ||
| 169 | const profile = await reconcileProfile(peer, bsky, reach, send, log) | |
| 170 | ||
| 171 | let sent = 0; let skipped = 0; let withheld = 0; let held = 0 | |
| 172 | for (const row of rows) { | |
| 173 | const plan = planRecord(row, { roomType, verb: 'post' }) | |
| 174 | const line = ' ' + row.name + ' [' + plan.networks.join(',') + ']' | |
| 175 | ||
| 176 | // ⛔ CONSENT FIRST, then capability. The reach lattice says what a room MAY reach; this says what | |
| 177 | // its owner has ASKED for. A post may be publishable and still must not be published. | |
| 178 | if (!reach.networks.includes('bsky')) { | |
| 179 | withheld++ | |
| 180 | log(line + ' bsky: not enabled for this identity') | |
| 181 | } else if (!plan.bsky.project) { | |
| 182 | withheld++ | |
| 183 | log(line + ' bsky: withheld — ' + plan.bsky.losses.join('; ')) | |
| 184 | } else if (have.has(tidFor(row.name))) { | |
| 185 | if (!reproject) { | |
| 186 | skipped++ | |
| 187 | log(line + ' bsky: already there') | |
| 188 | } else { | |
| 189 | const { embed, losses: imgLosses } = await buildEmbed(peer, bsky, plan.bsky.images, { maxImageBytes }) | |
| 190 | for (const l of imgLosses) log(line + ' bsky: ' + l) | |
| 191 | const desired = embed ? { ...plan.bsky.record, embed } : plan.bsky.record | |
| 192 | const rkey = tidFor(row.name) | |
| 193 | const actual = await bsky.getRecord(POST_COLLECTION, rkey) | |
| 194 | if (actual && actual.unknown) { | |
| 195 | log(line + ' bsky: could not read what is there — not touching it') | |
| 196 | } else if (samePost(desired, actual)) { | |
| 197 | skipped++ | |
| 198 | log(line + ' bsky: already matches') | |
| 199 | } else if (!send) { | |
| 200 | log(line + ' bsky: would REPROJECT ' + rkey + (embed ? ' (+' + embed.images.length + ' image)' : '')) | |
| 201 | } else { | |
| 202 | const r = await bsky.putRecord(POST_COLLECTION, rkey, desired) | |
| 203 | if (r.sent) { sent++; log(line + ' bsky: REPROJECTED ' + rkey + (embed ? ' (+' + embed.images.length + ' image)' : '')) } | |
| 204 | else log(line + ' bsky: reproject FAILED — ' + r.error) | |
| 205 | } | |
| 206 | } | |
| 207 | } else if (send && sent >= limit) { | |
| 208 | held++ | |
| 209 | log(line + ' bsky: held — limit ' + limit + ' reached this cycle') | |
| 210 | } else if (!send) { | |
| 211 | log(line + ' bsky: would send ' + JSON.stringify(plan.bsky.record.text.slice(0, 48)) + | |
| 212 | (plan.bsky.images.length ? ' + ' + plan.bsky.images.length + ' image(s)' : '') + | |
| 213 | (plan.bsky.losses.length ? ' — losing: ' + plan.bsky.losses.join('; ') : '')) | |
| 214 | } else { | |
| 215 | const { embed, losses: imgLosses } = await buildEmbed(peer, bsky, plan.bsky.images, { maxImageBytes }) | |
| 216 | for (const l of imgLosses) log(line + ' bsky: ' + l) | |
| 217 | const record = embed ? { ...plan.bsky.record, embed } : plan.bsky.record | |
| 218 | const r = await bsky.send(row.name, record) | |
| 219 | if (r.sent) { sent++; log(line + ' bsky: SENT ' + r.rkey + (plan.bsky.losses.length ? ' — lost: ' + plan.bsky.losses.join('; ') : '')) } | |
| 220 | else log(line + ' bsky: FAILED — ' + r.error) | |
| 221 | } | |
| 222 | ||
| 223 | // ⚠ Reported, never delivered, from this side. ap-bridge projects the fediverse from the feed it | |
| 224 | // HOLDS, so silence here would read as "the fediverse got nothing" when it is simply not our job. | |
| 225 | if (plan.fediverse.project) log(' fediverse: planned (' + plan.fediverse.content.length + ' bytes of html) — delivered by ap-bridge from the held feed') | |
| 226 | else log(' fediverse: withheld — ' + plan.fediverse.losses.join('; ')) | |
| 227 | } | |
| 228 | ||
| 229 | log(' ' + (send ? 'sent ' + sent : 'would send ' + (rows.length - withheld - skipped)) + | |
| 230 | ' · already there ' + skipped + ' · withheld ' + withheld + | |
| 231 | (held ? ' · held by limit ' + held : '')) | |
| 232 | ||
| 233 | return { refused: null, feed, rows: rows.length, sent, skipped, withheld, held, profile, reach } | |
| 234 | } | |
| 235 | ||
| 236 | export default { projectCycle, reconcileProfile, buildEmbed } |
| … | … | @@ -0,0 +1,120 @@ |
| 1 | // The projection pass, with no transport under it. | |
| 2 | // | |
| 3 | // ⛔ WHY IT IS EXTRACTED. This loop has to run in two places: on a desktop against unix sockets, and | |
| 4 | // inside the handset worklet, because bsky records are authored and SIGNED next to the DID key and | |
| 5 | // that key is on the phone. It lived inside index.js next to argv parsing, a socket path and | |
| 6 | // process.exit, so the only way to run it on a phone was to write it a second time. | |
| 7 | // | |
| 8 | // These pin the behaviours that must survive the move — the ones where getting it wrong is silent. | |
| 9 | ||
| 10 | import test from 'brittle' | |
| 11 | import { projectCycle } from '../lib/project.js' | |
| 12 | import { BskySink } from '../lib/bsky.js' | |
| 13 | import { tidFor } from '../lib/tid.js' | |
| 14 | ||
| 15 | const at = '2026-09-07T23:12:04.600Z' | |
| 16 | const row = (name, body) => ({ name, body, createdAt: at }) | |
| 17 | ||
| 18 | function fakePeer ({ networks = ['bsky'], rows = [], profile = null, unknown = false } = {}) { | |
| 19 | return { | |
| 20 | reach: async () => (unknown ? { networks: [], credential: null, unknown: true } : { networks, credential: null }), | |
| 21 | ownPosts: async () => ({ feed: 'f'.repeat(64), rows }), | |
| 22 | call: async (cmd) => { | |
| 23 | if (cmd === 'profile') return profile | |
| 24 | return null | |
| 25 | } | |
| 26 | } | |
| 27 | } | |
| 28 | ||
| 29 | function fakeSink ({ has = [], haveNull = false } = {}) { | |
| 30 | const written = [] | |
| 31 | const records = new Map() | |
| 32 | const client = { | |
| 33 | call: async (cmd, args = {}) => { | |
| 34 | if (cmd === 'get-state') { | |
| 35 | if (haveNull) throw new Error('cannot read') | |
| 36 | return { records: has.map((rkey) => ({ collection: 'app.bsky.feed.post', rkey })) } | |
| 37 | } | |
| 38 | if (cmd === 'apply-write') { written.push(args); records.set(args.collection + '/' + args.rkey, args.record); return { ok: true } } | |
| 39 | if (cmd === 'get-record') return records.get(args.collection + '/' + args.rkey) || null | |
| 40 | throw new Error('unexpected ' + cmd) | |
| 41 | } | |
| 42 | } | |
| 43 | return { sink: new BskySink(null, { connect: async () => client }), written, records } | |
| 44 | } | |
| 45 | ||
| 46 | test('★★★ consent gates every post — an identity with bsky off publishes nothing', async (t) => { | |
| 47 | const { sink, written } = fakeSink() | |
| 48 | const r = await projectCycle({ | |
| 49 | peer: fakePeer({ networks: ['fediverse'], rows: [row('1788822724600', 'hello')] }), | |
| 50 | bsky: sink, send: true | |
| 51 | }) | |
| 52 | t.is(r.withheld, 1, 'the post is withheld') | |
| 53 | t.is(r.sent, 0) | |
| 54 | t.is(written.length, 0, 'and nothing whatsoever was written') | |
| 55 | }) | |
| 56 | ||
| 57 | test('★★★ an unreadable repo REFUSES a send rather than duplicating everything', async (t) => { | |
| 58 | // ⛔ `have()` answering null means "could not find out", which is NOT an empty repo. Treating the | |
| 59 | // two alike re-sends every post the repo already holds. | |
| 60 | const { sink, written } = fakeSink({ haveNull: true }) | |
| 61 | const r = await projectCycle({ | |
| 62 | peer: fakePeer({ rows: [row('1788822724600', 'hello')] }), | |
| 63 | bsky: sink, send: true | |
| 64 | }) | |
| 65 | t.ok(r.refused, 'the run refuses: ' + r.refused) | |
| 66 | t.is(written.length, 0, 'and writes nothing') | |
| 67 | }) | |
| 68 | ||
| 69 | test('★★ a post already in the repo is not sent twice', async (t) => { | |
| 70 | const name = '1788822724600' | |
| 71 | const { sink, written } = fakeSink({ has: [tidFor(name)] }) | |
| 72 | const r = await projectCycle({ peer: fakePeer({ rows: [row(name, 'hello')] }), bsky: sink, send: true }) | |
| 73 | t.is(r.skipped, 1, 'recognised as already there') | |
| 74 | t.is(r.sent, 0) | |
| 75 | t.is(written.length, 0) | |
| 76 | }) | |
| 77 | ||
| 78 | test('★★ a public post is sent, under a TID rkey', async (t) => { | |
| 79 | const name = '1788822724600' | |
| 80 | const { sink, written } = fakeSink() | |
| 81 | const r = await projectCycle({ peer: fakePeer({ rows: [row(name, 'hello')] }), bsky: sink, send: true }) | |
| 82 | t.is(r.sent, 1) | |
| 83 | t.is(written.length, 1) | |
| 84 | t.is(written[0].collection, 'app.bsky.feed.post') | |
| 85 | t.is(written[0].rkey, tidFor(name), 'the AppView silently drops posts whose rkey is not a TID') | |
| 86 | t.is(written[0].record.text, 'hello') | |
| 87 | }) | |
| 88 | ||
| 89 | test('★★ a dry run writes nothing but still reports what it would do', async (t) => { | |
| 90 | const { sink, written } = fakeSink() | |
| 91 | const r = await projectCycle({ peer: fakePeer({ rows: [row('1788822724600', 'hello')] }), bsky: sink, send: false }) | |
| 92 | t.is(written.length, 0, 'nothing published') | |
| 93 | t.is(r.sent, 0) | |
| 94 | }) | |
| 95 | ||
| 96 | test('★★★ a members room reaches nothing, even with bsky switched on', async (t) => { | |
| 97 | // The reach lattice, not the switch, is what makes this safe: consent says where the owner WANTS to | |
| 98 | // publish; the room says what may leave at all. A private room must not escape a switched-on network. | |
| 99 | const { sink, written } = fakeSink() | |
| 100 | const r = await projectCycle({ | |
| 101 | peer: fakePeer({ networks: ['bsky'], rows: [row('1788822724600', 'secret')] }), | |
| 102 | bsky: sink, send: true, roomType: 'members' | |
| 103 | }) | |
| 104 | t.is(r.withheld, 1, 'withheld by the room, not by the switch') | |
| 105 | t.is(written.length, 0, 'and nothing left the device') | |
| 106 | }) | |
| 107 | ||
| 108 | test('★★ the profile is projected, and only when it differs', async (t) => { | |
| 109 | const { sink, written } = fakeSink() | |
| 110 | const peer = fakePeer({ rows: [], profile: { name: 'Fleeky', bio: 'hello' } }) | |
| 111 | ||
| 112 | const first = await projectCycle({ peer, bsky: sink, send: true }) | |
| 113 | t.ok(first.profile.done, 'written on the first pass') | |
| 114 | const profileWrites = written.filter((w) => w.collection === 'app.bsky.actor.profile') | |
| 115 | t.is(profileWrites.length, 1) | |
| 116 | t.is(profileWrites[0].record.displayName, 'Fleeky') | |
| 117 | ||
| 118 | const second = await projectCycle({ peer, bsky: sink, send: true }) | |
| 119 | t.absent(second.profile.done, 'and NOT rewritten when it already matches — each write is a repo version') | |
| 120 | }) |