def perform
return if delete_arrived_first?(object_uri) || unsupported_object_type?
- status = find_existing_status
+ RedisLock.acquire(lock_options) do |lock|
+ if lock.acquired?
+ @status = find_existing_status
+ process_status if @status.nil?
+ end
+ end
+
+ @status
+ end
- return status unless status.nil?
+ private
+ def process_status
ApplicationRecord.transaction do
- status = Status.create!(status_params)
+ @status = Status.create!(status_params)
- process_tags(status)
- process_attachments(status)
+ process_tags(@status)
+ process_attachments(@status)
end
- resolve_thread(status)
- distribute(status)
- forward_for_reply if status.public_visibility? || status.unlisted_visibility?
-
- status
+ resolve_thread(@status)
+ distribute(@status)
+ forward_for_reply if @status.public_visibility? || @status.unlisted_visibility?
end
- private
-
def find_existing_status
status = status_from_uri(object_uri)
status ||= Status.find_by(uri: @object['atomUri']) if @object['atomUri'].present?
return unless @json['signature'].present? && reply_to_local?
ActivityPub::RawDistributionWorker.perform_async(Oj.dump(@json), replied_to_status.account_id)
end
+
+ def lock_options
+ { redis: Redis.current, key: "create:#{@object['id']}" }
+ end
end