fix
This commit is contained in:
@@ -0,0 +1,409 @@
|
||||
// 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/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 := 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: processing.NewOpenAIClient(*llmKey, *llmBase, *llmModel, 0, 2),
|
||||
}
|
||||
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 {
|
||||
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
|
||||
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
|
||||
}
|
||||
|
||||
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".
|
||||
if _, err := pool.Exec(ctx, `
|
||||
UPDATE processed_products
|
||||
SET field_sources = field_sources - 'enhance_input_hash'
|
||||
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), "<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]) + "…"
|
||||
}
|
||||
Reference in New Issue
Block a user