Class: TextEmbeddingPopulationWorker

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

Overview

Nightly safety net and controlled backfill worker for text embeddings.

Uses Sidekiq::IterableJob so progress is saved after each record — a mid-run
deploy or worker restart resumes from the last successful record rather than
restarting from scratch.

Each iteration enqueues an EmbeddingWorker job rather than generating the
embedding inline, keeping this worker fast and letting the ai_embeddings
queue handle rate limiting and retries.

Supports all text-embedding types: Activity, Communication, Post, Article,
Showcase, Video, Item, ProductLine, SiteMap, ReviewsIo, CallRecord, and
AssistantBrainEntry.
(Images use a separate pipeline via ImageEmbeddingPopulationWorker.)

Scheduled: Nightly via config/sidekiq_production_schedule.yml

Examples:

Backfill activity embeddings (default: 1000/run)

TextEmbeddingPopulationWorker.perform_async('embeddable_type' => 'Activity')

Backfill with custom limit

TextEmbeddingPopulationWorker.perform_async('embeddable_type' => 'Activity', 'limit' => 5000)

Scan recently changed technical-support communications

TextEmbeddingPopulationWorker.perform_async(
  'source_key' => 'technical_support_communications',
  'lookback_hours' => 168,
  'limit' => 500
)

Constant Summary collapse

ALLOWED_TYPES =

Recognised allowed types.

%w[
  Activity Communication Post Article Showcase Video Item ProductLine
  SiteMap ReviewsIo CallRecord AssistantBrainEntry
].freeze
DEFAULT_LIMIT =

Default limit.

1000
TECHNICAL_SUPPORT_CASE_IDS =

One source of truth for every Technical Support evidence scope.

-> { SupportCase.where(case_type: 'Tech').select(:id) }.freeze
TECHNICAL_SUPPORT_REPLAY_CASE_IDS =

S.

lambda {
  SupportCase
    .where(
      case_type: TechnicalSupportReplayCriteria::CASE_TYPE,
      state: TechnicalSupportReplayCriteria::CLOSED_STATE,
      closed_reason: TechnicalSupportReplayCriteria::CLOSED_REASONS
    )
    .where(
      SupportCase[:closed_at].gteq(
        TechnicalSupportReplayCriteria::LOOKBACK.ago
      )
    )
    .select(:id)
}.freeze
REGISTERED_SOURCES =

Named record scopes that cannot be represented by a model type alone.
These stay in the existing population worker so adding a future business
funnel does not create a second embedding pipeline.

{
  'technical_support_activities' => {
    embeddable_type: 'Activity',
    scope: lambda {
      Activity.where(
        resource_type: 'SupportCase',
        resource_id: TECHNICAL_SUPPORT_CASE_IDS.call
      )
    }
  }.freeze,
  'technical_support_communications' => {
    embeddable_type: 'Communication',
    scope: lambda {
      Communication.where(
        resource_type: 'SupportCase',
        resource_id: TECHNICAL_SUPPORT_CASE_IDS.call
      )
    }
  }.freeze,
  'technical_support_replay_activities' => {
    embeddable_type: 'Activity',
    scope: lambda {
      Activity.where(
        resource_type: 'SupportCase',
        resource_id: TECHNICAL_SUPPORT_REPLAY_CASE_IDS.call
      )
    }
  }.freeze,
  'technical_support_replay_communications' => {
    embeddable_type: 'Communication',
    scope: lambda {
      Communication.where(
        resource_type: 'SupportCase',
        resource_id: TECHNICAL_SUPPORT_REPLAY_CASE_IDS.call
      )
    }
  }.freeze,
  'technical_support_replay_call_records' => {
    embeddable_type: 'CallRecord',
    scope: lambda {
      CallRecord.where(
        id: Activity.where(
          resource_type: 'SupportCase',
          resource_id: TECHNICAL_SUPPORT_REPLAY_CASE_IDS.call
        )
                    .where.not(call_record_id: nil)
                    .select(:call_record_id)
      )
    }
  }.freeze
}.freeze
RECENCY_COLUMNS =

Source timestamp used by the recent-change safety net. Call records do not
have +updated_at+; their transcript freshness is represented by
+transcribed_at+.

{ 'CallRecord' => :transcribed_at }.freeze

Instance Method Summary collapse

Instance Method Details

#build_enumerator(options = nil, cursor:) ⇒ Enumerator?

Builds a resumable enumerator over records that need a freshness check.

With +lookback_hours+, the worker scans recently changed eligible records
and lets Models::Embeddable#embedding_stale? decide whether to enqueue.
Without it, the worker selects records lacking a complete current-model
vector, which is the controlled historical-backfill mode.

Parameters:

  • options (Hash, nil) (defaults to: nil)
  • cursor (Object, nil) —

    Sidekiq IterableJob resume cursor

Options Hash (options):

  • :embeddable_type (String) —

    model type for a generic run

  • :source_key (String) —

    registered record scope

  • :lookback_hours (Integer) —

    recent-change window

  • :limit (Integer) —

    maximum records inspected

Returns:

  • (Enumerator, nil) 


139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'app/workers/text_embedding_population_worker.rb', line 139

def build_enumerator(options = nil, cursor:)
  options ||= {}
  @source_key = options[:source_key].presence
  @embeddable_type = resolve_embeddable_type(options)
  @lookback_hours = options[:lookback_hours].presence&.to_i
  @limit = (options[:limit] || DEFAULT_LIMIT).to_i
  @queued_count = 0

  return unless valid_configuration?

  scope = build_scope
  mode = @lookback_hours ? "recent #{@lookback_hours}h safety net" : 'missing-current backfill'
  source = @source_key || @embeddable_type
  log_info "Starting #{mode} for #{source}: #{scope.count} candidates (limit: #{@limit})"

  active_record_records_enumerator(scope.limit(@limit), cursor: cursor)
end

#each_iteration(record, *_args) ⇒ void

This method returns an undefined value.

Queues one stale record on the existing embedding worker.

Parameters:

  • record (ApplicationRecord) —

    embeddable source record

  • _args (Array<Object>) —

    unused IterableJob arguments



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'app/workers/text_embedding_population_worker.rb', line 162

def each_iteration(record, *_args)
  begin
    return unless record.embedding_stale?

    content = record.content_for_embedding
    skip_blank = content.blank? && !record.has_embedding?
  rescue StandardError => e
    # A record whose staleness/content check raises must not kill the run:
    # the IterableJob cursor would resume on that same record and fail again.
    # API failures keep their per-record retries inside EmbeddingWorker.
    Rails.event.notify(
      'warning.embedding.population_record_skipped',
      embeddable_type: @embeddable_type,
      embeddable_id: record.id,
      source_key: @source_key,
      exception_class: e.class.name,
      exception_message: e.message
    )
    return
  end

  if skip_blank
    log_debug "Skipping #{@embeddable_type}##{record.id}: no embeddable content"
    return
  end

  EmbeddingWorker.perform_async(@embeddable_type, record.id)
  @queued_count += 1

  log_info "Progress: #{@queued_count} queued" if (@queued_count % 200).zero?
end

#on_complete ⇒ void

This method returns an undefined value.

Logs the number of stale records queued by this run.



197
198
199
 
# File 'app/workers/text_embedding_population_worker.rb', line 197

def on_complete
  log_info "Complete: queued #{@queued_count} #{@embeddable_type || 'mixed'} records for embedding"
end