Class: Upkeep::Subscriptions::ActiveRecordSubscriptionPersistence

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

Defined Under Namespace

Classes: Job

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:, shape_index_record:, index_builder:) ⇒ ActiveRecordSubscriptionPersistence

Returns a new instance of ActiveRecordSubscriptionPersistence.



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

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

Instance Method Details

#count ⇒ Object 



106
107
108
109
110
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 106

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

#delete(ids) ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 68

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
        shape_keys = shape_keys_for_subscriptions(ids)
        index_record.where(subscription_id: ids).delete_all
        deleted = subscription_record.where(id: ids).delete_all
        delete_orphaned_shape_index_rows(shape_keys)
        decrement_count_cache(deleted)
      end
    end
  end
end

#fetch(id) ⇒ Object 



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

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

#fetch_with_index_entries(id) ⇒ Object 



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

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 



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

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)
      payload[:direct_index_rows] = result.fetch(:direct_index_rows)
      payload[:shape_index_rows] = result.fetch(:shape_index_rows)
    end
  else
    persist_jobs_without_instrumentation(jobs)
  end
end

#prune_stale!(older_than:, limit: nil) ⇒ Object 



58
59
60
61
62
63
64
65
66
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 58

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

  delete(stale_ids)
  stale_ids
end

#reset ⇒ Object 



99
100
101
102
103
104
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 99

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

#subscriptions ⇒ Object 



95
96
97
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 95

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

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



49
50
51
52
53
54
55
56
 
# File 'lib/upkeep/subscriptions/active_record_subscription_persistence.rb', line 49

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