Class: Mammoth::DeliveredEnvelopeStore

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

Overview

SQLite-backed ledger of downstream deliveries.

The PostgreSQL replication boundary is at-least-once. Mammoth therefore keeps a small delivery ledger so a transaction replayed by the upstream replication source after restart does not have to be delivered downstream again.

Constant Summary collapse

SCHEMA =

SQLite schema used to bootstrap the delivered-envelope ledger.

<<~SQL
  CREATE TABLE IF NOT EXISTS delivered_envelopes (
    id INTEGER PRIMARY KEY,
    idempotency_key TEXT NOT NULL,
    source_name TEXT NOT NULL,
    slot_name TEXT NOT NULL,
    destination_name TEXT NOT NULL,
    delivery_unit TEXT NOT NULL,
    transaction_id TEXT,
    source_position TEXT,
    delivered_at TEXT NOT NULL,

    UNIQUE (idempotency_key)
  );

  CREATE INDEX IF NOT EXISTS idx_delivered_envelopes_source
  ON delivered_envelopes(source_name, slot_name);

  CREATE INDEX IF NOT EXISTS idx_delivered_envelopes_destination
  ON delivered_envelopes(destination_name);

  CREATE INDEX IF NOT EXISTS idx_delivered_envelopes_source_position
  ON delivered_envelopes(source_position);
SQL

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(sqlite_store) ⇒ DeliveredEnvelopeStore

Returns a new instance of DeliveredEnvelopeStore.



40
41
42
43
 
# File 'lib/mammoth/delivered_envelope_store.rb', line 40

def initialize(sqlite_store)
  @sqlite_store = sqlite_store
  ensure_schema!
end

Instance Attribute Details

#sqlite_store ⇒ Object (readonly)

Returns the value of attribute sqlite_store.



38
39
40
 
# File 'lib/mammoth/delivered_envelope_store.rb', line 38

def sqlite_store
  @sqlite_store
end

Instance Method Details

#all ⇒ Object

rubocop:enable Metrics/MethodLength



81
82
83
 
# File 'lib/mammoth/delivered_envelope_store.rb', line 81

def all
  database.execute("SELECT * FROM delivered_envelopes ORDER BY id")
end

#count(destination: nil) ⇒ Integer

Count delivered envelopes.

Returns:

  • (Integer) —

    delivered envelope count



88
89
90
91
92
93
94
 
# File 'lib/mammoth/delivered_envelope_store.rb', line 88

def count(destination: nil)
  if destination
    database.get_first_value("SELECT COUNT(*) FROM delivered_envelopes WHERE destination_name = ?", [destination])
  else
    database.get_first_value("SELECT COUNT(*) FROM delivered_envelopes")
  end
end

#counts_by_destination ⇒ Array<Hash>

Count delivered envelopes grouped by destination.

Returns:

  • (Array<Hash>) —

    rows with destination_name and count



99
100
101
102
103
 
# File 'lib/mammoth/delivered_envelope_store.rb', line 99

def counts_by_destination
  database.execute(
    "SELECT destination_name, COUNT(*) AS count FROM delivered_envelopes GROUP BY destination_name"
  )
end

#delivered?(idempotency_key) ⇒ Boolean

Returns:

  • (Boolean) 


45
46
47
48
49
50
 
# File 'lib/mammoth/delivered_envelope_store.rb', line 45

def delivered?(idempotency_key)
  !database.execute(
    "SELECT 1 FROM delivered_envelopes WHERE idempotency_key = ? LIMIT 1",
    [idempotency_key]
  ).empty?
end

#record!(idempotency_key:, source_name:, slot_name:, destination_name:, delivery_unit:, transaction_id:, source_position:) ⇒ Object

rubocop:disable Metrics/MethodLength



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/mammoth/delivered_envelope_store.rb', line 53

def record!(idempotency_key:, source_name:, slot_name:, destination_name:, delivery_unit:, transaction_id:,
            source_position:)
  database.execute(
    <<~SQL,
      INSERT OR IGNORE INTO delivered_envelopes(
        idempotency_key, source_name, slot_name, destination_name, delivery_unit,
        transaction_id, source_position, delivered_at
      )
      VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    SQL
    [
      idempotency_key,
      source_name,
      slot_name,
      destination_name,
      delivery_unit,
      transaction_id,
      source_position,
      Time.now.utc.iso8601
    ]
  )
  database.get_first_row(
    "SELECT * FROM delivered_envelopes WHERE idempotency_key = ? LIMIT 1",
    [idempotency_key]
  )
end