Class: CallRecordBulkTranscriptionWorker

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

Overview

Nightly worker to backfill historical call record transcriptions using the
cheap Gemini 3.1 Flash-Lite native-audio model via RubyLLM (model resolved
from AiModelConstants.id(:transcription)).

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

Each iteration downloads the audio, transcribes synchronously, saves the
transcript, and queues an EmbeddingWorker job. This is intentionally serial
(not fan-out) to control API throughput and costs.

Processes newest-first (order: created_at DESC) within the eligible window —
any pending/error call older than min_age_days. The premium AssemblyAI
pipeline (at-import enqueue + DailyCallRecordTranscriptionWorker 24h catch-up

  • StaleTranscriptionRecoveryWorker) gets first crack on recent calls; this
    worker sweeps whatever it didn't complete once the call ages past min_age_days,
    which is what closes the former 24h–365d coverage gap.

Scheduled: Nightly via config/sidekiq_production_schedule.yml

Examples:

Backfill eligible calls older than 7 days (default: 200/run)

CallRecordBulkTranscriptionWorker.perform_async

Custom limit / model / age floor

CallRecordBulkTranscriptionWorker.perform_async(
  'limit' => 2000,
  'model' => 'gpt-4o-mini-transcribe',
  'min_age_days' => 30
)

Only calls involving specific parties (outlet-purchase accounts)

CallRecordBulkTranscriptionWorker.perform_async('party_ids' => [123, 456], 'limit' => 800)

Constant Summary collapse

DEFAULT_LIMIT =

Conservative default when no limit is passed. The scheduled entry overrides
this; raising the scheduled limit to drain the backlog faster is a
bulk-op decision (count-first + confirm) — see CLAUDE.md bulk-op protocol.

200
DEFAULT_MIN_AGE_DAYS =

Cheap pipeline handles any pending/error call older than this. The premium
AssemblyAI pipeline (at-import enqueue + DailyCallRecordTranscriptionWorker
24h catch-up + StaleTranscriptionRecoveryWorker + retries) gets first crack
on recent calls; a 7-day floor leaves that premium window comfortably clear,
and anything still un-transcribed after 7 days falls through to this cheap
backfill instead of sitting pending forever (the prior 365-day floor left a
24h–365d coverage gap).

7
MIN_DURATION_SECONDS =

Minimum duration seconds.

CallRecord::MIN_TRANSCRIPTION_DURATION
MIN_DURATION_SECONDS_VOICEMAIL =

Minimum duration seconds voicemail.

CallRecord::MIN_TRANSCRIPTION_DURATION_VOICEMAIL

Instance Method Summary collapse

Instance Method Details

#build_enumerator(options = nil, cursor:) ⇒ Object

Parameters:

  • options (Hash) (defaults to: nil) —

    job options (string or symbol keys)

  • cursor (Object) —

    iteration cursor managed by sidekiq-iteration

Options Hash (options):

  • limit (Integer) —

    maximum calls to transcribe this run (default DEFAULT_LIMIT)

  • model (String) —

    transcription model (default BulkTranscriptionService::DEFAULT_MODEL)

  • min_age_days (Integer) —

    only transcribe calls older than this many days (default DEFAULT_MIN_AGE_DAYS)

  • party_ids (Array<Integer>) —

    restrict the sweep to calls involving these parties



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'app/workers/call_record_bulk_transcription_worker.rb', line 67

def build_enumerator(options = nil, cursor:)
  # sidekiq-iteration omits the arg entirely when no options were enqueued.
  # Indifferent access here rather than trusting NormalizeArgsMiddleware to
  # have provided it: that is SERVER middleware, so a direct call (console,
  # tests under Sidekiq fake mode) hands this method plain string keys and
  # every `options[:sym]` read would silently answer nil — for `party_ids`,
  # that degrades a scoped backfill into the global sweep.
  options = (options || {}).with_indifferent_access
  @limit = (options[:limit] || DEFAULT_LIMIT).to_i
  @model = options[:model] || CallRecordProcessing::BulkTranscriptionService::DEFAULT_MODEL
  @min_age_days = (options[:min_age_days] || DEFAULT_MIN_AGE_DAYS).to_i
  @party_ids = Array(options[:party_ids]).compact
  @success_count = 0
  @skip_count = 0
  @error_count = 0

  scope = build_scope
  candidate_count = scope.count
  log_info "Starting bulk transcription: #{candidate_count} candidates, limit #{@limit}, model #{@model}, min_age #{@min_age_days}d"

  return nil if candidate_count.zero?

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

#each_iteration(call_record, *_args) ⇒ Object

Each iteration.

Parameters:

  • call_record (CallRecord) —

    the call record

  • _args (Array) —

    the args

Returns:

  • (Object) —

    the result



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'app/workers/call_record_bulk_transcription_worker.rb', line 97

def each_iteration(call_record, *_args)
  service = CallRecordProcessing::BulkTranscriptionService.new(call_record, model: @model)
  result = service.transcribe

  case result[:status]
  when :success
    @success_count += 1
    log_info "Transcribed ##{call_record.id}: #{result[:word_count]} words (#{result[:duration_secs]}s)" if (@success_count % 20).zero?
  when :skipped
    @skip_count += 1
  when :error
    @error_count += 1
    log_error "Failed ##{call_record.id}: #{result[:reason]} — #{result[:message]}"
  end

  sleep(rand(0.5..1.5))
rescue StandardError => e
  @error_count += 1
  log_error "Unexpected error for ##{call_record.id}: #{e.message}"
  ErrorReporting.error(e)
end

#on_complete ⇒ Object

On complete.

Returns:

  • (Object) —

    the result



122
123
124
 
# File 'app/workers/call_record_bulk_transcription_worker.rb', line 122

def on_complete
  log_info "Complete: #{@success_count} transcribed, #{@skip_count} skipped, #{@error_count} errors"
end