Class: Mammoth::EventSerializer

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

Overview

Serializes CDC-core change events into webhook payloads.

The serializer projects Mammoth's sink payload from CDC vocabulary rather than pgoutput protocol vocabulary. That keeps webhook delivery independent from PostgreSQL-specific message shapes while preserving source metadata such as commit LSN and transaction identity when available.

Constant Summary collapse

DEFAULT_SOURCE =

Default source label used in serialized webhook payloads.

"postgresql"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(event) ⇒ EventSerializer

Returns a new instance of EventSerializer.

Parameters:

  • event (CDC::Core::ChangeEvent) —

    normalized CDC event

Raises:

  • (ArgumentError) 


27
28
29
30
31
 
# File 'lib/mammoth/event_serializer.rb', line 27

def initialize(event)
  raise ArgumentError, "event must be a CDC::Core::ChangeEvent" unless event.is_a?(CDC::Core::ChangeEvent)

  @event = event.to_h
end

Class Method Details

.call(event) ⇒ Hash

Serialize a core change event into a webhook-ready Hash.

Parameters:

  • event (CDC::Core::ChangeEvent) —

    normalized CDC event

Returns:

  • (Hash) —

    webhook payload



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

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

Instance Method Details

#call ⇒ Hash

Return the webhook payload.

Returns:

  • (Hash) —

    webhook payload



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

def call
  event_hash = stringify_keys(@event)
  metadata = stringify_keys(event_hash["metadata"] || {})
  {
    "event_id" => event_id(metadata),
    "source" => source(metadata),
    "operation" => normalize_operation(event_hash.fetch("operation")),
    "namespace" => event_hash["schema"],
    "entity" => event_hash["table"],
    "identity" => event_hash["primary_key"],
    "source_position" => event_hash["commit_lsn"],
    "transaction_id" => event_hash["transaction_id"],
    "occurred_at" => occurred_at(event_hash),
    "data" => event_data(event_hash),
    "changes" => metadata["changes"] || [],
    "metadata" => metadata
  }
end

#to_json(*_args) ⇒ String

Return JSON representation of the webhook payload.

Returns:

  • (String) —

    JSON representation of the payload



58
59
60
 
# File 'lib/mammoth/event_serializer.rb', line 58

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