Fixed deletes with routing and callbacks - closes #1182

Andrew
1 parent 9155938d
Showing 4 changed files with 30 additions and 3 deletions Show diff stats
CHANGELOG.md
lib/searchkick/process_batch_job.rb
lib/searchkick/record_indexer.rb
test/routing_test.rb
 ## 3.1.2 [unreleased]
  
 - Fixed deletes with routing and `async` callbacks
+- Fixed deletes with routing and `queue` callbacks
  
 ## 3.1.1
  
@@ -3,6 +3,10 @@ module Searchkick
     queue_as { Searchkick.queue_name }
  
     def perform(class_name:, record_ids:)
+      # separate routing from id
+      routing = Hash[record_ids.map { |r| r.split(/(?<!\|)\|(?!\|)/, 2).map { |v| v.gsub("||", "|") } }]
+      record_ids = routing.keys
+
       klass = class_name.constantize
       scope = Searchkick.load_records(klass, record_ids)
       scope = scope.search_import if scope.respond_to?(:search_import)
@@ -10,7 +14,16 @@ module Searchkick
  
       # determine which records to delete
       delete_ids = record_ids - records.map { |r| r.id.to_s }
-      delete_records = delete_ids.map { |id| m = klass.new; m.id = id; m }
+      delete_records = delete_ids.map do |id|
+        m = klass.new
+        m.id = id
+        if routing[id]
+          m.define_singleton_method(:search_routing) do
+            routing[id]
+          end
+        end
+        m
+      end
  
       # bulk reindex
       index = klass.searchkick_index
@@ -20,7 +20,16 @@ module Searchkick
           raise Searchkick::Error, "Partial reindex not supported with queue option"
         end
  
-        index.reindex_queue.push(record.id.to_s)
+        # always pass routing in case record is deleted
+        # before the queue job runs
+        if record.respond_to?(:search_routing)
+          routing = record.search_routing
+        end
+
+        # escape pipe with double pipe
+        value = queue_escape(record.id.to_s)
+        value = "#{value}|#{queue_escape(routing)}" if routing
+        index.reindex_queue.push(value)
       when :async
         unless defined?(ActiveJob)
           raise Searchkick::Error, "Active Job not found"
@@ -47,6 +56,10 @@ module Searchkick
  
     private
  
+    def queue_escape(value)
+      value.gsub("|", "||")
+    end
+
     def reindex_record(method_name)
       if record.destroyed? || !record.persisted? || !record.should_index?
         begin
@@ -31,7 +31,7 @@ class RoutingTest &lt; Minitest::Test
   end
  
   def test_routing_queue
-    skip # unless defined?(ActiveJob) && defined?(Redis)
+    skip unless defined?(ActiveJob) && defined?(Redis)
  
     with_options(Song, routing: true, callbacks: :queue) do
       store_names ["Dollar Tree"], Song
1	1	## 3.1.2 [unreleased]
2	2
3	3	- Fixed deletes with routing and `async` callbacks
	4	+- Fixed deletes with routing and `queue` callbacks
4	5
5	6	## 3.1.1
6	7
...	...
...	...	@@ -3,6 +3,10 @@ module Searchkick
3	3	queue_as { Searchkick.queue_name }
4	4
5	5	def perform(class_name:, record_ids:)
	6	+ # separate routing from id
	7	+ routing = Hash[record_ids.map { \|r\| r.split(/(?<!\\|)\\|(?!\\|)/, 2).map { \|v\| v.gsub("\|\|", "\|") } }]
	8	+ record_ids = routing.keys
	9	+
6	10	klass = class_name.constantize
7	11	scope = Searchkick.load_records(klass, record_ids)
8	12	scope = scope.search_import if scope.respond_to?(:search_import)
...	...	@@ -10,7 +14,16 @@ module Searchkick
10	14
11	15	# determine which records to delete
12	16	delete_ids = record_ids - records.map { \|r\| r.id.to_s }
13		- delete_records = delete_ids.map { \|id\| m = klass.new; m.id = id; m }
	17	+ delete_records = delete_ids.map do \|id\|
	18	+ m = klass.new
	19	+ m.id = id
	20	+ if routing[id]
	21	+ m.define_singleton_method(:search_routing) do
	22	+ routing[id]
	23	+ end
	24	+ end
	25	+ m
	26	+ end
14	27
15	28	# bulk reindex
16	29	index = klass.searchkick_index
...	...
...	...	@@ -20,7 +20,16 @@ module Searchkick
20	20	raise Searchkick::Error, "Partial reindex not supported with queue option"
21	21	end
22	22
23		- index.reindex_queue.push(record.id.to_s)
	23	+ # always pass routing in case record is deleted
	24	+ # before the queue job runs
	25	+ if record.respond_to?(:search_routing)
	26	+ routing = record.search_routing
	27	+ end
	28	+
	29	+ # escape pipe with double pipe
	30	+ value = queue_escape(record.id.to_s)
	31	+ value = "#{value}\|#{queue_escape(routing)}" if routing
	32	+ index.reindex_queue.push(value)
24	33	when :async
25	34	unless defined?(ActiveJob)
26	35	raise Searchkick::Error, "Active Job not found"
...	...	@@ -47,6 +56,10 @@ module Searchkick
47	56
48	57	private
49	58
	59	+ def queue_escape(value)
	60	+ value.gsub("\|", "\|\|")
	61	+ end
	62	+
50	63	def reindex_record(method_name)
51	64	if record.destroyed? \|\| !record.persisted? \|\| !record.should_index?
52	65	begin
...	...
...	...	@@ -31,7 +31,7 @@ class RoutingTest < Minitest::Test
31	31	end
32	32
33	33	def test_routing_queue
34		- skip # unless defined?(ActiveJob) && defined?(Redis)
	34	+ skip unless defined?(ActiveJob) && defined?(Redis)
35	35
36	36	with_options(Song, routing: true, callbacks: :queue) do
37	37	store_names ["Dollar Tree"], Song
...	...