Add context resolver

config/locales/server.en.yml

@@ -71,7 +71,6 @@ en:
         object_not_ready: "Object is not ready"
         activity_already_processed: "Activity has already been processed"
         activity_not_supported: "Activity is not supported for actor and object"
-        activity_not_valid: "Activity is not valid"
         invalid_create_actor: "Actor cannot create posts"
         cannot_reply_to_deleted_post: "Cannot create an object inReplyTo a deleted Post"
         undo_actor_must_match_object_actor: "Undo Actor must match Object Actor"
@@ -84,6 +83,7 @@ en:
         only_followed_actors_create_new_topics: "Only followed actors can create new topics."
         actor_already_following: "Actor is already following that object"
         invalid_collection_item: "Invalid item"
+        cannot_resolve_context: "Cannot resolve object context"
       error:
         failed_to_create_user: "Failed to create user for %{actor_id}"
         failed_to_create_post: "Failed to create post for %{object_id}"

lib/discourse_activity_pub/ap/activity.rb

@@ -23,25 +23,27 @@ def object
       end
 
       def process
-        return false unless process_actor_and_object
-        return false unless perform_validate_activity
         return false unless perform_transactions
-
         forward_activity
-
         self
       end
 
-      def perform_transactions
+      def perform_transactions(skip_process: false)
         performed = true
 
         ActiveRecord::Base.transaction do
           begin
+            process_activity unless skip_process
+            validate_activity
             perform_activity
             store_activity
             respond_to_activity
+          rescue DiscourseActivityPub::AP::Handlers::Warning => warning
+            DiscourseActivityPub::Logger.warn(warning.message) if warning.message
+            performed = false
+            raise ActiveRecord::Rollback
           rescue DiscourseActivityPub::AP::Handlers::Error => error
-            DiscourseActivityPub::Logger.error(error.message)
+            DiscourseActivityPub::Logger.error(error.message) if error.message
             performed = false
             raise ActiveRecord::Rollback
           end
@@ -50,14 +52,12 @@ def perform_transactions
         performed
       end
 
-      def perform_validate_activity
-        return true if validate_activity
-        process_failed("activity_not_valid")
-        false
+      def process_activity
+        raise DiscourseActivityPub::AP::Handlers::Warning unless process_actor_and_object
       end
 
       def validate_activity
-        apply_handlers(type, :validate)
+        apply_handlers(type, :validate, raise_errors: true)
       end
 
       def perform_activity

lib/discourse_activity_pub/ap/activity/announce.rb

@@ -22,14 +22,18 @@ def process
             unless actor.stored.can_perform_activity?(type, object.type)
               return process_failed("activity_not_supported")
             end
-            return false unless perform_validate_activity
 
-            perform_transactions
+            perform_transactions(skip_process: true)
             forward_activity
           else
             # If the Announce wraps an activity we process the Activity and discard the Announce.
             # See https://codeberg.org/fediverse/fep/src/branch/main/fep/1b12/fep-1b12.md#the-announce-activity
-            return false unless perform_validate_activity
+            begin
+              apply_handlers(type, :validate, raise_errors: true)
+            rescue DiscourseActivityPub::AP::Handlers::Warning => warning
+              DiscourseActivityPub::Logger.warn(warning.message) if warning.message
+              return false
+            end
             object.parent = self
             object.delivered_to << delivered_to if delivered_to
             object.process

lib/discourse_activity_pub/ap/activity/compose.rb

@@ -9,7 +9,9 @@ def types
         end
 
         def validate_activity
-          return false unless activity_host_matches_object_host?
+          unless activity_host_matches_object_host?
+            raise DiscourseActivityPub::AP::Handlers::Warning
+          end
           super
         end
       end

lib/discourse_activity_pub/ap/handlers.rb

@@ -3,9 +3,11 @@
 module DiscourseActivityPub
   module AP
     module Handlers
-      class Error < StandardError
-        class Validate < Error
+      class Warning < StandardError
+        class Validate < Warning
         end
+      end
+      class Error < StandardError
         class Perform < Error
         end
         class Store < Error
@@ -73,7 +75,8 @@ def apply_handlers(activity, object_type, handler_type, parent: nil, raise_error
               begin
                 proc.call(activity, { parent: parent })
                 true
-              rescue DiscourseActivityPub::AP::Handlers::Error => error
+              rescue DiscourseActivityPub::AP::Handlers::Warning,
+                     DiscourseActivityPub::AP::Handlers::Error => error
                 activity.add_error(error)
                 raise_errors ? raise : false
               end

lib/discourse_activity_pub/ap/object.rb

@@ -22,7 +22,8 @@ def initialize(json: nil, stored: nil, parent: nil)
       end
 
       def id
-        stored&.ap_id
+        return stored.ap_id if stored
+        json["id"] if json
       end
 
       def type
@@ -181,7 +182,7 @@ def self.resolve_and_store(raw_object, parent = nil)
         object
       end
 
-      def self.resolve(raw_object)
+      def self.resolve(raw_object, resolve_attribution: true)
         object_id = DiscourseActivityPub::JsonLd.resolve_id(raw_object)
         return process_failed(raw_object, "cant_resolve_object") if object_id.blank?
 
@@ -205,7 +206,7 @@ def self.resolve(raw_object)
           return process_failed(resolved_object["id"], "object_not_supported")
         end
 
-        if object.json[:attributedTo]
+        if resolve_attribution && object.json[:attributedTo]
           attributed_to = Actor.resolve_and_store(object.json[:attributedTo])
           object.attributed_to = attributed_to if attributed_to.present?
         end

lib/discourse_activity_pub/ap/object/article.rb

@@ -12,7 +12,8 @@ def content
         end
 
         def in_reply_to
-          stored&.reply_to_id
+          return stored.reply_to_id if stored
+          json["inReplyTo"] if json
         end
 
         def can_belong_to

lib/discourse_activity_pub/ap/object/note.rb

@@ -12,7 +12,8 @@ def content
         end
 
         def in_reply_to
-          stored&.reply_to_id
+          return stored.reply_to_id if stored
+          json["inReplyTo"] if json
         end
 
         def can_belong_to

lib/discourse_activity_pub/context_resolver.rb

@@ -0,0 +1,109 @@
+# frozen_string_literal: true
+
+module DiscourseActivityPub
+  class ContextResolver
+    include HasErrors
+
+    REPLY_DEPTH_LIMIT = 3
+
+    attr_reader :object
+    attr_accessor :local_object, :remote_objects
+
+    def initialize(object)
+      @object = object
+    end
+
+    def perform
+      return unless resolve_context?
+
+      traverse_replies
+
+      if local_object
+        resolve_and_store_remote_objects
+      else
+        add_error(I18n.t("discourse_activity_pub.process.warning.cannot_resolve_context"))
+      end
+    end
+
+    def success?
+      errors.blank?
+    end
+
+    def self.perform(object)
+      new(object).perform
+    end
+
+    protected
+
+    def resolve_context?
+      # Resolve the context of unhandled replies without a post in reply to
+      object.model_id.blank? && object.reply_to_id.present? && object.in_reply_to_post.blank?
+    end
+
+    def traverse_replies
+      reply_depth = 1
+      reply_to_object = object.ap
+      @local_object = nil
+      @remote_objects = []
+
+      while local_object.nil? && reply_to_object&.in_reply_to.present? &&
+              reply_depth <= REPLY_DEPTH_LIMIT
+        reply_to_object =
+          DiscourseActivityPub::AP::Object.resolve(
+            reply_to_object.in_reply_to,
+            resolve_attribution: false,
+          )
+        break if reply_to_object.blank?
+
+        object = DiscourseActivityPubObject.find_by(ap_id: reply_to_object.id)
+        if object
+          # We only resolve a context in a full_topic topic
+          if object.model&.topic&.activity_pub_full_topic_enabled
+            @local_object = object
+          else
+            reply_to_object = nil
+          end
+        else
+          remote_objects << reply_to_object
+        end
+
+        reply_depth += 1
+      end
+    end
+
+    def resolve_and_store_remote_objects
+      # If anything fails everything has to be rolled back
+      ActiveRecord::Base.transaction do
+        remote_objects.each do |remote_object|
+          new_local_object =
+            DiscourseActivityPub::AP::Object.resolve_and_store(
+              remote_object.json,
+              DiscourseActivityPub::AP::Activity.factory(
+                { type: DiscourseActivityPub::AP::Activity::Create.type },
+              ),
+            )
+          rollback_remote_store if new_local_object&.stored.blank?
+
+          user =
+            DiscourseActivityPub::ActorHandler.update_or_create_user(
+              new_local_object.stored.attributed_to,
+            )
+          rollback_remote_store if user.blank?
+
+          post =
+            DiscourseActivityPub::PostHandler.create(
+              user,
+              new_local_object.stored,
+              topic_id: local_object.model.topic_id,
+            )
+          rollback_remote_store if post.blank?
+        end
+      end
+    end
+
+    def rollback_remote_store
+      add_error(I18n.t("discourse_activity_pub.process.warning.cannot_resolve_context"))
+      raise ActiveRecord::Rollback
+    end
+  end
+end

lib/discourse_activity_pub/post_handler.rb

@@ -16,7 +16,7 @@ def create(
       import_mode: false
     )
       if !user || !object || object.model_id ||
-           (!object.in_reply_to_post && !category_id && !tag_id)
+           (!object.in_reply_to_post && !category_id && !tag_id && !topic_id)
         return nil
       end
 
@@ -28,7 +28,7 @@ def create(
 
       new_topic = !object.reply_to_id && !topic_id && (category || tag)
       reply_to = object.in_reply_to_post
-      return nil if !import_mode && !new_topic && !reply_to
+      return nil if !import_mode && !new_topic && !reply_to && !topic_id
 
       params = {
         raw: object.content,
@@ -86,7 +86,7 @@ def create(
           object.update(
             model_type: "Post",
             model_id: post.id,
-            collection_id: post.topic.activity_pub_object.id,
+            collection_id: post.topic.activity_pub_object&.id,
           )
         end
       end
@@ -116,17 +116,17 @@ def self.ensure_activity_has_post(activity)
       post = activity.object.stored.model
 
       unless post
-        raise DiscourseActivityPub::AP::Handlers::Error::Validate,
+        raise DiscourseActivityPub::AP::Handlers::Warning::Validate,
               I18n.t("discourse_activity_pub.process.warning.cant_find_post")
       end
 
       if post.trashed?
-        raise DiscourseActivityPub::AP::Handlers::Error::Validate,
+        raise DiscourseActivityPub::AP::Handlers::Warning::Validate,
               I18n.t("discourse_activity_pub.process.warning.post_is_deleted")
       end
 
       unless post.activity_pub_full_topic
-        raise DiscourseActivityPub::AP::Handlers::Error::Validate,
+        raise DiscourseActivityPub::AP::Handlers::Warning::Validate,
               I18n.t("discourse_activity_pub.process.warning.full_topic_not_enabled")
       end
     end