fix: Persist inbound stripe webhooks

app/controllers/webhooks_controller.rb

@@ -2,20 +2,16 @@
 
 class WebhooksController < ApplicationController
   def stripe
-    result = PaymentProviders::Stripe::HandleIncomingWebhookService.call(
+    result = InboundWebhooks::CreateService.call(
       organization_id: params[:organization_id],
+      webhook_source: :stripe,
       code: params[:code].presence,
-      body: request.body.read,
-      signature: request.headers['HTTP_STRIPE_SIGNATURE']
+      payload: request.body.read,
+      signature: request.headers["HTTP_STRIPE_SIGNATURE"],
+      event_type: params[:type]
     )
 
-    unless result.success?
-      if result.error.is_a?(BaseService::ServiceFailure) && result.error.code == 'webhook_error'
-        return head(:bad_request)
-      end
-
-      result.raise_if_error!
-    end
+    return head(:bad_request) unless result.success?
 
     head(:ok)
   end

app/jobs/clock/inbound_webhooks_cleanup_job.rb

@@ -0,0 +1,13 @@
+# frozen_string_literal: true
+
+module Clock
+  class InboundWebhooksCleanupJob < ApplicationJob
+    include SentryCronConcern
+
+    queue_as "clock"
+
+    def perform
+      InboundWebhook.where("updated_at < ?", 90.days.ago).destroy_all
+    end
+  end
+end

app/jobs/clock/inbound_webhooks_retry_job.rb

@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+module Clock
+  class InboundWebhooksRetryJob < ApplicationJob
+    include SentryCronConcern
+
+    queue_as "clock"
+
+    def perform
+      InboundWebhook.retriable.find_each do |inbound_webhook|
+        InboundWebhooks::ProcessJob.perform_later(inbound_webhook:)
+      end
+    end
+  end
+end

app/jobs/inbound_webhooks/process_job.rb

@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+module InboundWebhooks
+  class ProcessJob < ApplicationJob
+    queue_as :default
+
+    def perform(inbound_webhook:)
+      InboundWebhooks::ProcessService.call(inbound_webhook:).raise_if_error!
+    end
+  end
+end

app/models/inbound_webhook.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+class InboundWebhook < ApplicationRecord
+  WEBHOOK_PROCESSING_WINDOW = 2.hours
+
+  belongs_to :organization
+
+  validates :event_type, :payload, :source, :status, presence: true
+
+  STATUSES = {
+    pending: "pending",
+    processing: "processing",
+    processed: "processed",
+    failed: "failed"
+  }
+
+  enum :status, STATUSES
+
+  scope :retriable, -> { reprocessable.or(old_pending) }
+  scope :reprocessable, -> { processing.where("processing_at <= ?", WEBHOOK_PROCESSING_WINDOW.ago) }
+  scope :old_pending, -> { pending.where("created_at <= ?", WEBHOOK_PROCESSING_WINDOW.ago) }
+
+  def processing!
+    update!(status: :processing, processing_at: Time.zone.now)
+  end
+end
+
+# == Schema Information
+#
+# Table name: inbound_webhooks
+#
+#  id              :uuid             not null, primary key
+#  code            :string
+#  event_type      :string           not null
+#  payload         :jsonb            not null
+#  processing_at   :datetime
+#  signature       :string
+#  source          :string           not null
+#  status          :string           default("pending"), not null
+#  created_at      :datetime         not null
+#  updated_at      :datetime         not null
+#  organization_id :uuid             not null
+#
+# Indexes
+#
+#  index_inbound_webhooks_on_organization_id  (organization_id)
+#
+# Foreign Keys
+#
+#  fk_rails_...  (organization_id => organizations.id)
+#

app/services/inbound_webhooks/create_service.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module InboundWebhooks
+  class CreateService < BaseService
+    def initialize(organization_id:, webhook_source:, payload:, event_type:, code: nil, signature: nil)
+      @organization_id = organization_id
+      @webhook_source = webhook_source
+      @code = code
+      @payload = payload
+      @signature = signature
+      @event_type = event_type
+
+      super
+    end
+
+    def call
+      inbound_webhook = InboundWebhook.create!(
+        organization_id:,
+        source: webhook_source,
+        code:,
+        payload:,
+        signature:,
+        event_type:
+      )
+
+      after_commit do
+        InboundWebhooks::ProcessJob.perform_later(inbound_webhook:)
+      end
+
+      result.inbound_webhook = inbound_webhook
+      result
+    rescue ActiveRecord::RecordInvalid => e
+      result.record_validation_failure!(record: e.record)
+    end
+
+    private
+
+    attr_reader :organization_id, :webhook_source, :code, :payload, :signature, :event_type
+  end
+end

app/services/inbound_webhooks/process_service.rb

@@ -0,0 +1,56 @@
+# frozen_string_literal: true
+
+module InboundWebhooks
+  class ProcessService < BaseService
+    WEBHOOK_HANDLER_SERVICES = {
+      stripe: PaymentProviders::Stripe::HandleIncomingWebhookService
+    }
+
+    def initialize(inbound_webhook:)
+      @inbound_webhook = inbound_webhook
+
+      super
+    end
+
+    def call
+      return result if within_processing_window?
+      return result if inbound_webhook.failed?
+      return result if inbound_webhook.processed?
+
+      inbound_webhook.processing!
+
+      handler_result = handler_service_klass.call(inbound_webhook:)
+
+      unless handler_result.success?
+        inbound_webhook.failed!
+        return handler_result
+      end
+
+      inbound_webhook.processed!
+
+      result.inbound_webhook = inbound_webhook
+      result
+    rescue
+      inbound_webhook.failed!
+      raise
+    end
+
+    private
+
+    attr_reader :inbound_webhook
+
+    def handler_service_klass
+      WEBHOOK_HANDLER_SERVICES.fetch(webhook_source) do
+        raise NameError, "Invalid inbound webhook source: #{webhook_source}"
+      end
+    end
+
+    def webhook_source
+      inbound_webhook.source.to_sym
+    end
+
+    def within_processing_window?
+      inbound_webhook.processing? && inbound_webhook.processing_at > InboundWebhook::WEBHOOK_PROCESSING_WINDOW.ago
+    end
+  end
+end

app/services/payment_providers/stripe/handle_incoming_webhook_service.rb

@@ -3,11 +3,10 @@
 module PaymentProviders
   module Stripe
     class HandleIncomingWebhookService < BaseService
-      def initialize(organization_id:, body:, signature:, code: nil)
-        @organization_id = organization_id
-        @body = body
-        @signature = signature
-        @code = code
+      extend Forwardable
+
+      def initialize(inbound_webhook:)
+        @inbound_webhook = inbound_webhook
 
         super
       end
@@ -22,7 +21,7 @@ def call
         return payment_provider_result unless payment_provider_result.success?
 
         event = ::Stripe::Webhook.construct_event(
-          body,
+          payload,
           signature,
           payment_provider_result.payment_provider&.webhook_secret
         )
@@ -42,7 +41,7 @@ def call
 
       private
 
-      attr_reader :organization_id, :body, :signature, :code
+      def_delegators :@inbound_webhook, :code, :organization_id, :payload, :signature
     end
   end
 end

clock.rb

@@ -119,6 +119,12 @@ module Clockwork
       .perform_later
   end
 
+  every(1.day, 'schedule:clean_inbound_webhooks', at: '01:10') do
+    Clock::InboundWebhooksCleanupJob
+      .set(sentry: {"slug" => 'lago_clean_inbound_webhooks', "cron" => '5 1 * * *'})
+      .perform_later
+  end
+
   unless ActiveModel::Type::Boolean.new.cast(ENV['LAGO_DISABLE_EVENTS_VALIDATION'])
     every(1.hour, 'schedule:post_validate_events', at: '*:05') do
       Clock::EventsValidationJob
@@ -146,4 +152,10 @@ module Clockwork
       .set(sentry: {"slug" => "lago_retry_failed_invoices", "cron" => '*/15 * * * *'})
       .perform_later
   end
+
+  every(15.minutes, "schedule:retry_inbound_webhooks") do
+    Clock::InboundWebhooksRetryJob
+      .set(sentry: {"slug" => "lago_retry_inbound_webhooks", "cron" => '*/15 * * * *'})
+      .perform_later
+  end
 end

db/migrate/20241213182343_create_inbound_webhooks.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+class CreateInboundWebhooks < ActiveRecord::Migration[7.1]
+  def change
+    create_table :inbound_webhooks, id: :uuid do |t|
+      t.string :source, null: false
+      t.string :event_type, null: false
+      t.jsonb :payload, null: false
+      t.string :status, null: false, default: 'pending'
+      t.belongs_to :organization, null: false, foreign_key: true, type: :uuid, index: true
+      t.string :code
+      t.string :signature
+
+      t.timestamps
+    end
+  end
+end

db/migrate/20241219122151_add_processing_at_to_inbound_webhooks.rb

@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class AddProcessingAtToInboundWebhooks < ActiveRecord::Migration[7.1]
+  def change
+    add_column :inbound_webhooks, :processing_at, :timestamp, precision: nil
+  end
+end

spec/clockwork_spec.rb

@@ -177,4 +177,46 @@
       expect(Clock::ProcessDunningCampaignsJob).to have_been_enqueued
     end
   end
+
+  describe "schedule:clean_inbound_webhooks" do
+    let(:job) { "schedule:clean_inbound_webhooks" }
+    let(:start_time) { Time.zone.parse("1 Apr 2022 00:01:00") }
+    let(:end_time) { Time.zone.parse("2 Apr 2022 00:00:00") }
+
+    it "enqueue a clean inbound webhooks job" do
+      Clockwork::Test.run(
+        file: clock_file,
+        start_time:,
+        end_time:,
+        tick_speed: 1.minute
+      )
+
+      expect(Clockwork::Test).to be_ran_job(job)
+      expect(Clockwork::Test.times_run(job)).to eq(1)
+
+      Clockwork::Test.block_for(job).call
+      expect(Clock::InboundWebhooksCleanupJob).to have_been_enqueued
+    end
+  end
+
+  describe "schedule:retry_inbound_webhooks" do
+    let(:job) { "schedule:retry_inbound_webhooks" }
+    let(:start_time) { Time.zone.parse("1 Apr 2022 00:05:00") }
+    let(:end_time) { Time.zone.parse("1 Apr 2022 00:20:00") }
+
+    it "enqueue a retry inbound webhooks job" do
+      Clockwork::Test.run(
+        file: clock_file,
+        start_time:,
+        end_time:,
+        tick_speed: 1.minute
+      )
+
+      expect(Clockwork::Test).to be_ran_job(job)
+      expect(Clockwork::Test.times_run(job)).to eq(1)
+
+      Clockwork::Test.block_for(job).call
+      expect(Clock::InboundWebhooksRetryJob).to have_been_enqueued
+    end
+  end
 end