moreminimore-chat/app/services/analytics/account_daily_processor.rb

# Aggregates one account's conversation activity into the immutable daily metrics
# (conversation_daily_metrics + customer_daily_metrics) for a single date in the
# account's reporting timezone, and writes the LLM classification tags onto the
# live conversations.
#
# Flow per account/date:
#   1. Compute the UTC day window from account.reporting_timezone (default UTC).
#   2. Select conversations with activity inside that window.
#   3. For each conversation, classify via Llm::AnalyticsClassifier (skipped when
#      disabled), re-tag it (replace prior M2-managed tags, keep manual labels),
#      and collect per-conversation / per-customer metrics.
#   4. Upsert conversation_daily_metric (unique account+date) and
#      customer_daily_metric (unique account+contact+date).
#
# Daily snapshots are immutable: counts captured here are a point-in-time view; the
# live conversation tags may be re-tagged later without changing historical snapshots.
class Analytics::AccountDailyProcessor
  TOPIC_TAG_PREFIX = 'topic:'.freeze
  DEAL_TAG_PREFIX = 'deal:'.freeze

  # @param account [Account]
  # @param date [Date] the reporting date in the account's timezone
  def self.perform(account:, date:)
    new(account: account, date: date).perform
  end

  def initialize(account:, date:)
    @account = account
    @date = date
  end

  def perform
    return unless scope_enabled?

    window = day_window
    conversations = active_conversations(window)
    return if conversations.empty?

    catalog_paths = catalog_paths
    metrics = build_metrics(conversations, window, catalog_paths)
    upsert_metrics(metrics)
  end

  private

  def scope_enabled?
    @account.reporting_timezone.blank? || ActiveSupport::TimeZone[@account.reporting_timezone].present?
  end

  def timezone
    @timezone ||= ActiveSupport::TimeZone[@account.reporting_timezone] || Time.zone
  end

  # [start_utc, end_utc) covering the account-local date.
  def day_window
    start_tz = timezone.parse(@date.to_s)
    end_tz = timezone.parse((@date + 1).to_s)
    [start_tz.utc, end_tz.utc]
  end

  # Conversations with activity (created or last activity) inside the account-local day.
  def active_conversations(window)
    start_utc, end_utc = window
    @account.conversations
            .where('created_at < ?', end_utc)
            .where('created_at >= ? OR last_activity_at >= ?', start_utc, start_utc)
  end

  # All tag-hierarchy paths in the account's catalog — used to detect & replace
  # previously-written product tags on re-classification.
  def catalog_paths
    @account.product_catalog_entries
            .order(:group_name, :subgroup_name, :product_name)
            .map(&:tag_hierarchy)
            .map { |parts| parts.join('>') }
  end

  def build_metrics(conversations, _window, catalog_paths)
    {
      conversation: {
        message_count: 0,
        conversation_count: 0,
        resolved_count: 0,
        unresolved_count: 0,
        top_tags: [],
        sale_tags: [],
        deal_outcomes: {},
        agent_breakdown: Hash.new(0),
        team_breakdown: Hash.new(0),
        channel_breakdown: Hash.new(0),
        inbox_breakdown: Hash.new(0)
      },
      customers: {}
    }.tap do |acc|
      conversations.each { |conversation| accumulate(acc, conversation, catalog_paths) }
    end
  end

  def accumulate(acc, conversation, catalog_paths)
    classification = Llm::AnalyticsClassifier.classify(account: @account, conversation: conversation)
    return if classification.error.present? || classification.disabled?

    apply_tags(conversation, classification, catalog_paths)

    cm = acc[:conversation]
    cm[:conversation_count] += 1
    message_count = chat_message_count(conversation)
    cm[:message_count] += message_count

    if conversation.resolved?
      cm[:resolved_count] += 1
    else
      cm[:unresolved_count] += 1
    end

    cm[:top_tags] |= (classification.topics + classification.products).compact
    cm[:sale_tags] |= classification.products
    cm[:deal_outcomes]['totals'] ||= {}
    cm[:deal_outcomes]['totals'][classification.deal] = cm[:deal_outcomes]['totals'][classification.deal].to_i + 1

    cm[:agent_breakdown][conversation.assignee_id] += 1 if conversation.assignee_id
    cm[:team_breakdown][conversation.team_id] += 1 if conversation.team_id
    cm[:channel_breakdown][channel_key(conversation)] += 1
    cm[:inbox_breakdown][conversation.inbox_id] += 1 if conversation.inbox_id

    customer_key = conversation.contact_id
    customer = acc[:customers][customer_key] ||= {
      message_count: 0, conversation_count: 0, resolved_count: 0, unresolved_count: 0,
      top_tags: [], deal_outcomes: {}, agent_ids: []
    }
    customer[:conversation_count] += 1
    customer[:message_count] += message_count
    if conversation.resolved?
      customer[:resolved_count] += 1
    else
      customer[:unresolved_count] += 1
    end
    customer[:top_tags] |= (classification.topics + classification.products).compact
    customer[:deal_outcomes][classification.deal] = customer[:deal_outcomes][classification.deal].to_i + 1
    customer[:agent_ids] = (customer[:agent_ids] | [conversation.assignee_id]).compact if conversation.assignee_id
  end

  def chat_message_count(conversation)
    conversation.messages.chat.count
  end

  def channel_key(conversation)
    conversation.inbox&.channel_type&.demodulize
  end

  # Write classifier tags onto the conversation: remove any prior M2-managed tags
  # (topic:/deal: prefixes + current catalog paths), then add the fresh ones.
  # Manual labels are preserved.
  def apply_tags(conversation, classification, catalog_paths)
    current = conversation.label_list.to_a
    m2_managed = current.select do |tag|
      tag.start_with?(TOPIC_TAG_PREFIX) || tag.start_with?(DEAL_TAG_PREFIX) || catalog_paths.include?(tag)
    end
    next_tags = current - m2_managed

    next_tags += classification.products
    next_tags += classification.topics.map { |topic| "#{TOPIC_TAG_PREFIX}#{topic}" }
    # Persist the deal outcome on the live conversation so reports can filter by it.
    next_tags << "#{DEAL_TAG_PREFIX}#{classification.deal}" if classification.deal.present?

    conversation.update!(label_list: next_tags.uniq)
  rescue ActiveRecord::RecordInvalid => e
    Rails.logger.error("[Analytics] conversation #{conversation.id} tag update failed: #{e.message}")
  end

  def upsert_metrics(metrics)
    conv = metrics[:conversation]
    conv_attrs = {
      account_id: @account.id,
      date: @date,
      timezone: @account.reporting_timezone.presence || 'UTC',
      message_count: conv[:message_count],
      conversation_count: conv[:conversation_count],
      resolved_count: conv[:resolved_count],
      unresolved_count: conv[:unresolved_count],
      top_tags: conv[:top_tags],
      sale_tags: conv[:sale_tags],
      deal_outcomes: conv[:deal_outcomes],
      agent_breakdown: conv[:agent_breakdown].map { |id, count| { 'agent_id' => id, 'count' => count } },
      team_breakdown: conv[:team_breakdown].map { |id, count| { 'team_id' => id, 'count' => count } },
      channel_breakdown: conv[:channel_breakdown].map { |ch, count| { 'channel' => ch, 'count' => count } },
      inbox_breakdown: conv[:inbox_breakdown].map { |id, count| { 'inbox_id' => id, 'count' => count } }
    }

    ConversationDailyMetric.upsert(
      conv_attrs,
      unique_by: %i[account_id date],
      update_only: %i[message_count conversation_count resolved_count unresolved_count top_tags sale_tags
                      deal_outcomes agent_breakdown team_breakdown channel_breakdown inbox_breakdown timezone]
    )

    metrics[:customers].each do |contact_id, data|
      CustomerDailyMetric.upsert(
        {
          account_id: @account.id,
          contact_id: contact_id,
          date: @date,
          timezone: @account.reporting_timezone.presence || 'UTC',
          message_count: data[:message_count],
          conversation_count: data[:conversation_count],
          resolved_count: data[:resolved_count],
          unresolved_count: data[:unresolved_count],
          top_tags: data[:top_tags],
          deal_outcomes: data[:deal_outcomes],
          agent_ids: data[:agent_ids].to_a
        },
        unique_by: %i[account_id contact_id date],
        update_only: %i[message_count conversation_count resolved_count unresolved_count top_tags
                        deal_outcomes agent_ids timezone]
      )
    end
  end
end