scope based upsert or delete

lib/thinking_sphinx/processor.rb

@@ -1,6 +1,9 @@
 # frozen_string_literal: true
 
 class ThinkingSphinx::Processor
+  # @param instance [ActiveRecord::Base] an ActiveRecord object
+  # @param model [Class] the ActiveRecord model of the instance
+  # @param id [Integer] the instance indices primary key (might be different from model primary key)
   def initialize(instance: nil, model: nil, id: nil)
     raise ArgumentError if instance.nil? && (model.nil? || id.nil?)
 
@@ -12,16 +15,27 @@ def initialize(instance: nil, model: nil, id: nil)
   def delete
     return if instance&.new_record?
 
-    indices.each { |index|
-      ThinkingSphinx::Deletion.perform(
-        index, id || instance.public_send(index.primary_key)
-      )
-    }
+    indices.each { |index| perform_deletion(index) }
   end
 
+  # Will insert instance into all matching indices
   def upsert
     real_time_indices.each do |index|
-      ThinkingSphinx::RealTime::Transcriber.new(index).copy loaded_instance
+      found = loaded_instance(index)
+      ThinkingSphinx::RealTime::Transcriber.new(index).copy found if found
+    end
+  end
+
+  # Will upsert or delete instance into all matching indices based on index scope
+  def stage
+    real_time_indices.each do |index|
+      found = find_in(index)
+
+      if found
+        ThinkingSphinx::RealTime::Transcriber.new(index).copy found
+      else
+        ThinkingSphinx::Deletion.perform(index, index_id(index))
+      end
     end
   end
 
@@ -35,11 +49,23 @@ def indices
     ).to_a
   end
 
-  def loaded_instance
-    @loaded_instance ||= instance || model.find(id)
+  def find_in(index)
+    index.scope.find_by(index.primary_key => index_id(index))
+  end
+
+  def loaded_instance(index)
+    instance || find_in(index)
   end
 
   def real_time_indices
     indices.select { |index| index.is_a? ThinkingSphinx::RealTime::Index }
   end
+
+  def perform_deletion(index)
+    ThinkingSphinx::Deletion.perform(index, index_id(index))
+  end
+
+  def index_id(index)
+    id || instance.public_send(index.primary_key)
+  end
 end

spec/acceptance/real_time_updates_spec.rb

@@ -28,7 +28,7 @@
     expect(Admin::Person.search('Mort').to_a).to eq([person])
   end
 
-  it "can use a direct interface for processing records" do
+  it "can use direct interface for upserting records" do
     Admin::Person.connection.execute <<~SQL
       INSERT INTO admin_people (name, created_at, updated_at)
       VALUES ('Pat', now(), now());
@@ -52,4 +52,64 @@
 
     expect(Admin::Person.search('Patrick').to_a).to eq([instance])
   end
+
+  it "can use direct interface for processing records outside scope" do
+    Article.connection.execute <<~SQL
+      INSERT INTO articles (title, published, created_at, updated_at)
+      VALUES ('Nice Title', TRUE, now(), now());
+    SQL
+
+    article  = Article.last
+
+    ThinkingSphinx::Processor.new(model: article.class, id: article.id).stage
+
+    expect(ThinkingSphinx.search('Nice', :indices => ["published_articles_core"])).to include(article)
+
+    Article.connection.execute <<~SQL
+      UPDATE articles SET published = FALSE WHERE title = 'Nice Title';
+    SQL
+    ThinkingSphinx::Processor.new(model: article.class, id: article.id).stage
+
+    expect(ThinkingSphinx.search('Nice', :indices => ["published_articles_core"])).to be_empty
+  end
+
+  it "can use direct interface for processing deleted records" do
+    Article.connection.execute <<~SQL
+      INSERT INTO articles (title, published, created_at, updated_at)
+      VALUES ('Nice Title', TRUE, now(), now());
+    SQL
+
+    article  = Article.last
+    ThinkingSphinx::Processor.new(:instance => article).stage
+
+    expect(ThinkingSphinx.search('Nice', :indices => ["published_articles_core"])).to include(article)
+
+    Article.connection.execute <<~SQL
+      DELETE FROM articles where title = 'Nice Title';
+    SQL
+
+    ThinkingSphinx::Processor.new(:instance => article).stage
+
+    expect(ThinkingSphinx.search('Nice', :indices => ["published_articles_core"])).to be_empty
+  end
+
+  it "stages records in real-time index with alternate ids" do
+    Album.connection.execute <<~SQL
+      INSERT INTO albums (id, name, artist, integer_id)
+      VALUES ('#{("a".."z").to_a.sample}', 'Sing to the Moon', 'Laura Mvula', #{rand(10000)});
+    SQL
+
+    album  = Album.last
+    ThinkingSphinx::Processor.new(:model => Album, id: album.integer_id).stage
+
+    expect(ThinkingSphinx.search('Laura', :indices => ["album_real_core"])).to include(album)
+
+    Article.connection.execute <<~SQL
+      DELETE FROM albums where id = '#{album.id}';
+    SQL
+
+    ThinkingSphinx::Processor.new(:instance => album).stage
+
+    expect(ThinkingSphinx.search('Laura', :indices => ["album_real_core"])).to be_empty
+  end
 end

spec/internal/app/indices/article_index.rb

@@ -23,3 +23,9 @@
 
   set_property :morphology => 'stem_en'
 end
+
+ThinkingSphinx::Index.define :article, :name => :published_articles, :with => :real_time do
+  indexes title, content
+
+  scope { Article.where :published => true }
+end