Class: Ask::State::Providers::Postgres

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

Overview

A persistent key-value store backed by PostgreSQL.

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

Examples:

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

Constant Summary collapse

MIGRATIONS = 
<<~SQL
  CREATE TABLE IF NOT EXISTS state_store (
    key         TEXT PRIMARY KEY NOT NULL,
    value       TEXT NOT NULL,
    expires_at  TIMESTAMPTZ
  );

  CREATE TABLE IF NOT EXISTS locks (
    key         TEXT PRIMARY KEY NOT NULL,
    token       TEXT NOT NULL,
    expires_at  TIMESTAMPTZ NOT NULL
  );

  CREATE TABLE IF NOT EXISTS queues (
    id           BIGSERIAL PRIMARY KEY,
    queue_name   TEXT NOT NULL,
    value        TEXT NOT NULL,
    enqueued_at  TIMESTAMPTZ NOT NULL
  );
  CREATE INDEX IF NOT EXISTS idx_queues_queue_name
    ON queues (queue_name, id);

  CREATE TABLE IF NOT EXISTS lists (
    id         BIGSERIAL PRIMARY KEY,
    list_key   TEXT NOT NULL,
    value      TEXT NOT NULL
  );
  CREATE INDEX IF NOT EXISTS idx_lists_list_key
    ON lists (list_key, id);
SQL

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(url: ENV.fetch("DATABASE_URL", "postgres://localhost:5432/ask_state"), pool_size: 5) ⇒ Postgres

Creates a new PostgreSQL-backed state store.

Parameters:

  • url (String) (defaults to: ENV.fetch("DATABASE_URL", "postgres://localhost:5432/ask_state")) —

    Postgres connection URL

  • pool_size (Integer) (defaults to: 5) —

    connection pool size (default 5)



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/ask/state/providers/postgres.rb', line 61

def initialize(url: ENV.fetch("DATABASE_URL", "postgres://localhost:5432/ask_state"),
               pool_size: 5)
  # On macOS, libpq's default GSSAPI negotiation can segfault
  # when a forked process (SolidQueue worker) opens a fresh
  # connection — the same crash the sibling apps guard with
  # `gssencmode: disable` in database.yml. The store connects
  # directly (not through ActiveRecord), so the guard lives here.
  url = self.class.darwin_safe_url(url) if RUBY_PLATFORM =~ /darwin/

  @pool = ConnectionPool.new(size: pool_size) do
    ::PG.connect(url)
  end

  migrate
end

Class Method Details

.darwin_safe_url(url) ⇒ Object

Append gssencmode=disable to a URL on macOS (idempotent).



78
79
80
81
82
83
 
# File 'lib/ask/state/providers/postgres.rb', line 78

def self.darwin_safe_url(url)
  return url if url.include?("gssencmode")

  separator = url.include?("?") ? "&" : "?"
  "#{url}#{separator}gssencmode=disable"
end

Instance Method Details

#acquire_lock(key, ttl: 10) ⇒ Object

-- distributed locking --



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

def acquire_lock(key, ttl: 10)
  now = Time.now.utc
  expires_at_time = now + ttl
  token = SecureRandom.hex(16)

  acquired = with_connection do |conn|
    # First clean up expired locks
    conn.exec_params("DELETE FROM locks WHERE key = $1 AND expires_at <= $2",
                   [key, now])

    result = conn.exec_params(<<~SQL, [key, expires_at_time, token])
      INSERT INTO locks (key, expires_at, token)
      SELECT $1::text, $2::timestamptz, $3::text
      WHERE NOT EXISTS (
        SELECT 1 FROM locks
        WHERE key = $1
      )
    SQL
    result.cmd_tuples > 0
  end

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

#clear ⇒ Object 



145
146
147
148
149
150
151
152
 
# File 'lib/ask/state/providers/postgres.rb', line 145

def clear
  with_connection do |conn|
    conn.exec("DELETE FROM state_store")
    conn.exec("DELETE FROM locks")
    conn.exec("DELETE FROM queues")
    conn.exec("DELETE FROM lists")
  end
end

#close ⇒ Object 



332
333
334
 
# File 'lib/ask/state/providers/postgres.rb', line 332

def close
  @pool&.shutdown { |c| c.close unless c.finished? }
end

#delete(key) ⇒ Object 



111
112
113
114
115
116
117
118
 
# File 'lib/ask/state/providers/postgres.rb', line 111

def delete(key)
  with_connection do |conn|
    conn.exec_params("DELETE FROM state_store WHERE key = $1", [key])
    # delete removes everything under the key, including ordered
    # lists (consumers store event feeds as lists).
    conn.exec_params("DELETE FROM lists WHERE list_key = $1", [key])
  end
end

#dequeue(queue) ⇒ Object 



233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
 
# File 'lib/ask/state/providers/postgres.rb', line 233

def dequeue(queue)
  with_connection do |conn|
    result = conn.exec_params(<<~SQL, [queue])
      DELETE FROM queues
      WHERE id = (
        SELECT id FROM queues
        WHERE queue_name = $1
        ORDER BY id ASC
        LIMIT 1
      )
      RETURNING value, enqueued_at
    SQL
    return nil if result.ntuples == 0

    QueueEntry.new(
      id: SecureRandom.uuid,
      value: JSON.parse(result[0]["value"]),
      enqueued_at: Time.parse(result[0]["enqueued_at"])
    )
  end
end

#enqueue(queue, value) ⇒ Object

-- message queues --



220
221
222
223
224
225
226
227
228
229
230
231
 
# File 'lib/ask/state/providers/postgres.rb', line 220

def enqueue(queue, value)
  id = SecureRandom.uuid

  with_connection do |conn|
    conn.exec_params(<<~SQL, [queue, JSON.generate(value), Time.now.utc])
      INSERT INTO queues (queue_name, value, enqueued_at)
      VALUES ($1, $2, $3)
    SQL
  end

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

#exists?(key) ⇒ Boolean

Returns:

  • (Boolean) 


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

def exists?(key)
  with_connection do |conn|
    result = conn.exec_params(<<~SQL, [key, Time.now.utc])
      SELECT 1 FROM state_store
      WHERE key = $1 AND (expires_at IS NULL OR expires_at > $2)
    SQL
    result.ntuples > 0
  end
end

#get(key) ⇒ Object

-- key-value --



87
88
89
90
91
92
93
94
95
 
# File 'lib/ask/state/providers/postgres.rb', line 87

def get(key)
  with_connection do |conn|
    row = conn.exec_params(<<~SQL, [key, Time.now.utc])
      SELECT value FROM state_store
      WHERE key = $1 AND (expires_at IS NULL OR expires_at > $2)
    SQL
    row.ntuples > 0 ? JSON.parse(row[0]["value"]) : nil
  end
end

#keys(pattern: nil) ⇒ Object 



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/ask/state/providers/postgres.rb', line 164

def keys(pattern: nil)
  with_connection do |conn|
    sql, params = if pattern
      like = self.class.glob_to_like(pattern)
      [<<~SQL, [like, Time.now.utc]]
        SELECT key FROM state_store
        WHERE key LIKE $1 AND (expires_at IS NULL OR expires_at > $2)
      SQL
    else
      [<<~SQL, [Time.now.utc]]
        SELECT key FROM state_store
        WHERE (expires_at IS NULL OR expires_at > $1)
      SQL
    end
    conn.exec_params(sql, params).map { |r| r["key"] }
  end
end

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

-- ordered lists --



266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
 
# File 'lib/ask/state/providers/postgres.rb', line 266

def list_append(key, value, max_length: nil)
  with_connection do |conn|
    conn.exec_params(<<~SQL, [key, JSON.generate(value)])
      INSERT INTO lists (list_key, value)
      VALUES ($1, $2)
    SQL

    return unless max_length

    # Keep the newest max_length entries: the cutoff is the
    # oldest id inside the newest max_length — delete strictly
    # older rows (and only this list's).
    conn.exec_params(<<~SQL, [key, max_length, key])
      DELETE FROM lists WHERE list_key = $3 AND id < (
        SELECT COALESCE(MIN(id), 0) FROM (
          SELECT id FROM lists
          WHERE list_key = $1
          ORDER BY id DESC
          LIMIT $2::bigint
        ) sub
      )
    SQL
  end
end

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



291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
 
# File 'lib/ask/state/providers/postgres.rb', line 291

def list_range(key, start = 0, stop = -1)
  with_connection do |conn|
    rows = if stop == -1
      conn.exec_params(<<~SQL, [key, start])
        SELECT value FROM lists
        WHERE list_key = $1
        ORDER BY id ASC
        OFFSET $2
      SQL
    else
      limit = stop - start + 1
      conn.exec_params(<<~SQL, [key, limit, start])
        SELECT value FROM lists
        WHERE list_key = $1
        ORDER BY id ASC
        LIMIT $2 OFFSET $3
      SQL
    end
    rows.map { |r| JSON.parse(r["value"]) }
  end
end

#list_remove(key, value) ⇒ Object 



313
314
315
316
317
318
319
320
321
322
 
# File 'lib/ask/state/providers/postgres.rb', line 313

def list_remove(key, value)
  serialized = JSON.generate(value)
  with_connection do |conn|
    result = conn.exec_params(
      "DELETE FROM lists WHERE list_key = $1 AND value = $2",
      [key, serialized]
    )
    result.cmd_tuples
  end
end

#queue_depth(queue) ⇒ Object 



255
256
257
258
259
260
261
262
 
# File 'lib/ask/state/providers/postgres.rb', line 255

def queue_depth(queue)
  with_connection do |conn|
    result = conn.exec_params(
      "SELECT COUNT(*) AS cnt FROM queues WHERE queue_name = $1", [queue]
    )
    result[0]["cnt"].to_i
  end
end

#release_lock(key, lock) ⇒ Object 



208
209
210
211
212
213
214
215
216
 
# File 'lib/ask/state/providers/postgres.rb', line 208

def release_lock(key, lock)
  with_connection do |conn|
    result = conn.exec_params(
      "DELETE FROM locks WHERE key = $1 AND token = $2",
      [key, lock.token]
    )
    result.cmd_tuples > 0
  end
end

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



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

def set(key, value, ttl: nil)
  with_connection do |conn|
    # Explicit casts: $3 is nil when no ttl, and PostgreSQL can't
    # infer a type for a bare NULL parameter.
    conn.exec_params(<<~SQL, [key, JSON.generate(value), ttl ? Time.now.utc + ttl : nil])
      INSERT INTO state_store (key, value, expires_at)
      VALUES ($1::text, $2::text, $3::timestamptz)
      ON CONFLICT (key) DO UPDATE SET
        value = EXCLUDED.value,
        expires_at = EXCLUDED.expires_at
    SQL
  end
end

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



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/ask/state/providers/postgres.rb', line 120

def set_if_not_exists(key, value, ttl: nil)
  with_connection do |conn|
    now = Time.now.utc
    expires = ttl ? now + ttl : nil

    # Insert when the key is missing or its row is expired. An
    # expired row still conflicts on the primary key, so the
    # ON CONFLICT branch overwrites it — but only when it is
    # actually expired, never a live value.
    result = conn.exec_params(<<~SQL, [key, JSON.generate(value), expires, now])
      INSERT INTO state_store (key, value, expires_at)
      SELECT $1::text, $2::text, $3::timestamptz
      WHERE NOT EXISTS (
        SELECT 1 FROM state_store
        WHERE key = $1 AND (expires_at IS NULL OR expires_at > $4)
      )
      ON CONFLICT (key) DO UPDATE SET
        value = EXCLUDED.value,
        expires_at = EXCLUDED.expires_at
      WHERE state_store.expires_at IS NOT NULL AND state_store.expires_at <= $4
    SQL
    result.cmd_tuples > 0
  end
end

#setup! ⇒ Object

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



328
329
330
 
# File 'lib/ask/state/providers/postgres.rb', line 328

def setup!
  with_connection { |conn| conn.exec(MIGRATIONS) }
end