Class: WebhookLog

Inherits:
ApplicationRecord show all
Defined in:
app/models/webhook_log.rb

Overview

== Schema Information

Table name: webhook_logs
Database name: primary

id :bigint not null, primary key
category :string not null
data :jsonb not null
next_attempt :datetime
notes :text
process_attempts :integer default(0), not null
processed_at :datetime
provider :string not null
resource_type :string
response_data :jsonb
state :string default("ready"), not null
created_at :datetime not null
updated_at :datetime not null
external_id :string
resource_id :integer

Indexes

idx_webhook_logs_provider_category_state (provider,category,state)
idx_webhook_logs_resource (resource_type,resource_id)
idx_webhook_logs_state_created (state,created_at)
index_webhook_logs_on_category (category)
index_webhook_logs_on_external_id (external_id)
index_webhook_logs_on_next_attempt (next_attempt) USING brin

Constant Summary collapse

PROVIDERS =

Providers that can send webhooks

%w[assemblyai oxylabs sendgrid shipengine switchvox].freeze
CATEGORIES =

Categories by provider

{
  'assemblyai' => %w[transcription_complete],
  'oxylabs' => %w[price_check price_check_complete],
  'sendgrid' => %w[delivery bounce engagement suppression unknown],
  'shipengine' => %w[tracking_update],
  'switchvox' => %w[new_voicemail checked_voicemail incoming_call route_to_extension call_answered call_hangup outgoing_call agent_login agent_logout unknown]
}.freeze
MAX_RETRY_ATTEMPTS =

Maximum retry attempts before moving to exception

5
RETRY_DELAYS =

Retry delay calculation (exponential backoff)

[5.minutes, 15.minutes, 1.hour, 4.hours, 24.hours].freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from ApplicationRecord

ransortable_attributes, #to_relation

Methods included from Models::EventPublishable

#publish_event

Instance Attribute Details

#category ⇒ Object (readonly) 



52
 
# File 'app/models/webhook_log.rb', line 52

validates :category, presence: true

#data ⇒ Object (readonly) 



53
 
# File 'app/models/webhook_log.rb', line 53

validates :data, presence: true

#provider ⇒ Object (readonly) 



51
 
# File 'app/models/webhook_log.rb', line 51

validates :provider, presence: true

Class Method Details

.awaiting_callback ⇒ ActiveRecord::Relation<WebhookLog>

A relation of WebhookLogs that are awaiting callback. Active Record Scope

Returns:

See Also:



98
99
100
 
# File 'app/models/webhook_log.rb', line 98

scope :awaiting_callback, -> {
  where(state: 'pending')
}

.create_pending!(provider:, category:, resource_type:, resource_id:, external_id: nil, data: {}, notes: nil) ⇒ WebhookLog

Class method to create a pending entry when submitting a job
Call this when you submit a job that expects a webhook callback

Parameters:

  • provider (String) —

    The webhook provider (e.g., 'assemblyai')

  • category (String) —

    The webhook category (e.g., 'transcription_complete')

  • resource_type (String) —

    Associated resource type (e.g., 'CallRecord', 'Video')

  • resource_id (Integer) —

    Associated resource ID

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

    Optional external ID from provider (e.g., job_id)

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

    Optional metadata about the submission

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

    Optional notes about the submission

Returns:



214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'app/models/webhook_log.rb', line 214

def self.create_pending!(provider:, category:, resource_type:, resource_id:, external_id: nil, data: {}, notes: nil)
  create!(
    provider: provider,
    category: category,
    resource_type: resource_type,
    resource_id: resource_id,
    external_id: external_id,
    data: data,
    state: 'pending',
    notes: notes
  )
end

.find_by_external_id(provider, external_id) ⇒ Object

Find any entry by external_id (for duplicate detection)
Uses the external_id column, not JSON data



292
293
294
295
296
297
298
 
# File 'app/models/webhook_log.rb', line 292

def self.find_by_external_id(provider, external_id)
  return nil if external_id.blank?

  where(provider: provider, external_id: external_id)
    .order(created_at: :desc)
    .first
end

.find_pending_for_resource(provider, category, resource_type, resource_id) ⇒ Object

Find a pending entry for a specific resource



301
302
303
304
305
306
307
308
309
310
311
 
# File 'app/models/webhook_log.rb', line 301

def self.find_pending_for_resource(provider, category, resource_type, resource_id)
  return nil if resource_type.blank? || resource_id.blank?

  where(
    provider: provider,
    category: category,
    resource_type: resource_type,
    resource_id: resource_id,
    state: 'pending'
  ).order(created_at: :desc).first
end

.for_resource ⇒ ActiveRecord::Relation<WebhookLog>

A relation of WebhookLogs that are for resource. Active Record Scope

Returns:

See Also:



102
103
104
 
# File 'app/models/webhook_log.rb', line 102

scope :for_resource, ->(resource) {
  where(resource_type: resource.class.name, resource_id: resource.id)
}

.ingest!(provider:, category:, data:, resource_type: nil, resource_id: nil, external_id: nil, notes: nil) ⇒ WebhookLog

Class method to ingest a webhook callback
Finds existing pending entry or creates a new ready entry
Prevents duplicates by checking external_id

Parameters:

  • provider (String) —

    The webhook provider (e.g., 'assemblyai')

  • category (String) —

    The webhook category (e.g., 'transcription_complete')

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

    Associated resource type

  • resource_id (Integer, nil) (defaults to: nil) —

    Associated resource ID

  • data (Hash) —

    The webhook payload

Returns:

  • (WebhookLog) —

    The updated or created log entry



236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
 
# File 'app/models/webhook_log.rb', line 236

def self.ingest!(provider:, category:, data:, resource_type: nil, resource_id: nil, external_id: nil, notes: nil)
  # First check for duplicate by external_id (prevents duplicate processing on webhook retries)
  if external_id.present?
    existing_any = find_by_external_id(provider, external_id)
    if existing_any
      # Already processed or being processed - skip duplicate
      if existing_any.processed? || existing_any.processing?
        Rails.logger.info "[WebhookLog] Duplicate webhook ignored: #{provider}/#{external_id} already #{existing_any.state}"
        return existing_any
      end
      # Existing pending entry - transition to ready
      if existing_any.pending?
        existing_any.data = existing_any.data.merge(data)
        existing_any.notes = notes if notes.present?
        existing_any.receive_callback!
        return existing_any
      end
      # Existing in retry/exception state - could be a retry from provider, update and reprocess
      if existing_any.retry? || existing_any.exception?
        existing_any.data = existing_any.data.merge(data)
        existing_any.notes = notes if notes.present?
        existing_any.reprocess!
        return existing_any
      end
    end
  end

  # Try to find existing pending entry by resource
  existing = find_pending_for_resource(provider, category, resource_type, resource_id)

  if existing
    # Update the pending entry with callback data and transition to ready
    existing.data = existing.data.merge(data)
    existing.external_id = external_id if external_id.present? && existing.external_id.blank?
    existing.notes = notes if notes.present?
    existing.receive_callback!
    existing
  else
    # No pending entry found - create a new one in ready state
    # This handles webhooks that arrive without a corresponding pending entry
    Rails.logger.info "[WebhookLog] Creating new ready entry for #{provider}/#{category} #{resource_type}:#{resource_id}"
    create!(
      provider: provider,
      category: category,
      resource_type: resource_type,
      resource_id: resource_id,
      external_id: external_id,
      data: data,
      notes: notes,
      state: 'ready'
    )
  end
end

.payload_contains ⇒ ActiveRecord::Relation<WebhookLog>

A relation of WebhookLogs that are payload contains. Active Record Scope

Returns:

See Also:



112
113
114
115
116
117
 
# File 'app/models/webhook_log.rb', line 112

scope :payload_contains, ->(term) {
  return none if term.blank?

  sanitized = "%#{sanitize_sql_like(term)}%"
  where('data::text ILIKE :term OR response_data::text ILIKE :term', term: sanitized)
}

.providers_for_select ⇒ Object

Helper for tom-select dropdowns



369
370
371
 
# File 'app/models/webhook_log.rb', line 369

def self.providers_for_select
  PROVIDERS.map { |p| [p.titleize, p] }
end

.ransackable_associations(_auth_object = nil) ⇒ Object 



125
126
127
 
# File 'app/models/webhook_log.rb', line 125

def self.ransackable_associations(_auth_object = nil)
  []
end

.ransackable_attributes(_auth_object = nil) ⇒ Object

Ransack attributes for search



120
121
122
123
 
# File 'app/models/webhook_log.rb', line 120

def self.ransackable_attributes(_auth_object = nil)
  %w[id provider category resource_type resource_id external_id state process_attempts
     created_at updated_at processed_at next_attempt notes]
end

.ransackable_scopes(_auth_object = nil) ⇒ Object

Custom Ransack scopes for advanced searching



130
131
132
 
# File 'app/models/webhook_log.rb', line 130

def self.ransackable_scopes(_auth_object = nil)
  %i[payload_contains]
end

.recent ⇒ ActiveRecord::Relation<WebhookLog>

A relation of WebhookLogs that are recent. Active Record Scope

Returns:

See Also:



106
107
108
 
# File 'app/models/webhook_log.rb', line 106

scope :recent, ->(hours = 24) {
  where(created_at: hours.hours.ago..)
}

.requiring_processing ⇒ ActiveRecord::Relation<WebhookLog>

A relation of WebhookLogs that are requiring processing. Active Record Scope

Returns:

See Also:



84
85
86
87
88
 
# File 'app/models/webhook_log.rb', line 84

scope :requiring_processing, -> {
  where(state: 'ready')
    .or(where(state: 'retry').where('next_attempt <= ?', Time.current))
    .order(:created_at)
}

.stale_pending ⇒ ActiveRecord::Relation<WebhookLog>

A relation of WebhookLogs that are stale pending. Active Record Scope

Returns:

See Also:



91
92
93
94
95
 
# File 'app/models/webhook_log.rb', line 91

scope :stale_pending, ->(threshold = 1.hour) {
  where(state: 'pending')
    .where(created_at: ..threshold.ago)
    .order(:created_at)
}

.states_for_select ⇒ Object

Helper for tom-select dropdowns



364
365
366
 
# File 'app/models/webhook_log.rb', line 364

def self.states_for_select
  state_machine.states.map { |s| [s.name.to_s.titleize, s.name.to_s] }
end

Instance Method Details

#data_json ⇒ Object

Virtual attribute for editing data as JSON string
Used in the edit form to allow manual payload editing



58
59
60
61
62
 
# File 'app/models/webhook_log.rb', line 58

def data_json
  JSON.pretty_generate(data) if data.present?
rescue JSON::GeneratorError
  data.to_s
end

#data_json=(json_string) ⇒ Object 



64
65
66
67
68
69
70
71
72
 
# File 'app/models/webhook_log.rb', line 64

def data_json=(json_string)
  return if json_string.blank?

  self.data = JSON.parse(json_string)
rescue JSON::ParserError => e
  @data_json_error = e.message
  # Keep the raw string so we can show an error and preserve user input
  @data_json_raw = json_string
end

#display_name ⇒ Object

Display name for logs



379
380
381
 
# File 'app/models/webhook_log.rb', line 379

def display_name
  "#{provider.titleize} - #{category} (##{id})"
end

#human_state_name ⇒ Object

Human readable state name



374
375
376
 
# File 'app/models/webhook_log.rb', line 374

def human_state_name
  state.to_s.titleize
end

#process! ⇒ Object

Process this webhook log entry
Delegates to the appropriate processor based on provider/category



315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
 
# File 'app/models/webhook_log.rb', line 315

def process!
  return unless can_start_processing?

  start_processing!

  begin
    result = processor_class.call(self)
    self.response_data = result if result.is_a?(Hash)
    self.notes = nil # Clear any previous error notes on success
    complete!
  rescue StandardError => e
    self.notes = "#{e.class}: #{e.message}\n#{e.backtrace&.first(5)&.join("\n")}"

    if process_attempts >= MAX_RETRY_ATTEMPTS
      fail!
      ErrorReporting.error(e, webhook_log_id: id, provider: provider, category: category)
    else
      schedule_retry!
      Rails.logger.warn "[WebhookLog] Scheduled retry for #{id}: #{e.message}"
    end
  end
end

#processor_class ⇒ Object

Get the appropriate processor class for this webhook



339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
 
# File 'app/models/webhook_log.rb', line 339

def processor_class
  case provider
  when 'assemblyai'
    WebhookProcessors::AssemblyaiProcessor
  when 'oxylabs'
    WebhookProcessors::OxylabsProcessor
  when 'sendgrid'
    WebhookProcessors::SendgridProcessor
  when 'shipengine'
    WebhookProcessors::ShipengineProcessor
  when 'switchvox'
    WebhookProcessors::SwitchvoxProcessor
  else
    raise NotImplementedError, "No processor for provider: #{provider}"
  end
end

#resource ⇒ Object

Find the associated resource



357
358
359
360
361
 
# File 'app/models/webhook_log.rb', line 357

def resource
  return nil if resource_type.blank? || resource_id.blank?

  resource_type.constantize.find_by(id: resource_id)
end