Class: AdSpendSyncWorker

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

Overview

Nightly per-campaign ad spend into SourceDataPoint, so ROI/ROAS is
answerable from the database across every paid channel.

Assistant conversation 4167 exposed the gap this closes: asked for marketing
ROI, Sunny could only reach Google (its only spend tool) and silently omitted
the Microsoft and OpenAI money. Spend lands on the campaign's Source — the
same key revenue attributes to (doc/architecture/SOURCE_ATTRIBUTION.md) — so
return-on-ad-spend is a join, not a reconciliation.

This worker is an orchestrator: it fans the providers out into a Sidekiq Pro
batch, one job each (AdSpendProviderSyncWorker), and AdSpendDayFinalizer
decides what the day's outcome means once they all land. The providers used
to run sequentially inside this job, which meant an Amazon throttle failed
the whole day and a re-run re-fetched Google, Microsoft and OpenAI as well —
spending their quota to recover someone else's. Fanning out keeps a retry
scoped to the provider that actually failed. Amazon report waiting is now a
chain of short polling jobs, so no worker thread is pinned between polls.

Idempotent: metrics upsert on (source, metric_type, period), so re-running a
date overwrites rather than doubling. Runs after every campaign sync (Google
01:00 → Amazon 02:15 CT) so a campaign created today already has its Source
row before its spend arrives.

Constant Summary collapse

ADAPTERS =

Provider key → adapter. Keys are what gets passed across the job boundary,
so they must stay stable; the adapter class itself can't be serialized.

{
  'google' => Marketing::AdSpend::GoogleAdapter,
  'microsoft_ads' => Marketing::AdSpend::MicrosoftAdapter,
  'openai_ads' => Marketing::AdSpend::OpenaiAdapter,
  'amazon_ads' => Marketing::AdSpend::AmazonAdapter
}.freeze
DEEP_HISTORY_PROVIDERS =

Providers whose history reaches further back than Amazon's, which retains
60 days for Sponsored Brands and gates the whole provider-day on it. A
historical backfill scopes itself to these, or every pre-retention day
fires an Amazon request that is certain to be rejected — and each rejection
opens the cohort-wide cooldown, stalling the chain for everyone.

%w[google microsoft_ads openai_ads].freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.providers_for(providers) ⇒ Array<String>

Only nil means "every provider". An explicitly empty array is refused
rather than widened: perform(date, nil, []) reading as a full run would
fan Amazon out across a historical range — the exact doomed request this
scoping exists to avoid — and it would do it silently.

Parameters:

  • providers (Array<String>, nil) —

    nil means every provider

Returns:

  • (Array<String>) —

    the providers to fan out, validated

Raises:

  • (ArgumentError) —

    on an empty or unrecognised selection



54
55
56
57
58
59
60
61
62
63
64
 
# File 'app/workers/ad_spend_sync_worker.rb', line 54

def self.providers_for(providers)
  return ADAPTERS.keys if providers.nil?

  requested = Array(providers).map(&:to_s).uniq
  raise ArgumentError, 'ad-spend provider list is empty; pass nil to run every provider' if requested.empty?

  unknown = requested - ADAPTERS.keys
  raise ArgumentError, "unknown ad-spend provider(s): #{unknown.join(', ')}" if unknown.any?

  requested
end

.scope_for(providers) ⇒ String?

Run-state keyspace for a provider selection. A full run gets nil so its
markers stay on the shared key. See Marketing::AdSpend::RunState.

Parameters:

  • providers (Array<String>, nil)

Returns:

  • (String, nil) 


71
72
73
74
75
76
 
# File 'app/workers/ad_spend_sync_worker.rb', line 71

def self.scope_for(providers)
  selected = providers_for(providers).sort
  return nil if selected == ADAPTERS.keys.sort

  selected.join('+')
end

.throttleable?(providers) ⇒ Boolean

Only Amazon opens the cohort-wide cooldown, so a run without it has no
reason to stand down for one — which is the whole point of scoping a
historical backfill to the other three.

Parameters:

  • providers (Array<String>, nil)

Returns:

  • (Boolean) 


84
85
86
 
# File 'app/workers/ad_spend_sync_worker.rb', line 84

def self.throttleable?(providers)
  providers_for(providers).include?(Marketing::AdSpend::AmazonAdapter::PROVIDER)
end

Instance Method Details

#perform(date_string = nil, backfill_end = nil, providers = nil) ⇒ String

Returns the batch id, useful for tracing a day in the Sidekiq UI.

Parameters:

  • date_string (String, nil) (defaults to: nil) —

    ISO date; defaults to yesterday (providers
    finalize a day's numbers after midnight in their own timezone)

  • backfill_end (String, nil) (defaults to: nil) —

    when driving a backfill range, the last
    date in it — the finalizer uses this to advance the chain

  • providers (Array<String>, nil) (defaults to: nil) —

    restrict the fan-out; nil runs every
    provider, which is what the nightly and the self-heal sweep want

Returns:

  • (String) —

    the batch id, useful for tracing a day in the Sidekiq UI



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
 
# File 'app/workers/ad_spend_sync_worker.rb', line 95

def perform(date_string = nil, backfill_end = nil, providers = nil)
  date = date_string.present? ? Date.parse(date_string) : Date.current - 1
  selected = self.class.providers_for(providers)
  scope = self.class.scope_for(providers)
  batch_id = SecureRandom.uuid

  # Claim the day HERE, not in the callers, because every dispatcher routes
  # through this method — the nightly cron, the backfill chain, the self-heal
  # sweep, a console re-run. While the claim lived in AdSpendBackfillWorker
  # only the chain recorded it, so `in_flight?` was blind to nightly runs and
  # the finalizer's release cleared a key nobody had set.
  #
  # The claim has to come FIRST: the batch can complete — and its finalizer
  # clear this very key — before a claim written afterwards lands, leaving a
  # stale marker that blocks the date for its whole three-hour TTL. But a claim
  # written first must be released if the enqueue raises, or nothing will ever
  # clear it: no batch exists, so no :complete callback fires.
  Marketing::AdSpend::RunState.mark_in_flight!(date, scope:)
  claimed = true

  batch = Sidekiq::Batch.new
  batch.description = "Ad spend #{date}#{" (#{scope})" if scope}"
  batch.on(:complete, AdSpendDayFinalizer,
           'date' => date.to_s, 'backfill_end' => backfill_end, 'providers' => providers)
  batch.jobs do
    selected.each { |provider| AdSpendProviderSyncWorker.perform_async(provider, date.to_s, batch_id) }
  end

  batch.bid
rescue StandardError
  # Only release a claim we actually took. An invalid provider list raises
  # before the claim, and clearing then would drop a concurrent full run's
  # in-flight marker for this date.
  Marketing::AdSpend::RunState.clear_in_flight!(date, scope: scope) if claimed
  raise
end