Class: Analytic::BudgetFact::Refresher

Inherits:
Object
  • Object
show all
Defined in:
app/models/analytic/budget_fact/refresher.rb

Overview

Per-run, instance-isolated recomputation of Analytic::BudgetFact rows for a
single (year, month).

Why this is an instance and not a pile of class methods

A refresh builds large in-memory indexes — ledger/budget aggregates keyed by
company / ledger-account / business-unit / project / supplier, plus the
project-id and supplier-id sets derived from them. Those indexes hold the
data for one (year, month) run.

Previously they lived in class instance variables on Analytic::BudgetFact
with no reset, so two refreshes running at once shared (and clobbered) the
same indexes. That is not hypothetical here: BudgetRefresherAllWorker fans
out twelve monthly BudgetRefresherWorker jobs onto the :budgets
queue, which the default Sidekiq capsule serves at concurrency 16 (it is not
an isolation capsule), and a daily BudgetRefresherIncrementalWorker cron
refreshes the current month. Multiple (year, month) runs genuinely overlap.
The earlier stopgap — a single class-level Mutex around the whole refresh —
only serialized runs within one process (zero protection across processes)
and turned every concurrent refresh into a global bottleneck, without curing
the underlying shared-mutable-state design.

Holding the indexes on a fresh Refresher instance per run makes concurrent
refreshes naturally isolated: each owns its own ivars, nothing is shared, no
mutex is needed, and it is correct under any thread/process topology. See
doc/tasks/202606252210_MEMOIZATION_THREAD_SAFETY_AUDIT.md §3e.

The public entry point stays refresh_data_for_month;
this class is its implementation.

rubocop:disable Metrics/ClassLength -- one cohesive refresh pipeline; the index
lifecycle (build → look up → persist) reads best in one place rather than split
across collaborators purely to satisfy the length metric.

:reek:TooManyInstanceVariables -- the many multi-level index maps ARE the point:
holding the per-run aggregates as instance state (not class ivars) is what makes
concurrent refreshes isolated.

Instance Method Summary collapse

Constructor Details

#initialize(year, month, incremental: false, progress_block: nil) ⇒ Refresher

Returns a new instance of Refresher.

Parameters:

  • year (Integer) —

    the year to refresh

  • month (Integer) —

    the month to refresh

  • incremental (Boolean) (defaults to: false) —

    if true, upsert (update or insert) instead of
    delete-all-then-reinsert

  • progress_block (Proc, nil) (defaults to: nil) —

    optional ->(percent, message) callback
    for progress reporting



48
49
50
51
52
53
 
# File 'app/models/analytic/budget_fact/refresher.rb', line 48

def initialize(year, month, incremental: false, progress_block: nil)
  @year = year
  @month = month
  @incremental_mode = incremental
  @progress_block = progress_block
end

Instance Method Details

#run ⇒ Boolean

Recompute and persist budget facts for @year/@month.
rubocop:disable Metrics/AbcSize -- linear orchestration: preload → aggregate → process → persist, with progress/log calls inline.

Returns:

  • (Boolean) —

    true on completion



58
59
60
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
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
193
194
195
196
197
198
199
200
 
# File 'app/models/analytic/budget_fact/refresher.rb', line 58

def run
  start_time = Time.current

  refresh_mode = @incremental_mode ? 'incremental (upsert)' : 'full (delete-reinsert)'
  report_progress(0, "Starting #{refresh_mode} refresh for #{@year}-#{@month}")
  logger.info "[BudgetFact] Starting #{refresh_mode} refresh for #{@year}-#{@month}"

  # The full-mode delete of this month's existing facts is deferred to the
  # persistence phase below, where it shares one transaction with the reinsert
  # (atomic swap). Deleting here — before minutes of preload / aggregate work —
  # risked leaving the report empty if anything failed.

  business_units = BusinessUnit.order(:number).to_a
  budget_groups = BudgetGroup.all.to_a
  leaf_budget_groups = budget_groups.select(&:leaf?)

  @company_account_array = LedgerAccount.

  # ============================================================
  # OPTIMIZATION 1: Pre-load projects (suppliers loaded after aggregates)
  # ============================================================
  report_progress(10, 'Pre-loading projects...')
  t1 = Time.current
  @projects_by_id = LedgerProject.all.index_by(&:id)
  logger.info "[BudgetFact] Pre-loaded #{@projects_by_id.size} projects in #{(Time.current - t1).round(2)}s"

  # ============================================================
  # OPTIMIZATION 2: Pre-compute all aggregates with multi-level indexing for O(1) lookups
  # ============================================================
  report_progress(20, 'Computing ledger aggregates...')
  t2 = Time.current
  precompute_ledger_aggregates(@year, @month)
  logger.info "[BudgetFact] Ledger aggregates computed in #{(Time.current - t2).round(2)}s"

  report_progress(35, 'Computing budget aggregates...')
  t3 = Time.current
  precompute_budget_aggregates(@year, @month)
  logger.info "[BudgetFact] Budget aggregates computed in #{(Time.current - t3).round(2)}s"

  # ============================================================
  # OPTIMIZATION 2.5: Load only suppliers we need (from pre-computed aggregates)
  # This replaces loading ALL parties (600K+) with only needed suppliers (~1-5K)
  # ============================================================
  report_progress(37, 'Loading required suppliers...')
  t_suppliers = Time.current
  supplier_ids = collect_all_supplier_ids_from_aggregates
  @parties_by_id = supplier_ids.any? ? Party.where(id: supplier_ids).index_by(&:id) : {}
  logger.info "[BudgetFact] Pre-loaded #{@parties_by_id.size} suppliers (from #{supplier_ids.size} IDs) in #{(Time.current - t_suppliers).round(2)}s"

  # ============================================================
  # OPTIMIZATION 3: Pre-load ALL existing dimensions into cache
  # This eliminates database lookups during processing
  # ============================================================
  report_progress(38, 'Pre-loading existing dimensions...')
  t4 = Time.current
  @dimension_cache = {}
  Analytic::BudgetDimension.find_each do |dim|
    cache_key = {
      company_id: dim.company_id,
      ledger_account_ids: dim.,
      business_unit_id: dim.business_unit_id,
      ledger_project_id: dim.ledger_project_id,
      supplier_id: dim.supplier_id,
      budget_group_id: dim.budget_group_id,
      parent_id: dim.parent_id
    }.to_s
    @dimension_cache[cache_key] = dim
  end
  logger.info "[BudgetFact] Pre-loaded #{@dimension_cache.size} dimensions in #{(Time.current - t4).round(2)}s"

  @facts_to_insert = []

  # Company configurations for budget report generation
  company_configs = [
    { dimension: '1', id: Company::USA, ledger_field: :company_amount, budget_field: :amount },
    { dimension: '2', id: Company::CAN, ledger_field: :company_amount, budget_field: :amount },
    { dimension: '4', id: Company::NLD, ledger_field: :company_amount, budget_field: :amount },
    { dimension: 'all', id: [Company::USA, Company::CAN], ledger_field: :consolidated_amount, budget_field: :consolidated_amount }
  ]

  report_progress(40, 'Processing budget groups...')
  t5 = Time.current
  total_iterations = company_configs.size * budget_groups.size
  current_iteration = 0

  company_configs.each do |company|
    budget_groups.each do |bg|
      current_iteration += 1
      # Progress from 40% to 85% during main processing
      progress_pct = 40 + ((current_iteration.to_f / total_iterations) * 45).round
      report_progress(progress_pct, "Processing #{company[:dimension]}: #{bg.description}")

      process_budget_group(@year, @month, company, bg, business_units, leaf_budget_groups)
    end
  end
  logger.info "[BudgetFact] Main processing loop completed in #{(Time.current - t5).round(2)}s"

  # ============================================================
  # OPTIMIZATION 4: Batch insert/upsert all facts at once
  # ============================================================
  if @incremental_mode
    report_progress(90, "Upserting #{@facts_to_insert.size} facts...")
    if @facts_to_insert.any?
      logger.info "[BudgetFact] Upserting #{@facts_to_insert.size} facts (incremental mode)"
      # Upsert in batches of 1000 to avoid memory issues
      # Uses unique index on (year, month, budget_dimension_id)
      @facts_to_insert.each_slice(1000) do |batch|
        Analytic::BudgetFact.upsert_all(
          batch,
          unique_by: :idx_budget_facts_year_month_dimension_unique,
          update_only: %i[
            budget_month actual_month budget_month_diff budget_month_percent
            budget_accumulated actual_accumulated budget_accumulated_diff budget_accumulated_percent
            actual_month_previous actual_month_previous_diff actual_month_previous_percent
            actual_accumulated_previous actual_accumulated_previous_diff actual_accumulated_previous_percent
            is_revenue
          ]
        )
      end
    end
  else
    report_progress(90, "Batch inserting #{@facts_to_insert.size} facts...")
    logger.info "[BudgetFact] Batch inserting #{@facts_to_insert.size} facts"
    # Atomic swap: delete this month's existing facts and reinsert them in a
    # single transaction, so a mid-refresh failure can't leave the month
    # empty and readers never observe a half-rebuilt report. Delete is
    # unconditional (a full refresh clears the month even when it yields no
    # facts); the batched inserts are a no-op when there's nothing to insert.
    Analytic::BudgetFact.transaction do
      Analytic::BudgetFact.where(year: @year, month: @month).delete_all
      @facts_to_insert.each_slice(1000) do |batch|
        Analytic::BudgetFact.insert_all!(batch)
      end
    end
  end

  elapsed = Time.current - start_time
  refresh_mode = @incremental_mode ? 'incremental' : 'full'
  report_progress(100, "Completed #{refresh_mode} refresh in #{elapsed.round(1)}s")
  logger.info "[BudgetFact] Completed #{refresh_mode} refresh for #{@year}-#{@month} in #{elapsed.round(2)}s"

  true
end