Add scheduler to synchronize moderation lists and populate suggestions

app/services/process_moderation_lists_service.rb

@@ -0,0 +1,177 @@
+# frozen_string_literal: true
+
+class ProcessModerationListsService < BaseService
+  Suggestion = Struct.new(:target_type, :target_key, :action, :moderation_subscription_id, :apply_automatically)
+  Retraction = Struct.new(:target_type, :action, :moderation_subscription_id, :retract_automatically)
+
+  def call
+    applicable_actions = Rails.configuration.x.mastodon.limited_federation_mode ? ['accept'] : ['reject', 'limit']
+
+    @retractions = {}
+
+    ModerationSubscription.order(priority: :desc).each do |subscription|
+      advisories = subscription.advisories.where(action: applicable_actions).to_a
+
+      # TODO: optimize this
+      advisories.delete_if { |advisory| SubscribedAdvisory.superseding_advisories(advisory).exists? }
+
+      # 1. apply automatic suggestions, if possible/needed
+      if subscription.apply_automatically
+        if subscription.preserve_relationships
+          domains_with_follows = Instance.with_domain_follows(advisories.filter_map { |advisory| advisory.target_key if advisory.target_type == 'domain' && advisory.action == 'reject' }).pluck(:domain)
+        end
+
+        advisories.delete_if do |advisory|
+          next false if subscription.preserve_relationships && advisory.target_type == 'domain' && domains_with_follows.include?(advisory.target_key)
+          next false if SubscribedAdvisory.conflicting_advisories(advisory).exists?
+
+          apply_automatic_advisory!(advisory)
+        end
+      end
+
+      # 2. upsert remaining suggestions
+      ModerationSuggestion.upsert_all(
+        advisories.map do |advisory|
+          {
+            target_type: advisory.target_type,
+            target_key: advisory.target_key,
+            action: advisory.action,
+            moderation_subscription_id: subscription.id,
+          }
+        end,
+        unique_by: [:target_type, :target_key, :action]
+      )
+
+      # 3. orphan suggestions
+      grouped_target_keys = advisories.group_by { |advisory| [advisory.target_type, advisory.action] }.transform_values { |advisories| advisories.map(&:target_key) }
+      SubscribedAdvisory::TARGET_TYPES.each do |target_type|
+        SubscribedAdvisory::ACTIONS.each do |action|
+          subscription
+            .suggestions
+            .where(target_type: target_type)
+            .where(action: action)
+            .where.not(target_key: grouped_target_keys.fetch([target_type, action], []))
+            .update_all(moderation_subscription_id: nil)
+        end
+      end
+
+      # 4. retractions
+      collect_retractions!(subscription)
+    end
+
+    # Cleanup: is it really it?
+    ModerationSuggestion.where(moderation_subscription_id: nil).delete_all
+
+    # Handle retractions
+    handle_retractions!
+  end
+
+  private
+
+  def collect_retractions!(subscription)
+    DomainAllow.where(moderation_subscription_id: subscription.id).where.not(domain: subscription.advisories.domain_target_type.accept_action.pluck(:target_key)).find_each do |domain_allow|
+      @retractions[domain_allow.domain] = Retraction.new(
+        action: 'accept',
+        target_type: 'domain',
+        retract_automatically: subscription.retract_automatically,
+        moderation_subscription_id: subscription.id
+      )
+    end
+
+    DomainBlock.where(moderation_subscription_id: subscription.id).where.not(domain: subscription.advisories.domain_target_type.where(action: ['limit', 'reject']).pluck(:target_key)).find_each do |domain_block|
+      next if domain_block.noop?
+
+      action = begin
+        case domain_block.severity
+        when 'silence'
+          'limit'
+        when 'suspend'
+          'reject'
+        end
+      end
+
+      @retractions[domain_block.domain] = Retraction.new(
+        action:,
+        target_type: 'domain',
+        retract_automatically: subscription.retract_automatically,
+        moderation_subscription_id: subscription.id
+      )
+    end
+  end
+
+  def handle_retractions!
+    representative = Account.representative
+
+    @retractions.each do |domain, attributes|
+      suggestion = ModerationSuggestion.find_by(target_type: attributes[:target_type], target_key: domain, action: attributes[:action])
+      if suggestion
+        # TODO: log
+        domain_block = DomainBlock.find_by(domain: domain)
+        domain_block&.update(moderation_subscription_id: suggestion.moderation_subscription_id)
+        suggestion.destroy
+      elsif attributes[:retract_automatically]
+        domain_block = DomainBlock.find_by(domain: domain)
+        representative.action_logs.create!(action: 'destroy', target: domain_block, moderation_subscription_id: domain_block.moderation_subscription_id)
+        UnblockDomainService.new.call(domain_block)
+      else
+        # TODO: what about if there is another kind of suggestion for the same target?
+        ModerationSuggestion.upsert(
+          {
+            target_type: attributes[:target_type],
+            target_key: domain,
+            action: 'retract',
+            moderation_subscription_id: attributes[:moderation_subscription_id],
+          },
+          unique_by: [:target_type, :target_key, :action]
+        )
+      end
+    end
+  end
+
+  def apply_automatic_advisory!(advisory)
+    representative = Account.representative
+
+    case [advisory.target_type, advisory.action]
+    when ['domain', 'accept']
+      # TODO: error handling
+      domain_allow = DomainAllow.create!(domain: advisory.target_key, moderation_subscription_id: advisory.moderation_subscription_id)
+      representative.action_logs.create!(action: 'create', target: domain_allow, moderation_subscription_id: domain_allow.moderation_subscription_id)
+      true
+    when ['domain', 'reject'], ['domain', 'limit']
+      existing_domain_block = DomainBlock.rule_for(advisory.target_key)
+      domain_block = DomainBlock.new(domain: advisory.target_key, moderation_subscription_id: advisory.moderation_subscription_id, obfuscate: false, severity: advisory.action == 'reject' ? 'suspend' : 'silence')
+
+      update = false
+
+      if existing_domain_block.present?
+        same_domain = existing_domain_block.domain == TagManager.instance.normalize_domain(domain_block.domain)
+
+        # Skip if it would downgrade a wider existing block
+        return unless same_domain || domain_block.stricter_than?(existing_domain_block)
+
+        if same_domain
+          # Allow updating existing domain blocks if they are already managed by the current subscription,
+          # or if the subscription is configured to be able to override existing ones
+          return unless existing_domain_block.moderation_subscription_id == advisory.moderation_subscription_id || (existing_domain_block.moderation_subscription_id.nil? && advisory.moderation_subscription.override_unattributed?)
+
+          # Only allow downgrades if retracting automatically
+          return unless domain_block.stricter_than?(existing_domain_block) || advisory.moderation_subscription.retract_automatically?
+
+          # Allow transparently upgrading a domain block
+          existing_domain_block.assign_attributes(moderation_subscription_id: advisory.moderation_subscription_id, severity: domain_block.severity)
+          domain_block = existing_domain_block
+
+          update = true
+        end
+      end
+
+      # TODO: error handling
+      domain_block.save!
+      representative.action_logs.create!(action: update ? 'update' : 'create', target: domain_block, moderation_subscription_id: domain_block.moderation_subscription_id)
+      DomainBlockWorker.perform_async(domain_block.id)
+      true
+    else
+      false
+    end
+  end
+end

app/workers/scheduler/moderation_list_synchronization_scheduler.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+class Scheduler::ModerationListSynchronizationScheduler
+  include Sidekiq::Worker
+
+  sidekiq_options retry: 0, lock: :until_executed, lock_ttl: 6.hours.to_i
+
+  def perform
+    return unless update_lists!
+
+    ProcessModerationListsService.new.call
+
+    # TODO: mail suggestions
+  end
+
+  private
+
+  def update_lists!
+    ModerationSubscription.all.map do |subscription|
+      ModerationSubscriptionSyncService.new.call(subscription)
+    end.any?
+  end
+end

config/sidekiq.yml

@@ -76,3 +76,7 @@
       interval: 1 hour
       class: Scheduler::CollectionItemCleanupScheduler
       queue: scheduler
+    moderation_list_synchronization_scheduler:
+      cron: '<%= Random.rand(0..59) %> <%= Random.rand(6..9) %> * * *'
+      class: ModerationListSynchronizationScheduler
+      queue: scheduler

spec/services/process_moderation_lists_service_spec.rb

@@ -0,0 +1,95 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe ProcessModerationListsService do
+  subject { described_class.new }
+
+  describe '#call' do
+    let(:high_priority_subscription) { Fabricate(:moderation_subscription, priority: 0, retract_automatically: false) }
+    let(:intermediate_priority_subscription) { Fabricate(:moderation_subscription, priority: 10, list_action: :accept) }
+    let(:subscription_with_automatic_application) { Fabricate(:moderation_subscription, priority: 20, apply_automatically: true, retract_automatically: true) }
+    let(:subscription_with_unsafe_application) { Fabricate(:moderation_subscription, priority: 21, apply_automatically: true, preserve_relationships: false, retract_automatically: false) }
+    let(:subscription_with_unattributed_override) { Fabricate(:moderation_subscription, priority: 22, apply_automatically: true, override_unattributed: true) }
+    let(:low_priority_subscription) { Fabricate(:moderation_subscription, priority: 30, retract_automatically: true) }
+
+    let(:local_account) { Fabricate(:account) }
+
+    before do
+      Fabricate(:domain_block, domain: 'retracted-block.com', moderation_subscription: high_priority_subscription, severity: :suspend)
+      Fabricate(:domain_block, domain: 'retract-me.com', moderation_subscription: low_priority_subscription, severity: :suspend)
+      Fabricate(:domain_block, domain: 'carried-over-block.com', moderation_subscription: subscription_with_automatic_application, severity: :suspend)
+      Fabricate(:domain_block, domain: 'upgradable-block.com', moderation_subscription: subscription_with_automatic_application, severity: :silence)
+      Fabricate(:domain_block, domain: 'prevented-automatic-block-upgrade.org', moderation_subscription: subscription_with_automatic_application, severity: :silence)
+      Fabricate(:domain_block, domain: 'bypassed-automatic-block-upgrade.org', moderation_subscription: subscription_with_unsafe_application, severity: :silence)
+      Fabricate(:domain_block, domain: 'downgrade-block.org', severity: :suspend)
+      Fabricate(:domain_block, domain: 'unattributed-upgradable-block.org', severity: :silence)
+      Fabricate(:domain_block, domain: 'second-unattributed-upgradable-block.org', severity: :silence)
+
+      # Remote followed users that prevent automatic block creation
+      %w(prevented-automatic-block.org prevented-automatic-block-upgrade.org bypassed-automatic-block.org bypassed-automatic-block-upgrade.org).each do |domain|
+        local_account.follow!(Fabricate(:account, domain: domain))
+      end
+
+      %w(example.com).each do |domain|
+        high_priority_subscription.advisories.create!(action: :reject, target_type: :domain, target_key: domain)
+      end
+
+      %w(automatic-block.org upgradable-block.com prevented-automatic-block.org prevented-automatic-block-upgrade.org).each do |domain|
+        subscription_with_automatic_application.advisories.create!(action: :reject, target_type: :domain, target_key: domain)
+      end
+
+      %w(bypassed-automatic-block.org bypassed-automatic-block-upgrade.org unattributed-upgradable-block.org).each do |domain|
+        subscription_with_unsafe_application.advisories.create!(action: :reject, target_type: :domain, target_key: domain)
+      end
+
+      subscription_with_unsafe_application.advisories.create!(action: :limit, target_type: :domain, target_key: 'downgrade-block.org')
+
+      subscription_with_unattributed_override.advisories.create!(action: :reject, target_type: :domain, target_key: 'second-unattributed-upgradable-block.org')
+
+      %w(example.com good.org).each do |domain|
+        intermediate_priority_subscription.advisories.create!(action: :accept, target_type: :domain, target_key: domain)
+      end
+
+      %w(example.com good.org evil.com carried-over-block.com).each do |domain|
+        low_priority_subscription.advisories.create!(action: :reject, target_type: :domain, target_key: domain)
+      end
+    end
+
+    it 'creates the epxected suggestions and audit log entries' do
+      expect { subject.call }
+        .to change(ModerationSuggestion, :count)
+        .and change { DomainBlock.find_by(domain: 'upgradable-block.com').severity }.from('silence').to('suspend')
+        .and change { DomainBlock.find_by(domain: 'bypassed-automatic-block-upgrade.org').severity }.from('silence').to('suspend')
+        .and change { DomainBlock.exists?(domain: 'retract-me.com') }.from(true).to(false)
+        .and change { DomainBlock.exists?(domain: 'automatic-block.org') }.from(false).to(true)
+        .and change { DomainBlock.exists?(domain: 'bypassed-automatic-block.org') }.from(false).to(true)
+        .and change { DomainBlock.find_by(domain: 'carried-over-block.com').moderation_subscription_id }.from(subscription_with_automatic_application.id).to(low_priority_subscription.id)
+        .and change { DomainBlock.find_by(domain: 'second-unattributed-upgradable-block.org').severity }.from('silence').to('suspend')
+        .and(not_change { DomainBlock.find_by(domain: 'unattributed-upgradable-block.org').severity })
+        .and(not_change { DomainBlock.find_by(domain: 'downgrade-block.org').severity })
+
+      expect(Admin::ActionLog.pluck(:action, :target_type, :human_identifier, :moderation_subscription_id))
+        .to contain_exactly(
+          ['destroy', 'DomainBlock', 'retract-me.com', low_priority_subscription.id],
+          ['create', 'DomainBlock', 'automatic-block.org', subscription_with_automatic_application.id],
+          ['update', 'DomainBlock', 'upgradable-block.com', subscription_with_automatic_application.id],
+          ['create', 'DomainBlock', 'bypassed-automatic-block.org', subscription_with_unsafe_application.id],
+          ['update', 'DomainBlock', 'bypassed-automatic-block-upgrade.org', subscription_with_unsafe_application.id],
+          ['update', 'DomainBlock', 'second-unattributed-upgradable-block.org', subscription_with_unattributed_override.id]
+        )
+
+      expect(ModerationSuggestion.pluck(:target_type, :target_key, :action, :moderation_subscription_id))
+        .to contain_exactly(
+          ['domain', 'example.com', 'reject', high_priority_subscription.id],
+          ['domain', 'good.org', 'reject', low_priority_subscription.id],
+          ['domain', 'evil.com', 'reject', low_priority_subscription.id],
+          ['domain', 'retracted-block.com', 'retract', high_priority_subscription.id],
+          ['domain', 'prevented-automatic-block.org', 'reject', subscription_with_automatic_application.id],
+          ['domain', 'prevented-automatic-block-upgrade.org', 'reject', subscription_with_automatic_application.id],
+          ['domain', 'downgrade-block.org', 'limit', subscription_with_unsafe_application.id],
+          ['domain', 'unattributed-upgradable-block.org', 'reject', subscription_with_unsafe_application.id]
+        )
+    end
+  end
+end