Module: AceMQ::AMQP::Patterns::SQL

Defined in:
lib/acemq/amqp/patterns/sql.rb

Overview

The little that the database-backed stores need from a database.

Ruby has no DBI. sqlite3, pg, mysql2 and everything built on them spell a query differently, bind parameters differently and report how many rows changed differently, so a store written against one of them is a store that works with one of them. This is the seam that avoids that, and it is deliberately three methods wide: anything more and it becomes an object-relational mapper nobody asked for, anything less and the stores cannot tell whether they won a race.

A connection is anything answering:

run(sql, params)              # => Result
placeholder(index)            # "?" or "$1", by driver
constraint_violation?(error)  # true for a duplicate key and nothing else

SQLite3Connection and PGConnection are here because those are the two this library is tested against. Wrapping a pool, a Sequel database or an ActiveRecord connection is a dozen lines of the same shape.

None of this is a runtime dependency. Neither driver is required until something hands one over, so a process that only publishes messages does not install a database driver to do it.

Defined Under Namespace

Modules: Statements Classes: PGConnection, Result, SQLite3Connection

Constant Summary collapse

SAFE_TABLE_NAME =

A table name reaches SQL by concatenation because no database lets one be bound as a parameter. Rejecting anything but a plain identifier is what keeps that safe.

/\A[A-Za-z_][A-Za-z0-9_]{0,62}\z/
TIMESTAMP =

Fixed-width UTC, so that a string comparison in the database orders instants the way instants are ordered.

There is no timestamp type spelled the same way in SQLite and PostgreSQL, and no portable way to bind a Time either. ISO-8601 in UTC is understood by both — PostgreSQL casts it into the column's type, SQLite keeps the text — and because every value is the same width and the same zone, expires_at <= ? means what it says on both.

"%Y-%m-%dT%H:%M:%S.%6NZ"

Class Method Summary collapse

Class Method Details

.at_utc(time) ⇒ String

Returns the instant as fixed-width UTC text.

Parameters:

  • time (Time)

Returns:

  • (String) —

    the instant as fixed-width UTC text



109
 
# File 'lib/acemq/amqp/patterns/sql.rb', line 109

def at_utc(time) = time.utc.strftime(TIMESTAMP)

.bind(template, connection) ⇒ String

Renders a statement's placeholders in the driver's own style.

The templates are written with ? and belong to this library, so counting them positionally is safe in a way it would not be for arbitrary SQL: none of them contains a literal question mark.

Parameters:

  • template (String)
  • connection (Object)

Returns:

  • (String) 


206
207
208
209
210
211
212
 
# File 'lib/acemq/amqp/patterns/sql.rb', line 206

def bind(template, connection)
  index = 0
  template.gsub("?") do
    index += 1
    connection.placeholder(index)
  end
end

.connect(driver) ⇒ Object

Wraps a driver's connection, or passes one through that is already wrapped.

Recognised by the methods the object has rather than by its class, so neither driver has to be loaded for this file to load.

Parameters:

  • driver (Object) —

    a SQLite3::Database, a PG::Connection, or anything already answering run

Returns:

  • (Object) —

    a connection

Raises:

  • (ArgumentError) 


86
87
88
89
90
91
92
93
94
95
 
# File 'lib/acemq/amqp/patterns/sql.rb', line 86

def connect(driver)
  return driver if driver.respond_to?(:run) && driver.respond_to?(:placeholder)
  return PGConnection.new(driver) if driver.respond_to?(:exec_params)
  return SQLite3Connection.new(driver) if driver.respond_to?(:execute)

  raise ArgumentError,
        "#{driver.class} is not a database connection this understands. Hand over " \
        "a SQLite3::Database, a PG::Connection, or an object answering run(sql, " \
        "params), placeholder(index) and constraint_violation?(error)."
end

.key(value) ⇒ String

An identifier as a database will compare it.

Two Ruby strings that are == have to be one key. Under the sqlite3 gem they are not: a String tagged ASCII-8BIT is bound with sqlite3_bind_blob and everything else with sqlite3_bind_text, and SQLite never considers a blob equal to a text value however identical the bytes. typeof() is the only way to see the difference — a blob key prints as the same characters as a text one, so SELECT * shows a row that WHERE message_id = ? then cannot find.

That is not a theoretical encoding: every string bunny hands back off the wire is ASCII-8BIT, message ids included, so a key taken from a delivery is a blob key and a key typed into a console or read from anywhere else is a text one. Without this, the two are different rows in a table whose primary key is supposed to make that impossible.

force_encoding rather than encode: this re-tags and does not transcode, so the bytes reaching the column are the bytes handed in. A key that is not valid UTF-8 is still bound as text and SQLite compares text byte by byte, so it still matches itself. Duplicated only when the tag has to change, because to_s on a String is that same String and force_encoding would otherwise edit the caller's.

PostgreSQL has never had the problem — pg binds by the connection's client encoding rather than by the Ruby tag — and this is harmless there, which is why it lives here rather than in one driver's wrapper.

Parameters:

  • value (Object)

Returns:

  • (String) 


142
143
144
145
146
147
 
# File 'lib/acemq/amqp/patterns/sql.rb', line 142

def key(value)
  text = value.to_s
  return text if text.encoding == Encoding::UTF_8

  text.dup.force_encoding(Encoding::UTF_8)
end

.positional(rows) ⇒ Array<Array>

Rows as this library reads them: an array per row, in the order the statement named its columns.

A driver may hand back either shape and both are legitimate. The sqlite3 gem returns arrays by default and hashes when the database has results_as_hash = true — which ActiveRecord sets on the connection it opens, so a service handing over ActiveRecord::Base.connection.raw_connection gets hashes without ever asking for them. Sequel and anything built on exec_query hand back hashes as their only shape.

Every store here reads row[0], row[1], and on a hash keyed by column name every one of those is nil. Nothing raises: the insert succeeds, the select "works", and an outbox relay reads a batch of records whose ids and bodies are all nil, publishes none of them and reports no error. Silent data loss, and invisible without a spec that sets the flag.

The keys of a hash row are its column names in the order the statement selected them, so the values in insertion order are the positional row. Integer keys are dropped first: older sqlite3 releases put the column index in the same hash alongside the name, and counting those would double every row.

Parameters:

  • rows (Array) —

    whatever a driver handed back

Returns:

  • (Array<Array>) —

    the same rows, positionally



187
188
189
190
191
192
193
194
195
 
# File 'lib/acemq/amqp/patterns/sql.rb', line 187

def positional(rows)
  return rows unless rows.is_a?(Array)

  rows.map do |row|
    next row unless row.is_a?(Hash)

    row.reject { |column, _| column.is_a?(Integer) }.values
  end
end

.table_name!(name) ⇒ Object

Raises:

  • (ArgumentError) —

    when name is not a plain SQL identifier



98
99
100
101
102
103
104
105
 
# File 'lib/acemq/amqp/patterns/sql.rb', line 98

def table_name!(name)
  return name.to_s if SAFE_TABLE_NAME.match?(name.to_s)

  raise ArgumentError,
        "a table must be a plain SQL identifier, was #{name.inspect}. It reaches " \
        "the statement by concatenation, because no database binds one as a " \
        "parameter."
end

.time(value) ⇒ Time

Returns whatever a driver handed back for a timestamp column.

Returns:

  • (Time) —

    whatever a driver handed back for a timestamp column



150
151
152
153
154
155
156
157
158
159
 
# File 'lib/acemq/amqp/patterns/sql.rb', line 150

def time(value)
  return value if value.is_a?(Time)
  return Time.now.utc if value.nil?

  # SQLite hands back the text that was written; PostgreSQL hands back
  # its own rendering of a timestamptz. Time.parse reads both.
  Time.parse(value.to_s)
rescue ArgumentError
  Time.now.utc
end