Class: Ask::State::Providers::MySQL

Inherits:
Adapter
  • Object
show all
Defined in:
lib/ask/state/providers/mysql.rb

Overview

A persistent key-value store backed by MySQL.

Implements the full Adapter contract: key-value storage, distributed locking, message queues, and ordered lists. Uses the mysql2 gem for database access.

Examples:

store = Ask::State::Providers::MySQL.new(
  url: ENV["MYSQL_URL"]
)
store.set("key", { hello: "world" })
store.get("key")  # => { "hello" => "world" }

Constant Summary collapse

MIGRATIONS = 
<<~SQL
  CREATE TABLE IF NOT EXISTS state_store (
    `key`       VARCHAR(255) PRIMARY KEY NOT NULL,
    `value`     TEXT NOT NULL,
    expires_at  DATETIME(3) NULL
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

  CREATE TABLE IF NOT EXISTS locks (
    `key`       VARCHAR(255) PRIMARY KEY NOT NULL,
    token       VARCHAR(64) NOT NULL,
    expires_at  DATETIME(3) NOT NULL
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

  CREATE TABLE IF NOT EXISTS queues (
    id           BIGINT AUTO_INCREMENT PRIMARY KEY,
    queue_name   VARCHAR(255) NOT NULL,
    `value`      TEXT NOT NULL,
    enqueued_at  DATETIME(3) NOT NULL,
    INDEX idx_queues_queue_name (queue_name, id)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

  CREATE TABLE IF NOT EXISTS lists (
    id         BIGINT AUTO_INCREMENT PRIMARY KEY,
    list_key   VARCHAR(255) NOT NULL,
    `value`    TEXT NOT NULL,
    INDEX idx_lists_list_key (list_key, id)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
SQL

Instance Method Summary collapse

Constructor Details

#initialize(url: ENV.fetch("MYSQL_URL", "mysql2://root@localhost:3306/ask_state")) ⇒ MySQL

Creates a new MySQL-backed state store.

Parameters:

  • url (String) (defaults to: ENV.fetch("MYSQL_URL", "mysql2://root@localhost:3306/ask_state")) —

    MySQL connection URL



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/ask/state/providers/mysql.rb', line 56

def initialize(url: ENV.fetch("MYSQL_URL", "mysql2://root@localhost:3306/ask_state"))
  require "mysql2"

  uri = URI.parse(url)
  @client = Mysql2::Client.new(
    host: uri.host || "localhost",
    port: uri.port || 3306,
    username: uri.user || "root",
    password: uri.password,
    database: uri.path.sub("/", ""),
    charset: "utf8mb4"
  )

  migrate
end

Instance Method Details

#acquire_lock(key, ttl: 10) ⇒ Object

-- distributed locking --



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/ask/state/providers/mysql.rb', line 151

def acquire_lock(key, ttl: 10)
  now = Time.now.utc
  expires_at_time = now + ttl
  token = SecureRandom.hex(16)
  now_f = now.strftime("%Y-%m-%d %H:%M:%S.%3N")

  # Clean up expired lock
  @client.prepare("DELETE FROM locks WHERE `key` = ? AND expires_at <= ?").execute(key, now_f)

  result = @client.prepare(<<~SQL).execute(key, expires_at_time.strftime("%Y-%m-%d %H:%M:%S.%3N"), token)
    INSERT INTO locks (`key`, expires_at, token)
    SELECT ?, ?, ?
    WHERE NOT EXISTS (
      SELECT 1 FROM locks WHERE `key` = ?
    )
  SQL
  result = @client.prepare(<<~SQL).execute(key)
    SELECT COUNT(*) AS cnt FROM locks WHERE `key` = ?
  SQL

  # Check if we got the lock
  row = result.first
  return nil unless row && row["cnt"].to_i > 0

  Lock.new(id: key, token: token, expires_at: expires_at_time)
end

#clear ⇒ Object 



115
116
117
118
119
120
 
# File 'lib/ask/state/providers/mysql.rb', line 115

def clear
  @client.query("DELETE FROM state_store")
  @client.query("DELETE FROM locks")
  @client.query("DELETE FROM queues")
  @client.query("DELETE FROM lists")
end

#close ⇒ Object

-- lifecycle --



268
269
270
 
# File 'lib/ask/state/providers/mysql.rb', line 268

def close
  @client&.close
end

#delete(key) ⇒ Object 



93
94
95
 
# File 'lib/ask/state/providers/mysql.rb', line 93

def delete(key)
  @client.prepare("DELETE FROM state_store WHERE `key` = ?").execute(key)
end

#dequeue(queue) ⇒ Object 



196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/ask/state/providers/mysql.rb', line 196

def dequeue(queue)
  row = @client.prepare(<<~SQL).execute(queue).first
    SELECT id, `value`, enqueued_at FROM queues
    WHERE queue_name = ?
    ORDER BY id ASC
    LIMIT 1
  SQL
  return nil unless row

  @client.prepare("DELETE FROM queues WHERE id = ?").execute(row["id"])

  QueueEntry.new(
    id: row["id"].to_s,
    value: JSON.parse(row["value"]),
    enqueued_at: Time.parse(row["enqueued_at"])
  )
end

#enqueue(queue, value) ⇒ Object

-- message queues --



185
186
187
188
189
190
191
192
193
194
 
# File 'lib/ask/state/providers/mysql.rb', line 185

def enqueue(queue, value)
  stmt = @client.prepare(<<~SQL)
    INSERT INTO queues (queue_name, `value`, enqueued_at)
    VALUES (?, ?, ?)
  SQL
  stmt.execute(queue, JSON.generate(value), Time.now.utc.strftime("%Y-%m-%d %H:%M:%S.%3N"))
  id = @client.last_id

  QueueEntry.new(id: id.to_s, value: value, enqueued_at: Time.now)
end

#exists?(key) ⇒ Boolean

Returns:

  • (Boolean) 


122
123
124
125
126
127
128
 
# File 'lib/ask/state/providers/mysql.rb', line 122

def exists?(key)
  result = @client.prepare(<<~SQL).execute(key, Time.now.utc.strftime("%Y-%m-%d %H:%M:%S.%3N"))
    SELECT 1 FROM state_store
    WHERE `key` = ? AND (expires_at IS NULL OR expires_at > ?)
  SQL
  result.count > 0
end

#get(key) ⇒ Object

-- key-value --



74
75
76
77
78
79
80
 
# File 'lib/ask/state/providers/mysql.rb', line 74

def get(key)
  row = @client.prepare(<<~SQL).execute(key, Time.now.utc.strftime("%Y-%m-%d %H:%M:%S.%3N")).first
    SELECT `value` FROM state_store
    WHERE `key` = ? AND (expires_at IS NULL OR expires_at > ?)
  SQL
  row ? JSON.parse(row["value"]) : nil
end

#keys(pattern: nil) ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/ask/state/providers/mysql.rb', line 130

def keys(pattern: nil)
  sql, params = if pattern
    like = self.class.glob_to_like(pattern)
    [<<~SQL, [like]]
      SELECT `key` FROM state_store
      WHERE `key` LIKE ? AND (expires_at IS NULL OR expires_at > ?)
    SQL
  else
    [<<~SQL, []]
      SELECT `key` FROM state_store
      WHERE (expires_at IS NULL OR expires_at > ?)
    SQL
  end
  now = Time.now.utc.strftime("%Y-%m-%d %H:%M:%S.%3N")
  stmt = @client.prepare(sql)
  rows = stmt.execute(*params, now)
  rows.map { |r| r["key"] }
end

#list_append(key, value, max_length: nil) ⇒ Object

-- ordered lists --



222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'lib/ask/state/providers/mysql.rb', line 222

def list_append(key, value, max_length: nil)
  stmt = @client.prepare("INSERT INTO lists (list_key, `value`) VALUES (?, ?)")
  stmt.execute(key, JSON.generate(value))

  return unless max_length

  @client.prepare(<<~SQL).execute(key, key, max_length)
    DELETE FROM lists WHERE id <= (
      SELECT COALESCE(MIN(id), 0) FROM (
        SELECT id FROM lists
        WHERE list_key = ?
        ORDER BY id DESC
        LIMIT ?
      ) sub
    )
  SQL
end

#list_range(key, start = 0, stop = -1)) ⇒ Object 



240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'lib/ask/state/providers/mysql.rb', line 240

def list_range(key, start = 0, stop = -1)
  rows = if stop == -1
    @client.prepare(<<~SQL).execute(key, start)
      SELECT `value` FROM lists
      WHERE list_key = ?
      ORDER BY id ASC
      LIMIT 18446744073709551615 OFFSET ?
    SQL
  else
    limit = stop - start + 1
    @client.prepare(<<~SQL).execute(key, limit, start)
      SELECT `value` FROM lists
      WHERE list_key = ?
      ORDER BY id ASC
      LIMIT ? OFFSET ?
    SQL
  end
  rows.map { |r| JSON.parse(r["value"]) }
end

#list_remove(key, value) ⇒ Object 



260
261
262
263
264
 
# File 'lib/ask/state/providers/mysql.rb', line 260

def list_remove(key, value)
  serialized = JSON.generate(value)
  @client.prepare("DELETE FROM lists WHERE list_key = ? AND `value` = ?").execute(key, serialized)
  @client.affected_rows
end

#queue_depth(queue) ⇒ Object 



214
215
216
217
218
 
# File 'lib/ask/state/providers/mysql.rb', line 214

def queue_depth(queue)
  result = @client.prepare("SELECT COUNT(*) AS cnt FROM queues WHERE queue_name = ?").execute(queue)
  row = result.first
  row["cnt"]
end

#release_lock(key, lock) ⇒ Object 



178
179
180
181
 
# File 'lib/ask/state/providers/mysql.rb', line 178

def release_lock(key, lock)
  @client.prepare("DELETE FROM locks WHERE `key` = ? AND token = ?").execute(key, lock.token)
  @client.affected_rows > 0
end

#set(key, value, ttl: nil) ⇒ Object 



82
83
84
85
86
87
88
89
90
91
 
# File 'lib/ask/state/providers/mysql.rb', line 82

def set(key, value, ttl: nil)
  stmt = @client.prepare(<<~SQL)
    INSERT INTO state_store (`key`, `value`, expires_at)
    VALUES (?, ?, ?)
    ON DUPLICATE KEY UPDATE
      `value` = VALUES(`value`),
      expires_at = VALUES(expires_at)
  SQL
  stmt.execute(key, JSON.generate(value), ttl ? (Time.now.utc + ttl).strftime("%Y-%m-%d %H:%M:%S.%3N") : nil)
end

#set_if_not_exists(key, value, ttl: nil) ⇒ Object 



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/ask/state/providers/mysql.rb', line 97

def set_if_not_exists(key, value, ttl: nil)
  now = Time.now.utc
  expires = ttl ? (now + ttl).strftime("%Y-%m-%d %H:%M:%S.%3N") : nil
  now_f = now.strftime("%Y-%m-%d %H:%M:%S.%3N")

  stmt = @client.prepare(<<~SQL)
    INSERT INTO state_store (`key`, `value`, expires_at)
    SELECT ?, ?, ?
    WHERE NOT EXISTS (
      SELECT 1 FROM state_store
      WHERE `key` = ? AND (expires_at IS NULL OR expires_at > ?)
    )
  SQL
  # MySQL's INSERT...SELECT won't fail on DUPLICATE KEY if WHERE is false
  stmt.execute(key, JSON.generate(value), expires, key, now_f)
  @client.affected_rows > 0
end