Class: GoodJob::ElasticsearchProcessBatch

Inherits:
HasUtils::GoodJob::BaseWorker
  • Object
show all
Defined in:
app/workers/good_job/elasticsearch_process_batch.rb

Instance Method Summary collapse

Instance Method Details

#perform(index_class_name, index_name, record_ids, routing_map = {}) ⇒ Object 



7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'app/workers/good_job/elasticsearch_process_batch.rb', line 7

def perform(index_class_name, index_name, record_ids, routing_map = {})
  index_klass = index_class_name.constantize
  is_keyword_index = index_name.include?("keyword")

  records = ActiveRecord::Base.connection.select_all <<~SQL
    SELECT * FROM #{index_klass.view_name} WHERE id IN (#{record_ids.map { |id| "'#{id}'" }.join(",")})
  SQL

  # Converts PG arrays to Ruby arrays
  records.each do |record|
    record.each do |k, v|
      if records.column_types[k].is_a?(ActiveRecord::ConnectionAdapters::PostgreSQL::OID::Array)
        record[k] = records.column_types[k].deserialize(v)
      end
    end
  end.map(&:symbolize_keys!)

  record_type = index_klass.type

  indexes = records.map do |record|
    attributes = ::HasHelpers::Index.default_attributes.merge(record.except(index_klass.active_record.primary_key))

    if is_keyword_index
      data = attributes
      data.each do |field, field_value|
        data[field] = field_value&.to_s if field_value.is_a?(::Integer) || field_value.is_a?(::Float) # Storing text instead of numbers allows for easier wildcard searching
      end
    else
      # Exclude fields that are large, display-only, and non-searchable to prevent bloat in full-text aggregation.
      # `:result_type` is marked `enabled: false` in the mapping and thus excluded from indexing.
      excluded_fields = [:result_type]
      all_fields = index_klass.all_values_from_attributes(attributes.except(*excluded_fields))
      data = attributes.merge(all_fields: all_fields)
    end
    routing = index_klass.is_global? ? 0 : record[:organization_id]

    {
      index: {
        routing: routing,
        _index: index_name,
        _id: record[index_klass.active_record.primary_key.to_sym],
        _type: record_type,
        data: data
      }
    }
  end

  found_ids = records.map { |r| r[index_klass.active_record.primary_key.to_sym].to_s }
  delete_ids = record_ids.map(&:to_s) - found_ids
  deletes = delete_ids.filter_map do |record_id|
    routing = index_klass.is_global? ? 0 : routing_map[record_id]
    next if routing.blank?

    {
      delete: {
        routing: routing,
        _index: index_name,
        _id: record_id,
        _type: record_type
      }
    }
  end

  if (index_klass.index_name != index_name && index_klass.keyword_search_index_name != index_name) ||
     ::Searchkick.client.indices.exists_alias?(name: index_name)
    ::Searchkick.indexer.queue([*indexes, *deletes])
  else
    retry_job(wait: 1.minute)
  end
end