Class: Webhookdb::Backfiller

Inherits:
Object
  • Object
show all
Defined in:
lib/webhookdb/backfiller.rb

Overview

Implementation of a generic backfill pattern.

Defined Under Namespace

Modules: Bulk

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.do_retry_wait(seconds) ⇒ Object

Make this easy to mock



50
51
52
 
# File 'lib/webhookdb/backfiller.rb', line 50

def self.do_retry_wait(seconds)
  Kernel.sleep(seconds)
end

Instance Method Details

#_fetch_backfill_page_with_retry(pagination_token, last_backfilled: nil, attempt: 1) ⇒ Object 



54
55
56
57
58
59
60
61
62
 
# File 'lib/webhookdb/backfiller.rb', line 54

def _fetch_backfill_page_with_retry(pagination_token, last_backfilled: nil, attempt: 1)
  return self.fetch_backfill_page(pagination_token, last_backfilled:)
rescue Webhookdb::Http::BaseError => e
  raise e if attempt >= self.max_backfill_retry_attempts
  # Assume we'll never succeed on a 401, so don't bother retrying.
  raise e if e.is_a?(Webhookdb::Http::Error) && e.status == 401
  self.wait_for_retry_attempt(attempt:)
  return self._fetch_backfill_page_with_retry(pagination_token, last_backfilled:, attempt: attempt + 1)
end

#backfill(last_backfilled) ⇒ Object

Use nil last_backfilled for a full sync, pass it for an incremental. Should be service integration last_backfilled_at, the timestamp of the latest resource, etc.



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/webhookdb/backfiller.rb', line 17

def backfill(last_backfilled)
  pagination_token = nil
  loop do
    page, next_pagination_token = self._fetch_backfill_page_with_retry(
      pagination_token, last_backfilled:,
    )
    if page.nil?
      msg = "Fetching a page should return an empty array, not nil. The service response probably is missing a key?"
      raise TypeError, msg
    end
    pagination_token = next_pagination_token
    page.each do |item|
      self.handle_item(item)
    end
    Amigo::DurableJob.heartbeat
    break if pagination_token.blank?
    if Webhookdb.regression_mode?
      Webhookdb.logger.warn("regression_mode_backfill_termination", backfiller: self.to_s, pagination_token:)
      break
    end
  end
  self.flush_pending_inserts if self.respond_to?(:flush_pending_inserts)
end

#fetch_backfill_page(pagination_token, last_backfilled:) ⇒ Object

Raises:

  • (NotImplementedError) 


10
11
12
 
# File 'lib/webhookdb/backfiller.rb', line 10

def fetch_backfill_page(pagination_token, last_backfilled:)
  raise NotImplementedError
end

#handle_item(item) ⇒ Object

Called for each item.

Raises:

  • (NotImplementedError) 


6
7
8
 
# File 'lib/webhookdb/backfiller.rb', line 6

def handle_item(item)
  raise NotImplementedError
end

#max_backfill_retry_attempts ⇒ Object 



41
42
43
 
# File 'lib/webhookdb/backfiller.rb', line 41

def max_backfill_retry_attempts
  return 3
end

#wait_for_retry_attempt(attempt:) ⇒ Object 



45
46
47
 
# File 'lib/webhookdb/backfiller.rb', line 45

def wait_for_retry_attempt(attempt:)
  Webhookdb::Backfiller.do_retry_wait(attempt)
end