Class: PromptObjects::Operations::Retention

Inherits:
Object
  • Object
show all
Defined in:
lib/prompt_objects/operations/retention.rb

Overview

Bounds replay logs while preserving current snapshots and durable domain rows. Pruned clients recover from workspace/thread snapshots.

Constant Summary collapse

DEFAULT_RUN_EVENT_LIMIT = 
500
DEFAULT_WORKSPACE_EVENT_LIMIT = 
5_000

Instance Method Summary collapse

Constructor Details

#initialize(store) ⇒ Retention

Returns a new instance of Retention.



11
12
13
 
# File 'lib/prompt_objects/operations/retention.rb', line 11

def initialize(store)
  @store = store
end

Instance Method Details

#compact! ⇒ Object 



54
55
56
57
58
59
60
 
# File 'lib/prompt_objects/operations/retention.rb', line 54

def compact!
  @store.with_connection do |db|
    db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    db.execute("VACUUM")
  end
  true
end

#prune_events!(run_event_limit: DEFAULT_RUN_EVENT_LIMIT, workspace_event_limit: DEFAULT_WORKSPACE_EVENT_LIMIT) ⇒ Object

Raises:

  • (ArgumentError) 


15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/prompt_objects/operations/retention.rb', line 15

def prune_events!(run_event_limit: DEFAULT_RUN_EVENT_LIMIT,
                  workspace_event_limit: DEFAULT_WORKSPACE_EVENT_LIMIT)
  raise ArgumentError, "run_event_limit must be positive" unless run_event_limit.positive?
  raise ArgumentError, "workspace_event_limit must be positive" unless workspace_event_limit.positive?

  @store.transaction do |db|
    run_event_ids = db.execute(<<~SQL, [run_event_limit]).map { |row| row["id"] }
      SELECT id FROM (
        SELECT id, ROW_NUMBER() OVER (PARTITION BY run_id ORDER BY sequence DESC) AS position
        FROM run_events
      ) WHERE position > ?
    SQL
    workspace_cursors = run_event_ids.empty? ? [] : db.execute(
      "SELECT workspace_cursor FROM run_events WHERE id IN (#{placeholders(run_event_ids)})",
      run_event_ids
    ).map { |row| row["workspace_cursor"] }
    delete_ids(db, "run_events", run_event_ids)
    delete_ids(db, "workspace_events", workspace_cursors)

    workspace_event_ids = db.execute(<<~SQL, [workspace_event_limit]).map { |row| row["id"] }
      SELECT id FROM (
        SELECT id, ROW_NUMBER() OVER (
          PARTITION BY workspace_session_id ORDER BY id DESC
        ) AS position
        FROM workspace_events
      ) WHERE position > ?
    SQL
    workspace_event_ids.reject! do |id|
      db.get_first_value("SELECT 1 FROM run_events WHERE workspace_cursor = ?", [id])
    end
    delete_ids(db, "workspace_events", workspace_event_ids)

    {
      run_events_deleted: run_event_ids.length,
      workspace_events_deleted: (workspace_cursors + workspace_event_ids).uniq.length
    }
  end
end