Class: Sentiero::Stores::SQLite

Inherits:
Sentiero::Store show all
Defined in:
lib/sentiero/stores/sqlite.rb,
lib/sentiero/stores/sqlite/schema.rb

Defined Under Namespace

Modules: Schema Classes: Where

Constant Summary collapse

SCAN_IN_CHUNK =

Batched scan: avoids the base's get_session + get_events-per-window N+1.

500

Constants inherited from Sentiero::Store

Sentiero::Store::MAX_METADATA_KEYS, Sentiero::Store::MAX_METADATA_VALUE_SIZE, Sentiero::Store::PROBLEM_TITLE_MAX, Sentiero::Store::VALID_ID, Sentiero::Store::VALID_STATUS

Instance Attribute Summary

Attributes inherited from Sentiero::Store

#limits

Instance Method Summary collapse

Constructor Details

#initialize(path: "sentiero.db", limits: nil) ⇒ SQLite

Returns a new instance of SQLite.



46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/sentiero/stores/sqlite.rb', line 46

def initialize(path: "sentiero.db", limits: nil)
  unless defined?(::SQLite3)
    raise LoadError, "The sqlite3 gem is required for Sentiero::Stores::SQLite. Add `gem 'sqlite3'` to your Gemfile."
  end

  @limits = limits
  @monitor = Monitor.new
  @path = path
  @db = create_database
  Schema.create(@db)
end

Instance Method Details

#clear! ⇒ Object 



429
430
431
432
433
434
435
436
437
438
 
# File 'lib/sentiero/stores/sqlite.rb', line 429

def clear!
  @db.transaction do
    @db.execute("DELETE FROM events")
    @db.execute("DELETE FROM sessions")
    @db.execute("DELETE FROM problems")
    @db.execute("DELETE FROM occurrences")
    @db.execute("DELETE FROM server_events")
  end
  nil
end

#count_occurrences(problem_id, after: nil) ⇒ Object

COUNT(*) on the (fingerprint, timestamp) index, no row materialization.



350
351
352
353
354
355
 
# File 'lib/sentiero/stores/sqlite.rb', line 350

def count_occurrences(problem_id, after: nil)
  validate_id!(problem_id)
  where = Where.new.add("fingerprint = ?", problem_id)
  where.add("timestamp > ?", after.to_f) if after
  @db.get_first_value("SELECT COUNT(*) FROM occurrences #{where.clause}", where.params)
end

#delete_session(session_id) ⇒ Object 



245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/sentiero/stores/sqlite.rb', line 245

def delete_session(session_id)
  validate_id!(session_id)

  @db.transaction do
    @db.execute("DELETE FROM events WHERE session_id = ?", [session_id])
    @db.execute("DELETE FROM sessions WHERE session_id = ?", [session_id])
    @db.execute("DELETE FROM occurrences WHERE session_id = ?", [session_id])
    @db.execute("DELETE FROM server_events WHERE session_id = ?", [session_id])
  end
  nil
end

#delete_window(ref) ⇒ Object 



257
258
259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'lib/sentiero/stores/sqlite.rb', line 257

def delete_window(ref)
  validate_window_ref!(ref)
  session_id, window_id = ref.session_id, ref.window_id

  @db.transaction do
    @db.execute("DELETE FROM events WHERE session_id = ? AND window_id = ?", [session_id, window_id])

    remaining = @db.get_first_value("SELECT COUNT(*) FROM events WHERE session_id = ?", [session_id])
    if remaining == 0
      @db.execute("DELETE FROM sessions WHERE session_id = ?", [session_id])
    end
  end
  nil
end

#each_session_events(limit: nil, since: nil, until_time: nil) ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/sentiero/stores/sqlite.rb', line 106

def each_session_events(limit: nil, since: nil, until_time: nil)
  return enum_for(:each_session_events, limit: limit, since: since, until_time: until_time) unless block_given?

  cap = limit || limits.analytics_max_scan_sessions
  rows = scan_session_rows(cap, since, until_time)
  return if rows.empty?

  events = events_by_session_window(rows.map { |row| row["session_id"] })
  rows.each do |row|
    windows = events[row["session_id"]]
    next unless windows

    summary = scan_summary(row, windows)
    windows.each { |window_id, window_events| yield summary, window_id, window_events }
  end
end

#get_events(ref, after: nil, limit: nil) ⇒ Object 



207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/sentiero/stores/sqlite.rb', line 207

def get_events(ref, after: nil, limit: nil)
  validate_window_ref!(ref)
  session_id, window_id = ref.session_id, ref.window_id

  conditions = ["session_id = ?", "window_id = ?"]
  params = [session_id, window_id]

  if after
    conditions << "timestamp > ?"
    params << after.to_f
  end

  sql = "SELECT data FROM events WHERE #{conditions.join(" AND ")} ORDER BY timestamp ASC"
  if limit
    sql += " LIMIT ?"
    params << limit.to_i
  end

  @db.execute(sql, params).map { |event_row| JSON.parse(event_row["data"]) }
end

#get_occurrences(problem_id, after: nil, limit: nil) ⇒ Object 



336
337
338
339
340
341
342
343
344
345
346
347
 
# File 'lib/sentiero/stores/sqlite.rb', line 336

def get_occurrences(problem_id, after: nil, limit: nil)
  validate_id!(problem_id)
  where = Where.new.add("fingerprint = ?", problem_id)
  where.add("timestamp > ?", after.to_f) if after
  sql = "SELECT data FROM occurrences #{where.clause} ORDER BY timestamp ASC"
  params = where.params
  if limit
    sql += " LIMIT ?"
    params += [limit.to_i]
  end
  @db.execute(sql, params).map { |row| JSON.parse(row["data"]) }
end

#get_problem(problem_id) ⇒ Object 



330
331
332
333
334
 
# File 'lib/sentiero/stores/sqlite.rb', line 330

def get_problem(problem_id)
  validate_id!(problem_id)
  row = @db.get_first_row("SELECT * FROM problems WHERE fingerprint = ?", [problem_id])
  row && problem_row_to_hash(row)
end

#get_server_event(event_id) ⇒ Object 



379
380
381
382
383
 
# File 'lib/sentiero/stores/sqlite.rb', line 379

def get_server_event(event_id)
  validate_id!(event_id)
  row = @db.get_first_row("SELECT data FROM server_events WHERE event_id = ?", [event_id])
  row && JSON.parse(row["data"])
end

#get_session(session_id) ⇒ Object 



172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
 
# File 'lib/sentiero/stores/sqlite.rb', line 172

def get_session(session_id)
  validate_id!(session_id)

  row = @db.get_first_row("SELECT * FROM sessions WHERE session_id = ?", [session_id])
  return nil unless row

  window_stats = @db.execute(
    "SELECT window_id, COUNT(*) AS cnt, MIN(timestamp) AS min_ts, MAX(timestamp) AS max_ts FROM events WHERE session_id = ? GROUP BY window_id",
    [session_id]
  )

  window_data = window_stats.map { |stats|
    window = {window_id: stats["window_id"], event_count: stats["cnt"]}
    window[:first_event_at] = stats["min_ts"] if stats["min_ts"]
    window[:last_event_at] = stats["max_ts"] if stats["max_ts"]
    window
  }

  result = {
    session_id: session_id,
    windows: window_data,
    created_at: row["created_at"],
    updated_at: row["updated_at"],
    first_event_at: row["first_event_at"],
    last_event_at: row["last_event_at"]
  }

  if row["metadata"]
    parsed = JSON.parse(row["metadata"])
    result[:metadata] = parsed unless empty_metadata?(parsed)
  end

  result
end

#list_problems(project:, limit:, offset: 0, status: nil, sort_by: nil, search: nil, since: nil, until_time: nil) ⇒ Object

Native-SQL mirror of the base filter_and_page_problems.



311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
 
# File 'lib/sentiero/stores/sqlite.rb', line 311

def list_problems(project:, limit:, offset: 0, status: nil, sort_by: nil, search: nil, since: nil, until_time: nil)
  where = Where.new
  where.add("project = ?", project) unless project.nil?
  where.add("status = ?", status) if status
  where.add("last_seen >= ?", since.to_f) if since
  where.add("last_seen <= ?", until_time.to_f) if until_time
  if search && !search.empty?
    pattern = "%#{search}%"
    where.add("(title LIKE ? OR exception_class LIKE ?)", pattern, pattern)
  end
  order = case sort_by
  when "first_seen" then "ORDER BY first_seen DESC"
  when "count" then "ORDER BY count DESC"
  else "ORDER BY last_seen DESC"
  end
  @db.execute("SELECT * FROM problems #{where.clause} #{order} LIMIT ? OFFSET ?", where.params + [limit, offset])
    .map { |row| problem_row_to_hash(row) }
end

#list_server_events(project:, limit:, name: nil, level: nil, session_id: nil, after: nil) ⇒ Object 



385
386
387
388
389
390
391
392
393
394
 
# File 'lib/sentiero/stores/sqlite.rb', line 385

def list_server_events(project:, limit:, name: nil, level: nil, session_id: nil, after: nil)
  where = Where.new
  where.add("project = ?", project) unless project.nil?
  where.add("name = ?", name) if name
  where.add("level = ?", level) if level
  where.add("session_id = ?", session_id) if session_id
  where.add("timestamp > ?", after.to_f) if after
  @db.execute("SELECT data FROM server_events #{where.clause} ORDER BY timestamp ASC LIMIT ?", where.params + [limit])
    .map { |row| JSON.parse(row["data"]) }
end

#list_sessions(limit:, offset: 0, since: nil, until_time: nil, sort_by: nil, search: nil) ⇒ Object 



123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/sentiero/stores/sqlite.rb', line 123

def list_sessions(limit:, offset: 0, since: nil, until_time: nil, sort_by: nil, search: nil)
  where = Where.new
  where.add("s.updated_at >= ?", since.to_f) if since
  where.add("s.updated_at <= ?", until_time.to_f) if until_time
  if search && !search.empty?
    pattern = "%#{search}%"
    where.add("(s.session_id LIKE ? OR COALESCE(s.metadata, '') LIKE ?)", pattern, pattern)
  end

  order_clause = case sort_by
  when "created_at"
    "ORDER BY s.created_at DESC"
  when "event_count"
    "ORDER BY event_count DESC"
  else
    "ORDER BY s.updated_at DESC"
  end

  sql = <<~SQL
    SELECT s.session_id, s.created_at, s.updated_at, s.first_event_at, s.last_event_at, s.metadata,
           COUNT(e.id) AS event_count
    FROM sessions s
    LEFT JOIN events e ON e.session_id = s.session_id
    #{where.clause}
    GROUP BY s.id
    #{order_clause}
    LIMIT ? OFFSET ?
  SQL

  rows = @db.execute(sql, where.params + [limit, offset])

  rows.map { |row|
    window_ids = @db.execute(
      "SELECT DISTINCT window_id FROM events WHERE session_id = ?", [row["session_id"]]
    ).map { |window_row| window_row["window_id"] }

    summary_hash(
      session_id: row["session_id"],
      window_ids: window_ids,
      event_count: row["event_count"],
      created_at: row["created_at"],
      updated_at: row["updated_at"],
      first_event_at: row["first_event_at"],
      last_event_at: row["last_event_at"],
      metadata: row["metadata"] && JSON.parse(row["metadata"])
    )
  }
end

#occurrences_for_session(session_id, limit: nil) ⇒ Object 



396
397
398
399
400
401
402
403
404
405
 
# File 'lib/sentiero/stores/sqlite.rb', line 396

def occurrences_for_session(session_id, limit: nil)
  validate_id!(session_id)
  sql = "SELECT data FROM occurrences WHERE session_id = ? ORDER BY timestamp ASC"
  params = [session_id]
  if limit
    sql += " LIMIT ?"
    params << limit.to_i
  end
  @db.execute(sql, params).map { |row| JSON.parse(row["data"]) }
end

#purge_older_than(seconds) ⇒ Object 



440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
 
# File 'lib/sentiero/stores/sqlite.rb', line 440

def purge_older_than(seconds)
  cutoff = Time.now.to_f - seconds
  session_count = nil

  @db.transaction do
    @db.execute(
      "DELETE FROM events WHERE session_id IN (SELECT session_id FROM sessions WHERE updated_at < ?)",
      [cutoff]
    )
    @db.execute("DELETE FROM sessions WHERE updated_at < ?", [cutoff])
    session_count = @db.changes

    @db.execute("DELETE FROM server_events WHERE timestamp < ?", [cutoff])
    @db.execute("DELETE FROM occurrences WHERE timestamp < ?", [cutoff])
    @db.execute(
      "DELETE FROM occurrences WHERE fingerprint IN (SELECT fingerprint FROM problems WHERE last_seen < ?)",
      [cutoff]
    )
    @db.execute("DELETE FROM problems WHERE last_seen < ?", [cutoff])
  end

  session_count
end

#save_events(ref, events) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/sentiero/stores/sqlite.rb', line 58

def save_events(ref, events)
  return if events.nil? || events.empty?

  validate_window_ref!(ref)
  session_id, window_id = ref.session_id, ref.window_id

  now = Time.now.to_f
  event_timestamps = events.filter_map { |e| e["timestamp"]&.to_f }
  batch_min = event_timestamps.min
  batch_max = event_timestamps.max

  @db.transaction do
    existing = @db.get_first_row("SELECT id, first_event_at, last_event_at FROM sessions WHERE session_id = ?", [session_id])

    if existing
      new_first = batch_min ? [existing["first_event_at"], batch_min].compact.min : existing["first_event_at"]
      new_last = batch_max ? [existing["last_event_at"], batch_max].compact.max : existing["last_event_at"]

      @db.execute(
        "UPDATE sessions SET updated_at = ?, first_event_at = ?, last_event_at = ? WHERE session_id = ?",
        [now, new_first, new_last, session_id]
      )
    else
      @db.execute(
        "INSERT INTO sessions (session_id, created_at, updated_at, first_event_at, last_event_at, metadata) VALUES (?, ?, ?, ?, ?, NULL)",
        [session_id, now, now, batch_min, batch_max]
      )
    end

    stmt = @db.prepare("INSERT INTO events (session_id, window_id, timestamp, data) VALUES (?, ?, ?, ?)")
    begin
      events.each do |event|
        stmt.execute(session_id, window_id, event["timestamp"]&.to_f, JSON.generate(event))
      end
    ensure
      stmt.close
    end

    enforce_max_events(session_id)
    enforce_max_sessions(session_id)
  end

  nil
end

#save_metadata(session_id, metadata) ⇒ Object 



228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
 
# File 'lib/sentiero/stores/sqlite.rb', line 228

def save_metadata(session_id, metadata)
  return unless metadata.is_a?(Hash) && !metadata.empty?

  validate_id!(session_id)
  validate_metadata!(metadata)

  @db.transaction do
    row = @db.get_first_row("SELECT metadata FROM sessions WHERE session_id = ?", [session_id])
    return unless row

    existing = row["metadata"] ? JSON.parse(row["metadata"]) : {}
    merged = existing.merge(metadata.transform_keys(&:to_s))
    @db.execute("UPDATE sessions SET metadata = ? WHERE session_id = ?", [JSON.generate(merged), session_id])
  end
  nil
end

#save_occurrence(occurrence) ⇒ Object 



272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
 
# File 'lib/sentiero/stores/sqlite.rb', line 272

def save_occurrence(occurrence)
  validate_occurrence!(occurrence)
  fp = occurrence["fingerprint"]
  ts = occurrence["timestamp"].to_f
  occ_id = SecureRandom.uuid
  stored = occurrence.merge("id" => occ_id)

  # Native-SQL upsert mirroring the base new_problem_attrs/touched_problem_attrs.
  @db.transaction do
    existing = @db.get_first_row(
      "SELECT count, first_seen, last_seen, status, resolved_at FROM problems WHERE fingerprint = ?", [fp]
    )
    if existing
      reopening = existing["status"] == "resolved"
      @db.execute(
        "UPDATE problems SET count = count + 1, first_seen = ?, last_seen = ?, message = ?, status = ?, resolved_at = ? WHERE fingerprint = ?",
        [[existing["first_seen"], ts].min, [existing["last_seen"], ts].max, occurrence["message"],
          reopening ? "open" : existing["status"], reopening ? nil : existing["resolved_at"], fp]
      )
    else
      @db.execute(
        "INSERT INTO problems (fingerprint, project, exception_class, title, message, count, status, first_seen, last_seen, resolved_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)",
        [fp, occurrence["project"], occurrence["exception_class"], build_problem_title(occurrence),
          occurrence["message"], 1, "open", ts, ts]
      )
    end

    @db.execute(
      "INSERT INTO occurrences (occurrence_id, fingerprint, session_id, timestamp, data) VALUES (?, ?, ?, ?, ?)",
      [occ_id, fp, occurrence["session_id"], ts, JSON.generate(stored)]
    )

    enforce_max_problems
  end
  save_metadata(occurrence["session_id"], {"has_errors" => true}) if occurrence["session_id"]
  fp
end

#save_server_event(event) ⇒ Object 



365
366
367
368
369
370
371
372
373
374
375
376
377
 
# File 'lib/sentiero/stores/sqlite.rb', line 365

def save_server_event(event)
  validate_server_event!(event)
  ev_id = SecureRandom.uuid
  stored = event.merge("id" => ev_id)
  @db.transaction do
    @db.execute(
      "INSERT INTO server_events (event_id, project, name, level, session_id, timestamp, data) VALUES (?, ?, ?, ?, ?, ?, ?)",
      [ev_id, event["project"], event["name"], event["level"], event["session_id"], event["timestamp"].to_f, JSON.generate(stored)]
    )
    enforce_max_server_events
  end
  nil
end

#server_events_for_session(session_id, limit: nil) ⇒ Object 



407
408
409
410
411
412
413
414
415
416
 
# File 'lib/sentiero/stores/sqlite.rb', line 407

def server_events_for_session(session_id, limit: nil)
  validate_id!(session_id)
  sql = "SELECT data FROM server_events WHERE session_id = ? ORDER BY timestamp ASC"
  params = [session_id]
  if limit
    sql += " LIMIT ?"
    params << limit.to_i
  end
  @db.execute(sql, params).map { |row| JSON.parse(row["data"]) }
end

#session_ids_for_problem(problem_id, limit: nil) ⇒ Object 



418
419
420
421
422
423
424
425
426
427
 
# File 'lib/sentiero/stores/sqlite.rb', line 418

def session_ids_for_problem(problem_id, limit: nil)
  validate_id!(problem_id)
  sql = "SELECT session_id, MAX(timestamp) AS ts FROM occurrences WHERE fingerprint = ? AND session_id IS NOT NULL GROUP BY session_id ORDER BY ts DESC"
  params = [problem_id]
  if limit
    sql += " LIMIT ?"
    params << limit.to_i
  end
  @db.execute(sql, params).map { |row| row["session_id"] }
end

#update_problem_status(problem_id, status) ⇒ Object 



357
358
359
360
361
362
363
 
# File 'lib/sentiero/stores/sqlite.rb', line 357

def update_problem_status(problem_id, status)
  validate_id!(problem_id)
  validate_status!(status)
  resolved_at = (status == "resolved") ? Time.now.to_f : nil
  @db.execute("UPDATE problems SET status = ?, resolved_at = ? WHERE fingerprint = ?", [status, resolved_at, problem_id])
  nil
end