Class: Hecks::Adapters::Postgres

Inherits:
Object
  • Object
show all
Includes:
Codec, SchemaBuilder, SqlQueryBuilder
Defined in:
lib/hecks/adapters/driven/postgres.rb,
lib/hecks/adapters/driven/postgres/codec.rb,
lib/hecks/adapters/driven/postgres/schema_builder.rb

Overview

The plain Postgres store — flat, one table per aggregate, real typed columns for scalars and jsonb for nested/list attributes, exactly the shape Sqlite already made for its own table. Sibling to PostgresEra (postgres_era.rb), which is the same database with full lineage/era machinery on top — pick this one unless a domain actually needs to survive a live shape change. See docs/implemented/postgres-era-adapter-split-plan.md for why the two are split.

No hecks_eras, no lineage, no advisory-lock-per-write for era tracking, no lineage_capable?/era_check! — this class simply doesn't define those methods at all, and the capability idiom elsewhere already treats their absence as "not lineage-capable".

Storage shape (see postgres/schema_builder.rb for the DDL, postgres/codec.rb for the encode/decode):

  • One real column per attribute, typed for a scalar (SQL_TYPES, text default), jsonb for a nested (value-object) or list-typed attribute — never JSON-in-TEXT the way Sqlite has to, since Postgres has a native jsonb type.
  • append and project are two real Postgres statements — save/ delete wrap them in ONE transaction, same "the journal insert and the snapshot stay atomic" reasoning PostgresEra#append's own comment gives: a crash between the two must never leave a half-written state. append/project stay plain, individually- callable methods (never wrapping their own transaction) so AppendOnly#recover!'s replay — project alone, no append — keeps working the same way it does on every other adapter.

Defined Under Namespace

Modules: Codec, SchemaBuilder

Constant Summary collapse

SQL_TYPES = 
{ "Integer" => "bigint", "Float" => "double precision" }.freeze

Constants included from SqlQueryBuilder

SqlQueryBuilder::COMPARATORS

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from SqlQueryBuilder

#query

Constructor Details

#initialize(aggregate:, settings: {}, root: nil) ⇒ Postgres

Returns a new instance of Postgres.



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 92

def initialize(aggregate:, settings: {}, root: nil)
  @aggregate = aggregate
  @db = self.class.connect_for(aggregate.name, settings)
  # THE OPTIONAL saga-persistence capability's own scoping column
  # (§2/§4) — falls back to the aggregate's own storage name for a
  # directly-instantiated adapter (specs), same fallback shape
  # Sqlite's own @domain already uses.
  @domain = (
    if settings.key?(:domain)
      settings[:domain]
    elsif settings.key?("domain")
      settings["domain"]
    else
      aggregate.storage_name
    end
  ).to_s

  create_aggregate_table!
  create_entry_table!
  create_event_table!
  create_saga_table!
end

Instance Attribute Details

#aggregate ⇒ Object (readonly)

Returns the value of attribute aggregate.



50
51
52
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 50

def aggregate
  @aggregate
end

Class Method Details

.connect_for(name, settings) ⇒ Object 



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
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 54

def self.connect_for(name, settings)
  # LAZY, ON PURPOSE — same reasoning as PostgresEra's own
  # connect_for: a domain that never wires Postgres should never
  # need the gem installed.
  require "pg"

  declared = settings.key?(:database) ? settings[:database] : settings["database"]
  if declared.to_s.empty?
    raise Runtime::WiringError,
          "#{name} binds Postgres, which needs a database connection, " \
          "but its world declares no \"database\"."
  end

  connection =
    if declared.start_with?("postgres://", "postgresql://")
      PG.connect(declared)
    else
      PG.connect(dbname: declared)
    end

  # SHARED-INSTANCE ISOLATION — same as PostgresEra's own: a
  # domain that declares `schema` is sharing its Postgres instance
  # with other domains, so every unqualified reference this
  # adapter constructs resolves through search_path. A domain with
  # no `schema` setting keeps Postgres's own default (public).
  schema = settings.key?(:schema) ? settings[:schema] : settings["schema"]
  connection.exec("SET search_path TO #{connection.quote_ident(schema)}") if schema.to_s != ""

  # QUIET ON PURPOSE — same reasoning as PostgresEra's own: a
  # schema/table that already exists is the ORDINARY case on every
  # boot after the first, not news.
  connection.exec("SET client_min_messages = warning")
  connection
rescue PG::Error => error
  raise Runtime::WiringError,
        "cannot bind Postgres at #{declared} for #{name}: #{error.message.strip}"
end

Instance Method Details

#all(order_by: nil, direction: :asc) ⇒ Object

order_by IS A RUNTIME VALUE — see Sqlite#all's own reasoning; whitelisted the identical way before it ever reaches order_expression.



127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 127

def all(order_by: nil, direction: :asc)
  order_sql = "ORDER BY id"
  if order_by
    name = order_by.to_s.split(".").first
    raise Runtime::WiringError, "#{@aggregate.name} has no attribute #{order_by.inspect} to order by" unless @aggregate.lifecycle&.field.to_s == name || @aggregate.attribute(name)

    spec = QuerySpecification::Common::OrderBy.new(field: order_by, direction: direction)
    order_sql = "ORDER BY #{order_clause(spec, nil)}"
  end

  @db.exec("SELECT * FROM #{quoted_table} #{order_sql}").map { |row| instance_from_row(row) }
end

#append(entry) ⇒ Object 



142
143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 142

def append(entry)
  @db.exec_params(
    "INSERT INTO #{quoted_entry_table} (aggregate_id, operation, state, mirrors) VALUES ($1, $2, $3, $4)",
    # `mirrors` (unlike `state`) is a NULLABLE column — an absent
    # mirrors hash must bind a real SQL NULL, not the four-character
    # JSON text `"null"` (`JSON.generate(nil)`), or a future `IS NULL`
    # check against it would never match. Same guard `sqlite.rb`/
    # `d1.rb`/`postgres_era.rb` already use for their own journal's
    # `mirrors` column.
    [entry.id, entry.operation, JSON.generate(entry.state), entry.mirrors && JSON.generate(entry.mirrors)]
  )
  entry
end

#atomic_put(entry, insert_only: false) ⇒ Object

Classification, journal append and snapshot replacement are one Postgres transaction. A row lock cannot serialize two first writers — there is no row to lock yet — so a transaction-scoped advisory lock on schema/table + aggregate id owns that missing-row race as well. Once a writer acquires it, the preceding writer has committed and status can be read from the materialized table without a runtime-side find.



233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 233

def atomic_put(entry, insert_only: false)
  status = nil
  @db.transaction do
    @db.exec_params(
      "SELECT pg_advisory_xact_lock(" \
      "hashtext(current_schema() || ':' || $1), hashtext($2))",
      [table, entry.id.to_s]
    )
    exists = !@db.exec_params(
      "SELECT 1 FROM #{quoted_table} WHERE id = $1",
      [entry.id.to_s]
    ).ntuples.zero?
    if insert_only && exists
      status = :conflicted
      next
    end
    status = exists ? :replaced : :inserted
    append(entry)
    project(entry)
  end
  status
end

#count ⇒ Object 



140
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 140

def count = @db.exec("SELECT COUNT(*) FROM #{quoted_table}")[0]["count"].to_i

#delete(id) ⇒ Object 



256
257
258
259
260
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 256

def delete(id)
  entry = Ports::Persistence::Entry.new(operation: "delete", id: id.to_s, state: nil)
  @db.transaction { append(entry); project(entry) }
  true
end

#delete_saga(process_manager:, correlation:) ⇒ Object 



295
296
297
298
299
300
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 295

def delete_saga(process_manager:, correlation:)
  @db.exec_params(
    "DELETE FROM hecks_saga_instances WHERE domain = $1 AND process_manager = $2 AND correlation = $3",
    [@domain, process_manager.to_s, correlation.to_s]
  )
end

#each_saga ⇒ Object 



302
303
304
305
306
307
308
309
310
311
312
313
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 302

def each_saga
  return enum_for(:each_saga) unless block_given?

  @db.exec_params(
    "SELECT process_manager, correlation, state, memory, completed_compensations FROM hecks_saga_instances WHERE domain = $1",
    [@domain]
  ).each do |row|
    yield row["process_manager"], row["correlation"], row["state"],
          JSON.parse(row["memory"], symbolize_names: true),
          JSON.parse(row["completed_compensations"] || "[]", symbolize_names: true)
  end
end

#entries ⇒ Object 



197
198
199
200
201
202
203
204
205
206
207
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 197

def entries
  @db.exec("SELECT aggregate_id, operation, state, mirrors FROM #{quoted_entry_table} ORDER BY sequence").map do |row|
    state = JSON.parse(row["state"])
    Ports::Persistence::Entry.new(
      operation: row["operation"] || "save",
      id:        row["aggregate_id"],
      state:     state&.transform_keys(&:to_sym),
      mirrors:   row["mirrors"] && JSON.parse(row["mirrors"])
    )
  end
end

#events ⇒ Object 



269
270
271
272
273
274
275
276
277
278
279
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 269

def events
  @db.exec("SELECT * FROM events ORDER BY id").map do |row|
    Runtime::Event.new(
      name:        row["name"],
      aggregate:   row["aggregate"],
      id:          row["aggregate_id"],
      payload:     JSON.parse(row["payload"], symbolize_names: true),
      occurred_at: row["occurred_at"]
    )
  end
end

#find(id) ⇒ Object 



117
118
119
120
121
122
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 117

def find(id)
  result = @db.exec_params("SELECT * FROM #{quoted_table} WHERE id = $1", [id.to_s])
  return nil if result.ntuples.zero?

  instance_from_row(result[0])
end

#persistence_capabilities ⇒ Object 



52
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 52

def persistence_capabilities = [:atomic_put, :optimistic_concurrency]

#project(entry, expected_version: nil) ⇒ Object

expected_version: requests optimistic-concurrency CAS (see persistence_capabilities/Ports::Persistence::AppendOnly#save). hecks_version is ADAPTER BOOKKEEPING — never in persisted_fields (Codec), so it never reaches decode's domain-state hash. It goes in the INSERT column list at 1 (a genuinely new row) and bumps by one in the ON CONFLICT DO UPDATE branch; when expected_version is given, that UPDATE branch additionally requires hecks_version = expected_version to apply at all — Postgres's own INSERT ... ON CONFLICT DO UPDATE ... WHERE, which gates only whether the CONFLICT branch's update applies. A genuinely new row never reaches that branch at all, so it always inserts regardless of this WHERE. RETURNING hecks_version plus ntuples.zero? is how a real version mismatch is told apart from an ordinary write: zero rows back means the conflict branch's WHERE excluded the row entirely — the version had already moved — so nil is returned for the caller (AppendOnly#save) to treat as "stale, no-op".



172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 172

def project(entry, expected_version: nil)
  return @db.exec_params("DELETE FROM #{quoted_table} WHERE id = $1", [entry.id]) if entry.delete?

  instance = Runtime::Instance.new(aggregate: @aggregate, id: entry.id, state: entry.state)
  columns  = (["id"] + persisted_fields.map { |field| field[:name].to_s } + ["hecks_version"])
  values   = [instance.id.to_s] + persisted_fields.map { |field| encode_field(field, instance[field[:name]]) } + [1]
  updates  = persisted_fields.map { |field| "#{quote_ident(field[:name])} = EXCLUDED.#{quote_ident(field[:name])}" } +
             ["hecks_version = #{quoted_table}.hecks_version + 1"]

  sql = "INSERT INTO #{quoted_table} (#{columns.map { |c| quote_ident(c) }.join(', ')}) " \
        "VALUES (#{(1..columns.size).map { |n| "$#{n}" }.join(', ')}) " \
        "ON CONFLICT (id) DO UPDATE SET #{updates.join(', ')}"
  if expected_version
    values += [expected_version]
    sql += " WHERE #{quoted_table}.hecks_version = $#{values.size}"
  end
  sql += " RETURNING hecks_version"

  result = @db.exec_params(sql, values)
  return nil if result.ntuples.zero?

  instance.version = result[0]["hecks_version"].to_i
  instance
end

#record_event(event) ⇒ Object 



262
263
264
265
266
267
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 262

def record_event(event)
  @db.exec_params(
    "INSERT INTO events (name, aggregate, aggregate_id, payload, occurred_at) VALUES ($1, $2, $3, $4, $5)",
    [event.name, event.aggregate, event.id.to_s, JSON.generate(event.payload), event.occurred_at]
  )
end

#reset! ⇒ Object 



209
210
211
212
213
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 209

def reset!
  @db.exec("DELETE FROM #{quoted_table}")
  @db.exec("DELETE FROM #{quoted_entry_table}")
  self
end

#save(instance) ⇒ Object

ONE TRANSACTION, not the plain append-then-project two-step a file-based adapter needs a crash-recovery replay for (Heki) — real Postgres ACID atomicity is sitting right there, so a crash between the journal insert and the table upsert must not leave the two disagreeing. append/project themselves stay plain, transaction-free methods (see the class comment above) — the transaction lives here, the one caller that runs both together.



222
223
224
225
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 222

def save(instance)
  entry = Ports::Persistence::Entry.new(operation: "save", id: instance.id.to_s, state: instance.state.dup)
  @db.transaction { append(entry); project(entry) }
end

#save_saga(process_manager:, correlation:, state:, memory:, completed_compensations: []) ⇒ Object

── the OPTIONAL saga-persistence capability (§2) — same DDL and shape as PostgresEra's own (postgres_era.rb), not lineage- specific, copied verbatim.



284
285
286
287
288
289
290
291
292
293
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 284

def save_saga(process_manager:, correlation:, state:, memory:, completed_compensations: [])
  @db.exec_params(
    "INSERT INTO hecks_saga_instances (domain, process_manager, correlation, state, memory, completed_compensations) " \
    "VALUES ($1, $2, $3, $4, $5, $6) " \
    "ON CONFLICT (domain, process_manager, correlation) DO UPDATE " \
    "SET state = EXCLUDED.state, memory = EXCLUDED.memory, " \
    "completed_compensations = EXCLUDED.completed_compensations, updated_at = now()",
    [@domain, process_manager.to_s, correlation.to_s, state.to_s, JSON.generate(memory), JSON.generate(completed_compensations)]
  )
end

#table ⇒ Object 



115
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 115

def table = @aggregate.storage_name