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

package feeds

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/csv"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"sort"
	"strings"
	"time"
	"unicode"

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

const (
	exportChunkHint           = 500   // flush HTTP/CSV buffer every N products
	exportBatchSize           = 1000  // SQL keyset page size (bounded memory)
	exportMaxProducts         = 50000 // hard ceiling for selected product_ids
	exportSelectedMaxProducts = 2000  // in-memory selected export buffer cap
	defaultExportRoot         = "products"
	defaultExportItem         = "product"
)

// exportProductSeq yields products one at a time without collecting the full set.
type exportProductSeq func(yield func(exportProduct) error) error

// CreateExportInput creates an export feed (XML or CSV).
type CreateExportInput struct {
	Name         string
	SourceFeedID *string
	Format       string
	Template     any
	Filters      any
}

type exportFeedRow struct {
	ID           uuid.UUID
	CompanyID    uuid.UUID
	Name         string
	SourceFeedID *uuid.UUID
	Format       string
	Template     []byte
	Filters      []byte
	IsActive     bool
}

type exportField struct {
	Key    string `json:"key"`
	Source string `json:"source"`
}

type exportTemplate struct {
	Root     string            `json:"root"`
	Item     string            `json:"item"`
	Fields   []exportField     `json:"fields"`
	Mappings map[string]string `json:"mappings"` // key -> source (legacy-ish shorthand)
}

type exportFilters struct {
	Statuses []string `json:"statuses"`
	FeedID   string   `json:"feed_id"`
}

type exportProduct struct {
	ProductID            *string
	Name                 *string
	Category             *string
	Description          *string
	ProcessedDescription *string
	ProcessedName        *string
	Status               *string
	Attributes           []byte
	ProcessedAttributes  []byte
	MappedData           []byte
	FeedID               *uuid.UUID
}

// Known EPREL / energy-label source aliases resolved from processed JSON.
var eprelSourceAliases = map[string][]string{
	"eprel_id":           {"eprel_id", "eprelId", "EPRELID", "EprelId", "eprelID"},
	"energy_class":       {"energy_class", "energyClass", "eprel_energy_class", "eprel_class", "eprelClass"},
	"eprel_energy_class": {"eprel_energy_class", "energy_class", "energyClass", "eprel_class"},
	"eprel_class":        {"eprel_class", "energy_class", "energyClass", "eprel_energy_class"},
	"energy_scale":       {"energy_scale", "energyScale", "eprel_energy_scale", "eprel_scale"},
	"eprel_energy_scale": {"eprel_energy_scale", "energy_scale", "energyScale", "eprel_scale"},
	"eprel_scale":        {"eprel_scale", "energy_scale", "energyScale", "eprel_energy_scale"},
	"eprel_label":        {"eprel_label", "eprel_label_url", "label"},
	"eprel_label_url":    {"eprel_label_url", "eprel_label", "label"},
	"eprel_pdf":          {"eprel_pdf", "eprel_pdf_url", "pdf"},
	"eprel_pdf_url":      {"eprel_pdf_url", "eprel_pdf", "pdf"},
	"eprel_brand":        {"eprel_brand", "brand"},
	"eprel_model":        {"eprel_model", "model", "model_identifier"},
	"eprel_gtin":         {"eprel_gtin", "gtin", "ean"},
}

// publicExportTokenBytes is CSPRNG entropy for new public export tokens (256 bits).
// Historical DB defaults used 16 bytes (128 bits / 32 hex); validation still accepts those.
const publicExportTokenBytes = 32

// ValidPublicToken reports whether a public export token has the expected hex shape.
// Used by HTTP middleware to reject probes without a DB round-trip.
func ValidPublicToken(token string) bool {
	return validPublicToken(token)
}

// validPublicToken rejects undersized or non-hex tokens before DB lookup (scrape probing).
// Floor is 32 hex chars (128 bits) matching historical gen_random_bytes(16) defaults;
// new tokens are 64 hex chars (256 bits).
func validPublicToken(token string) bool {
	n := len(token)
	if n < 32 || n > 64 || n%2 != 0 {
		return false
	}
	for _, r := range token {
		if unicode.Is(unicode.ASCII_Hex_Digit, r) {
			continue
		}
		return false
	}
	return true
}

func newPublicExportToken() (string, error) {
	b := make([]byte, publicExportTokenBytes)
	if _, err := rand.Read(b); err != nil {
		return "", err
	}
	return hex.EncodeToString(b), nil
}

func sanitizeXMLName(name, fallback string) string {
	name = strings.TrimSpace(name)
	if name == "" {
		return fallback
	}
	var b strings.Builder
	colonUsed := false
	for i, r := range name {
		ok := unicode.IsLetter(r) || unicode.IsDigit(r) || r == '_' || r == '-' || r == '.'
		// Allow one namespace colon (e.g. g:id for Google Shopping XML).
		if r == ':' && !colonUsed && i > 0 && i < len(name)-1 {
			ok = true
			colonUsed = true
		}
		if i == 0 && (unicode.IsDigit(r) || r == '-' || r == '.' || r == ':') {
			b.WriteByte('_')
			if r == ':' {
				continue
			}
		}
		if ok {
			b.WriteRune(r)
		} else {
			b.WriteByte('_')
		}
	}
	out := b.String()
	if out == "" || out == "_" {
		return fallback
	}
	return out
}

func parseExportTemplate(raw []byte) exportTemplate {
	tpl := exportTemplate{
		Root: defaultExportRoot,
		Item: defaultExportItem,
		Fields: []exportField{
			{Key: "product_id", Source: "product_id"},
			{Key: "name", Source: "name"},
			{Key: "category", Source: "category"},
			{Key: "description", Source: "description"},
			{Key: "status", Source: "status"},
		},
	}
	if len(raw) == 0 || string(raw) == "{}" || string(raw) == "null" {
		return tpl
	}
	var parsed exportTemplate
	if err := json.Unmarshal(raw, &parsed); err != nil {
		return tpl
	}
	if parsed.Root != "" {
		tpl.Root = parsed.Root
	}
	if parsed.Item != "" {
		tpl.Item = parsed.Item
	}
	if len(parsed.Fields) > 0 {
		tpl.Fields = parsed.Fields
	} else if len(parsed.Mappings) > 0 {
		keys := make([]string, 0, len(parsed.Mappings))
		for key := range parsed.Mappings {
			keys = append(keys, key)
		}
		sort.Strings(keys)
		fields := make([]exportField, 0, len(keys))
		for _, key := range keys {
			source := parsed.Mappings[key]
			if source == "" {
				source = key
			}
			fields = append(fields, exportField{Key: key, Source: source})
		}
		tpl.Fields = fields
	}
	return tpl
}

func parseExportFilters(raw []byte) exportFilters {
	var f exportFilters
	if len(raw) == 0 || string(raw) == "{}" || string(raw) == "null" {
		return f
	}
	_ = json.Unmarshal(raw, &f)
	return f
}

func (s *Service) loadExportFeedByToken(ctx context.Context, token string) (exportFeedRow, error) {
	token = strings.ToLower(strings.TrimSpace(token))
	if !validPublicToken(token) {
		return exportFeedRow{}, pgx.ErrNoRows
	}
	var row exportFeedRow
	err := s.Pool.QueryRow(ctx, `
		SELECT id, company_id, name, source_feed_id, format, template, filters, is_active
		FROM export_feeds
		WHERE public_token = $1 AND is_active = true`, token).Scan(
		&row.ID, &row.CompanyID, &row.Name, &row.SourceFeedID, &row.Format, &row.Template, &row.Filters, &row.IsActive,
	)
	return row, err
}

func (s *Service) loadExportFeedByID(ctx context.Context, companyID, id uuid.UUID) (exportFeedRow, error) {
	var row exportFeedRow
	err := s.Pool.QueryRow(ctx, `
		SELECT id, company_id, name, source_feed_id, format, template, filters, is_active
		FROM export_feeds
		WHERE id = $1 AND company_id = $2`, id, companyID).Scan(
		&row.ID, &row.CompanyID, &row.Name, &row.SourceFeedID, &row.Format, &row.Template, &row.Filters, &row.IsActive,
	)
	return row, err
}

func resolveStatuses(f exportFilters) []string {
	if len(f.Statuses) > 0 {
		out := make([]string, 0, len(f.Statuses))
		for _, st := range f.Statuses {
			st = strings.TrimSpace(st)
			if st != "" {
				out = append(out, st)
			}
		}
		if len(out) > 0 {
			return out
		}
	}
	return []string{"processed", "completed"}
}

func resolveFeedFilter(row exportFeedRow, f exportFilters) *uuid.UUID {
	if f.FeedID != "" {
		id, err := uuid.Parse(f.FeedID)
		if err == nil {
			return &id
		}
	}
	return row.SourceFeedID
}

// queryExportProductsBatch loads one keyset page of export products (newest first).
// Pass cursorUpdatedAt/cursorID as nil/uuid.Nil for the first page; subsequent pages
// continue after the previous page's last (updated_at, id) pair.
func (s *Service) queryExportProductsBatch(
	ctx context.Context,
	row exportFeedRow,
	cursorUpdatedAt *time.Time,
	cursorID uuid.UUID,
	limit int,
) (pgx.Rows, error) {
	if limit <= 0 {
		limit = exportBatchSize
	}
	filters := parseExportFilters(row.Filters)
	statuses := resolveStatuses(filters)
	feedID := resolveFeedFilter(row, filters)
	// Prefer processed_products; LEFT JOIN raw only for mapped_data fallback (eprel_id, etc.).
	// Keyset on (updated_at DESC, id DESC) keeps each round-trip bounded to `limit` rows.
	return s.Pool.Query(ctx, `
		SELECT p.id, p.updated_at, p.product_id, p.name, p.category, p.description, p.processed_description, p.processed_name,
			p.status, p.attributes, p.processed_attributes,
			COALESCE(r.mapped_data, '{}'::jsonb), p.feed_id
		FROM processed_products p
		LEFT JOIN raw_products r ON r.id = p.raw_product_id AND r.company_id = p.company_id
		WHERE p.company_id = $1
			AND p.status = ANY($2::text[])
			AND ($3::uuid IS NULL OR p.feed_id = $3)
			AND ($4::timestamptz IS NULL OR (p.updated_at, p.id) < ($4::timestamptz, $5::uuid))
		ORDER BY p.updated_at DESC, p.id DESC
		LIMIT $6`,
		row.CompanyID, statuses, feedID, cursorUpdatedAt, cursorID, limit,
	)
}

// forEachExportProduct walks matching products in keyset batches so export never
// materializes the full result set in memory.
func (s *Service) forEachExportProduct(ctx context.Context, row exportFeedRow, yield func(exportProduct) error) error {
	var (
		cursorUpdatedAt *time.Time
		cursorID        uuid.UUID
	)
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		rows, err := s.queryExportProductsBatch(ctx, row, cursorUpdatedAt, cursorID, exportBatchSize)
		if err != nil {
			return err
		}
		n := 0
		var lastUpdated time.Time
		var lastID uuid.UUID
		for rows.Next() {
			p, id, updatedAt, err := scanExportProductWithCursor(rows)
			if err != nil {
				rows.Close()
				return err
			}
			if err := yield(p); err != nil {
				rows.Close()
				return err
			}
			lastUpdated = updatedAt
			lastID = id
			n++
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			return err
		}
		if n == 0 {
			return nil
		}
		if n < exportBatchSize {
			return nil
		}
		u := lastUpdated
		cursorUpdatedAt = &u
		cursorID = lastID
	}
}

func rowsToExportSeq(rows pgx.Rows) exportProductSeq {
	return func(yield func(exportProduct) error) error {
		for rows.Next() {
			p, err := scanExportProduct(rows)
			if err != nil {
				return err
			}
			if err := yield(p); err != nil {
				return err
			}
		}
		return rows.Err()
	}
}

func productFieldValue(p exportProduct, source string) string {
	source = strings.TrimSpace(source)
	switch source {
	case "product_id", "id", "sku":
		return derefStr(p.ProductID)
	case "gtin", "ean", "upc", "barcode":
		attrs := flattenProductAttrs(p)
		for _, key := range []string{"gtin", "ean", "upc", "barcode", "eprel_gtin"} {
			if v := lookupFlattened(attrs, key); v != "" {
				return v
			}
		}
		return ""
	case "name", "title":
		if p.ProcessedName != nil && *p.ProcessedName != "" {
			return *p.ProcessedName
		}
		return derefStr(p.Name)
	case "category":
		return derefStr(p.Category)
	case "description":
		if p.ProcessedDescription != nil && *p.ProcessedDescription != "" {
			return *p.ProcessedDescription
		}
		return derefStr(p.Description)
	case "processed_description":
		return derefStr(p.ProcessedDescription)
	case "processed_name":
		return derefStr(p.ProcessedName)
	case "status":
		return derefStr(p.Status)
	case "feed_id":
		if p.FeedID != nil {
			return p.FeedID.String()
		}
		return ""
	case "attributes":
		return jsonOrEmpty(p.Attributes)
	case "processed_attributes":
		return jsonOrEmpty(p.ProcessedAttributes)
	case "specifications", "specifications.*", "specs", "specs.*":
		return formatFlattenedSpecs(flattenProductAttrs(p))
	default:
		if aliases, ok := eprelSourceAliases[source]; ok {
			attrs := flattenProductAttrs(p)
			for _, key := range aliases {
				if v := attrs[key]; v != "" {
					return v
				}
				if v := attrs["eprel."+key]; v != "" {
					return v
				}
			}
			return ""
		}
		key := source
		switch {
		case strings.HasPrefix(source, "attr."):
			key = strings.TrimPrefix(source, "attr.")
		case strings.HasPrefix(source, "spec."):
			key = strings.TrimPrefix(source, "spec.")
		case strings.HasPrefix(source, "specifications."):
			key = strings.TrimPrefix(source, "specifications.")
		case strings.HasPrefix(source, "eprel."):
			key = strings.TrimPrefix(source, "eprel.")
		}
		attrs := flattenProductAttrs(p)
		if v := lookupFlattened(attrs, key); v != "" {
			return v
		}
		if v := lookupFlattened(attrs, source); v != "" {
			return v
		}
		return ""
	}
}

// flattenProductAttrs merges processed_attributes → attributes → mapped_data and
// flattens nested specifications / eprel objects into exportable scalar keys.
func flattenProductAttrs(p exportProduct) map[string]string {
	out := map[string]string{}
	// Lowest priority first so higher-priority sources overwrite.
	mergeAttrBlob(out, p.MappedData)
	mergeAttrBlob(out, p.Attributes)
	mergeAttrBlob(out, p.ProcessedAttributes)
	return out
}

func mergeAttrBlob(out map[string]string, raw []byte) {
	if len(raw) == 0 || string(raw) == "null" {
		return
	}
	var m map[string]any
	if err := json.Unmarshal(raw, &m); err != nil || m == nil {
		return
	}
	for k, v := range m {
		putAttrValue(out, k, v)
	}
	// Nested specifications → flat keys.
	if specs, ok := m["specifications"]; ok {
		flattenSpecifications(out, specs)
	}
	if specs, ok := m["specs"]; ok {
		flattenSpecifications(out, specs)
	}
	// Nested eprel object → eprel_* / energy_* aliases.
	if eprel, ok := m["eprel"]; ok {
		flattenEprelObject(out, eprel)
	}
}

func flattenSpecifications(out map[string]string, specs any) {
	switch s := specs.(type) {
	case map[string]any:
		for k, v := range s {
			putAttrValue(out, k, v)
			putAttrValue(out, "spec."+k, v)
			putAttrValue(out, "specifications."+k, v)
		}
	case []any:
		for _, item := range s {
			obj, ok := item.(map[string]any)
			if !ok {
				continue
			}
			key := firstString(obj, "key", "name", "id", "attribute_key")
			if key == "" {
				continue
			}
			val := obj["value"]
			if val == nil {
				val = obj["name"]
			}
			putAttrValue(out, key, val)
			putAttrValue(out, "spec."+key, val)
			putAttrValue(out, "specifications."+key, val)
		}
	}
}

func flattenEprelObject(out map[string]string, eprel any) {
	obj, ok := eprel.(map[string]any)
	if !ok || obj == nil {
		return
	}
	for k, v := range obj {
		putAttrValue(out, k, v)
		putAttrValue(out, "eprel."+k, v)
		switch strings.ToLower(k) {
		case "id", "eprelid", "eprel_id":
			putAttrValue(out, "eprel_id", v)
		case "energyclass", "energy_class", "class":
			putAttrValue(out, "energy_class", v)
			putAttrValue(out, "eprel_energy_class", v)
			putAttrValue(out, "eprel_class", v)
		case "energyscale", "energy_scale", "scale":
			putAttrValue(out, "energy_scale", v)
			putAttrValue(out, "eprel_energy_scale", v)
			putAttrValue(out, "eprel_scale", v)
		case "label", "label_url", "labelurl":
			putAttrValue(out, "eprel_label", v)
			putAttrValue(out, "eprel_label_url", v)
		case "pdf", "pdf_url", "pdfurl":
			putAttrValue(out, "eprel_pdf", v)
			putAttrValue(out, "eprel_pdf_url", v)
		case "brand":
			putAttrValue(out, "eprel_brand", v)
		case "model", "model_identifier":
			putAttrValue(out, "eprel_model", v)
		case "gtin", "ean":
			putAttrValue(out, "eprel_gtin", v)
		}
	}
}

func putAttrValue(out map[string]string, key string, v any) {
	key = strings.TrimSpace(key)
	if key == "" || v == nil {
		return
	}
	if s := scalarAttrString(v); s != "" {
		out[key] = s
	}
}

func scalarAttrString(v any) string {
	if v == nil {
		return ""
	}
	switch t := v.(type) {
	case string:
		return t
	case float64:
		if t == float64(int64(t)) {
			return fmt.Sprintf("%d", int64(t))
		}
		return fmt.Sprint(t)
	case bool:
		return fmt.Sprint(t)
	case map[string]any:
		// Prefer display name/value from structured attribute objects.
		if s := firstString(t, "value", "name", "#text", "text"); s != "" {
			return s
		}
		b, err := json.Marshal(t)
		if err != nil {
			return ""
		}
		return string(b)
	default:
		b, err := json.Marshal(t)
		if err != nil {
			return ""
		}
		s := string(b)
		if s == "null" {
			return ""
		}
		return s
	}
}

func firstString(m map[string]any, keys ...string) string {
	for _, k := range keys {
		if v, ok := m[k]; ok && v != nil {
			if s, ok := v.(string); ok && strings.TrimSpace(s) != "" {
				return s
			}
			if s := scalarAttrString(v); s != "" && !strings.HasPrefix(s, "{") && !strings.HasPrefix(s, "[") {
				return s
			}
		}
	}
	return ""
}

func lookupFlattened(attrs map[string]string, key string) string {
	if key == "" || attrs == nil {
		return ""
	}
	if v, ok := attrs[key]; ok && v != "" {
		return v
	}
	// Case-insensitive fallback for vendor key variants.
	lower := strings.ToLower(key)
	for k, v := range attrs {
		if strings.ToLower(k) == lower && v != "" {
			return v
		}
	}
	return ""
}

func formatFlattenedSpecs(attrs map[string]string) string {
	keys := make([]string, 0)
	seen := map[string]struct{}{}
	for k := range attrs {
		if strings.HasPrefix(k, "spec.") {
			base := strings.TrimPrefix(k, "spec.")
			if _, ok := seen[base]; ok {
				continue
			}
			seen[base] = struct{}{}
			keys = append(keys, base)
		}
	}
	if len(keys) == 0 {
		return ""
	}
	sort.Strings(keys)
	parts := make([]string, 0, len(keys))
	for _, k := range keys {
		parts = append(parts, k+": "+attrs["spec."+k])
	}
	return strings.Join(parts, "; ")
}

// expandSpecFields returns sorted (key, value) pairs for specifications.* expansion.
func expandSpecFields(p exportProduct) []exportField {
	attrs := flattenProductAttrs(p)
	keys := make([]string, 0)
	seen := map[string]struct{}{}
	for k := range attrs {
		if !strings.HasPrefix(k, "spec.") {
			continue
		}
		base := strings.TrimPrefix(k, "spec.")
		if base == "" {
			continue
		}
		if _, ok := seen[base]; ok {
			continue
		}
		seen[base] = struct{}{}
		keys = append(keys, base)
	}
	sort.Strings(keys)
	fields := make([]exportField, 0, len(keys))
	for _, k := range keys {
		fields = append(fields, exportField{Key: k, Source: "spec." + k})
	}
	return fields
}

func isSpecExpandSource(source string) bool {
	switch strings.TrimSpace(source) {
	case "specifications.*", "specs.*", "attr.specifications.*":
		return true
	default:
		return false
	}
}

func derefStr(p *string) string {
	if p == nil {
		return ""
	}
	return *p
}

func jsonOrEmpty(b []byte) string {
	if len(b) == 0 || string(b) == "null" {
		return ""
	}
	return string(b)
}

func scanExportProduct(rows pgx.Rows) (exportProduct, error) {
	var p exportProduct
	err := rows.Scan(
		&p.ProductID, &p.Name, &p.Category, &p.Description, &p.ProcessedDescription, &p.ProcessedName,
		&p.Status, &p.Attributes, &p.ProcessedAttributes, &p.MappedData, &p.FeedID,
	)
	return p, err
}

func scanExportProductWithCursor(rows pgx.Rows) (exportProduct, uuid.UUID, time.Time, error) {
	var (
		p         exportProduct
		id        uuid.UUID
		updatedAt time.Time
	)
	err := rows.Scan(
		&id, &updatedAt,
		&p.ProductID, &p.Name, &p.Category, &p.Description, &p.ProcessedDescription, &p.ProcessedName,
		&p.Status, &p.Attributes, &p.ProcessedAttributes, &p.MappedData, &p.FeedID,
	)
	return p, id, updatedAt, err
}

func (s *Service) touchLastGenerated(ctx context.Context, id uuid.UUID) error {
	_, err := s.Pool.Exec(ctx, `
		UPDATE export_feeds SET last_generated_at = now(), updated_at = now() WHERE id = $1`, id)
	return err
}

// StreamPublicExport writes XML or CSV for an active export feed identified by public_token.
// Products are streamed from the DB in company scope (tenant isolation via token → company_id).
func (s *Service) StreamPublicExport(ctx context.Context, w io.Writer, token, wantFormat string) error {
	row, err := s.loadExportFeedByToken(ctx, token)
	if err != nil {
		return err
	}
	format := strings.ToLower(strings.TrimSpace(row.Format))
	if wantFormat != "" && format != "" && format != strings.ToLower(wantFormat) {
		return ErrFormatMismatch
	}
	if format == "" {
		format = strings.ToLower(wantFormat)
	}
	count, err := s.streamExport(ctx, w, row, format)
	if err != nil {
		return err
	}
	_ = count
	_ = s.touchLastGenerated(ctx, row.ID)
	return nil
}

// PublicExportXML streams XML for a public export token.
func (s *Service) PublicExportXML(ctx context.Context, w io.Writer, token string) error {
	return s.StreamPublicExport(ctx, w, token, "xml")
}

// PublicExportCSV streams CSV for a public export token.
func (s *Service) PublicExportCSV(ctx context.Context, w io.Writer, token string) error {
	return s.StreamPublicExport(ctx, w, token, "csv")
}

// GenerateExportFeed runs an on-demand generation for a company-owned export feed and
// updates last_generated_at. Content is not persisted to disk; public URLs stream live.
func (s *Service) GenerateExportFeed(ctx context.Context, companyID, id uuid.UUID) (map[string]any, error) {
	row, err := s.loadExportFeedByID(ctx, companyID, id)
	if err != nil {
		return nil, err
	}
	if !row.IsActive {
		return nil, ClientMsg("export feed inactive")
	}
	format := strings.ToLower(row.Format)
	if format == "" {
		format = "xml"
	}
	n, err := s.streamExport(ctx, io.Discard, row, format)
	if err != nil {
		return nil, err
	}
	if err := s.touchLastGenerated(ctx, row.ID); err != nil {
		return nil, err
	}
	var lastGen *string
	_ = s.Pool.QueryRow(ctx, `
		SELECT to_char(last_generated_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS"Z"')
		FROM export_feeds WHERE id = $1`, id).Scan(&lastGen)
	return map[string]any{
		"id":                id,
		"format":            format,
		"products_exported": n,
		"last_generated_at": lastGen,
		"status":            "completed",
	}, nil
}

func (s *Service) streamExport(ctx context.Context, w io.Writer, row exportFeedRow, format string) (int, error) {
	tpl := parseExportTemplate(row.Template)
	seq := exportProductSeq(func(yield func(exportProduct) error) error {
		return s.forEachExportProduct(ctx, row, yield)
	})
	switch format {
	case "csv":
		return streamCSV(w, seq, tpl)
	default:
		return streamXML(w, seq, tpl, row.Name)
	}
}

func streamXML(w io.Writer, seq exportProductSeq, tpl exportTemplate, feedName string) (int, error) {
	root := sanitizeXMLName(tpl.Root, defaultExportRoot)
	item := sanitizeXMLName(tpl.Item, defaultExportItem)
	if _, err := fmt.Fprintf(w, "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<%s feed=\"%s\">\n",
		root, xmlEscape(feedName)); err != nil {
		return 0, err
	}
	count := 0
	err := seq(func(p exportProduct) error {
		if err := writeXMLProduct(w, item, tpl, p); err != nil {
			return err
		}
		count++
		if count%exportChunkHint == 0 {
			if f, ok := w.(interface{ Flush() }); ok {
				f.Flush()
			}
		}
		return nil
	})
	if err != nil {
		return count, err
	}
	_, err = fmt.Fprintf(w, "</%s>\n", root)
	return count, err
}

func writeXMLProduct(w io.Writer, item string, tpl exportTemplate, p exportProduct) error {
	if _, err := fmt.Fprintf(w, "  <%s>\n", item); err != nil {
		return err
	}
	for _, field := range tpl.Fields {
		if isSpecExpandSource(field.Source) {
			for _, spec := range expandSpecFields(p) {
				key := sanitizeXMLName(spec.Key, "field")
				val := productFieldValue(p, spec.Source)
				if val == "" {
					continue
				}
				if _, err := fmt.Fprintf(w, "    <%s>%s</%s>\n", key, xmlEscape(val), key); err != nil {
					return err
				}
			}
			continue
		}
		key := sanitizeXMLName(field.Key, "field")
		val := productFieldValue(p, field.Source)
		if val == "" {
			continue
		}
		if _, err := fmt.Fprintf(w, "    <%s>%s</%s>\n", key, xmlEscape(val), key); err != nil {
			return err
		}
	}
	_, err := fmt.Fprintf(w, "  </%s>\n", item)
	return err
}

func streamCSV(w io.Writer, seq exportProductSeq, tpl exportTemplate) (int, error) {
	cw := csv.NewWriter(w)
	headers := make([]string, len(tpl.Fields))
	for i, f := range tpl.Fields {
		headers[i] = f.Key
		if headers[i] == "" {
			headers[i] = f.Source
		}
	}
	if err := cw.Write(headers); err != nil {
		return 0, err
	}
	count := 0
	record := make([]string, len(tpl.Fields))
	err := seq(func(p exportProduct) error {
		for i, field := range tpl.Fields {
			record[i] = productFieldValue(p, field.Source)
		}
		if err := cw.Write(record); err != nil {
			return err
		}
		count++
		if count%exportChunkHint == 0 {
			cw.Flush()
		}
		return nil
	})
	cw.Flush()
	if err != nil {
		return count, err
	}
	if err := cw.Error(); err != nil {
		return count, err
	}
	return count, nil
}

// UpdateExportFeedTemplate stores template/filters JSON for a company export feed.
func (s *Service) UpdateExportFeedTemplate(ctx context.Context, companyID, id uuid.UUID, template, filters any) (map[string]any, error) {
	tplBytes, err := json.Marshal(template)
	if err != nil {
		return nil, err
	}
	if template == nil {
		tplBytes = []byte("{}")
	}
	filterBytes, err := json.Marshal(filters)
	if err != nil {
		return nil, err
	}
	if filters == nil {
		filterBytes = []byte("{}")
	}
	ct, err := s.Pool.Exec(ctx, `
		UPDATE export_feeds
		SET template = $3::jsonb, filters = $4::jsonb, updated_at = now()
		WHERE id = $1 AND company_id = $2`, id, companyID, tplBytes, filterBytes)
	if err != nil {
		return nil, err
	}
	if ct.RowsAffected() == 0 {
		return nil, errors.New("not found")
	}
	return map[string]any{"id": id, "template": template, "filters": filters}, nil
}

// ExportSelectedProducts renders XML/CSV for the given processed product IDs using the export feed template.
func (s *Service) ExportSelectedProducts(ctx context.Context, companyID, feedID uuid.UUID, productIDs []uuid.UUID) (filename, mimeType string, content []byte, count int, err error) {
	if len(productIDs) == 0 {
		return "", "", nil, 0, ClientMsg("product_ids is required")
	}
	if len(productIDs) > exportSelectedMaxProducts {
		return "", "", nil, 0, ClientMsg(fmt.Sprintf("at most %d product_ids allowed", exportSelectedMaxProducts))
	}
	row, err := s.loadExportFeedByID(ctx, companyID, feedID)
	if err != nil {
		return "", "", nil, 0, err
	}
	if !row.IsActive {
		return "", "", nil, 0, ClientMsg("export feed inactive")
	}
	format := strings.ToLower(row.Format)
	if format == "" {
		format = "xml"
	}
	rows, err := s.queryExportProductsByIDs(ctx, companyID, productIDs)
	if err != nil {
		return "", "", nil, 0, err
	}
	defer rows.Close()

	tpl := parseExportTemplate(row.Template)
	seq := rowsToExportSeq(rows)
	var buf bytes.Buffer
	switch format {
	case "csv":
		count, err = streamCSV(&buf, seq, tpl)
		mimeType = "text/csv; charset=utf-8"
	default:
		count, err = streamXML(&buf, seq, tpl, row.Name)
		mimeType = "application/xml; charset=utf-8"
		format = "xml"
	}
	if err != nil {
		return "", "", nil, count, err
	}
	if count == 0 {
		return "", "", nil, 0, ClientMsg("selected products could not be found or are not exportable")
	}
	_ = s.touchLastGenerated(ctx, row.ID)
	safe := sanitizeExportFileName(row.Name)
	filename = fmt.Sprintf("%s-selected-%d.%s", safe, count, format)
	return filename, mimeType, buf.Bytes(), count, nil
}

func (s *Service) queryExportProductsByIDs(ctx context.Context, companyID uuid.UUID, productIDs []uuid.UUID) (pgx.Rows, error) {
	return s.Pool.Query(ctx, `
		SELECT p.product_id, p.name, p.category, p.description, p.processed_description, p.processed_name,
			p.status, p.attributes, p.processed_attributes,
			COALESCE(r.mapped_data, '{}'::jsonb), p.feed_id
		FROM processed_products p
		LEFT JOIN raw_products r ON r.id = p.raw_product_id AND r.company_id = p.company_id
		WHERE p.company_id = $1
			AND (p.id = ANY($2::uuid[]) OR p.raw_product_id = ANY($2::uuid[]))
		ORDER BY p.updated_at DESC
		LIMIT $3`,
		companyID, productIDs, exportSelectedMaxProducts,
	)
}

// renderExportSnippet builds an in-memory XML or CSV snippet for products + template (no DB).
func renderExportSnippet(format string, tpl exportTemplate, products []exportProduct, feedName string) (string, int, error) {
	var buf bytes.Buffer
	switch strings.ToLower(strings.TrimSpace(format)) {
	case "csv":
		cw := csv.NewWriter(&buf)
		headers := make([]string, len(tpl.Fields))
		for i, f := range tpl.Fields {
			headers[i] = f.Key
			if headers[i] == "" {
				headers[i] = f.Source
			}
		}
		if err := cw.Write(headers); err != nil {
			return "", 0, err
		}
		record := make([]string, len(tpl.Fields))
		for i, p := range products {
			for j, field := range tpl.Fields {
				record[j] = productFieldValue(p, field.Source)
			}
			if err := cw.Write(record); err != nil {
				return buf.String(), i, err
			}
		}
		cw.Flush()
		return buf.String(), len(products), cw.Error()
	default:
		root := sanitizeXMLName(tpl.Root, defaultExportRoot)
		item := sanitizeXMLName(tpl.Item, defaultExportItem)
		if _, err := fmt.Fprintf(&buf, "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<%s feed=\"%s\">\n",
			root, xmlEscape(feedName)); err != nil {
			return "", 0, err
		}
		for i, p := range products {
			if err := writeXMLProduct(&buf, item, tpl, p); err != nil {
				return buf.String(), i, err
			}
		}
		if _, err := fmt.Fprintf(&buf, "</%s>\n", root); err != nil {
			return buf.String(), len(products), err
		}
		return buf.String(), len(products), nil
	}
}

func sanitizeExportFileName(name string) string {
	name = strings.TrimSpace(name)
	if name == "" {
		return "export"
	}
	var b strings.Builder
	for _, r := range strings.ToLower(name) {
		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '-' || r == '_' {
			b.WriteRune(r)
		} else if r == ' ' {
			b.WriteByte('_')
		}
	}
	out := b.String()
	if out == "" {
		return "export"
	}
	return out
}