descrybe/apps/api/cmd/seed-a1/recover_jobs.go

package main

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"log"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/descrybe/descrybe-v2/apps/api/internal/billing"
	"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

// recoverJobsFromMySQLDump streams a mysqldump and upserts every A1
// processing_jobs (+ joinable processing_job_products) into live Postgres.
//
// Needed because live A1 often only retains a couple of recent jobs (retention
// deletes non-migrated terminal rows after 30 days; jobs domain may never have
// been imported). Export alone cannot invent history that is missing from PG.
//
// Job products resolve raw_product_id via dump GTIN → Postgres raw_products.
func recoverJobsFromMySQLDump(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID, dumpPath string) error {
	legacyCompany := billing.A1LegacyCompanyID
	var legacy string
	_ = pg.QueryRow(ctx, `SELECT COALESCE(legacy_company_id::text, '') FROM companies WHERE id = $1`, companyID).Scan(&legacy)
	if legacy != "" {
		legacyCompany = legacy
	}

	fi, err := os.Stat(dumpPath)
	if err != nil {
		return fmt.Errorf("stat mysql dump %q: %w", dumpPath, err)
	}
	log.Printf("recover-jobs: dump=%s size=%d legacy_company=%s", dumpPath, fi.Size(), legacyCompany)

	userByLegacy, err := loadUserLegacyMap(ctx, pg, companyID)
	if err != nil {
		return err
	}

	f, err := os.Open(dumpPath)
	if err != nil {
		return fmt.Errorf("open mysql dump: %w", err)
	}
	defer f.Close()

	jobs, err := scanA1ProcessingJobs(f, legacyCompany)
	if err != nil {
		return err
	}
	if len(jobs) == 0 {
		return fmt.Errorf("no processing_jobs for legacy company %s in dump", legacyCompany)
	}
	log.Printf("dump: %d A1 processing_jobs (incl. history)", len(jobs))

	if _, err := f.Seek(0, io.SeekStart); err != nil {
		return err
	}
	jobIDs := make(map[string]struct{}, len(jobs))
	for _, j := range jobs {
		jobIDs[j.id] = struct{}{}
	}
	pjpRows, rawLegacyIDs, err := scanA1JobProducts(f, jobIDs)
	if err != nil {
		return err
	}
	log.Printf("dump: %d A1 processing_job_products across %d raw ids", len(pjpRows), len(rawLegacyIDs))

	if _, err := f.Seek(0, io.SeekStart); err != nil {
		return err
	}
	gtinByLegacy, err := scanRawProductGTINs(f, rawLegacyIDs, legacyCompany)
	if err != nil {
		return err
	}
	log.Printf("dump: resolved %d/%d raw→gtin mappings", len(gtinByLegacy), len(rawLegacyIDs))

	rawByGTIN, err := loadRawByGTIN(ctx, pg, companyID, gtinByLegacy)
	if err != nil {
		return err
	}

	var jobsUpserted, jobsSkipped int
	for _, j := range jobs {
		jobUUID, err := parseDumpJobID(j.id)
		if err != nil {
			jobsSkipped++
			continue
		}
		var userID *uuid.UUID
		if uid, ok := userByLegacy[j.userLegacy]; ok {
			userID = &uid
		}
		status := normalizeProcessingJobStatus(j.status)
		ptype := strings.TrimSpace(j.processingType)
		if ptype == "" {
			ptype = "full"
		}
		var errPtr *string
		if j.errText != "" {
			v := j.errText
			errPtr = &v
		}
		_, err = pg.Exec(ctx, `
			INSERT INTO processing_jobs (
				id, company_id, user_id, status, total_products, processed_products,
				error, processing_type, priority, estimated_tokens,
				started_at, completed_at, created_at, updated_at,
				current_step, step_progress, ai_provider_mode
			) VALUES (
				$1, $2, $3, $4, $5, $6,
				$7, $8, $9, $10,
				$11, $12, $13, $14,
				'', '[]'::jsonb, 'migrated'
			)
			ON CONFLICT (id) DO UPDATE SET
				company_id = EXCLUDED.company_id,
				user_id = COALESCE(EXCLUDED.user_id, processing_jobs.user_id),
				status = EXCLUDED.status,
				total_products = EXCLUDED.total_products,
				processed_products = EXCLUDED.processed_products,
				error = EXCLUDED.error,
				processing_type = EXCLUDED.processing_type,
				priority = EXCLUDED.priority,
				estimated_tokens = EXCLUDED.estimated_tokens,
				started_at = EXCLUDED.started_at,
				completed_at = EXCLUDED.completed_at,
				created_at = EXCLUDED.created_at,
				updated_at = EXCLUDED.updated_at,
				ai_provider_mode = 'migrated'`,
			jobUUID, companyID, userID, status,
			j.totalProducts, j.processedProducts,
			errPtr, ptype, j.priority, j.estimatedTokens,
			j.startedAt, j.completedAt, j.createdAt, j.updatedAt,
		)
		if err != nil {
			log.Printf("job %s: %v", j.id, err)
			jobsSkipped++
			continue
		}
		jobsUpserted++
	}

	var pjpUpserted, pjpSkipped int
	for _, row := range pjpRows {
		jobUUID, err := parseDumpJobID(row.jobID)
		if err != nil {
			pjpSkipped++
			continue
		}
		gtin, ok := gtinByLegacy[row.rawLegacy]
		if !ok || gtin == "" {
			pjpSkipped++
			continue
		}
		rawUUID, ok := rawByGTIN[gtin]
		if !ok {
			pjpSkipped++
			continue
		}
		prodID := uuid.NewSHA1(uuid.NameSpaceOID, []byte("pjp:"+strconv.FormatInt(row.legacyID, 10)))
		var errPtr *string
		if row.errText != "" {
			v := row.errText
			errPtr = &v
		}
		_, err = pg.Exec(ctx, `
			INSERT INTO processing_job_products (
				id, job_id, raw_product_id, status, error, processed_product_id, created_at, updated_at
			) VALUES ($1, $2, $3, $4, $5, NULL, $6, $7)
			ON CONFLICT (id) DO UPDATE SET
				status = EXCLUDED.status,
				error = EXCLUDED.error,
				raw_product_id = EXCLUDED.raw_product_id,
				updated_at = EXCLUDED.updated_at`,
			prodID, jobUUID, rawUUID, normalizeJobProductStatus(row.status), errPtr,
			row.createdAt, row.updatedAt,
		)
		if err != nil {
			pjpSkipped++
			continue
		}
		pjpUpserted++
	}

	var liveJobs, livePJP int
	_ = pg.QueryRow(ctx, `SELECT count(*) FROM processing_jobs WHERE company_id=$1`, companyID).Scan(&liveJobs)
	_ = pg.QueryRow(ctx, `SELECT count(*) FROM processing_job_products WHERE job_id IN (SELECT id FROM processing_jobs WHERE company_id=$1)`, companyID).Scan(&livePJP)
	log.Printf("recover-jobs done: jobs_upserted=%d skipped=%d pjp_upserted=%d skipped=%d live_jobs=%d live_job_products=%d",
		jobsUpserted, jobsSkipped, pjpUpserted, pjpSkipped, liveJobs, livePJP)
	return nil
}

type dumpJob struct {
	id                               string
	userLegacy                       string
	status, processingType, errText  string
	totalProducts, processedProducts int
	priority, estimatedTokens        int
	startedAt, completedAt           *time.Time
	createdAt, updatedAt             time.Time
}

type dumpJobProduct struct {
	legacyID             int64
	jobID, rawLegacy     string
	status, errText      string
	createdAt, updatedAt time.Time
}

func loadUserLegacyMap(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID) (map[string]uuid.UUID, error) {
	rows, err := pg.Query(ctx, `
		SELECT u.id, COALESCE(u.legacy_user_id, '')
		FROM users u
		JOIN memberships m ON m.user_id = u.id
		WHERE m.company_id = $1 AND COALESCE(u.legacy_user_id, '') <> ''`, companyID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[string]uuid.UUID{}
	for rows.Next() {
		var id uuid.UUID
		var legacy string
		if err := rows.Scan(&id, &legacy); err != nil {
			return nil, err
		}
		out[legacy] = id
	}
	return out, rows.Err()
}

func loadRawByGTIN(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID, gtinByLegacy map[string]string) (map[string]uuid.UUID, error) {
	uniq := make([]string, 0, len(gtinByLegacy))
	seen := map[string]struct{}{}
	for _, g := range gtinByLegacy {
		g = strings.TrimSpace(g)
		if g == "" {
			continue
		}
		if _, ok := seen[g]; ok {
			continue
		}
		seen[g] = struct{}{}
		uniq = append(uniq, g)
	}
	out := map[string]uuid.UUID{}
	if len(uniq) == 0 {
		return out, nil
	}
	rows, err := pg.Query(ctx, `
		SELECT gtin, id FROM raw_products
		WHERE company_id = $1 AND gtin = ANY($2)`, companyID, uniq)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var gtin string
		var id uuid.UUID
		if err := rows.Scan(&gtin, &id); err != nil {
			return nil, err
		}
		out[gtin] = id
	}
	return out, rows.Err()
}

func scanA1ProcessingJobs(r io.Reader, legacyCompany string) ([]dumpJob, error) {
	br := bufio.NewReaderSize(r, 4<<20)
	mode := false
	var out []dumpJob
	for {
		line, err := br.ReadString('\n')
		if strings.HasPrefix(line, "INSERT INTO `processing_jobs`") {
			mode = true
		} else if mode && dumpSectionEnded(line, "processing_jobs") {
			mode = false
		}
		if mode && strings.Contains(line, "'"+legacyCompany+"'") {
			fields := parseMySQLTupleFields(line)
			// id, user_id, company_id, status, total, processed, error, started, completed, created, updated, type, priority, estimated
			if len(fields) >= 14 && fields[2] == legacyCompany {
				j := dumpJob{
					id:                fields[0],
					userLegacy:        nullish(fields[1]),
					status:            fields[3],
					totalProducts:     atoiDefault(fields[4], 0),
					processedProducts: atoiDefault(fields[5], 0),
					errText:           nullish(fields[6]),
					startedAt:         parseDumpTimePtr(fields[7]),
					completedAt:       parseDumpTimePtr(fields[8]),
					createdAt:         parseDumpTime(fields[9]),
					updatedAt:         parseDumpTime(fields[10]),
					processingType:    nullish(fields[11]),
					priority:          atoiDefault(fields[12], 0),
					estimatedTokens:   atoiDefault(fields[13], 0),
				}
				if j.createdAt.IsZero() {
					j.createdAt = time.Now().UTC()
				}
				if j.updatedAt.IsZero() {
					j.updatedAt = j.createdAt
				}
				out = append(out, j)
			}
		}
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, err
		}
	}
	return out, nil
}

func scanA1JobProducts(r io.Reader, jobIDs map[string]struct{}) ([]dumpJobProduct, map[string]struct{}, error) {
	br := bufio.NewReaderSize(r, 4<<20)
	mode := false
	var out []dumpJobProduct
	rawIDs := map[string]struct{}{}
	for {
		line, err := br.ReadString('\n')
		if strings.HasPrefix(line, "INSERT INTO `processing_job_products`") {
			mode = true
		} else if mode && dumpSectionEnded(line, "processing_job_products") {
			mode = false
		}
		if mode && looksLikeTupleLine(line) {
			fields := parseMySQLTupleFields(line)
			// id, job_id, raw_product_id, status, error, processed_product_id, created, updated
			if len(fields) >= 8 {
				if _, ok := jobIDs[fields[1]]; ok {
					legacyID, _ := strconv.ParseInt(fields[0], 10, 64)
					row := dumpJobProduct{
						legacyID:  legacyID,
						jobID:     fields[1],
						rawLegacy: fields[2],
						status:    fields[3],
						errText:   nullish(fields[4]),
						createdAt: parseDumpTime(fields[6]),
						updatedAt: parseDumpTime(fields[7]),
					}
					if row.createdAt.IsZero() {
						row.createdAt = time.Now().UTC()
					}
					if row.updatedAt.IsZero() {
						row.updatedAt = row.createdAt
					}
					out = append(out, row)
					rawIDs[row.rawLegacy] = struct{}{}
				}
			}
		}
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, nil, err
		}
	}
	return out, rawIDs, nil
}

func scanRawProductGTINs(r io.Reader, want map[string]struct{}, legacyCompany string) (map[string]string, error) {
	br := bufio.NewReaderSize(r, 4<<20)
	mode := false
	out := map[string]string{}
	remaining := len(want)
	for remaining > 0 {
		line, err := br.ReadString('\n')
		if strings.HasPrefix(line, "INSERT INTO `raw_products`") {
			mode = true
		} else if mode && dumpSectionEnded(line, "raw_products") {
			mode = false
		}
		if mode && looksLikeTupleLine(line) {
			// Need id + gtin; company is field index 4 in this dump schema:
			// id, gtin, feed_id, feed_ids, company_id, raw_data, ...
			fields := parseMySQLTupleFieldsN(line, 5)
			if len(fields) >= 5 {
				id := fields[0]
				if _, ok := want[id]; ok && fields[4] == legacyCompany {
					gtin := strings.TrimSpace(nullish(fields[1]))
					if gtin != "" {
						out[id] = gtin
						remaining--
					}
				}
			}
		}
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, err
		}
	}
	return out, nil
}

func dumpSectionEnded(line, table string) bool {
	if strings.HasPrefix(line, "CREATE TABLE") || strings.HasPrefix(line, "UNLOCK TABLES") || strings.HasPrefix(line, "LOCK TABLES") {
		return true
	}
	if strings.HasPrefix(line, "INSERT INTO `") && !strings.HasPrefix(line, "INSERT INTO `"+table+"`") {
		return true
	}
	if strings.HasPrefix(line, "DROP TABLE") {
		return true
	}
	return false
}

func looksLikeTupleLine(line string) bool {
	return processing.LooksLikeMySQLTupleLine(line)
}

func parseMySQLTupleFields(line string) []string {
	return processing.ParseMySQLTupleFields(line)
}

func parseMySQLTupleFieldsN(line string, maxFields int) []string {
	return processing.ParseMySQLTupleFieldsN(line, maxFields)
}

func parseDumpJobID(raw string) (uuid.UUID, error) {
	raw = strings.TrimSpace(raw)
	if id, err := uuid.Parse(raw); err == nil {
		return id, nil
	}
	// Legacy dump mixes UUID and numeric string PKs — keep remaps stable (migrator parity).
	return uuid.NewSHA1(uuid.NameSpaceOID, []byte("processing_job:"+raw)), nil
}

func nullish(s string) string {
	s = strings.TrimSpace(s)
	if s == "" || strings.EqualFold(s, "NULL") {
		return ""
	}
	return s
}

func atoiDefault(s string, def int) int {
	s = nullish(s)
	if s == "" {
		return def
	}
	n, err := strconv.Atoi(s)
	if err != nil {
		return def
	}
	return n
}

func parseDumpTime(s string) time.Time {
	s = nullish(s)
	if s == "" {
		return time.Time{}
	}
	for _, layout := range []string{
		"2006-01-02 15:04:05",
		time.RFC3339,
		"2006-01-02 15:04:05.000",
	} {
		if t, err := time.ParseInLocation(layout, s, time.UTC); err == nil {
			return t.UTC()
		}
	}
	return time.Time{}
}

func parseDumpTimePtr(s string) *time.Time {
	t := parseDumpTime(s)
	if t.IsZero() {
		return nil
	}
	return &t
}

func normalizeProcessingJobStatus(raw string) string {
	switch strings.ToLower(strings.TrimSpace(raw)) {
	case "completed", "success", "done":
		return "completed"
	case "failed", "error":
		return "failed"
	case "cancelled", "canceled", "skipped":
		return "cancelled"
	case "running", "processing":
		return "running"
	case "pending", "queued":
		return "pending"
	default:
		return "failed"
	}
}

func normalizeJobProductStatus(raw string) string {
	switch strings.ToLower(strings.TrimSpace(raw)) {
	case "processed", "completed", "success", "done":
		return "processed"
	case "failed", "error":
		return "failed"
	case "cancelled", "canceled", "skipped":
		return "cancelled"
	case "processing", "running":
		return "processing"
	case "pending", "queued":
		return "pending"
	default:
		return "failed"
	}
}