Class: Mammoth::TransactionEnvelopeSerializer

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

Overview

Serializes CDC transaction envelopes into webhook payloads.

Mammoth uses transaction envelopes as the safest delivery and checkpointing boundary for concurrent delivery. A transaction payload preserves the commit position and groups the row-level changes that belong to the same database transaction.

Constant Summary collapse

PAYLOAD_TYPE =

Default payload type for transaction webhook delivery.

"transaction.committed"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(envelope) ⇒ TransactionEnvelopeSerializer

Returns a new instance of TransactionEnvelopeSerializer.

Parameters:

  • envelope (CDC::Core::TransactionEnvelope) —

    transaction envelope



27
28
29
30
31
32
33
 
# File 'lib/mammoth/transaction_envelope_serializer.rb', line 27

def initialize(envelope)
  unless envelope.is_a?(CDC::Core::TransactionEnvelope)
    raise ArgumentError, "envelope must be a CDC::Core::TransactionEnvelope"
  end

  @envelope = envelope
end

Class Method Details

.call(envelope) ⇒ Hash

Serialize a CDC::Core::TransactionEnvelope into a Hash.

Parameters:

  • envelope (CDC::Core::TransactionEnvelope) —

    transaction envelope

Returns:

  • (Hash) —

    webhook-ready transaction payload



22
23
24
 
# File 'lib/mammoth/transaction_envelope_serializer.rb', line 22

def self.call(envelope)
  new(envelope).call
end

Instance Method Details

#call ⇒ Hash

Return the webhook payload.

Returns:

  • (Hash) —

    transaction webhook payload



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/mammoth/transaction_envelope_serializer.rb', line 38

def call
  event_payloads = envelope.events.map { |event| EventSerializer.call(event) }
  {
    "event_id" => envelope_metadata["event_id"] || SecureRandom.uuid,
    "type" => PAYLOAD_TYPE,
    "source" => first_event_value(event_payloads, "source") || EventSerializer::DEFAULT_SOURCE,
    "transaction_id" => envelope.transaction_id,
    "source_position" => source_position(event_payloads),
    "commit_lsn" => source_position(event_payloads),
    "committed_at" => committed_at,
    "event_count" => event_payloads.length,
    "events" => event_payloads,
    "metadata" => envelope_metadata
  }
end

#to_json(*_args) ⇒ String

Return JSON representation of the transaction payload.

Returns:

  • (String) —

    JSON representation



57
58
59
 
# File 'lib/mammoth/transaction_envelope_serializer.rb', line 57

def to_json(*_args)
  JSON.generate(call)
end