Add XenForo direct messages importer with improved recipient handling for XF2.0+ (#31508)

Have tested the import of direct messages on Xenforo 2.3 to Discourse
latest version with half a million conversations.

script/import_scripts/xenforo.rb

@@ -309,54 +309,119 @@ class ImportScripts::XenForo < ImportScripts::Base
   end
 
   def import_private_messages
-    puts "", "importing private messages..."
-    post_count = mysql_query("SELECT COUNT(*) count FROM xf_conversation_message").first["count"]
+    puts "", "Importing private messages..."
+    total_count =
+      mysql_query("SELECT COUNT(*) AS count FROM xf_conversation_message").first["count"]
+
     batches(BATCH_SIZE) do |offset|
-      posts = mysql_query <<-SQL
-        SELECT c.conversation_id, c.recipients, c.title, m.message, m.user_id, m.message_date, m.message_id, IF(c.first_message_id != m.message_id, c.first_message_id, 0) as topic_id
-        FROM xf_conversation_master c
-        LEFT JOIN xf_conversation_message m ON m.conversation_id = c.conversation_id
-        ORDER BY c.conversation_id, m.message_id
-        LIMIT #{BATCH_SIZE}
-        OFFSET #{offset}
+      rows = mysql_query(<<-SQL).to_a
+        SELECT
+          cm.conversation_id,
+          cm.title,
+          cm.user_id AS conversation_owner,
+          cm.recipients,
+          cm.first_message_id,
+          cmsg.message_id,
+          cmsg.message,
+          cmsg.user_id AS message_user_id,
+          cmsg.message_date,
+          cmsg.username AS message_username
+        FROM xf_conversation_master cm
+        JOIN xf_conversation_message cmsg
+          ON cmsg.conversation_id = cm.conversation_id
+        ORDER BY cm.conversation_id, cmsg.message_date
+        LIMIT #{BATCH_SIZE} OFFSET #{offset}
       SQL
-      break if posts.size < 1
-      next if all_records_exist? :posts, posts.map { |post| "pm_#{post["message_id"]}" }
-      create_posts(posts, total: post_count, offset: offset) do |post|
-        user_id = user_id_from_imported_user_id(post["user_id"]) || Discourse::SYSTEM_USER_ID
-        title = post["title"]
-        message_id = "pm_#{post["message_id"]}"
-        raw = process_xenforo_post(post["message"], 0)
-        if raw.present?
-          msg = {
-            id: message_id,
-            user_id: user_id,
-            raw: raw,
-            created_at: Time.zone.at(post["message_date"].to_i),
-            import_mode: true,
-          }
-          if post["topic_id"] <= 0
-            topic_id = post["topic_id"]
-            if t = topic_lookup_from_imported_post_id("pm_#{topic_id}")
-              msg[:topic_id] = t[:topic_id]
-            else
-              puts "Topic ID #{topic_id} not found, skipping post #{post["message_id"]} from #{post["user_id"]}"
-              next
-            end
+
+      break if rows.empty?
+      next if all_records_exist?(:posts, rows.map { |row| "pm_#{row["message_id"]}" })
+
+      create_posts(rows, total: total_count, offset: offset) do |row|
+        # Determine sender: use conversation_owner for the first message; otherwise, use message_user_id.
+        mapped_user_id =
+          if row["message_id"].to_i == row["first_message_id"].to_i
+            user_id_from_imported_user_id(row["conversation_owner"])
           else
-            msg[:title] = post["title"]
-            msg[:archetype] = Archetype.private_message
-            to_user_array = PHP.unserialize(post["recipients"])
-            if to_user_array.size > 0
-              discourse_user_ids = to_user_array.keys.map { |id| user_id_from_imported_user_id(id) }
-              usernames = User.where(id: [discourse_user_ids]).pluck(:username)
-              msg[:target_usernames] = usernames.join(",")
+            user_id_from_imported_user_id(row["message_user_id"])
+          end
+
+        # If mapping fails, fall back to the system user.
+        unless User.exists?(id: mapped_user_id)
+          puts "Row #{row["message_id"]}: User mapping invalid for XenForo user #{row["message_user_id"]} or conversation owner #{row["conversation_owner"]}, using system user."
+          mapped_user_id = Discourse::SYSTEM_USER_ID
+        end
+
+        # Process the message body and timestamp.
+        raw = process_xenforo_post(row["message"], row["message_id"])
+        created_at = Time.zone.at(row["message_date"].to_i)
+
+        if row["message_id"].to_i == row["first_message_id"].to_i
+          # For the first message (PM topic), get target recipients.
+          recipients_raw =
+            (
+              begin
+                PHP.unserialize(row["recipients"])
+              rescue StandardError
+                nil
+              end
+            )
+          if recipients_raw.blank?
+            begin
+              recipients_raw = JSON.parse(row["recipients"])
+            rescue StandardError
+              recipients_raw = {}
             end
           end
-          msg
+
+          # Fallback: if still empty, query the recipient table.
+          if recipients_raw.blank? || recipients_raw.empty?
+            recipient_rows =
+              mysql_query(
+                "SELECT user_id FROM #{TABLE_PREFIX}conversation_recipient WHERE conversation_id = #{row["conversation_id"]} AND recipient_state = 'active'",
+              ).to_a
+            if recipient_rows.any?
+              recipients_raw = {}
+              recipient_rows.each { |r| recipients_raw[r["user_id"].to_s] = true }
+            end
+          end
+
+          target_usernames =
+            if recipients_raw.present? && recipients_raw.keys.any?
+              discourse_user_ids =
+                recipients_raw.keys.map { |id| user_id_from_imported_user_id(id.to_s) }
+              User.where(id: discourse_user_ids).pluck(:username).join(",")
+            end
+
+          if target_usernames.blank?
+            puts "Row #{row["message_id"]}: No valid target usernames found, skipping conversation."
+            next
+          end
+
+          {
+            id: "pm_#{row["message_id"]}",
+            user_id: mapped_user_id,
+            title: row["title"].presence || "No title",
+            raw: raw,
+            created_at: created_at,
+            archetype: Archetype.private_message,
+            import_mode: true,
+            target_usernames: target_usernames,
+          }
         else
-          puts "Empty message, skipping post #{post["message_id"]}"
-          next
+          # For replies, find the topic by the first message.
+          parent = topic_lookup_from_imported_post_id("pm_#{row["first_message_id"]}")
+          unless parent
+            puts "Row #{row["message_id"]}: Topic for conversation #{row["conversation_id"]} not found, skipping message."
+            next
+          end
+          {
+            id: "pm_#{row["message_id"]}",
+            user_id: mapped_user_id,
+            raw: raw,
+            created_at: created_at,
+            topic_id: parent[:topic_id],
+            import_mode: true,
+          }
         end
       end
     end