descrybe/apps/api/cmd/formula-e2e/main.go

// 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(&gtin, &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), "<h") && !strings.Contains(strings.ToLower(desc), "<p") {
			problems = append(problems, "description is not formula HTML")
		}
		if len(problems) == 0 {
			fmt.Printf("VERDICT    : OK — generated from formula\n")
		} else {
			fails++
			fmt.Printf("VERDICT    : FAIL — %s\n", strings.Join(problems, "; "))
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if fails > 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]) + "…"
}