descrybe/apps/api/internal/feeds/service.go

package feeds

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

type Service struct {
	Pool      *pgxpool.Pool
	UploadDir string
}

// CreateInput is the payload for creating an input feed (URL and/or uploaded CSV).
// Legacy clients send name + item_path (url optional). V2 clients send name + url/file
// plus optional feed_type / sync_interval_minutes (or legacy sync_frequency in hours).
type CreateInput struct {
	Name                string
	URL                 string
	ItemPath            string
	FeedType            string
	SyncIntervalMinutes int
	SyncFrequencyHours  int // legacy alias; converted to minutes when SyncIntervalMinutes unset
	Options             map[string]any
}

func (s *Service) List(ctx context.Context, companyID uuid.UUID, limit, offset int, q string) ([]map[string]any, int64, int64, int64, error) {
	q = strings.TrimSpace(q)
	where := `company_id = $1`
	args := []any{companyID}
	if q != "" {
		where += ` AND (
			COALESCE(name, '') ILIKE '%' || $2 || '%' OR
			COALESCE(url, '') ILIKE '%' || $2 || '%' OR
			COALESCE(feed_type, '') ILIKE '%' || $2 || '%' OR
			COALESCE(status, '') ILIKE '%' || $2 || '%' OR
			COALESCE(options->>'source_filename', '') ILIKE '%' || $2 || '%'
		)`
		args = append(args, q)
	}
	var total int64
	if err := s.Pool.QueryRow(ctx, `SELECT count(*) FROM input_feeds WHERE `+where, args...).Scan(&total); err != nil {
		return nil, 0, 0, 0, err
	}
	// active_total = truly syncing; mapped_total = fields saved but not activated.
	var activeTotal, mappedTotal int64
	if err := s.Pool.QueryRow(ctx,
		`SELECT
			count(*) FILTER (WHERE lower(COALESCE(status, '')) = 'active'),
			count(*) FILTER (WHERE lower(COALESCE(status, '')) = 'mapped')
		FROM input_feeds WHERE `+where,
		args...,
	).Scan(&activeTotal, &mappedTotal); err != nil {
		return nil, 0, 0, 0, err
	}
	limitArg := len(args) + 1
	offsetArg := len(args) + 2
	query := fmt.Sprintf(`
		SELECT id, name, url, feed_type, status, sync_interval_minutes, last_synced_at, options, created_at, updated_at,
			(SELECT count(*)::int FROM raw_products rp WHERE rp.feed_id = input_feeds.id AND rp.company_id = input_feeds.company_id) AS product_count,
			(SELECT max(rp.updated_at) FROM raw_products rp WHERE rp.feed_id = input_feeds.id AND rp.company_id = input_feeds.company_id) AS products_updated_at
		FROM input_feeds WHERE %s
		ORDER BY created_at DESC LIMIT $%d OFFSET $%d`, where, limitArg, offsetArg)
	queryArgs := append(append([]any{}, args...), limit, offset)
	rows, err := s.Pool.Query(ctx, query, queryArgs...)
	if err != nil {
		return nil, 0, 0, 0, err
	}
	defer rows.Close()
	items, err := scanMaps(rows, []string{"id", "name", "url", "feed_type", "status", "sync_interval_minutes", "last_synced_at", "options", "created_at", "updated_at", "product_count", "products_updated_at"})
	if err != nil {
		return nil, 0, 0, 0, err
	}
	if err := s.attachMappingFieldCounts(ctx, companyID, items); err != nil {
		return nil, 0, 0, 0, err
	}
	return items, total, activeTotal, mappedTotal, nil
}

// ProductTotals is company-scoped catalog counts across all feeds.
type ProductTotals struct {
	Total       int64
	Processed   int64
	Unprocessed int64
}

// CompanyProductTotals returns company-scoped catalog counts for feeds/dashboard cards.
// ASSUMPTION: Total = count(raw_products); Processed = count(processed_products);
// Unprocessed = count(raw where processing_status='unprocessed'). These are not a
// partition of Total (P+U≠Total by design). Do not redefine without documenting a new ASSUMPTION.
func (s *Service) CompanyProductTotals(ctx context.Context, companyID uuid.UUID) (ProductTotals, error) {
	var t ProductTotals
	err := s.Pool.QueryRow(ctx, `
		SELECT
			(SELECT count(*)::bigint FROM raw_products WHERE company_id = $1),
			(SELECT count(*)::bigint FROM processed_products WHERE company_id = $1),
			(SELECT count(*)::bigint FROM raw_products
				WHERE company_id = $1 AND lower(COALESCE(processing_status, '')) = 'unprocessed')`,
		companyID,
	).Scan(&t.Total, &t.Processed, &t.Unprocessed)
	return t, err
}

func (s *Service) Create(ctx context.Context, companyID uuid.UUID, in CreateInput) (map[string]any, error) {
	name := strings.TrimSpace(in.Name)
	if name == "" {
		return nil, ClientMsg("name required")
	}
	url := strings.TrimSpace(in.URL)
	if err := ValidateFeedURL(ctx, url); err != nil {
		return nil, err
	}
	opts := in.Options
	if opts == nil {
		opts = map[string]any{}
	}
	itemPath := strings.TrimSpace(in.ItemPath)
	if itemPath == "" {
		if p, ok := opts["item_path"].(string); ok {
			itemPath = strings.TrimSpace(p)
		}
	}
	if itemPath != "" {
		opts["item_path"] = itemPath
	}
	hasLocal := sourcePathFromOptions(opts) != ""
	// Dual-support: legacy create allows name + item_path without url/file.
	if url == "" && !hasLocal && itemPath == "" {
		return nil, errSourceRequired
	}
	feedType := strings.ToLower(strings.TrimSpace(in.FeedType))
	if feedType == "" {
		if hasLocal {
			feedType = "csv"
		} else {
			feedType = "xml"
		}
	}
	if feedType != "xml" && feedType != "csv" {
		return nil, ClientMsg("feed_type must be xml or csv")
	}
	interval := in.SyncIntervalMinutes
	if interval <= 0 && in.SyncFrequencyHours > 0 {
		interval = in.SyncFrequencyHours * 60
	}
	if interval <= 0 {
		interval = 60
	}
	optsBytes, err := json.Marshal(opts)
	if err != nil {
		return nil, err
	}
	var id uuid.UUID
	err = s.Pool.QueryRow(ctx, `
		INSERT INTO input_feeds (company_id, name, url, feed_type, status, sync_interval_minutes, options)
		VALUES ($1, $2, $3, $4, 'unmapped', $5, $6::jsonb) RETURNING id`,
		companyID, name, nullStr(url), feedType, interval, optsBytes).Scan(&id)
	if err != nil {
		return nil, err
	}
	return s.Get(ctx, companyID, id)
}

func (s *Service) Get(ctx context.Context, companyID, id uuid.UUID) (map[string]any, error) {
	row := s.Pool.QueryRow(ctx, `
		SELECT id, name, url, feed_type, status, sync_interval_minutes, last_synced_at, options, created_at, updated_at,
			(SELECT count(*)::int FROM raw_products rp WHERE rp.feed_id = input_feeds.id AND rp.company_id = input_feeds.company_id) AS product_count,
			(SELECT max(rp.updated_at) FROM raw_products rp WHERE rp.feed_id = input_feeds.id AND rp.company_id = input_feeds.company_id) AS products_updated_at
		FROM input_feeds WHERE id = $1 AND company_id = $2`, id, companyID)
	item, err := scanMap(row, []string{"id", "name", "url", "feed_type", "status", "sync_interval_minutes", "last_synced_at", "options", "created_at", "updated_at", "product_count", "products_updated_at"})
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	if err := s.attachMappingFieldCounts(ctx, companyID, []map[string]any{item}); err != nil {
		return nil, err
	}
	return item, nil
}

// attachMappingFieldCounts sets mapping_field_count / has_mappings / mapping_incomplete
// on each feed row from the active feed_mappings document (one query for the page).
// mapping_incomplete mirrors list-chip blocking preflight (empty/required/item_path).
func (s *Service) attachMappingFieldCounts(ctx context.Context, companyID uuid.UUID, items []map[string]any) error {
	if len(items) == 0 {
		return nil
	}
	ids := make([]uuid.UUID, 0, len(items))
	index := make(map[uuid.UUID]map[string]any, len(items))
	for _, item := range items {
		id, ok := asUUID(item["id"])
		if !ok {
			continue
		}
		ids = append(ids, id)
		index[id] = item
		item["mapping_field_count"] = 0
		item["has_mappings"] = false
		item["mapping_incomplete"] = true
	}
	if len(ids) == 0 {
		return nil
	}
	required, err := s.loadRequiredStandardFields(ctx, companyID)
	if err != nil {
		return err
	}
	rows, err := s.Pool.Query(ctx, `
		SELECT DISTINCT ON (feed_id) feed_id, mappings
		FROM feed_mappings
		WHERE company_id = $1 AND is_active = true AND feed_id = ANY($2::uuid[])
		ORDER BY feed_id, version DESC`, companyID, ids)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var feedID uuid.UUID
		var raw []byte
		if err := rows.Scan(&feedID, &raw); err != nil {
			return err
		}
		item := index[feedID]
		if item == nil {
			continue
		}
		var parsed any
		if err := json.Unmarshal(raw, &parsed); err != nil {
			continue
		}
		mappings := parseMappings(parsed)
		n := len(mappings)
		item["mapping_field_count"] = n
		item["has_mappings"] = n > 0
		item["mapping_incomplete"] = mappingDocIncomplete(item, parsed, mappings, required)
	}
	return rows.Err()
}

func isCSVFeedType(feedType string) bool {
	t := strings.ToLower(strings.TrimSpace(feedType))
	return t == "csv" || t == "excel"
}

func feedItemPathHint(item map[string]any, mappingsRaw any) string {
	if p := itemPathFromMappings(mappingsRaw); p != "" {
		return p
	}
	if opts, ok := item["options"].(map[string]any); ok {
		if v, ok := opts["item_path"].(string); ok {
			if p := strings.TrimSpace(v); p != "" {
				return p
			}
		}
	}
	if v, ok := item["item_path"].(string); ok {
		return strings.TrimSpace(v)
	}
	return ""
}

// mappingDocIncomplete reports list-chip blocking gaps (empty mappings, required targets, XML item_path).
func mappingDocIncomplete(item map[string]any, mappingsRaw any, mappings []FieldMapping, required []requiredStandardField) bool {
	if err := validateMappingsForSync(mappings, required); err != nil {
		return true
	}
	feedType, _ := item["feed_type"].(string)
	if !isCSVFeedType(feedType) && feedItemPathHint(item, mappingsRaw) == "" {
		return true
	}
	return false
}

func asUUID(v any) (uuid.UUID, bool) {
	switch t := v.(type) {
	case uuid.UUID:
		return t, true
	case [16]byte:
		return uuid.UUID(t), true
	case string:
		id, err := uuid.Parse(t)
		return id, err == nil
	default:
		return uuid.Nil, false
	}
}

func (s *Service) Update(ctx context.Context, companyID, id uuid.UUID, body map[string]any) (map[string]any, error) {
	name, _ := body["name"].(string)
	url, _ := body["url"].(string)
	status, _ := body["status"].(string)
	feedType, _ := body["feed_type"].(string)
	itemPath, _ := body["item_path"].(string)
	if err := ValidateFeedURL(ctx, url); err != nil {
		return nil, err
	}
	feedType = strings.ToLower(strings.TrimSpace(feedType))
	if feedType != "" && feedType != "xml" && feedType != "csv" {
		return nil, ClientMsg("feed_type must be xml or csv")
	}
	interval := 0
	switch v := body["sync_interval_minutes"].(type) {
	case float64:
		interval = int(v)
	case int:
		interval = v
	case json.Number:
		n, _ := v.Int64()
		interval = int(n)
	}
	if interval <= 0 {
		switch v := body["sync_frequency"].(type) {
		case float64:
			interval = int(v) * 60
		case int:
			interval = v * 60
		}
	}

	ct, err := s.Pool.Exec(ctx, `
		UPDATE input_feeds SET
			name = CASE WHEN $3 <> '' THEN $3 ELSE name END,
			url = CASE WHEN $4 <> '' THEN $4 ELSE url END,
			status = CASE WHEN $5 <> '' THEN $5 ELSE status END,
			feed_type = CASE WHEN $6 <> '' THEN $6 ELSE feed_type END,
			sync_interval_minutes = CASE WHEN $7 > 0 THEN $7 ELSE sync_interval_minutes END,
			options = CASE
				WHEN $8 <> '' THEN COALESCE(options, '{}'::jsonb) || jsonb_build_object('item_path', to_jsonb($8::text))
				ELSE options
			END,
			updated_at = now()
		WHERE id = $1 AND company_id = $2`,
		id, companyID, name, url, status, feedType, interval, strings.TrimSpace(itemPath))
	if err != nil {
		return nil, err
	}
	if ct.RowsAffected() == 0 {
		return nil, ErrNotFound
	}
	return s.Get(ctx, companyID, id)
}

func (s *Service) Delete(ctx context.Context, companyID, id uuid.UUID) error {
	ct, err := s.Pool.Exec(ctx, `DELETE FROM input_feeds WHERE id = $1 AND company_id = $2`, id, companyID)
	if err != nil {
		return err
	}
	if ct.RowsAffected() == 0 {
		return ErrNotFound
	}
	return nil
}

func (s *Service) GetMappings(ctx context.Context, companyID, feedID uuid.UUID) (map[string]any, error) {
	var id uuid.UUID
	var version int
	var mappings []byte
	err := s.Pool.QueryRow(ctx, `
		SELECT id, version, mappings FROM feed_mappings
		WHERE feed_id = $1 AND company_id = $2 AND is_active = true
		ORDER BY version DESC LIMIT 1`, feedID, companyID).Scan(&id, &version, &mappings)
	if err != nil {
		return nil, err
	}
	var m any
	_ = json.Unmarshal(mappings, &m)
	return map[string]any{"id": id, "version": version, "mappings": m}, nil
}

func (s *Service) PutMappings(ctx context.Context, companyID, feedID uuid.UUID, mappings any) (map[string]any, error) {
	b, err := json.Marshal(mappings)
	if err != nil {
		return nil, err
	}
	var version int
	_ = s.Pool.QueryRow(ctx, `
		SELECT COALESCE(MAX(version), 0) FROM feed_mappings WHERE feed_id = $1`, feedID).Scan(&version)
	version++
	_, _ = s.Pool.Exec(ctx, `UPDATE feed_mappings SET is_active = false WHERE feed_id = $1`, feedID)
	var id uuid.UUID
	err = s.Pool.QueryRow(ctx, `
		INSERT INTO feed_mappings (feed_id, company_id, version, mappings, is_active)
		VALUES ($1, $2, $3, $4, true) RETURNING id`, feedID, companyID, version, b).Scan(&id)
	if err != nil {
		return nil, err
	}
	// Keep feed.options.item_path in sync for Sync() XML item selection, and
	// flip unmapped -> mapped whenever at least one field mapping is saved.
	path := itemPathFromMappings(mappings)
	hasFields := len(parseMappings(mappings)) > 0
	switch {
	case path != "":
		_, _ = s.Pool.Exec(ctx, `
			UPDATE input_feeds SET
				options = COALESCE(options, '{}'::jsonb) || jsonb_build_object('item_path', to_jsonb($3::text)),
				status = CASE WHEN status = 'unmapped' THEN 'mapped' ELSE status END,
				updated_at = now()
			WHERE id = $1 AND company_id = $2`, feedID, companyID, path)
	case hasFields:
		_, _ = s.Pool.Exec(ctx, `
			UPDATE input_feeds SET
				status = CASE WHEN status = 'unmapped' THEN 'mapped' ELSE status END,
				updated_at = now()
			WHERE id = $1 AND company_id = $2`, feedID, companyID)
	}
	return map[string]any{"id": id, "version": version, "mappings": mappings}, nil
}

func (s *Service) ListExportFeeds(ctx context.Context, companyID uuid.UUID, limit, offset int) ([]map[string]any, int64, error) {
	var total int64
	if err := s.Pool.QueryRow(ctx, `SELECT count(*) FROM export_feeds WHERE company_id = $1`, companyID).Scan(&total); err != nil {
		return nil, 0, err
	}
	rows, err := s.Pool.Query(ctx, `
		SELECT id, name, source_feed_id, format, public_token, is_active, last_generated_at, created_at, updated_at
		FROM export_feeds WHERE company_id = $1
		ORDER BY created_at DESC LIMIT $2 OFFSET $3`, companyID, limit, offset)
	if err != nil {
		return nil, 0, err
	}
	defer rows.Close()
	items, err := scanMaps(rows, []string{"id", "name", "source_feed_id", "format", "public_token", "is_active", "last_generated_at", "created_at", "updated_at"})
	if err != nil {
		return nil, 0, err
	}
	return items, total, nil
}

func (s *Service) GetExportFeed(ctx context.Context, companyID, id uuid.UUID) (map[string]any, error) {
	rows, err := s.Pool.Query(ctx, `
		SELECT id, name, source_feed_id, format, public_token, template, filters, is_active, last_generated_at, created_at, updated_at
		FROM export_feeds WHERE id = $1 AND company_id = $2`, id, companyID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items, err := scanMaps(rows, []string{"id", "name", "source_feed_id", "format", "public_token", "template", "filters", "is_active", "last_generated_at", "created_at", "updated_at"})
	if err != nil {
		return nil, err
	}
	if len(items) == 0 {
		return nil, errors.New("not found")
	}
	return items[0], nil
}

func (s *Service) DeleteExportFeed(ctx context.Context, companyID, id uuid.UUID) error {
	ct, err := s.Pool.Exec(ctx, `DELETE FROM export_feeds WHERE id = $1 AND company_id = $2`, id, companyID)
	if err != nil {
		return err
	}
	if ct.RowsAffected() == 0 {
		return errors.New("not found")
	}
	return nil
}

func (s *Service) UpdateExportFeed(ctx context.Context, companyID, id uuid.UUID, name *string, isActive *bool, template, filters any) (map[string]any, error) {
	current, err := s.GetExportFeed(ctx, companyID, id)
	if err != nil {
		return nil, err
	}
	if name != nil {
		n := strings.TrimSpace(*name)
		if n == "" {
			return nil, ClientMsg("name required")
		}
		if _, err := s.Pool.Exec(ctx, `
			UPDATE export_feeds SET name = $3, updated_at = now() WHERE id = $1 AND company_id = $2`, id, companyID, n); err != nil {
			return nil, err
		}
	}
	if isActive != nil {
		if _, err := s.Pool.Exec(ctx, `
			UPDATE export_feeds SET is_active = $3, updated_at = now() WHERE id = $1 AND company_id = $2`, id, companyID, *isActive); err != nil {
			return nil, err
		}
	}
	if template != nil || filters != nil {
		tpl := template
		flt := filters
		if tpl == nil {
			tpl = current["template"]
		}
		if flt == nil {
			flt = current["filters"]
		}
		if _, err := s.UpdateExportFeedTemplate(ctx, companyID, id, tpl, flt); err != nil {
			return nil, err
		}
	}
	return s.GetExportFeed(ctx, companyID, id)
}

func (s *Service) CreateExportFeed(ctx context.Context, companyID uuid.UUID, in CreateExportInput) (map[string]any, error) {
	if strings.TrimSpace(in.Name) == "" {
		return nil, ClientMsg("name required")
	}
	format := strings.ToLower(strings.TrimSpace(in.Format))
	if format == "" {
		format = "xml"
	}
	if format != "xml" && format != "csv" {
		return nil, ClientMsg("format must be xml or csv")
	}
	var src *uuid.UUID
	if in.SourceFeedID != nil && *in.SourceFeedID != "" {
		id, err := uuid.Parse(*in.SourceFeedID)
		if err != nil {
			return nil, ClientMsg("invalid source_feed_id")
		}
		src = &id
	}
	tplBytes := []byte("{}")
	if in.Template != nil {
		b, err := json.Marshal(in.Template)
		if err != nil {
			return nil, err
		}
		tplBytes = b
	}
	filterBytes := []byte("{}")
	if in.Filters != nil {
		b, err := json.Marshal(in.Filters)
		if err != nil {
			return nil, err
		}
		filterBytes = b
	}
	token, err := newPublicExportToken()
	if err != nil {
		return nil, err
	}
	var id uuid.UUID
	err = s.Pool.QueryRow(ctx, `
    INSERT INTO export_feeds (company_id, name, source_feed_id, format, template, filters, public_token)
    VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7) RETURNING id, public_token`,
		companyID, in.Name, src, format, tplBytes, filterBytes, token).Scan(&id, &token)
	if err != nil {
		return nil, err
	}
	return map[string]any{
		"id": id, "name": in.Name, "format": format, "public_token": token,
		"template": in.Template, "filters": in.Filters,
	}, nil
}

// RotateExportFeedPublicToken replaces the public URL token (revokes the previous URL).
func (s *Service) RotateExportFeedPublicToken(ctx context.Context, companyID, id uuid.UUID) (map[string]any, error) {
	token, err := newPublicExportToken()
	if err != nil {
		return nil, err
	}
	ct, err := s.Pool.Exec(ctx, `
		UPDATE export_feeds SET public_token = $3, updated_at = now()
		WHERE id = $1 AND company_id = $2`, id, companyID, token)
	if err != nil {
		return nil, err
	}
	if ct.RowsAffected() == 0 {
		return nil, errors.New("not found")
	}
	return s.GetExportFeed(ctx, companyID, id)
}

func nullStr(s string) *string {
	if s == "" {
		return nil
	}
	return &s
}

func xmlEscape(s string) string {
	r := strings.NewReplacer("&", "&amp;", "<", "&lt;", ">", "&gt;", `"`, "&quot;")
	return r.Replace(s)
}

func scanMaps(rows pgx.Rows, cols []string) ([]map[string]any, error) {
	out := make([]map[string]any, 0)
	for rows.Next() {
		vals := make([]any, len(cols))
		ptrs := make([]any, len(cols))
		for i := range vals {
			ptrs[i] = &vals[i]
		}
		if err := rows.Scan(ptrs...); err != nil {
			return nil, err
		}
		m := make(map[string]any, len(cols))
		for i, c := range cols {
			m[c] = normalize(vals[i])
		}
		out = append(out, m)
	}
	return out, rows.Err()
}

func scanMap(row pgx.Row, cols []string) (map[string]any, error) {
	vals := make([]any, len(cols))
	ptrs := make([]any, len(cols))
	for i := range vals {
		ptrs[i] = &vals[i]
	}
	if err := row.Scan(ptrs...); err != nil {
		return nil, err
	}
	m := make(map[string]any, len(cols))
	for i, c := range cols {
		m[c] = normalize(vals[i])
	}
	return m, nil
}

func normalize(v any) any {
	switch t := v.(type) {
	case []byte:
		var j any
		if json.Unmarshal(t, &j) == nil {
			return j
		}
		return string(t)
	case [16]byte:
		return uuid.UUID(t).String()
	default:
		return v
	}
}