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

package feeds

import (
	"encoding/json"
	"strings"
)

// Reserved mapped_data key holding seller-facing change flags from the latest
// sync that modified the row. Cleared/replaced on the next content update.
const syncChangesMappedKey = "_sync_changes"

// Seller-high-value mapped field groups compared during feed upserts.
var (
	syncPriceFields = []string{"price", "sale_price", "purchase_price"}
	syncStockFields = []string{"stock", "quantity", "qty"}
	syncAvailFields = []string{"availability", "stock_status", "in_stock"}
	syncTitleFields = []string{"title", "name", "product_name"}
)

// syncDeltaCounts is the MVP seller summary written to input_feeds.options.last_sync_deltas.
type syncDeltaCounts struct {
	JobID                string `json:"job_id,omitempty"`
	New                  int    `json:"new"`
	PriceChanged         int    `json:"price_changed"`
	StockChanged         int    `json:"stock_changed"`
	AvailabilityChanged  int    `json:"availability_changed"`
	TitleChanged         int    `json:"title_changed"`
	OtherChanged         int    `json:"other_changed"`
	Unchanged            int    `json:"unchanged"`
	Skipped              int    `json:"skipped"`
}

func (d *syncDeltaCounts) addChanges(changes []string) {
	if len(changes) == 0 {
		d.OtherChanged++
		return
	}
	seen := map[string]struct{}{}
	for _, c := range changes {
		if _, ok := seen[c]; ok {
			continue
		}
		seen[c] = struct{}{}
		switch c {
		case "new":
			d.New++
		case "price":
			d.PriceChanged++
		case "stock":
			d.StockChanged++
		case "availability":
			d.AvailabilityChanged++
		case "title":
			d.TitleChanged++
		default:
			d.OtherChanged++
		}
	}
}

func (d syncDeltaCounts) asMap() map[string]any {
	return map[string]any{
		"job_id":                d.JobID,
		"new":                   d.New,
		"price_changed":         d.PriceChanged,
		"stock_changed":         d.StockChanged,
		"availability_changed":  d.AvailabilityChanged,
		"title_changed":         d.TitleChanged,
		"other_changed":         d.OtherChanged,
		"unchanged":             d.Unchanged,
		"skipped":               d.Skipped,
	}
}

func mappedScalar(m map[string]any, key string) string {
	if m == nil {
		return ""
	}
	v, ok := m[key]
	if !ok || v == nil {
		return ""
	}
	switch t := v.(type) {
	case string:
		return strings.TrimSpace(t)
	case float64:
		return strings.TrimSpace(strings.TrimRight(strings.TrimRight(
			strings.ReplaceAll(jsonNumber(t), "e+0", "e+"), "0"), "."))
	case json.Number:
		return strings.TrimSpace(t.String())
	case bool:
		if t {
			return "true"
		}
		return "false"
	default:
		b, err := json.Marshal(t)
		if err != nil {
			return ""
		}
		return strings.TrimSpace(string(b))
	}
}

func jsonNumber(f float64) string {
	b, err := json.Marshal(f)
	if err != nil {
		return ""
	}
	return string(b)
}

func fieldGroupChanged(oldM, newM map[string]any, keys []string) bool {
	for _, k := range keys {
		if mappedScalar(oldM, k) != mappedScalar(newM, k) {
			return true
		}
	}
	return false
}

// detectMappedChanges compares prior mapped_data JSON to the new mapped map.
// Returns change kind tags: price, stock, availability, title, other.
func detectMappedChanges(oldJSON []byte, newMapped map[string]any) []string {
	var oldM map[string]any
	if len(oldJSON) > 0 {
		_ = json.Unmarshal(oldJSON, &oldM)
	}
	if oldM == nil {
		oldM = map[string]any{}
	}
	cleanNew := stripSyncChanges(newMapped)
	var out []string
	if fieldGroupChanged(oldM, cleanNew, syncPriceFields) {
		out = append(out, "price")
	}
	if fieldGroupChanged(oldM, cleanNew, syncStockFields) {
		out = append(out, "stock")
	}
	if fieldGroupChanged(oldM, cleanNew, syncAvailFields) {
		out = append(out, "availability")
	}
	if fieldGroupChanged(oldM, cleanNew, syncTitleFields) {
		out = append(out, "title")
	}
	// Any other mapped key change (excluding reserved meta).
	if len(out) == 0 && !mappedEqualIgnoringSyncMeta(oldM, cleanNew) {
		out = append(out, "other")
	}
	return out
}

func stripSyncChanges(m map[string]any) map[string]any {
	if m == nil {
		return map[string]any{}
	}
	out := make(map[string]any, len(m))
	for k, v := range m {
		if k == syncChangesMappedKey {
			continue
		}
		out[k] = v
	}
	return out
}

// preserveSyncedExtras keeps fields that feed mappings never set (notably
// category unique_id from legacy assignment / seed backfill) when a sync
// remaps the row. Without this, upserts replace mapped_data wholesale and
// wipe category even though the feed has no category column.
func preserveSyncedExtras(existingJSON []byte, mapped map[string]any) map[string]any {
	out := stripSyncChanges(mapped)
	if len(existingJSON) == 0 {
		return out
	}
	var old map[string]any
	if err := json.Unmarshal(existingJSON, &old); err != nil || old == nil {
		return out
	}
	old = stripSyncChanges(old)
	if mappedScalar(out, "category") == "" {
		if cat := mappedScalar(old, "category"); cat != "" && !strings.EqualFold(cat, "none") {
			out["category"] = cat
		}
	}
	return out
}

func mappedEqualIgnoringSyncMeta(a, b map[string]any) bool {
	aa := stripSyncChanges(a)
	bb := stripSyncChanges(b)
	ab, err1 := json.Marshal(aa)
	bb2, err2 := json.Marshal(bb)
	if err1 != nil || err2 != nil {
		return false
	}
	return bytesEqualJSON(ab, bb2)
}

func withSyncChanges(mapped map[string]any, changes []string) map[string]any {
	out := stripSyncChanges(mapped)
	if len(changes) == 0 {
		return out
	}
	tags := make([]any, 0, len(changes))
	seen := map[string]struct{}{}
	for _, c := range changes {
		c = strings.TrimSpace(strings.ToLower(c))
		if c == "" {
			continue
		}
		if _, ok := seen[c]; ok {
			continue
		}
		seen[c] = struct{}{}
		tags = append(tags, c)
	}
	if len(tags) > 0 {
		out[syncChangesMappedKey] = tags
	}
	return out
}