package main import ( "bufio" "context" "fmt" "io" "log" "os" "strings" "github.com/descrybe/descrybe-v2/apps/api/internal/billing" "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. Legacy dumps store category on // processed rows (unique_id codes); feed mappings never mapped a category field, // so re-processing without this backfill yields Uncategorized / grey C coverage. func backfillMappedCategoriesFromProcessed(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID) (int64, error) { ct, err := pg.Exec(ctx, ` UPDATE raw_products r SET mapped_data = jsonb_set( COALESCE(r.mapped_data, '{}'::jsonb), '{category}', to_jsonb(p.category), true ), updated_at = now() FROM processed_products p WHERE p.raw_product_id = r.id AND p.company_id = $1 AND r.company_id = $1 AND COALESCE(NULLIF(trim(p.category), ''), '') <> '' AND lower(trim(p.category)) <> 'none' AND COALESCE(NULLIF(trim(r.mapped_data->>'category'), ''), '') = ''`, companyID) if err != nil { return 0, fmt.Errorf("backfill mapped category: %w", err) } return ct.RowsAffected(), nil } // 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) }