Class: Ask::State::Providers::SQLite

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

Overview

A persistent key-value store backed by SQLite.

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

Thread-safe via internal Mutex.

Examples:

store = Ask::State::Providers::SQLite.new(path: "sessions.db")
store.set("key", { hello: "world" })
store.get("key")  # => { "hello" => "world" }

Instance Method Summary collapse

Constructor Details

#initialize(path: "sessions.db", **pragmas) ⇒ SQLite

Creates or opens the database at path.



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/ask/state/providers/sqlite.rb', line 25

def initialize(path: "sessions.db", **pragmas)
  require "sqlite3"

  @mutex = Mutex.new
  @db = SQLite3::Database.new(path)
  @db.results_as_hash = true
  @db.busy_timeout = 5000

  defaults = {
    journal_mode: "WAL",
    synchronous: "NORMAL",
    foreign_keys: "ON",
    cache_size: -64_000
  }

  defaults.merge(pragmas).each do |key, value|
    @db.execute("PRAGMA #{key} = #{value}")
  end

  migrate
end

Instance Method Details

#acquire_lock(key, ttl: 10) ⇒ Object

-- distributed locking --



138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/ask/state/providers/sqlite.rb', line 138

def acquire_lock(key, ttl: 10)
  @mutex.synchronize do
    now = Time.now.to_f
    expires_at_time = Time.now + ttl
    token = SecureRandom.hex(16)

    row = @db.get_first_row(
      "SELECT 1 FROM locks WHERE key = ? AND expires_at > ?",
      [key, now]
    )
    return nil if row

    @db.execute("DELETE FROM locks WHERE key = ?", [key])
    @db.execute(<<~SQL, [key, expires_at_time.to_f, token])
      INSERT INTO locks (key, expires_at, token)
      VALUES (?, ?, ?)
    SQL

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

#clear ⇒ Object 



98
99
100
101
102
103
104
105
 
# File 'lib/ask/state/providers/sqlite.rb', line 98

def clear
  @mutex.synchronize do
    @db.execute("DELETE FROM state_store")
    @db.execute("DELETE FROM locks")
    @db.execute("DELETE FROM queues")
    @db.execute("DELETE FROM lists")
  end
end

#close ⇒ Object 



286
287
288
289
290
 
# File 'lib/ask/state/providers/sqlite.rb', line 286

def close
  @mutex.synchronize do
    @db.close
  end
end

#delete(key) ⇒ Object 



68
69
70
71
72
73
74
75
 
# File 'lib/ask/state/providers/sqlite.rb', line 68

def delete(key)
  @mutex.synchronize do
    @db.execute("DELETE FROM state_store WHERE key = ?", [key])
    # delete removes everything under the key, including ordered
    # lists (consumers store event feeds as lists).
    @db.execute("DELETE FROM lists WHERE list_key = ?", [key])
  end
end

#dequeue(queue) ⇒ Object 



183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/ask/state/providers/sqlite.rb', line 183

def dequeue(queue)
  @mutex.synchronize do
    row = @db.get_first_row(<<~SQL, [queue])
      DELETE FROM queues
      WHERE id = (
        SELECT id FROM queues
        WHERE queue_name = ?
        ORDER BY id ASC
        LIMIT 1
      )
      RETURNING id, value, enqueued_at
    SQL
    return nil unless row

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

#enqueue(queue, value) ⇒ Object

-- message queues --



172
173
174
175
176
177
178
179
180
181
 
# File 'lib/ask/state/providers/sqlite.rb', line 172

def enqueue(queue, value)
  @mutex.synchronize do
    @db.execute(<<~SQL, [queue, JSON.generate(value), Time.now.iso8601])
      INSERT INTO queues (queue_name, value, enqueued_at)
      VALUES (?, ?, ?)
    SQL
    id = @db.last_insert_row_id
    QueueEntry.new(id: id.to_s, value: value, enqueued_at: Time.now)
  end
end

#exists?(key) ⇒ Boolean

Returns:

  • (Boolean) 


107
108
109
110
111
112
113
114
115
 
# File 'lib/ask/state/providers/sqlite.rb', line 107

def exists?(key)
  @mutex.synchronize do
    row = @db.get_first_row(<<~SQL, [key, Time.now.to_f])
      SELECT 1 FROM state_store
      WHERE key = ? AND (expires_at IS NULL OR expires_at > ?)
    SQL
    !row.nil?
  end
end

#get(key) ⇒ Object

-- key-value --



49
50
51
52
53
54
55
56
57
 
# File 'lib/ask/state/providers/sqlite.rb', line 49

def get(key)
  @mutex.synchronize do
    row = @db.get_first_row(<<~SQL, [key, Time.now.to_f])
      SELECT value FROM state_store
      WHERE key = ? AND (expires_at IS NULL OR expires_at > ?)
    SQL
    row ? JSON.parse(row["value"]) : nil
  end
end

#keys(pattern: nil) ⇒ Object 



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/ask/state/providers/sqlite.rb', line 117

def keys(pattern: nil)
  @mutex.synchronize do
    now = Time.now.to_f
    sql, params = if pattern
      like = self.class.glob_to_like(pattern)
      [<<~SQL, [like, now]]
        SELECT key FROM state_store
        WHERE key LIKE ? AND (expires_at IS NULL OR expires_at > ?)
      SQL
    else
      [<<~SQL, [now]]
        SELECT key FROM state_store
        WHERE (expires_at IS NULL OR expires_at > ?)
      SQL
    end
    @db.execute(sql, params).map { |r| r["key"] }
  end
end

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

-- ordered lists --



216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
 
# File 'lib/ask/state/providers/sqlite.rb', line 216

def list_append(key, value, max_length: nil)
  @mutex.synchronize do
    serialized = JSON.generate(value)

    @db.execute(<<~SQL, [key, serialized])
      INSERT INTO lists (list_key, value)
      VALUES (?, ?)
    SQL

    return unless max_length

    row = @db.get_first_row(<<~SQL, [key, max_length])
      SELECT MIN(id) AS cutoff FROM (
        SELECT id FROM lists
        WHERE list_key = ?
        ORDER BY id DESC
        LIMIT ?
      )
    SQL
    return unless row && row["cutoff"]

    @db.execute(<<~SQL, [key, row["cutoff"]])
      DELETE FROM lists WHERE list_key = ? AND id < ?
    SQL
  end
end

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



243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/ask/state/providers/sqlite.rb', line 243

def list_range(key, start = 0, stop = -1)
  @mutex.synchronize do
    rows = if stop == -1
      @db.execute(<<~SQL, [key, start])
        SELECT value FROM lists
        WHERE list_key = ?
        ORDER BY id ASC
        LIMIT -1 OFFSET ?
      SQL
    else
      limit = stop - start + 1
      @db.execute(<<~SQL, [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
end

#list_remove(key, value) ⇒ Object 



265
266
267
268
269
270
271
272
273
274
 
# File 'lib/ask/state/providers/sqlite.rb', line 265

def list_remove(key, value)
  @mutex.synchronize do
    serialized = JSON.generate(value)
    @db.execute(
      "DELETE FROM lists WHERE list_key = ? AND value = ?",
      [key, serialized]
    )
    @db.changes
  end
end

#queue_depth(queue) ⇒ Object 



205
206
207
208
209
210
211
212
 
# File 'lib/ask/state/providers/sqlite.rb', line 205

def queue_depth(queue)
  @mutex.synchronize do
    row = @db.get_first_row(
      "SELECT COUNT(*) AS cnt FROM queues WHERE queue_name = ?", [queue]
    )
    row["cnt"]
  end
end

#release_lock(key, lock) ⇒ Object 



160
161
162
163
164
165
166
167
168
 
# File 'lib/ask/state/providers/sqlite.rb', line 160

def release_lock(key, lock)
  @mutex.synchronize do
    @db.execute(
      "DELETE FROM locks WHERE key = ? AND token = ?",
      [key, lock.token]
    )
    @db.changes > 0
  end
end

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



59
60
61
62
63
64
65
66
 
# File 'lib/ask/state/providers/sqlite.rb', line 59

def set(key, value, ttl: nil)
  @mutex.synchronize do
    @db.execute(<<~SQL, [key, JSON.generate(value), ttl ? Time.now.to_f + ttl : nil])
      INSERT OR REPLACE INTO state_store (key, value, expires_at)
      VALUES (?, ?, ?)
    SQL
  end
end

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



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/ask/state/providers/sqlite.rb', line 77

def set_if_not_exists(key, value, ttl: nil)
  @mutex.synchronize do
    now = Time.now.to_f
    expires = ttl ? now + ttl : nil

    row = @db.get_first_row(
      "SELECT 1 FROM state_store WHERE key = ? AND (expires_at IS NULL OR expires_at > ?)",
      [key, now]
    )
    return false if row

    # Key doesn't exist or is expired — delete any leftovers, then insert
    @db.execute("DELETE FROM state_store WHERE key = ?", [key])
    @db.execute(<<~SQL, [key, JSON.generate(value), expires])
      INSERT INTO state_store (key, value, expires_at)
      VALUES (?, ?, ?)
    SQL
    true
  end
end

#setup! ⇒ Object

Idempotent setup — creates tables if they don't exist. Called automatically on initialize. Safe to call multiple times.



280
281
282
283
284
 
# File 'lib/ask/state/providers/sqlite.rb', line 280

def setup!
  @mutex.synchronize do
    migrate
  end
end