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
 
# 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 = (settings[:domain] || settings["domain"] || aggregate.storage_name).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[: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[: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.



119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 119

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 do |row|
    Runtime::Instance.new(aggregate: @aggregate, id: row["id"], state: decode(row))
  end
end

#append(entry) ⇒ Object 



136
137
138
139
140
141
142
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 136

def append(entry)
  @db.exec_params(
    "INSERT INTO #{quoted_entry_table} (aggregate_id, operation, state, mirrors) VALUES ($1, $2, $3, $4)",
    [entry.id, entry.operation, JSON.generate(entry.state), 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.



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 197

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 



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

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

#delete(id) ⇒ Object 



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

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 



258
259
260
261
262
263
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 258

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 



265
266
267
268
269
270
271
272
273
274
275
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 265

def each_saga
  return enum_for(:each_saga) unless block_given?

  @db.exec_params(
    "SELECT process_manager, correlation, state, memory 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)
  end
end

#entries ⇒ Object 



161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 161

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 



233
234
235
236
237
238
239
240
241
242
243
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 233

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 



109
110
111
112
113
114
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 109

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

  Runtime::Instance.new(aggregate: @aggregate, id: result[0]["id"], state: decode(result[0]))
end

#persistence_capabilities ⇒ Object 



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

def persistence_capabilities = [:atomic_put]

#project(entry) ⇒ Object 



144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 144

def project(entry)
  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 })
  values   = [instance.id.to_s] + persisted_fields.map { |field| encode_field(field, instance[field[:name]]) }
  updates  = persisted_fields.map { |field| "#{quote_ident(field[:name])} = EXCLUDED.#{quote_ident(field[:name])}" }

  @db.exec_params(
    "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(', ')}",
    values
  )
  instance
end

#record_event(event) ⇒ Object 



226
227
228
229
230
231
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 226

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 



173
174
175
176
177
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 173

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.



186
187
188
189
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 186

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:) ⇒ Object

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



248
249
250
251
252
253
254
255
256
 
# File 'lib/hecks/adapters/driven/postgres.rb', line 248

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

#table ⇒ Object 



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

def table = @aggregate.storage_name