DEV: Add support for converting and importing topic_tags (#35041)

This adds converter(Discourse-only, for now) and importer steps for
`topic_tags`.

migrations/config/intermediate_db.yml

@@ -154,6 +154,11 @@ schema:
           - "highest_staff_post_number"
           - "reviewable_score"
           - "slow_mode_seconds"
+    topic_tags:
+      primary_key_column_names: [ "topic_id", "tag_id" ]
+      columns:
+        exclude:
+          - "id"
     user_associated_accounts:
       primary_key_column_names: [ "user_id", "provider_name" ]
       columns:
@@ -466,7 +471,6 @@ schema:
         - "topic_links"
         - "topic_localizations"
         - "topic_search_data"
-        - "topic_tags"
         - "topic_thumbnails"
         - "topic_timers"
         - "topic_users"

migrations/db/intermediate_db_schema/100-base-schema.sql

@@ -206,6 +206,14 @@ CREATE TABLE topic_allowed_users
     PRIMARY KEY (topic_id, user_id)
 );
 
+CREATE TABLE topic_tags
+(
+    tag_id     NUMERIC  NOT NULL,
+    topic_id   NUMERIC  NOT NULL,
+    created_at DATETIME,
+    PRIMARY KEY (topic_id, tag_id)
+);
+
 CREATE TABLE topics
 (
     original_id          NUMERIC  NOT NULL PRIMARY KEY,

migrations/lib/converters/discourse/steps/topic_tags.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+module Migrations::Converters::Discourse
+  class TopicTags < ::Migrations::Converters::Base::ProgressStep
+    attr_accessor :source_db
+
+    def max_progress
+      @source_db.count <<~SQL
+        SELECT COUNT(*)
+        FROM topic_tags
+      SQL
+    end
+
+    def items
+      @source_db.query <<~SQL
+        SELECT *
+        FROM topic_tags
+        ORDER BY topic_id, tag_id
+      SQL
+    end
+
+    def process_item(item)
+      IntermediateDB::TopicTag.create(
+        topic_id: item[:topic_id],
+        tag_id: item[:tag_id],
+        created_at: item[:created_at],
+      )
+    end
+  end
+end

migrations/lib/database/intermediate_db/topic_tag.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+# This file is auto-generated from the IntermediateDB schema. To make changes,
+# update the "config/intermediate_db.yml" configuration file and then run
+# `bin/cli schema generate` to regenerate this file.
+
+module Migrations::Database::IntermediateDB
+  module TopicTag
+    SQL = <<~SQL
+      INSERT INTO topic_tags (
+        tag_id,
+        topic_id,
+        created_at
+      )
+      VALUES (
+        ?, ?, ?
+      )
+    SQL
+    private_constant :SQL
+
+    def self.create(tag_id:, topic_id:, created_at: nil)
+      ::Migrations::Database::IntermediateDB.insert(
+        SQL,
+        tag_id,
+        topic_id,
+        ::Migrations::Database.format_datetime(created_at),
+      )
+    end
+  end
+end

migrations/lib/importer/steps/topic_tags.rb

@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+module Migrations::Importer::Steps
+  class TopicTags < ::Migrations::Importer::CopyStep
+    depends_on :topics, :tags
+
+    requires_set :existing_topic_tags, "SELECT topic_id, tag_id FROM topic_tags"
+
+    column_names %i[topic_id tag_id created_at updated_at]
+
+    total_rows_query <<~SQL, MappingType::TOPICS, MappingType::TAGS
+      SELECT COUNT(*)
+        FROM topic_tags
+             JOIN mapped.ids mapped_topic
+               ON topic_tags.topic_id = mapped_topic.original_id AND mapped_topic.type = ?1
+             JOIN mapped.ids mapped_tag
+               ON topic_tags.tag_id = mapped_tag.original_id AND mapped_tag.type = ?2
+    SQL
+
+    rows_query <<~SQL, MappingType::TOPICS, MappingType::TAGS
+      SELECT topic_tags.*,
+             mapped_topic.discourse_id AS discourse_topic_id,
+             mapped_tag.discourse_id   AS discourse_tag_id
+        FROM topic_tags
+             JOIN mapped.ids mapped_topic
+               ON topic_tags.topic_id = mapped_topic.original_id AND mapped_topic.type = ?1
+             JOIN mapped.ids mapped_tag
+               ON topic_tags.tag_id = mapped_tag.original_id AND mapped_tag.type = ?2
+      ORDER BY topic_tags.topic_id, topic_tags.tag_id
+    SQL
+
+    private
+
+    def transform_row(row)
+      topic_id = row[:discourse_topic_id]
+      tag_id = row[:discourse_tag_id]
+
+      return nil unless @existing_topic_tags.add?(topic_id, tag_id)
+
+      row[:topic_id] = topic_id
+      row[:tag_id] = tag_id
+
+      super
+    end
+  end
+end