atingester: Bluesky Syncing Tools

This library for syncing data from the bluesky network.

Forked from @atproto/sync

I used this code as a reference.

This tool can subscribes...

• Firehose is an authenticated stream of events used to efficiently sync user updates (posts, likes, follows, handle changes, etc).
• Jetstream is a streaming service that consumes Firehose and converts it into lightweight, friendly JSON.
• Turbostream is a real-time, hydrated repeater service built on top of Jetstream.

...in same functions.

Links

• JetstreamProxy is a server that proxies Jetstream. Using this with atingester can further reduce network traffic. !!! JetstreamProxy currently only supports legacy-v1. !!!
• feed-generator is an example of using atingester with cursor.
• flow-measurement-bot is an example of using atingester.

Usage

It's basically the same as here.

Make sure to execute await initIngester() exactly once before calling Ingester, Jetstream, or similar classes!

!!! You can set compress: true only in the legacy-v1 environment. !!!

Ingester ( Firehose / Jetstream legacy-v1, v1, v2 / Turbostream )

import { Ingester, initIngester } from 'atingester'
import { IdResolver } from '@atproto/identity'

const run = async () => {
  await initIngester()

  const ingester = new Ingester('Firehose', {
    idResolver: new IdResolver(),
    handleEvent: (evt) => {
      if (evt.event === 'create') {
        console.log(evt.record.text)
      }
    },
    onInfo: (info) => {
      console.info(info)
    },
    onError: (err: Error) => {
      console.error(err)
    },
    /*
    getCursor?: () => Awaited<number | undefined>
    runner?: EventRunner // should only set getCursor *or* runner
    */
    service: 'wss://bsky.network',
    subscriptionReconnectDelay: 3000,
    unauthenticatedCommits: false,
    unauthenticatedHandles: false,
    compress: true, // this set only in the legacy-v1 environment
    filterCollections: ['app.bsky.feed.post'],
    filterDids: ['did:plc:abcde....'],
    excludeIdentity: true,
    excludeAccount: true,
    excludeCommit: false,
    excludeSync: true,
  })

  ingester.start()

  await ingester.destroy()
}

run()

Firehose

import { Firehose, initIngester } from 'atingester'
import { IdResolver } from '@atproto/identity'

const run = async () => {
  await initIngester()

  const firehose = new Firehose({
    idResolver: new IdResolver(),
    handleEvent: (evt) => {
      if (evt.event === 'create') {
        console.log(evt.record.text)
      }
    },
    onInfo: (info) => {
      console.info(info)
    },
    onError: (err: Error) => {
      console.error(err)
    },
    service: 'wss://bsky.network',
    filterCollections: ['app.bsky.feed.post'],
    filterDids: ['did:plc:abcde....'],
    excludeIdentity: true,
    excludeAccount: true,
    excludeSync: true,
  })

  firehose.start()

  await firehose.destroy()
}

run()

Jetstream legacy-v1, v1

import { JetstreamV1, initIngester } from 'atingester'
import { IdResolver } from '@atproto/identity'

const run = async () => {
  await initIngester()

  const jetstreamV1 = new JetstreamV1({
    idResolver: new IdResolver(),
    handleEvent: (evt) => {
      if (evt.event === 'create') {
        console.log(evt.record.text)
      }
    },
    onInfo: (info) => {
      console.info(info)
    },
    onError: (err: Error) => {
      console.error(err)
    },
    service: 'wss://jetstream1.us-west.bsky.network',
    compress: true, // this set only in the legacy-v1 environment
    filterCollections: ['app.bsky.feed.post'],
    filterDids: ['did:plc:abcde....'],
    excludeIdentity: true,
    excludeAccount: true,
  })

  jetstreamV1.start()

  await jetstreamV1.destroy()
}

run()

Jetstream v2

!!! At this stage, only the Live Tail feature has been implemented. We expect to roll out the other new features in stages. !!!

Please obtain your apiKey here.

With a single subscription, you can register up to 100 collections and 10,000 DIDs. If you exceed these limits, please create multiple Ingesters and combine them.

import { Jetstream, initIngester } from 'atingester'
import { IdResolver } from '@atproto/identity'

const run = async () => {
  await initIngester()

  const jetstream = new Jetstream({
    idResolver: new IdResolver(),
    handleEvent: (evt) => {
      if (evt.event === 'create') {
        console.log(evt.record.text)
      }
    },
    onInfo: (info) => {
      console.info(info)
    },
    onError: (err: Error) => {
      console.error(err)
    },
    service: 'wss://jetstream.us-west.bsky.network',
    compress: true,
    filterCollections: ['app.bsky.feed.post'],
    filterDids: ['did:plc:abcde....'],
    excludeIdentity: true,
    excludeAccount: true,
  })

  jetstream.start()

  await jetstream.destroy()
}

run()

Turbostream

import { initIngester, Turbostream } from 'atingester'

const run = async () => {
  await initIngester()

  const turbostream = new Turbostream({
    handleEvent: (evt) => {
      if (evt.event === 'create') {
        console.log(evt.record.text)
      }
    },
    onInfo: (info) => {
      console.info(info)
    },
    onError: (err: Error) => {
      console.error(err)
    },
    service: 'wss://api.graze.social',
    filterDids: ['did:plc:abcde....'],
  })

  turbostream.start()

  await turbostream.destroy()
}

run()