package controlplane

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

	bootstate "github.com/nexuscrm/db/pkg/bootstate"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

const (
	migrationIdMultiPicklistStorageBackfill = "20260618_multipicklist_storage_backfill"
	multiPicklistBackfillPageSize           = 500
	multiPicklistFastBackfillPageSize       = 5000
)

// safeSQLIdentifierRE matches identifiers that are safe to interpolate into
// backtick-delimited SQL. The connector push backfill uses the same pattern for
// its resolved table/column identifiers.
var safeSQLIdentifierRE = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)

// MigrateMultiPicklistStorageBackfill normalizes legacy semicolon-delimited
// multi-select picklist values (stored as JSON STRING type, e.g. "A;B") to
// the canonical JSON array shape (["A","B"]).
//
// Scope: this migration handles the common legacy storage shape — JSON STRING
// rows where the string value uses semicolons as the picklist-value delimiter.
// Runtime membership predicates intentionally dual-read this legacy shape, so
// this migration is physical cleanup and should not be required for deploy
// readiness.
// It does NOT attempt to repair rare historical double-wrapper arrays (e.g.
// ["[\"A\",\"B\"]"]) which require field-metadata (Options) to disambiguate.
// Those are a pre-existing data-quality issue that a follow-up field-aware
// migration will address with a separate migration ID.
//
// Physical table and column names are resolved through the platform's
// canonical read path: ResolveReadSourceTable for objects, and the same
// backtick-aware PhysicalName resolution used by runtime query code for
// columns.
//
// Idempotent: stamp-guarded by _System_Migration. The normalizer itself is
// also idempotent — already-canonical JSON arrays round-trip unchanged.
func MigrateMultiPicklistStorageBackfill(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return fmt.Errorf("MigrateMultiPicklistStorageBackfill requires a database")
	}
	applied, err := bootstate.MigrationApplied(ctx, db, migrationIdMultiPicklistStorageBackfill)
	if err != nil {
		return err
	}
	if applied {
		log.Println("   ⏭  MigrateMultiPicklistStorageBackfill: already applied; skipping")
		return nil
	}

	totalUpdated, err := backfillMultiPicklistStorage(ctx, db)
	if err != nil {
		return fmt.Errorf("MigrateMultiPicklistStorageBackfill: %w", err)
	}

	if err := bootstate.RecordMigrationApplied(
		ctx,
		db,
		migrationIdMultiPicklistStorageBackfill,
		1,
		"Normalize legacy semicolon-delimited multi-select picklist values to canonical JSON array storage.",
	); err != nil {
		return err
	}

	if totalUpdated > 0 {
		log.Printf("   🧼 MigrateMultiPicklistStorageBackfill: normalized %d row(s)", totalUpdated)
	} else {
		log.Println("   ✅ MigrateMultiPicklistStorageBackfill: no rows needed normalization")
	}
	return nil
}

// backfillMultiPicklistStorage is the idempotent inner workhorse (not
// stamp-guarded). Exported for direct use in tests.
func backfillMultiPicklistStorage(ctx context.Context, db *sql.DB) (int, error) {
	// Step 1: build a per-object ObjectMetadata map so we can call the
	// canonical ResolveReadSourceTable() instead of raw JSON extraction.
	objMetaByID := make(map[string]*models.ObjectMetadata)
	objRows, err := db.QueryContext(ctx,
		fmt.Sprintf(
			"SELECT `%s`, `%s`, `%s` FROM `%s` WHERE COALESCE(`%s`,0) = 0",
			constants.ColSysObject_ID,
			constants.ColSysObject_APIName,
			constants.FieldSysObject_ReadSource,
			constants.TableObject,
			constants.FieldSysObject_IsDeleted,
		),
	)
	if err != nil {
		return 0, fmt.Errorf("query _System_Object for read source resolution: %w", err)
	}
	for objRows.Next() {
		var id, apiName string
		var readSourceRaw sql.NullString
		if err := objRows.Scan(&id, &apiName, &readSourceRaw); err != nil {
			objRows.Close()
			return 0, fmt.Errorf("scan _System_Object: %w", err)
		}
		var readSource *models.ObjectReadSourceConfig
		if readSourceRaw.Valid && strings.TrimSpace(readSourceRaw.String) != "" {
			// Decode — NormalizeObjectReadSourceConfig inside ResolveReadSourceTable
			// will validate; here we just parse the JSON.
			var cfg models.ObjectReadSourceConfig
			if err := json.Unmarshal([]byte(readSourceRaw.String), &cfg); err != nil {
				objRows.Close()
				return 0, fmt.Errorf("decode _System_Object ReadSource JSON for %s: %w", apiName, err)
			}
			readSource = &cfg
		}
		objMetaByID[id] = &models.ObjectMetadata{
			APIName:    strings.TrimSpace(apiName),
			ReadSource: readSource,
		}
	}
	objRows.Close()
	if err := objRows.Err(); err != nil {
		return 0, fmt.Errorf("iterate _System_Object: %w", err)
	}

	// Step 2: find all multipicklist fields that have a physical column
	// (FieldRequiresPhysicalColumn semantics from shared/pkg/formulastorage).
	// Two categories:
	//   (a) Stored (non-formula) multipicklist fields: Type IN ('multipicklist',
	//       'multiselectpicklist') + not an active view-only formula field.
	//   (b) Formula-backed multipicklist fields: ReturnType IN ('multipicklist',
	//       'multiselectpicklist') + FormulaStorageMode='write_materialized'.
	// View-only formula fields have no physical column and are excluded.
	// For each field we capture the raw PhysicalName and ApiName so we can apply
	// the canonical backtick-aware column resolution.
	type fieldCandidate struct {
		objID        string
		apiName      string
		physicalName string // may be empty; may be backtick-quoted
	}
	var candidates []fieldCandidate
	fieldRows, err := db.QueryContext(ctx,
		fmt.Sprintf(
			"SELECT f.`%s`, f.`%s`, f.`%s` FROM `%s` f "+
				"WHERE COALESCE(f.`%s`,0) = 0 "+
				"AND ( "+
				"  (LOWER(TRIM(f.`%s`)) IN ('multipicklist','multiselectpicklist') "+
				"   AND (f.`%s` IS NULL OR LOWER(TRIM(f.`%s`)) = '' "+
				"     OR (LOWER(TRIM(f.`%s`)) = 'active' AND LOWER(TRIM(COALESCE(f.`%s`,''))) = 'write_materialized')))"+
				"  OR "+
				"  (LOWER(TRIM(COALESCE(f.`%s`,''))) IN ('multipicklist','multiselectpicklist') "+
				"   AND LOWER(TRIM(COALESCE(f.`%s`,''))) = 'write_materialized')"+
				")",
			constants.ColSysField_ObjectID,
			constants.ColSysField_APIName,
			constants.ColSysField_PhysicalName,
			constants.TableField,
			constants.FieldSysField_IsDeleted,
			constants.FieldSysField_Type,
			constants.FieldSysField_FormulaMode,
			constants.FieldSysField_FormulaMode,
			constants.FieldSysField_FormulaMode,
			constants.FieldSysField_FormulaStorageMode,
			constants.FieldSysField_ReturnType,
			constants.FieldSysField_FormulaStorageMode,
		),
	)
	if err != nil {
		return 0, fmt.Errorf("query multipicklist fields: %w", err)
	}
	for fieldRows.Next() {
		var objID, apiName string
		var physicalName sql.NullString
		if err := fieldRows.Scan(&objID, &apiName, &physicalName); err != nil {
			fieldRows.Close()
			return 0, fmt.Errorf("scan multipicklist field: %w", err)
		}
		candidates = append(candidates, fieldCandidate{
			objID:        objID,
			apiName:      strings.TrimSpace(apiName),
			physicalName: strings.TrimSpace(physicalName.String),
		})
	}
	fieldRows.Close()
	if err := fieldRows.Err(); err != nil {
		return 0, fmt.Errorf("iterate multipicklist fields: %w", err)
	}
	log.Printf("   MigrateMultiPicklistStorageBackfill: found %d multipicklist field candidate(s)", len(candidates))

	// Step 3: resolve table and column names through canonical helpers,
	// validate, and backfill.
	totalUpdated := 0
	tableColumns := make(map[string][]string)
	tableColumnSeen := make(map[string]map[string]struct{})
	var tableOrder []string
	for _, c := range candidates {
		obj, ok := objMetaByID[c.objID]
		if !ok {
			continue
		}
		tableName, err := obj.ResolveReadSourceTable()
		if err != nil {
			return totalUpdated, fmt.Errorf("resolve read source table for %s: %w", c.apiName, err)
		}
		if !safeSQLIdentifierRE.MatchString(tableName) {
			return totalUpdated, fmt.Errorf("unsafe physical table name %q", tableName)
		}
		colName := resolvePhysicalColumnName(c.apiName, c.physicalName)
		if !safeSQLIdentifierRE.MatchString(colName) {
			return totalUpdated, fmt.Errorf("unsafe physical column name %q for field %s.%s", colName, c.apiName, c.physicalName)
		}
		if _, ok := tableColumnSeen[tableName]; !ok {
			tableColumnSeen[tableName] = make(map[string]struct{})
			tableOrder = append(tableOrder, tableName)
		}
		if _, ok := tableColumnSeen[tableName][colName]; !ok {
			tableColumnSeen[tableName][colName] = struct{}{}
			tableColumns[tableName] = append(tableColumns[tableName], colName)
		}
	}
	for _, tableName := range tableOrder {
		cols := tableColumns[tableName]
		log.Printf("   MigrateMultiPicklistStorageBackfill: fast-path scanning %s simple JSON string candidates across %d column(s)", tableName, len(cols))
		n, err := fastBackfillSimpleMultiPicklistTable(ctx, db, tableName, cols)
		if err != nil {
			return totalUpdated, err
		}
		totalUpdated += n
		if n > 0 {
			log.Printf("   MigrateMultiPicklistStorageBackfill: %s fast-path updated simple JSON string rows=%d", tableName, n)
		}
	}
	for _, tableName := range tableOrder {
		cols := tableColumns[tableName]
		log.Printf("   MigrateMultiPicklistStorageBackfill: scanning %s remaining legacy JSON string candidates across %d column(s)", tableName, len(cols))
		n, err := backfillMultiPicklistTablePagedFallback(ctx, db, tableName, cols)
		if err != nil {
			return totalUpdated, fmt.Errorf("backfill %s: %w", tableName, err)
		}
		totalUpdated += n
	}
	return totalUpdated, nil
}

// resolvePhysicalColumnName applies the same PhysicalName resolution used by
// runtime query code (connector_push_legacy_mirror_backfill.go:483-492):
//   - empty PhysicalName → API name
//   - backtick-wrapped PhysicalName (`name`) → API name (alias; real column is API name)
//   - otherwise → PhysicalName
func resolvePhysicalColumnName(apiName, physicalName string) string {
	apiName = strings.TrimSpace(apiName)
	physicalName = strings.TrimSpace(physicalName)
	if physicalName == "" {
		return apiName
	}
	if strings.HasPrefix(physicalName, "`") && strings.HasSuffix(physicalName, "`") {
		return apiName
	}
	return physicalName
}

// backfillMultiPicklistColumn normalizes a single multipicklist column in a
// single SObject table. It selects rows whose JSON type is STRING (legacy
// semicolon-delimited values), normalizes them to canonical JSON arrays, and
// writes them back with a compare-and-swap guard.
//
// Non-STRING rows (JSON arrays, NULL) are not touched. Rare double-wrapper
// arrays are handled in a follow-up field-aware migration.
func backfillMultiPicklistColumn(ctx context.Context, db *sql.DB, tableName, columnName string) (int, error) {
	return backfillMultiPicklistColumnWithPageSize(ctx, db, tableName, columnName, multiPicklistBackfillPageSize)
}

func backfillMultiPicklistColumnWithPageSize(ctx context.Context, db *sql.DB, tableName, columnName string, pageSize int) (int, error) {
	if pageSize <= 0 {
		return 0, fmt.Errorf("invalid multipicklist backfill page size %d", pageSize)
	}
	if err := ctx.Err(); err != nil {
		return 0, fmt.Errorf("scan %s.%s canceled before fast-path backfill: %w", tableName, columnName, err)
	}
	selectQuery := fmt.Sprintf(
		"SELECT `Id`, `%s` FROM `%s` WHERE `%s` IS NOT NULL AND JSON_TYPE(`%s`) = 'STRING' AND `Id` > ? ORDER BY `Id` LIMIT ?",
		columnName, tableName, columnName, columnName,
	)
	updated := 0
	log.Printf("   MigrateMultiPicklistStorageBackfill: %s.%s paging legacy JSON string candidates with page_size=%d", tableName, columnName, pageSize)

	fastUpdated, err := fastBackfillSimpleMultiPicklistStrings(ctx, db, tableName, columnName)
	if err != nil {
		return updated, err
	}
	updated += fastUpdated
	if fastUpdated > 0 {
		log.Printf("   MigrateMultiPicklistStorageBackfill: %s.%s fast-path updated simple JSON string rows=%d", tableName, columnName, fastUpdated)
	}

	return backfillMultiPicklistColumnPagedFallbackWithQuery(ctx, db, tableName, columnName, pageSize, selectQuery, updated)
}

func backfillMultiPicklistTablePagedFallback(ctx context.Context, db *sql.DB, tableName string, columnNames []string) (int, error) {
	return backfillMultiPicklistTablePagedFallbackWithPageSize(ctx, db, tableName, columnNames, multiPicklistBackfillPageSize)
}

func backfillMultiPicklistTablePagedFallbackWithPageSize(ctx context.Context, db *sql.DB, tableName string, columnNames []string, pageSize int) (int, error) {
	if pageSize <= 0 {
		return 0, fmt.Errorf("invalid multipicklist backfill page size %d", pageSize)
	}
	if len(columnNames) == 0 {
		return 0, nil
	}

	var selectCols strings.Builder
	for _, columnName := range columnNames {
		fmt.Fprintf(
			&selectCols,
			", CASE WHEN %s THEN CAST(`%s` AS CHAR) ELSE NULL END",
			remainingMultiPicklistStringPredicate(columnName),
			columnName,
		)
	}
	selectQuery := fmt.Sprintf(
		"SELECT `Id`%s FROM `%s` WHERE (%s) AND `Id` > ? ORDER BY `Id` LIMIT ?",
		selectCols.String(),
		tableName,
		remainingMultiPicklistAnyStringPredicate(columnNames),
	)

	updated := 0
	scanned := 0
	lastID := ""
	log.Printf("   MigrateMultiPicklistStorageBackfill: %s paging remaining legacy JSON string candidates with page_size=%d", tableName, pageSize)
	for {
		if err := ctx.Err(); err != nil {
			return updated, fmt.Errorf("scan %s canceled after %d candidate row(s), %d update(s): %w", tableName, scanned, updated, err)
		}
		rows, err := db.QueryContext(ctx, selectQuery, lastID, pageSize)
		if err != nil {
			return updated, fmt.Errorf("query legacy values in %s after Id=%q: %w", tableName, lastID, err)
		}

		batch := make([]multiPicklistTableBackfillRow, 0, pageSize)
		for rows.Next() {
			rv := multiPicklistTableBackfillRow{
				values: make([]sql.NullString, len(columnNames)),
			}
			dest := make([]interface{}, 0, len(columnNames)+1)
			dest = append(dest, &rv.id)
			for i := range rv.values {
				dest = append(dest, &rv.values[i])
			}
			if err := rows.Scan(dest...); err != nil {
				rows.Close()
				return updated, fmt.Errorf("scan %s after Id=%q: %w", tableName, lastID, err)
			}
			batch = append(batch, rv)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return updated, fmt.Errorf("iterate %s after Id=%q: %w", tableName, lastID, err)
		}

		if len(batch) == 0 {
			log.Printf("   MigrateMultiPicklistStorageBackfill: %s completed candidate_rows=%d updated=%d", tableName, scanned, updated)
			return updated, nil
		}
		previousLastID := lastID
		lastID = batch[len(batch)-1].id
		if lastID == previousLastID {
			return updated, fmt.Errorf("query legacy values in %s did not advance past Id=%q", tableName, previousLastID)
		}

		updatesByColumn := make(map[string][]multiPicklistBackfillUpdate, len(columnNames))
		for _, rv := range batch {
			for i, value := range rv.values {
				if !value.Valid {
					continue
				}
				if err := ctx.Err(); err != nil {
					return updated, fmt.Errorf("update %s canceled after %d candidate row(s), %d update(s): %w", tableName, scanned, updated, err)
				}
				normalized := schemautil.NormalizeMultiPicklistStorageBackfill(value.String)
				normStr, ok := normalized.(string)
				columnName := columnNames[i]
				if ok && normStr == value.String {
					continue
				}
				if ok {
					updatesByColumn[columnName] = append(updatesByColumn[columnName], multiPicklistBackfillUpdate{
						id:         rv.id,
						previous:   value.String,
						normalized: &normStr,
					})
				} else {
					updatesByColumn[columnName] = append(updatesByColumn[columnName], multiPicklistBackfillUpdate{
						id:       rv.id,
						previous: value.String,
					})
				}
			}
		}

		batchUpdated := 0
		for _, columnName := range columnNames {
			columnUpdated, err := bulkUpdateMultiPicklistBackfillBatch(ctx, db, tableName, columnName, updatesByColumn[columnName])
			if err != nil {
				return updated, err
			}
			batchUpdated += columnUpdated
		}
		updated += batchUpdated

		scanned += len(batch)
		log.Printf("   MigrateMultiPicklistStorageBackfill: %s processed candidate_row_batch=%d updated_batch=%d candidate_rows_total=%d updated_total=%d",
			tableName, len(batch), batchUpdated, scanned, updated)
		if len(batch) < pageSize {
			log.Printf("   MigrateMultiPicklistStorageBackfill: %s completed candidate_rows=%d updated=%d", tableName, scanned, updated)
			return updated, nil
		}
	}
}

func backfillMultiPicklistColumnPagedFallbackWithQuery(ctx context.Context, db *sql.DB, tableName, columnName string, pageSize int, selectQuery string, updated int) (int, error) {
	scanned := 0
	lastID := ""
	for {
		if err := ctx.Err(); err != nil {
			return updated, fmt.Errorf("scan %s.%s canceled after %d candidate row(s), %d update(s): %w", tableName, columnName, scanned, updated, err)
		}
		rows, err := db.QueryContext(ctx, selectQuery, lastID, pageSize)
		if err != nil {
			return updated, fmt.Errorf("query legacy values in %s.%s after Id=%q: %w", tableName, columnName, lastID, err)
		}

		batch := make([]multiPicklistBackfillRow, 0, pageSize)
		for rows.Next() {
			var rv multiPicklistBackfillRow
			if err := rows.Scan(&rv.id, &rv.value); err != nil {
				rows.Close()
				return updated, fmt.Errorf("scan %s.%s after Id=%q: %w", tableName, columnName, lastID, err)
			}
			batch = append(batch, rv)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return updated, fmt.Errorf("iterate %s.%s after Id=%q: %w", tableName, columnName, lastID, err)
		}

		if len(batch) == 0 {
			log.Printf("   MigrateMultiPicklistStorageBackfill: %s.%s completed candidates=%d updated=%d", tableName, columnName, scanned, updated)
			return updated, nil
		}

		previousLastID := lastID
		lastID = batch[len(batch)-1].id
		if lastID == previousLastID {
			return updated, fmt.Errorf("query legacy values in %s.%s did not advance past Id=%q", tableName, columnName, previousLastID)
		}

		updates := make([]multiPicklistBackfillUpdate, 0, len(batch))
		for _, rv := range batch {
			if err := ctx.Err(); err != nil {
				return updated, fmt.Errorf("update %s.%s canceled after %d candidate row(s), %d update(s): %w", tableName, columnName, scanned, updated, err)
			}
			normalized := schemautil.NormalizeMultiPicklistStorageBackfill(rv.value)
			normStr, ok := normalized.(string)
			if ok && normStr == rv.value {
				continue // already canonical
			}
			if ok {
				updates = append(updates, multiPicklistBackfillUpdate{
					id:         rv.id,
					previous:   rv.value,
					normalized: &normStr,
				})
			} else {
				updates = append(updates, multiPicklistBackfillUpdate{
					id:       rv.id,
					previous: rv.value,
				})
			}
		}
		batchUpdated, err := bulkUpdateMultiPicklistBackfillBatch(ctx, db, tableName, columnName, updates)
		if err != nil {
			return updated, err
		}
		updated += batchUpdated

		scanned += len(batch)
		log.Printf("   MigrateMultiPicklistStorageBackfill: %s.%s processed candidate_batch=%d updated_batch=%d candidates_total=%d updated_total=%d",
			tableName, columnName, len(batch), batchUpdated, scanned, updated)
		if len(batch) < pageSize {
			log.Printf("   MigrateMultiPicklistStorageBackfill: %s.%s completed candidates=%d updated=%d", tableName, columnName, scanned, updated)
			return updated, nil
		}
	}
}

func fastBackfillSimpleMultiPicklistStrings(ctx context.Context, db *sql.DB, tableName, columnName string) (int, error) {
	selectQuery := fmt.Sprintf(
		"SELECT `Id` FROM `%s` WHERE %s AND `Id` > ? ORDER BY `Id` LIMIT ?",
		tableName,
		simpleMultiPicklistFastPathPredicate(columnName),
	)
	updated := 0
	scanned := 0
	lastID := ""
	for {
		if err := ctx.Err(); err != nil {
			return updated, fmt.Errorf("fast-path scan %s.%s canceled after %d candidate id(s), %d update(s): %w", tableName, columnName, scanned, updated, err)
		}
		rows, err := db.QueryContext(ctx, selectQuery, lastID, multiPicklistFastBackfillPageSize)
		if err != nil {
			return updated, fmt.Errorf("fast-path query simple legacy values in %s.%s after Id=%q: %w", tableName, columnName, lastID, err)
		}
		ids := make([]string, 0, multiPicklistFastBackfillPageSize)
		for rows.Next() {
			var id string
			if err := rows.Scan(&id); err != nil {
				rows.Close()
				return updated, fmt.Errorf("fast-path scan %s.%s after Id=%q: %w", tableName, columnName, lastID, err)
			}
			ids = append(ids, id)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return updated, fmt.Errorf("fast-path iterate %s.%s after Id=%q: %w", tableName, columnName, lastID, err)
		}

		if len(ids) == 0 {
			return updated, nil
		}
		previousLastID := lastID
		lastID = ids[len(ids)-1]
		if lastID == previousLastID {
			return updated, fmt.Errorf("fast-path query simple legacy values in %s.%s did not advance past Id=%q", tableName, columnName, previousLastID)
		}

		batchUpdated, err := updateSimpleMultiPicklistFastPathIDs(ctx, db, tableName, columnName, ids)
		if err != nil {
			return updated, err
		}
		scanned += len(ids)
		updated += batchUpdated
		log.Printf("   MigrateMultiPicklistStorageBackfill: %s.%s fast-path processed id_batch=%d updated_batch=%d ids_total=%d fast_updated_total=%d",
			tableName, columnName, len(ids), batchUpdated, scanned, updated)
		if len(ids) < multiPicklistFastBackfillPageSize {
			return updated, nil
		}
	}
}

func fastBackfillSimpleMultiPicklistTable(ctx context.Context, db *sql.DB, tableName string, columnNames []string) (int, error) {
	if len(columnNames) == 0 {
		return 0, nil
	}
	anyPredicate := simpleMultiPicklistAnyFastPathPredicate(columnNames)
	selectQuery := fmt.Sprintf(
		"SELECT `Id` FROM `%s` WHERE (%s) AND `Id` > ? ORDER BY `Id` LIMIT ?",
		tableName,
		anyPredicate,
	)
	updated := 0
	scanned := 0
	lastID := ""
	for {
		if err := ctx.Err(); err != nil {
			return updated, fmt.Errorf("fast-path scan %s canceled after %d candidate id(s), %d update(s): %w", tableName, scanned, updated, err)
		}
		rows, err := db.QueryContext(ctx, selectQuery, lastID, multiPicklistFastBackfillPageSize)
		if err != nil {
			return updated, fmt.Errorf("fast-path query simple legacy values in %s after Id=%q: %w", tableName, lastID, err)
		}
		ids := make([]string, 0, multiPicklistFastBackfillPageSize)
		for rows.Next() {
			var id string
			if err := rows.Scan(&id); err != nil {
				rows.Close()
				return updated, fmt.Errorf("fast-path scan %s after Id=%q: %w", tableName, lastID, err)
			}
			ids = append(ids, id)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return updated, fmt.Errorf("fast-path iterate %s after Id=%q: %w", tableName, lastID, err)
		}

		if len(ids) == 0 {
			return updated, nil
		}
		previousLastID := lastID
		lastID = ids[len(ids)-1]
		if lastID == previousLastID {
			return updated, fmt.Errorf("fast-path query simple legacy values in %s did not advance past Id=%q", tableName, previousLastID)
		}

		batchUpdated, err := updateSimpleMultiPicklistFastPathTableIDs(ctx, db, tableName, columnNames, ids)
		if err != nil {
			return updated, err
		}
		scanned += len(ids)
		updated += batchUpdated
		log.Printf("   MigrateMultiPicklistStorageBackfill: %s fast-path processed id_batch=%d updated_batch=%d ids_total=%d fast_updated_total=%d",
			tableName, len(ids), batchUpdated, scanned, updated)
		if len(ids) < multiPicklistFastBackfillPageSize {
			return updated, nil
		}
	}
}

func updateSimpleMultiPicklistFastPathTableIDs(ctx context.Context, db *sql.DB, tableName string, columnNames []string, ids []string) (int, error) {
	if len(ids) == 0 || len(columnNames) == 0 {
		return 0, nil
	}
	placeholders := strings.TrimRight(strings.Repeat("?,", len(ids)), ",")
	var b strings.Builder
	fmt.Fprintf(&b, "UPDATE `%s` SET ", tableName)
	for i, columnName := range columnNames {
		if i > 0 {
			b.WriteString(", ")
		}
		fmt.Fprintf(&b, "`%s` = CASE WHEN %s THEN %s ELSE `%s` END",
			columnName,
			simpleMultiPicklistFastPathPredicate(columnName),
			simpleMultiPicklistFastPathValueExpression(columnName),
			columnName,
		)
	}
	fmt.Fprintf(&b, " WHERE `Id` IN (%s) AND (%s)", placeholders, simpleMultiPicklistAnyFastPathPredicate(columnNames))
	args := make([]interface{}, 0, len(ids))
	for _, id := range ids {
		args = append(args, id)
	}
	result, err := db.ExecContext(ctx, b.String(), args...)
	if err != nil {
		return 0, fmt.Errorf("fast-path update simple legacy values in %s: %w", tableName, err)
	}
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("read fast-path update count for %s: %w", tableName, err)
	}
	return int(rowsAffected), nil
}

func updateSimpleMultiPicklistFastPathIDs(ctx context.Context, db *sql.DB, tableName, columnName string, ids []string) (int, error) {
	if len(ids) == 0 {
		return 0, nil
	}
	placeholders := strings.TrimRight(strings.Repeat("?,", len(ids)), ",")
	updateQuery := fmt.Sprintf(
		"UPDATE `%s` SET `%s` = CASE WHEN TRIM(JSON_UNQUOTE(`%s`)) = '' THEN NULL ELSE JSON_ARRAY(JSON_UNQUOTE(`%s`)) END "+
			"WHERE `Id` IN (%s) AND %s",
		tableName,
		columnName,
		columnName,
		columnName,
		placeholders,
		simpleMultiPicklistFastPathPredicate(columnName),
	)
	args := make([]interface{}, 0, len(ids))
	for _, id := range ids {
		args = append(args, id)
	}
	result, err := db.ExecContext(ctx, updateQuery, args...)
	if err != nil {
		return 0, fmt.Errorf("fast-path update simple legacy values in %s.%s: %w", tableName, columnName, err)
	}
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("read fast-path update count for %s.%s: %w", tableName, columnName, err)
	}
	return int(rowsAffected), nil
}

func simpleMultiPicklistFastPathPredicate(columnName string) string {
	return fmt.Sprintf(
		"`%s` IS NOT NULL AND JSON_TYPE(`%s`) = 'STRING' "+
			"AND (TRIM(JSON_UNQUOTE(`%s`)) = '' "+
			"OR (JSON_UNQUOTE(`%s`) = TRIM(JSON_UNQUOTE(`%s`)) AND JSON_UNQUOTE(`%s`) NOT LIKE '%%;%%'))",
		columnName,
		columnName,
		columnName,
		columnName,
		columnName,
		columnName,
	)
}

func simpleMultiPicklistFastPathValueExpression(columnName string) string {
	return fmt.Sprintf(
		"CASE WHEN TRIM(JSON_UNQUOTE(`%s`)) = '' THEN NULL ELSE JSON_ARRAY(JSON_UNQUOTE(`%s`)) END",
		columnName,
		columnName,
	)
}

func simpleMultiPicklistAnyFastPathPredicate(columnNames []string) string {
	var b strings.Builder
	for i, columnName := range columnNames {
		if i > 0 {
			b.WriteString(" OR ")
		}
		b.WriteString("(")
		b.WriteString(simpleMultiPicklistFastPathPredicate(columnName))
		b.WriteString(")")
	}
	return b.String()
}

func remainingMultiPicklistStringPredicate(columnName string) string {
	return fmt.Sprintf("`%s` IS NOT NULL AND JSON_TYPE(`%s`) = 'STRING'", columnName, columnName)
}

func remainingMultiPicklistAnyStringPredicate(columnNames []string) string {
	var b strings.Builder
	for i, columnName := range columnNames {
		if i > 0 {
			b.WriteString(" OR ")
		}
		b.WriteString("(")
		b.WriteString(remainingMultiPicklistStringPredicate(columnName))
		b.WriteString(")")
	}
	return b.String()
}

type multiPicklistBackfillRow struct {
	id    string
	value string
}

type multiPicklistTableBackfillRow struct {
	id     string
	values []sql.NullString
}

type multiPicklistBackfillUpdate struct {
	id         string
	previous   string
	normalized *string
}

func bulkUpdateMultiPicklistBackfillBatch(ctx context.Context, db *sql.DB, tableName, columnName string, updates []multiPicklistBackfillUpdate) (int, error) {
	if len(updates) == 0 {
		return 0, nil
	}

	var b strings.Builder
	args := make([]interface{}, 0, len(updates)*4)
	fmt.Fprintf(&b, "UPDATE `%s` SET `%s` = CASE `Id` ", tableName, columnName)
	for _, update := range updates {
		if update.normalized == nil {
			b.WriteString("WHEN ? THEN NULL ")
			args = append(args, update.id)
			continue
		}
		b.WriteString("WHEN ? THEN CAST(? AS JSON) ")
		args = append(args, update.id, *update.normalized)
	}
	fmt.Fprintf(&b, "ELSE `%s` END WHERE ", columnName)
	for i, update := range updates {
		if i > 0 {
			b.WriteString(" OR ")
		}
		fmt.Fprintf(&b, "(`Id` = ? AND `%s` = CAST(? AS JSON))", columnName)
		args = append(args, update.id, update.previous)
	}

	result, err := db.ExecContext(ctx, b.String(), args...)
	if err != nil {
		return 0, fmt.Errorf("bulk update %s.%s candidate batch: %w", tableName, columnName, err)
	}
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("read bulk update count for %s.%s candidate batch: %w", tableName, columnName, err)
	}
	return int(rowsAffected), nil
}