BlueSky Firehose Stuff

hose.rb

require "faye/websocket"
require "cbor"
require "base32"

SubscribeReposEndpoint = "wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos"

# Decode CBOR data into JSON as it appears
def decode_cbor_data(data)
  CBOR::Unpacker.new(StringIO.new(data))
    .each.to_a
end

# if you want to understand what is happening here (decoding the CID of a commit), read
# https://ipld.io/specs/codecs/dag-cbor/spec/
def decode_cid(data)
  data
    &.value
    &.[](1..-1)
    &.then { Base32.encode(_1) }
    &.downcase&.gsub(/=+$/, "")
    &.then { "b" + _1 }
end

EM.run do
  Firehose = Faye::WebSocket::Client.new(SubscribeReposEndpoint)

  Firehose.on :open do |event|
    puts "Connected to WebSocket endpoint"
  end

  Firehose.on :message do |event|
    begin # So we can catch exceptions 
      # Decode CBOR data into JSON
      cbor_data = event.data.pack("C*") # turn the bytes into a string
      json_data = decode_cbor_data(cbor_data) # decode the string

      # for every commit that appears
      if json_data[0].is_a?(Hash) && json_data[0]["op"] && json_data[0]["t"] == "#commit"
        puts "Received commit"
        ops = json_data[1]["ops"]
        ops.each do |op|
          puts "#{decode_cid(op["cid"])}"                       # Print the CID
          puts op["action"]                                     # Print the action (create, update, delete)
          puts "at://#{json_data[1]["repo"]}/#{op["path"]}"     # Print the at:// uri
          if op["path"].split("/").first == "app.bsky.feed.post"
            puts "It's a post!"                                 # Print a message for every post!
          end
          puts                                                  # Print a blank line
        end
      end
    rescue => e
      puts "Error: #{e.message}"
    end
  end

  Firehose.on :close do |event|
    puts "WebSocket connection closed (Code: #{event.code}, Reason: #{event.reason})"
    EM.stop
  end

  Firehose.on :error do |event|
    puts "WebSocket error: #{event.message}"
    EM.stop
  end
end