added skeleton of a firehose client from bluefeeds

Gemfile

···
 
 gem 'minisky', '~> 0.5'
 gem 'didkit', '~> 0.2', git: 'https://tangled.sh/@mackuba.eu/didkit'
+gem 'skyfall', '~> 0.6'
 
 gem 'base58'
 gem 'jwt'

Gemfile.lock

···
       minitest (>= 5.1)
       securerandom (>= 0.3)
       tzinfo (~> 2.0, >= 2.0.5)
+    base32 (0.3.4)
     base58 (0.2.3)
     base64 (0.3.0)
     bcrypt_pbkdf (1.1.1)
···
       net-sftp (>= 2.0.0)
       net-ssh (>= 2.0.14)
       net-ssh-gateway (>= 1.1.0)
+    cbor (0.5.10.1)
     concurrent-ruby (1.3.5)
     connection_pool (2.5.3)
     date (3.4.1)
     drb (2.2.3)
     ed25519 (1.4.0)
     erb (5.0.2)
+    eventmachine (1.2.7)
+    faye-websocket (0.12.0)
+      eventmachine (>= 0.12.0)
+      websocket-driver (>= 0.8.0)
     highline (3.1.2)
       reline
     i18n (1.14.7)
···
     sinatra-activerecord (2.0.28)
       activerecord (>= 4.1)
       sinatra (>= 1.0)
+    skyfall (0.6.0)
+      base32 (~> 0.3, >= 0.3.4)
+      base64 (~> 0.1)
+      cbor (~> 0.5, >= 0.5.9.6)
+      eventmachine (~> 1.2, >= 1.2.7)
+      faye-websocket (~> 0.12)
     stringio (3.1.7)
     tilt (2.6.1)
     timeout (0.4.3)
     tzinfo (2.0.6)
       concurrent-ruby (~> 1.0)
+    websocket-driver (0.8.0)
+      base64
+      websocket-extensions (>= 0.1.0)
+    websocket-extensions (0.1.5)
 
 PLATFORMS
   aarch64-linux
···
   rake
   sinatra
   sinatra-activerecord (~> 2.0)
+  skyfall (~> 0.6)
 
 BUNDLED WITH
    2.7.0

app/firehose_client.rb

···
+require 'skyfall'
+
+require_relative 'init'
+require_relative 'models/subscription'
+
+class FirehoseClient
+  attr_accessor :start_cursor, :service
+
+  DEFAULT_RELAY = 'bsky.network'
+
+  def initialize
+    @env = (ENV['APP_ENV'] || ENV['RACK_ENV'] || :development).to_sym
+    @service = DEFAULT_RELAY
+  end
+
+  def start
+    return if @sky
+
+    log "Starting firehose process (YJIT = #{RubyVM::YJIT.enabled? ? 'on' : 'off'})"
+
+    last_cursor = load_or_init_cursor
+    cursor = @start_cursor || last_cursor
+
+    @sky = Skyfall::Firehose.new(@service, :subscribe_repos, cursor)
+    @sky.user_agent = "Lycan (https://tangled.sh/@mackuba.eu/lycan) #{@sky.version_string}"
+    @sky.check_heartbeat = true
+
+    @sky.on_message do |m|
+      start_time = Time.now
+      diff = start_time - @last_update
+
+      if diff > 30
+        log "Receiving messages again after #{sprintf('%.1f', diff)}s, starting from #{m.time.getlocal}"
+      end
+
+      @last_update = start_time
+      process_message(m)
+    end
+
+    @sky.on_connecting { |u| log "Connecting to #{u}..." }
+    @sky.on_connect {
+      log "Connected ✓"
+
+      @replaying = true
+      @last_update = Time.now
+
+      @timer ||= EM::PeriodicTimer.new(20) do
+        now = Time.now
+        diff = now - @last_update
+
+        if diff > 30
+          log "Timer: last update #{sprintf('%.1f', diff)}s ago"
+        end
+      end
+    }
+
+    @sky.on_disconnect {
+      log "Disconnected."
+    }
+
+    @sky.on_reconnect {
+      log "Connection lost, reconnecting..."
+
+      @timer&.cancel
+      @timer = nil
+    }
+
+    @sky.on_timeout {
+      log "Trying to reconnect..."
+    }
+
+    @sky.on_error { |e| log "ERROR: #{e.class} #{e.message}" }
+
+    @sky.connect
+  end
+
+  def stop
+    save_cursor(@sky.cursor) unless @sky.nil?
+
+    @sky&.disconnect
+    @sky = nil
+  end
+
+  def load_or_init_cursor
+    if sub = Subscription.find_by(service: @service)
+      sub.cursor
+    else
+      Subscription.create!(service: @service, cursor: 0)
+      nil
+    end
+  end
+
+  def save_cursor(cursor)
+    Subscription.where(service: @service).update_all(cursor: cursor)
+  end
+
+  def process_message(msg)
+    save_cursor(msg.seq) if msg.seq % 1000 == 0
+
+    case msg.type
+    when :info
+      log "InfoMessage: #{msg}"
+    when :account
+      process_account_event(msg)
+    when :commit
+      if @replaying
+        log "Replaying events since #{msg.time.getlocal} -->"
+        @replaying = false
+      end
+
+      msg.operations.each do |op|
+        case op.type
+        when :bsky_post
+          # ...
+        end
+      end
+    end
+  end
+
+  def process_account_event(msg)
+    if msg.status == :deleted
+      # ...
+    end
+  end
+
+  def log(text)
+    puts "[#{Time.now}] #{text}"
+  end
+
+  def inspect
+    vars = instance_variables - [:@timer]
+    values = vars.map { |v| "#{v}=#{instance_variable_get(v).inspect}" }.join(", ")
+    "#<#{self.class}:0x#{object_id} #{values}>"
+  end
+end

app/models/subscription.rb

···
+require 'active_record'
+
+class Subscription < ActiveRecord::Base
+  validates_presence_of :service, :cursor
+end

bin/firehose

···
+#!/usr/bin/env ruby
+
+$LOAD_PATH.unshift(File.expand_path('..', __dir__))
+
+require 'bundler/setup'
+require 'app/firehose_client'
+
+$stdout.sync = true
+
+if ENV['ARLOG'] == '1'
+  ActiveRecord::Base.logger = Logger.new(STDOUT)
+else
+  ActiveRecord::Base.logger = nil
+end
+
+def print_help
+  puts "Usage: #{$0} [options...]"
+  puts "Options:"
+  puts "  -r12345 = start from cursor 12345"
+end
+
+firehose = FirehoseClient.new
+
+args = ARGV.dup
+
+while arg = args.shift
+  case arg
+  when /^\-r(\d+)$/
+    firehose.start_cursor = $1.to_i
+  when '-h', '--help'
+    print_help
+    exit 0
+  else
+    puts "Unrecognized option: #{arg}"
+    print_help
+    exit 1
+  end
+end
+
+trap("SIGINT") {
+  firehose.log "Stopping..."
+
+  EM.add_timer(0) {
+    firehose.stop
+  }
+}
+
+trap("SIGTERM") {
+  firehose.log "Shutting down the service..."
+
+  EM.add_timer(0) {
+    firehose.stop
+  }
+}
+
+firehose.start

db/migrate/20250918024627_add_subscriptions.rb

···
+class AddSubscriptions < ActiveRecord::Migration[7.2]
+  def change
+    create_table :subscriptions do |t|
+      t.string "service", null: false
+      t.bigint "cursor", null: false
+    end
+  end
+end

db/schema.rb

···
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[7.2].define(version: 2025_09_06_233017) do
+ActiveRecord::Schema[7.2].define(version: 2025_09_18_024627) do
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
 
···
     t.string "post_uri"
     t.index ["actor_id", "rkey"], name: "index_reposts_on_actor_id_and_rkey", unique: true
     t.index ["actor_id", "time", "id"], name: "index_reposts_on_actor_id_and_time_and_id", order: { time: :desc, id: :desc }
+  end
+
+  create_table "subscriptions", force: :cascade do |t|
+    t.string "service", null: false
+    t.bigint "cursor", null: false
   end
 
   create_table "users", id: :serial, force: :cascade do |t|