// Command formula-e2e runs the real processing pipeline against the local database // and reports whether category formulas actually drove the generated copy. // // It is a local proof, not a test double: it builds processing.Pipeline the way // cmd/worker does (same AI resolution, same prompts service, same Engine), starts a // real job, processes it, and then reads processed_products back. // // go run ./cmd/mock-llm -addr 127.0.0.1:18767 & // go run ./cmd/formula-e2e -company "A1 Slovenija" -limit 3 // go run ./cmd/formula-e2e -company "A1 Slovenija" -gtin 8022068075495 // go run ./cmd/formula-e2e -new-tenant # English defaults on a fresh company package main import ( "context" "encoding/json" "flag" "fmt" "log" "strings" "time" "github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit" "github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts" "github.com/descrybe/descrybe-v2/apps/api/internal/aiprovider" "github.com/descrybe/descrybe-v2/apps/api/internal/catalog" "github.com/descrybe/descrybe-v2/apps/api/internal/config" "github.com/descrybe/descrybe-v2/apps/api/internal/platformsettings" "github.com/descrybe/descrybe-v2/apps/api/internal/processing" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" ) func main() { companyName := flag.String("company", "A1 Slovenija", "company name to process") gtin := flag.String("gtin", "", "specific GTIN to process (default: first products with a category)") limit := flag.Int("limit", 2, "how many products to process") ptype := flag.String("type", "full", "processing_type") newTenant := flag.Bool("new-tenant", false, "create a throwaway English tenant + category + product and process that") keep := flag.Bool("keep", false, "keep the throwaway tenant instead of deleting it") llmBase := flag.String("llm-base", "", "override the resolved LLM base URL (e.g. http://127.0.0.1:18767/v1 for cmd/mock-llm)") llmKey := flag.String("llm-key", "local-test", "API key for -llm-base") llmModel := flag.String("llm-model", "mock-llm", "model id for -llm-base") flag.Parse() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) defer cancel() cfg, err := config.Load() if err != nil { log.Fatalf("config: %v", err) } pool, err := pgxpool.New(ctx, cfg.DatabaseURL) if err != nil { log.Fatalf("db: %v", err) } defer pool.Close() if err := pool.Ping(ctx); err != nil { log.Fatalf("db ping: %v", err) } pipeline, auditedSvc := newWorkerLikePipeline(ctx, pool, cfg) if strings.TrimSpace(*llmBase) != "" { // Everything else stays production-identical; only the provider endpoint is // pinned so a local run does not depend on the configured remote model. pipeline.AI = fixedCompleter{ client: auditedSvc.WrapAudit( processing.NewOpenAIClient(*llmKey, *llmBase, *llmModel, 0, 2), uuid.Nil, "processing", processing.AIProviderInternal), } log.Printf("LLM override: base=%s model=%s", *llmBase, *llmModel) } if *newTenant { if err := runNewTenant(ctx, pool, pipeline, *keep); err != nil { log.Fatalf("new tenant: %v", err) } return } if err := runExisting(ctx, pool, pipeline, *companyName, *gtin, *limit, *ptype); err != nil { log.Fatalf("run: %v", err) } } // fixedCompleter pins every job to one endpoint (local mock-llm) while leaving the // rest of the pipeline exactly as the worker builds it. type fixedCompleter struct{ client processing.Completer } func (f fixedCompleter) ResolveCompleter(context.Context, uuid.UUID) (processing.Completer, string, bool, error) { return f.client, processing.AIProviderInternal, false, nil } func (f fixedCompleter) ResolveCompleterForRole(ctx context.Context, id uuid.UUID, _ string) (processing.Completer, string, bool, error) { return f.ResolveCompleter(ctx, id) } // newWorkerLikePipeline mirrors cmd/worker/main.go so this proof exercises the same // code path production does. func newWorkerLikePipeline(ctx context.Context, pool *pgxpool.Pool, cfg config.Config) (*processing.Pipeline, *aiprovider.Service) { platSettings := platformsettings.NewService(pool, platformsettings.EnvConfig{ AppEncryptionKey: cfg.AppEncryptionKey, CredentialsEncryptionKey: cfg.CredentialsEncryptionKey, TokenSigningSecret: cfg.TokenSigningSecret, DatabaseURL: cfg.DatabaseURL, OpenAIAPIKey: cfg.OpenAIAPIKey, OpenAIBaseURL: cfg.OpenAIBaseURL, OpenAIModel: cfg.OpenAIModel, }) aiSvc := aiprovider.NewService(pool, aiprovider.EnvConfig{ AppEncryptionKey: cfg.AppEncryptionKey, CredentialsEncryptionKey: cfg.CredentialsEncryptionKey, TokenSigningSecret: cfg.TokenSigningSecret, DatabaseURL: cfg.DatabaseURL, OpenAIAPIKey: cfg.OpenAIAPIKey, OpenAIBaseURL: cfg.OpenAIBaseURL, OpenAIModel: cfg.OpenAIModel, ProcessingRPM: cfg.ProcessingRPM, ProcessingMaxRetries: cfg.ProcessingMaxRetries, }) aiSvc.Platform = platSettings aiSvc.WithAudit(aiaudit.NewRecorder(pool)) p := processing.NewPipeline(pool) p.AI = aiSvc p.Prompts = aiprompts.NewService(pool) p.Limiter = nil p.Engine = &processing.Engine{ Vector: processing.NoopVectorCategorizer{}, ProviderMode: processing.AIProviderInternal, } if oi, err := platSettings.ResolveOpenAI(ctx); err != nil { log.Printf("WARNING: platform OpenAI resolve failed: %v", err) } else if strings.TrimSpace(oi.APIKey) == "" { log.Printf("WARNING: platform OpenAI unset — AI enhance will be skipped") } else { log.Printf("OpenAI: source=%s base=%s model=%s", oi.Source, oi.BaseURL, oi.Model) } return p, aiSvc } func runExisting(ctx context.Context, pool *pgxpool.Pool, p *processing.Pipeline, companyName, gtin string, limit int, ptype string) error { var companyID uuid.UUID var lang string err := pool.QueryRow(ctx, ` SELECT id, COALESCE(language, '') FROM companies WHERE name = $1`, companyName).Scan(&companyID, &lang) if err != nil { return fmt.Errorf("company %q: %w", companyName, err) } fmt.Printf("\n=== %s (%s, language=%s)\n", companyName, companyID, lang) var rawIDs []uuid.UUID if strings.TrimSpace(gtin) != "" { rows, err := pool.Query(ctx, ` SELECT id FROM raw_products WHERE company_id = $1 AND gtin = $2 LIMIT $3`, companyID, strings.TrimSpace(gtin), limit) if err != nil { return err } rawIDs, err = collectIDs(rows) if err != nil { return err } } else { rows, err := pool.Query(ctx, ` SELECT id FROM raw_products WHERE company_id = $1 AND COALESCE(NULLIF(trim(mapped_data->>'category'), ''), '') <> '' AND length(COALESCE(mapped_data->>'description', '')) > 120 ORDER BY id LIMIT $2`, companyID, limit) if err != nil { return err } rawIDs, err = collectIDs(rows) if err != nil { return err } } if len(rawIDs) == 0 { return fmt.Errorf("no matching raw products") } return processAndReport(ctx, pool, p, companyID, rawIDs, ptype) } func processAndReport(ctx context.Context, pool *pgxpool.Pool, p *processing.Pipeline, companyID uuid.UUID, rawIDs []uuid.UUID, ptype string) error { // Clear prior enhance hashes so the run cannot be skipped as "unchanged". // The hash lives twice: once on field_sources and once per language inside // localized_content — clearing only the first still hash-skips the LLM. if _, err := pool.Exec(ctx, ` UPDATE processed_products SET field_sources = field_sources - 'enhance_input_hash', localized_content = COALESCE(( SELECT jsonb_object_agg(k, v - 'enhance_input_hash') FROM jsonb_each(COALESCE(localized_content, '{}'::jsonb)) AS t(k, v) ), '{}'::jsonb) WHERE company_id = $1 AND raw_product_id = ANY($2)`, companyID, rawIDs); err != nil { return err } var userID uuid.UUID if err := pool.QueryRow(ctx, ` SELECT user_id FROM memberships WHERE company_id = $1 ORDER BY created_at LIMIT 1`, companyID).Scan(&userID); err != nil { return fmt.Errorf("membership: %w", err) } jobs, err := p.StartJob(ctx, companyID, userID, rawIDs, ptype) if err != nil { return fmt.Errorf("start job: %w", err) } for _, j := range jobs { if err := p.ProcessJob(ctx, j.ID); err != nil { return fmt.Errorf("process job %s: %w", j.ID, err) } } return report(ctx, pool, companyID, rawIDs) } func report(ctx context.Context, pool *pgxpool.Pool, companyID uuid.UUID, rawIDs []uuid.UUID) error { rows, err := pool.Query(ctx, ` SELECT COALESCE(rp.gtin, ''), COALESCE(pp.category, ''), COALESCE(rp.mapped_data->>'name', ''), COALESCE(rp.mapped_data->>'description', ''), COALESCE(pp.processed_name, ''), COALESCE(pp.processed_description, ''), COALESCE(pp.field_sources::text, '{}'), COALESCE(pp.processed_attributes::text, '{}') FROM raw_products rp LEFT JOIN processed_products pp ON pp.raw_product_id = rp.id AND pp.company_id = rp.company_id WHERE rp.company_id = $1 AND rp.id = ANY($2)`, companyID, rawIDs) if err != nil { return err } defer rows.Close() fails := 0 for rows.Next() { var gtin, category, feedName, feedDesc, name, desc, sources, attrs string if err := rows.Scan(>in, &category, &feedName, &feedDesc, &name, &desc, &sources, &attrs); err != nil { return err } fmt.Printf("\n---------------- GTIN %s category=%q\n", gtin, category) fmt.Printf("FEED name : %s\n", trunc(feedName, 140)) fmt.Printf("FEED desc : %s\n", trunc(plain(feedDesc), 200)) fmt.Printf("NEW name : %s\n", trunc(name, 140)) fmt.Printf("NEW desc : %s\n", trunc(desc, 400)) fmt.Printf("attrs : %s\n", trunc(attrs, 200)) fmt.Printf("sources : %s\n", trunc(sources, 240)) var problems []string if strings.TrimSpace(name) == "" || strings.TrimSpace(desc) == "" { problems = append(problems, "empty processed output") } if strings.EqualFold(strings.TrimSpace(name), strings.TrimSpace(feedName)) { problems = append(problems, "NAME copied from feed") } if strings.EqualFold(strings.TrimSpace(plain(desc)), strings.TrimSpace(plain(feedDesc))) { problems = append(problems, "DESCRIPTION copied from feed") } if !strings.Contains(strings.ToLower(desc), " 0 { // Returned, never os.Exit: the throwaway-tenant cleanup is a defer. return fmt.Errorf("%d product(s) did not generate from their category formula", fails) } fmt.Printf("\nall products generated from their category formula\n") return nil } // runNewTenant proves the platform default: a brand-new company and category get // English formulas at create time and enhance into that shape. func runNewTenant(ctx context.Context, pool *pgxpool.Pool, p *processing.Pipeline, keep bool) error { suffix := uuid.NewString()[:8] companyName := "formula-e2e-" + suffix var companyID uuid.UUID if err := pool.QueryRow(ctx, ` INSERT INTO companies (name, language, content_languages) VALUES ($1, 'en', ARRAY['en']) RETURNING id`, companyName).Scan(&companyID); err != nil { return err } if !keep { defer func() { if _, err := pool.Exec(context.Background(), `DELETE FROM companies WHERE id = $1`, companyID); err != nil { log.Printf("cleanup: %v", err) } else { fmt.Printf("\ncleaned up throwaway tenant %s\n", companyName) } }() } var userID uuid.UUID if err := pool.QueryRow(ctx, `SELECT id FROM users ORDER BY created_at LIMIT 1`).Scan(&userID); err != nil { return fmt.Errorf("need at least one user: %w", err) } if _, err := pool.Exec(ctx, ` INSERT INTO memberships (user_id, company_id, role) VALUES ($1, $2, 'admin') ON CONFLICT DO NOTHING`, userID, companyID); err != nil { return err } if err := grantAIPlan(ctx, pool, companyID); err != nil { return err } // Category created through the real service — that is where defaults are applied. svc := &catalog.Service{Pool: pool} cat, err := svc.CreateCategory(ctx, companyID, "High chairs", "high-chairs", nil, nil) if err != nil { return fmt.Errorf("create category: %w", err) } fmt.Printf("\n=== new tenant %s (%s)\n", companyName, companyID) printCategoryFormulas(cat) mapped := map[string]any{ "name": "BRUNNER folding camping chair ONE SHOT grey black 0404164N.C20", "description": "An elegant and very light chair designed for art directors. Folding aluminium frame, comfortable seat and armrests.", "brand": "BRUNNER", "gtin": "8022068075495", "category": "high-chairs", "specifications": map[string]any{ "Colour": "grey black", "Material": "aluminium", }, } mappedJSON, _ := json.Marshal(mapped) var rawID uuid.UUID if err := pool.QueryRow(ctx, ` INSERT INTO raw_products (company_id, gtin, mapped_data, raw_data) VALUES ($1, $2, $3::jsonb, '{}'::jsonb) RETURNING id`, companyID, "8022068075495", string(mappedJSON)).Scan(&rawID); err != nil { return err } return processAndReport(ctx, pool, p, companyID, []uuid.UUID{rawID}, "full") } func grantAIPlan(ctx context.Context, pool *pgxpool.Pool, companyID uuid.UUID) error { // Reuse whichever plan the A1 tenant is on so entitlements (can_use_ai) match a // real paid tenant rather than a hand-built feature map. var planID int64 err := pool.QueryRow(ctx, ` SELECT cp.plan_id FROM company_plans cp JOIN companies c ON c.id = cp.company_id WHERE c.name = 'A1 Slovenija' LIMIT 1`).Scan(&planID) if err != nil { log.Printf("no reference plan found (%v) — enhance may be gated", err) return nil } if _, err := pool.Exec(ctx, ` INSERT INTO company_plans (company_id, plan_id, is_active, billing_cycle_start, next_billing_date, total_credits_allocated) VALUES ($1, $2, true, now(), now() + interval '30 days', 100000)`, companyID, planID); err != nil { return err } _, err = pool.Exec(ctx, ` INSERT INTO credit_balances (company_id, total_credits, used_credits) VALUES ($1, 100000, 0) ON CONFLICT (company_id) DO UPDATE SET total_credits = 100000, used_credits = 0`, companyID) return err } func printCategoryFormulas(cat map[string]any) { for _, k := range []string{"prompts", "title_template", "description_template"} { v, ok := cat[k] if !ok || v == nil { fmt.Printf("%-22s: (none)\n", k) continue } b, _ := json.Marshal(v) fmt.Printf("%-22s: %s\n", k, trunc(string(b), 420)) } } func collectIDs(rows interface { Next() bool Scan(...any) error Err() error Close() }) ([]uuid.UUID, error) { defer rows.Close() var out []uuid.UUID for rows.Next() { var id uuid.UUID if err := rows.Scan(&id); err != nil { return nil, err } out = append(out, id) } return out, rows.Err() } func plain(s string) string { var b strings.Builder inTag := false for _, r := range s { switch { case r == '<': inTag = true case r == '>': inTag = false case !inTag: b.WriteRune(r) } } return strings.Join(strings.Fields(b.String()), " ") } func trunc(s string, n int) string { s = strings.ReplaceAll(strings.TrimSpace(s), "\n", " ") rs := []rune(s) if len(rs) <= n { return s } return string(rs[:n]) + "…" }