bridge

tcmts.dev.devenv.bridge · #pyhuzza6 · mjs file

// Stand-ins, fed only by the real PDS, for the two services dev-env doesn't have:
// - Jetstream (v2's network.bsky.jetstream.subscribeEvents, xrpc.v1.json frames): the PDS's own firehose
//   (com.atproto.sync.subscribeRepos), decoded by @atproto/sync (commits unchecked: its resolver won't fetch an http PLC;
//   the registry checks each record against its signed repository itself), then sent as Jetstream's JSON;
// - a relay's com.atproto.sync.listReposByCollection (collectiondir's): from the PDS's listRepos and describeRepo.
import { Firehose } from '@atproto/sync'
import { IdResolver } from '@atproto/identity'
import { lexToJson } from '@atproto/lex-json'
import { createHash } from 'node:crypto'
import http from 'node:http'
import fs from 'node:fs'

const envFile = new URL('./env.json', import.meta.url)
const env = JSON.parse(fs.readFileSync(envFile, 'utf8'))
const stream = []
const listening = new Set()
const frame = (text) => {
  const data = Buffer.from(text)
  const n = data.length
  const head = n < 126 ? Buffer.from([0x81, n]) : n < 65536 ? Buffer.from([0x81, 126, n >> 8, n & 0xff]) : (() => { const b = Buffer.alloc(10); b[0] = 0x81; b[1] = 127; b.writeBigUInt64BE(BigInt(n), 2); return b })()
  return Buffer.concat([head, data])
}
const firehose = new Firehose({
  idResolver: new IdResolver({ plcUrl: env.plc }),
  service: env.pds.replace(/^http/, 'ws'),
  getCursor: () => 0,
  // (the PLC is http here, which @atproto/sync's resolver refuses: commits pass unchecked, and the registry checks each record itself)
  unauthenticatedCommits: true,
  excludeIdentity: true, excludeAccount: true, excludeSync: true,
  onError: (e) => console.error('firehose:', e.message, e.cause?.message ?? e.cause, e.cause?.stack?.split('\n').slice(0,4).join(' / ')),
  handleEvent: (e) => {
    if (!('collection' in e) || !e.collection.startsWith('codes.tcm.') && e.collection !== 'app.bsky.graph.follow') return
    const p = {
      $type: 'network.bsky.jetstream.subscribeEvents#commit', seq: e.seq, did: e.did, time: e.time, witnessedAt: e.time,
      rev: e.rev, operation: e.event, collection: e.collection, rkey: e.rkey,
      ...(e.event === 'delete' ? {} : { record: lexToJson(e.record), cid: e.cid.toString() }),
    }
    stream.push(p)
    console.log(`event ${p.seq} ${p.operation} ${p.did} ${p.collection}/${p.rkey}`)
    for (const l of listening) if (l.wants(p)) l.socket.write(frame(JSON.stringify({ $type: 'message', payload: p })))
  },
})
firehose.start()

const server = http.createServer(async (req, res) => {
  const url = new URL(req.url, 'http://x')
  if (url.pathname === '/xrpc/com.atproto.sync.listReposByCollection') {
    const want = url.searchParams.get('collection')
    const repos = []
    const list = await (await fetch(`${env.pds}/xrpc/com.atproto.sync.listRepos?limit=1000`)).json()
    for (const r of list.repos) {
      const d = await (await fetch(`${env.pds}/xrpc/com.atproto.repo.describeRepo?repo=${r.did}`)).json()
      if (d.collections?.includes(want)) repos.push({ did: r.did })
    }
    res.writeHead(200, { 'content-type': 'application/json' })
    return res.end(JSON.stringify({ repos }))
  }
  res.writeHead(404).end()
})
server.on('upgrade', (req, socket) => {
  const url = new URL(req.url, 'http://x')
  if (url.pathname !== '/xrpc/network.bsky.jetstream.subscribeEvents') return socket.destroy()
  const accept = createHash('sha1').update(`${req.headers['sec-websocket-key']}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64')
  const asked = String(req.headers['sec-websocket-protocol'] ?? '').split(',').map((x) => x.trim())
  socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n${asked.includes('xrpc.v1.json') ? 'Sec-WebSocket-Protocol: xrpc.v1.json\r\n' : ''}\r\n`)
  const collections = url.searchParams.getAll('collections')
  const cursor = Number(url.searchParams.get('cursor') ?? NaN)
  const wants = (p) => !collections.length || collections.some((c) => c.endsWith('.*') ? p.collection.startsWith(c.slice(0, -1)) : p.collection === c)
  const since = (p) => cursor >= 1e15 ? Date.parse(p.time) * 1000 >= cursor : p.seq > cursor
  console.log(`subscriber: cursor ${url.searchParams.get('cursor')} ${collections}`)
  if (!Number.isNaN(cursor)) for (const p of stream) if (wants(p) && since(p)) socket.write(frame(JSON.stringify({ $type: 'message', payload: p })))
  const l = { socket, wants }
  listening.add(l)
  socket.on('data', (b) => {
    const op = b[0] & 0x0f
    if (op === 0x8) socket.end(Buffer.from([0x88, 0]))
    else if (op === 0x9) { const len = b[1] & 0x7f; const mask = b.subarray(2, 6); const data = Buffer.from(b.subarray(6, 6 + len).map((x, i) => x ^ mask[i % 4])); socket.write(Buffer.concat([Buffer.from([0x8a, data.length]), data])) }
  })
  socket.on('close', () => listening.delete(l))
  socket.on('error', () => listening.delete(l))
})
server.listen(0, '127.0.0.1', () => {
  const now = JSON.parse(fs.readFileSync(envFile, 'utf8'))
  now.bridge = `http://127.0.0.1:${server.address().port}`
  fs.writeFileSync(envFile, JSON.stringify(now, null, 2))
  console.log(`bridge at ${now.bridge}`)
})

used by

• tcmts.dev.atprotoTrial