# frozen_string_literal: true class Analytics::CycleAnalytics::Aggregation < ApplicationRecord include FromUnion belongs_to :group, optional: false validates :incremental_runtimes_in_seconds, :incremental_processed_records, :last_full_run_runtimes_in_seconds, :last_full_run_processed_records, presence: true, length: { maximum: 10 }, allow_blank: true scope :priority_order, -> (column_to_sort = :last_incremental_run_at) { order(arel_table[column_to_sort].asc.nulls_first) } scope :enabled, -> { where('enabled IS TRUE') } def estimated_next_run_at return unless enabled return if last_incremental_run_at.nil? estimation = duration_until_the_next_aggregation_job + average_aggregation_duration + (last_incremental_run_at - earliest_last_run_at) estimation < 1 ? nil : estimation.from_now end def self.safe_create_for_group(group) top_level_group = group.root_ancestor aggregation = find_by(group_id: top_level_group.id) return aggregation if aggregation.present? insert({ group_id: top_level_group.id }, unique_by: :group_id) find_by(group_id: top_level_group.id) end private # The aggregation job is scheduled every 10 minutes: */10 * * * * def duration_until_the_next_aggregation_job (10 - (DateTime.current.minute % 10)).minutes.seconds end def average_aggregation_duration return 0.seconds if incremental_runtimes_in_seconds.empty? average = incremental_runtimes_in_seconds.sum.fdiv(incremental_runtimes_in_seconds.size) average.seconds end def earliest_last_run_at max = self.class.select(:last_incremental_run_at) .where(enabled: true) .where.not(last_incremental_run_at: nil) .priority_order .limit(1) .to_sql connection.select_value("(#{max})") end def self.load_batch(last_run_at, column_to_query = :last_incremental_run_at, batch_size = 100) last_run_at_not_set = Analytics::CycleAnalytics::Aggregation .enabled .where(column_to_query => nil) .priority_order(column_to_query) .limit(batch_size) last_run_at_before = Analytics::CycleAnalytics::Aggregation .enabled .where(arel_table[column_to_query].lt(last_run_at)) .priority_order(column_to_query) .limit(batch_size) Analytics::CycleAnalytics::Aggregation .from_union([last_run_at_not_set, last_run_at_before], remove_order: false, remove_duplicates: false) .limit(batch_size) end end