firehose.js

'use strict';

const { cborToLexRecord, readCar } = require('@atproto/repo');
const { Subscription } = require('@atproto/xrpc-server');

const SERVICE = 'bsky.network';

const subscription = new Subscription({
  service: `wss://${SERVICE}`,
  method: 'com.atproto.sync.subscribeRepos',
  getState: () => ({}),
  validate: (value) => value,
});

(async () => {
  for await (const event of subscription) {
    const car = await readCar(event.blocks);

    for (const op of event.ops) {
      if (op.action !== 'create') {
        continue;
      }

      const recordBytes = car.blocks.get(op.cid)
      if (!recordBytes) {
        continue;
      }

      const record = cborToLexRecord(recordBytes);

      const collection = op.path.split('/')[0];
      if (collection !== 'app.bsky.feed.post') {
        continue;
      }

      console.log(record.text);
    }
  }
})();