save cursor and use it on reconnect to replay events

app/firehose_stream.rb

···
 require_relative 'config'
 require_relative 'models/feed_post'
 require_relative 'models/post'
+require_relative 'models/subscription'
 
 class FirehoseStream
-  attr_accessor :show_progress, :log_status, :log_posts, :save_posts
+  attr_accessor :show_progress, :log_status, :log_posts, :save_posts, :replay_events
 
   def initialize
     @env = (ENV['APP_ENV'] || ENV['RACK_ENV'] || :development).to_sym
+    @service = 'bsky.social'
 
     @show_progress = (@env == :development) ? true : false
     @log_status = true
     @log_posts = (@env == :development) ? :matching : false
     @save_posts = (@env == :development) ? :all : :matching
+    @replay_events = (@env == :development) ? false : true
 
     @feeds = BlueFactory.all_feeds
   end
···
   def start
     return if @sky
 
-    @sky = Skyfall::Stream.new('bsky.social', :subscribe_repos)
+    last_cursor = load_or_init_cursor
+    cursor = @replay_events ? last_cursor : nil
+
+    @sky = Skyfall::Stream.new(@service, :subscribe_repos, cursor)
 
     @sky.on_message do |m|
       handle_message(m)
     end
 
     if @log_status
-      @sky.on_connect { puts "Connected #{Time.now} ✓" }
+      @sky.on_connect { @replaying = !!(cursor); puts "Connected #{Time.now} ✓" }
       @sky.on_disconnect { puts; puts "Disconnected #{Time.now}" }
       @sky.on_reconnect { puts "Reconnecting..." }
       @sky.on_error { |e| puts "ERROR: #{Time.now} #{e}" }
···
     @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 handle_message(msg)
+    if msg.seq % 10 == 0
+      save_cursor(msg.seq)
+    end
+
+    if @replaying
+      puts "Replaying events since #{msg.time.getlocal} -->"
+      @replaying = false
+    end
+
     return if msg.type != :commit
 
     msg.operations.each do |op|

app/models/subscription.rb

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

db/migrate/20230727134424_add_subscriptions.rb

···
+class AddSubscriptions < ActiveRecord::Migration[6.1]
+  def change
+    create_table :subscriptions do |t|
+      t.string :service, null: false
+      t.integer :cursor, null: false
+    end
+
+    add_index :subscriptions, :service, unique: true
+  end
+end

db/schema.rb

···
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 2023_06_15_155215) do
+ActiveRecord::Schema.define(version: 2023_07_27_134424) do
 
   create_table "feed_posts", force: :cascade do |t|
     t.integer "feed_id", null: false
···
     t.text "data", null: false
     t.string "rkey", null: false
     t.index ["rkey"], name: "index_posts_on_rkey"
+  end
+
+  create_table "subscriptions", force: :cascade do |t|
+    t.string "service", null: false
+    t.integer "cursor", null: false
+    t.index ["service"], name: "index_subscriptions_on_service", unique: true
   end
 
 end

firehose.rb

···
   puts "  -da = save all posts to database"
   puts "  -dm = save only matching posts to database"
   puts "  -nd = don't save any posts"
+  puts
+  puts "  * Replaying missed events: [default: -nr in development, -r in production]"
+  puts "  -r = pass a cursor param when connecting to replay any missed events"
+  puts "  -nr = don't replay missed events"
 end
 
 firehose = FirehoseStream.new
···
     firehose.save_posts = :all
   when '-nd'
     firehose.save_posts = false
+  when '-r'
+    firehose.replay_events = true
+  when '-nr'
+    firehose.replay_events = false
   when '-h', '--help'
     print_help
     exit 0