Class: Hecks::Adapters::Sqlite

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

Overview

The SQLite store: one table per aggregate head, an append-only entry table beside it. The DDL lives in sqlite/schema_builder.rb, the column codec in sqlite/codec.rb, and the query compilation is the shared SqlQueryBuilder — this file supplies only SQLite's dialect.

Direct Known Subclasses

SqlitePersistence, SqliteProjection

Defined Under Namespace

Modules: Codec, SchemaBuilder

Constant Summary collapse

SQL_TYPES = 
{ "Integer" => "INTEGER", "Float" => "REAL" }.freeze

Constants included from SqlQueryBuilder

Hecks::Adapters::SqlQueryBuilder::COMPARATORS

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from SqlQueryBuilder

#query

Constructor Details

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

Returns a new instance of Sqlite.



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 31

def initialize(aggregate:, settings: {}, root: nil)
  # LAZY, ON PURPOSE — a domain that never wires Sqlite should never
  # need the gem installed. `require "hecks"` alone must not
  # force a database client library nobody asked for.
  require "sqlite3"

  @aggregate = aggregate
  @path      = resolve_path(settings, root)
  # THE OPTIONAL saga-persistence capability's own scoping column
  # (§2/§4) — falls back to the aggregate's own name for a
  # directly-instantiated adapter (specs), same fallback shape
  # Postgres's own @domain already uses.
  @domain    = (
    if settings.key?(:domain)
      settings[:domain]
    elsif settings.key?("domain")
      settings["domain"]
    else
      aggregate.name
    end
  ).to_s

  FileUtils.mkdir_p(File.dirname(@path))
  @db = SQLite3::Database.new(@path)
  @db.results_as_hash = true
  # The append is the recovery commit point. Keep SQLite's fsync policy
  # explicit instead of inheriting a process-wide pragma choice.
  @db.execute("PRAGMA synchronous = FULL")

  create_aggregate_table!
  create_entry_table!
  ensure_entry_operation_column!
  ensure_entry_mirrors_column!
  create_event_table!
  create_saga_table!
end

Instance Attribute Details

#aggregate ⇒ Object (readonly)

Returns the value of attribute aggregate.



27
28
29
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 27

def aggregate
  @aggregate
end

#path ⇒ Object (readonly)

Returns the value of attribute path.



27
28
29
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 27

def path
  @path
end

Instance Method Details

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

order_by IS A RUNTIME VALUE — see postgres.rb's own all for the full reasoning; whitelisted the identical way before it ever reaches order_expression.



80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 80

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.execute("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 



97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 97

def append(entry)
  @db.execute(
    "INSERT INTO #{quoted_entry_table} (aggregate_id, operation, state, mirrors) VALUES (?, ?, ?, ?)",
    # `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 `postgres_era.rb`
    # already uses for its 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

The outcome lookup, journal append and snapshot replacement share one SQLite transaction. The runtime performs no preliminary find; this adapter-native operation owns both concurrency and outcome reporting.



152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 152

def atomic_put(entry, insert_only: false)
  status = nil
  @db.transaction do
    exists = !@db.get_first_value("SELECT 1 FROM #{quoted_table} WHERE id = ?", [entry.id.to_s]).nil?
    if insert_only && exists
      status = :conflicted
      next
    end
    status = exists ? :replaced : :inserted
    append(entry)
    project(entry)
  end
  status
end

#count ⇒ Object 



95
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 95

def count = @db.get_first_value("SELECT COUNT(*) FROM #{quoted_table}").to_i

#delete(id) ⇒ Object 



167
168
169
170
171
172
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 167

def delete(id)
  entry = Ports::Persistence::Entry.new(operation: "delete", id: id.to_s, state: nil)
  append(entry)
  project(entry)
  true
end

#delete_saga(process_manager:, correlation:) ⇒ Object 



213
214
215
216
217
218
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 213

def delete_saga(process_manager:, correlation:)
  @db.execute(
    "DELETE FROM hecks_saga_instances WHERE domain = ? AND process_manager = ? AND correlation = ?",
    [@domain, process_manager.to_s, correlation.to_s]
  )
end

#each_saga ⇒ Object 



220
221
222
223
224
225
226
227
228
229
230
231
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 220

def each_saga
  return enum_for(:each_saga) unless block_given?

  @db.execute(
    "SELECT process_manager, correlation, state, memory, completed_compensations FROM hecks_saga_instances WHERE domain = ?",
    [@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 



125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 125

def entries
  @db.execute("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 



181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 181

def events
  @db.execute("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 



70
71
72
73
74
75
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 70

def find(id)
  row = @db.get_first_row("SELECT * FROM #{quoted_table} WHERE id = ?", [id.to_s])
  return nil unless row

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

#persistence_capabilities ⇒ Object 



29
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 29

def persistence_capabilities = [:atomic_put]

#project(entry) ⇒ Object 



110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 110

def project(entry)
  return @db.execute("DELETE FROM #{quoted_table} WHERE id = ?", [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 }).map { |c| quote_ident(c) }
  values  = [instance.id.to_s] + persisted_fields.map { |field| encode_field(field, instance[field[:name]]) }
  slots   = Array.new(columns.size, "?").join(", ")

  @db.execute(
    "INSERT OR REPLACE INTO #{quoted_table} (#{columns.join(', ')}) VALUES (#{slots})",
    values
  )
  instance
end

#record_event(event) ⇒ Object 



174
175
176
177
178
179
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 174

def record_event(event)
  @db.execute(
    "INSERT INTO events (name, aggregate, aggregate_id, payload, occurred_at) VALUES (?, ?, ?, ?, ?)",
    [event.name, event.aggregate, event.id.to_s, JSON.generate(event.payload), event.occurred_at]
  )
end

#reset! ⇒ Object 



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

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

#save(instance) ⇒ Object 



143
144
145
146
147
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 143

def save(instance)
  entry = Ports::Persistence::Entry.new(operation: "save", id: instance.id.to_s, state: instance.state.dup)
  append(entry)
  project(entry)
end

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

── the OPTIONAL saga-persistence capability (§2) — reuses the DDL every SQLite-backed aggregate table already lives beside (create_saga_table!, Sqlite::SchemaBuilder, shared with D1). SQLite's resolve_path defaults to one .db file PER AGGREGATE unless a domain shares one database setting across its aggregates — since saga persistence resolves through whichever adapter instance backs the domain's FIRST aggregate (Registry#saga_persistence), this table ends up living inside THAT one aggregate's own file by default. Correct and durable either way; a domain that wants an obviously-named saga store already gets one by sharing database across its aggregates, the recommended, common case.



205
206
207
208
209
210
211
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 205

def save_saga(process_manager:, correlation:, state:, memory:, completed_compensations: [])
  @db.execute(
    "INSERT OR REPLACE INTO hecks_saga_instances (domain, process_manager, correlation, state, memory, completed_compensations) " \
    "VALUES (?, ?, ?, ?, ?, ?)",
    [@domain, process_manager.to_s, correlation.to_s, state.to_s, JSON.generate(memory), JSON.generate(completed_compensations)]
  )
end

#table ⇒ Object 



68
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 68

def table = @aggregate.storage_name