Class: Upkeep::Subscriptions::ActiveRecordSubscriptionPersistence

Inherits:
Object
  • Object
show all
Defined in:
lib/upkeep/subscriptions/active_record_subscription_persistence.rb

Constant Summary collapse

PERSIST_NOTIFICATION = 
"persist_subscription_store.upkeep"
INDEX_ENTRIES_SNAPSHOT_KEY = 
"__upkeep_index_entries"

Instance Method Summary collapse

Constructor Details

#initialize(subscription_record:, index_record:, index_builder:) ⇒ ActiveRecordSubscriptionPersistence

Returns a new instance of ActiveRecordSubscriptionPersistence.



16
17
18
19
20
21
22
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 16

def initialize(subscription_record:, index_record:, index_builder:)
  @subscription_record = subscription_record
  @index_record = index_record
  @index_builder = index_builder
  @count_mutex = Mutex.new
  @count_cache = nil
end

Instance Method Details

#count ⇒ Object 



97
98
99
100
101
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 97

def count
  @count_mutex.synchronize do
    @count_cache ||= subscription_record.count
  end
end

#delete(ids) ⇒ Object 



62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 62

def delete(ids)
  ids = Array(ids)
  return if ids.empty?

  silence_active_record_logging do
    ActiveRecord::Base.connection_pool.with_connection do
      ActiveRecord::Base.transaction do
        index_record.where(subscription_id: ids).delete_all
        deleted = subscription_record.where(id: ids).delete_all
        decrement_count_cache(deleted)
      end
    end
  end
end

#fetch(id) ⇒ Object 



77
78
79
80
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 77

def fetch(id)
  record = subscription_record.find(id)
  subscription_with_metadata(record)
end

#fetch_with_index_entries(id) ⇒ Object 



82
83
84
85
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 82

def fetch_with_index_entries(id)
  record = subscription_record.find(id)
  [subscription_with_metadata(record), index_entries_from_snapshot(record.recorder_snapshot)]
end

#persist_jobs(jobs) ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 24

def persist_jobs(jobs)
  if ActiveSupport::Notifications.notifier.listening?(PERSIST_NOTIFICATION)
    payload = {
      store: "active_record",
      jobs: jobs.size,
      subscriptions: jobs.count { |job| persist_subscription?(job) },
      index_jobs: jobs.count { |job| persist_index?(job) },
      dependency_entries: jobs.sum { |job| persist_index?(job) ? job.entries.size : 0 },
      pending_index_entries: jobs.sum { |job| persist_subscription?(job) ? job.entries.size : 0 },
      operations: operation_counts(jobs)
    }
    ActiveSupport::Notifications.instrument(PERSIST_NOTIFICATION, payload) do
      result = persist_jobs_without_instrumentation(jobs)
      payload[:subscription_rows] = result.fetch(:subscription_rows)
      payload[:index_rows] = result.fetch(:index_rows)
    end
  else
    persist_jobs_without_instrumentation(jobs)
  end
end

#prune_stale!(older_than:) ⇒ Object 



54
55
56
57
58
59
60
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 54

def prune_stale!(older_than:)
  stale_ids = subscription_record.where(subscription_record.arel_table[:updated_at].lt(older_than)).pluck(:id)
  return [] if stale_ids.empty?

  delete(stale_ids)
  stale_ids
end

#reset ⇒ Object 



91
92
93
94
95
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 91

def reset
  index_record.delete_all
  subscription_record.delete_all
  write_count_cache(0)
end

#subscriptions ⇒ Object 



87
88
89
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 87

def subscriptions
  subscription_record.order(:created_at, :id).map { |record| subscription_with_metadata(record) }
end

#touch(id, metadata:, now:) ⇒ Object 



45
46
47
48
49
50
51
52
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 45

def touch(id, metadata:, now:)
  subscription_record.where(id: id).find_each do |record|
    record.update_columns(
      metadata: record.metadata.to_h.merge(metadata),
      updated_at: now
    )
  end
end