send_message.diff

diff --git a/app/services/p2p/message/base.rb b/app/services/p2p/message/base.rb
index 5e8d961793fc..601231ed1d9f 100644
--- a/app/services/p2p/message/base.rb
+++ b/app/services/p2p/message/base.rb
@@ -24,12 +24,17 @@ module P2p
       # be sent to another Procore instance. The `resource` parameter is the resource
       # that is being shared. The `project_connection` is the target for where
       # the resource should send to. The send_message method doesn't return anything.
-      def send_message(resource:, project_connection: nil)
+      def send_message(data:, project_connection: nil, header_thread_id: nil)
         body = context(squad: squad, tool: tool) do
-          serialize(resource)
+          serialize(data)
         end
 
-        create_and_share_sent_message(body, resource, project_connection)
+        create_and_share_sent_message(
+          body,
+          data,
+          project_connection,
+          header_thread_id: header_thread_id
+        )
       rescue StandardError => e
         raise ProcoreConnect::Errors::MessageSendingError.new(
           e.message,
@@ -47,13 +52,16 @@ module P2p
           user: current_user
         )
 
-        resource = context(squad: squad, tool: tool) do
-          project_connection = ::ProjectConnection.find_by_received_message(psp_received_message)
-          # When initiating or reactivating project connections, the receiver necessarily has either a nonexistent or nonactive record.
-          if !psp_received_message.body_type.to_s.start_with?('project_connection.') && !project_connection.active?
-            raise ProcoreConnect::Errors::InvalidProjectConnectionError
-          end
+        project_connection = ::ProjectConnection.find_by_received_message(psp_received_message)
+
+        # When initiating or reactivating project connections, the receiver necessarily has either a nonexistent or nonactive record.
+        if !psp_received_message.body_type.to_s.start_with?('project_connection.') && !project_connection.active?
+          raise ProcoreConnect::Errors::InvalidProjectConnectionError
+        end
 
+        psp_thread_join.project_connection ||= ::ProjectConnection.find_by_received_message(psp_received_message)
+
+        resource = context(squad: squad, tool: tool) do
           handle(
             project_connection: project_connection,
             message: psp_received_message,
@@ -62,16 +70,18 @@ module P2p
           )
         end
 
-        psp_thread_join.project_connection ||= ::ProjectConnection.find_by_received_message(psp_received_message)
         psp_thread_join.threadable ||= resource
-        # connected_resource_metadata may be changed through the handle method
         psp_thread_join.save_with_unique_guard! if psp_thread_join.changed?
       rescue StandardError => e
+        psp_thread_join.save_with_unique_guard! if psp_thread_join.changed?
+
         error = ProcoreConnect::Errors::MessageHandlingError.new(
           e.message,
           message_type: message_type
         )
 
+        handle_failure(error: e, psp_received_message: psp_received_message)
+
         # If we simply raise the error, tool/squad attribution is lost while this gets swept up in
         # the Sidekiq retry flow. Instead we report as a handled exception and then send it off
         # to the retry cycle without being re-reported to Bugsnag.
@@ -117,6 +127,17 @@ module P2p
         raise NotImplementedError("Message must override 'handle'")
       end
 
+      def handle_failure(error:, psp_received_message:)
+        project_connection = ::ProjectConnection.find_by_received_message(psp_received_message)
+
+        ::P2p::Message::Shared::Failure.new(current_context).
+          send_message(
+            project_connection: project_connection,
+            data: error,
+            header_thread_id: psp_received_message.header_thread_id
+          )
+      end
+
       # Defines the tool that this message type belongs to (for bugsnags / reporting).
       def tool
         raise NotImplementedError("Message must override 'tool'")
@@ -157,10 +178,16 @@ module P2p
         name.constantize
       end
 
-      def create_and_share_sent_message(body, resource, project_connection)
+      def create_and_share_sent_message(body, data, project_connection, header_thread_id: nil)
+        resource = data if data.is_a?(ActiveRecord::Base)
+
         ActiveRecord::Base.transaction do
           if project_connection
-            psp_thread_join = find_or_create_psp_thread_join(resource, project_connection)
+            psp_thread_join = find_or_create_psp_thread_join(
+              resource,
+              project_connection,
+              header_thread_id: header_thread_id
+            )
           else
             # The first PspThreadJoin's connection will correspond with the original sender
             # *only* if the current sender is not themselves the original sender. At the moment,
@@ -188,11 +215,18 @@ module P2p
         end
       end
 
-      def find_or_create_psp_thread_join(resource, project_connection)
-        existing_psp_thread_join = PspThreadJoin.
-          for_threadable(resource).
-          for_project_connection(project_connection.id).
-          first
+      def find_or_create_psp_thread_join(resource, project_connection, header_thread_id: nil)
+        existing_psp_thread_join = if header_thread_id
+          PspThreadJoin.
+            for_thread_id(header_thread_id).
+            for_project_connection(project_connection).
+            first
+        else
+          PspThreadJoin.
+            for_threadable(resource).
+            for_project_connection(project_connection.id).
+            first
+        end
 
         existing_psp_thread_join || PspThreadJoin.create!(
           company: current_company,