Fix incomplete account records being read (#4998)

* Fix incomplete account records being read

- Put account processing into redis lock
- Do not save until record is complete

* Fix spaces

1 files changed, 56 insertions, 24 deletions

diff --git a/app/services/activitypub/process_account_service.rb b/app/services/activitypub/process_account_service.rb
index a45681078..811209537 100644
--- a/app/services/activitypub/process_account_service.rb
+++ b/app/services/activitypub/process_account_service.rb
@@ -12,14 +12,21 @@ class ActivityPub::ProcessAccountService < BaseService
     @uri         = @json['id']
     @username    = username
     @domain      = domain
-    @account     = Account.find_by(uri: @uri)
     @collections = {}
 
-    old_public_key = @account&.public_key
-    create_account  if @account.nil?
-    upgrade_account if @account.ostatus?
-    update_account
-    RefollowWorker.perform_async(@account.id) if !old_public_key.nil? && old_public_key != @account.public_key
+    RedisLock.acquire(lock_options) do |lock|
+      if lock.acquired?
+        @account        = Account.find_by(uri: @uri)
+        @old_public_key = @account&.public_key
+        @old_protocol   = @account&.protocol
+
+        create_account if @account.nil?
+        update_account
+      end
+    end
+
+    after_protocol_change! if protocol_changed?
+    after_key_change! if key_changed?
 
     @account
   rescue Oj::ParseError
@@ -37,33 +44,46 @@ class ActivityPub::ProcessAccountService < BaseService
     @account.suspended   = true if auto_suspend?
     @account.silenced    = true if auto_silence?
     @account.private_key = nil
-    @account.save!
   end
 
   def update_account
     @account.last_webfingered_at = Time.now.utc
     @account.protocol            = :activitypub
-    @account.inbox_url           = @json['inbox'] || ''
-    @account.outbox_url          = @json['outbox'] || ''
-    @account.shared_inbox_url    = (@json['endpoints'].is_a?(Hash) ? @json['endpoints']['sharedInbox'] : @json['sharedInbox']) || ''
-    @account.followers_url       = @json['followers'] || ''
-    @account.url                 = url || @uri
-    @account.display_name        = @json['name'] || ''
-    @account.note                = @json['summary'] || ''
-    @account.avatar_remote_url   = image_url('icon')  unless skip_download?
-    @account.header_remote_url   = image_url('image') unless skip_download?
-    @account.public_key          = public_key || ''
-    @account.locked              = @json['manuallyApprovesFollowers'] || false
-    @account.statuses_count      = outbox_total_items    if outbox_total_items.present?
-    @account.following_count     = following_total_items if following_total_items.present?
-    @account.followers_count     = followers_total_items if followers_total_items.present?
+
+    set_immediate_attributes!
+    set_fetchable_attributes!
+
     @account.save_with_optional_media!
   end
 
-  def upgrade_account
+  def set_immediate_attributes!
+    @account.inbox_url        = @json['inbox'] || ''
+    @account.outbox_url       = @json['outbox'] || ''
+    @account.shared_inbox_url = (@json['endpoints'].is_a?(Hash) ? @json['endpoints']['sharedInbox'] : @json['sharedInbox']) || ''
+    @account.followers_url    = @json['followers'] || ''
+    @account.url              = url || @uri
+    @account.display_name     = @json['name'] || ''
+    @account.note             = @json['summary'] || ''
+    @account.locked           = @json['manuallyApprovesFollowers'] || false
+  end
+
+  def set_fetchable_attributes!
+    @account.avatar_remote_url = image_url('icon')  unless skip_download?
+    @account.header_remote_url = image_url('image') unless skip_download?
+    @account.public_key        = public_key || ''
+    @account.statuses_count    = outbox_total_items    if outbox_total_items.present?
+    @account.following_count   = following_total_items if following_total_items.present?
+    @account.followers_count   = followers_total_items if followers_total_items.present?
+  end
+
+  def after_protocol_change!
     ActivityPub::PostUpgradeWorker.perform_async(@account.domain)
   end
 
+  def after_key_change!
+    RefollowWorker.perform_async(@account.id)
+  end
+
   def image_url(key)
     value = first_of_value(@json[key])
 
@@ -122,15 +142,27 @@ class ActivityPub::ProcessAccountService < BaseService
   end
 
   def auto_suspend?
-    domain_block && domain_block.suspend?
+    domain_block&.suspend?
   end
 
   def auto_silence?
-    domain_block && domain_block.silence?
+    domain_block&.silence?
   end
 
   def domain_block
     return @domain_block if defined?(@domain_block)
     @domain_block = DomainBlock.find_by(domain: @domain)
   end
+
+  def key_changed?
+    !@old_public_key.nil? && @old_public_key != @account.public_key
+  end
+
+  def protocol_changed?
+    !@old_protocol.nil? && @old_protocol != @account.protocol
+  end
+
+  def lock_options
+    { redis: Redis.current, key: "process_account:#{@uri}" }
+  end
 end