descrybe/apps/api/internal/httpapi/v1_process_handlers.go

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
		}
		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 && *job.Error != "" {
			errMsg = *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 (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)
}