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
 
# File 'lib/ask/state/providers/sqlite.rb', line 25

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

  @path = path
  @pid = Process.pid
  @mutex = Mutex.new
  @pragmas = {
    journal_mode: "WAL",
    synchronous: "NORMAL",
    foreign_keys: "ON",
    cache_size: -64_000
  }.merge(pragmas)

  connect
  migrate
end

Instance Method Details

#acquire_lock(key, ttl: 10) ⇒ Object

-- distributed locking --



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/ask/state/providers/sqlite.rb', line 133

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 



93
94
95
96
97
98
99
100
 
# File 'lib/ask/state/providers/sqlite.rb', line 93

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 



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

def close
  @mutex.synchronize do
    @db&.close unless @db&.closed?
  end
end

#delete(key) ⇒ Object 



63
64
65
66
67
68
69
70
 
# File 'lib/ask/state/providers/sqlite.rb', line 63

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 



178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/ask/state/providers/sqlite.rb', line 178

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 --



167
168
169
170
171
172
173
174
175
176
 
# File 'lib/ask/state/providers/sqlite.rb', line 167

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) 


102
103
104
105
106
107
108
109
110
 
# File 'lib/ask/state/providers/sqlite.rb', line 102

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 --



44
45
46
47
48
49
50
51
52
 
# File 'lib/ask/state/providers/sqlite.rb', line 44

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 



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/ask/state/providers/sqlite.rb', line 112

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 --



211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/ask/state/providers/sqlite.rb', line 211

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 



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

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 



260
261
262
263
264
265
266
267
268
269
 
# File 'lib/ask/state/providers/sqlite.rb', line 260

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 



200
201
202
203
204
205
206
207
 
# File 'lib/ask/state/providers/sqlite.rb', line 200

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 



155
156
157
158
159
160
161
162
163
 
# File 'lib/ask/state/providers/sqlite.rb', line 155

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 



54
55
56
57
58
59
60
61
 
# File 'lib/ask/state/providers/sqlite.rb', line 54

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 



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/ask/state/providers/sqlite.rb', line 72

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.



275
276
277
278
279
 
# File 'lib/ask/state/providers/sqlite.rb', line 275

def setup!
  @mutex.synchronize do
    migrate
  end
end