descrybe/apps/api/cmd/_thin_synth_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"
)

// timeoutCompleter forces enhance deadline so synthesis path is exercised live.
type timeoutCompleter struct{}

func (timeoutCompleter) Complete(context.Context, string, string) (processing.Completion, error) {
	return processing.Completion{}, context.DeadlineExceeded
}

func main() {
	log.SetOutput(logredact.Writer(os.Stderr))
	ean := strings.TrimSpace(os.Getenv("PROBE_EAN"))
	if ean == "" {
		ean = "4719331854218"
	}
	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()

	pipeline := processing.NewPipeline(pool)
	pipeline.BatchSize = cfg.ProcessingBatchSize
	pipeline.AI = nil
	pipeline.Prompts = aiprompts.NewService(pool)
	pipeline.Engine = &processing.Engine{
		Completer:    timeoutCompleter{},
		Vector:       processing.NoopVectorCategorizer{},
		EPREL:        nil,
		ProviderMode: processing.AIProviderInternal,
	}

	var rawID uuid.UUID
	if err := pool.QueryRow(ctx, `
		SELECT id FROM raw_products
		WHERE company_id = $1 AND gtin = $2`, companyID, ean).Scan(&rawID); err != nil {
		log.Fatalf("raw product %s: %v", ean, err)
	}
	log.Printf("ean=%s raw_id=%s", ean, rawID)

	_, _ = pool.Exec(ctx, `
		UPDATE processed_products
		SET processed_description = '', description = '',
			field_sources = COALESCE(field_sources, '{}'::jsonb) - 'enhance_input_hash',
			updated_at = now()
		WHERE company_id = $1 AND product_id = $2`, companyID, ean)

	_, _ = pool.Exec(ctx, `
		UPDATE processing_jobs
		SET status = 'failed', error = 'yielded to thin-synth probe', updated_at = now()
		WHERE status IN ('pending','running','processing')`)

	jobs, err := pipeline.StartJob(ctx, companyID, userID, []uuid.UUID{rawID}, "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)

	defend := make(chan struct{})
	go func() {
		t := time.NewTicker(300 * 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 thin-synth 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 '%parked%' OR error ILIKE '%yield%' OR error ILIKE '%reclaim%'
					       OR error ILIKE '%exclusive%' OR error ILIKE '%cleared%' OR error ILIKE '%P0%')`, 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`, 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, 2*time.Minute)
	defer runCancel()
	if err := pipeline.ProcessJob(runCtx, jobID); err != nil {
		close(defend)
		log.Fatalf("ProcessJob: %v", err)
	}
	close(defend)

	var name, desc, mode string
	var descLen int
	if err := pool.QueryRow(ctx, `
		SELECT COALESCE(name,''), COALESCE(processed_description,''), length(COALESCE(processed_description,'')),
			COALESCE(ai_provider_mode,'')
		FROM processed_products
		WHERE company_id = $1 AND product_id = $2`, companyID, ean).
		Scan(&name, &desc, &descLen, &mode); err != nil {
		log.Fatalf("load processed: %v", err)
	}
	fmt.Printf("EAN %s\n", ean)
	fmt.Printf("name=%q\n", name)
	fmt.Printf("processed_description_len=%d\n", descLen)
	fmt.Printf("processed_description=%q\n", desc)
	fmt.Printf("ai_provider_mode=%q\n", mode)
	echo := strings.EqualFold(strings.TrimSpace(desc), strings.TrimSpace(name))
	pass := descLen > 0 && !echo && !strings.Contains(strings.ToLower(desc), "ready for retail listing")
	fmt.Printf("PASS_DESC_NE_TITLE=%v\n", pass)
	if !pass {
		os.Exit(1)
	}

	out := os.Getenv("PROBE_OUT_DIR")
	if out == "" {
		out = `F:\laragon\www\_MY\descrybe-v2\.codehelper\_process_probe_thin_synth`
	}
	_ = os.MkdirAll(out, 0o755)
	_ = os.WriteFile(out+string(os.PathSeparator)+"process_id.txt", []byte(jobID.String()+"\n"), 0o644)
	_ = os.WriteFile(out+string(os.PathSeparator)+"score.txt", []byte(fmt.Sprintf(
		"ean=%s\nname=%s\npd_len=%d\npd=%s\nmode=%s\nPASS=%v\n", ean, name, descLen, desc, mode, pass,
	)), 0o644)
}