Elasticsearch #

# Install the gem
gem install elasticsearch

# Or in the Gemfile

# Gemfile
gem 'elasticsearch', '~> 8.0'   # make sure the version matches the Elasticsearch server

# For Rails with searchkick
gem 'searchkick', '~> 5.3'

require 'elasticsearch'

# Connect to local Elasticsearch
client = Elasticsearch::Client.new(
  host: "http://localhost:9200",
  log:  false   # set true to debug queries
)

# Connect with authentication (Elasticsearch 8+ with security enabled)
client = Elasticsearch::Client.new(
  host:     "https://localhost:9200",
  user:     "elastic",
  password: ENV["ES_PASSWORD"],
  ca_fingerprint: ENV["ES_CA_FINGERPRINT"]  # certificate fingerprint
)

# Connect to Elastic Cloud
client = Elasticsearch::Client.new(
  cloud_id: ENV["ELASTIC_CLOUD_ID"],
  api_key:  ENV["ELASTIC_API_KEY"]
)

# Connect with multiple nodes (for high availability)
client = Elasticsearch::Client.new(
  hosts: [
    { host: "es-node-1", port: 9200 },
    { host: "es-node-2", port: 9200 },
    { host: "es-node-3", port: 9200 }
  ],
  retry_on_failure: 3,
  reload_connections: true
)

# Check the connection
puts client.info.dig("version", "number")
puts client.ping ? "Connected!" : "Failed!"

# Create an index with a mapping
client.indices.create(
  index: "products",
  body: {
    settings: {
      number_of_shards:   1,
      number_of_replicas: 0,   # 0 for development, 1+ for production
      analysis: {
        analyzer: {
          # Custom analyzer for Indonesian
          indonesian_analyzer: {
            type:      "custom",
            tokenizer: "standard",
            filter:    ["lowercase", "stop", "asciifolding"]
          }
        }
      }
    },
    mappings: {
      properties: {
        name: {
          type:     "text",
          analyzer: "indonesian_analyzer",
          fields: {
            keyword: { type: "keyword" }  # for exact match and sort
          }
        },
        description: {
          type:     "text",
          analyzer: "indonesian_analyzer"
        },
        price:      { type: "double" },
        stock:      { type: "integer" },
        active:     { type: "boolean" },
        category:   { type: "keyword" },   # keyword = exact match, not analyzed
        tags:       { type: "keyword" },   # keyword array
        created_at: { type: "date" },
        location: {
          type: "geo_point"   # for geospatial search
        },
        specs: {
          type: "object",   # nested object
          properties: {
            processor: { type: "keyword" },
            ram:       { type: "keyword" },
            storage:   { type: "keyword" }
          }
        }
      }
    }
  }
)

# Check whether the index exists
puts client.indices.exists?(index: "products")

# Delete the index
client.indices.delete(index: "products")

# Alias — enables zero-downtime reindexing
client.indices.put_alias(index: "products_v2", name: "products")

# INDEX (insert/replace) a single document
client.index(
  index: "products",
  id:    "prod-001",   # optional — ES will generate one if absent
  body: {
    name:      "Pro Gaming Laptop 15",
    description: "A high-powered laptop for gaming and content creators",
    price:     22_000_000,
    stock:     10,
    active:    true,
    category:  "Electronics",
    tags:      ["laptop", "gaming", "high-performance"],
    specs: {
      processor: "Intel Core i9-13900H",
      ram:       "32GB DDR5",
      storage:   "1TB NVMe SSD"
    },
    created_at: Time.now.iso8601
  }
)

# GET a single document by ID
doc = client.get(index: "products", id: "prod-001")
puts doc["_source"]["name"]  # => "Pro Gaming Laptop 15"
puts doc["_id"]              # => "prod-001"

# Check whether a document exists
puts client.exists?(index: "products", id: "prod-001")

# UPDATE — partial update (only the mentioned fields)
client.update(
  index: "products",
  id:    "prod-001",
  body: {
    doc: {
      price:  21_500_000,
      stock:  9,
      updated_at: Time.now.iso8601
    }
  }
)

# UPDATE with a script (for operations like incrementing)
client.update(
  index: "products",
  id:    "prod-001",
  body: {
    script: {
      source: "ctx._source.stock -= params.amount",
      params: { amount: 1 }
    }
  }
)

# DELETE
client.delete(index: "products", id: "prod-001")

# INDEX many documents at once with the Bulk API
# (far more efficient than inserting one by one)
body = []
product_list.each do |p|
  body << { index: { _index: "products", _id: p[:id] } }
  body << {
    name:      p[:name],
    price:     p[:price],
    stock:     p[:stock],
    category:  p[:category],
    active:    p[:active],
    created_at: Time.now.iso8601
  }
end

response = client.bulk(body: body)
puts "Errors: #{response['errors']}"
puts "Items processed: #{response['items'].length}"

# match — analyzes the query and documents, suitable for full-text search
response = client.search(
  index: "products",
  body: {
    query: {
      match: {
        name: {
          query:    "laptop gaming",
          operator: "and"   # all words must be present (default: "or")
        }
      }
    }
  }
)

# Iterate the results
hits = response["hits"]["hits"]
hits.each do |hit|
  puts "#{hit['_score'].round(2)}: #{hit['_source']['name']}"
end

puts "Total: #{response.dig('hits', 'total', 'value')} documents"

# multi_match — search multiple fields at once
response = client.search(
  index: "products",
  body: {
    query: {
      multi_match: {
        query:  "cheap gaming laptop",
        fields: ["name^3", "description", "tags"],  # ^3 = name is 3x more important
        type:   "best_fields"
      }
    }
  }
)

# term — exact match, not analyzed (for keyword fields)
response = client.search(
  index: "products",
  body: {
    query: {
      term: { category: "Electronics" }
    }
  }
)

# terms — match any of an array of values
response = client.search(
  index: "products",
  body: {
    query: {
      terms: { category: ["Electronics", "Accessories", "Monitors"] }
    }
  }
)

# bool query — combine must, should, must_not, filter
response = client.search(
  index: "products",
  body: {
    query: {
      bool: {
        # must — MUST match, affects the relevance score
        must: [
          { match: { name: "laptop" } }
        ],
        # filter — MUST match, does NOT affect the score (faster, cached)
        filter: [
          { term:  { active: true } },
          { range: { price: { gte: 5_000_000, lte: 25_000_000 } } },
          { term:  { "tags" => "gaming" } }
        ],
        # should — MAY match, boosts the score if it matches
        should: [
          { term:  { category: "Electronics" } },
          { match: { description: "high performance" } }
        ],
        minimum_should_match: 1,  # at least 1 should must match
        # must_not — MUST NOT match
        must_not: [
          { term: { stock: 0 } }
        ]
      }
    }
  }
)

# range — value or date range search
client.search(
  index: "products",
  body: {
    query: {
      range: {
        price: { gte: 1_000_000, lte: 10_000_000 }
      }
    }
  }
)

# Date ranges
client.search(
  index: "products",
  body: {
    query: {
      range: {
        created_at: {
          gte: "2024-01-01",
          lte: "2024-12-31",
          format: "yyyy-MM-dd"
        }
      }
    }
  }
)

# wildcard — match a pattern (* = many characters, ? = one character)
client.search(
  index: "products",
  body: { query: { wildcard: { "name.keyword" => "Laptop*Pro*" } } }
)

# prefix — match a beginning
client.search(
  index: "products",
  body: { query: { prefix: { "name.keyword" => "Laptop" } } }
)

# fuzzy — tolerant of typos (edit distance)
client.search(
  index: "products",
  body: {
    query: {
      fuzzy: {
        name: { value: "laptoop", fuzziness: "AUTO" }
      }
    }
  }
)

# Sort
client.search(
  index: "products",
  body: {
    query: { term: { active: true } },
    sort: [
      { price: { order: "asc" } },   # cheapest first
      { "name.keyword": { order: "asc" } }   # then name A-Z
    ]
  }
)

# Pagination with from/size (max from: 10,000)
page = 2
per_page = 20
client.search(
  index: "products",
  body: {
    query: { match_all: {} },
    from:  (page - 1) * per_page,
    size:  per_page
  }
)

# search_after — for deep pagination (> 10,000 documents)
# First: fetch the first page with sort + PIT (Point in Time)
pit = client.open_point_in_time(index: "products", keep_alive: "1m")
pit_id = pit["id"]

response = client.search(
  body: {
    size: 20,
    query: { match_all: {} },
    sort: [{ price: "asc" }, { _id: "asc" }],
    pit: { id: pit_id, keep_alive: "1m" }
  }
)

# Next: use the sort values from the last hit
last_hit    = response["hits"]["hits"].last
sort_values = last_hit["sort"]

response_2 = client.search(
  body: {
    size: 20,
    query: { match_all: {} },
    sort: [{ price: "asc" }, { _id: "asc" }],
    pit: { id: pit_id, keep_alive: "1m" },
    search_after: sort_values
  }
)

# Close the PIT when done
client.close_point_in_time(body: { id: pit_id })

# Source filtering — fetch only certain fields
client.search(
  index: "products",
  body: {
    query: { match_all: {} },
    _source: ["name", "price", "stock"]   # only fetch 3 fields
  }
)

# Or exclude certain fields
client.search(
  index: "products",
  body: {
    query: { match_all: {} },
    _source: { excludes: ["description", "specs"] }
  }
)

response = client.search(
  index: "products",
  body: {
    query: { match: { description: "gaming high performance" } },
    highlight: {
      fields: {
        name:        { number_of_fragments: 0 },  # highlight the whole field
        description: {
          number_of_fragments: 3,
          fragment_size: 150,
          pre_tags:  ["<mark>"],
          post_tags: ["</mark>"]
        }
      }
    }
  }
)

response["hits"]["hits"].each do |hit|
  puts hit["_source"]["name"]
  puts hit.dig("highlight", "description")&.join("... ")
end
# Output: "Laptop for <mark>gaming</mark> with <mark>high performance</mark>"

response = client.search(
  index: "products",
  body: {
    query: { term: { active: true } },
    size: 0,   # 0 = only aggregations, no documents needed

    aggs: {
      # Count documents per category
      per_category: {
        terms: { field: "category", size: 20 }
      },

      # Price statistics
      price_stats: {
        stats: { field: "price" }
        # Produces: count, min, max, avg, sum
      },

      # Price ranges for faceted filters
      price_ranges: {
        range: {
          field: "price",
          ranges: [
            { to:   1_000_000,              key: "Under 1 Million" },
            { from: 1_000_000, to: 5_000_000, key: "1-5 Million" },
            { from: 5_000_000, to: 15_000_000, key: "5-15 Million" },
            { from: 15_000_000,             key: "Above 15 Million" }
          ]
        }
      },

      # Price histogram per 1 million
      price_histogram: {
        histogram: {
          field:    "price",
          interval: 1_000_000,
          min_doc_count: 1
        }
      },

      # Top tags
      top_tags: {
        terms: { field: "tags", size: 10 }
      },

      # Nested aggregation — average price per category
      per_category_with_avg_price: {
        terms: { field: "category", size: 20 },
        aggs: {
          avg_price: { avg: { field: "price" } },
          total_stock: { sum: { field: "stock" } }
        }
      }
    }
  }
)

# Access the aggregation results
puts "Number of products per category:"
response.dig("aggregations", "per_category", "buckets").each do |bucket|
  puts "  #{bucket['key']}: #{bucket['doc_count']} products"
end

stats = response.dig("aggregations", "price_stats")
puts "Price: min=#{stats['min']}, max=#{stats['max']}, avg=#{stats['avg'].round}"

# Completion suggester — fast autocomplete
# First, define the suggest field in the mapping
client.indices.put_mapping(
  index: "products",
  body: {
    properties: {
      name_suggest: {
        type: "completion",
        analyzer: "standard"
      }
    }
  }
)

# Index a document with the suggest field
client.index(
  index: "products",
  id:    "prod-001",
  body: {
    name:         "Pro Gaming Laptop",
    name_suggest: {
      input:  ["Pro Gaming Laptop", "Gaming Laptop", "Laptop"],
      weight: 10   # relevance weight
    }
  }
)

# Autocomplete query
response = client.search(
  index: "products",
  body: {
    suggest: {
      product_autocomplete: {
        prefix: "lap",    # text the user typed
        completion: {
          field: "name_suggest",
          size:  5,        # maximum of 5 suggestions
          skip_duplicates: true
        }
      }
    }
  }
)

suggestions = response.dig("suggest", "product_autocomplete", 0, "options")
suggestions.each { |s| puts s["text"] }
# => "Laptop", "Gaming Laptop", "Pro Gaming Laptop"

# Term suggester — typo correction ("laptoop" → "laptop")
response = client.search(
  index: "products",
  body: {
    suggest: {
      word_correction: {
        text: "laptoop gamng",
        term: {
          field: "name",
          suggest_mode: "missing"  # only suggest if the word is absent from the index
        }
      }
    }
  }
)

# Gemfile
# gem 'searchkick', '~> 5.3'

# app/models/product.rb
class Product < ApplicationRecord
  searchkick(
    word_start:  [:name],              # per-word autocomplete
    text_start:  [:name],              # start-of-text autocomplete
    language:    "english",            # language analyzer
    index_name:  "products_#{Rails.env}",
    settings: {
      number_of_shards:   1,
      number_of_replicas: Rails.env.production? ? 1 : 0
    }
  )

  scope :search_import, -> { includes(:category).where(active: true) }

  # Customize the indexed data
  def search_data
    {
      name:        name,
      description: description,
      price:       price,
      stock:       stock,
      active:      active,
      category:    category&.name,
      tags:        tags,
      created_at:  created_at
    }
  end
end

# Indexing
Product.reindex          # reindex all products
Product.reindex(async: true)  # asynchronously (background job)

# Basic search
results = Product.search("laptop gaming")
results.each { |p| puts p.name }
puts "Total: #{results.total_count}"

# Search with filters, sort, and pagination
results = Product.search(
  "laptop",
  where: {
    active:   true,
    price:    { gte: 5_000_000, lte: 25_000_000 },
    category: ["Electronics", "Accessories"]
  },
  order: { price: :asc },
  page:    params[:page] || 1,
  per_page: 20,
  includes: [:category]   # eager load relations
)

# Aggregations (facets)
results = Product.search(
  "laptop",
  aggs: [:category, :tags],
  where: { active: true }
)

# Access facets
puts results.aggs.dig("category", "buckets")

# Highlight
results = Product.search(
  "laptop gaming",
  highlight: { fields: [:name, :description] }
)
results.with_highlights.each do |product, highlight|
  puts highlight[:name] || product.name
end

# Autocomplete / Suggest
Product.search("lap", fields: ["name^5", "description"], match: :word_start, limit: 5)

# Automatic sync when data changes
product = Product.find(1)
product.update!(price: 20_000_000)   # automatically updates the ES index
product.destroy                      # automatically removed from ES

# Or disable auto-sync and do it manually
Product.searchkick_callbacks(false) do
  Product.update_all(price: 15_000_000)  # doesn't trigger ES updates
end
Product.reindex   # manual sync after a bulk update

# Strategy 1: Automatic sync (searchkick default)
# Every save/update/destroy → updates ES synchronously
# Suitable for small applications, but can slow down HTTP requests

# Strategy 2: Async with a background job
# config/initializers/searchkick.rb
Searchkick.callbacks = :async   # use ActiveJob

# Strategy 3: Scheduled batch reindex
# config/schedule.rb (with the whenever gem)
every 1.hour do
  runner "Product.reindex"
end

# Strategy 4: Delta sync — only sync what changed
# app/models/product.rb
after_commit :reindex, if: :saved_change_to_relevant_field?

def saved_change_to_relevant_field?
  saved_changes.keys.any? { |k| %w[name description price stock active].include?(k) }
end

# Searchkick handles this automatically with aliases
Product.reindex

# Behind the scenes:
# 1. Create a new index: products_20240815120000
# 2. Index all data into the new index
# 3. Swap the "products" alias to the new index
# 4. Delete the old index

# Manual with elasticsearch-ruby:
client = client

# 1. Create the new index
client.indices.create(index: "products_v2", body: new_mapping)

# 2. Reindex from the old index to the new one
client.reindex(
  body: {
    source: { index: "products_v1" },
    dest:   { index: "products_v2" }
  },
  wait_for_completion: false  # async for large datasets
)

# 3. Monitor progress
# client.tasks.get(task_id: task_id)

# 4. Swap the alias
client.indices.update_aliases(
  body: {
    actions: [
      { remove: { index: "products_v1", alias: "products" } },
      { add:    { index: "products_v2", alias: "products" } }
    ]
  }
)

# 5. Delete the old index
client.indices.delete(index: "products_v1")