Class: Mammoth::DeliveryWorker

Inherits:
Object
  • Object
show all
Defined in:
lib/mammoth/delivery_worker.rb

Overview

Delivers normalized events with retry, checkpoint, and dead-letter handling.

DeliveryWorker is Mammoth's first reliable delivery unit. It intentionally keeps the delivery contract small: attempt webhook delivery, advance the checkpoint after success, and persist the failed event to the dead letter queue after retry exhaustion.

Constant Summary collapse

DEFAULT_SOURCE =

Default source name used when an event does not provide one.

"postgresql"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(sink:, checkpoint_store:, dead_letter_store:, source_name:, slot_name:, publication_name:, max_attempts:, retry_schedule:, delivered_envelope_store: nil, sleeper: Kernel.method(:sleep), route_filter: nil, enabled: true) ⇒ DeliveryWorker

Returns a new instance of DeliveryWorker.

Parameters:

  • sink (#deliver) —

    destination sink

  • checkpoint_store (Mammoth::CheckpointStore) —

    checkpoint persistence

  • dead_letter_store (Mammoth::DeadLetterStore) —

    dead letter persistence

  • delivered_envelope_store (Mammoth::DeliveredEnvelopeStore, nil) (defaults to: nil) —

    downstream delivery ledger

  • source_name (String) —

    logical source name

  • slot_name (String) —

    replication slot name

  • publication_name (String) —

    publication name

  • max_attempts (Integer) —

    maximum delivery attempts

  • retry_schedule (Array<Integer>) —

    retry wait schedule in seconds

  • sleeper (#call) (defaults to: Kernel.method(:sleep)) —

    sleep strategy, injectable for tests

  • route_filter (Mammoth::RouteFilter, nil) (defaults to: nil) —

    optional destination route matcher

  • enabled (Boolean) (defaults to: true) —

    whether this destination accepts new deliveries



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/mammoth/delivery_worker.rb', line 29

def initialize(sink:, checkpoint_store:, dead_letter_store:, source_name:, slot_name:, publication_name:,
               max_attempts:, retry_schedule:, delivered_envelope_store: nil, sleeper: Kernel.method(:sleep),
               route_filter: nil, enabled: true)
  @sink = sink
  @checkpoint_store = checkpoint_store
  @dead_letter_store = dead_letter_store
  @delivered_envelope_store = delivered_envelope_store || DeliveredEnvelopeStore.new(checkpoint_store.sqlite_store)
  @source_name = source_name
  @slot_name = slot_name
  @publication_name = publication_name
  @max_attempts = max_attempts
  @retry_schedule = retry_schedule
  @sleeper = sleeper
  @route_filter = route_filter || RouteFilter.new
  @enabled = enabled
end

Instance Attribute Details

#checkpoint_store ⇒ Object (readonly)

Returns the value of attribute checkpoint_store.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def checkpoint_store
  @checkpoint_store
end

#dead_letter_store ⇒ Object (readonly)

Returns the value of attribute dead_letter_store.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def dead_letter_store
  @dead_letter_store
end

#delivered_envelope_store ⇒ Object (readonly)

Returns the value of attribute delivered_envelope_store.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def delivered_envelope_store
  @delivered_envelope_store
end

#enabled ⇒ Object (readonly)

Returns the value of attribute enabled.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def enabled
  @enabled
end

#max_attempts ⇒ Object (readonly)

Returns the value of attribute max_attempts.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def max_attempts
  @max_attempts
end

#publication_name ⇒ Object (readonly)

Returns the value of attribute publication_name.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def publication_name
  @publication_name
end

#retry_schedule ⇒ Object (readonly)

Returns the value of attribute retry_schedule.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def retry_schedule
  @retry_schedule
end

#route_filter ⇒ Object (readonly)

Returns the value of attribute route_filter.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def route_filter
  @route_filter
end

#sink ⇒ Object (readonly)

Returns the value of attribute sink.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def sink
  @sink
end

#sleeper ⇒ Object (readonly)

Returns the value of attribute sleeper.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def sleeper
  @sleeper
end

#slot_name ⇒ Object (readonly)

Returns the value of attribute slot_name.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def slot_name
  @slot_name
end

#source_name ⇒ Object (readonly)

Returns the value of attribute source_name.



14
15
16
 
# File 'lib/mammoth/delivery_worker.rb', line 14

def source_name
  @source_name
end

Class Method Details

.from_config(config, sink:, checkpoint_store:, dead_letter_store:, sleeper: Kernel.method(:sleep), delivery_policy: {}) ⇒ Mammoth::DeliveryWorker

Build a delivery worker from Mammoth configuration and stores.

Parameters:

Returns:



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/mammoth/delivery_worker.rb', line 54

def self.from_config(config, sink:, checkpoint_store:, dead_letter_store:, sleeper: Kernel.method(:sleep),
                     delivery_policy: {})
  new(
    sink: sink,
    checkpoint_store: checkpoint_store,
    dead_letter_store: dead_letter_store,
    source_name: config.dig("mammoth", "name"),
    slot_name: config.dig("replication", "slot"),
    publication_name: Array(config.dig("replication", "publications")).join(","),
    max_attempts: delivery_policy.fetch("max_attempts", config.dig("retry", "max_attempts")),
    retry_schedule: delivery_policy.fetch("schedule_seconds", config.dig("retry", "schedule_seconds")),
    sleeper: sleeper,
    route_filter: delivery_policy.fetch("route_filter", RouteFilter.new),
    enabled: delivery_policy.fetch("enabled", true)
  )
end

Instance Method Details

#deliver(event) ⇒ Hash

Deliver an event with retry, checkpoint, and DLQ handling.

Parameters:

  • event (Hash, #to_h) —

    normalized event

Returns:

  • (Hash) —

    delivery summary



83
84
85
 
# File 'lib/mammoth/delivery_worker.rb', line 83

def deliver(event)
  deliver_work(event, serializer: EventSerializer, delivery_method: :deliver)
end

#deliver_transaction(envelope) ⇒ Hash

Deliver a transaction envelope with retry, checkpoint, and DLQ handling.

Parameters:

  • envelope (#events, #transaction_id) —

    CDC transaction envelope

Returns:

  • (Hash) —

    delivery summary



75
76
77
 
# File 'lib/mammoth/delivery_worker.rb', line 75

def deliver_transaction(envelope)
  deliver_work(envelope, serializer: TransactionEnvelopeSerializer, delivery_method: :deliver_transaction)
end