Move relevant reductions to extract attr_accessor (#636)

* Remove user_ids from prepared reductions

* Move relevant reductions to extract attr_accessor for external reducers

* SORRY FOR YELLING

app/models/extract.rb

@@ -16,6 +16,8 @@ class Extract < ApplicationRecord
     field :updatedAt, !Types::TimeType, property: :updated_at
   end
 
+  attr_accessor :relevant_reduction
+
   belongs_to :workflow, counter_cache: true
   belongs_to :subject
   has_and_belongs_to_many_with_deferred_save :subject_reduction

app/models/reducer.rb

@@ -62,9 +62,12 @@ def process(extract_fetcher, reduction_fetcher, relevant_reductions=[])
         reduction = get_reduction(reduction_fetcher, group_key)
         extracts = filter_extracts(grouped, reduction)
 
+        # Set relevant reduction on each extract if required by external reducer
+        augmented_extracts = add_relevant_reductions(extracts, relevant_reductions)
+
         # relevant_reductions are any previously reduced user or subject reductions
         # that are required by this reducer to properly calculate
-        reduce_into(extracts, reduction, relevant_reductions).tap do |r|
+        reduce_into(augmented_extracts, reduction).tap do |r|
           r.expired = false
 
           # note that because we use deferred associations, this won't actually hit the database
@@ -95,7 +98,7 @@ def associate_extracts(reduction, extracts)
     end
   end
 
-  def reduce_into(extracts, reduction, relevant_reductions)
+  def reduce_into(extracts, reduction)
     raise NotImplementedError
   end
 
@@ -121,4 +124,16 @@ def grouping
 
   def nilify_empty_fields
   end
+
+  def add_relevant_reductions(extracts, relevant_reductions)
+    extracts.map do |ex|
+      ex.relevant_reduction = case topic
+                              when 0, "reduce_by_subject"
+                                relevant_reductions.find { |rr| rr.user_id == ex.user_id }
+                              when 1, "reduce_by_user"
+                                relevant_reductions.find { |rr| rr.subject_id == ex.subject_id }
+                              end
+    end
+    extracts
+  end
 end

app/models/reducers/consensus_reducer.rb

@@ -2,7 +2,7 @@ module Reducers
   class ConsensusReducer < Reducer
     config_field :ignore_empty_extracts, default: false
 
-    def reduce_into(extractions, reduction, _relevant_reductions=[])
+    def reduce_into(extractions, reduction)
       store_value = reduction.store || {}
       counter = CountingHash.new(store_value)
 

app/models/reducers/count_reducer.rb

@@ -7,7 +7,7 @@
 #
 module Reducers
   class CountReducer < Reducer
-    def reduce_into(extracts, reduction, _relevant_reductions=[])
+    def reduce_into(extracts, reduction)
       data = reduction.data || {}
 
       classifications_count = data.fetch("classifications", 0)

app/models/reducers/external_reducer.rb

@@ -22,12 +22,12 @@ class ExternalReducerFailed < StandardError; end
       end
     end
 
-    def reduce_into(extractions, reduction, relevant_reductions=[])
+    def reduce_into(extractions, reduction)
       if url
         response = if default_reduction?
           RestClient.post(url, extractions.to_json, {content_type: :json, accept: :json})
         elsif running_reduction?
-          RestClient.post(url, { extracts: extractions, store: reduction.store, relevant_reductions: relevant_reductions }.to_json, {content_type: :json, accept: :json})
+          RestClient.post(url, { extracts: extractions, store: reduction.store }.to_json, {content_type: :json, accept: :json})
         else
           raise StandardError.new("Impossible reducer configuration #{id}")
         end

app/models/reducers/first_extract_reducer.rb

@@ -6,7 +6,7 @@
 #
 module Reducers
   class FirstExtractReducer < Reducer
-    def reduce_into(extractions, reduction, _relevant_reductions=[])
+    def reduce_into(extractions, reduction)
       reduction.tap do |r|
         r.data = if reduction.data.blank? then (extractions&.fetch(0, nil)&.data || {}) else reduction.data end
       end

app/models/reducers/placeholder_reducer.rb

@@ -1,6 +1,6 @@
 module Reducers
   class PlaceholderReducer < Reducer
-    def reduce_into(extracts, reduction, _relevant_reductions=[])
+    def reduce_into(extracts, reduction)
       reduction.tap do |r|
         r.data = nil
       end

app/models/reducers/sqs_reducer.rb

@@ -9,7 +9,7 @@ class SqsReducer < Reducer
       end
     end
 
-    def reduce_into(extracts, reduction, _relevant_reductions=[])
+    def reduce_into(extracts, reduction)
       extracts.map do |extract|
         {
           message_body: prepare_extract(extract).to_json,

app/models/reducers/stats_reducer.rb

@@ -1,6 +1,6 @@
 module Reducers
   class StatsReducer < Reducer
-    def reduce_into(extractions, reduction, _relevant_reductions=[])
+    def reduce_into(extractions, reduction)
       data = reduction.data || {}
 
       reduction.tap do |r|

app/models/reducers/summary_statistics_reducer.rb

@@ -68,7 +68,7 @@ def operations
     end
 
 
-    def reduce_into(extracts, reduction, _relevant_reductions=[])
+    def reduce_into(extracts, reduction)
       @old_store = reduction.store || {}
       @new_store = {}
 

app/models/reducers/unique_count_reducer.rb

@@ -2,7 +2,7 @@ module Reducers
   class UniqueCountReducer < Reducer
     config_field :field
 
-    def reduce_into(extracts, reduction, _relevant_reductions=[])
+    def reduce_into(extracts, reduction)
       store = reduction.store || {}
       store["items"] = [] unless store.key? "items"
 

app/models/subject_reduction.rb

@@ -23,7 +23,6 @@ def prepare
       id: id,
       reducible: { id: reducible_id, type: reducible_type },
       data: data,
-      user_ids: extracts.pluck(:user_id),
       subject: subject.attributes,
       created_at: created_at,
       updated_at: updated_at

spec/controllers/user_reductions_controller_spec.rb

@@ -26,8 +26,6 @@
       response = get :index, params: { workflow_id: workflow.id, user_id: user1_id }
       results = JSON.parse(response.body)
 
-      # binding.pry
-
       # there are four total reductions but only two belong to this user in this workflow
       expect(response).to be_successful
       expect(results.size).to be(2)

spec/models/reducer_spec.rb

@@ -232,7 +232,62 @@ def reduce_into(extracts, reductions=nil)
       allow(subject_reduction_double).to receive(:expired=)
 
       running_reducer.process(extract_fetcher, reduction_fetcher)
-      expect(running_reducer).to have_received(:reduce_into).with([extract2], subject_reduction_double, [])
+      expect(running_reducer).to have_received(:reduce_into).with([extract2], subject_reduction_double)
+    end
+  end
+
+  describe "relevant reductions" do
+    let(:workflow) { create(:workflow) }
+
+    it 'include user reductions' do
+      subject_reducer = create(:stats_reducer,
+                                reducible: workflow,
+                                topic: Reducer.topics[:reduce_by_subject],
+                                config: {user_reducer_keys: "skillz"}
+                              )
+      new_extracts = [
+        build(:extract, classification_id: 1, subject_id: 1234, user_id: 1, data: {x: 1, y: 2}),
+        build(:extract, classification_id: 2, subject_id: 1234, user_id: 2, data: {x: 2, y: 2}),
+        build(:extract, classification_id: 3, subject_id: 1234, user_id: 3, data: {x: 3, y: 1})
+      ]
+
+      reductions = [
+        create(:user_reduction, data: {skill: 15}, user_id: 1, reducible: workflow, reducer_key: 'skillz'),
+        create(:user_reduction, data: {skill: 22}, user_id: 2, reducible: workflow, reducer_key: 'skillz')
+      ]
+
+      augmented_extracts = subject_reducer.add_relevant_reductions(new_extracts, reductions)
+
+      expect(augmented_extracts[0]).to have_attributes(relevant_reduction: reductions[0])
+      expect(augmented_extracts[1]).to have_attributes(relevant_reduction: reductions[1])
+      expect(augmented_extracts[2]).to have_attributes(relevant_reduction: nil)
+    end
+
+    it 'include subject reductions' do
+      subjects = create_list(:subject, 2)
+      user_reducer = create(:stats_reducer,
+                             reducible: workflow,
+                             topic: Reducer.topics[:reduce_by_user],
+                             reduction_mode: Reducer.reduction_modes[:running_reduction],
+                             config: {user_reducer_keys: "difficulty"}
+                           )
+
+      new_extracts = [
+        build(:extract, workflow_id: workflow.id, subject_id: subjects[0].id, user_id: 1, data: { feedback: {} } ),
+        build(:extract, workflow_id: workflow.id, subject_id: subjects[1].id, user_id: 1, data: { feedback: {} } ),
+        build(:extract, workflow_id: workflow.id, subject_id: 999, user_id: 1, data: { feedback: {} } ),
+      ]
+
+      reductions = [
+        create(:subject_reduction, data: {difficulty: [0.7, 0.3, 0.1] }, subject_id: subjects[0].id, reducible: workflow, reducer_key: 'difficulty'),
+        create(:subject_reduction, data: {difficulty: [0.4, 0.2, 0.8] }, subject_id: subjects[1].id, reducible: workflow, reducer_key: 'difficulty')
+      ]
+
+      augmented_extracts = user_reducer.add_relevant_reductions(new_extracts, reductions)
+
+      expect(augmented_extracts[0]).to have_attributes(relevant_reduction: reductions[0])
+      expect(augmented_extracts[1]).to have_attributes(relevant_reduction: reductions[1])
+      expect(augmented_extracts[2]).to have_attributes(relevant_reduction: nil)
     end
   end
 end

spec/models/reducers/external_reducer_spec.rb

@@ -3,6 +3,7 @@
 describe Reducers::ExternalReducer do
   let(:valid_url){ "https://example.org/post/extracts/here" }
   let(:workflow) { create :workflow }
+  let(:subject) { create :subject}
   let(:reducer) { described_class.new(
     key: 'external',
     reducible_id: workflow.id,
@@ -12,8 +13,8 @@
 
   let(:extracts) {
     [
-      Extract.new(data: {"foo" => "bar"}, workflow: workflow),
-      Extract.new(data: {"foo" => "baz"}, workflow: workflow)
+      Extract.new(data: {"foo" => "bar"}, workflow: workflow, subject: subject, user_id: 123),
+      Extract.new(data: {"foo" => "baz"}, workflow: workflow, subject: subject, user_id: 123)
     ]
   }
 
@@ -83,8 +84,7 @@
 
     let(:request_data){{
       extracts: extracts,
-      store: running_reduction.store,
-      relevant_reductions: []
+      store: running_reduction.store
     }}
 
     it 'sends the extracts and the store' do
@@ -113,43 +113,5 @@
       reducer.reduce_into(extracts, running_reduction)
       expect(running_reduction.store).to have_key('bar')
     end
-
-    it 'includes relevant user reductions' do
-      relevant_reduction = create :user_reduction, data: {skill: 15}, user_id: 111, reducible: workflow, reducer_key: 'skillz'
-      running_reducer.user_reducer_keys = "skillz"
-
-      request_data[:relevant_reductions] = [relevant_reduction]
-
-      stub_request(:post, valid_url)
-        .with(:headers => {'Accept'=>'application/json',
-                          'Content-Type'=>'application/json',
-                          'Host'=>'example.org'})
-        .to_return(:status => 200, :body => request_data.to_json, :headers => {})
-
-      reducer.reduce_into(extracts, running_reduction, [relevant_reduction])
-      expect(a_request(:post, valid_url)
-              .with(body: request_data.to_json))
-        .to have_been_made.once
-    end
-
-    it 'includes relevant subject reductions' do
-      subject = create(:subject)
-      relevant_reduction = create :subject_reduction, data: {probability: 0.94}, subject_id: subject.id, reducible: workflow, reducer_key: 'skillz'
-
-      running_reducer.subject_reducer_keys = "skillz"
-
-      request_data[:relevant_reductions] = [relevant_reduction]
-
-      stub_request(:post, valid_url)
-        .with(:headers => {'Accept'=>'application/json',
-                          'Content-Type'=>'application/json',
-                          'Host'=>'example.org'})
-        .to_return(:status => 200, :body => request_data.to_json, :headers => {})
-
-      reducer.reduce_into(extracts, running_reduction, [relevant_reduction])
-      expect(a_request(:post, valid_url)
-              .with(body: request_data.to_json))
-        .to have_been_made.once
-    end
   end
-end
+end

spec/models/runs_reducers_spec.rb

@@ -145,6 +145,30 @@
     expect(UserReduction.first.data).to eq({"LN" => 2})
   end
 
+  it 'includes relevant reductions' do
+    subject = create :subject
+    reducible = create :workflow
+    reducer = create(:stats_reducer,
+                      topic: Reducer.topics[:reduce_by_subject],
+                      reducible: reducible,
+                      config: {user_reducer_keys: "testing"}
+                    )
+    runner = described_class.new(reducible, [reducer])
+
+    extracts = [
+      create(:extract, extractor_key: 's', workflow_id: reducible.id, user_id: 1, subject_id: subject.id, data: {}),
+      create(:extract, extractor_key: 's', workflow_id: reducible.id, user_id: 2, subject_id: subject.id, data: {})
+    ]
+
+    reductions = [
+      create(:user_reduction, data: {skill: 15}, user_id: 1, reducible: reducible, reducer_key: 'testing'),
+      create(:user_reduction, data: {skill: 22}, user_id: 2, reducible: reducible, reducer_key: 'testing')
+    ]
+
+    expect(reducer).to receive(:process).with(any_args, reductions).and_call_original
+    runner.reduce(subject.id, nil)
+  end
+
   context "reducing by project" do
     let(:project) { create :project }
     let(:reducer) { create(:stats_reducer, key: 's', reducible: project) }

spec/models/subject_reduction_spec.rb

@@ -13,7 +13,6 @@
     reduction = create(:subject_reduction, extracts: [extract1, extract2, extract3])
 
     prepared = reduction.prepare
-    expect(prepared[:user_ids]).to include(extract1.user_id, extract2.user_id, extract3.user_id)
     expect(prepared[:subject]).to eq(reduction.subject.attributes)
   end
 end