Rollup service to compact DisplayAdEvent records (#18508)

* Restore DisplayAdEvents.counts_for

* Try a rollup service for DisplayAdEvent

* Schedule daily rollup worker for one month prior

* Refactor aggregator

* More testing for rollup

* Add test for rollup attribute fidelity

* DisplayAd success_rate based on counts_for

app/controllers/display_ad_events_controller.rb

@@ -18,8 +18,8 @@ class DisplayAdEventsController < ApplicationMetalController
     ThrottledCall.perform(:display_ads_data_update, throttle_for: 15.minutes) do
       @display_ad = DisplayAd.find(display_ad_event_params[:display_ad_id])
 
-      num_impressions = @display_ad.display_ad_events.impressions.size
-      num_clicks = @display_ad.display_ad_events.clicks.size
+      num_impressions = @display_ad.display_ad_events.impressions.sum(:counts_for)
+      num_clicks = @display_ad.display_ad_events.clicks.sum(:counts_for)
       rate = num_clicks.to_f / num_impressions
 
       @display_ad.update_columns(

app/services/display_ad_event_rollup.rb

@@ -0,0 +1,92 @@
+class DisplayAdEventRollup
+  ATTRIBUTES_PRESERVED = %i[user_id display_ad_id category context_type created_at].freeze
+  ATTRIBUTES_DESTROYED = %i[id counts_for updated_at].freeze
+
+  class EventAggregator
+    Compact = Struct.new(:events, :user_id, :display_ad_id, :category, :context_type) do
+      def to_h
+        super.except(:events).merge({ counts_for: events.sum(&:counts_for) })
+      end
+    end
+
+    def initialize
+      @aggregator = Hash.new do |level1, user_id|
+        level1[user_id] = Hash.new do |level2, display_ad_id|
+          level2[display_ad_id] = Hash.new do |level3, category|
+            level3[category] = Hash.new do |level4, context_type|
+              level4[context_type] = []
+            end
+          end
+        end
+      end
+    end
+
+    def <<(event)
+      @aggregator[event.user_id][event.display_ad_id][event.category][event.context_type] << event
+    end
+
+    def each
+      @aggregator.each_pair do |user_id, grouped_by_user_id|
+        grouped_by_user_id.each_pair do |display_ad_id, grouped_by_display_ad_id|
+          grouped_by_display_ad_id.each_pair do |category, grouped_by_category|
+            grouped_by_category.each_pair do |context_type, events|
+              next unless events.size > 1
+
+              yield Compact.new(events, user_id, display_ad_id, category, context_type)
+            end
+          end
+        end
+      end
+    end
+
+    private
+
+    attr_reader :aggregator
+  end
+
+  def self.rollup(date, relation: DisplayAdEvent)
+    new(relation: relation).rollup(date)
+  end
+
+  def initialize(relation:)
+    @aggregator = EventAggregator.new
+    @relation = relation
+  end
+
+  attr_reader :aggregator, :relation
+
+  def rollup(date)
+    created = []
+
+    rows = relation.where(created_at: date.all_day)
+    aggregate_into_groups(rows).each do |compacted_events|
+      created << compact_records(date, compacted_events)
+    end
+
+    created
+  end
+
+  private
+
+  def aggregate_into_groups(rows)
+    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.events).delete_all
+    end
+
+    result
+  end
+end

app/workers/display_ad_event_rollup_worker.rb

@@ -0,0 +1,10 @@
+class DisplayAdEventRollupWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: :low_priority
+
+  def perform
+    month_ago = Date.current - 32.days
+    DisplayAdEventRollup.rollup month_ago
+  end
+end

config/schedule.yml

@@ -150,3 +150,7 @@ notify_about_published_articles:
   description: "Sends notifications about the new published articles (runs every minute)"
   cron: "*/5 * * * *"
   class: "Articles::PublishWorker"
+display_ad_event_rollup:
+  description: "Compact rows in the display_ad_events table"
+  cron: "5 3 * * *"
+  class: "DisplayAdEventRollupWorker"

db/migrate/20220927091957_add_counts_for_to_display_ad_events.rb

@@ -0,0 +1,5 @@
+class AddCountsForToDisplayAdEvents < ActiveRecord::Migration[7.0]
+  def change
+    add_column :display_ad_events, :counts_for, :integer, default: 1, null: false
+  end
+end

db/schema.rb

@@ -444,6 +444,7 @@ ActiveRecord::Schema[7.0].define(version: 2022_10_03_103855) do
   create_table "display_ad_events", force: :cascade do |t|
     t.string "category"
     t.string "context_type"
+    t.integer "counts_for", default: 1, null: false
     t.datetime "created_at", precision: nil, null: false
     t.bigint "display_ad_id"
     t.datetime "updated_at", precision: nil, null: false

spec/services/display_ad_event_rollup_spec.rb

@@ -0,0 +1,116 @@
+require "rails_helper"
+
+RSpec.describe DisplayAdEventRollup, type: :service do
+  let(:ad1) { create :display_ad }
+  let(:ad2) { create :display_ad }
+  let(:user1) { create :user }
+  let(:user2) { create :user }
+
+  def override_timestamps
+    DisplayAdEvent.record_timestamps = false
+    yield
+    DisplayAdEvent.record_timestamps = true
+  end
+
+  def days_ago_as_range(num)
+    (Date.current - num.days).all_day
+  end
+
+  it "fails if new attributes would be lost" do
+    attributes_considered = DisplayAdEventRollup::ATTRIBUTES_PRESERVED + DisplayAdEventRollup::ATTRIBUTES_DESTROYED
+    expect(DisplayAdEvent.column_names.map(&:to_sym)).to contain_exactly(*attributes_considered)
+  end
+
+  context "when compacting many rows" do
+    before do
+      override_timestamps do
+        create(:display_ad_event, created_at: Date.current - 2, display_ad: ad1, user_id: nil, updated_at: Date.current)
+        create(:display_ad_event, created_at: Date.current - 2, display_ad: ad1, user_id: nil, updated_at: Date.current)
+        create(:display_ad_event, created_at: Date.current - 2, display_ad: ad1, user_id: nil, updated_at: Date.current)
+
+        create(:display_ad_event, created_at: Date.current - 2, display_ad: ad1, user_id: user1.id,
+                                  updated_at: Date.current)
+        create(:display_ad_event, created_at: Date.current - 2, display_ad: ad2, user_id: nil, updated_at: Date.current)
+      end
+    end
+
+    it "compacts one day's display_ad_events" do
+      expect(DisplayAdEvent.where(created_at: days_ago_as_range(2)).count).to eq(5)
+
+      described_class.rollup(Date.current - 2)
+
+      expectations = [
+        [ad1.id, nil, 3],
+        [ad1.id, user1.id, 1],
+        [ad2.id, nil, 1],
+      ]
+      results_mapped = DisplayAdEvent.where(created_at: days_ago_as_range(2)).map do |event|
+        [event.display_ad_id, event.user_id, event.counts_for]
+      end
+      expect(results_mapped).to contain_exactly(*expectations)
+    end
+  end
+
+  # separate category
+  it "groups by category" do
+    create(:display_ad_event, category: "impression", display_ad: ad1, user_id: nil)
+    create(:display_ad_event, category: "impression", display_ad: ad1, user_id: nil)
+    create(:display_ad_event, category: "impression", display_ad: ad1, user_id: nil)
+    create(:display_ad_event, category: "click", display_ad: ad1, user_id: nil)
+    create(:display_ad_event, category: "click", display_ad: ad1, user_id: nil)
+
+    described_class.rollup(Date.current)
+    results = DisplayAdEvent.where(created_at: Date.current.all_day)
+    by_category = results.index_by { |r| r["category"] }
+    expect(by_category["impression"]["counts_for"]).to eq(3)
+    expect(by_category["click"]["counts_for"]).to eq(2)
+  end
+
+  # separate display_ad_id
+  it "groups by display_ad_id" do
+    create(:display_ad_event, display_ad: ad1, user_id: nil)
+    create(:display_ad_event, display_ad: ad1, user_id: nil)
+    create(:display_ad_event, display_ad: ad1, user_id: nil)
+    create(:display_ad_event, display_ad: ad2, user_id: nil)
+    create(:display_ad_event, display_ad: ad2, user_id: nil)
+
+    described_class.rollup(Date.current)
+    results = DisplayAdEvent.where(created_at: Date.current.all_day)
+    by_ad = results.index_by { |r| r["display_ad_id"] }
+    expect(by_ad[ad1.id]["counts_for"]).to eq(3)
+    expect(by_ad[ad2.id]["counts_for"]).to eq(2)
+  end
+
+  # separate user_id / null
+  it "groups by user_id (including null / logged-out user)" do
+    create(:display_ad_event, display_ad: ad1, user: user1)
+    create(:display_ad_event, display_ad: ad1, user: user2)
+    create(:display_ad_event, display_ad: ad1, user: user2)
+    create(:display_ad_event, display_ad: ad1, user: nil)
+    create(:display_ad_event, display_ad: ad1, user: nil)
+    create(:display_ad_event, display_ad: ad1, user: nil)
+    create(:display_ad_event, display_ad: ad1, user: nil)
+    create(:display_ad_event, display_ad: ad1, user: nil)
+
+    described_class.rollup(Date.current)
+    results = DisplayAdEvent.where(created_at: Date.current.all_day)
+    by_user = results.index_by { |r| r["user_id"] }
+    expect(by_user[user1.id]["counts_for"]).to eq(1)
+    expect(by_user[user2.id]["counts_for"]).to eq(2)
+    expect(by_user[nil]["counts_for"]).to eq(5)
+  end
+
+  # sums counts_for > 1
+  it "counts previously crunched" do
+    create(:display_ad_event, display_ad: ad1, counts_for: 10, user_id: nil)
+    create(:display_ad_event, display_ad: ad1, counts_for: 15, user_id: nil)
+    create(:display_ad_event, display_ad: ad1, user_id: nil)
+    create(:display_ad_event, display_ad: ad1, user_id: nil)
+    create(:display_ad_event, display_ad: ad1, user_id: nil)
+
+    described_class.rollup(Date.current)
+    results = DisplayAdEvent.where(created_at: Date.current.all_day)
+    expect(results.count).to eq(1)
+    expect(results.first.counts_for).to eq(28)
+  end
+end

spec/workers/display_ad_event_rollup_worker_spec.rb

@@ -0,0 +1,19 @@
+require "rails_helper"
+
+RSpec.describe DisplayAdEventRollupWorker, type: :worker do
+  subject(:worker) { described_class.new }
+
+  before do
+    allow(DisplayAdEventRollup).to receive(:rollup)
+  end
+
+  include_examples "#enqueues_on_correct_queue", "low_priority"
+
+  describe "#perform" do
+    it "rollups one month ago" do
+      month_ago = Date.current - 32.days
+      worker.perform
+      expect(DisplayAdEventRollup).to have_received(:rollup).with(month_ago)
+    end
+  end
+end