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

package httpapi

import (
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"

	"github.com/descrybe/descrybe-v2/apps/api/internal/catalog"
	"github.com/descrybe/descrybe-v2/apps/api/internal/feeds"
	"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
	"github.com/go-chi/chi/v5"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

func (s *Server) handleListFeeds(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	limit, offset := ParseLimitOffset(r)
	page, total, activeTotal, mappedTotal, err := s.Feeds.List(r.Context(), cid, limit, offset, QuerySearch(r))
	if err != nil {
		Error(w, http.StatusInternalServerError, "list failed")
		return
	}
	products, err := s.Feeds.CompanyProductTotals(r.Context(), cid)
	if err != nil {
		Error(w, http.StatusInternalServerError, "list failed")
		return
	}
	JSON(w, http.StatusOK, map[string]any{
		"feeds": feeds.PresentFeeds(page), "total": total, "active_total": activeTotal, "mapped_total": mappedTotal,
		"product_total": products.Total, "processed_total": products.Processed, "unprocessed_total": products.Unprocessed,
		"limit": limit, "offset": offset,
	})
}

func (s *Server) handleCreateFeed(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	uid, _ := UserIDFromContext(r.Context())

	ct := strings.ToLower(strings.TrimSpace(r.Header.Get("Content-Type")))
	if strings.HasPrefix(ct, "multipart/form-data") {
		s.createFeedFromMultipart(w, r, cid, uid)
		return
	}

	var body struct {
		Name                string `json:"name"`
		URL                 string `json:"url"`
		ItemPath            string `json:"item_path"`
		FeedType            string `json:"feed_type"`
		SyncIntervalMinutes int    `json:"sync_interval_minutes"`
	}
	if err := DecodeJSON(r, &body); err != nil {
		Error(w, http.StatusBadRequest, "invalid json")
		return
	}
	item, err := s.Feeds.Create(r.Context(), cid, feeds.CreateInput{
		Name:                body.Name,
		URL:                 body.URL,
		ItemPath:            body.ItemPath,
		FeedType:            body.FeedType,
		SyncIntervalMinutes: body.SyncIntervalMinutes,
	})
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not create feed", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusCreated, feeds.PresentFeed(item))
}

func (s *Server) createFeedFromMultipart(w http.ResponseWriter, r *http.Request, cid, uid uuid.UUID) {
	if err := r.ParseMultipartForm(catalogMaxUpload); err != nil {
		Error(w, http.StatusBadRequest, "invalid multipart form")
		return
	}

	name := strings.TrimSpace(r.FormValue("name"))
	url := strings.TrimSpace(r.FormValue("url"))
	feedType := strings.TrimSpace(r.FormValue("feed_type"))
	itemPath := strings.TrimSpace(r.FormValue("item_path"))
	interval, _ := strconv.Atoi(strings.TrimSpace(r.FormValue("sync_interval_minutes")))

	file, header, fileErr := r.FormFile("file")
	var options map[string]any
	if fileErr == nil {
		defer file.Close()
		meta, err := s.Catalog.SaveUpload(
			r.Context(),
			cid,
			uid,
			s.Config.UploadDir,
			header.Filename,
			header.Header.Get("Content-Type"),
			"feed",
			file,
		)
		if err != nil {
			ClientOrLog(w, http.StatusBadRequest, "could not save upload", err, catalog.ClientError)
			return
		}
		pathStr, _ := meta["path"].(string)
		fileID, _ := meta["id"].(string)
		fileName, _ := meta["name"].(string)
		options = map[string]any{
			"source_path":     pathStr,
			"source_file_id":  fileID,
			"source_filename": fileName,
			"source_kind":     "csv",
		}
		if feedType == "" {
			feedType = "csv"
		}
		if fid, err := uuid.Parse(fileID); err == nil {
			_, _ = s.Catalog.UpdateFileStatus(r.Context(), cid, fid, "uploaded", map[string]any{
				"kind": "feed",
				"feed": true,
				"name": name,
			})
		}
	} else if url == "" && itemPath == "" {
		Error(w, http.StatusBadRequest, "url or file field required")
		return
	}

	item, err := s.Feeds.Create(r.Context(), cid, feeds.CreateInput{
		Name:                name,
		URL:                 url,
		ItemPath:            itemPath,
		FeedType:            feedType,
		SyncIntervalMinutes: interval,
		Options:             options,
	})
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not create feed", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusCreated, feeds.PresentFeed(item))
}

func (s *Server) handleGetFeed(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	item, err := s.Feeds.Get(r.Context(), cid, id)
	if err != nil {
		if feeds.IsNotFound(err) {
			Error(w, http.StatusNotFound, "not found")
			return
		}
		Error(w, http.StatusInternalServerError, "get failed")
		return
	}
	JSON(w, http.StatusOK, feeds.PresentFeed(item))
}

func (s *Server) handleUpdateFeed(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	var body map[string]any
	if err := DecodeJSON(r, &body); err != nil {
		Error(w, http.StatusBadRequest, "invalid json")
		return
	}
	item, err := s.Feeds.Update(r.Context(), cid, id, body)
	if err != nil {
		if feeds.IsNotFound(err) {
			Error(w, http.StatusNotFound, "not found")
			return
		}
		ClientOrLog(w, http.StatusBadRequest, "could not update feed", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusOK, feeds.PresentFeed(item))
}

func (s *Server) handleDeleteFeed(w http.ResponseWriter, r *http.Request) {
	if !s.allowCompanyAdminOrPlatform(w, r) {
		return
	}
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	if err := s.Feeds.Delete(r.Context(), cid, id); err != nil {
		if feeds.IsNotFound(err) {
			Error(w, http.StatusNotFound, "not found")
			return
		}
		Error(w, http.StatusInternalServerError, "delete failed")
		return
	}
	JSON(w, http.StatusOK, map[string]any{"id": id.String(), "deleted": true})
}

func (s *Server) handleSyncFeed(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	jobID, err := s.Feeds.EnqueueSync(r.Context(), cid, id)
	if err != nil {
		if feeds.IsNotFound(err) {
			Error(w, http.StatusNotFound, "Feed not found")
			return
		}
		ClientOrLog(w, http.StatusBadRequest, "could not sync feed", err, feeds.ClientError)
		return
	}
	if s.Jobs != nil {
		_ = s.Jobs.EnqueueFeedSyncJob(r.Context(), jobID)
	}
	job, err := s.Feeds.GetSyncJob(r.Context(), cid, id, jobID)
	if err != nil {
		Error(w, http.StatusInternalServerError, "could not load sync job")
		return
	}
	JSON(w, http.StatusAccepted, job)
}

func (s *Server) handleListSyncJobs(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	limit, _ := ParseLimitOffset(r)
	items, err := s.Feeds.ListSyncJobs(r.Context(), cid, id, limit)
	if err != nil {
		Error(w, http.StatusInternalServerError, "list failed")
		return
	}
	JSON(w, http.StatusOK, map[string]any{"jobs": items, "limit": limit})
}

func (s *Server) handleGetSyncJob(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	feedID, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	jobID, err := uuid.Parse(chi.URLParam(r, "jobID"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid job id")
		return
	}
	job, err := s.Feeds.GetSyncJob(r.Context(), cid, feedID, jobID)
	if err != nil {
		Error(w, http.StatusNotFound, "not found")
		return
	}
	JSON(w, http.StatusOK, job)
}

func (s *Server) handleGetFeedMappings(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	item, err := s.Feeds.GetMappings(r.Context(), cid, id)
	if err != nil {
		JSON(w, http.StatusOK, map[string]any{"mappings": []any{}})
		return
	}
	JSON(w, http.StatusOK, item)
}

func (s *Server) handlePutFeedMappings(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	var body struct {
		Mappings any `json:"mappings"`
	}
	if err := DecodeJSON(r, &body); err != nil {
		Error(w, http.StatusBadRequest, "invalid json")
		return
	}
	item, err := s.Feeds.PutMappings(r.Context(), cid, id, body.Mappings)
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not save mappings", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusOK, item)
}

func (s *Server) handleExtractFeedSchema(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	var body struct {
		ItemPath string `json:"item_path"`
	}
	if err := DecodeJSONOptional(r, &body); err != nil {
		Error(w, http.StatusBadRequest, "invalid json")
		return
	}
	result, err := s.Feeds.ExtractSchema(r.Context(), cid, id, body.ItemPath)
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not extract schema", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusOK, result)
}

// handleSyncAndProcessSample syncs a company-scoped feed then starts a processing job
// for up to N of that feed's raw products (default 10, max 100).
func (s *Server) handleSyncAndProcessSample(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	uid, _ := UserIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	var body struct {
		Limit          int    `json:"limit"`
		SkipSync       bool   `json:"skip_sync"`
		ProcessingType string `json:"processing_type"`
	}
	if err := DecodeJSONOptional(r, &body); err != nil {
		Error(w, http.StatusBadRequest, "invalid json")
		return
	}
	limit := body.Limit
	if limit <= 0 {
		limit = 10
	}
	if limit > 100 {
		limit = 100
	}
	processingType := strings.TrimSpace(body.ProcessingType)
	if processingType == "" {
		processingType = "full"
	}

	var syncJob map[string]any
	if !body.SkipSync {
		job, syncErr := s.Feeds.Sync(r.Context(), cid, id)
		if syncErr != nil {
			ClientOrLog(w, http.StatusBadRequest, "could not sync feed", syncErr, feeds.ClientError)
			return
		}
		syncJob = job
	}

	rawIDs, err := s.Catalog.ListRawProductIDsByFeed(r.Context(), cid, id, limit)
	if err != nil {
		Error(w, http.StatusInternalServerError, "list raw products failed")
		return
	}
	if len(rawIDs) == 0 {
		JSON(w, http.StatusOK, map[string]any{
			"sync_job":         syncJob,
			"processing_job":   nil,
			"raw_product_ids":  []string{},
			"sample_requested": limit,
			"sample_queued":    0,
			"message":          "Sync completed but no raw products found for this feed",
		})
		return
	}

	procJobs, err := s.Processing.StartJob(r.Context(), cid, uid, rawIDs, processingType)
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not start processing", err, processing.ClientError)
		return
	}
	for _, procJob := range procJobs {
		if err := s.Jobs.EnqueueProcessingJob(r.Context(), procJob.ID); err != nil {
			Error(w, http.StatusInternalServerError, "enqueue failed")
			return
		}
	}
	var primary any
	if len(procJobs) > 0 {
		primary = processing.FormatStartJobsResponse(procJobs)
	}

	idStrs := make([]string, 0, len(rawIDs))
	for _, rid := range rawIDs {
		idStrs = append(idStrs, rid.String())
	}
	JSON(w, http.StatusAccepted, map[string]any{
		"sync_job":         syncJob,
		"processing_job":   primary,
		"processing_jobs":  procJobs,
		"raw_product_ids":  idStrs,
		"sample_requested": limit,
		"sample_queued":    len(rawIDs),
		"message":          fmt.Sprintf("Queued %d product(s) for processing", len(rawIDs)),
	})
}

func (s *Server) handleListExportFeeds(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	limit, offset := ParseLimitOffset(r)
	page, total, err := s.Feeds.ListExportFeeds(r.Context(), cid, limit, offset)
	if err != nil {
		Error(w, http.StatusInternalServerError, "list failed")
		return
	}
	JSON(w, http.StatusOK, map[string]any{"export_feeds": page, "total": total, "limit": limit, "offset": offset})
}

func (s *Server) handleCreateExportFeed(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	var body struct {
		Name         string  `json:"name"`
		SourceFeedID *string `json:"source_feed_id"`
		Format       string  `json:"format"`
		Template     any     `json:"template"`
		Filters      any     `json:"filters"`
	}
	if err := DecodeJSON(r, &body); err != nil {
		Error(w, http.StatusBadRequest, "invalid json")
		return
	}
	item, err := s.Feeds.CreateExportFeed(r.Context(), cid, feeds.CreateExportInput{
		Name: body.Name, SourceFeedID: body.SourceFeedID, Format: body.Format,
		Template: body.Template, Filters: body.Filters,
	})
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not create export feed", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusCreated, item)
}

func (s *Server) handleGetExportFeed(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	item, err := s.Feeds.GetExportFeed(r.Context(), cid, id)
	if err != nil {
		Error(w, http.StatusNotFound, "not found")
		return
	}
	JSON(w, http.StatusOK, item)
}

func (s *Server) handleUpdateExportFeed(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	var body struct {
		Name     *string `json:"name"`
		IsActive *bool   `json:"is_active"`
		Template any     `json:"template"`
		Filters  any     `json:"filters"`
	}
	if err := DecodeJSON(r, &body); err != nil {
		Error(w, http.StatusBadRequest, "invalid json")
		return
	}
	item, err := s.Feeds.UpdateExportFeed(r.Context(), cid, id, body.Name, body.IsActive, body.Template, body.Filters)
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not update export feed", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusOK, item)
}

func (s *Server) handleUpdateExportFeedTemplate(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	var body struct {
		Template any `json:"template"`
		Filters  any `json:"filters"`
	}
	if err := DecodeJSON(r, &body); err != nil {
		Error(w, http.StatusBadRequest, "invalid json")
		return
	}
	item, err := s.Feeds.UpdateExportFeedTemplate(r.Context(), cid, id, body.Template, body.Filters)
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not update export template", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusOK, item)
}

func (s *Server) handleDeleteExportFeed(w http.ResponseWriter, r *http.Request) {
	if !s.allowCompanyAdminOrPlatform(w, r) {
		return
	}
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	if err := s.Feeds.DeleteExportFeed(r.Context(), cid, id); err != nil {
		Error(w, http.StatusNotFound, "not found")
		return
	}
	JSON(w, http.StatusOK, map[string]string{"status": "ok"})
}

func (s *Server) handleRotateExportFeedPublicToken(w http.ResponseWriter, r *http.Request) {
	if !s.allowCompanyAdminOrPlatform(w, r) {
		return
	}
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	item, err := s.Feeds.RotateExportFeedPublicToken(r.Context(), cid, id)
	if err != nil {
		Error(w, http.StatusNotFound, "not found")
		return
	}
	JSON(w, http.StatusOK, item)
}

func (s *Server) handleGenerateExportFeed(w http.ResponseWriter, r *http.Request) {
	cid, _ := CompanyIDFromContext(r.Context())
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		Error(w, http.StatusBadRequest, "invalid id")
		return
	}
	item, err := s.Feeds.GenerateExportFeed(r.Context(), cid, id)
	if err != nil {
		ClientOrLog(w, http.StatusBadRequest, "could not generate export", err, feeds.ClientError)
		return
	}
	JSON(w, http.StatusOK, item)
}

func (s *Server) handlePublicExportXML(w http.ResponseWriter, r *http.Request) {
	token := chi.URLParam(r, "token")
	lw := &lazyHeaderWriter{ResponseWriter: w, contentType: "application/xml; charset=utf-8"}
	if err := s.Feeds.PublicExportXML(r.Context(), lw, token); err != nil {
		if !lw.wrote {
			writePublicExportError(w, err)
		}
		return
	}
}

func (s *Server) handlePublicExportCSV(w http.ResponseWriter, r *http.Request) {
	token := chi.URLParam(r, "token")
	lw := &lazyHeaderWriter{ResponseWriter: w, contentType: "text/csv; charset=utf-8"}
	if err := s.Feeds.PublicExportCSV(r.Context(), lw, token); err != nil {
		if !lw.wrote {
			writePublicExportError(w, err)
		}
		return
	}
}

func writePublicExportError(w http.ResponseWriter, err error) {
	// Format mismatch must look identical to unknown tokens so probing .xml/.csv
	// cannot confirm whether a guessed public_token exists.
	switch {
	case errors.Is(err, feeds.ErrFormatMismatch), errors.Is(err, pgx.ErrNoRows):
		Error(w, http.StatusNotFound, "export feed not found")
	default:
		Error(w, http.StatusNotFound, "export feed not found")
	}
}

type lazyHeaderWriter struct {
	http.ResponseWriter
	contentType string
	wrote       bool
}

func (l *lazyHeaderWriter) Write(p []byte) (int, error) {
	if !l.wrote {
		l.Header().Set("Content-Type", l.contentType)
		l.wrote = true
	}
	return l.ResponseWriter.Write(p)
}

func (l *lazyHeaderWriter) Flush() {
	if f, ok := l.ResponseWriter.(http.Flusher); ok {
		f.Flush()
	}
}