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

package main

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

	"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"
)

// Classic A1 Postman / Elkotex fixture EANs — must never live on Platform Demo.
var a1FixtureEANs = []string{
	"5905575903198",
	"6970995789942",
}

type categoryBackfillResult struct {
	ProcessedUpdated    int64
	ProcessedInserted   int64
	MappedUpdated       int64
	DumpPairs           int
	PurgedOtherRaw      int64
	PurgedOtherPP       int64
	ProcessedWithCat    int
	ProcessedWithoutCat int
	MappedWithCat       int
	MappedWithoutCat    int
}

// backfillMappedCategoriesFromProcessed copies processed_products.category into
// raw_products.mapped_data.category for A1 only. Delegates to processing.
func backfillMappedCategoriesFromProcessed(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID) (int64, error) {
	return processing.BackfillMappedCategoriesFromProcessed(ctx, pg, companyID)
}

// backfillCategoriesFromMySQLDump streams dump processed_products for the A1
// legacy company and writes product_id (GTIN) → category onto A1 Postgres
// mapped_data.category (and updates any existing processed_products.category).
// It does not insert processed rows — A1 demo seed stays at processed=0.
func backfillCategoriesFromMySQLDump(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID, dumpPath string) (categoryBackfillResult, error) {
	var out categoryBackfillResult
	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
	}

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

	byGTIN, err := scanA1ProcessedCategories(f, legacyCompany)
	if err != nil {
		return out, err
	}
	out.DumpPairs = len(byGTIN)
	if len(byGTIN) == 0 {
		return out, fmt.Errorf("no A1 processed_products categories for legacy %s in dump", legacyCompany)
	}
	log.Printf("dump: %d A1 gtin→category pairs", len(byGTIN))

	gtins := make([]string, 0, len(byGTIN))
	cats := make([]string, 0, len(byGTIN))
	for g, c := range byGTIN {
		gtins = append(gtins, g)
		cats = append(cats, c)
	}

	ct, err := pg.Exec(ctx, `
		UPDATE processed_products p
		SET category = v.category,
			updated_at = now()
		FROM unnest($2::text[], $3::text[]) AS v(gtin, category)
		WHERE p.company_id = $1
		  AND p.product_id = v.gtin
		  AND COALESCE(NULLIF(trim(v.category), ''), '') <> ''
		  AND (
			COALESCE(NULLIF(trim(p.category), ''), '') = ''
			OR lower(trim(p.category)) = 'none'
			OR p.category IS DISTINCT FROM v.category
		  )`, companyID, gtins, cats)
	if err != nil {
		return out, fmt.Errorf("update processed category from dump: %w", err)
	}
	out.ProcessedUpdated = ct.RowsAffected()

	// A1 demo seed keeps processed=0. Do not INSERT processed rows from the dump —
	// only refresh mapped_data.category (and any existing processed rows if present).
	ct, err = pg.Exec(ctx, `
		UPDATE raw_products r
		SET mapped_data = jsonb_set(
			COALESCE(r.mapped_data, '{}'::jsonb),
			'{category}',
			to_jsonb(v.category),
			true
		),
		updated_at = now()
		FROM unnest($2::text[], $3::text[]) AS v(gtin, category)
		WHERE r.company_id = $1
			AND r.gtin = v.gtin
			AND COALESCE(NULLIF(trim(v.category), ''), '') <> ''
			AND (
				COALESCE(NULLIF(trim(r.mapped_data->>'category'), ''), '') = ''
				OR r.mapped_data->>'category' IS DISTINCT FROM v.category
			)`, companyID, gtins, cats)
	if err != nil {
		return out, fmt.Errorf("update mapped category from dump: %w", err)
	}
	out.MappedUpdated = ct.RowsAffected()

	n, err := backfillMappedCategoriesFromProcessed(ctx, pg, companyID)
	if err != nil {
		return out, err
	}
	out.MappedUpdated += n
	if err := fillCategoryCoverage(ctx, pg, companyID, &out); err != nil {
		return out, err
	}
	return out, nil
}

func fillCategoryCoverage(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID, out *categoryBackfillResult) error {
	err := pg.QueryRow(ctx, `
		SELECT
			COUNT(*) FILTER (
				WHERE COALESCE(NULLIF(trim(category), ''), '') <> ''
					AND lower(trim(category)) <> 'none'
			),
			COUNT(*) FILTER (
				WHERE COALESCE(NULLIF(trim(category), ''), '') = ''
					OR lower(trim(category)) = 'none'
			)
		FROM processed_products
		WHERE company_id = $1`, companyID).Scan(&out.ProcessedWithCat, &out.ProcessedWithoutCat)
	if err != nil {
		return fmt.Errorf("count processed categories: %w", err)
	}
	err = pg.QueryRow(ctx, `
		SELECT
			COUNT(*) FILTER (
				WHERE COALESCE(NULLIF(trim(mapped_data->>'category'), ''), '') <> ''
			),
			COUNT(*) FILTER (
				WHERE COALESCE(NULLIF(trim(mapped_data->>'category'), ''), '') = ''
			)
		FROM raw_products
		WHERE company_id = $1`, companyID).Scan(&out.MappedWithCat, &out.MappedWithoutCat)
	if err != nil {
		return fmt.Errorf("count mapped categories: %w", err)
	}
	return nil
}

// purgeA1FixtureEANsFromOtherTenants deletes the Postman Elkotex fixture EANs
// from every company except A1 (Platform Demo must not mirror A1 fixtures).
func purgeA1FixtureEANsFromOtherTenants(ctx context.Context, pg *pgxpool.Pool, a1CompanyID uuid.UUID) (rawN, ppN int64, err error) {
	ct, err := pg.Exec(ctx, `
		DELETE FROM processing_job_products pjp
		WHERE pjp.raw_product_id IN (
			SELECT id FROM raw_products
			WHERE company_id <> $1 AND gtin = ANY($2::text[])
		)
		OR pjp.processed_product_id IN (
			SELECT id FROM processed_products
			WHERE company_id <> $1 AND product_id = ANY($2::text[])
		)`, a1CompanyID, a1FixtureEANs)
	if err != nil {
		return 0, 0, fmt.Errorf("purge fixture job products: %w", err)
	}
	_ = ct

	ct, err = pg.Exec(ctx, `
		DELETE FROM processed_products
		WHERE company_id <> $1 AND product_id = ANY($2::text[])`, a1CompanyID, a1FixtureEANs)
	if err != nil {
		return 0, 0, fmt.Errorf("purge fixture processed: %w", err)
	}
	ppN = ct.RowsAffected()

	ct, err = pg.Exec(ctx, `
		DELETE FROM raw_products
		WHERE company_id <> $1 AND gtin = ANY($2::text[])`, a1CompanyID, a1FixtureEANs)
	if err != nil {
		return 0, 0, fmt.Errorf("purge fixture raw: %w", err)
	}
	rawN = ct.RowsAffected()
	return rawN, ppN, nil
}

func scanA1ProcessedCategories(r io.Reader, legacyCompany string) (map[string]string, error) {
	br := bufio.NewReaderSize(r, 1<<20)
	inTable := false
	out := make(map[string]string, 4096)
	for {
		line, err := br.ReadString('\n')
		if len(line) > 0 {
			trimmed := strings.TrimSpace(line)
			if strings.HasPrefix(trimmed, "INSERT INTO `processed_products`") ||
				strings.HasPrefix(trimmed, "INSERT INTO processed_products") {
				inTable = true
			} else if inTable && strings.HasPrefix(trimmed, "CREATE TABLE") {
				break
			} else if inTable && strings.HasPrefix(trimmed, "INSERT INTO `") &&
				!strings.Contains(trimmed, "processed_products") {
				break
			} else if inTable && looksLikeTupleLine(line) && strings.Contains(line, legacyCompany) {
				fields := parseMySQLTupleFieldsN(line, 6)
				if len(fields) >= 5 {
					gtin := strings.TrimSpace(fields[2])
					cat := strings.TrimSpace(fields[4])
					if gtin != "" && cat != "" && !strings.EqualFold(cat, "NULL") {
						out[gtin] = cat
					}
				}
			}
		}
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, err
		}
	}
	return out, nil
}

func logCategoryBackfillResult(res categoryBackfillResult, source string) {
	log.Printf("category backfill (%s): dump_pairs=%d processed_updated=%d processed_inserted=%d mapped_updated=%d purged_other_raw=%d purged_other_pp=%d",
		source, res.DumpPairs, res.ProcessedUpdated, res.ProcessedInserted, res.MappedUpdated, res.PurgedOtherRaw, res.PurgedOtherPP)
	log.Printf("A1 coverage: processed with_cat=%d without_cat=%d | raw mapped with_cat=%d without_cat=%d",
		res.ProcessedWithCat, res.ProcessedWithoutCat, res.MappedWithCat, res.MappedWithoutCat)
}