Add scheduler job to repair collection attribution (#39550)

app/workers/scheduler/repair_remote_collections_scheduler.rb

@@ -0,0 +1,39 @@
+# frozen_string_literal: true
+
+# TODO: Remove in the next version
+class Scheduler::RepairRemoteCollectionsScheduler
+  include Sidekiq::Worker
+  include JsonLdHelper
+  include Redisable
+
+  sidekiq_options retry: 0, lock: :until_executed, lock_ttl: 1.day.to_i
+
+  def perform
+    max_id = Collection.maximum(:id)
+    last_known_good = redis.get('remote_collection_repair:last_known_good')
+
+    affected_collections = Collection.joins(:account).where(local: false)
+      .where("substring(collections.uri from '/ap/users/\\d+/') != substring(accounts.collections_url from '/ap/users/\\d+/')")
+    affected_collections = affected_collections.where(id: last_known_good...) if last_known_good
+
+    successful = true
+    affected_collections.find_each do |collection|
+      json = fetch_resource(collection.uri, true)
+      if json.nil? || !json.is_a?(Hash) || !json.key?('attributedTo')
+        successful = false
+        next
+      end
+
+      account = Account.find_by(uri: json['attributedTo'])
+      account ||= ActivityPub::FetchRemoteAccountService.new.call(json['attributedTo'])
+      if account.nil?
+        successful = false
+        next
+      end
+
+      collection.update(account:)
+    end
+
+    redis.set('remote_collection_repair:last_known_good', max_id.to_s) if successful
+  end
+end

config/sidekiq.yml

@@ -76,3 +76,7 @@
       interval: 1 hour
       class: Scheduler::CollectionItemCleanupScheduler
       queue: scheduler
+    repair_remote_collections_scheduler:
+      every: ['24h', first_in: '1s']
+      class: Scheduler::RepairRemoteCollectionsScheduler
+      queue: scheduler

spec/workers/scheduler/repair_remote_collections_scheduler_spec.rb

@@ -0,0 +1,106 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe Scheduler::RepairRemoteCollectionsScheduler do
+  include Redisable
+
+  let(:worker) { described_class.new }
+
+  describe 'perform' do
+    let(:owner) do
+      Fabricate(:remote_account, uri: 'https://example.com/accounts/123', collections_url: 'https://example.com/ap/users/123/featured_collections')
+    end
+    let(:other_account) do
+      Fabricate(:remote_account, collections_url: 'https://example.com/ap/users/234/featured_collections')
+    end
+    let(:collection) do
+      Fabricate(:remote_collection,
+                uri: 'https://example.com/ap/users/123/collections/1',
+                account: other_account,
+                created_at: 2.hours.ago)
+    end
+    let(:response) do
+      {
+        '@context' => 'https://www.w3.org/ns/activitystreams',
+        'id' => collection.uri,
+        'type' => 'FeaturedCollection',
+        'name' => collection.name,
+        'summary' => collection.description_html,
+        'attributedTo' => 'https://example.com/accounts/123',
+        'sensitive' => false,
+        'discoverable' => true,
+        'totalItems' => 0,
+      }
+    end
+
+    before do
+      stub_request(:get, collection.uri)
+        .to_return_json(
+          status: 200,
+          body: response,
+          headers: { 'Content-Type' => 'application/activity+json' }
+        )
+    end
+
+    after do
+      redis.del('remote_collection_repair:last_known_good')
+    end
+
+    context 'without flag in redis set' do
+      context 'when the proper account is known' do
+        before { owner }
+
+        it 'fixes the wrong attribution and sets the flag in redis' do
+          worker.perform
+
+          expect(collection.reload.account).to eq owner
+          expect(redis.get('remote_collection_repair:last_known_good')).to_not be_nil
+        end
+      end
+
+      context 'when the proper account is unknown' do
+        let(:stubbed_service) { instance_double(ActivityPub::FetchRemoteAccountService) }
+
+        before do
+          allow(stubbed_service).to receive(:call).with('https://example.com/accounts/123') { service_result }
+          allow(ActivityPub::FetchRemoteAccountService).to receive(:new).and_return(stubbed_service)
+        end
+
+        context 'when the service can fetch the account' do
+          let(:service_result) { owner }
+
+          it 'fixes the wrong attribution and sets the flag in redis' do
+            worker.perform
+
+            expect(collection.reload.account).to eq owner
+            expect(redis.get('remote_collection_repair:last_known_good')).to_not be_nil
+          end
+        end
+
+        context 'when the service cannot fetch the account' do
+          let(:service_result) { nil }
+
+          it 'does not fix the collection and does not set the flag in redis' do
+            worker.perform
+
+            expect(collection.reload.account).to eq other_account
+            expect(redis.get('remote_collection_repair:last_known_good')).to be_nil
+          end
+        end
+      end
+    end
+
+    context 'with flag in redis set' do
+      before do
+        redis.set('remote_collection_repair:last_known_good', collection.id + 1)
+      end
+
+      it 'does nothing with collections older than the last known good time' do
+        worker.perform
+
+        expect(collection.reload.account).to eq other_account
+      end
+    end
+  end
+end