Replays events for projects that belong to selected groups to replicate

app/models/geo/event_log.rb

@@ -13,5 +13,15 @@ module Geo
     belongs_to :repository_renamed_event,
       class_name: 'Geo::RepositoryRenamedEvent',
       foreign_key: :repository_renamed_event_id
+
+    def event
+      repository_updated_event ||
+      repository_deleted_event ||
+      repository_renamed_event
+    end
+
+    def project_id
+      event&.project_id
+    end
   end
 end

lib/gitlab/geo/log_cursor/daemon.rb

@@ -61,6 +61,8 @@ module Gitlab
 
         def handle_events(batch)
           batch.each do |event|
+            next unless can_replay?(event)
+
             # Update repository
             if event.repository_updated_event
               handle_repository_update(event.repository_updated_event)
@@ -88,6 +90,12 @@ module Gitlab
           @exit = true
         end
 
+        def can_replay?(event)
+          return true if Gitlab::Geo.current_node.project_ids.nil?
+
+          Gitlab::Geo.current_node.project_ids.include?(event.project_id)
+        end
+
         def handle_repository_update(updated_event)
           registry = ::Geo::ProjectRegistry.find_or_initialize_by(project_id: updated_event.project_id)
 

spec/factories/geo/event_log.rb

 FactoryGirl.define do
   factory :geo_event_log, class: Geo::EventLog do
     trait :updated_event do
-      repository_updated_event factory: :geo_repository_update_event
+      repository_updated_event factory: :geo_repository_updated_event
     end
 
     trait :deleted_event do
-      repository_deleted_event factory: :geo_repository_delete_event
+      repository_deleted_event factory: :geo_repository_deleted_event
     end
   end
 
-  factory :geo_repository_update_event, class: Geo::RepositoryUpdatedEvent do
+  factory :geo_repository_updated_event, class: Geo::RepositoryUpdatedEvent do
     source 0
     branches_affected 0
     tags_affected 0
     project
   end
 
-  factory :geo_repository_delete_event, class: Geo::RepositoryDeletedEvent do
+  factory :geo_repository_deleted_event, class: Geo::RepositoryDeletedEvent do
     project
 
     repository_storage_name { project.repository_storage }
@@ -24,4 +24,8 @@ FactoryGirl.define do
     deleted_path { project.path_with_namespace }
     deleted_project_name { project.name }
   end
+
+  factory :geo_repository_renamed_event, class: Geo::RepositoryRenamedEvent do
+    project
+  end
 end

spec/lib/gitlab/geo/log_cursor/daemon_spec.rb

@@ -21,12 +21,14 @@ describe Gitlab::Geo::LogCursor::Daemon do
       end
     end
 
-    context 'when processing a repository updated event' do
+    context 'when replaying a repository updated event' do
+      let!(:geo_node) { create(:geo_node) }
       let(:event_log) { create(:geo_event_log, :updated_event) }
       let!(:event_log_state) { create(:geo_event_log_state, event_id: event_log.id - 1) }
       let(:repository_updated_event) { event_log.repository_updated_event }
 
       before do
+        allow(Gitlab::Geo).to receive(:current_node).and_return(geo_node)
         allow(subject).to receive(:exit?).and_return(false, true)
       end
 
@@ -53,7 +55,7 @@ describe Gitlab::Geo::LogCursor::Daemon do
       end
     end
 
-    context 'when processing a repository deleted event' do
+    context 'when replaying a repository deleted event' do
       let(:event_log) { create(:geo_event_log, :deleted_event) }
       let(:project) { event_log.repository_deleted_event.project }
       let!(:event_log_state) { create(:geo_event_log_state, event_id: event_log.id - 1) }
@@ -79,5 +81,31 @@ describe Gitlab::Geo::LogCursor::Daemon do
         subject.run!
       end
     end
+
+    context 'when node have group restrictions' do
+      let(:geo_node) { create(:geo_node) }
+      let(:group) { create(:group) }
+      let(:project) { create(:empty_project, group: group) }
+      let(:repository_updated_event) { create(:geo_repository_updated_event, project: project) }
+      let(:event_log) { create(:geo_event_log, repository_updated_event: repository_updated_event) }
+      let!(:event_log_state) { create(:geo_event_log_state, event_id: event_log.id - 1) }
+
+      before do
+        allow(Gitlab::Geo).to receive(:current_node).and_return(geo_node)
+        allow(subject).to receive(:exit?).and_return(false, true)
+      end
+
+      it 'replays events for projects that belong to selected groups to replicate' do
+        geo_node.update_attribute(:groups, [group])
+
+        expect { subject.run! }.to change(Geo::ProjectRegistry, :count).by(1)
+      end
+
+      it 'does not replay events for projects that do not belong to selected groups to replicate' do
+        geo_node.update_attribute(:groups, [create(:group)])
+
+        expect { subject.run! }.not_to change(Geo::ProjectRegistry, :count)
+      end
+    end
   end
 end

spec/models/geo/event_log_spec.rb

@@ -3,6 +3,47 @@ require 'spec_helper'
 RSpec.describe Geo::EventLog, type: :model do
   describe 'relationships' do
     it { is_expected.to belong_to(:repository_updated_event).class_name('Geo::RepositoryUpdatedEvent').with_foreign_key('repository_updated_event_id') }
+    it { is_expected.to belong_to(:repository_deleted_event).class_name('Geo::RepositoryDeletedEvent').with_foreign_key('repository_deleted_event_id') }
     it { is_expected.to belong_to(:repository_renamed_event).class_name('Geo::RepositoryRenamedEvent').with_foreign_key('repository_renamed_event_id') }
   end
+
+  describe '#event' do
+    it 'returns nil when having no event associated' do
+      expect(subject.event).to be_nil
+    end
+
+    it 'returns repository_updated_event when set' do
+      repository_updated_event = build(:geo_repository_updated_event)
+      subject.repository_updated_event = repository_updated_event
+
+      expect(subject.event).to eq repository_updated_event
+    end
+
+    it 'returns repository_deleted_event when set' do
+      repository_deleted_event = build(:geo_repository_deleted_event)
+      subject.repository_deleted_event = repository_deleted_event
+
+      expect(subject.event).to eq repository_deleted_event
+    end
+
+    it 'returns repository_renamed_event when set' do
+      repository_renamed_event = build(:geo_repository_renamed_event)
+      subject.repository_renamed_event = repository_renamed_event
+
+      expect(subject.event).to eq repository_renamed_event
+    end
+  end
+
+  describe '#project_id' do
+    it 'returns nil when having no event associated' do
+      expect(subject.project_id).to be_nil
+    end
+
+    it 'returns event#project_id when an event is present' do
+      repository_updated_event = build(:geo_repository_updated_event)
+      subject.repository_updated_event = repository_updated_event
+
+      expect(subject.project_id).to eq repository_updated_event.project_id
+    end
+  end
 end