descrybe/apps/api/cmd/seed-a1-reset-processing/main.go

package main

// Command seed-a1-reset-processing returns the A1 Slovenija tenant to a fresh
// processing state without wiping catalog inputs.
//
// Deletes/cancels processing jobs and job-product links, deletes processed_products,
// and sets all raw_products to unprocessed. Retains feeds, mappings, raw/mapped
// product payloads, attributes, categories, standard fields, and export feeds.
//
// Idempotent and company-scoped. Run AFTER seed-a1 reimport — do not bake into
// the main seed path.
//
// Usage:
//
//	go run ./cmd/seed-a1-reset-processing
//	go run ./cmd/seed-a1-reset-processing -company 604f23a8-b66e-4b21-8b45-0d72b68f4790
//	go run ./cmd/seed-a1-reset-processing -dry-run
//
// DATABASE_URL / -postgres required.
import (
	"context"
	"flag"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

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

const defaultA1CompanyID = "604f23a8-b66e-4b21-8b45-0d72b68f4790"

type counts struct {
	Raw          int64
	Processed    int64
	Unprocessed  int64
	Jobs         int64
	JobProducts  int64
	FeedSyncJobs int64
}

func main() {
	postgresURL := flag.String("postgres", os.Getenv("DATABASE_URL"), "Postgres URL (DATABASE_URL)")
	company := flag.String("company", defaultA1CompanyID, "Postgres companies.id (default A1 Slovenija)")
	byName := flag.String("name", "", "Resolve company by name (e.g. \"A1 Slovenija\") when -company omitted/wrong")
	dryRun := flag.Bool("dry-run", false, "print before counts only; do not mutate")
	flag.Parse()

	if strings.TrimSpace(*postgresURL) == "" {
		log.Fatal("-postgres / DATABASE_URL is required")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
	defer cancel()

	pg, err := pgxpool.New(ctx, *postgresURL)
	if err != nil {
		log.Fatalf("postgres: %v", err)
	}
	defer pg.Close()

	companyID, err := resolveCompany(ctx, pg, *company, *byName)
	if err != nil {
		log.Fatal(err)
	}

	before, err := loadCounts(ctx, pg, companyID)
	if err != nil {
		log.Fatalf("counts: %v", err)
	}
	log.Printf("company %s before: raw=%d processed=%d unprocessed=%d jobs=%d job_products=%d feed_sync_jobs=%d",
		companyID, before.Raw, before.Processed, before.Unprocessed, before.Jobs, before.JobProducts, before.FeedSyncJobs)

	if *dryRun {
		log.Printf("dry-run: no changes")
		return
	}

	if err := resetProcessing(ctx, pg, companyID); err != nil {
		log.Fatalf("reset: %v", err)
	}

	after, err := loadCounts(ctx, pg, companyID)
	if err != nil {
		log.Fatalf("counts after: %v", err)
	}
	log.Printf("company %s after:  raw=%d processed=%d unprocessed=%d jobs=%d job_products=%d feed_sync_jobs=%d",
		companyID, after.Raw, after.Processed, after.Unprocessed, after.Jobs, after.JobProducts, after.FeedSyncJobs)
	if after.Processed != 0 || after.Jobs != 0 || after.JobProducts != 0 {
		log.Fatalf("expected processed=0 jobs=0 job_products=0; got processed=%d jobs=%d job_products=%d",
			after.Processed, after.Jobs, after.JobProducts)
	}
	if after.Raw != before.Raw {
		log.Fatalf("raw catalog changed (%d → %d) — abort expectation failed", before.Raw, after.Raw)
	}
	if after.Unprocessed != after.Raw {
		log.Fatalf("expected all raw unprocessed (%d), got %d", after.Raw, after.Unprocessed)
	}
	log.Printf("ok: catalog retained, processing state cleared")
}

func resolveCompany(ctx context.Context, pg *pgxpool.Pool, companyFlag, nameFlag string) (uuid.UUID, error) {
	nameFlag = strings.TrimSpace(nameFlag)
	if nameFlag != "" {
		var id uuid.UUID
		err := pg.QueryRow(ctx, `
			SELECT id FROM companies
			WHERE lower(name) = lower($1)
			ORDER BY created_at ASC LIMIT 1`, nameFlag).Scan(&id)
		if err != nil {
			return uuid.Nil, fmt.Errorf("resolve -name %q: %w", nameFlag, err)
		}
		return id, nil
	}
	id, err := uuid.Parse(strings.TrimSpace(companyFlag))
	if err != nil {
		return uuid.Nil, fmt.Errorf("-company: %w", err)
	}
	var name string
	err = pg.QueryRow(ctx, `SELECT name FROM companies WHERE id = $1`, id).Scan(&name)
	if err != nil {
		return uuid.Nil, fmt.Errorf("company %s not found: %w", id, err)
	}
	log.Printf("resolved company %s (%s)", id, name)
	return id, nil
}

func loadCounts(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID) (counts, error) {
	var c counts
	err := pg.QueryRow(ctx, `
		SELECT
			(SELECT count(*) FROM raw_products WHERE company_id = $1),
			(SELECT count(*) FROM processed_products WHERE company_id = $1),
			(SELECT count(*) FROM raw_products WHERE company_id = $1 AND processing_status = 'unprocessed' AND is_processed = false),
			(SELECT count(*) FROM processing_jobs WHERE company_id = $1),
			(SELECT count(*) FROM processing_job_products pjp
				JOIN processing_jobs pj ON pj.id = pjp.job_id WHERE pj.company_id = $1),
			(SELECT count(*) FROM feed_sync_jobs WHERE company_id = $1)
	`, companyID).Scan(&c.Raw, &c.Processed, &c.Unprocessed, &c.Jobs, &c.JobProducts, &c.FeedSyncJobs)
	return c, err
}

func resetProcessing(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID) error {
	tx, err := pg.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)

	// Job products before jobs (FK).
	if _, err := tx.Exec(ctx, `
		DELETE FROM processing_job_products
		WHERE job_id IN (SELECT id FROM processing_jobs WHERE company_id = $1)`, companyID); err != nil {
		return fmt.Errorf("delete job products: %w", err)
	}
	if _, err := tx.Exec(ctx, `
		DELETE FROM processing_jobs WHERE company_id = $1`, companyID); err != nil {
		return fmt.Errorf("delete jobs: %w", err)
	}
	// Ephemeral sync job rows (optional clutter); raw.sync_job_id SET NULL on delete.
	if _, err := tx.Exec(ctx, `
		DELETE FROM feed_sync_jobs WHERE company_id = $1`, companyID); err != nil {
		return fmt.Errorf("delete feed sync jobs: %w", err)
	}
	if _, err := tx.Exec(ctx, `
		DELETE FROM processed_products WHERE company_id = $1`, companyID); err != nil {
		return fmt.Errorf("delete processed: %w", err)
	}
	ct, err := tx.Exec(ctx, `
		UPDATE raw_products
		SET is_processed = false,
			processing_status = 'unprocessed',
			updated_at = now()
		WHERE company_id = $1
		  AND (is_processed = true OR processing_status <> 'unprocessed')`, companyID)
	if err != nil {
		return fmt.Errorf("reset raw: %w", err)
	}
	log.Printf("raw rows reset to unprocessed: %d", ct.RowsAffected())

	return tx.Commit(ctx)
}