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



72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 72

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 



89
90
91
92
93
94
95
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 89

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



139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 139

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 



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

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

#delete(id) ⇒ Object 



154
155
156
157
158
159
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 154

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 



200
201
202
203
204
205
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 200

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 



207
208
209
210
211
212
213
214
215
216
217
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 207

def each_saga
  return enum_for(:each_saga) unless block_given?

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

#entries ⇒ Object 



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

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 



168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 168

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 



62
63
64
65
66
67
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 62

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 



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

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 



161
162
163
164
165
166
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 161

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 



124
125
126
127
128
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 124

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

#save(instance) ⇒ Object 



130
131
132
133
134
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 130

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



192
193
194
195
196
197
198
 
# File 'lib/hecks/adapters/driven/sqlite.rb', line 192

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

#table ⇒ Object 



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

def table = @aggregate.storage_name