11 files changed, 200 insertions, 1 deletions
diff --git a/app/controllers/activitypub/followers_synchronizations_controller.rb b/app/controllers/activitypub/followers_synchronizations_controller.rb
new file mode 100644
index 000000000..525031105
--- /dev/null
+++ b/app/controllers/activitypub/followers_synchronizations_controller.rb
@@ -0,0 +1,36 @@
+# frozen_string_literal: true
+
+class ActivityPub::FollowersSynchronizationsController < ActivityPub::BaseController
+  include SignatureVerification
+  include AccountOwnedConcern
+
+  before_action :require_signature!
+  before_action :set_items
+  before_action :set_cache_headers
+
+  def show
+    expires_in 0, public: false
+    render json: collection_presenter,
+           serializer: ActivityPub::CollectionSerializer,
+           adapter: ActivityPub::Adapter,
+           content_type: 'application/activity+json'
+  end
+
+  private
+
+  def uri_prefix
+    signed_request_account.uri[/http(s?):\/\/[^\/]+\//]
+  end
+
+  def set_items
+    @items = @account.followers.where(Account.arel_table[:uri].matches(uri_prefix + '%', false, true)).pluck(:uri)
+  end
+
+  def collection_presenter
+    ActivityPub::CollectionPresenter.new(
+      id: account_followers_synchronization_url(@account),
+      type: :ordered,
+      items: @items
+    )
+  end
+end
diff --git a/app/controllers/activitypub/inboxes_controller.rb b/app/controllers/activitypub/inboxes_controller.rb
index 0a561e7f0..fdb60d590 100644
--- a/app/controllers/activitypub/inboxes_controller.rb
+++ b/app/controllers/activitypub/inboxes_controller.rb
@@ -11,6 +11,7 @@ class ActivityPub::InboxesController < ActivityPub::BaseController
 
   def create
     upgrade_account
+    process_collection_synchronization
     process_payload
     head 202
   end
@@ -52,6 +53,19 @@ class ActivityPub::InboxesController < ActivityPub::BaseController
     DeliveryFailureTracker.reset!(signed_request_account.inbox_url)
   end
 
+  def process_collection_synchronization
+    raw_params = request.headers['Collection-Synchronization']
+    return if raw_params.blank? || ENV['DISABLE_FOLLOWERS_SYNCHRONIZATION'] == 'true'
+
+    # Re-using the syntax for signature parameters
+    tree   = SignatureParamsParser.new.parse(raw_params)
+    params = SignatureParamsTransformer.new.apply(tree)
+
+    ActivityPub::PrepareFollowersSynchronizationService.new.call(signed_request_account, params)
+  rescue Parslet::ParseFailed
+    Rails.logger.warn 'Error parsing Collection-Synchronization header'
+  end
+
   def process_payload
     ActivityPub::ProcessingWorker.perform_async(signed_request_account.id, body, @account&.id)
   end
diff --git a/app/lib/activitypub/tag_manager.rb b/app/lib/activitypub/tag_manager.rb
index 3f98dad2e..3f2ae1106 100644
--- a/app/lib/activitypub/tag_manager.rb
+++ b/app/lib/activitypub/tag_manager.rb
@@ -40,6 +40,10 @@ class ActivityPub::TagManager
     end
   end
 
+  def uri_for_username(username)
+    account_url(username: username)
+  end
+
   def generate_uri_for(_target)
     URI.join(root_url, 'payloads', SecureRandom.uuid)
   end
diff --git a/app/models/account.rb b/app/models/account.rb
index 5acc8d621..59d338f5a 100644
--- a/app/models/account.rb
+++ b/app/models/account.rb
@@ -352,6 +352,12 @@ class Account < ApplicationRecord
     shared_inbox_url.presence || inbox_url
   end
 
+  def synchronization_uri_prefix
+    return 'local' if local?
+
+    @synchronization_uri_prefix ||= uri[/http(s?):\/\/[^\/]+\//]
+  end
+
   class Field < ActiveModelSerializers::Model
     attributes :name, :value, :verified_at, :account, :errors
 
diff --git a/app/models/concerns/account_interactions.rb b/app/models/concerns/account_interactions.rb
index 6a0ad5aa9..e2c4b8acf 100644
--- a/app/models/concerns/account_interactions.rb
+++ b/app/models/concerns/account_interactions.rb
@@ -243,6 +243,26 @@ module AccountInteractions
          .where('users.current_sign_in_at > ?', User::ACTIVE_DURATION.ago)
   end
 
+  def remote_followers_hash(url_prefix)
+    Rails.cache.fetch("followers_hash:#{id}:#{url_prefix}") do
+      digest = "\x00" * 32
+      followers.where(Account.arel_table[:uri].matches(url_prefix + '%', false, true)).pluck_each(:uri) do |uri|
+        Xorcist.xor!(digest, Digest::SHA256.digest(uri))
+      end
+      digest.unpack('H*')[0]
+    end
+  end
+
+  def local_followers_hash
+    Rails.cache.fetch("followers_hash:#{id}:local") do
+      digest = "\x00" * 32
+      followers.where(domain: nil).pluck_each(:username) do |username|
+        Xorcist.xor!(digest, Digest::SHA256.digest(ActivityPub::TagManager.instance.uri_for_username(username)))
+      end
+      digest.unpack('H*')[0]
+    end
+  end
+
   private
 
   def remove_potential_friendship(other_account, mutual = false)
diff --git a/app/models/follow.rb b/app/models/follow.rb
index 0b4ddbf3f..55a9da792 100644
--- a/app/models/follow.rb
+++ b/app/models/follow.rb
@@ -41,8 +41,10 @@ class Follow < ApplicationRecord
 
   before_validation :set_uri, only: :create
   after_create :increment_cache_counters
+  after_create :invalidate_hash_cache
   after_destroy :remove_endorsements
   after_destroy :decrement_cache_counters
+  after_destroy :invalidate_hash_cache
 
   private
 
@@ -63,4 +65,10 @@ class Follow < ApplicationRecord
     account&.decrement_count!(:following_count)
     target_account&.decrement_count!(:followers_count)
   end
+
+  def invalidate_hash_cache
+    return if account.local? && target_account.local?
+
+    Rails.cache.delete("followers_hash:#{target_account_id}:#{account.synchronization_uri_prefix}")
+  end
 end
diff --git a/app/services/activitypub/prepare_followers_synchronization_service.rb b/app/services/activitypub/prepare_followers_synchronization_service.rb
new file mode 100644
index 000000000..2d22ed701
--- /dev/null
+++ b/app/services/activitypub/prepare_followers_synchronization_service.rb
@@ -0,0 +1,13 @@
+# frozen_string_literal: true
+
+class ActivityPub::PrepareFollowersSynchronizationService < BaseService
+  include JsonLdHelper
+
+  def call(account, params)
+    @account = account
+
+    return if params['collectionId'] != @account.followers_url || invalid_origin?(params['url']) || @account.local_followers_hash == params['digest']
+
+    ActivityPub::FollowersSynchronizationWorker.perform_async(@account.id, params['url'])
+  end
+end
diff --git a/app/services/activitypub/synchronize_followers_service.rb b/app/services/activitypub/synchronize_followers_service.rb
new file mode 100644
index 000000000..d83fcf55e
--- /dev/null
+++ b/app/services/activitypub/synchronize_followers_service.rb
@@ -0,0 +1,74 @@
+# frozen_string_literal: true
+
+class ActivityPub::SynchronizeFollowersService < BaseService
+  include JsonLdHelper
+  include Payloadable
+
+  def call(account, partial_collection_url)
+    @account = account
+
+    items = collection_items(partial_collection_url)
+    return if items.nil?
+
+    # There could be unresolved accounts (hence the call to .compact) but this
+    # should never happen in practice, since in almost all cases we keep an
+    # Account record, and should we not do that, we should have sent a Delete.
+    # In any case there is not much we can do if that occurs.
+    @expected_followers = items.map { |uri| ActivityPub::TagManager.instance.uri_to_resource(uri, Account) }.compact
+
+    remove_unexpected_local_followers!
+    handle_unexpected_outgoing_follows!
+  end
+
+  private
+
+  def remove_unexpected_local_followers!
+    @account.followers.local.where.not(id: @expected_followers.map(&:id)).each do |unexpected_follower|
+      UnfollowService.new.call(unexpected_follower, @account)
+    end
+  end
+
+  def handle_unexpected_outgoing_follows!
+    @expected_followers.each do |expected_follower|
+      next if expected_follower.following?(@account)
+
+      if expected_follower.requested?(@account)
+        # For some reason the follow request went through but we missed it
+        expected_follower.follow_requests.find_by(target_account: @account)&.authorize!
+      else
+        # Since we were not aware of the follow from our side, we do not have an
+        # ID for it that we can include in the Undo activity. For this reason,
+        # the Undo may not work with software that relies exclusively on
+        # matching activity IDs and not the actor and target
+        follow = Follow.new(account: expected_follower, target_account: @account)
+        ActivityPub::DeliveryWorker.perform_async(build_undo_follow_json(follow), follow.account_id, follow.target_account.inbox_url)
+      end
+    end
+  end
+
+  def build_undo_follow_json(follow)
+    Oj.dump(serialize_payload(follow, ActivityPub::UndoFollowSerializer))
+  end
+
+  def collection_items(collection_or_uri)
+    collection = fetch_collection(collection_or_uri)
+    return unless collection.is_a?(Hash)
+
+    collection = fetch_collection(collection['first']) if collection['first'].present?
+    return unless collection.is_a?(Hash)
+
+    case collection['type']
+    when 'Collection', 'CollectionPage'
+      collection['items']
+    when 'OrderedCollection', 'OrderedCollectionPage'
+      collection['orderedItems']
+    end
+  end
+
+  def fetch_collection(collection_or_uri)
+    return collection_or_uri if collection_or_uri.is_a?(Hash)
+    return if invalid_origin?(collection_or_uri)
+
+    fetch_resource_without_id_validation(collection_or_uri, nil, true)
+  end
+end
diff --git a/app/workers/activitypub/delivery_worker.rb b/app/workers/activitypub/delivery_worker.rb
index 60775787a..6c5a576a7 100644
--- a/app/workers/activitypub/delivery_worker.rb
+++ b/app/workers/activitypub/delivery_worker.rb
@@ -2,6 +2,7 @@
 
 class ActivityPub::DeliveryWorker
   include Sidekiq::Worker
+  include RoutingHelper
   include JsonLdHelper
 
   STOPLIGHT_FAILURE_THRESHOLD = 10
@@ -38,9 +39,18 @@ class ActivityPub::DeliveryWorker
     Request.new(:post, @inbox_url, body: @json, http_client: http_client).tap do |request|
       request.on_behalf_of(@source_account, :uri, sign_with: @options[:sign_with])
       request.add_headers(HEADERS)
+      request.add_headers({ 'Collection-Synchronization' => synchronization_header }) if ENV['DISABLE_FOLLOWERS_SYNCHRONIZATION'] != 'true' && @options[:synchronize_followers]
     end
   end
 
+  def synchronization_header
+    "collectionId=\"#{account_followers_url(@source_account)}\", digest=\"#{@source_account.remote_followers_hash(inbox_url_prefix)}\", url=\"#{account_followers_synchronization_url(@source_account)}\""
+  end
+
+  def inbox_url_prefix
+    @inbox_url[/http(s?):\/\/[^\/]+\//]
+  end
+
   def perform_request
     light = Stoplight(@inbox_url) do
       request_pool.with(@host) do |http_client|
diff --git a/app/workers/activitypub/distribution_worker.rb b/app/workers/activitypub/distribution_worker.rb
index e4997ba0e..9b4814644 100644
--- a/app/workers/activitypub/distribution_worker.rb
+++ b/app/workers/activitypub/distribution_worker.rb
@@ -13,7 +13,7 @@ class ActivityPub::DistributionWorker
     return if skip_distribution?
 
     ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
-      [payload, @account.id, inbox_url]
+      [payload, @account.id, inbox_url, { synchronize_followers: !@status.distributable? }]
     end
 
     relay! if relayable?
diff --git a/app/workers/activitypub/followers_synchronization_worker.rb b/app/workers/activitypub/followers_synchronization_worker.rb
new file mode 100644
index 000000000..35a3ef0b9
--- /dev/null
+++ b/app/workers/activitypub/followers_synchronization_worker.rb
@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+class ActivityPub::FollowersSynchronizationWorker
+  include Sidekiq::Worker
+
+  sidekiq_options queue: 'push', lock: :until_executed
+
+  def perform(account_id, url)
+    @account = Account.find_by(id: account_id)
+    return true if @account.nil?
+
+    ActivityPub::SynchronizeFollowersService.new.call(@account, url)
+  end
+end