descrybe/apps/api/cmd/_electronics_probe_tmp/main.go

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
	"github.com/descrybe/descrybe-v2/apps/api/internal/config"
	"github.com/descrybe/descrybe-v2/apps/api/internal/db"
	"github.com/descrybe/descrybe-v2/apps/api/internal/logredact"
	"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
	"github.com/google/uuid"
)

func main() {
	log.SetOutput(logredact.Writer(os.Stderr))
	eans := []string{
		"4548736132597",
		"194644207427",
		"194644034894",
		"195949822384",
		"3830061960930",
	}
	if v := strings.TrimSpace(os.Getenv("PROBE_EANS")); v != "" {
		eans = strings.Split(v, ",")
		for i := range eans {
			eans[i] = strings.TrimSpace(eans[i])
		}
	}
	companyID, err := uuid.Parse("2b3159b0-fc08-415b-b248-35ed02a6baab")
	if err != nil {
		log.Fatal(err)
	}
	userID, err := uuid.Parse("0ce3305c-d810-4b56-b1d4-3c1ed510db76")
	if err != nil {
		log.Fatal(err)
	}

	cfg, err := config.Load()
	if err != nil {
		log.Fatalf("config: %v", err)
	}
	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer cancel()

	pool, err := db.NewPool(ctx, cfg.DatabaseURL, db.PoolOptions{
		MaxConns:              int32(cfg.DBMaxConns),
		MinConns:              int32(cfg.DBMinConns),
		MaxConnLifetime:       cfg.DBMaxConnLifetime,
		MaxConnLifetimeJitter: cfg.DBMaxConnLifetimeJitter,
		MaxConnIdleTime:       cfg.DBMaxConnIdleTime,
		HealthCheckPeriod:     cfg.DBHealthCheckPeriod,
		StatementTimeout:      cfg.DBStatementTimeout,
	})
	if err != nil {
		log.Fatalf("db: %v", err)
	}
	defer pool.Close()

	pipeline := processing.NewPipeline(pool)
	pipeline.BatchSize = cfg.ProcessingBatchSize
	// Force mock-llm: platform DB OpenAI binding would otherwise override env.
	pipeline.AI = nil
	pipeline.Prompts = aiprompts.NewService(pool)
	mockKey := strings.TrimSpace(cfg.OpenAIAPIKey)
	if mockKey == "" {
		mockKey = "local-test"
	}
	mockBase := strings.TrimSpace(cfg.OpenAIBaseURL)
	if mockBase == "" {
		mockBase = "http://127.0.0.1:18767/v1"
	}
	mockModel := strings.TrimSpace(cfg.OpenAIModel)
	if mockModel == "" {
		mockModel = "mock-llm"
	}
	log.Printf("forcing completer base=%s model=%s", mockBase, mockModel)
	pipeline.Engine = &processing.Engine{
		Completer:    processing.NewOpenAIClient(mockKey, mockBase, mockModel, cfg.ProcessingRPM, cfg.ProcessingMaxRetries),
		Vector:       processing.NoopVectorCategorizer{},
		EPREL:        nil, // skip remote EPREL hangs during local scorecard
		ProviderMode: processing.AIProviderInternal,
	}

	rawIDs := make([]uuid.UUID, 0, len(eans))
	for _, ean := range eans {
		var id uuid.UUID
		err := pool.QueryRow(ctx, `
			SELECT id FROM raw_products
			WHERE company_id = $1 AND gtin = $2`, companyID, ean).Scan(&id)
		if err != nil {
			log.Fatalf("raw product %s: %v", ean, err)
		}
		rawIDs = append(rawIDs, id)
		log.Printf("ean=%s raw_id=%s", ean, id)
	}

	// Clear competing queue noise.
	_, _ = pool.Exec(ctx, `
		UPDATE processing_jobs
		SET status = 'failed', error = 'yielded to electronics in-process probe', updated_at = now()
		WHERE status IN ('pending','running','processing')`)

	jobs, err := pipeline.StartJob(ctx, companyID, userID, rawIDs, "full")
	if err != nil {
		log.Fatalf("StartJob: %v", err)
	}
	if len(jobs) == 0 {
		log.Fatal("StartJob returned no jobs")
	}
	jobID := jobs[0].ID
	log.Printf("job_id=%s chunks=%d", jobID, len(jobs))

	defend := make(chan struct{})
	go func() {
		t := time.NewTicker(400 * time.Millisecond)
		defer t.Stop()
		for {
			select {
			case <-defend:
				return
			case <-ctx.Done():
				return
			case <-t.C:
				_, _ = pool.Exec(context.Background(), `
					UPDATE processing_jobs
					SET status = 'failed', error = 'yielded to electronics in-process probe', updated_at = now()
					WHERE status IN ('pending','running','processing') AND id <> $1`, jobID)
				_, _ = pool.Exec(context.Background(), `
					UPDATE processing_jobs
					SET status = 'running', error = NULL, updated_at = now()
					WHERE id = $1 AND status IN ('failed','cancelled')
					  AND (error ILIKE '%P0%' OR error ILIKE '%parked%' OR error ILIKE '%yield%' OR error ILIKE '%reclaim%' OR error ILIKE '%cleared%' OR error ILIKE '%superseded%' OR error ILIKE '%abandoned%')`, jobID)
				_, _ = pool.Exec(context.Background(), `
					UPDATE processing_job_products
					SET status = 'pending', error = NULL, updated_at = now()
					WHERE job_id = $1 AND status IN ('failed','cancelled')
					  AND processed_product_id IS NULL
					  AND (error ILIKE '%P0%' OR error ILIKE '%parked%' OR error ILIKE '%yield%' OR error ILIKE '%reclaim%' OR error ILIKE '%cleared%' OR error ILIKE '%superseded%' OR error ILIKE '%abandoned%')`, jobID)
			}
		}
	}()

	_, _ = pool.Exec(ctx, `
		UPDATE processing_jobs
		SET status = 'running', started_at = COALESCE(started_at, now()), updated_at = now()
		WHERE id = $1`, jobID)

	runCtx, runCancel := context.WithTimeout(ctx, 3*time.Minute)
	defer runCancel()
	if err := pipeline.ProcessJob(runCtx, jobID); err != nil {
		close(defend)
		log.Fatalf("ProcessJob: %v", err)
	}
	close(defend)

	rows, err := pool.Query(ctx, `
		SELECT rp.gtin, pjp.status, COALESCE(pjp.error,''), pjp.processed_product_id IS NOT NULL
		FROM processing_job_products pjp
		JOIN raw_products rp ON rp.id = pjp.raw_product_id
		WHERE pjp.job_id = $1
		ORDER BY rp.gtin`, jobID)
	if err != nil {
		log.Fatalf("query items: %v", err)
	}
	defer rows.Close()
	fmt.Printf("JOB %s\n", jobID)
	for rows.Next() {
		var gtin, status, errMsg string
		var hasPP bool
		if err := rows.Scan(&gtin, &status, &errMsg, &hasPP); err != nil {
			log.Fatal(err)
		}
		fmt.Printf("ITEM %s status=%s has_pp=%v err=%s\n", gtin, status, hasPP, errMsg)
	}
	out := os.Getenv("PROBE_OUT_DIR")
	if out == "" {
		out = `F:\laragon\www\_MY\descrybe-v2\.codehelper\_process_probe_electronics`
	}
	_ = os.MkdirAll(out, 0o755)
	_ = os.WriteFile(out+string(os.PathSeparator)+"process_id.txt", []byte(jobID.String()+"\n"), 0o644)
}