Class: Upkeep::Subscriptions::ActiveRecordStore

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

Defined Under Namespace

Classes: DeferredIndexWrite, IndexEntryRecord, ShapeIndexEntryRecord, SubscriptionRecord

Constant Summary collapse

LOOKUP_NOTIFICATION = 
LayeredReverseIndex::LOOKUP_NOTIFICATION
REGISTER_NOTIFICATION = 
"register_subscription_store.upkeep"
ACTIVATE_NOTIFICATION = 
"activate_subscription_store.upkeep"
PERSIST_NOTIFICATION = 
ActiveRecordSubscriptionPersistence::PERSIST_NOTIFICATION
DURABILITY_MODE = 
"sync_subscription_row_index_on_subscribe"
INDEX_DURABILITY = 
"on_subscribe"
REQUIRED_SCHEMA = 
{
  "upkeep_subscriptions" => {
    "id" => :string,
    "subscriber_id" => :string,
    "recorder_snapshot" => :json,
    "metadata" => :json,
    "subscription_shape_key" => :string,
    "created_at" => :datetime,
    "updated_at" => :datetime
  },
  "upkeep_subscription_index_entries" => {
    "subscription_id" => :string,
    "lookup_key_digest" => :string,
    "dependency_source" => :string,
    "lookup_table" => :string,
    "lookup_record_id_snapshot" => :json,
    "lookup_attribute" => :string,
    "dependency_table" => :string,
    "dependency_predicate_digest" => :string,
    "dependency_metadata_snapshot" => :json,
    "owner_ids_snapshot" => :json,
    "created_at" => :datetime,
    "updated_at" => :datetime
  },
  "upkeep_subscription_shape_index_entries" => {
    "subscription_shape_key" => :string,
    "lookup_key_digest" => :string,
    "dependency_source" => :string,
    "lookup_table" => :string,
    "lookup_record_id_snapshot" => :json,
    "lookup_attribute" => :string,
    "dependency_table" => :string,
    "dependency_predicate_digest" => :string,
    "dependency_metadata_snapshot" => :json,
    "owner_ids_snapshot" => :json,
    "created_at" => :datetime,
    "updated_at" => :datetime
  }
}.freeze

Constants inherited from BaseStore

BaseStore::DEFAULT_SUBSCRIPTION_TTL, BaseStore::PRUNE_NOTIFICATION, BaseStore::TRIM_BATCH_LIMIT, BaseStore::TRIM_EVERY

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from BaseStore

#explain, #fetch, #touch, #unregister

Constructor Details

#initialize(subscription_record: SubscriptionRecord, index_record: IndexEntryRecord, shape_index_record: ShapeIndexEntryRecord) ⇒ ActiveRecordStore

Returns a new instance of ActiveRecordStore.



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 90

def initialize(subscription_record: SubscriptionRecord, index_record: IndexEntryRecord, shape_index_record: ShapeIndexEntryRecord)
  @subscription_record = subscription_record
  @index_record = index_record
  @shape_index_record = shape_index_record
  @index_builder = ReverseIndex.new
  @pending_registry = ActiveRegistry.new
  @active_registry = ActiveRegistry.new
  @deferred_index_writes = {}
  @deferred_index_mutex = Mutex.new
  @persistence = ActiveRecordSubscriptionPersistence.new(
    subscription_record: subscription_record,
    index_record: index_record,
    shape_index_record: shape_index_record,
    index_builder: index_builder
  )
  persistent_index = PersistentReverseIndex.new(
    reverse_index: index_builder,
    index_record: index_record,
    shape_index_record: shape_index_record,
    subscription_record: subscription_record
  )
  @reverse_index = LayeredReverseIndex.new(
    active_index: active_registry,
    persistent_index: persistent_index,
    persistent_count: -> { persistence.count },
    store: "active_record",
    pending_index: pending_registry
  )
end

Instance Attribute Details

#reverse_index ⇒ Object (readonly)

Returns the value of attribute reverse_index.



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

def reverse_index
  @reverse_index
end

Class Method Details

.available?(connect: false) ⇒ Boolean

Returns:

  • (Boolean) 


120
121
122
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 120

def self.available?(connect: false)
  schema_errors(connect: connect).empty?
end

.schema_errors(connect: false) ⇒ Object 



124
125
126
127
128
129
130
131
132
133
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 124

def self.schema_errors(connect: false)
  return ["Active Record is not connected"] unless ActiveRecord::Base.connected? || connect

  connection = ActiveRecord::Base.connection
  REQUIRED_SCHEMA.flat_map { |table, columns| schema_errors_for_table(connection, table, columns) }
rescue ActiveRecord::ConnectionNotEstablished, ActiveRecord::NoDatabaseError => error
  [error.message]
rescue ActiveRecord::StatementInvalid => error
  ["database schema could not be inspected: #{error.message}"]
end

Instance Method Details

#activate(id) ⇒ Object 



183
184
185
186
187
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 183

def activate(id)
  with_optional_notification(ACTIVATE_NOTIFICATION, { store: "active_record", subscription_id: id }) do |payload|
    activate_subscription(id, payload: payload)
  end
end

#fetch_for_composition(id) ⇒ Object 



189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 189

def fetch_for_composition(id)
  local = active_registry.fetch(id) || pending_registry.fetch(id)
  return local if local

  subscription, entries = persistence.fetch_with_index_entries(id)
  entries.each do |entry|
    subscription.graph.add_dependency(entry.owner_id, entry.dependency)
  end
  recorder = Runtime::Recorder.new(graph: subscription.graph)
  Subscription.new(
    subscription.id,
    subscription.subscriber_id,
    recorder,
    recorder.graph,
    subscription.metadata
  )
rescue ActiveRecord::RecordNotFound
  raise NotFound, id
end

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



209
210
211
212
213
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 209

def prune_stale!(older_than: stale_threshold, limit: nil)
  stale_ids = persistence.prune_stale!(older_than: older_than, limit: limit)
  active_registry.unregister(stale_ids)
  stale_ids.size
end

#register(subscriber_id:, recorder:, metadata: {}, entries: nil) ⇒ Object 



171
172
173
174
175
176
177
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 171

def register(subscriber_id:, recorder:, metadata: {}, entries: nil)
  subscription = with_optional_notification(REGISTER_NOTIFICATION, { store: "active_record" }) do |payload|
    register_subscription(subscriber_id, recorder, metadata, entries: entries, payload: payload)
  end
  trim_opportunistically
  subscription
end

#reset ⇒ Object 



230
231
232
233
234
235
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 230

def reset
  clear_deferred_index_writes
  pending_registry.reset
  active_registry.reset
  persistence.reset
end

#shutdown ⇒ Object 



179
180
181
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 179

def shutdown
  clear_deferred_index_writes
end

#subscriptions ⇒ Object 



215
216
217
218
219
220
221
222
223
224
225
226
227
228
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 215

def subscriptions
  persistent_count = persistence.count
  in_memory_subscriptions = (active_registry.subscriptions + pending_registry.subscriptions).to_h do |subscription|
    [subscription.id, subscription]
  end
  return in_memory_subscriptions.values if in_memory_subscriptions.size >= persistent_count

  seen_ids = {}
  persisted = persistence.subscriptions.map do |subscription|
    seen_ids[subscription.id] = true
    in_memory_subscriptions.fetch(subscription.id, subscription)
  end
  persisted + in_memory_subscriptions.values.reject { |subscription| seen_ids[subscription.id] }
end

#summary ⇒ Object 



237
238
239
240
241
242
243
244
245
246
247
248
249
 
# File 'lib/upkeep/subscriptions/active_record_store.rb', line 237

def summary
  persistent_count = persistence.count
  pending_count = pending_registry.count
  active_count = active_registry.count
  {
    subscriptions: [persistent_count, active_count + pending_count].max,
    persistent_subscriptions: persistent_count,
    pending_subscriptions: pending_count,
    active_subscriptions: active_count,
    deferred_index_subscriptions: deferred_index_count,
    reverse_index: reverse_index.summary
  }
end