package processing import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "log" "math" "math/rand" "net" "net/http" "net/url" "strings" "sync" "time" "github.com/descrybe/descrybe-v2/apps/api/internal/company" "github.com/descrybe/descrybe-v2/apps/api/internal/config" "github.com/descrybe/descrybe-v2/apps/api/internal/security" ) // OpenAIClient calls an OpenAI-compatible Chat Completions API with rate limiting and retries. type OpenAIClient struct { APIKey string BaseURL string Model string HTTPClient *http.Client MinInterval time.Duration MaxRetries int // ModeLabel is recorded on products/jobs for analytics // ("internal" | "popular:" | "custom"). Defaults to internal. ModeLabel string mu sync.Mutex lastCall time.Time } const maxOpenAIRetries = 8 // openAIHTTPTimeout is the full-request limit for OpenAI-compatible chat/embeddings. // Reasoning models (e.g. code-fast) often need 60–180s; 240s reduces false timeouts. // Timeouts are not retried (see classifyOpenAITransportErr) so a hung upstream fails // the item in ~4m instead of MaxRetries×240s of silent PROCESSING. const openAIHTTPTimeout = 240 * time.Second // openAIWaitLogInterval emits progress while HTTPClient.Do blocks (cancel is only // checked between job items, so mid-call silence looks like a stuck job). const openAIWaitLogInterval = 30 * time.Second func NewOpenAIClient(apiKey, baseURL, model string, rpm, maxRetries int) *OpenAIClient { if baseURL == "" { baseURL = "https://api.openai.com/v1" } if model == "" { model = "gpt-4o-mini" } if maxRetries <= 0 { maxRetries = 3 } else if maxRetries > maxOpenAIRetries { maxRetries = maxOpenAIRetries } interval := time.Duration(0) if rpm > 0 { interval = time.Minute / time.Duration(rpm) } // Dial-time SSRF. Loopback when base URL is local; RFC1918 only in non-prod. policy := openAIDialPolicy(baseURL) // Prefer synthesize fallback on timeout/error/empty (RunSteps) over raising this unboundedly. return &OpenAIClient{ APIKey: apiKey, BaseURL: strings.TrimRight(baseURL, "/"), Model: model, HTTPClient: security.SafeHTTPClientPolicy(openAIHTTPTimeout, policy), MinInterval: interval, MaxRetries: maxRetries, ModeLabel: AIProviderInternal, } } func openAIDialPolicy(baseURL string) security.DialPolicy { if openAIBaseAllowsLoopback(baseURL) { return security.DialPolicy{AllowLoopback: true} } if openAIBaseAllowsPrivate(baseURL) { return security.DialPolicy{AllowLoopback: true, AllowPrivate: true} } return security.DialPolicy{} } func openAIBaseAllowsLoopback(baseURL string) bool { u, err := url.Parse(baseURL) if err != nil || u.Hostname() == "" { return false } host := strings.ToLower(u.Hostname()) if host == "localhost" { return true } ip := net.ParseIP(host) return ip != nil && ip.IsLoopback() } // openAIBaseAllowsPrivate permits RFC1918/ULA literal OPENAI_BASE_URL hosts // when APP_ENV is not production/prod (local LAN OpenAI-compatible proxies). func openAIBaseAllowsPrivate(baseURL string) bool { if config.IsProductionEnv() { return false } u, err := url.Parse(baseURL) if err != nil || u.Hostname() == "" { return false } ip := net.ParseIP(strings.ToLower(u.Hostname())) if ip == nil || ip.IsLinkLocalUnicast() { return false } return ip.IsPrivate() } func (c *OpenAIClient) Enabled() bool { return c != nil && strings.TrimSpace(c.APIKey) != "" } // ProviderModeLabel implements ProviderLabeler for analytics writes. // Loopback / mock-llm base URLs must not report as managed "internal". func (c *OpenAIClient) ProviderModeLabel() string { if c == nil { return AIProviderUnknown } label := strings.TrimSpace(c.ModeLabel) if IsMockOrLoopbackBaseURL(c.BaseURL) && (label == "" || label == AIProviderInternal) { return AIProviderCustom } if label != "" { return label } return AIProviderInternal } // IsMockOrLoopbackBaseURL reports local mock / loopback OpenAI-compatible bases // (e.g. mock-llm on 127.0.0.1:18767). Used for analytics ModeLabel accuracy. func IsMockOrLoopbackBaseURL(baseURL string) bool { if openAIBaseAllowsLoopback(baseURL) { return true } lower := strings.ToLower(strings.TrimSpace(baseURL)) return strings.Contains(lower, "mock-llm") } // startWaitLogger logs request start and periodic progress while Do blocks. // Call the returned stop with the Do error (or nil) when the request finishes. func (c *OpenAIClient) startWaitLogger(op string, maxTokens int) func(error) { if c == nil { return func(error) {} } start := time.Now() log.Printf("openai: request start op=%s model=%s base=%s max_tokens=%d timeout=%s", op, c.Model, c.BaseURL, maxTokens, openAIHTTPTimeout) done := make(chan struct{}) var once sync.Once go func() { t := time.NewTicker(openAIWaitLogInterval) defer t.Stop() for { select { case <-done: return case <-t.C: log.Printf("openai: still waiting op=%s model=%s base=%s elapsed=%s timeout=%s", op, c.Model, c.BaseURL, time.Since(start).Round(time.Second), openAIHTTPTimeout) } } }() return func(err error) { once.Do(func() { close(done) elapsed := time.Since(start).Round(time.Millisecond) if err != nil { log.Printf("openai: request done op=%s model=%s elapsed=%s err=%s", op, c.Model, elapsed, TruncateError(err)) return } log.Printf("openai: request done op=%s model=%s elapsed=%s ok=1", op, c.Model, elapsed) }) } } // isOpenAITimeoutErr reports HTTP client / context deadline timeouts. // These must not be retried: each attempt costs openAIHTTPTimeout and leaves the job looking stuck. func isOpenAITimeoutErr(err error) bool { if err == nil { return false } if errors.Is(err, context.DeadlineExceeded) { return true } var ne net.Error if errors.As(err, &ne) && ne.Timeout() { return true } s := err.Error() return strings.Contains(s, "Client.Timeout exceeded") || strings.Contains(s, "context deadline exceeded") } // classifyOpenAITransportErr maps Do/Read errors to (wrappedErr, retryable). // Timeouts and cancel are fatal for this attempt chain; other network errors may retry. func classifyOpenAITransportErr(model string, err error) (error, bool) { if err == nil { return nil, false } if errors.Is(err, context.Canceled) { return err, false } if isOpenAITimeoutErr(err) { return fmt.Errorf( "openai timed out waiting for model %q after %s (not retrying; use a faster model or fix upstream latency): %w", model, openAIHTTPTimeout, err), false } return err, true } type chatRequest struct { Model string `json:"model"` Messages []chatMessage `json:"messages"` Temperature float64 `json:"temperature"` MaxTokens int `json:"max_tokens,omitempty"` } type chatMessage struct { Role string `json:"role"` Content string `json:"content"` } type chatResponse struct { Model string `json:"model"` Choices []struct { FinishReason string `json:"finish_reason"` Message struct { Content json.RawMessage `json:"content"` ReasoningContent string `json:"reasoning_content"` Reasoning string `json:"reasoning"` } `json:"message"` } `json:"choices"` Usage struct { PromptTokens int `json:"prompt_tokens"` CompletionTokens int `json:"completion_tokens"` TotalTokens int `json:"total_tokens"` } `json:"usage"` Error *struct { Message string `json:"message"` Type string `json:"type"` } `json:"error"` } var ( errEmptyModelResponse = errors.New("empty model response") errLengthCappedResponse = errors.New("response truncated at max_tokens") ) func isLengthBudgetErr(err error) bool { return errors.Is(err, errEmptyModelResponse) || errors.Is(err, errLengthCappedResponse) } func (c *OpenAIClient) Complete(ctx context.Context, system, user string) (Completion, error) { return c.CompleteWithOptions(ctx, system, user, CompleteOptions{}) } func (c *OpenAIClient) CompleteWithOptions(ctx context.Context, system, user string, opts CompleteOptions) (Completion, error) { if !c.Enabled() { return Completion{}, errors.New("openai api key not configured") } system = SanitizeText(system) user = SanitizeText(user) temp := opts.Temperature if temp <= 0 { temp = DefaultStructuredTemp } if temp > 0.3 { temp = 0.3 } maxTok := opts.MaxTokens var lastErr error for attempt := 0; attempt <= c.MaxRetries; attempt++ { if attempt > 0 { backoff := time.Duration(math.Pow(2, float64(attempt-1))) * 200 * time.Millisecond jitter := time.Duration(rand.Intn(100)) * time.Millisecond select { case <-ctx.Done(): return Completion{}, ctx.Err() case <-time.After(backoff + jitter): } } if err := c.waitRate(ctx); err != nil { return Completion{}, err } comp, retryable, err := c.doComplete(ctx, system, user, temp, maxTok) if err == nil { return comp, nil } lastErr = err if isLengthBudgetErr(err) { if maxTok <= 0 || maxTok < MaxTokensEnhanceRetry { prev := maxTok if maxTok < MaxTokensEnhanceRetry { maxTok = MaxTokensEnhanceRetry } log.Printf("openai: length-cap retry model=%s prev_max_tokens=%d next_max_tokens=%d err=%s", c.Model, prev, maxTok, TruncateError(err)) retryable = true } else { // Already at enhance retry ceiling — another long call will not help. return Completion{}, fmt.Errorf( "openai length-capped at max_tokens=%d for model %q (prefer a faster non-reasoning product-enhance model): %w", maxTok, c.Model, err) } } if !retryable { return Completion{}, err } } return Completion{}, fmt.Errorf("openai retries exhausted: %w", lastErr) } func (c *OpenAIClient) waitRate(ctx context.Context) error { c.mu.Lock() if c.MinInterval <= 0 { c.lastCall = time.Now() c.mu.Unlock() return nil } now := time.Now() wait := c.MinInterval - now.Sub(c.lastCall) if wait < 0 { wait = 0 } // Reserve the next slot under the lock so concurrent callers cannot both // observe the same lastCall and bypass MinInterval. c.lastCall = now.Add(wait) c.mu.Unlock() if wait > 0 { select { case <-ctx.Done(): return ctx.Err() case <-time.After(wait): } } return nil } type embeddingRequest struct { Model string `json:"model"` Input []string `json:"input"` } type embeddingResponse struct { Data []struct { Embedding []float32 `json:"embedding"` Index int `json:"index"` } `json:"data"` Error *struct { Message string `json:"message"` } `json:"error"` } // Embed implements Embedder via OpenAI-compatible POST /embeddings. func (c *OpenAIClient) Embed(ctx context.Context, texts []string) ([][]float32, error) { if !c.Enabled() { return nil, errors.New("openai api key not configured") } if len(texts) == 0 { return nil, errors.New("empty embedding input") } clean := make([]string, 0, len(texts)) for _, t := range texts { t = SanitizeText(t) if t == "" { return nil, errors.New("empty embedding input") } clean = append(clean, t) } var lastErr error for attempt := 0; attempt <= c.MaxRetries; attempt++ { if attempt > 0 { backoff := time.Duration(math.Pow(2, float64(attempt-1))) * 200 * time.Millisecond jitter := time.Duration(rand.Intn(100)) * time.Millisecond select { case <-ctx.Done(): return nil, ctx.Err() case <-time.After(backoff + jitter): } } if err := c.waitRate(ctx); err != nil { return nil, err } vecs, retryable, err := c.doEmbed(ctx, clean) if err == nil { return vecs, nil } lastErr = err if !retryable { return nil, err } } return nil, fmt.Errorf("openai embedding retries exhausted: %w", lastErr) } func (c *OpenAIClient) doEmbed(ctx context.Context, texts []string) ([][]float32, bool, error) { body, err := json.Marshal(embeddingRequest{Model: c.Model, Input: texts}) if err != nil { return nil, false, err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+"/embeddings", bytes.NewReader(body)) if err != nil { return nil, false, err } req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+c.APIKey) stopWait := c.startWaitLogger("embeddings", 0) res, err := c.HTTPClient.Do(req) stopWait(err) if err != nil { wrapped, retryable := classifyOpenAITransportErr(c.Model, err) return nil, retryable, wrapped } defer res.Body.Close() raw, err := io.ReadAll(io.LimitReader(res.Body, 4<<20)) if err != nil { wrapped, retryable := classifyOpenAITransportErr(c.Model, err) return nil, retryable, wrapped } var parsed embeddingResponse if err := json.Unmarshal(raw, &parsed); err != nil { return nil, false, fmt.Errorf("openai embeddings decode: %w", err) } if res.StatusCode == http.StatusTooManyRequests || res.StatusCode >= 500 { msg := "rate limited or server error" if parsed.Error != nil && parsed.Error.Message != "" { msg = TruncateError(errors.New(parsed.Error.Message)) } return nil, true, errors.New(msg) } if res.StatusCode >= 400 { msg := fmt.Sprintf("openai embeddings http %d", res.StatusCode) if parsed.Error != nil && parsed.Error.Message != "" { msg = TruncateError(errors.New(parsed.Error.Message)) } return nil, false, errors.New(msg) } if len(parsed.Data) == 0 { return nil, false, errors.New("empty embedding response") } out := make([][]float32, len(texts)) for _, row := range parsed.Data { if row.Index < 0 || row.Index >= len(out) { return nil, false, errors.New("embedding index out of range") } out[row.Index] = row.Embedding } for i, v := range out { if len(v) == 0 { return nil, false, fmt.Errorf("missing embedding at index %d", i) } } return out, false, nil } func (c *OpenAIClient) doComplete(ctx context.Context, system, user string, temperature float64, maxTokens int) (Completion, bool, error) { reqBody := chatRequest{ Model: c.Model, Messages: []chatMessage{ {Role: "system", Content: system}, {Role: "user", Content: user}, }, Temperature: temperature, MaxTokens: maxTokens, } body, err := json.Marshal(reqBody) if err != nil { return Completion{}, false, err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+"/chat/completions", bytes.NewReader(body)) if err != nil { return Completion{}, false, err } req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+c.APIKey) stopWait := c.startWaitLogger("chat/completions", maxTokens) res, err := c.HTTPClient.Do(req) if err != nil { stopWait(err) wrapped, retryable := classifyOpenAITransportErr(c.Model, err) return Completion{}, retryable, wrapped } defer res.Body.Close() raw, err := io.ReadAll(io.LimitReader(res.Body, 1<<20)) if err != nil { stopWait(err) wrapped, retryable := classifyOpenAITransportErr(c.Model, err) return Completion{}, retryable, wrapped } var parsed chatResponse if err := json.Unmarshal(raw, &parsed); err != nil { decErr := fmt.Errorf("openai decode: %w", err) stopWait(decErr) return Completion{}, false, decErr } if res.StatusCode == http.StatusTooManyRequests || res.StatusCode >= 500 { msg := "rate limited or server error" if parsed.Error != nil && parsed.Error.Message != "" { msg = TruncateError(errors.New(parsed.Error.Message)) } httpErr := errors.New(msg) stopWait(httpErr) return Completion{}, true, httpErr } if res.StatusCode >= 400 { msg := fmt.Sprintf("openai http %d", res.StatusCode) if parsed.Error != nil && parsed.Error.Message != "" { msg = TruncateError(errors.New(parsed.Error.Message)) } httpErr := errors.New(msg) stopWait(httpErr) return Completion{}, false, httpErr } text := "" finishReason := "" if len(parsed.Choices) > 0 { finishReason = strings.TrimSpace(parsed.Choices[0].FinishReason) text = SanitizeOutput(choiceMessageText(parsed.Choices[0].Message.Content, parsed.Choices[0].Message.ReasoningContent, parsed.Choices[0].Message.Reasoning)) } usagePrompt := parsed.Usage.PromptTokens usageOut := parsed.Usage.CompletionTokens usageTotal := parsed.Usage.TotalTokens canBump := maxTokens <= 0 || maxTokens < MaxTokensEnhanceRetry lengthCapped := strings.EqualFold(finishReason, "length") if text == "" { // Reasoning models often return empty content when max_tokens cuts mid-thought. // Only retry when CompleteWithOptions can still raise max_tokens. // Never log HTTP-ok as ok=1 when content is empty (prod looked "successful" at 30–272ms). retryable := lengthCapped && canBump emptyErr := fmt.Errorf("%w (finish_reason=%s max_tokens=%d prompt_tokens=%d completion_tokens=%d)", errEmptyModelResponse, finishReason, maxTokens, usagePrompt, usageOut) stopWait(emptyErr) log.Printf("openai: chat content empty model=%s finish_reason=%s max_tokens=%d prompt_tokens=%d completion_tokens=%d total_tokens=%d retryable=%t", c.Model, finishReason, maxTokens, usagePrompt, usageOut, usageTotal, retryable) return Completion{}, retryable, errEmptyModelResponse } if lengthCapped { // Truncated JSON was previously treated as success → parse_failed → synthesize. // Accept only when the truncated body still parses as a JSON object. if _, err := ParseJSONObject(text); err != nil { retryable := canBump truncErr := fmt.Errorf("%w (finish_reason=length max_tokens=%d prompt_tokens=%d completion_tokens=%d content_runes=%d)", errLengthCappedResponse, maxTokens, usagePrompt, usageOut, len([]rune(text))) stopWait(truncErr) log.Printf("openai: chat content truncated model=%s finish_reason=length max_tokens=%d prompt_tokens=%d completion_tokens=%d total_tokens=%d content_runes=%d retryable=%t", c.Model, maxTokens, usagePrompt, usageOut, usageTotal, len([]rune(text)), retryable) return Completion{}, retryable, errLengthCappedResponse } } stopWait(nil) log.Printf("openai: chat content ok model=%s finish_reason=%s content_runes=%d prompt_tokens=%d completion_tokens=%d total_tokens=%d", c.Model, finishReason, len([]rune(text)), usagePrompt, usageOut, usageTotal) return Completion{ Text: text, PromptTokens: usagePrompt, OutputTokens: usageOut, TotalTokens: usageTotal, Model: parsed.Model, Raw: map[string]any{ "model": parsed.Model, "usage": parsed.Usage, "status": res.StatusCode, "finish_reason": finishReason, }, }, false, nil } // choiceMessageText reads assistant text from OpenAI-compatible chat responses. // Prefer message.content (string or multipart text parts). When empty — common for // reasoning models that only fill reasoning_content under a tight max_tokens — // fall back to reasoning fields and prefer an embedded JSON object when present. func choiceMessageText(content json.RawMessage, reasoningContent, reasoning string) string { if text := decodeChatContent(content); text != "" { return text } for _, alt := range []string{reasoningContent, reasoning} { alt = strings.TrimSpace(alt) if alt == "" { continue } if obj := StripJSONFences(alt); strings.HasPrefix(obj, "{") { if _, err := ParseJSONObject(obj); err == nil { return obj } } } return "" } func decodeChatContent(raw json.RawMessage) string { raw = bytes.TrimSpace(raw) if len(raw) == 0 || string(raw) == "null" { return "" } var s string if err := json.Unmarshal(raw, &s); err == nil { return strings.TrimSpace(s) } var parts []struct { Type string `json:"type"` Text string `json:"text"` } if err := json.Unmarshal(raw, &parts); err == nil { var b strings.Builder for _, p := range parts { if strings.TrimSpace(p.Text) != "" { b.WriteString(p.Text) } } return strings.TrimSpace(b.String()) } return "" } // HeuristicCompleter is used when OpenAI is not configured (local/dev fallback). type HeuristicCompleter struct{} // ProviderModeLabel labels heuristic output for analytics (not a paid provider). func (h HeuristicCompleter) ProviderModeLabel() string { return AIProviderInternal } func (h HeuristicCompleter) Complete(_ context.Context, system, user string) (Completion, error) { systemL := strings.ToLower(system) user = SanitizeText(user) text := "General" switch { case strings.Contains(systemL, "categoryid") || (strings.Contains(systemL, "categor") && strings.Contains(systemL, "available categories")): // Taxonomy pick: return first Available Categories (ID: …) from the user prompt. id := firstAvailableCategoryIDFromPrompt(user) if id == "" { id = "unknown" } b, _ := json.Marshal(map[string]any{"categoryId": id, "confidence": 0.5}) text = string(b) case strings.Contains(systemL, `"name"`) || strings.Contains(systemL, "titles and descriptions"): // Prefer explicit Name:/Desc: (ProductEnhanceUser) or Current name: labels. // Never use firstLine(user) alone — that line is often "Category: …". name := labeledPromptValue(user, "name:", "current name:") if name == "" || isPromptLabelTitle(name) { name = "Product" } desc := labeledPromptValue(user, "desc:", "description:", "current description:") if desc == "" || descriptionEchoesTitle(desc, name) || isWeakPriorEnhanceDescription(desc, name) { desc = inventHeuristicDescription(system, user, name) } b, _ := json.Marshal(map[string]string{"name": name, "description": desc}) text = string(b) case strings.Contains(systemL, "attributes") && strings.Contains(systemL, "json"): text = `{"material":"unknown","brand":"unknown"}` default: // Never echo ProductEnhanceUser's first line ("Category: …") as title/output. text = labeledPromptValue(user, "name:", "current name:") if text == "" || isPromptLabelTitle(text) { text = "ok" } } return Completion{ Text: text, TotalTokens: 0, Model: "heuristic", Raw: map[string]any{"provider": "heuristic"}, }, nil } // parseAttrsFromPrompt extracts the Attrs:/Attributes: JSON object from an enhance user message. func parseAttrsFromPrompt(user string) map[string]any { lower := strings.ToLower(user) label := "attrs:" at := strings.Index(lower, label) if at < 0 { label = "attributes:" at = strings.Index(lower, label) if at < 0 { return nil } } rest := strings.TrimSpace(user[at+len(label):]) if i := strings.Index(rest, "{"); i >= 0 { rest = rest[i:] if end := strings.Index(rest, "}"); end >= 0 { rest = rest[:end+1] } } else { rest = firstLine(rest) } var m map[string]any if err := json.Unmarshal([]byte(rest), &m); err != nil || len(m) == 0 { return nil } return m } // inventHeuristicDescription builds a short factual description from Category + Attrs // (+ brand / model / dims) for thin inputs. Matches Slovenian vs English from the // enhance system/user prompt language hint. Never returns sole retail-filler copy // when a title or attributes exist. func inventHeuristicDescription(system, user, name string) string { name = strings.TrimSpace(name) if name == "" || name == "" || isPromptLabelTitle(name) { name = labeledPromptValue(user, "name:", "current name:") } if name == "" || isPromptLabelTitle(name) { name = "Product" } cat := labeledPromptValue(user, "category:") if isUnusableCategoryValue(cat, name) { cat = "" } lang := languageCodeFromEnhancePrompt(system, user) attrs := CompactAttrs(parseAttrsFromPrompt(user), MaxAttrKeys) if out := synthesizeDescriptionFromTitle(name, cat, lang, attrs); out != "" { return out } return SanitizeOutput(fmt.Sprintf("%s is a catalog product with the known attributes.", name)) } func languageCodeFromEnhancePrompt(system, user string) string { blob := strings.ToLower(system + "\n" + user) for _, code := range company.ContentLanguages { label := strings.ToLower(company.LanguageLabel(code)) if label == "" { continue } if strings.Contains(blob, "in "+label) || strings.Contains(blob, "language: "+label) { return code } } if code, err := company.ParseLanguage(labeledPromptValue(user, "language:", "content language:"), false); err == nil { return code } return company.DefaultLanguage }