diff --git a/app/controllers/display_ad_events_controller.rb b/app/controllers/display_ad_events_controller.rb index 5a28ffa12..e6e6bddb3 100644 --- a/app/controllers/display_ad_events_controller.rb +++ b/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( diff --git a/app/services/display_ad_event_rollup.rb b/app/services/display_ad_event_rollup.rb new file mode 100644 index 000000000..a43c7c8f6 --- /dev/null +++ b/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 diff --git a/app/workers/display_ad_event_rollup_worker.rb b/app/workers/display_ad_event_rollup_worker.rb new file mode 100644 index 000000000..cd4c05155 --- /dev/null +++ b/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 diff --git a/config/schedule.yml b/config/schedule.yml index 2577f0a75..7316e7b69 100644 --- a/config/schedule.yml +++ b/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" diff --git a/db/migrate/20220927091957_add_counts_for_to_display_ad_events.rb b/db/migrate/20220927091957_add_counts_for_to_display_ad_events.rb new file mode 100644 index 000000000..0bf6c9874 --- /dev/null +++ b/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 diff --git a/db/schema.rb b/db/schema.rb index 116cfe51a..58e2cc91a 100644 --- a/db/schema.rb +++ b/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 diff --git a/spec/services/display_ad_event_rollup_spec.rb b/spec/services/display_ad_event_rollup_spec.rb new file mode 100644 index 000000000..997d7437c --- /dev/null +++ b/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 diff --git a/spec/workers/display_ad_event_rollup_worker_spec.rb b/spec/workers/display_ad_event_rollup_worker_spec.rb new file mode 100644 index 000000000..60cb0dd08 --- /dev/null +++ b/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