CV2-6314: Sync sources_count value between PG & ES (#2278)

* CV2-6314: use cached field for sources_count

* CV2-6314: fix tests

* CV2-6314: delete targets_count field

* CV2-6314: fix tests

* CV2-6314: apply PR comments

Author: melsawy

app/models/concerns/project_media_cached_fields.rb

@@ -107,6 +107,23 @@ def title_or_description_update
       recalculate: :recalculate_suggestions_count,
       update_on: [SIMILARITY_EVENT]
 
+    cached_field :sources_count,
+      start_as: 0,
+      update_es: true,
+      update_pg: true,
+      recalculate: :recalculate_sources_count,
+      update_on: [
+        {
+          model: Relationship,
+          if: proc { |r| r.is_confirmed? },
+          affected_ids: proc { |r| [r.source_id, r.target_id] },
+          events: {
+            save: :recalculate,
+            destroy: :recalculate
+          }
+        }
+      ]
+
     cached_field :related_count,
       start_as: 0,
       update_es: true,
@@ -538,6 +555,10 @@ def recalculate_suggestions_count
       Relationship.send('suggested').where(source_id: self.id).count
     end
 
+    def recalculate_sources_count
+      Relationship.where(target_id: self.id).where('relationship_type = ?', Relationship.confirmed_type.to_yaml).count
+    end
+
     def recalculate_is_suggested
       Relationship.where('relationship_type = ?', Relationship.suggested_type.to_yaml).where(target_id: self.id).exists?
     end

app/models/concerns/relationship_bulk.rb

@@ -29,6 +29,9 @@ def bulk_update(ids, updates, team)
         }
         self.delay.run_update_callbacks(ids.to_json, extra_options.to_json)
         delete_cached_fields(pm_source.id, relationships.map(&:target_id))
+        # Call the cached field sources_count to update its value, as the field is not called from the UI.
+        ids = [pm_source.id, relationships.map(&:target_id)].flatten
+        ProjectMedia.where(id: ids).each{ |pm| pm.sources_count(true) }
         { source_project_media: pm_source }
       end
     end
@@ -66,7 +69,7 @@ def run_update_callbacks(ids_json, extra_options_json)
       index_alias = CheckElasticSearchModel.get_index_alias
       es_body = []
       versions = []
-      callbacks = [:reset_counters, :update_counters, :propagate_inversion]
+      callbacks = [:turn_off_unmatched_field, :move_explainers_to_source, :apply_status_to_target]
       target_ids = []
       Relationship.where(id: ids, source_id: extra_options['source_id']).find_each do |r|
         target_ids << r.target_id
@@ -106,11 +109,8 @@ def run_update_callbacks(ids_json, extra_options_json)
         # Send report if needed
         if extra_options['action'] == 'accept'
           begin Relationship.smooch_send_report(r.id) rescue nil end
-          Relationship.replicate_status_to_children(r.source_id, whodunnit, team_id)
         end
       end
-      # Update un-matched field
-      ProjectMedia.where(id: target_ids).update_all(unmatched: 0)
       # Update ES docs
       $repository.client.bulk body: es_body unless es_body.blank?
       # Import versions
@@ -128,9 +128,6 @@ def run_destroy_callbacks(relationship_target_json, extra_options_json)
       source = ProjectMedia.find_by_id(source_id)
       version_metadata = nil
       unless source.nil?
-        source.skip_check_ability = true
-        source.targets_count = Relationship.where(source_id: source.id).where('relationship_type = ? OR relationship_type = ?', Relationship.confirmed_type.to_yaml, Relationship.suggested_type.to_yaml).count
-        source.save!
         # Get version metadata
         version_metadata = {
           source: {
@@ -141,13 +138,8 @@ def run_destroy_callbacks(relationship_target_json, extra_options_json)
           }
         }.to_json
       end
-
-      ProjectMedia.where(id: target_ids).find_each do |target|
-        target.skip_check_ability = true
-        target.unmatched = 1
-        target.sources_count = Relationship.where(target_id: target.id).where('relationship_type = ?', Relationship.confirmed_type.to_yaml).count
-        target.save!
-      end
+      # Update un-matched field
+      ProjectMedia.where(id: target_ids).update_all(unmatched: 1)
       # Update ES
       options = {
         index: CheckElasticSearchModel.get_index_alias,

app/models/relationship.rb

@@ -22,15 +22,12 @@ class Relationship < ApplicationRecord
   before_create :point_targets_to_new_source
   before_create :destroy_same_suggested_item, if: proc { |r| r.is_confirmed? }
   after_create :move_to_same_project_as_main, prepend: true
-  after_create :update_counters, prepend: true
-  after_update :reset_counters, prepend: true
   after_update :propagate_inversion
   after_save :turn_off_unmatched_field, if: proc { |r| r.is_confirmed? || r.is_suggested? }
   after_save :move_explainers_to_source, :apply_status_to_target, if: proc { |r| r.is_confirmed? }
   before_destroy :archive_detach_to_list
-  after_destroy :update_counters, prepend: true
   after_destroy :turn_on_unmatched_field, if: proc { |r| r.is_confirmed? || r.is_suggested? }
-  after_commit :update_counter_and_elasticsearch, on: [:create, :update]
+  after_commit :update_elasticsearch_data, on: [:create, :update]
   after_commit :destroy_elasticsearch_relation, on: :destroy
 
   has_paper_trail on: [:create, :update, :destroy], if: proc { |x| User.current.present? && !x.is_being_copied? }, versions: { class_name: 'Version' }
@@ -99,13 +96,6 @@ def self.default_type
     { source: 'parent', target: 'child' }
   end
 
-  def self.propagate_inversion(ids, source_id)
-    Relationship.where(id: ids.split(',')).each do |r|
-      r.source_id = source_id
-      r.send(:reset_counters)
-    end
-  end
-
   def is_being_copied?
     (self.source && self.source.is_being_copied) || self.is_being_copied
   end
@@ -138,22 +128,6 @@ def is_being_confirmed?
     self.send(method).to_json == Relationship.suggested_type.to_json && self.relationship_type.to_json == Relationship.confirmed_type.to_json
   end
 
-  def update_counters
-    return if self.is_default?
-    unless self.target.nil?
-      target = self.target
-      target.skip_check_ability = true
-      target.sources_count = Relationship.where(target_id: target.id).where('relationship_type = ?', Relationship.confirmed_type.to_yaml).count
-      target.save!
-    end
-    unless self.source.nil?
-      source = self.source
-      source.skip_check_ability = true
-      source.targets_count = Relationship.where(source_id: source.id).where('relationship_type = ? OR relationship_type = ?', Relationship.confirmed_type.to_yaml, Relationship.suggested_type.to_yaml).count
-      source.save!
-    end
-  end
-
   def create_or_update_parent_id
     self.source_id
   end
@@ -293,15 +267,6 @@ def relationship_type_is_valid
     end
   end
 
-  def reset_counters
-    if (self.source_id_before_last_save && self.source_id_before_last_save != self.source_id) || (self.target_id_before_last_save && self.target_id_before_last_save != self.target_id)
-      previous = Relationship.new(source_id: self.source_id_before_last_save, target_id: self.target_id_before_last_save)
-      previous.update_counters
-      current = Relationship.new(source_id: self.source_id, target_id: self.target_id)
-      current.update_counters
-    end
-  end
-
   def propagate_inversion
     if self.source_id_before_last_save == self.target_id && self.target_id_before_last_save == self.source_id
       ids = Relationship.where(source_id: self.target_id).map(&:id).join(',')
@@ -322,7 +287,6 @@ def propagate_inversion
       end
       self.source&.clear_cached_fields
       self.target&.clear_cached_fields
-      Relationship.delay_for(1.second).propagate_inversion(ids, self.source_id)
     end
   end
 
@@ -382,8 +346,7 @@ def turn_on_unmatched_field
     set_unmatched_field(1)
   end
 
-  def update_counter_and_elasticsearch
-    self.update_counters
+  def update_elasticsearch_data
     self.update_elasticsearch_parent
   end
 

db/migrate/20250408164959_remove_targets_count_from_project_media.rb

@@ -0,0 +1,5 @@
+class RemoveTargetsCountFromProjectMedia < ActiveRecord::Migration[6.1]
+  def change
+    remove_column :project_medias, :targets_count
+  end
+end

db/schema.rb

@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 2025_03_28_140809) do
+ActiveRecord::Schema.define(version: 2025_04_08_164959) do
 
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
@@ -349,7 +349,7 @@
     t.string "tags", default: [], array: true
     t.boolean "trashed", default: false
     t.bigint "author_id"
-    t.integer "channel"
+    t.integer "channel", null: false
     t.index "date_trunc('day'::text, created_at)", name: "explainer_created_at_day"
     t.index ["author_id"], name: "index_explainers_on_author_id"
     t.index ["channel"], name: "index_explainers_on_channel"
@@ -376,7 +376,7 @@
     t.boolean "imported", default: false
     t.boolean "trashed", default: false
     t.bigint "author_id"
-    t.integer "channel"
+    t.integer "channel", null: false
     t.index "date_trunc('day'::text, created_at)", name: "fact_check_created_at_day"
     t.index ["author_id"], name: "index_fact_checks_on_author_id"
     t.index ["channel"], name: "index_fact_checks_on_channel"
@@ -560,7 +560,6 @@
     t.boolean "read", default: false, null: false
     t.integer "sources_count", default: 0, null: false
     t.integer "archived", default: 0
-    t.integer "targets_count", default: 0, null: false
     t.integer "last_seen"
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false

test/controllers/elastic_search_5_test.rb

@@ -14,6 +14,7 @@ def setup
   end
 
   test "should match secondary items and show items based on show_similar option" do
+    RequestStore.store[:skip_cached_field_update] = false
     t = create_team
     parent = create_project_media team: t, disable_es_callbacks: false
     child_1 = create_project_media team: t, quote: 'child_media a', disable_es_callbacks: false

test/models/project_media_test.rb

@@ -11,6 +11,7 @@ def setup
 
   test "should query media" do
     setup_elasticsearch
+    RequestStore.store[:skip_cached_field_update] = false
     t = create_team
     pm = create_project_media team: t, disable_es_callbacks: false
     create_project_media team: t, disable_es_callbacks: false

test/models/relationship_2_test.rb

@@ -94,28 +94,18 @@ def teardown
     end
   end
 
-  test "should start with targets count zero" do
-    pm = create_project_media project: @project
-    assert_equal 0, pm.targets_count
-  end
-
   test "should increment and decrement counters when relationship is created or destroyed" do
+    RequestStore.store[:skip_cached_field_update] = false
     s = create_project_media project: @project
     t = create_project_media project: @project
-    assert_equal 0, s.targets_count
     assert_equal 0, s.sources_count
-    assert_equal 0, t.targets_count
     assert_equal 0, t.sources_count
     r = create_relationship source_id: s.id, target_id: t.id, relationship_type: Relationship.confirmed_type
-    assert_equal 1, s.reload.targets_count
     assert_equal 0, s.reload.sources_count
     assert_equal 1, t.reload.sources_count
-    assert_equal 0, t.reload.targets_count
     r.destroy
-    assert_equal 0, s.reload.targets_count
     assert_equal 0, s.reload.sources_count
     assert_equal 0, t.reload.sources_count
-    assert_equal 0, t.reload.targets_count
   end
 
   test "should create related report" do

test/models/relationship_test.rb

@@ -34,6 +34,7 @@ def setup
   end
 
   test "should update sources_count and parent_id for confirmed item" do
+    RequestStore.store[:skip_cached_field_update] = false
     setup_elasticsearch
     t = create_team
     pm_s = create_project_media team: t, disable_es_callbacks: false
@@ -99,16 +100,16 @@ def setup
   end
 
   test "should verify versions and ES for bulk-update" do
+    setup_elasticsearch
+    RequestStore.store[:skip_cached_field_update] = false
+    t = create_team
+    u = create_user
+    create_team_user team: t, user: u, role: 'admin'
     with_versioning do
-      setup_elasticsearch
-      RequestStore.store[:skip_cached_field_update] = false
-      t = create_team
-      u = create_user
-      create_team_user team: t, user: u, role: 'admin'
       with_current_user_and_team(u, t) do
         # Try to create an item with title that trigger a version metadata error(CV2-2910)
         pm_s = create_project_media team: t, quote: "Rahul Gandhi's interaction with Indian?param:test&Journalists Association in London"
-        pm_t1 = create_project_media team: t
+        pm_t1 = create_project_media team: t, disable_es_callbacks: false
         pm_t2 = create_project_media team: t
         pm_t3 = create_project_media team: t
         r1 = create_relationship source_id: pm_s.id, target_id: pm_t1.id, relationship_type: Relationship.suggested_type
@@ -157,18 +158,16 @@ def setup
 
   test "should bulk-reject similar items" do
     RequestStore.store[:skip_cached_field_update] = false
+    setup_elasticsearch
+    t = create_team
+    u = create_user
+    create_team_user team: t, user: u, role: 'admin'
     with_versioning do
-      setup_elasticsearch
-      t = create_team
-      u = create_user
-      p = create_project team: t
-      p2 = create_project team: t
-      create_team_user team: t, user: u, role: 'admin'
       with_current_user_and_team(u, t) do
-        pm_s = create_project_media team: t, project: p
-        pm_t1 = create_project_media team: t, project: p
-        pm_t2 = create_project_media team: t, project: p
-        pm_t3 = create_project_media team: t, project: p
+        pm_s = create_project_media team: t
+        pm_t1 = create_project_media team: t
+        pm_t2 = create_project_media team: t
+        pm_t3 = create_project_media team: t
         r1 = create_relationship source_id: pm_s.id, target_id: pm_t1.id, relationship_type: Relationship.suggested_type
         r2 = create_relationship source_id: pm_s.id, target_id: pm_t2.id, relationship_type: Relationship.suggested_type
         r3 = create_relationship source_id: pm_s.id, target_id: pm_t3.id, relationship_type: Relationship.suggested_type
@@ -275,14 +274,6 @@ def setup
     assert_equal 1, pm2.explainer_items.count
   end
 
-  test "should not attempt to update source count if source does not exist" do
-    r = create_relationship relationship_type: Relationship.confirmed_type
-    r.source.delete
-    assert_nothing_raised do
-      r.reload.send :update_counters
-    end
-  end
-
   test "should cache the name of who created a similar item" do
     RequestStore.store[:skip_cached_field_update] = false
     t = create_team