Added Backfill service for Geo

Geo nodes now can be synced without rsync. The sync will be triggered using our own transport method instead
and it will be triggered from the primary node at the moment a secondary node is created.

app/models/geo_node.rb

@@ -17,6 +17,7 @@ class GeoNode < ActiveRecord::Base
   validates :relative_url_root, length: { minimum: 0, allow_nil: false }
 
   after_initialize :build_dependents
+  after_create :backfill_repositories
   after_save :refresh_bulk_notify_worker_status
   after_destroy :refresh_bulk_notify_worker_status
   before_validation :update_dependents_attributes
@@ -121,4 +122,8 @@ class GeoNode < ActiveRecord::Base
     self.system_hook.push_events = true
     self.system_hook.tag_push_events = true
   end
+
+  def backfill_repositories
+    GeoScheduleBackfillWorker.perform_async(id) unless primary?
+  end
 end

app/services/geo/repository_backfill_service.rb

+module Geo
+  class RepositoryBackfillService
+    attr_reader :project, :geo_node
+
+    def initialize(project, geo_node)
+      @project = project
+      @geo_node = geo_node
+    end
+
+    def execute
+      geo_node.system_hook.execute(hook_data, 'system_hooks')
+    end
+
+    private
+
+    def hook_data
+      {
+        event_name: 'push',
+        project_id: project.id,
+        project: project.hook_attrs
+      }
+    end
+  end
+end

app/services/geo/schedule_backfill_service.rb

+module Geo
+  class ScheduleBackfillService
+    attr_accessor :geo_node_id
+
+    def initialize(geo_node_id)
+      @geo_node_id = geo_node_id
+    end
+
+    def execute
+      return if geo_node_id.nil?
+
+      Project.all.select(&:valid_repo?).each do |project|
+        GeoBackfillWorker.perform_async(geo_node_id, project.id)
+      end
+    end
+  end
+end

app/workers/geo_backfill_worker.rb

+class GeoBackfillWorker
+  include Sidekiq::Worker
+  include ::GeoDynamicBackoff
+  include GeoQueue
+
+  def perform(geo_node_id, project_id)
+    project = Project.find(project_id)
+    geo_node = GeoNode.find(geo_node_id)
+
+    return unless project && geo_node
+
+    Geo::RepositoryBackfillService.new(project, geo_node).execute
+  end
+end

app/workers/geo_schedule_backfill_worker.rb

+class GeoScheduleBackfillWorker
+  include Sidekiq::Worker
+  include ::GeoDynamicBackoff
+  include GeoQueue
+
+  def perform(geo_node_id)
+    Geo::ScheduleBackfillService.new(geo_node_id).execute
+  end
+end

spec/models/geo_node_spec.rb

@@ -91,6 +91,30 @@ describe GeoNode, type: :model do
         expect(node.system_hook.push_events).to be_truthy
         expect(node.system_hook.tag_push_events).to be_truthy
       end
+
+      context 'backfill mechanism for a secondary node' do
+        before do
+          Sidekiq::Worker.clear_all
+        end
+
+        it 'schedules the scheduler worker' do
+          Sidekiq::Testing.fake! do
+            expect{ create(:geo_node, primary: false) }.to change(GeoScheduleBackfillWorker.jobs, :size).by(1)
+          end
+        end
+
+        it 'schedules the correct worker for the number of projects' do
+          Sidekiq::Testing.fake! do
+            2.times do
+              create(:project)
+            end
+
+            create(:geo_node, primary: false)
+
+            expect{ GeoScheduleBackfillWorker.drain }.to change(GeoBackfillWorker.jobs, :size).by(2)
+          end
+        end
+      end
     end
   end
 

spec/requests/api/geo_spec.rb

@@ -78,6 +78,14 @@ describe API::API, api: true do
       post api('/geo/receive_events'), push_payload, geo_token_header
       expect(response.status).to eq 201
     end
+
+    it 'can start a refresh process from the backfill service' do
+      project = create(:project)
+      backfill = Geo::RepositoryBackfillService.new(project, geo_node)
+      post api('/geo/receive_events'), backfill.send(:hook_data), geo_token_header
+
+      expect(response.status).to eq 201
+    end
   end
 
   describe 'POST /geo/receive_events push_tag events' do

spec/services/geo/repository_backfill_service_spec.rb

+require 'spec_helper'
+
+describe Geo::RepositoryBackfillService, services: true do
+  subject { Geo::RepositoryBackfillService.new(project, geo_node) }
+  let(:project) { create(:project) }
+  let(:geo_node) { create(:geo_node) }
+
+  describe '#execute' do
+    it 'calls upon the system hook of the Geo Node' do
+      WebMock.stub_request(:post, geo_node.geo_events_url)
+
+      subject.execute
+
+      expect(WebMock).to have_requested(:post, geo_node.geo_events_url).with(
+        headers: { 'Content-Type' => 'application/json', 'X-Gitlab-Event' => 'System Hook' },
+        body: {
+          event_name: 'push',
+          project_id: project.id,
+          project: project.hook_attrs
+        }
+      ).once
+    end
+  end
+end

spec/services/geo/schedule_backfill_service_spec.rb

+require 'spec_helper'
+
+describe Geo::ScheduleBackfillService, services: true do
+  subject { Geo::ScheduleBackfillService.new(geo_node.id) }
+  let(:geo_node) { create(:geo_node) }
+
+  describe '#execute' do
+    it 'schedules the backfill service' do
+      Sidekiq::Worker.clear_all
+
+      Sidekiq::Testing.fake! do
+        2.times do
+          create(:project)
+        end
+
+        expect{ subject.execute }.to change(GeoBackfillWorker.jobs, :size).by(2)
+      end
+    end
+  end
+end