Class: EmbeddingRefreshWorker

Inherits:
Object
  • Object
show all
Includes:
Sidekiq::Job
Defined in:
app/workers/embedding_refresh_worker.rb

Overview

Scheduled worker to detect and regenerate stale embeddings.
Runs weekly across all corpora and supports bounded, type-scoped passes for
higher-frequency repair of small critical corpora such as Articles.

For Images, checks for missing unified embeddings and Vision analysis.
Uses the Gemini pipeline: pHash → Gemini Embedding 2 → Gemini Flash Vision

Scheduled: Sundays at 3:30am (after sitemap generation at 1:10am)
Queue: low priority to avoid impacting user-facing operations

Examples:

Manual invocation

EmbeddingRefreshWorker.new.perform

Constant Summary collapse

EMBEDDABLE_TYPES =
Note:

CallRecord transcription is handled by DailyCallRecordTranscriptionWorker,
but we still check here for:

  • LeMUR re-runs that update content without regenerating embeddings
  • Failed embedding jobs (transcribed but no embedding)
  • Manual edits to call record metadata

Content types to check for stale embeddings
Order matters: check smaller collections first to spread load

%w[
  Post
  Article
  Showcase
  Video
  Image
  SiteMap
  ReviewsIo
  Item
  CallRecord
  AssistantBrainEntry
].freeze
MAX_RECORDS_PER_TYPE =

Give every corpus a bounded share of the weekly repair pass.

50
MAX_RECORDS_PER_RUN =

Hard ceiling across text repair and full image analysis.

500
MAX_IMAGES_FULL_ANALYSIS =

Maximum images to queue for full analysis per run

100
TYPE_CHECK_DELAY =

Delay between checking each type (seconds)

2

Instance Method Summary collapse

Instance Method Details

#perform(options = {}) ⇒ Hash{String => Integer}

Checks eligible records for stale embeddings and queues regeneration.

Parameters:

  • options (Hash) (defaults to: {}) —

    serialized worker options

Options Hash (options):

  • types (Array<String>) —

    subset of embeddable types to inspect (defaults to all)

  • max_records (Integer) —

    cap on total records queued this run

  • max_records_per_type (Integer) —

    cap on records queued per type

Returns:

  • (Hash{String => Integer}) —

    queued counts by embeddable type



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'app/workers/embedding_refresh_worker.rb', line 61

def perform(options = {})
  types = requested_types(options)
  max_records = bounded_limit(options[:max_records], default: MAX_RECORDS_PER_RUN)
  per_type_limit = bounded_limit(options[:max_records_per_type], default: MAX_RECORDS_PER_TYPE)
  log_info "Starting embedding refresh check for #{types.join(', ')}..."

  total_queued = 0
  stats = {}

  types.each do |type|
    break if total_queued >= max_records

    # Small delay between types to spread database load
    sleep(TYPE_CHECK_DELAY) unless total_queued.zero?

    remaining = max_records - total_queued
    count = check_and_queue_stale(type, limit: [per_type_limit, remaining].min)
    stats[type] = count
    total_queued += count

    log_info "#{type}: #{count} stale embeddings queued" if count.positive?
  end

  # Queue images missing Vision analysis or unified embedding for full pipeline
  remaining = max_records - total_queued
  image_limit = [MAX_IMAGES_FULL_ANALYSIS, per_type_limit, remaining].min
  if types.include?('Image')
    full_analysis_count = image_limit.positive? ? queue_images_for_full_analysis(limit: image_limit) : 0
    stats['Image_FullAnalysis'] = full_analysis_count
    total_queued += full_analysis_count
  end

  # Also extract fresh content for static pages before embedding
  extract_static_pages_if_needed if types.include?('SiteMap')

  log_info "Embedding refresh complete. Total queued: #{total_queued}"
  log_info "Stats: #{stats.select { |_, v| v.positive? }}" if total_queued.positive?

  stats
rescue StandardError => e
  log_error "Embedding refresh failed: #{e.message}"
  ErrorReporting.error(e)
  raise
end