Add follower synchronization mechanism (#14510)

* Add support for followers synchronization on the receiving end

Check the `collectionSynchronization` attribute on `Create` and `Announce`
activities and synchronize followers from provided collection if possible.

* Add tests for followers synchronization on the receiving end

* Add support for follower synchronization on the sender's end

* Add tests for the sending end

* Switch from AS attributes to HTTP header

Replace the custom `collectionSynchronization` ActivityStreams attribute by
an HTTP header (`X-AS-Collection-Synchronization`) with the same syntax as
the `Signature` header and the following fields:
- `collectionId` to specify which collection to synchronize
- `digest` for the SHA256 hex-digest of the list of followers known on the
   receiving instance (where “receiving instance” is determined by accounts
   sharing the same host name for their ActivityPub actor `id`)
- `url` of a collection that should be fetched by the instance actor

Internally, move away from the webfinger-based `domain` attribute and use
account `uri` prefix to group accounts.

* Add environment variable to disable followers synchronization

Since the whole mechanism relies on some new preconditions that, in some
extremely rare cases, might not be met, add an environment variable
(DISABLE_FOLLOWERS_SYNCHRONIZATION) to disable the mechanism altogether and
avoid followers being incorrectly removed.

The current conditions are:
1. all managed accounts' actor `id` and inbox URL have the same URI scheme and
   netloc.
2. all accounts whose actor `id` or inbox URL share the same URI scheme and
   netloc as a managed account must be managed by the same Mastodon instance
   as well.

As far as Mastodon is concerned, breaking those preconditions require extensive
configuration changes in the reverse proxy and might also cause other issues.

Therefore, this environment variable provides a way out for people with highly
unusual configurations, and can be safely ignored for the overwhelming majority
of Mastodon administrators.

* Only set follower synchronization header on non-public statuses

This is to avoid unnecessary computations and allow Follow-related
activities to be handled by the usual codepath instead of going through
the synchronization mechanism (otherwise, any Follow/Undo/Accept activity
would trigger the synchronization mechanism even if processing the activity
itself would be enough to re-introduce synchronization)

* Change how ActivityPub::SynchronizeFollowersService handles follow requests

If the remote lists a local follower which we only know has sent a follow
request, consider the follow request as accepted instead of sending an Undo.

* Integrate review feeback

- rename X-AS-Collection-Synchronization to Collection-Synchronization
- various minor refactoring and code style changes

* Only select required fields when computing followers_hash

* Use actor URI rather than webfinger domain in synchronization endpoint

* Change hash computation to be a XOR of individual hashes

Makes it much easier to be memory-efficient, and avoid sorting discrepancy issues.

* Marginally improve followers_hash computation speed

* Further improve hash computation performances by using pluck_each

5 files changed, 264 insertions, 2 deletions

diff --git a/spec/controllers/activitypub/followers_synchronizations_controller_spec.rb b/spec/controllers/activitypub/followers_synchronizations_controller_spec.rb
new file mode 100644
index 000000000..a24d3f8e0
--- /dev/null
+++ b/spec/controllers/activitypub/followers_synchronizations_controller_spec.rb
@@ -0,0 +1,58 @@
+require 'rails_helper'
+
+RSpec.describe ActivityPub::FollowersSynchronizationsController, type: :controller do
+  let!(:account)    { Fabricate(:account) }
+  let!(:follower_1) { Fabricate(:account, domain: 'example.com', uri: 'https://example.com/users/a') }
+  let!(:follower_2) { Fabricate(:account, domain: 'example.com', uri: 'https://example.com/users/b') }
+  let!(:follower_3) { Fabricate(:account, domain: 'foo.com', uri: 'https://foo.com/users/a') }
+
+  before do
+    follower_1.follow!(account)
+    follower_2.follow!(account)
+    follower_3.follow!(account)
+  end
+
+  before do
+    allow(controller).to receive(:signed_request_account).and_return(remote_account)
+  end
+
+  describe 'GET #show' do
+    context 'without signature' do
+      let(:remote_account) { nil }
+
+      before do
+        get :show, params: { account_username: account.username }
+      end
+
+      it 'returns http not authorized' do
+        expect(response).to have_http_status(401)
+      end
+    end
+
+    context 'with signature from example.com' do
+      let(:remote_account) { Fabricate(:account, domain: 'example.com', uri: 'https://example.com/instance') }
+
+      before do
+        get :show, params: { account_username: account.username }
+      end
+
+      it 'returns http success' do
+        expect(response).to have_http_status(200)
+      end
+
+      it 'returns application/activity+json' do
+        expect(response.content_type).to eq 'application/activity+json'
+      end
+
+      it 'returns orderedItems with followers from example.com' do
+        json = body_as_json
+        expect(json[:orderedItems]).to be_an Array
+        expect(json[:orderedItems].sort).to eq [follower_1.uri, follower_2.uri]
+      end
+
+      it 'returns private Cache-Control header' do
+        expect(response.headers['Cache-Control']).to eq 'max-age=0, private'
+      end
+    end
+  end
+end
diff --git a/spec/controllers/activitypub/inboxes_controller_spec.rb b/spec/controllers/activitypub/inboxes_controller_spec.rb
index f3bc23953..e5c004611 100644
--- a/spec/controllers/activitypub/inboxes_controller_spec.rb
+++ b/spec/controllers/activitypub/inboxes_controller_spec.rb
@@ -22,6 +22,56 @@ RSpec.describe ActivityPub::InboxesController, type: :controller do
       end
     end
 
+    context 'with Collection-Synchronization header' do
+      let(:remote_account)             { Fabricate(:account, followers_url: 'https://example.com/followers', domain: 'example.com', uri: 'https://example.com/actor', protocol: :activitypub) }
+      let(:synchronization_collection) { remote_account.followers_url }
+      let(:synchronization_url)        { 'https://example.com/followers-for-domain' }
+      let(:synchronization_hash)       { 'somehash' }
+      let(:synchronization_header)     { "collectionId=\"#{synchronization_collection}\", digest=\"#{synchronization_hash}\", url=\"#{synchronization_url}\"" }
+
+      before do
+        allow(ActivityPub::FollowersSynchronizationWorker).to receive(:perform_async).and_return(nil)
+        allow_any_instance_of(Account).to receive(:local_followers_hash).and_return('somehash')
+
+        request.headers['Collection-Synchronization'] = synchronization_header
+        post :create, body: '{}'
+      end
+
+      context 'with mismatching target collection' do
+        let(:synchronization_collection) { 'https://example.com/followers2' }
+
+        it 'does not start a synchronization job' do
+          expect(ActivityPub::FollowersSynchronizationWorker).not_to have_received(:perform_async)
+        end
+      end
+
+      context 'with mismatching domain in partial collection attribute' do
+        let(:synchronization_url) { 'https://example.org/followers' }
+
+        it 'does not start a synchronization job' do
+          expect(ActivityPub::FollowersSynchronizationWorker).not_to have_received(:perform_async)
+        end
+      end
+
+      context 'with matching digest' do
+        it 'does not start a synchronization job' do
+          expect(ActivityPub::FollowersSynchronizationWorker).not_to have_received(:perform_async)
+        end
+      end
+
+      context 'with mismatching digest' do
+        let(:synchronization_hash) { 'wronghash' }
+
+        it 'starts a synchronization job' do
+          expect(ActivityPub::FollowersSynchronizationWorker).to have_received(:perform_async)
+        end
+      end
+
+      it 'returns http accepted' do
+        expect(response).to have_http_status(202)
+      end
+    end
+
     context 'without signature' do
       before do
         post :create, body: '{}'
diff --git a/spec/models/concerns/account_interactions_spec.rb b/spec/models/concerns/account_interactions_spec.rb
index f0380179c..85fbf7e79 100644
--- a/spec/models/concerns/account_interactions_spec.rb
+++ b/spec/models/concerns/account_interactions_spec.rb
@@ -539,6 +539,49 @@ describe AccountInteractions do
     end
   end
 
+  describe '#followers_hash' do
+    let(:me) { Fabricate(:account, username: 'Me') }
+    let(:remote_1) { Fabricate(:account, username: 'alice', domain: 'example.org', uri: 'https://example.org/users/alice') }
+    let(:remote_2) { Fabricate(:account, username: 'bob', domain: 'example.org', uri: 'https://example.org/users/bob') }
+    let(:remote_3) { Fabricate(:account, username: 'eve', domain: 'foo.org', uri: 'https://foo.org/users/eve') }
+
+    before do
+      remote_1.follow!(me)
+      remote_2.follow!(me)
+      remote_3.follow!(me)
+      me.follow!(remote_1)
+    end
+
+    context 'on a local user' do
+      it 'returns correct hash for remote domains' do
+        expect(me.remote_followers_hash('https://example.org/')).to eq '707962e297b7bd94468a21bc8e506a1bcea607a9142cd64e27c9b106b2a5f6ec'
+        expect(me.remote_followers_hash('https://foo.org/')).to eq 'ccb9c18a67134cfff9d62c7f7e7eb88e6b803446c244b84265565f4eba29df0e'
+      end
+
+      it 'invalidates cache as needed when removing or adding followers' do
+        expect(me.remote_followers_hash('https://example.org/')).to eq '707962e297b7bd94468a21bc8e506a1bcea607a9142cd64e27c9b106b2a5f6ec'
+        remote_1.unfollow!(me)
+        expect(me.remote_followers_hash('https://example.org/')).to eq '241b00794ce9b46aa864f3220afadef128318da2659782985bac5ed5bd436bff'
+        remote_1.follow!(me)
+        expect(me.remote_followers_hash('https://example.org/')).to eq '707962e297b7bd94468a21bc8e506a1bcea607a9142cd64e27c9b106b2a5f6ec'
+      end
+    end
+
+    context 'on a remote user' do
+      it 'returns correct hash for remote domains' do
+        expect(remote_1.local_followers_hash).to eq Digest::SHA256.hexdigest(ActivityPub::TagManager.instance.uri_for(me))
+      end
+
+      it 'invalidates cache as needed when removing or adding followers' do
+        expect(remote_1.local_followers_hash).to eq Digest::SHA256.hexdigest(ActivityPub::TagManager.instance.uri_for(me))
+        me.unfollow!(remote_1)
+        expect(remote_1.local_followers_hash).to eq '0000000000000000000000000000000000000000000000000000000000000000'
+        me.follow!(remote_1)
+        expect(remote_1.local_followers_hash).to eq Digest::SHA256.hexdigest(ActivityPub::TagManager.instance.uri_for(me))
+      end
+    end
+  end
+
   describe 'muting an account' do
     let(:me) { Fabricate(:account, username: 'Me') }
     let(:you) { Fabricate(:account, username: 'You') }
diff --git a/spec/services/activitypub/synchronize_followers_service_spec.rb b/spec/services/activitypub/synchronize_followers_service_spec.rb
new file mode 100644
index 000000000..75dcf204b
--- /dev/null
+++ b/spec/services/activitypub/synchronize_followers_service_spec.rb
@@ -0,0 +1,105 @@
+require 'rails_helper'
+
+RSpec.describe ActivityPub::SynchronizeFollowersService, type: :service do
+  let(:actor)          { Fabricate(:account, domain: 'example.com', uri: 'http://example.com/account', inbox_url: 'http://example.com/inbox') }
+  let(:alice)          { Fabricate(:account, username: 'alice') }
+  let(:bob)            { Fabricate(:account, username: 'bob') }
+  let(:eve)            { Fabricate(:account, username: 'eve') }
+  let(:mallory)        { Fabricate(:account, username: 'mallory') }
+  let(:collection_uri) { 'http://example.com/partial-followers' }
+
+  let(:items) do
+    [
+      ActivityPub::TagManager.instance.uri_for(alice),
+      ActivityPub::TagManager.instance.uri_for(eve),
+      ActivityPub::TagManager.instance.uri_for(mallory),
+    ]
+  end
+
+  let(:payload) do
+    {
+      '@context': 'https://www.w3.org/ns/activitystreams',
+      type: 'Collection',
+      id: collection_uri,
+      items: items,
+    }.with_indifferent_access
+  end
+
+  subject { described_class.new }
+
+  shared_examples 'synchronizes followers' do
+    before do
+      alice.follow!(actor)
+      bob.follow!(actor)
+      mallory.request_follow!(actor)
+
+      allow(ActivityPub::DeliveryWorker).to receive(:perform_async)
+
+      subject.call(actor, collection_uri)
+    end
+
+    it 'keeps expected followers' do
+      expect(alice.following?(actor)).to be true
+    end
+
+    it 'removes local followers not in the remote list' do
+      expect(bob.following?(actor)).to be false
+    end
+
+    it 'converts follow requests to follow relationships when they have been accepted' do
+      expect(mallory.following?(actor)).to be true
+    end
+
+    it 'sends an Undo Follow to the actor' do
+      expect(ActivityPub::DeliveryWorker).to have_received(:perform_async).with(anything, eve.id, actor.inbox_url)
+    end
+  end
+
+  describe '#call' do
+    context 'when the endpoint is a Collection of actor URIs' do
+      before do
+        stub_request(:get, collection_uri).to_return(status: 200, body: Oj.dump(payload))
+      end
+
+      it_behaves_like 'synchronizes followers'
+    end
+
+    context 'when the endpoint is an OrderedCollection of actor URIs' do
+      let(:payload) do
+        {
+          '@context': 'https://www.w3.org/ns/activitystreams',
+          type: 'OrderedCollection',
+          id: collection_uri,
+          orderedItems: items,
+        }.with_indifferent_access
+      end
+
+      before do
+        stub_request(:get, collection_uri).to_return(status: 200, body: Oj.dump(payload))
+      end
+
+      it_behaves_like 'synchronizes followers'
+    end
+
+    context 'when the endpoint is a paginated Collection of actor URIs' do
+      let(:payload) do
+        {
+          '@context': 'https://www.w3.org/ns/activitystreams',
+          type: 'Collection',
+          id: collection_uri,
+          first: {
+            type: 'CollectionPage',
+            partOf: collection_uri,
+            items: items,
+          }
+        }.with_indifferent_access
+      end
+
+      before do
+        stub_request(:get, collection_uri).to_return(status: 200, body: Oj.dump(payload))
+      end
+
+      it_behaves_like 'synchronizes followers'
+    end
+  end
+end
diff --git a/spec/workers/activitypub/delivery_worker_spec.rb b/spec/workers/activitypub/delivery_worker_spec.rb
index 351be185c..f4633731e 100644
--- a/spec/workers/activitypub/delivery_worker_spec.rb
+++ b/spec/workers/activitypub/delivery_worker_spec.rb
@@ -3,16 +3,22 @@
 require 'rails_helper'
 
 describe ActivityPub::DeliveryWorker do
+  include RoutingHelper
+
   subject { described_class.new }
 
   let(:sender)  { Fabricate(:account) }
   let(:payload) { 'test' }
 
+  before do
+    allow_any_instance_of(Account).to receive(:remote_followers_hash).with('https://example.com/').and_return('somehash')
+  end
+
   describe 'perform' do
     it 'performs a request' do
       stub_request(:post, 'https://example.com/api').to_return(status: 200)
-      subject.perform(payload, sender.id, 'https://example.com/api')
-      expect(a_request(:post, 'https://example.com/api')).to have_been_made.once
+      subject.perform(payload, sender.id, 'https://example.com/api', { synchronize_followers: true })
+      expect(a_request(:post, 'https://example.com/api').with(headers: { 'Collection-Synchronization' => "collectionId=\"#{account_followers_url(sender)}\", digest=\"somehash\", url=\"#{account_followers_synchronization_url(sender)}\"" })).to have_been_made.once
     end
 
     it 'raises when request fails' do