Implement PageViewRollup (#20670)

* Refactor

* Create PageViewRollup WIP

* Create PageViewRollupWorker and specs

* Add spec WIP

* Write spec

* Refactor

* Add spec

* Prevent retry

* Update baseline date

* Improve specs

* Update ATTRIBUTES_

* Update spec again

app/services/page_view_rollup.rb

@@ -0,0 +1,92 @@
+class PageViewRollup
+  ATTRIBUTES_PRESERVED = %i[article_id created_at user_id].freeze
+  ATTRIBUTES_DESTROYED = %i[id domain path referrer updated_at user_agent counts_for_number_of_views
+                            time_tracked_in_seconds].freeze
+
+  class ViewAggregator
+    Compact = Struct.new(:views, :article_id, :user_id) do
+      def to_h
+        super.except(:views).merge({
+                                     counts_for_number_of_views: views.sum(&:counts_for_number_of_views),
+                                     time_tracked_in_seconds: views.sum(&:time_tracked_in_seconds)
+                                   })
+      end
+    end
+
+    def initialize
+      @aggregator = Hash.new do |level1, article_id|
+        level1[article_id] = Hash.new do |level2, user_id|
+          level2[user_id] = []
+        end
+      end
+    end
+
+    def <<(view)
+      @aggregator[view.article_id][view.user_id] << view
+    end
+
+    def each
+      @aggregator.each_pair do |article_id, grouped_by_article_id|
+        grouped_by_article_id.each_pair do |user_id, views|
+          next unless views.size > 1
+
+          yield Compact.new(views, article_id, user_id)
+        end
+      end
+    end
+
+    private
+
+    attr_reader :aggregator
+  end
+
+  def self.rollup(date, relation: PageView)
+    new(relation: relation).rollup(date.to_datetime)
+  end
+
+  def initialize(relation:)
+    @relation = relation
+  end
+
+  attr_reader :relation
+
+  def rollup(date)
+    created = []
+
+    (0..23).each do |hour|
+      start_hour = date.change(hour: hour)
+      end_hour = date.change(hour: hour + 1)
+      rows = relation.where(user_id: nil, created_at: start_hour...end_hour)
+      aggregate_into_groups(rows).each do |compacted_views|
+        created << compact_records(start_hour, compacted_views)
+      end
+    end
+
+    created
+  end
+
+  private
+
+  def aggregate_into_groups(rows)
+    aggregator = ViewAggregator.new
+    rows.in_batches.each_record do |event|
+      aggregator << event
+    end
+
+    aggregator
+  end
+
+  def compact_records(date, compacted)
+    result = nil
+
+    relation.transaction do
+      result = relation.create!(compacted.to_h) do |event|
+        event.created_at = date
+      end
+
+      relation.where(id: compacted.views).delete_all
+    end
+
+    result
+  end
+end

app/workers/articles/update_page_views_worker.rb

@@ -10,8 +10,7 @@ module Articles
     # @see Articles::PageViewUpdater
     def perform(create_params)
       article = Article.find_by(id: create_params["article_id"])
-      return unless article
-      return unless article.published?
+      return unless article&.published?
       return if create_params[:user_id] && article.user_id == create_params[:user_id]
 
       PageView.create!(create_params)

app/workers/page_view_rollup_worker.rb

@@ -0,0 +1,10 @@
+class PageViewRollupWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: :low_priority, retry: false
+
+  def perform
+    one_year_ago = 1.year.ago
+    PageViewRollup.rollup one_year_ago
+  end
+end

spec/services/page_view_rollup_spec.rb

@@ -0,0 +1,75 @@
+require "rails_helper"
+
+RSpec.describe PageViewRollup, type: :service do
+  let(:article1) { create(:article) }
+  let(:article2) { create(:article) }
+  let(:user1) { create(:user) }
+  let(:user2) { create(:user) }
+
+  def days_ago_as_range(num)
+    (Date.current - num.days).all_day
+  end
+
+  it "fails if new attributes would be lost" do
+    attributes_considered = described_class::ATTRIBUTES_PRESERVED + described_class::ATTRIBUTES_DESTROYED
+    expect(PageView.column_names.map(&:to_sym)).to match_array(attributes_considered)
+  end
+
+  context "when compacting many rows" do
+    it "does not compact signed-in user's views" do
+      user_view1 = create(:page_view, article: article1, user: user1, created_at: 2.days.ago)
+      user_view2 = create(:page_view, article: article1, user: user1, created_at: 2.days.ago)
+      rando_view1 = create(:page_view, article: article1, user: nil, created_at: 2.days.ago)
+      rando_view2 = create(:page_view, article: article1, user: nil, created_at: 2.days.ago)
+
+      expect do
+        described_class.rollup(2.days.ago)
+      end.to change(PageView, :count).from(4).to(3)
+
+      expect(user_view1.reload).to be_persisted
+      expect(user_view2.reload).to be_persisted
+      expect(PageView.where(id: [rando_view1.id, rando_view2.id])).to be_empty
+    end
+
+    it "compacts by the hour" do
+      [[1, 59], [1, 0], [3, 0], [3, 1], [3, 2]].each do |hour, min|
+        create(:page_view, article: article1, user: nil, created_at: 2.days.ago.change(hour: hour, min: min))
+      end
+
+      expect do
+        described_class.rollup(2.days.ago)
+      end.to change(PageView, :count).from(5).to(2)
+
+      results_plucked = PageView.where(created_at: days_ago_as_range(2))
+        .pluck(:article_id, :user_id, :counts_for_number_of_views, :time_tracked_in_seconds)
+
+      expect(results_plucked).to contain_exactly(
+        [article1.id, nil, 2, 30],
+        [article1.id, nil, 3, 45],
+      )
+    end
+
+    it "does not compact views outside of the same hour" do
+      24.times do |hour|
+        create(:page_view, article: article1, user: nil, created_at: 2.days.ago.change(hour: hour))
+      end
+
+      expect do
+        described_class.rollup(2.days.ago)
+      end.not_to change(PageView, :count)
+    end
+  end
+
+  it "only compacts views of the same article" do
+    create(:page_view, article: article1, user: nil, created_at: 2.days.ago)
+    create(:page_view, article: article1, user: nil, created_at: 2.days.ago)
+    create(:page_view, article: article2, user: nil, created_at: 2.days.ago)
+    create(:page_view, article: article2, user: nil, created_at: 2.days.ago)
+
+    expect do
+      described_class.rollup(2.days.ago)
+    end.to change(PageView, :count).from(4).to(2)
+    expect(PageView.pluck(:article_id,
+                          :counts_for_number_of_views)).to contain_exactly([article1.id, 2], [article2.id, 2])
+  end
+end

spec/workers/page_view_rollup_worker_spec.rb

@@ -0,0 +1,21 @@
+require "rails_helper"
+
+RSpec.describe PageViewRollupWorker, type: :worker do
+  subject(:worker) { described_class.new }
+
+  before do
+    allow(PageViewRollup).to receive(:rollup)
+  end
+
+  include_examples "#enqueues_on_correct_queue", "low_priority"
+
+  describe "#perform" do
+    it "rollups five month ago" do
+      Time.freeze do
+        one_year_ago = 1.year.ago
+        worker.perform
+        expect(PageViewRollup).to have_received(:rollup).with(one_year_ago)
+      end
+    end
+  end
+end