Class: Edi::Amazon::FeedMessageSender

Inherits:
BaseEdiService show all
Defined in:
app/services/edi/amazon/feed_message_sender.rb

Overview

Service object: feed message sender.

Constant Summary

Constants included from RequestIdentifiable

RequestIdentifiable::REQUEST_ID_HEADERS

Constants included from Edi::AddressAbbreviator

Edi::AddressAbbreviator::MAX_LENGTH

Instance Attribute Summary

Attributes inherited from BaseEdiService

#orchestrator

Attributes inherited from BaseService

#options

Instance Method Summary collapse

Methods inherited from BaseEdiService

#amazon_feed_product_type, #duplicate_po_already_notified?, #initialize, #mark_duplicate_po_as_notified, #onboard_ordered_catalog_items, #report_order_creation_issues, #safe_process_edi_communication_log

Methods included from RequestIdentifiable

#partner_request_id

Methods included from Edi::AddressAbbreviator

#abbreviate_street, #collect_street_originals, #record_address_abbreviation_notes

Methods inherited from BaseService

#initialize, #log_debug, #log_error, #log_info, #log_warning, #logger, #tagged_logger

Constructor Details

This class inherits a constructor from Edi::BaseEdiService

Instance Method Details

#claim_budget(ecl) ⇒ Boolean

Returns false when this account has spent its window's budget.

Returns:

  • (Boolean) —

    false when this account has spent its window's budget



182
183
184
185
186
187
188
189
190
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 182

def claim_budget(ecl)
  return true if FeedSubmissionBudget.claim!(orchestrator.transporter_profile)

  defer(ecl, FeedSubmissionBudget::PERIOD.to_i,
        "Deferred: #{orchestrator.transporter_profile} feed budget spent for this window")
  logger.info "Deferring ECL #{ecl.id} (#{feed_category}/#{orchestrator.partner}) — " \
              "#{orchestrator.transporter_profile} budget spent"
  false
end

#create_feed(ecl, feed_transport, feed_document_id) ⇒ void

This method returns an undefined value.

STEP 4: create the feed itself, binding the uploaded document to the
marketplace. On success the returned feedId is the handle
Edi::Amazon::FeedSubmissionResultProcessor later polls.



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 141

def create_feed(ecl, feed_transport, feed_document_id)
  data = %(
          {
            "feedType":"#{feed_type}",
            "marketplaceIds": ["#{orchestrator.marketplace}"],
            "inputFeedDocumentId": "#{feed_document_id}"
          }
        )
  res = feed_transport.send_data(data, "#{orchestrator.feed_message_remote_path}/feeds", 'POST')
  ecl.notes += " | HTTP CODE: #{res[:http_result]&.status}, HTTP BODY: #{res[:http_result]&.body}, REQUEST ID: #{partner_request_id(res[:http_result])}, Timestamp: #{Time.current.to_datetime.to_fs(:crm_default)}"
  logger.info "Result: HTTP CODE: #{res[:http_result]&.status}, HTTP BODY: #{res[:http_result]&.body}"
  return ecl.error unless res[:success] && (body = res[:http_result]&.body.to_s).present?

  ecl.transaction_id = JSON.parse(body).with_indifferent_access[:feedId]
  ecl.transmit_datetime = Time.current
  ecl.start_process!
end

#defer(ecl, seconds, reason) ⇒ Object

Reschedules an unsent ECL and RECORDS why. It stays ready on purpose —
requiring_processing honours transmit_after — but a row that changes
nothing is unreadable: a deferred feed used to look exactly like a sender
that hung, which is how the EU stall hid for seven weeks. update_columns
keeps the state machine out of it, so updated_at is set explicitly.



197
198
199
200
201
202
203
204
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 197

def defer(ecl, seconds, reason)
  ecl.update_columns(
    transmit_after: Time.current + seconds,
    process_attempts: ecl.process_attempts.to_i + 1,
    notes: "#{reason} — Timestamp: #{Time.current.to_datetime.to_fs(:crm_default)}",
    updated_at: Time.current
  )
end

#ecl_in_queue ⇒ Object 



206
207
208
209
210
211
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 206

def ecl_in_queue
  EdiCommunicationLog.requiring_processing
                     .where(partner: orchestrator.partner)
                     .where(category: feed_category)
                     .order(:created_at)
end

#full_snapshot_feed? ⇒ Boolean

Whether every message of this feed category carries the COMPLETE state, so
an older unsent one is superseded rather than owed. False by default —
ListingMessageSender patches a single item per message, and dropping one
of those loses the edit. Overridden true by the inventory and price senders.

Returns:

  • (Boolean) 


73
74
75
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 73

def full_snapshot_feed?
  false
end

#instantiate_transporter(transporter, transporter_profile = nil, options = {}) ⇒ Transport::HttpSellerApiConnection, Transport::HttpApiUploadConnection

Builds the transport connection for the given transporter type.

Parameters:

  • transporter (Symbol) —

    :http_seller_api or :http_api_upload

  • transporter_profile (Symbol, nil) (defaults to: nil) —

    Heatwave::Configuration profile key

  • options (Hash) (defaults to: {}) —

    extra options merged into the transport constructor hash

Options Hash (options):

  • headers (Hash) —

    extra HTTP headers merged into the connection's headers (e.g. Content-Type for feed uploads)

  • logger (Logger) —

    logger to use instead of Rails.logger

Returns:

Raises:

  • (RuntimeError) —

    when transporter is not recognized



222
223
224
225
226
227
228
229
230
231
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 222

def instantiate_transporter(transporter, transporter_profile = nil, options = {})
  case transporter
  when :http_seller_api
    Transport::HttpSellerApiConnection.new({ profile: transporter_profile }.merge(options))
  when :http_api_upload
    Transport::HttpApiUploadConnection.new({ profile: transporter_profile }.merge(options))
  else
    raise "Unknown transporter: #{transporter}"
  end
end

#process(edi_communication_logs = nil) ⇒ Object 



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 77

def process(edi_communication_logs = nil)
  ecls = edi_communication_logs ? [edi_communication_logs].flatten : supersede_stale(ecl_in_queue.to_a)
  return ecls if ecls.empty?

  feed_transport = instantiate_transporter(orchestrator.transporter, orchestrator.transporter_profile)

  ecls.each do |ecl|
    next unless claim_budget(ecl)

    submit_feed(ecl, feed_transport)
  rescue HTTP::RateLimitExceededError => e
    retry_after_seconds = e.retry_after.to_i.positive? ? e.retry_after.to_i : 1.hour.to_i
    defer(ecl, retry_after_seconds, "Amazon SP-API rate limited (429). #{e.message}")
    logger.warn "Amazon SP-API rate limited for ECL #{ecl.id} (#{feed_category}/#{orchestrator.partner}). " \
                "Scheduled retry via transmit_after: #{ecl.transmit_after&.iso8601} " \
                "(#{retry_after_seconds}s from now). #{e.message}"
  end
  ecls
end

#submit_feed(ecl, feed_transport) ⇒ void

This method returns an undefined value.

The four-step SP-API feed dance, extracted from #process so the queue
loop reads as a queue loop. Each step's HTTP result is appended to
+ecl.notes+ and any failure short-circuits to +ecl.error+.

Parameters:



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 104

def submit_feed(ecl, feed_transport)
  logger.info "Sending feed data to #{orchestrator.partner}"
  # STEP 1 Create Feed Document
  # VERY IMPORTANT contentType must match Content-Type header in the subsequent upload PUT request exactly!
  ct = ecl.data_type == 'xml' ? 'text/xml; charset=UTF-8' : 'application/json'
  res = feed_transport.send_data(%({'contentType':'#{ct}'}), "#{orchestrator.feed_message_remote_path}/documents", 'POST')
  ecl.notes = "HTTP CODE: #{res[:http_result]&.status}, HTTP BODY: #{res[:http_result]&.body}, HTTP METHOD: 'POST', REQUEST ID: #{partner_request_id(res[:http_result])}, Timestamp: #{Time.current.to_datetime.to_fs(:crm_default)}"
  logger.info "Result: HTTP CODE: #{res[:http_result]&.status}, HTTP BODY: #{res[:http_result]&.body}"
  return ecl.error unless res[:success] && (body = res[:http_result]&.body.to_s).present?

  json_hash = JSON.parse(body).with_indifferent_access
  return ecl.error unless upload_feed_document(ecl, json_hash[:url], ct)

  create_feed(ecl, feed_transport, json_hash[:feedDocumentId])
end

#supersede_stale(ecls) ⇒ Array<EdiCommunicationLog>

Each full-snapshot message carries the complete state, so an older unsent
one is stale data, not a backlog item — and replaying it is actively
harmful: nine EU marketplaces sharing one SP-API account replayed 90+
superseded snapshots an hour, spent the account's whole feed allowance on
data from July, and 429'd the snapshot that mattered. Every one of those
ECLs sat in ready with NULL notes, indistinguishable from a hang.

Parameters:

Returns:



168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 168

def supersede_stale(ecls)
  return ecls unless full_snapshot_feed?
  return ecls if ecls.size <= 1

  *stale, current = ecls
  stale.each do |ecl|
    ecl.update(notes: "Superseded by ECL #{current.id} — snapshot never sent")
    ecl.archive
  end
  logger.info "Superseded #{stale.size} stale #{feed_category} snapshot(s) for #{orchestrator.partner}"
  [current]
end

#upload_feed_document(ecl, url, content_type) ⇒ Boolean

STEP 2 & 3: construct and upload the feed document to the signed URL
returned by step 1 — no auth needed on that URL.
VERY IMPORTANT Content-Type header must match contentType in the previous
Create Feed Document request exactly!

Returns:

  • (Boolean) —

    whether the upload succeeded



126
127
128
129
130
131
132
133
134
 
# File 'app/services/edi/amazon/feed_message_sender.rb', line 126

def upload_feed_document(ecl, url, content_type)
  # Both transports are now Faraday-backed and expose #status: the
  # feed_document_uploader (HttpApiUploadConnection) and feed_transport
  # (HttpSellerApiConnection, migrated to faraday-retry).
  uploader = instantiate_transporter(:http_api_upload, nil, { headers: { 'Content-Type': content_type } })
  res = uploader.send_data(ecl.data, url, 'POST')
  ecl.notes += " | HTTP CODE: #{res[:http_result]&.status}, HTTP BODY: #{res[:http_result]&.body}, REQUEST ID: #{partner_request_id(res[:http_result])}, Timestamp: #{Time.current.to_datetime.to_fs(:crm_default)}"
  res[:success]
end