Class: ChatSDK::State::Pg

Inherits:
Base
  • Object
show all
Defined in:
lib/chat_sdk/state/pg.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(url: nil, connection: nil, key_prefix: "chat_sdk", auto_migrate: true) ⇒ Pg

Returns a new instance of Pg.



10
11
12
13
14
 
# File 'lib/chat_sdk/state/pg.rb', line 10

def initialize(url: nil, connection: nil, key_prefix: "chat_sdk", auto_migrate: true)
  @conn = connection || PG.connect(url || ENV["DATABASE_URL"] || ENV["POSTGRES_URL"] || "postgresql://localhost/chat_sdk_dev")
  @key_prefix = key_prefix
  ensure_tables if auto_migrate
end

Instance Attribute Details

#conn ⇒ Object (readonly)

Returns the value of attribute conn.



16
17
18
 
# File 'lib/chat_sdk/state/pg.rb', line 16

def conn
  @conn
end

Instance Method Details

#acquire_lock(key, owner:, ttl:) ⇒ Object

Locks



44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/chat_sdk/state/pg.rb', line 44

def acquire_lock(key, owner:, ttl:)
  ttl_seconds = ttl.to_f
  @conn.exec_params(
    "DELETE FROM chat_sdk_locks WHERE key_prefix = $1 AND lock_key = $2 AND expires_at < NOW()",
    [@key_prefix, key]
  )
  result = @conn.exec_params(
    "INSERT INTO chat_sdk_locks (key_prefix, lock_key, owner, expires_at) " \
    "VALUES ($1, $2, $3, NOW() + INTERVAL '#{ttl_seconds} seconds') ON CONFLICT DO NOTHING",
    [@key_prefix, key, owner]
  )
  result.cmd_tuples > 0
end

#cleanup_expired ⇒ Object 



135
136
137
138
139
140
141
142
143
144
 
# File 'lib/chat_sdk/state/pg.rb', line 135

def cleanup_expired
  @conn.exec_params(
    "DELETE FROM chat_sdk_cache WHERE key_prefix = $1 AND expires_at IS NOT NULL AND expires_at < NOW()",
    [@key_prefix]
  )
  @conn.exec_params(
    "DELETE FROM chat_sdk_locks WHERE key_prefix = $1 AND expires_at < NOW()",
    [@key_prefix]
  )
end

#clear ⇒ Object

Cleanup



129
130
131
132
133
 
# File 'lib/chat_sdk/state/pg.rb', line 129

def clear
  @conn.exec_params("DELETE FROM chat_sdk_subscriptions WHERE key_prefix = $1", [@key_prefix])
  @conn.exec_params("DELETE FROM chat_sdk_locks WHERE key_prefix = $1", [@key_prefix])
  @conn.exec_params("DELETE FROM chat_sdk_cache WHERE key_prefix = $1", [@key_prefix])
end

#delete(key) ⇒ Object 



101
102
103
104
105
106
107
108
109
110
 
# File 'lib/chat_sdk/state/pg.rb', line 101

def delete(key)
  @conn.exec_params(
    "DELETE FROM chat_sdk_cache WHERE key_prefix = $1 AND cache_key = $2",
    [@key_prefix, key]
  )
  @conn.exec_params(
    "DELETE FROM chat_sdk_locks WHERE key_prefix = $1 AND lock_key = $2",
    [@key_prefix, key]
  )
end

#force_lock(key, owner:, ttl:) ⇒ Object 



66
67
68
69
70
71
72
73
74
75
 
# File 'lib/chat_sdk/state/pg.rb', line 66

def force_lock(key, owner:, ttl:)
  ttl_seconds = ttl.to_f
  @conn.exec_params(
    "INSERT INTO chat_sdk_locks (key_prefix, lock_key, owner, expires_at) " \
    "VALUES ($1, $2, $3, NOW() + INTERVAL '#{ttl_seconds} seconds') " \
    "ON CONFLICT (key_prefix, lock_key) DO UPDATE SET owner = EXCLUDED.owner, expires_at = EXCLUDED.expires_at",
    [@key_prefix, key, owner]
  )
  true
end

#get(key) ⇒ Object

Key-value store



79
80
81
82
83
84
85
86
87
 
# File 'lib/chat_sdk/state/pg.rb', line 79

def get(key)
  result = @conn.exec_params(
    "SELECT value FROM chat_sdk_cache WHERE key_prefix = $1 AND cache_key = $2 AND (expires_at IS NULL OR expires_at > NOW())",
    [@key_prefix, key]
  )
  return JSON.parse(result[0]["value"]) if result.ntuples > 0

  nil
end

#release_lock(key, owner:) ⇒ Object 



58
59
60
61
62
63
64
 
# File 'lib/chat_sdk/state/pg.rb', line 58

def release_lock(key, owner:)
  result = @conn.exec_params(
    "DELETE FROM chat_sdk_locks WHERE key_prefix = $1 AND lock_key = $2 AND owner = $3",
    [@key_prefix, key, owner]
  )
  result.cmd_tuples > 0
end

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



89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/chat_sdk/state/pg.rb', line 89

def set(key, value, ttl: nil)
  serialized = JSON.generate(value)
  expires = ttl ? "NOW() + INTERVAL '#{ttl.to_f} seconds'" : "NULL"
  @conn.exec_params(
    "INSERT INTO chat_sdk_cache (key_prefix, cache_key, value, expires_at) " \
    "VALUES ($1, $2, $3::jsonb, #{expires}) " \
    "ON CONFLICT (key_prefix, cache_key) DO UPDATE SET value = EXCLUDED.value, expires_at = EXCLUDED.expires_at",
    [@key_prefix, key, serialized]
  )
  value
end

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



112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/chat_sdk/state/pg.rb', line 112

def set_if_absent(key, value, ttl: nil)
  serialized = JSON.generate(value)
  @conn.exec_params(
    "DELETE FROM chat_sdk_cache WHERE key_prefix = $1 AND cache_key = $2 AND expires_at IS NOT NULL AND expires_at < NOW()",
    [@key_prefix, key]
  )
  expires = ttl ? "NOW() + INTERVAL '#{ttl.to_f} seconds'" : "NULL"
  result = @conn.exec_params(
    "INSERT INTO chat_sdk_cache (key_prefix, cache_key, value, expires_at) " \
    "VALUES ($1, $2, $3::jsonb, #{expires}) ON CONFLICT DO NOTHING",
    [@key_prefix, key, serialized]
  )
  result.cmd_tuples > 0
end

#subscribe(thread_id) ⇒ Object

Subscriptions



20
21
22
23
24
25
 
# File 'lib/chat_sdk/state/pg.rb', line 20

def subscribe(thread_id)
  @conn.exec_params(
    "INSERT INTO chat_sdk_subscriptions (key_prefix, thread_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
    [@key_prefix, thread_id]
  )
end

#subscribed?(thread_id) ⇒ Boolean

Returns:

  • (Boolean) 


34
35
36
37
38
39
40
 
# File 'lib/chat_sdk/state/pg.rb', line 34

def subscribed?(thread_id)
  result = @conn.exec_params(
    "SELECT 1 FROM chat_sdk_subscriptions WHERE key_prefix = $1 AND thread_id = $2",
    [@key_prefix, thread_id]
  )
  result.ntuples > 0
end

#unsubscribe(thread_id) ⇒ Object 



27
28
29
30
31
32
 
# File 'lib/chat_sdk/state/pg.rb', line 27

def unsubscribe(thread_id)
  @conn.exec_params(
    "DELETE FROM chat_sdk_subscriptions WHERE key_prefix = $1 AND thread_id = $2",
    [@key_prefix, thread_id]
  )
end