descrybe/apps/api/cmd/migrator/catalog_feeds.go

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"strconv"
	"strings"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

func migrateCatalogAndFeeds(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap, userMap map[string]string,
	allow map[string]bool,
	domains domainSet,
	report map[string]int,
	dryRun bool,
) (categoryMap, attributeMap, feedMap, rawMap, fileMap map[string]string) {
	categoryMap = map[string]string{}
	attributeMap = map[string]string{}
	feedMap = map[string]string{}
	rawMap = map[string]string{}
	fileMap = map[string]string{}

	// Feeds before products so feed_id can be remapped when possible.
	if domains.has("feeds") {
		migrateInputFeeds(ctx, mysqlDB, pg, companyMap, feedMap, allow, report, dryRun)
	}
	if domains.has("files") {
		migrateFiles(ctx, mysqlDB, pg, companyMap, userMap, fileMap, allow, report, dryRun)
	}
	if domains.has("catalog") {
		migrateCategories(ctx, mysqlDB, pg, companyMap, categoryMap, allow, report, dryRun)
		migrateAttributes(ctx, mysqlDB, pg, companyMap, attributeMap, allow, report, dryRun)
		migrateCategoryAttributes(ctx, mysqlDB, pg, companyMap, attributeMap, allow, report, dryRun)
		migrateCustomVariables(ctx, mysqlDB, pg, companyMap, allow, report, dryRun)
	}
	if domains.has("products") {
		migrateRawProducts(ctx, mysqlDB, pg, companyMap, feedMap, rawMap, allow, report, dryRun)
		migrateProcessedProducts(ctx, mysqlDB, pg, companyMap, feedMap, rawMap, allow, report, dryRun)
		backfillProcessedDescriptionsFromMapped(ctx, pg, companyMap, allow, report, dryRun)
		backfillProcessedNamesFromMapped(ctx, pg, companyMap, allow, report, dryRun)
	}
	if domains.has("feeds") {
		migrateExportFeeds(ctx, mysqlDB, pg, companyMap, feedMap, allow, report, dryRun)
	}

	return categoryMap, attributeMap, feedMap, rawMap, fileMap
}

func migrateCategories(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap, categoryMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if !mysqlTableExists(ctx, mysqlDB, "categories") {
		log.Printf("categories skipped: table missing")
		return
	}
	q := mysqlSelectList(
		"id",
		"company_id",
		"name",
		"unique_id",
		mysqlCol(ctx, mysqlDB, "categories", "parent_id", "NULL"),
		mysqlCol(ctx, mysqlDB, "categories", "path", "NULL"),
		mysqlCoalesce(ctx, mysqlDB, "categories", "level", "0"),
		mysqlCoalesce(ctx, mysqlDB, "categories", "position", "0"),
		mysqlCoalesce(ctx, mysqlDB, "categories", "is_active", "1"),
		mysqlCol(ctx, mysqlDB, "categories", "description", "NULL"),
		mysqlCol(ctx, mysqlDB, "categories", "prompt", "NULL"),
		mysqlCol(ctx, mysqlDB, "categories", "metadata", "NULL"),
		mysqlCol(ctx, mysqlDB, "categories", "config", "NULL"),
		mysqlCol(ctx, mysqlDB, "categories", "title_template", "NULL"),
		mysqlCol(ctx, mysqlDB, "categories", "description_template", "NULL"),
	) + " FROM categories WHERE 1=1"
	clause, cargs := mysqlCompanyFilter("company_id", allow)
	q += clause
	rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
	if err != nil {
		rows, err = mysqlDB.QueryContext(ctx, `
			SELECT id, company_id, name, unique_id, NULL, NULL, 0, 0, 1,
				NULL, NULL, NULL, NULL, NULL, NULL
			FROM categories WHERE 1=1`+clause, cargs...)
	}
	if err != nil {
		log.Printf("categories skipped: %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var legacyID int64
		var companyLegacy, name, uniqueID string
		var parentID, path, desc, prompt sql.NullString
		var level, position, isActive int
		var metadata, config, titleTpl, descTpl []byte
		if err := rows.Scan(&legacyID, &companyLegacy, &name, &uniqueID, &parentID, &path,
			&level, &position, &isActive, &desc, &prompt, &metadata, &config, &titleTpl, &descTpl); err != nil {
			log.Printf("category scan: %v", err)
			report["categories_skipped"]++
			continue
		}
		cid, ok := companyMap[companyLegacy]
		if !ok {
			report["categories_skipped"]++
			continue
		}
		newID := uuid.New()
		categoryMap[strconv.FormatInt(legacyID, 10)] = newID.String()
		if dryRun {
			report["categories"]++
			continue
		}
		_, err = pg.Exec(ctx, `
			INSERT INTO categories (
				id, company_id, name, unique_id, parent_unique_id, path, level, position,
				is_active, description, prompt, metadata, config, title_template, description_template
			) VALUES (
				$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11,
				COALESCE($12::jsonb, '{}'::jsonb), COALESCE($13::jsonb, '{}'::jsonb), $14::jsonb, $15::jsonb
			)
			ON CONFLICT (company_id, unique_id) DO UPDATE SET
				name = EXCLUDED.name,
				parent_unique_id = EXCLUDED.parent_unique_id,
				title_template = EXCLUDED.title_template,
				description_template = EXCLUDED.description_template,
				updated_at = now()`,
			newID, cid, name, uniqueID, nullString(parentID), nullString(path), level, position,
			isActive == 1, nullString(desc), nullString(prompt),
			jsonOrNull(metadata), jsonOrNull(config), jsonOrNull(titleTpl), jsonOrNull(descTpl))
		if err != nil {
			log.Printf("category insert %d: %v", legacyID, err)
			var existing uuid.UUID
			if err2 := pg.QueryRow(ctx, `SELECT id FROM categories WHERE company_id = $1 AND unique_id = $2`, cid, uniqueID).Scan(&existing); err2 == nil {
				categoryMap[strconv.FormatInt(legacyID, 10)] = existing.String()
			} else {
				report["categories_skipped"]++
				continue
			}
		} else {
			var existing uuid.UUID
			_ = pg.QueryRow(ctx, `SELECT id FROM categories WHERE company_id = $1 AND unique_id = $2`, cid, uniqueID).Scan(&existing)
			if existing != uuid.Nil {
				categoryMap[strconv.FormatInt(legacyID, 10)] = existing.String()
			}
		}
		report["categories"]++
	}
}

func migrateAttributes(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap, attributeMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if !mysqlTableExists(ctx, mysqlDB, "attributes") {
		log.Printf("attributes skipped: table missing")
		return
	}
	q := mysqlSelectList(
		"id",
		"company_id",
		"attribute_key",
		"name",
		mysqlCoalesce(ctx, mysqlDB, "attributes", "value_type", "'string'"),
		mysqlCol(ctx, mysqlDB, "attributes", "unit", "NULL"),
		mysqlCol(ctx, mysqlDB, "attributes", "example", "NULL"),
		mysqlCol(ctx, mysqlDB, "attributes", "parent_key", "NULL"),
	) + " FROM attributes WHERE 1=1"
	clause, cargs := mysqlCompanyFilter("company_id", allow)
	q += clause
	rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
	if err != nil {
		rows, err = mysqlDB.QueryContext(ctx, `
			SELECT id, company_id, attribute_key, name, 'string', NULL, NULL, NULL FROM attributes WHERE 1=1`+clause, cargs...)
	}
	if err != nil {
		log.Printf("attributes skipped: %v", err)
		return
	}
	defer rows.Close()

	const batchSize = 400
	type pending struct {
		legacyID                                 int64
		cid, key, name, valueType                string
		unit, example, parentKey                 *string
		newID                                    uuid.UUID
	}
	flush := func(batch []pending) {
		if len(batch) == 0 {
			return
		}
		if dryRun {
			report["attributes"] += len(batch)
			return
		}
		b := &pgx.Batch{}
		for _, p := range batch {
			b.Queue(`
				INSERT INTO attributes (id, company_id, attribute_key, name, value_type, unit, example, parent_key)
				VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
				ON CONFLICT (company_id, attribute_key) DO UPDATE SET
					name = EXCLUDED.name, value_type = EXCLUDED.value_type, updated_at = now()`,
				p.newID, p.cid, p.key, p.name, p.valueType, p.unit, p.example, p.parentKey)
		}
		br := pg.SendBatch(ctx, b)
		if err := br.Close(); err != nil {
			log.Printf("attributes batch: %v — resolving ids individually", err)
			for _, p := range batch {
				var existing uuid.UUID
				if err2 := pg.QueryRow(ctx, `SELECT id FROM attributes WHERE company_id = $1 AND attribute_key = $2`, p.cid, p.key).Scan(&existing); err2 == nil {
					attributeMap[strconv.FormatInt(p.legacyID, 10)] = existing.String()
					report["attributes"]++
				} else {
					report["attributes_skipped"]++
				}
			}
			return
		}
		for _, p := range batch {
			var existing uuid.UUID
			if err := pg.QueryRow(ctx, `SELECT id FROM attributes WHERE company_id = $1 AND attribute_key = $2`, p.cid, p.key).Scan(&existing); err == nil {
				attributeMap[strconv.FormatInt(p.legacyID, 10)] = existing.String()
			}
			report["attributes"]++
		}
	}

	batch := make([]pending, 0, batchSize)
	for rows.Next() {
		var legacyID int64
		var companyLegacy, key, name, valueType string
		var unit, example, parentKey sql.NullString
		if err := rows.Scan(&legacyID, &companyLegacy, &key, &name, &valueType, &unit, &example, &parentKey); err != nil {
			log.Printf("attribute scan: %v", err)
			report["attributes_skipped"]++
			continue
		}
		cid, ok := companyMap[companyLegacy]
		if !ok {
			report["attributes_skipped"]++
			continue
		}
		if valueType == "" {
			valueType = "string"
		}
		newID := uuid.New()
		attributeMap[strconv.FormatInt(legacyID, 10)] = newID.String()
		batch = append(batch, pending{
			legacyID: legacyID, cid: cid, key: key, name: name, valueType: valueType,
			unit: nullString(unit), example: nullString(example), parentKey: nullString(parentKey),
			newID: newID,
		})
		if len(batch) >= batchSize {
			flush(batch)
			batch = batch[:0]
		}
	}
	flush(batch)
}

func migrateCategoryAttributes(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap, attributeMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if !mysqlTableExists(ctx, mysqlDB, "category_attributes") {
		log.Printf("category_attributes skipped: table missing")
		return
	}
	clause, cargs := mysqlCompanyFilter("company_id", allow)
	rows, err := mysqlDB.QueryContext(ctx, `
		SELECT company_id, category_id, attribute_id, COALESCE(required, 0)
		FROM category_attributes WHERE 1=1`+clause, cargs...)
	if err != nil {
		log.Printf("category_attributes skipped: %v", err)
		return
	}
	defer rows.Close()
	const batchSize = 500
	batch := &pgx.Batch{}
	pending := 0
	flush := func() {
		if dryRun || pending == 0 {
			return
		}
		br := pg.SendBatch(ctx, batch)
		if err := br.Close(); err != nil {
			log.Printf("category_attributes batch: %v", err)
		}
		batch = &pgx.Batch{}
		pending = 0
	}
	for rows.Next() {
		var companyLegacy, categoryUnique string
		var attrLegacy int64
		var required int
		if err := rows.Scan(&companyLegacy, &categoryUnique, &attrLegacy, &required); err != nil {
			log.Printf("category_attribute scan: %v", err)
			report["category_attributes_skipped"]++
			continue
		}
		cid, okC := companyMap[companyLegacy]
		attrID, okA := attributeMap[strconv.FormatInt(attrLegacy, 10)]
		if !okC || !okA {
			report["category_attributes_skipped"]++
			continue
		}
		if dryRun {
			report["category_attributes"]++
			continue
		}
		batch.Queue(`
			INSERT INTO category_attributes (company_id, category_unique_id, attribute_id, required)
			VALUES ($1, $2, $3, $4)
			ON CONFLICT (company_id, category_unique_id, attribute_id) DO UPDATE SET required = EXCLUDED.required`,
			cid, categoryUnique, attrID, required == 1)
		pending++
		report["category_attributes"]++
		if pending >= batchSize {
			flush()
		}
	}
	flush()
}

func migrateCustomVariables(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if !mysqlTableExists(ctx, mysqlDB, "custom_variables") {
		log.Printf("custom_variables skipped: table missing")
		return
	}
	// Legacy has name/label/description/example; v2 has name/value/description.
	clause, cargs := mysqlCompanyFilter("company_id", allow)
	rows, err := mysqlDB.QueryContext(ctx, `
		SELECT company_id, name,
			COALESCE(NULLIF(label, ''), NULLIF(example, ''), ''),
			description
		FROM custom_variables WHERE 1=1`+clause, cargs...)
	if err != nil {
		rows, err = mysqlDB.QueryContext(ctx, `
			SELECT company_id, name, COALESCE(value, ''), description
			FROM custom_variables WHERE 1=1`+clause, cargs...)
	}
	if err != nil {
		log.Printf("custom_variables skipped: %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var companyLegacy, name, value string
		var desc sql.NullString
		if err := rows.Scan(&companyLegacy, &name, &value, &desc); err != nil {
			log.Printf("custom_variable scan: %v", err)
			report["custom_variables_skipped"]++
			continue
		}
		cid, ok := companyMap[companyLegacy]
		if !ok {
			report["custom_variables_skipped"]++
			continue
		}
		if dryRun {
			report["custom_variables"]++
			continue
		}
		_, err = pg.Exec(ctx, `
			INSERT INTO custom_variables (company_id, name, value, description)
			VALUES ($1, $2, $3, $4)
			ON CONFLICT (company_id, name) DO UPDATE SET
				value = EXCLUDED.value, description = EXCLUDED.description, updated_at = now()`,
			cid, name, value, nullString(desc))
		if err != nil {
			log.Printf("custom_variable: %v", err)
			report["custom_variables_skipped"]++
			continue
		}
		report["custom_variables"]++
	}
}

func migrateInputFeeds(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap, feedMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if mysqlTableExists(ctx, mysqlDB, "xml_feeds") {
		hasMappings := mysqlColumnExists(ctx, mysqlDB, "xml_feeds", "field_mappings")
		hasVersion := mysqlColumnExists(ctx, mysqlDB, "xml_feeds", "mapping_version")
		hasSourceType := mysqlColumnExists(ctx, mysqlDB, "xml_feeds", "source_type")
		q := `SELECT id, company_id, name, url, COALESCE(status, 'active'), COALESCE(sync_frequency, 24)`
		if hasMappings {
			q += `, field_mappings`
		} else {
			q += `, NULL`
		}
		if hasVersion {
			q += `, COALESCE(mapping_version, 1)`
		} else {
			q += `, 1`
		}
		if hasSourceType {
			q += `, COALESCE(source_type, 'url')`
		} else {
			q += `, 'url'`
		}
		clause, cargs := mysqlCompanyFilter("company_id", allow)
		q += ` FROM xml_feeds WHERE 1=1` + clause
		rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
		if err != nil {
			log.Printf("xml_feeds skipped: %v", err)
		} else {
			defer rows.Close()
			for rows.Next() {
				var legacyID int64
				var companyLegacy, name string
				var url, status sql.NullString
				var syncFreq, mapVersion int
				var fieldMappings []byte
				var sourceType string
				if err := rows.Scan(&legacyID, &companyLegacy, &name, &url, &status, &syncFreq, &fieldMappings, &mapVersion, &sourceType); err != nil {
					log.Printf("xml_feed scan: %v", err)
					report["input_feeds_skipped"]++
					continue
				}
				cid, ok := companyMap[companyLegacy]
				if !ok {
					report["input_feeds_skipped"]++
					continue
				}
				st := "active"
				if status.Valid && status.String != "" {
					st = status.String
				}
				interval := syncFreq * 60
				if interval <= 0 {
					interval = 60
				}
				feedType := "xml"
				if sourceType == "csv" || sourceType == "file" {
					feedType = sourceType
				}
				newID := uuid.New()
				feedMap[strconv.FormatInt(legacyID, 10)] = newID.String()
				if dryRun {
					report["input_feeds"]++
					if len(fieldMappings) > 0 {
						report["feed_mappings"]++
					}
					continue
				}
				_, err = pg.Exec(ctx, `
					INSERT INTO input_feeds (id, company_id, name, url, feed_type, status, sync_interval_minutes)
					VALUES ($1, $2, $3, $4, $5, $6, $7)`,
					newID, cid, name, nullString(url), feedType, st, interval)
				if err != nil {
					log.Printf("input_feed insert %d: %v", legacyID, err)
					report["input_feeds_skipped"]++
					delete(feedMap, strconv.FormatInt(legacyID, 10))
					continue
				}
				report["input_feeds"]++
				if err := insertFeedMappings(ctx, pg, newID, cid, fieldMappings, mapVersion, report); err != nil {
					log.Printf("feed_mappings insert %d: %v", legacyID, err)
				}
			}
			return
		}
	} else {
		log.Printf("xml_feeds skipped: table missing")
	}

	if !mysqlTableExists(ctx, mysqlDB, "product_feeds") {
		log.Printf("product_feeds skipped: table missing")
		return
	}
	rows, err := mysqlDB.QueryContext(ctx, `
		SELECT id, feed_name FROM product_feeds`)
	if err != nil {
		log.Printf("product_feeds skipped: %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var legacyID int64
		var feedName sql.NullString
		if err := rows.Scan(&legacyID, &feedName); err != nil {
			report["input_feeds_skipped"]++
			continue
		}
		name := feedName.String
		if name == "" {
			name = fmt.Sprintf("product_feed_%d", legacyID)
		}
		// product_feeds has no company_id — attach to first mapped company when only one, else skip.
		if len(companyMap) != 1 {
			report["input_feeds_skipped"]++
			continue
		}
		var cid string
		for _, v := range companyMap {
			cid = v
			break
		}
		newID := uuid.New()
		feedMap[strconv.FormatInt(legacyID, 10)] = newID.String()
		if dryRun {
			report["input_feeds"]++
			continue
		}
		_, err = pg.Exec(ctx, `
			INSERT INTO input_feeds (id, company_id, name, feed_type, status)
			VALUES ($1, $2, $3, 'xml', 'active')`, newID, cid, name)
		if err != nil {
			log.Printf("product_feed insert %d: %v", legacyID, err)
			delete(feedMap, strconv.FormatInt(legacyID, 10))
			report["input_feeds_skipped"]++
			continue
		}
		report["input_feeds"]++
	}
}

func migrateRawProducts(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap, feedMap, rawMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if !mysqlTableExists(ctx, mysqlDB, "raw_products") {
		log.Printf("raw_products skipped: table missing")
		return
	}
	q := mysqlSelectList(
		"id",
		"company_id",
		"gtin",
		mysqlCol(ctx, mysqlDB, "raw_products", "feed_id", "NULL"),
		mysqlCol(ctx, mysqlDB, "raw_products", "feed_ids", "NULL"),
		mysqlCoalesce(ctx, mysqlDB, "raw_products", "raw_data", "'{}'"),
		mysqlCol(ctx, mysqlDB, "raw_products", "mapped_data", "NULL"),
		mysqlCoalesce(ctx, mysqlDB, "raw_products", "is_processed", "0"),
		mysqlCoalesce(ctx, mysqlDB, "raw_products", "processing_status", "'unprocessed'"),
	) + " FROM raw_products WHERE 1=1"
	clause, cargs := mysqlCompanyFilter("company_id", allow)
	q += clause
	rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
	if err != nil {
		rows, err = mysqlDB.QueryContext(ctx, `
			SELECT id, company_id, gtin, feed_id, NULL, raw_data, NULL, 0, 'unprocessed'
			FROM raw_products WHERE 1=1`+clause, cargs...)
	}
	if err != nil {
		log.Printf("raw_products skipped: %v", err)
		return
	}
	defer rows.Close()

	const batchSize = 500
	type pending struct {
		legacyID                     int64
		cid, gtin, status            string
		feedID                       *uuid.UUID
		feedIDs, rawData, mappedData []byte
		isProcessed                  bool
		newID                        uuid.UUID
	}
	flush := func(batch []pending) {
		if len(batch) == 0 {
			return
		}
		if dryRun {
			report["raw_products"] += len(batch)
			return
		}
		tx, err := pg.Begin(ctx)
		if err != nil {
			log.Printf("raw_products tx: %v", err)
			report["raw_products_skipped"] += len(batch)
			return
		}
		defer tx.Rollback(ctx)

		copyRows := make([][]any, 0, len(batch))
		for _, p := range batch {
			copyRows = append(copyRows, []any{
				p.newID, p.cid, p.gtin, p.feedID,
				jsonOrNull(p.feedIDs), string(ensureJSON(p.rawData)), string(ensureJSON(p.mappedData)),
				p.isProcessed, p.status,
			})
		}
		_, err = tx.CopyFrom(ctx,
			pgx.Identifier{"raw_products"},
			[]string{"id", "company_id", "gtin", "feed_id", "feed_ids", "raw_data", "mapped_data", "is_processed", "processing_status"},
			pgx.CopyFromRows(copyRows),
		)
		if err != nil {
			log.Printf("raw_products copy: %v — falling back to per-row insert", err)
			for _, p := range batch {
				var inserted uuid.UUID
				err = tx.QueryRow(ctx, `
					INSERT INTO raw_products (
						id, company_id, gtin, feed_id, feed_ids, raw_data, mapped_data,
						is_processed, processing_status
					) VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7::jsonb, $8, $9)
					ON CONFLICT (company_id, gtin) DO UPDATE SET
						feed_id = COALESCE(EXCLUDED.feed_id, raw_products.feed_id),
						raw_data = EXCLUDED.raw_data,
						mapped_data = EXCLUDED.mapped_data,
						is_processed = EXCLUDED.is_processed,
						processing_status = EXCLUDED.processing_status,
						updated_at = now()
					RETURNING id`,
					p.newID, p.cid, p.gtin, p.feedID,
					jsonOrNull(p.feedIDs), string(ensureJSON(p.rawData)), string(ensureJSON(p.mappedData)),
					p.isProcessed, p.status).Scan(&inserted)
				if err != nil {
					log.Printf("raw_product insert %d: %v", p.legacyID, err)
					report["raw_products_skipped"]++
					continue
				}
				rawMap[strconv.FormatInt(p.legacyID, 10)] = inserted.String()
				report["raw_products"]++
			}
		} else {
			report["raw_products"] += len(batch)
		}
		if err := tx.Commit(ctx); err != nil {
			log.Printf("raw_products commit: %v", err)
		}
	}

	batch := make([]pending, 0, batchSize)
	gtinOwner := map[string]string{} // companyUUID|gtin → new UUID string (dedupe for PG unique index)
	for rows.Next() {
		var legacyID int64
		var companyLegacy string
		var gtin, status sql.NullString
		var feedID sql.NullInt64
		var feedIDs, rawData, mappedData []byte
		var isProcessed int
		if err := rows.Scan(&legacyID, &companyLegacy, &gtin, &feedID, &feedIDs, &rawData, &mappedData, &isProcessed, &status); err != nil {
			log.Printf("raw_product scan: %v", err)
			report["raw_products_skipped"]++
			continue
		}
		cid, ok := companyMap[companyLegacy]
		if !ok {
			report["raw_products_skipped"]++
			continue
		}
		gtinVal := ""
		if gtin.Valid {
			gtinVal = strings.TrimSpace(gtin.String)
		}
		if gtinVal == "" {
			// PG requires NOT NULL gtin; synthesize a stable placeholder from legacy id.
			gtinVal = fmt.Sprintf("legacy-missing-%d", legacyID)
			report["raw_products_gtin_synthesized"]++
		}
		statusVal := "unprocessed"
		if status.Valid && strings.TrimSpace(status.String) != "" {
			statusVal = strings.TrimSpace(status.String)
		}
		switch statusVal {
		case "unprocessed", "processing", "processed", "failed":
		default:
			statusVal = "unprocessed"
			report["raw_products_status_normalized"]++
		}
		rawObj := map[string]any{}
		if len(rawData) > 0 {
			_ = json.Unmarshal(rawData, &rawObj)
		}
		var mappedFeed *uuid.UUID
		if feedID.Valid {
			rawObj["_legacy_feed_id"] = feedID.Int64
			if fid, ok := feedMap[strconv.FormatInt(feedID.Int64, 10)]; ok {
				u, err := uuid.Parse(fid)
				if err == nil {
					mappedFeed = &u
				}
			}
		}
		rawBytes, _ := json.Marshal(rawObj)
		dedupeKey := cid + "|" + gtinVal
		if existing, ok := gtinOwner[dedupeKey]; ok {
			rawMap[strconv.FormatInt(legacyID, 10)] = existing
			report["raw_products_gtin_deduped"]++
			continue
		}
		newID := uuid.New()
		gtinOwner[dedupeKey] = newID.String()
		rawMap[strconv.FormatInt(legacyID, 10)] = newID.String()
		batch = append(batch, pending{
			legacyID: legacyID, cid: cid, gtin: gtinVal, status: statusVal,
			feedID: mappedFeed, feedIDs: feedIDs, rawData: rawBytes, mappedData: mappedData,
			isProcessed: isProcessed == 1,
			newID:       newID,
		})
		if len(batch) >= batchSize {
			flush(batch)
			batch = batch[:0]
		}
	}
	flush(batch)
}

func migrateProcessedProducts(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap, feedMap, rawMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if !mysqlTableExists(ctx, mysqlDB, "processed_products") {
		log.Printf("processed_products skipped: table missing")
		return
	}
	q := mysqlSelectList(
		"id",
		mysqlCol(ctx, mysqlDB, "processed_products", "company_id", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "product_id", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "name", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "category", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "description", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "processed_description", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "attributes", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "processed_attributes", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "status", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "gpt_response", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "total_tokens", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "feed_id", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "raw_product_id", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "processed_name", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "meta_title", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "meta_description", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "structured_description", "NULL"),
		mysqlCol(ctx, mysqlDB, "processed_products", "field_sources", "NULL"),
	) + " FROM processed_products WHERE 1=1"
	clause, cargs := mysqlCompanyFilter("company_id", allow)
	q += clause
	rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
	if err != nil {
		rows, err = mysqlDB.QueryContext(ctx, `
			SELECT id, company_id, product_id, name, category, description, processed_description,
				attributes, processed_attributes, status, gpt_response, total_tokens,
				feed_id, raw_product_id, NULL, NULL, NULL, NULL, NULL
			FROM processed_products WHERE 1=1`+clause, cargs...)
	}
	if err != nil {
		log.Printf("processed_products skipped: %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var legacyID int64
		var companyLegacy sql.NullString
		var productID, name, category, desc, procDesc, status, procName, metaTitle, metaDesc sql.NullString
		var attrs, procAttrs, gptResp, structured, fieldSources []byte
		var totalTokens sql.NullInt64
		var feedID, rawProductID sql.NullInt64
		if err := rows.Scan(&legacyID, &companyLegacy, &productID, &name, &category, &desc, &procDesc,
			&attrs, &procAttrs, &status, &gptResp, &totalTokens, &feedID, &rawProductID,
			&procName, &metaTitle, &metaDesc, &structured, &fieldSources); err != nil {
			log.Printf("processed_product scan: %v", err)
			report["processed_products_skipped"]++
			continue
		}
		if !companyLegacy.Valid || companyLegacy.String == "" {
			report["processed_products_skipped"]++
			continue
		}
		cid, ok := companyMap[companyLegacy.String]
		if !ok {
			report["processed_products_skipped"]++
			continue
		}
		var mappedFeed, mappedRaw *uuid.UUID
		if feedID.Valid {
			if fid, ok := feedMap[strconv.FormatInt(feedID.Int64, 10)]; ok {
				u, err := uuid.Parse(fid)
				if err == nil {
					mappedFeed = &u
				}
			}
		}
		if rawProductID.Valid {
			if rid, ok := rawMap[strconv.FormatInt(rawProductID.Int64, 10)]; ok {
				u, err := uuid.Parse(rid)
				if err == nil {
					mappedRaw = &u
				}
			}
		}
		if dryRun {
			report["processed_products"]++
			continue
		}
		var tokens *int
		if totalTokens.Valid {
			t := int(totalTokens.Int64)
			tokens = &t
		}
		_, err = pg.Exec(ctx, `
			INSERT INTO processed_products (
				company_id, product_id, name, category, description, processed_description,
				attributes, processed_attributes, status, gpt_response, total_tokens,
				feed_id, raw_product_id, processed_name, meta_title, meta_description,
				structured_description, field_sources
			) VALUES (
				$1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9, $10::jsonb, $11,
				$12, $13, $14, $15, $16, $17::jsonb, $18::jsonb
			)`,
			cid, nullString(productID), nullString(name), nullString(category), nullString(desc), nullString(procDesc),
			jsonOrNull(attrs), jsonOrNull(procAttrs), nullString(status), jsonOrNull(gptResp), tokens,
			mappedFeed, mappedRaw, nullString(procName), nullString(metaTitle), nullString(metaDesc),
			jsonOrNull(structured), jsonOrNull(fieldSources))
		if err != nil {
			log.Printf("processed_product insert %d: %v", legacyID, err)
			report["processed_products_skipped"]++
			continue
		}
		report["processed_products"]++
	}
}

// backfillProcessedDescriptionsFromMapped fills empty processed_products.description from the
// linked raw mapped_data.description. Legacy MySQL often left description blank while the feed
// original lived only on raw_products.mapped_data.
func backfillProcessedDescriptionsFromMapped(
	ctx context.Context,
	pg *pgxpool.Pool,
	companyMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if dryRun || pg == nil {
		return
	}
	cids := make([]uuid.UUID, 0, len(companyMap))
	for legacy, cid := range companyMap {
		if len(allow) > 0 && !allow[legacy] {
			continue
		}
		u, err := uuid.Parse(cid)
		if err != nil {
			continue
		}
		cids = append(cids, u)
	}
	if len(cids) == 0 {
		return
	}
	ct, err := pg.Exec(ctx, `
		UPDATE processed_products p
		SET description = NULLIF(r.mapped_data->>'description', ''),
			updated_at = now()
		FROM raw_products r
		WHERE p.raw_product_id = r.id
		  AND p.company_id = ANY($1::uuid[])
		  AND COALESCE(p.description, '') = ''
		  AND COALESCE(r.mapped_data->>'description', '') <> ''`, cids)
	if err != nil {
		log.Printf("processed description backfill: %v", err)
		return
	}
	n := int(ct.RowsAffected())
	if n > 0 {
		report["processed_descriptions_backfilled"] = n
		log.Printf("backfilled %d processed_products.description from mapped_data", n)
	}
}

// backfillProcessedNamesFromMapped fills empty processed_products.name from mapped_data name/title.
func backfillProcessedNamesFromMapped(
	ctx context.Context,
	pg *pgxpool.Pool,
	companyMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if dryRun || pg == nil {
		return
	}
	cids := make([]uuid.UUID, 0, len(companyMap))
	for legacy, cid := range companyMap {
		if len(allow) > 0 && !allow[legacy] {
			continue
		}
		u, err := uuid.Parse(cid)
		if err != nil {
			continue
		}
		cids = append(cids, u)
	}
	if len(cids) == 0 {
		return
	}
	ct, err := pg.Exec(ctx, `
		UPDATE processed_products p
		SET name = COALESCE(NULLIF(r.mapped_data->>'name', ''), NULLIF(r.mapped_data->>'title', '')),
			updated_at = now()
		FROM raw_products r
		WHERE p.raw_product_id = r.id
		  AND p.company_id = ANY($1::uuid[])
		  AND COALESCE(p.name, '') = ''
		  AND COALESCE(NULLIF(r.mapped_data->>'name', ''), NULLIF(r.mapped_data->>'title', ''), '') <> ''`, cids)
	if err != nil {
		log.Printf("processed name backfill: %v", err)
		return
	}
	n := int(ct.RowsAffected())
	if n > 0 {
		report["processed_names_backfilled"] = n
		log.Printf("backfilled %d processed_products.name from mapped_data", n)
	}
}

func migrateExportFeeds(
	ctx context.Context,
	mysqlDB *sql.DB,
	pg *pgxpool.Pool,
	companyMap, feedMap map[string]string,
	allow map[string]bool,
	report map[string]int,
	dryRun bool,
) {
	if !mysqlTableExists(ctx, mysqlDB, "export_feeds") {
		log.Printf("export_feeds skipped: table missing")
		return
	}
	q := mysqlSelectList(
		"id",
		"company_id",
		"name",
		mysqlCoalesce(ctx, mysqlDB, "export_feeds", "format", "'xml'"),
		mysqlCol(ctx, mysqlDB, "export_feeds", "source_feed_id", "NULL"),
		mysqlCol(ctx, mysqlDB, "export_feeds", "mappings", "NULL"),
		mysqlCol(ctx, mysqlDB, "export_feeds", "structure", "NULL"),
		mysqlCol(ctx, mysqlDB, "export_feeds", "last_generated_at", "NULL"),
	) + " FROM export_feeds WHERE 1=1"
	clause, cargs := mysqlCompanyFilter("company_id", allow)
	q += clause
	rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
	if err != nil {
		rows, err = mysqlDB.QueryContext(ctx, `
			SELECT id, company_id, name, format, source_feed_id, mappings, NULL, NULL
			FROM export_feeds WHERE 1=1`+clause, cargs...)
	}
	if err != nil {
		log.Printf("export_feeds skipped: %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var legacyID, companyLegacy, name, format string
		var sourceFeed sql.NullInt64
		var mappings, structure []byte
		var lastGen sql.NullTime
		if err := rows.Scan(&legacyID, &companyLegacy, &name, &format, &sourceFeed, &mappings, &structure, &lastGen); err != nil {
			log.Printf("export_feed scan: %v", err)
			report["export_feeds_skipped"]++
			continue
		}
		cid, ok := companyMap[companyLegacy]
		if !ok {
			report["export_feeds_skipped"]++
			continue
		}
		var src *uuid.UUID
		if sourceFeed.Valid {
			if fid, ok := feedMap[strconv.FormatInt(sourceFeed.Int64, 10)]; ok {
				u, err := uuid.Parse(fid)
				if err == nil {
					src = &u
				}
			}
		}
		tpl := map[string]any{}
		if len(mappings) > 0 {
			var m any
			if json.Unmarshal(mappings, &m) == nil {
				tpl["mappings"] = m
			}
		}
		if len(structure) > 0 {
			var s any
			if json.Unmarshal(structure, &s) == nil {
				tpl["structure"] = s
			}
		}
		tplBytes, _ := json.Marshal(tpl)
		if format == "" {
			format = "xml"
		}
		if dryRun {
			report["export_feeds"]++
			continue
		}
		newID := uuid.New()
		_, err = pg.Exec(ctx, `
			INSERT INTO export_feeds (id, company_id, name, source_feed_id, format, template, last_generated_at)
			VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7)`,
			newID, cid, name, src, format, string(tplBytes), nullTime(lastGen))
		if err != nil {
			log.Printf("export_feed insert %s: %v", legacyID, err)
			report["export_feeds_skipped"]++
			continue
		}
		report["export_feeds"]++
	}
}

func insertFeedMappings(
	ctx context.Context,
	pg *pgxpool.Pool,
	feedID uuid.UUID,
	companyID string,
	fieldMappings []byte,
	version int,
	report map[string]int,
) error {
	if len(fieldMappings) == 0 {
		return nil
	}
	payload := ensureJSON(fieldMappings)
	// Normalize object maps → JSON array of entries for feed_mappings.mappings default shape.
	var asObj map[string]any
	if json.Unmarshal(payload, &asObj) == nil && len(asObj) > 0 {
		arr := make([]any, 0, len(asObj))
		for k, v := range asObj {
			entry := map[string]any{"key": k, "mapping": normalizeLegacyMappingValue(v)}
			arr = append(arr, entry)
		}
		if b, err := json.Marshal(arr); err == nil {
			payload = b
		}
	} else {
		var asArr []any
		if json.Unmarshal(payload, &asArr) == nil && len(asArr) > 0 {
			for i, item := range asArr {
				m, ok := item.(map[string]any)
				if !ok {
					continue
				}
				if nested, ok := m["mapping"]; ok {
					m["mapping"] = normalizeLegacyMappingValue(nested)
					asArr[i] = m
				}
			}
			if b, err := json.Marshal(asArr); err == nil {
				payload = b
			}
		}
	}
	if version <= 0 {
		version = 1
	}
	_, err := pg.Exec(ctx, `
		INSERT INTO feed_mappings (feed_id, company_id, version, mappings, is_active)
		VALUES ($1, $2, $3, $4::jsonb, true)`,
		feedID, companyID, version, string(payload))
	if err != nil {
		report["feed_mappings_skipped"]++
		return err
	}
	report["feed_mappings"]++
	return nil
}

// normalizeLegacyMappingValue rewrites nested mapping objects so fieldName uses
// snake_case aliases the v2 mapping UI / ecommerce catalog understand.
func normalizeLegacyMappingValue(v any) any {
	m, ok := v.(map[string]any)
	if !ok {
		return v
	}
	out := make(map[string]any, len(m))
	for k, val := range m {
		out[k] = val
	}
	for _, key := range []string{"fieldName", "field", "target"} {
		raw, _ := out[key].(string)
		if strings.TrimSpace(raw) == "" {
			continue
		}
		out[key] = canonicalizeLegacyFieldKey(raw)
		break
	}
	return out
}

func canonicalizeLegacyFieldKey(raw string) string {
	compact := strings.ToLower(strings.TrimSpace(raw))
	compact = strings.ReplaceAll(compact, "_", "")
	compact = strings.ReplaceAll(compact, "-", "")
	compact = strings.ReplaceAll(compact, " ", "")
	aliases := map[string]string{
		"name": "title", "productname": "title", "title": "title",
		"purchaseprice": "purchase_price", "productmodel": "product_model",
		"moreimages": "additional_image_urls", "imageurl": "image_url",
		"producturl": "product_url", "officiallink": "official_link",
		"mainimage": "main_image", "eprelid": "eprel_id",
		"stockstatus": "availability", "videourl": "video_url",
		"netdepth": "net_depth", "netheight": "net_height",
		"netwidth": "net_width", "netmass": "net_mass",
	}
	if v, ok := aliases[compact]; ok {
		return v
	}
	return strings.TrimSpace(raw)
}

func mysqlTableExists(ctx context.Context, db *sql.DB, name string) bool {
	var n int
	err := db.QueryRowContext(ctx, `
		SELECT COUNT(*) FROM information_schema.tables
		WHERE table_schema = DATABASE() AND table_name = ?`, name).Scan(&n)
	return err == nil && n > 0
}

func mysqlColumnExists(ctx context.Context, db *sql.DB, table, column string) bool {
	var n int
	err := db.QueryRowContext(ctx, `
		SELECT COUNT(*) FROM information_schema.columns
		WHERE table_schema = DATABASE() AND table_name = ? AND column_name = ?`, table, column).Scan(&n)
	return err == nil && n > 0
}

func nullString(ns sql.NullString) *string {
	if !ns.Valid {
		return nil
	}
	s := ns.String
	return &s
}

func nullTime(nt sql.NullTime) any {
	if !nt.Valid {
		return nil
	}
	return nt.Time
}

func jsonOrNull(b []byte) *string {
	if len(b) == 0 || string(b) == "null" {
		return nil
	}
	s := string(b)
	if !json.Valid(b) {
		enc, err := json.Marshal(s)
		if err != nil {
			return nil
		}
		s = string(enc)
	}
	return &s
}

func ensureJSON(b []byte) []byte {
	if len(b) == 0 || !json.Valid(b) {
		return []byte("{}")
	}
	return b
}