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