Class: RetailerAnalysisWorker

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

Overview

Runs a retailer-catalog AI analysis: feeds Retailer::AnalysisPrompt to the
Sunny chat stack (Assistant::ChatService) on a dedicated
AssistantConversation, then flips retailer_analysis_status and broadcasts
the retailer dashboard's analysis panel. Enqueued by
Crm::RetailersController#analyze; follows the DailyFocusAnalysisRepWorker
shape (single ChatService generation path, metadata status, live stream
chunks + formatted completion broadcast to the chat page, Turbo broadcast
on completion).

Constant Summary collapse

HEARTBEAT_INTERVAL =

Seconds between processing-lock heartbeats during the run (mirrors
AssistantChatWorker) so the chat UI keeps showing "Sunny is thinking"
instead of the stale-heartbeat warning on long analyses.

30
STALE_PROCESSING_AFTER =

A processing conversation older than this no longer blocks a new run —
it means its worker died mid-flight.

30.minutes

Instance Method Summary collapse

Instance Method Details

#perform(catalog_id, conversation_id) ⇒ void

This method returns an undefined value.

Parameters:

  • catalog_id (Integer)
  • conversation_id (Integer) —

    the linked AssistantConversation



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'app/workers/retailer_analysis_worker.rb', line 37

def perform(catalog_id, conversation_id)
  catalog = Catalog.find_by(id: catalog_id)
  conversation = AssistantConversation.find_by(id: conversation_id)
  return unless catalog && conversation

  # Idempotency: one analysis per catalog at a time, and only for the
  # conversation still marked processing. Duplicate delivery or a
  # concurrent controller run must not pay for a second chat call —
  # the controller's create takes the same lock (Crm::RetailersController#analyze).
  AssistantConversation.with_advisory_lock_result("retailer_analysis:#{catalog_id}", timeout_seconds: 0) do
    next unless conversation.reload.retailer_analysis_status == 'processing'

    run_analysis(catalog, conversation)
  end
end