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

package main

import (
	"context"
	"encoding/json"
	"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/eprel"
	"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{
		"4242005342488",
		"4242005479948",
		"4242002996967",
		"4242005327065",
		"4242005382330",
	}
	companyID := uuid.MustParse("2b3159b0-fc08-415b-b248-35ed02a6baab")
	userID := uuid.MustParse("0ce3305c-d810-4b56-b1d4-3c1ed510db76")

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

	out := os.Getenv("PROBE_OUT_DIR")
	if out == "" {
		out = `F:\laragon\www\_MY\descrybe-v2\.codehelper\_probe_batch_a`
	}
	_ = os.MkdirAll(out, 0o755)

	pipeline := processing.NewPipeline(pool)
	pipeline.BatchSize = cfg.ProcessingBatchSize
	pipeline.AI = nil
	pipeline.Prompts = aiprompts.NewService(pool)

	mockKey := strings.TrimSpace(cfg.OpenAIAPIKey)
	if mockKey == "" {
		mockKey = "local-test"
	}
	mockBase := "http://127.0.0.1:18767/v1"
	mockModel := "mock-llm"
	log.Printf("forcing completer base=%s model=%s (ModeLabel=internal)", mockBase, mockModel)
	client := processing.NewOpenAIClient(mockKey, mockBase, mockModel, cfg.ProcessingRPM, cfg.ProcessingMaxRetries)
	client.ModeLabel = processing.AIProviderInternal

	eprelClient := eprel.NewClient(eprel.Options{Enabled: true, Timeout: 20 * time.Second})
	pipeline.Engine = &processing.Engine{
		Completer:    client,
		Vector:       processing.NoopVectorCategorizer{},
		EPREL:        eprelClient,
		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)
	}

	// Self-heal only: do NOT fail/reclaim other jobs.
	if n, err := processing.BackfillMissingMeta(ctx, pool, companyID); err != nil {
		log.Printf("BackfillMissingMeta: %v", err)
	} else {
		log.Printf("BackfillMissingMeta updated=%d", n)
	}

	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", jobID)
	_ = os.WriteFile(out+string(os.PathSeparator)+"process_id.txt", []byte(jobID.String()+"\n"), 0o644)

	defend := make(chan struct{})
	go func() {
		t := time.NewTicker(500 * time.Millisecond)
		defer t.Stop()
		for {
			select {
			case <-defend:
				return
			case <-ctx.Done():
				return
			case <-t.C:
				// Restore ONLY our job if a sibling parked/yielded it. Never touch other jobs.
				_, _ = 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 '%exclusive%' OR error ILIKE '%batch_%' OR error ILIKE '%inproc%')`, 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 '%exclusive%' OR error ILIKE '%batch_%' OR error ILIKE '%inproc%')`, 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, 12*time.Minute)
	defer runCancel()
	if err := pipeline.ProcessJob(runCtx, jobID); err != nil {
		close(defend)
		log.Fatalf("ProcessJob: %v", err)
	}
	close(defend)

	items, err := pipeline.LoadV1ProcessJobItems(ctx, companyID, jobID, "full")
	if err != nil {
		log.Fatalf("LoadV1ProcessJobItems: %v", err)
	}
	payload := map[string]any{
		"data": map[string]any{
			"process_id":      jobID.String(),
			"status":          "COMPLETED",
			"processing_type": "full",
			"total_items":     len(items),
			"items":           items,
		},
	}
	raw, err := json.MarshalIndent(payload, "", "  ")
	if err != nil {
		log.Fatalf("marshal: %v", err)
	}
	if err := os.WriteFile(out+string(os.PathSeparator)+"final.json", raw, 0o644); err != nil {
		log.Fatalf("write final: %v", err)
	}
	fmt.Printf("JOB %s items=%d\n", jobID, len(items))
	for _, it := range items {
		ean, _ := it["ean"].(string)
		st, _ := it["status"].(string)
		title, _ := it["title"].(string)
		if len(title) > 50 {
			title = title[:50]
		}
		_, hasEprel := it["eprel"].(map[string]any)
		ai, _ := it["ai_provider_mode"].(string)
		fmt.Printf("ITEM %s status=%s eprel=%v ai=%s title=%s\n", ean, st, hasEprel, ai, title)
	}
}