package httpapi import ( "context" "encoding/json" "errors" "fmt" "log" "net/http" "strings" "github.com/descrybe/descrybe-v2/apps/api/internal/billing" "github.com/descrybe/descrybe-v2/apps/api/internal/catalog" "github.com/descrybe/descrybe-v2/apps/api/internal/processing" "github.com/go-chi/chi/v5" "github.com/google/uuid" ) // v1StartProcessRequest accepts legacy items[] and v2-native raw_product_ids. // processing_type may be a string or array (decoded via json.RawMessage). type v1StartProcessRequest struct { RawProductIDs []string `json:"raw_product_ids"` ProcessingType json.RawMessage `json:"processing_type"` ProcessingTypes []string `json:"processing_types"` Items []catalog.V1ProcessItem `json:"items"` } func v1ErrFromProcessing(w http.ResponseWriter, err error) { if errors.Is(err, billing.ErrInsufficientCredits) || errors.Is(err, billing.ErrProductLimitExceeded) || errors.Is(err, billing.ErrAIRequiresUpgrade) || errors.Is(err, billing.ErrEPRELRequiresUpgrade) || errors.Is(err, billing.ErrFeatureDisabled) { code := planGateCode(err) msg := err.Error() if errors.Is(err, billing.ErrFeatureDisabled) { msg = "feature_disabled" } v1Err(w, http.StatusPaymentRequired, code, msg) return } if errors.Is(err, processing.ErrRateLimited) { v1Err(w, http.StatusTooManyRequests, "rate_limited", err.Error()) return } if msg, ok := processing.ClientError(err); ok { v1Err(w, http.StatusBadRequest, "validation_error", msg) return } // Log only — do not call LogAndError (writes FlatAPIError) then v1Err (would double-write). if err != nil { log.Printf("httpapi: could not start processing job: %s", redactForLog(err.Error())) } v1Err(w, http.StatusBadRequest, "validation_error", "could not start processing job") } // handleV1StartProcess implements legacy-compatible POST /api/v1/products/process. // POST /api/v1/process stays on handleStartProcessingJob (flat 202 ProcessingJob). func (s *Server) handleV1StartProcess(w http.ResponseWriter, r *http.Request) { cid, ok := CompanyIDFromContext(r.Context()) if !ok || cid == uuid.Nil { v1Err(w, http.StatusUnauthorized, "unauthorized", "Unauthorized") return } uid, _ := UserIDFromContext(r.Context()) var body v1StartProcessRequest if err := DecodeJSONAllowUnknown(r, &body); err != nil { v1Err(w, http.StatusBadRequest, "validation_error", "Request body is required") return } storageType, _, typeErr := parseV1ProcessingTypeRaw(body.ProcessingType) if typeErr != nil { v1Err(w, http.StatusBadRequest, "validation_error", typeErr.Error()) return } // SPA may send processing_types without processing_type; prefer explicit type when set. if len(body.ProcessingType) == 0 && len(body.ProcessingTypes) > 0 { parsed, _, err := processing.ParseV1ProcessingType(body.ProcessingTypes[0]) if err != nil { v1Err(w, http.StatusBadRequest, "validation_error", err.Error()) return } storageType = parsed } var rawIDs []uuid.UUID var itemErrs []string totalItems := 0 switch { case len(body.Items) > 0: totalItems = len(body.Items) if totalItems > processing.StartProductCap() { v1Err(w, http.StatusBadRequest, "validation_error", fmt.Sprintf("too many products (max %d)", processing.StartProductCap())) return } for _, it := range body.Items { if strings.TrimSpace(it.EAN) == "" { v1Err(w, http.StatusBadRequest, "validation_error", "All items must have a valid 'ean' field") return } } // Gate credits/features before EnsureRaw so insufficient-credit clients cannot spam catalog writes. if err := s.assertV1ProcessGates(r.Context(), cid, storageType, totalItems); err != nil { v1ErrFromProcessing(w, err) return } ids, _, errs, err := s.ensureRawV1Items(r.Context(), cid, body.Items) if err != nil { v1Err(w, http.StatusInternalServerError, "internal_server_error", "Internal server error") return } itemErrs = errs rawIDs = ids if len(rawIDs) == 0 { msg := "Failed to process any items" if len(errs) > 0 { msg = fmt.Sprintf("Failed to process any items. Errors: %s", strings.Join(errs, "; ")) } v1Err(w, http.StatusBadRequest, "validation_error", msg) return } case len(body.RawProductIDs) > 0: totalItems = len(body.RawProductIDs) if totalItems > processing.StartProductCap() { v1Err(w, http.StatusBadRequest, "validation_error", fmt.Sprintf("too many products (max %d)", processing.StartProductCap())) return } ids := make([]uuid.UUID, 0, len(body.RawProductIDs)) for _, sID := range body.RawProductIDs { id, err := uuid.Parse(sID) if err != nil { v1Err(w, http.StatusBadRequest, "validation_error", "invalid raw_product_id") return } ids = append(ids, id) } rawIDs = ids default: v1Err(w, http.StatusBadRequest, "validation_error", "'items' array is required") return } jobs, err := s.startV1Jobs(r.Context(), cid, uid, rawIDs, storageType) if err != nil { v1ErrFromProcessing(w, err) return } for _, job := range jobs { if err := s.enqueueV1Job(r.Context(), job.ID); err != nil { v1Err(w, http.StatusInternalServerError, "internal_server_error", "enqueue failed") return } } if len(jobs) == 0 { v1Err(w, http.StatusBadRequest, "validation_error", "could not start processing job") return } primary := jobs[0] // processed_items is the completed count — stay 0 until the worker finishes. // total_items is the enqueue size; status stays pending/processing until poll COMPLETED. resp := map[string]any{ "process_id": primary.ID.String(), "status": "pending", "message": fmt.Sprintf("Processing started for %d product(s)", len(rawIDs)), "total_items": totalItems, "processed_items": 0, } if len(jobs) > 1 { siblings := make([]string, 0, len(jobs)-1) for i := 1; i < len(jobs); i++ { siblings = append(siblings, jobs[i].ID.String()) } resp["job_count"] = len(jobs) resp["sibling_job_ids"] = siblings resp["total_products_queued"] = len(rawIDs) } if len(itemErrs) > 0 { resp["errors"] = itemErrs } v1OK(w, http.StatusOK, resp, nil) } func parseV1ProcessingTypeRaw(raw json.RawMessage) (storage string, response any, err error) { if len(raw) == 0 || string(raw) == "null" { return processing.ParseV1ProcessingType(nil) } var asString string if err := json.Unmarshal(raw, &asString); err == nil { return processing.ParseV1ProcessingType(asString) } var asArr []any if err := json.Unmarshal(raw, &asArr); err == nil { return processing.ParseV1ProcessingType(asArr) } return "", nil, fmt.Errorf("Invalid processing_type. Use \"full\", a single step (category, title, description, attributes), or an array of steps, e.g. [\"title\",\"attributes\"].") } // handleV1GetProcess implements legacy-compatible GET /api/v1/products/process/{id}. func (s *Server) handleV1GetProcess(w http.ResponseWriter, r *http.Request) { cid, ok := CompanyIDFromContext(r.Context()) if !ok || cid == uuid.Nil { v1Err(w, http.StatusUnauthorized, "unauthorized", "Unauthorized") return } id, err := uuid.Parse(chi.URLParam(r, "id")) if err != nil { v1Err(w, http.StatusBadRequest, "validation_error", "invalid id") return } job, err := s.getV1ProcessJob(r.Context(), cid, id) if err != nil { v1Err(w, http.StatusNotFound, "not_found", "Processing job not found") return } status := processing.MapJobStatusForV1(job.Status) ptype := processing.ProcessingTypeForAPIResponse(job.ProcessingType) switch status { case "COMPLETED": items, loadErr := s.loadV1ProcessJobItems(r.Context(), cid, id, job.ProcessingType) if loadErr != nil { v1Err(w, http.StatusInternalServerError, "internal_server_error", "Internal server error") return } items = processing.ApplyV1SEOMetaPolicy(cid, items) data := map[string]any{ "status": status, "process_id": job.ID.String(), "processing_type": ptype, "items": items, "total_items": len(items), } if job.CompletedAt != nil { data["processed_at"] = job.CompletedAt.UTC().Format("2006-01-02T15:04:05.000Z") } else { data["processed_at"] = job.CreatedAt.UTC().Format("2006-01-02T15:04:05.000Z") } if len(items) == 0 { data["message"] = "Processing completed but no products found" } v1OK(w, http.StatusOK, data, nil) case "FAILED": errMsg := "Processing failed" if job.Error != nil && strings.TrimSpace(*job.Error) != "" { log.Printf("httpapi: v1 process job %s failed: %s", job.ID, redactForLog(*job.Error)) errMsg = publicV1JobError(*job.Error) } v1OK(w, http.StatusOK, map[string]any{ "status": status, "process_id": job.ID.String(), "processing_type": ptype, "error": errMsg, }, nil) default: data := map[string]any{ "status": status, "process_id": job.ID.String(), "processing_type": ptype, "created_at": job.CreatedAt.UTC().Format("2006-01-02T15:04:05.000Z"), "started_at": nil, } if job.StartedAt != nil { data["started_at"] = job.StartedAt.UTC().Format("2006-01-02T15:04:05.000Z") } v1OK(w, http.StatusOK, data, nil) } } func publicV1JobError(raw string) string { raw = strings.TrimSpace(raw) if raw == "" { return "Processing failed" } out := processing.PublicV1Error(raw) if strings.TrimSpace(out) == "" { return "Processing failed" } return out } func presentV1Job(job processing.Job) map[string]any { out := map[string]any{ "id": job.ID, "status": job.Status, "total_products": job.TotalProducts, "processed_products": job.ProcessedProducts, "processing_type": processing.ProcessingTypeForAPIResponse(job.ProcessingType), "created_at": formatV1Timestamp(job.CreatedAt), } if job.StartedAt != nil { out["started_at"] = formatV1Timestamp(*job.StartedAt) } if job.CompletedAt != nil { out["completed_at"] = formatV1Timestamp(*job.CompletedAt) } if job.Error != nil && strings.TrimSpace(*job.Error) != "" { out["error"] = publicV1JobError(*job.Error) } return out } func presentV1JobList(jobs []processing.Job) []any { if len(jobs) == 0 { return []any{} } formatted := processing.FormatListJobsResponse(jobs) out := make([]any, len(formatted)) for i, row := range formatted { base := presentV1Job(jobs[i]) raw, err := json.Marshal(row) if err != nil { out[i] = base continue } var extra map[string]any if err := json.Unmarshal(raw, &extra); err != nil { out[i] = base continue } for _, k := range []string{"sibling_job_ids", "job_count", "total_products_queued"} { if v, ok := extra[k]; ok { base[k] = v } } out[i] = base } return out } // handleV1GetProcessJob serves GET /api/v1/process/{id} with a public job DTO // (no current_step, step notes, or company_id). Dashboard GET stays on handleGetProcessingJob. func (s *Server) handleV1GetProcessJob(w http.ResponseWriter, r *http.Request) { cid, ok := CompanyIDFromContext(r.Context()) if !ok || cid == uuid.Nil { Error(w, http.StatusUnauthorized, "unauthorized") return } id, err := uuid.Parse(chi.URLParam(r, "id")) if err != nil { Error(w, http.StatusBadRequest, "invalid id") return } job, err := s.getV1ProcessJob(r.Context(), cid, id) if err != nil { Error(w, http.StatusNotFound, "not found") return } out := presentV1Job(job) if processing.JobStatusIncludesProducts(job.Status) { items, loadErr := s.loadV1ProcessJobItems(r.Context(), cid, id, job.ProcessingType) if loadErr != nil { Error(w, http.StatusInternalServerError, "load failed") return } items = processing.ApplyV1SEOMetaPolicy(cid, items) out["items"] = items out["total_items"] = len(items) } JSON(w, http.StatusOK, out) } func (s *Server) ensureRawV1Items(ctx context.Context, companyID uuid.UUID, items []catalog.V1ProcessItem) ([]uuid.UUID, []catalog.EnsureRawResult, []string, error) { if s != nil && s.testEnsureRawV1Items != nil { return s.testEnsureRawV1Items(ctx, companyID, items) } return s.Catalog.EnsureRawProductsFromV1Items(ctx, companyID, items) } func (s *Server) startV1Jobs(ctx context.Context, companyID, userID uuid.UUID, rawIDs []uuid.UUID, processingType string) ([]processing.Job, error) { if s != nil && s.testStartJobs != nil { return s.testStartJobs(ctx, companyID, userID, rawIDs, processingType) } return s.Processing.StartJob(ctx, companyID, userID, rawIDs, processingType) } func (s *Server) enqueueV1Job(ctx context.Context, jobID uuid.UUID) error { if s != nil && s.testEnqueueJob != nil { return s.testEnqueueJob(ctx, jobID) } return s.Jobs.EnqueueProcessingJob(ctx, jobID) } func (s *Server) getV1ProcessJob(ctx context.Context, companyID, id uuid.UUID) (processing.Job, error) { if s != nil && s.testGetJob != nil { return s.testGetJob(ctx, companyID, id) } return s.Processing.GetJob(ctx, companyID, id) } func (s *Server) loadV1ProcessJobItems(ctx context.Context, companyID, jobID uuid.UUID, processingType string) ([]processing.V1ProcessJobItem, error) { if s != nil && s.testLoadV1ProcessJobItems != nil { return s.testLoadV1ProcessJobItems(ctx, companyID, jobID, processingType) } return s.Processing.LoadV1ProcessJobItems(ctx, companyID, jobID, processingType) } // assertV1ProcessGates runs credit/feature checks before EnsureRaw catalog writes. func (s *Server) assertV1ProcessGates(ctx context.Context, companyID uuid.UUID, processingType string, batchSize int) error { if s == nil || s.Billing == nil { return nil } if err := s.Billing.AssertProcessingFeatures(ctx, companyID, processingType); err != nil { return err } opts := billing.ProcessingGateOpts{ RequiresAI: billing.ProcessingTypeRequiresAI(processingType) || billing.ProcessingTypeIsEmailCampaignAI(processingType), RequiresEPREL: billing.ProcessingTypeRequiresEPREL(processingType), } return s.Billing.AssertCanStartProcessing(ctx, companyID, batchSize, opts) }