Class: Trifle::Stats::Driver::Postgres

Inherits:
Object
  • Object
show all
Includes:
Mixins::Packer
Defined in:
lib/trifle/stats/driver/postgres.rb

Overview

rubocop:disable Metrics/ClassLength

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Mixins::Packer

included

Constructor Details

#initialize(client = PG::Connection.new, table_name: 'trifle_stats', joined_identifier: :full, ping_table_name: nil, system_tracking: true) ⇒ Postgres

rubocop:disable Layout/LineLength



12
13
14
15
16
17
18
19
 
# File 'lib/trifle/stats/driver/postgres.rb', line 12

def initialize(client = PG::Connection.new, table_name: 'trifle_stats', joined_identifier: :full, ping_table_name: nil, system_tracking: true) # rubocop:disable Layout/LineLength
  @client = client
  @table_name = table_name
  @ping_table_name = ping_table_name || "#{table_name}_ping"
  @joined_identifier = self.class.normalize_joined_identifier(joined_identifier)
  @system_tracking = system_tracking
  @separator = '::'
end

Instance Attribute Details

#client ⇒ Object

Returns the value of attribute client.



10
11
12
 
# File 'lib/trifle/stats/driver/postgres.rb', line 10

def client
  @client
end

#ping_table_name ⇒ Object

Returns the value of attribute ping_table_name.



10
11
12
 
# File 'lib/trifle/stats/driver/postgres.rb', line 10

def ping_table_name
  @ping_table_name
end

#separator ⇒ Object (readonly)

Returns the value of attribute separator.



47
48
49
 
# File 'lib/trifle/stats/driver/postgres.rb', line 47

def separator
  @separator
end

#table_name ⇒ Object

Returns the value of attribute table_name.



10
11
12
 
# File 'lib/trifle/stats/driver/postgres.rb', line 10

def table_name
  @table_name
end

Class Method Details

.normalize_joined_identifier(value) ⇒ Object 



194
195
196
197
198
199
200
201
 
# File 'lib/trifle/stats/driver/postgres.rb', line 194

def self.normalize_joined_identifier(value)
  case value
  when nil, :full, 'full', :partial, 'partial'
    value.nil? ? nil : value.to_sym
  else
    raise ArgumentError, 'joined_identifier must be nil, :full, "full", :partial, or "partial"'
  end
end

.setup!(client = PG::Connection.new, table_name: 'trifle_stats', joined_identifier: :full, ping_table_name: nil) ⇒ Object

rubocop:disable Layout/LineLength, Metrics/MethodLength



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/trifle/stats/driver/postgres.rb', line 21

def self.setup!(client = PG::Connection.new, table_name: 'trifle_stats', joined_identifier: :full, ping_table_name: nil) # rubocop:disable Layout/LineLength, Metrics/MethodLength
  ping_table_name ||= "#{table_name}_ping"
  identifier_mode = normalize_joined_identifier(joined_identifier)

  case identifier_mode
  when :full
    client.exec("CREATE TABLE #{table_name} (key VARCHAR(255) PRIMARY KEY, data JSONB NOT NULL DEFAULT '{}'::jsonb)") # rubocop:disable Layout/LineLength
  when :partial
    client.exec("CREATE TABLE #{table_name} (key VARCHAR(255) NOT NULL, at TIMESTAMPTZ NOT NULL, data JSONB NOT NULL DEFAULT '{}'::jsonb, PRIMARY KEY (key, at))") # rubocop:disable Layout/LineLength
  else
    client.exec("CREATE TABLE #{table_name} (key VARCHAR(255) NOT NULL, granularity VARCHAR(255) NOT NULL, at TIMESTAMPTZ NOT NULL, data JSONB NOT NULL DEFAULT '{}'::jsonb, PRIMARY KEY (key, granularity, at))") # rubocop:disable Layout/LineLength

    # Create ping table for separated mode only
    client.exec("CREATE TABLE #{ping_table_name} (key VARCHAR(255) PRIMARY KEY, at TIMESTAMPTZ NOT NULL, data JSONB NOT NULL DEFAULT '{}'::jsonb)") # rubocop:disable Layout/LineLength
  end
end

Instance Method Details

#description ⇒ Object 



38
39
40
41
42
43
44
45
 
# File 'lib/trifle/stats/driver/postgres.rb', line 38

def description
  mode = if @joined_identifier == :full
           'J'
         else
           @joined_identifier == :partial ? 'P' : 'S'
         end
  "#{self.class.name}(#{mode})"
end

#get(keys:) ⇒ Object 



122
123
124
125
126
 
# File 'lib/trifle/stats/driver/postgres.rb', line 122

def get(keys:)
  identifiers = keys.map { |key| identifier_for(key) }
  data = get_all(identifiers: identifiers)
  identifiers.map { |identifier| self.class.unpack(hash: data.fetch(identifier, {})) }
end

#get_all(identifiers:) ⇒ Object

rubocop:disable Metrics/AbcSize



128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/trifle/stats/driver/postgres.rb', line 128

def get_all(identifiers:) # rubocop:disable Metrics/AbcSize
  query, params = get_query(identifiers: identifiers)
  results = client.exec_params(query, params).to_a
  sample = identifiers.first

  results.each_with_object(Hash.new({})) do |r, o|
    identifier = sample.to_h { |k, _| [k, k == :at ? Time.parse(r[k.to_s]) : r[k.to_s]] }

    o[identifier] = JSON.parse(r['data'])
  rescue JSON::ParserError
    nil
  end
end

#get_query(identifiers:) ⇒ Object

rubocop:disable Metrics/MethodLength



142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/trifle/stats/driver/postgres.rb', line 142

def get_query(identifiers:) # rubocop:disable Metrics/MethodLength
  params = []
  conditions = identifiers.map do |identifier|
    identifier.map do |k, v|
      params << query_param(v)
      "#{k} = $#{params.size}"
    end.join(' AND ')
  end.join(' OR ')

  query = <<-SQL
    SELECT * FROM #{table_name} WHERE #{conditions};
  SQL
  [query, params]
end

#inc(keys:, values:, count: 1, tracking_key: nil) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/trifle/stats/driver/postgres.rb', line 59

def inc(keys:, values:, count: 1, tracking_key: nil)
  data = self.class.pack(hash: values)
  client.transaction do |c|
    keys.map do |key|
      identifier = identifier_for(key)
      query, params = inc_query(identifier: identifier, data: data)
      c.exec_params(query, params)
      track_system_data(c, key, count, tracking_key)
    end
  end
end

#inc_query(identifier:, data:) ⇒ Object

rubocop:disable Metrics/AbcSize, Metrics/MethodLength



71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/trifle/stats/driver/postgres.rb', line 71

def inc_query(identifier:, data:) # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
  columns = identifier.keys.join(', ')
  placeholders = identifier.each_with_index.map { |_, i| "$#{i + 1}" }.join(', ')
  expression = data.inject("to_jsonb(#{table_name}.data)") do |o, (k, v)|
    path = escape_string(k.to_s)
    "jsonb_set(#{o}, '{#{path}}', (COALESCE(#{table_name}.data->>'#{path}', '0')::numeric + #{numeric_value(k, v)})::text::jsonb)" # rubocop:disable Layout/LineLength
  end

  query = <<-SQL
    INSERT INTO #{table_name} (#{columns}, data) VALUES (#{placeholders}, $#{identifier.size + 1})
    ON CONFLICT (#{columns}) DO UPDATE SET data = #{expression};
  SQL
  [query, query_params(identifier) + [data.to_json]]
end

#ping(key:, values:) ⇒ Object 



157
158
159
160
161
162
163
164
165
 
# File 'lib/trifle/stats/driver/postgres.rb', line 157

def ping(key:, values:)
  return [] if @joined_identifier

  data = self.class.pack(hash: { data: values, at: key.at })
  query, params = ping_query(key: key.key, at: key.at, data: data)
  client.transaction do |c|
    c.exec_params(query, params)
  end
end

#ping_query(key:, at:, data:) ⇒ Object 



167
168
169
170
171
172
173
 
# File 'lib/trifle/stats/driver/postgres.rb', line 167

def ping_query(key:, at:, data:)
  query = <<-SQL
    INSERT INTO #{ping_table_name} (key, at, data) VALUES ($1, $2, $3)
    ON CONFLICT (key) DO UPDATE SET at = $2, data = $3::jsonb;
  SQL
  [query, [key.to_s, at.iso8601, data.to_json]]
end

#scan(key:) ⇒ Object 



175
176
177
178
179
180
181
182
183
184
185
 
# File 'lib/trifle/stats/driver/postgres.rb', line 175

def scan(key:)
  return [] if @joined_identifier

  query, params = scan_query(key: key.key)
  result = client.exec_params(query, params).to_a.first
  return [] if result.nil?

  [Time.parse(result['at']), self.class.unpack(hash: JSON.parse(result['data']))]
rescue JSON::ParserError
  []
end

#scan_query(key:) ⇒ Object 



187
188
189
190
191
192
 
# File 'lib/trifle/stats/driver/postgres.rb', line 187

def scan_query(key:)
  query = <<-SQL
    SELECT at, data FROM #{ping_table_name} WHERE key = $1 ORDER BY at DESC LIMIT 1;
  SQL
  [query, [key.to_s]]
end

#set(keys:, values:, count: 1, tracking_key: nil) ⇒ Object 



86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/trifle/stats/driver/postgres.rb', line 86

def set(keys:, values:, count: 1, tracking_key: nil)
  data = self.class.pack(hash: values)
  client.transaction do |c|
    keys.map do |key|
      identifier = identifier_for(key)
      query, params = set_query(identifier: identifier, data: data)
      c.exec_params(query, params)
      track_system_data(c, key, count, tracking_key)
    end
  end
end

#set_query(identifier:, data:) ⇒ Object

rubocop:disable Metrics/AbcSize, Metrics/MethodLength



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/trifle/stats/driver/postgres.rb', line 106

def set_query(identifier:, data:) # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
  columns = identifier.keys.join(', ')
  placeholders = identifier.each_with_index.map { |_, i| "$#{i + 1}" }.join(', ')
  params = query_params(identifier) + [data.to_json]
  expression = data.inject("to_jsonb(#{table_name}.data)") do |o, (k, v)|
    params << v.to_json
    "jsonb_set(#{o}, '{#{escape_string(k.to_s)}}', $#{params.size}::jsonb)"
  end

  query = <<-SQL
    INSERT INTO #{table_name} (#{columns}, data) VALUES (#{placeholders}, $#{identifier.size + 1})
    ON CONFLICT (#{columns}) DO UPDATE SET data = #{expression}
  SQL
  [query, params]
end

#system_data_for(key:, count: 1, tracking_key: nil) ⇒ Object 



54
55
56
57
 
# File 'lib/trifle/stats/driver/postgres.rb', line 54

def system_data_for(key:, count: 1, tracking_key: nil)
  tracking_key ||= key.key
  self.class.pack(hash: { count: count, keys: { tracking_key => count } })
end

#system_identifier_for(key:) ⇒ Object 



49
50
51
52
 
# File 'lib/trifle/stats/driver/postgres.rb', line 49

def system_identifier_for(key:)
  key = Nocturnal::Key.new(key: '__system__key__', granularity: key.granularity, at: key.at)
  identifier_for(key)
end

#track_system_data(connection, key, count, tracking_key) ⇒ Object 



98
99
100
101
102
103
104
 
# File 'lib/trifle/stats/driver/postgres.rb', line 98

def track_system_data(connection, key, count, tracking_key)
  return unless @system_tracking

  system_data = system_data_for(key: key, count: count, tracking_key: tracking_key)
  query, params = inc_query(identifier: system_identifier_for(key: key), data: system_data)
  connection.exec_params(query, params)
end