Class: AmazonSqsNotificationPollerWorker

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

Overview

Sidekiq worker: drains the SP-API notifications SQS queue into the webhook
pipeline (ingest-then-process, see the webhooks skill).

SP-API pushes notifications to the queue — most types directly via the SQS
destination, the EventBridge-workflow types via an EventBridge rule that
targets the same queue. This worker long-polls on a 1-minute cron cadence,
lands each message as a WebhookLog, and deletes it from SQS only after
the log durably commits. Retries and error handling live in the WebhookLog
state machine, NOT in SQS redelivery; a message that fails to ingest stays
on the queue and reaches the DLQ after 5 receives. Standard queues are
at-least-once/unordered — WebhookLog.ingest! dedups by
(provider, external_id) where external_id is the SP-API notificationId.

No-ops (with a log line) until the amazon_notifications credentials
profile exists, so the cron entry is safe to deploy ahead of the runtime
IAM key.

See doc/tasks/202607152125_AMAZON_SPAPI_SQS_NOTIFICATIONS.md

Constant Summary collapse

PROVIDER =

Provider key in WebhookLog::PROVIDERS / CATEGORIES.

'amazon_sp_api'
RUN_BUDGET =

Stop draining shortly before the next cron tick would fire.

50.seconds
MAX_MESSAGES_PER_RECEIVE =

SQS maximum per receive call.

10
WAIT_TIME_SECONDS =

SQS long-poll maximum — an empty queue holds the call open this long.

20
REQUIRED_CONFIG_KEYS =

Every value Aws::SQS::Client needs — explicit nils would NOT fall back to
the SDK's default provider chain, they'd fail on the first call.

%i[queue_url region aws_access_key_id aws_secret_access_key].freeze
HEARTBEAT_CACHE_KEY =

Liveness heartbeat for the CRM monitor (Amazon::NotificationsHealth):
written after every completed drain cycle.

'amazon_sp_api_notifications:poller_ran_at'
LAST_DRAIN_CACHE_KEY =

Last non-empty drain: { at:, count: }.

'amazon_sp_api_notifications:poller_last_drain'

Instance Method Summary collapse

Instance Method Details

#perform ⇒ Object

Runs the job.

Returns:

  • (Object) —

    the result



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'app/workers/amazon_sqs_notification_poller_worker.rb', line 49

def perform
  # Guard on the full runtime profile, not mere presence: the credentials
  # are nested under `production:` so non-production environments must
  # never drain the live queue — in dev, Heatwave::Configuration returns
  # the unresolved outer hash (none of these keys) and this skips.
  if config.blank? || REQUIRED_CONFIG_KEYS.any? { |key| config[key].blank? }
    Rails.logger.info '[AmazonSqsNotificationPoller] amazon_notifications credentials not configured, skipping'
    return
  end

  deadline = RUN_BUDGET.from_now
  drained = 0
  while Time.current < deadline
    messages = receive_messages
    break if messages.empty?

    messages.each { |message| drained += 1 if ingest(message) }
  end

  Rails.cache.write(HEARTBEAT_CACHE_KEY, Time.current, expires_in: 2.days)
  return unless drained.positive?

  Rails.cache.write(LAST_DRAIN_CACHE_KEY, { at: Time.current, count: drained }, expires_in: 2.days)
  Rails.logger.info "[AmazonSqsNotificationPoller] drained #{drained} notification(s)"
end