go-cart-actor/cmd/cart/main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"net/http/pprof"
	"os"
	"os/signal"
	"strings"
	"time"

	"git.k6n.net/mats/go-cart-actor/internal/ucp"
	"git.k6n.net/mats/go-cart-actor/pkg/actor"
	"git.k6n.net/mats/go-cart-actor/pkg/cart"
	cartmcp "git.k6n.net/mats/go-cart-actor/pkg/cart/mcp"
	"git.k6n.net/mats/go-cart-actor/pkg/cart/recovery"
	"git.k6n.net/mats/go-cart-actor/pkg/promotions"
	"git.k6n.net/mats/go-cart-actor/pkg/proxy"
	"git.k6n.net/mats/go-cart-actor/pkg/telemetry"
	"git.k6n.net/mats/go-cart-actor/pkg/voucher"
	"git.k6n.net/mats/go-redis-inventory/pkg/inventory"
	"git.k6n.net/mats/platform/catalog"
	"git.k6n.net/mats/platform/config"
	"git.k6n.net/mats/platform/event"
	"git.k6n.net/mats/platform/rabbit"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/redis/go-redis/v9"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

var (
	// cartMetrics is the shared prometheus instrumentation for the
	// cart actor pool and event log. Built once at process start; nil
	// is not expected in production but the actor package's
	// touch sites are nil-safe so a test binary can pass nil.
	cartMetrics = actor.NewMetrics("cart")
)

func init() {
	os.Mkdir("data", 0755)
}

type App struct {
	pool   *actor.SimpleGrainPool[cart.CartGrain]
	server *PoolServer
}

var podIp = os.Getenv("POD_IP")
var name = os.Getenv("POD_NAME")
var amqpUrl = os.Getenv("AMQP_URL")
var redisAddress = os.Getenv("REDIS_ADDRESS")
var redisPassword = os.Getenv("REDIS_PASSWORD")

// normalizeListenAddr accepts either a bare port ("8080") or a full listen
// address (":8080", "0.0.0.0:8080") and returns a value usable by http.Server.Addr.
func normalizeListenAddr(v string) string {
	if strings.Contains(v, ":") {
		return v
	}
	return ":" + v
}

// loadUCPCartSigner loads the UCP ECDSA signing key from the path specified in
// the UCP_SIGNING_KEY_PATH environment variable. Returns nil if unset or unreadable.
func loadUCPCartSigner() *ucp.SigningConfig {
	path := os.Getenv("UCP_SIGNING_KEY_PATH")
	if path == "" {
		return nil
	}
	pemData, err := os.ReadFile(path)
	if err != nil {
		log.Printf("ucp signing: cannot read key %s: %v", path, err)
		return nil
	}
	cfg, err := ucp.NewSigningConfigFromPEM(pemData, "k6n-ecdsa-2026")
	if err != nil {
		log.Printf("ucp signing: invalid key at %s: %v", path, err)
		return nil
	}
	return cfg
}

func getCountryFromHost(host string) string {
	if strings.Contains(strings.ToLower(host), "-no") {
		return "no"
	}
	if strings.Contains(strings.ToLower(host), "-se") {
		return "se"
	}
	return "se"
}

type MutationContext struct {
	VoucherService voucher.Service
}

type CartChangeEvent struct {
	CartId    cart.CartId         `json:"cartId"`
	Mutations []actor.ApplyResult `json:"mutations"`
}

// catalogProjectionCache is the cart's per-pod, cold-start-ready catalog
// projection cache, backed by platform/catalog.ProjectionStore. It consumes the
// catalog.projection_published wire — a full snapshot (begin/chunk/end, framed by
// epoch) plus incremental deltas — into its OWN local .bin and serves catalog
// facts by SKU (price, tax_class, inventory_tracked, image, …) at add-to-cart /
// checkout-reserve decisions. Stock is NOT cached — queried synchronously from
// inventory per docs/inventory-shape.md.
//
// Clustering: the cart runs N replicas. Each pod has its OWN exclusive bus queue
// (rabbit.ListenToPattern declares an exclusive server-named queue, so every pod
// receives the full stream) and its OWN pod-local .bin. It's a replicated read
// cache — no leader, no shared volume; N pods writing one file would corrupt it.
// See docs/catalog-projection.md.
//
// The cart stores the full catalog.Projection (identity transform): its
// add-to-cart overlay (projection_overlay.go) consumes nearly every field, so
// there's nothing to subset. Deletes ride as tombstones in the store's overlay
// (IsDeleted) so a removed SKU isn't re-fetched before the next snapshot folds
// the delete into the base. Cold-grace: an SKU not yet in the base/overlay
// resolves via the existing PRODUCT_BASE_URL HTTP fetch + overlay.
type catalogProjectionCache struct {
	store *catalog.ProjectionStore[catalog.Projection]
}

// identityProjection is the cart's mandatory transform — it stores the full
// Projection because the overlay uses almost all of it.
func identityProjection(p catalog.Projection) catalog.Projection { return p }

// newCatalogProjectionCache opens the per-pod store under dir. dir MUST be
// pod-local (container fs / an emptyDir in k8s) — never a shared volume.
func newCatalogProjectionCache(dir string) (*catalogProjectionCache, error) {
	s, err := catalog.NewProjectionStore(dir, "cart-projection.bin", identityProjection, nil)
	if err != nil {
		return nil, err
	}
	return &catalogProjectionCache{store: s}, nil
}

// HandleFrame feeds one catalog.projection_published event (snapshot frame or
// delta) into the store; framing/epoch ordering live in platform/catalog.
func (c *catalogProjectionCache) HandleFrame(meta map[string]string, payload []byte) error {
	return c.store.HandleFrame(meta, payload)
}

func (c *catalogProjectionCache) Get(sku string) (catalog.Projection, bool) { return c.store.Get(sku) }

// IsDeleted reports whether the bus has explicitly deleted sku (overlay
// tombstone) — call sites skip the HTTP fallback for it.
func (c *catalogProjectionCache) IsDeleted(sku string) bool { return c.store.IsDeleted(sku) }

// Len reports cached entries (base + overlay) — debug/log gauge only.
func (c *catalogProjectionCache) Len() int {
	_, base, overlay := c.store.Stats()
	return base + overlay
}

func main() {

	// cartPort is the bare HTTP port. It drives both the local listener and the
	// port used to proxy to peer pods, so a cluster must run all pods on the
	// same CART_PORT.
	cartPort := config.EnvString("CART_PORT", "8080")
	if i := strings.LastIndex(cartPort, ":"); i >= 0 {
		cartPort = cartPort[i+1:]
	}

	controlPlaneConfig := actor.DefaultServerConfig()

	promotionsPath := os.Getenv("PROMOTIONS_FILE")
	if promotionsPath == "" {
		promotionsPath = "data/promotions.json"
	}
	promotionStore, err := promotions.NewStore(promotionsPath)
	if err != nil {
		log.Fatalf("Error loading promotions: %v\n", err)
	}

	log.Printf("loaded %d promotions", len(promotionStore.List()))

	promotionService := promotions.NewPromotionService(nil)

	//inventoryPubSub := actor.NewPubSub[inventory.InventoryChange]()

	rdb := redis.NewClient(&redis.Options{
		Addr:     redisAddress,
		Password: redisPassword,
		DB:       0,
	})
	inventoryReservationService, err := inventory.NewRedisCartReservationService(rdb)
	if err != nil {
		log.Fatalf("Error creating inventory reservation service: %v\n", err)
	}
	reservationPolicy := inventory.NewReservationPolicyAdapter(inventoryReservationService)

	reg := cart.NewCartMultationRegistry(cart.NewCartMutationContext(reservationPolicy))
	reg.RegisterProcessor(
		actor.NewMutationProcessor(func(ctx context.Context, g *cart.CartGrain) error {
			_, span := tracer.Start(ctx, "Totals and promotions")
			defer span.End()

			// Clear bypass flags first
			for _, v := range g.Vouchers {
				if v != nil {
					v.BypassedByPromotions = false
				}
			}

			g.UpdateTotals()
			promotionCtx := promotions.NewContextFromCart(g, promotions.WithNow(time.Now()))
			results, _ := promotionService.EvaluateAll(promotionStore.Snapshot(), promotionCtx)

			// Check if any qualified promotion rule has a coupon_code condition matching our vouchers
			hasBypassed := false
			for _, res := range results {
				if res.Applicable {
					promotions.WalkConditions(res.Rule.Conditions, func(c promotions.Condition) bool {
						if bc, ok := c.(promotions.BaseCondition); ok && bc.Type == promotions.CondCouponCode {
							var codes []string
							if s, ok := bc.Value.AsString(); ok {
								codes = append(codes, strings.ToLower(s))
							} else if arr, ok := bc.Value.AsStringSlice(); ok {
								for _, s := range arr {
									codes = append(codes, strings.ToLower(s))
								}
							}
							for _, code := range codes {
								for _, v := range g.Vouchers {
									if v != nil && strings.ToLower(v.Code) == code {
										if !v.BypassedByPromotions {
											v.BypassedByPromotions = true
											hasBypassed = true
										}
									}
								}
							}
						}
						return true
					})
				}
			}

			if hasBypassed {
				// Re-evaluate with bypassed vouchers
				g.UpdateTotals()
				promotionCtx = promotions.NewContextFromCart(g, promotions.WithNow(time.Now()))
				results, _ = promotionService.EvaluateAll(promotionStore.Snapshot(), promotionCtx)
			}

			// ApplyResults applies qualifying actions in priority order and records
			// every effect — both applied discounts and pending "spend X more for ..."
			// nudges with their progress — in g.AppliedPromotions.
			promotionService.ApplyResults(g, results, promotionCtx)
			g.Version++
			return nil
		}),
	)

	diskStorage := actor.NewDiskStorage[cart.CartGrain](config.EnvString("CART_DIR", "data"), reg)
	diskStorage.SetMetrics(cartMetrics)
	poolConfig := actor.GrainPoolConfig[cart.CartGrain]{
		MutationRegistry: reg,
		Storage:          diskStorage,
		Metrics:          cartMetrics,
		Spawn: func(ctx context.Context, id uint64) (actor.Grain[cart.CartGrain], error) {
			_, span := tracer.Start(ctx, fmt.Sprintf("Spawn cart id %d", id))
			defer span.End()
			ret := cart.NewCartGrain(id, time.Now())
			// Set baseline lastChange at spawn; replay may update it to last event timestamp.

			// inventoryPubSub.Subscribe(ret.HandleInventoryChange)
			err := diskStorage.LoadEvents(ctx, id, ret)
			// if err == nil && inventoryService != nil {
			// 	refs := make([]*inventory.InventoryReference, 0)
			// 	for _, item := range ret.Items {
			// 		refs = append(refs, &inventory.InventoryReference{
			// 			SKU:        inventory.SKU(item.Sku),
			// 			LocationID: getLocationId(item),
			// 		})
			// 	}
			// 	_, span := tracer.Start(ctx, "update inventory")
			// 	defer span.End()
			// 	res, err := inventoryService.GetInventoryBatch(ctx, refs...)
			// 	if err != nil {
			// 		log.Printf("unable to update inventory %v", err)
			// 	} else {
			// 		for _, update := range res {
			// 			for _, item := range ret.Items {
			// 				if matchesSkuAndLocation(update, *item) && update.Quantity != uint32(item.Stock) {
			// 					// maybe apply an update to give visibility to the cart
			// 					item.Stock = uint16(update.Quantity)
			// 				}
			// 			}
			// 		}
			// 	}
			// }

			return ret, err
		},
		Destroy: func(grain actor.Grain[cart.CartGrain]) error {
			// cart, err := grain.GetCurrentState()
			// if err != nil {
			// 	return err
			// }
			//inventoryPubSub.Unsubscribe(cart.HandleInventoryChange)

			return nil
		},
		SpawnHost: func(host string) (actor.Host[cart.CartGrain], error) {
			return proxy.NewRemoteHost[cart.CartGrain](host, cartPort)
		},
		TTL:      5 * time.Minute,
		PoolSize: 2 * 65535,
		Hostname: podIp,
	}

	pool, err := actor.NewSimpleGrainPool(poolConfig)
	if err != nil {
		log.Fatalf("Error creating cart pool: %v\n", err)
	}

	cartMCP := cartmcp.New(pool)

	// Stream each applied mutation to the shared "mutations" exchange (routing key
	// mutation.cart) so the backoffice /commerce live feed (and other consumers)
	// see cart activity. Best-effort: if AMQP is unreachable the cart still serves.
	if amqpUrl != "" {
		if conn, derr := rabbit.Dial(amqpUrl, "cart"); derr != nil {
			log.Printf("cart: AMQP connect failed, mutation feed disabled: %v", derr)
		} else {
			listener := actor.NewAmqpListener(conn.Connection(), "cart", actor.MutationSummary)
			listener.DefineTopics()
			pool.AddListener(listener)
			log.Printf("cart: mutation feed enabled (mutation.cart)")
		}
	}

	// catalog.projection_published consumer (the cart is the canonical
	// consumer of the projection wire per docs/inventory-shape.md; inventory
	// emits level crossings back to finder only, never the reverse). Own
	// connection keeps the cart's mutation-feed lifecycle decoupled from a
	// read-write-consume connection that could grow event handlers later
	// (e.g. promotions reacting to a price drop on a watched SKU).
	// Pod-local store dir — container fs / an emptyDir in k8s. NEVER a shared
	// volume: each replica owns its own .bin (replicated read cache, no leader).
	projectionDir := config.EnvString("CART_PROJECTION_DIR", "data/projection")
	projectionCache, err := newCatalogProjectionCache(projectionDir)
	if err != nil {
		log.Fatalf("Error opening catalog projection cache: %v", err)
	}
	if amqpUrl != "" {
		if pConn, perr := rabbit.Dial(amqpUrl, "cart-catalog-projection"); perr != nil {
			log.Printf("cart: catalog projection consumer DISABLED (dial failed): %v", perr)
		} else {
			pConn.NotifyOnReconnect(func() {
				ch, err := pConn.Channel()
				if err != nil {
					log.Printf("cart: channel on projection reconnect: %v", err)
					return
				}
				if err := rabbit.ListenToPattern(ch, "catalog", string(event.CatalogProjectionPublished), catalog.MakeAmqpHandler(projectionCache.store)); err != nil {
					log.Printf("cart: bind catalog.projection_published on reconnect: %v", err)
					_ = ch.Close()
				}
			})
			defer func() { _ = pConn.Close() }()
			log.Printf("cart: catalog projection consumer ENABLED (catalog.projection_published on 'catalog' exchange)")
		}
	}

	syncedServer := NewPoolServer(pool, fmt.Sprintf("%s, %s", name, podIp), projectionCache) //inventoryService, inventoryReservationService)

	app := &App{
		pool:   pool,
		server: syncedServer,
	}

	mux := http.NewServeMux()
	debugMux := http.NewServeMux()

	grpcSrv, err := actor.NewControlServer[cart.CartGrain](controlPlaneConfig, pool)
	if err != nil {
		log.Fatalf("Error starting control plane gRPC server: %v\n", err)
	}
	defer grpcSrv.GracefulStop()

	go diskStorage.SaveLoop(20 * time.Second)
	UseDiscovery(pool)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	defer stop()

	// Data Retention (C5) GDPR Purge
	retentionDays := config.EnvInt("CART_RETENTION_DAYS", 0, func(n int) bool { return n >= 0 })
	if retentionDays > 0 {
		purgeInterval := config.EnvDuration("CART_PURGE_INTERVAL", 24*time.Hour)
		log.Printf("cart: data retention enabled. Retaining carts for %d days, purging every %v", retentionDays, purgeInterval)
		go func() {
			ticker := time.NewTicker(purgeInterval)
			defer ticker.Stop()

			// Run immediately on startup
			runPurge(ctx, diskStorage, pool, retentionDays)

			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					runPurge(ctx, diskStorage, pool, retentionDays)
				}
			}
		}()
	} else {
		log.Print("cart: data retention / GDPR purge disabled (CART_RETENTION_DAYS not set or <= 0)")
	}

	// Abandoned-cart recovery check (C3-notifications seam, dry-run by default).
	// Scans local pod event-log files for carts whose modtime falls into the
	// "[now - delay - window, now - delay]" window and that have items + a
	// contact (email and/or push tokens). Hands each candidate to the
	// configured Notifier (LoggingNotifier v0). The scanner reads local shard
	// files only — the control plane doesn't fan out between pods on purpose
	// (each pod owns the carts it is the owner of). Idempotency is the next
	// pass; the v0 window size keeps duplicate notifications minute-scoped.
	abandonedDelay := config.EnvDuration("CART_ABANDONED_DELAY", time.Hour)
	if abandonedDelay > 0 {
		abandonedWindow := config.EnvDuration("CART_ABANDONED_WINDOW", time.Hour)
		abandonedInterval := config.EnvDuration("CART_ABANDONED_SCAN_INTERVAL", 15*time.Minute)
		notifier := recovery.NewLoggingNotifier()
		scanner := recovery.NewScanner(config.EnvString("CART_DIR", "data"), diskStorage)
		log.Printf("cart: abandonment recovery ENABLED. Delay=%v window=%v scan_interval=%v (notifier=LoggingNotifier dry-run)", abandonedDelay, abandonedWindow, abandonedInterval)
		go func() {
			ticker := time.NewTicker(abandonedInterval)
			defer ticker.Stop()

			runRecovery(ctx, scanner, notifier, abandonedDelay, abandonedWindow)

			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					runRecovery(ctx, scanner, notifier, abandonedDelay, abandonedWindow)
				}
			}
		}()
	} else {
		log.Print("cart: abandonment recovery DISABLED (CART_ABANDONED_DELAY not set or <= 0)")
	}

	otelShutdown, err := telemetry.SetupOTelSDK(ctx)
	if err != nil {
		log.Fatalf("Unable to start otel %v", err)
	}

	syncedServer.Serve(mux)

	// Cart MCP edge: inspect and mutate carts — get_cart, get_cart_items,
	// update_item_quantity, remove_cart_item, clear_cart, apply_voucher, etc.
	// Exposes 11 tools on the same grain pool, mounted at /cart-mcp so it does
	// not conflict with the Promotions MCP at /promotions-mcp.
	mux.Handle("/cart-mcp", cartMCP.Handler())
	mux.Handle("/cart-mcp/", cartMCP.Handler())

	// UCP REST adapter — Universal Commerce Protocol
	cartUCP := ucp.CartHandler(pool)
	if signer := loadUCPCartSigner(); signer != nil {
		cartUCP = ucp.WithSigning(cartUCP, signer)
		log.Print("ucp signing enabled")
	}
	// StripPrefix is required because the sub-mux (CartHandler) registers
	// routes like "GET /{id}" which expect a relative path; Go's ServeMux
	// passes the full request path (e.g. /ucp/v1/carts/123) without stripping
	// the prefix, so the sub-mux would never match.
	//
	// Only the subtree pattern (/ucp/v1/carts/) is registered, NOT the exact
	// pattern (/ucp/v1/carts), because registering both causes Go's ServeMux
	// to redirect the exact match to the subtree root, and StripPrefix
	// interferes with the redirect target (producing a 307 to "/" instead
	// of "/ucp/v1/carts/"). Consumers should use the trailing-slash variant.
	mux.Handle("/ucp/v1/carts/", http.StripPrefix("/ucp/v1/carts", cartUCP))

	// Stateless promotion evaluation: POST a (possibly partial) eval context and
	// get back the resulting totals plus the applied/pending effect breakdown,
	// without creating or mutating a real cart.
	mux.HandleFunc("POST /promotions/evaluate", newPromotionEvaluateHandler(promotionStore, promotionService))

	// only for local
	mux.HandleFunc("GET /add/remote/{host}", func(w http.ResponseWriter, r *http.Request) {
		pool.AddRemote(r.PathValue("host"))
	})

	debugMux.HandleFunc("/debug/pprof/", pprof.Index)
	debugMux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
	debugMux.HandleFunc("/debug/pprof/profile", pprof.Profile)
	debugMux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
	debugMux.HandleFunc("/debug/pprof/trace", pprof.Trace)
	debugMux.Handle("/metrics", promhttp.Handler())

	// Projection cache probe (dev/verification). GET /debug/projection/{sku} →
	// the resolved projection + the bus-fed map size. map_entries counts both
	// live projections and tombstones; a found=true response confirms the bus
	// consumer delivered the SKU to this pod.
	debugMux.HandleFunc("/debug/projection/", func(w http.ResponseWriter, r *http.Request) {
		sku := strings.TrimPrefix(r.URL.Path, "/debug/projection/")
		p, found := projectionCache.Get(sku)
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{
			"sku":         sku,
			"found":       found,
			"projection":  p,
			"map_entries": projectionCache.Len(),
			"deleted":     projectionCache.IsDeleted(sku),
		})
	})
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
		// Grain pool health: simple capacity check (mirrors previous GrainHandler.IsHealthy)
		grainCount, capacity := app.pool.LocalUsage()
		if grainCount >= capacity {
			w.WriteHeader(http.StatusInternalServerError)
			w.Write([]byte("grain pool at capacity"))
			return
		}
		if !pool.IsHealthy() {
			w.WriteHeader(http.StatusInternalServerError)
			w.Write([]byte("control plane not healthy"))
			return
		}
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})
	mux.HandleFunc("/livez", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})

	mux.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("1.0.0"))
	})

	mux.HandleFunc("/openapi.json", ServeEmbeddedOpenAPI)

	httpAddr := normalizeListenAddr(cartPort)
	debugAddr := normalizeListenAddr(config.EnvString("CART_DEBUG_PORT", "8081"))

	srv := &http.Server{
		Addr:        httpAddr,
		BaseContext: func(net.Listener) context.Context { return ctx },
		ReadTimeout: 10 * time.Second,
		// Close idle keep-alive connections so a load test with many short-lived
		// clients doesn't pin file handles open indefinitely.
		IdleTimeout:  60 * time.Second,
		WriteTimeout: 20 * time.Second,
		Handler:      otelhttp.NewHandler(mux, "/"),
	}

	defer func() {

		fmt.Println("Shutting down due to signal")
		otelShutdown(context.Background())
		diskStorage.Close()
		pool.Close()

	}()

	srvErr := make(chan error, 1)
	go func() {
		srvErr <- srv.ListenAndServe()
	}()

	// Inventory change consumption used to live here over the bare Redis
	// `inventory_changed` channel; it is now owned by the cart-inventory
	// service, which translates quantity changes into inventory.level_changed
	// bus crossings. The cart reads exact stock synchronously when it needs it.
	// See docs/inventory-shape.md.

	log.Printf("Server started at %s (debug %s)", httpAddr, debugAddr)

	go http.ListenAndServe(debugAddr, debugMux)

	select {
	case err = <-srvErr:
		// Error when starting HTTP server.
		log.Fatalf("Unable to start server: %v", err)
	case <-ctx.Done():
		// Wait for first CTRL+C.
		// Stop receiving signal notifications as soon as possible.
		stop()
	}

}

func runPurge(ctx context.Context, diskStorage *actor.DiskStorage[cart.CartGrain], pool *actor.SimpleGrainPool[cart.CartGrain], retentionDays int) {
	log.Printf("cart: starting data retention purge...")
	retention := time.Duration(retentionDays) * 24 * time.Hour

	localIds := make(map[uint64]bool)
	for _, id := range pool.GetLocalIds() {
		localIds[id] = true
	}
	isGrainActive := func(id uint64) bool {
		return localIds[id]
	}

	purged, err := diskStorage.PurgeOlderThan(ctx, retention, isGrainActive)
	if err != nil {
		log.Printf("cart: data retention purge failed: %v", err)
	} else {
		log.Printf("cart: data retention purge completed. Purged %d expired cart logs", purged)
	}
}

// runRecovery is the abandoned-cart scanner pass. It finds carts that crossed
// the abandonment threshold in the recent window and hands each to the
// configured Notifier. Each pass is independent; idempotency across ticks is
// the next pass (logged duplicates are not catastrophic for the dry-run
// notifier, but a real MailerSend/FCM impl will need a SETNX side-key).
func runRecovery(ctx context.Context, scanner *recovery.Scanner, notifier recovery.Notifier, delay, window time.Duration) {
	candidates, err := scanner.Scan(ctx, delay, window, time.Now())
	if err != nil {
		log.Printf("cart: recovery scan failed: %v", err)
		return
	}
	if len(candidates) == 0 {
		log.Printf("cart: recovery scan complete, 0 candidates (window delay=%v width=%v)", delay, window)
		return
	}
	log.Printf("cart: recovery scan complete, %d candidates", len(candidates))
	for _, c := range candidates {
		if err := notifier.NotifyAbandoned(ctx, c); err != nil {
			log.Printf("cart: recovery notify (cart %d) failed: %v", c.CartID, err)
		}
	}
}