go-cart-actor/pkg/cart/recovery/scanner.go

package recovery

import (
	"context"
	"fmt"
	"net/mail"
	"os"
	"strconv"
	"strings"
	"time"

	"git.k6n.net/mats/go-cart-actor/pkg/actor"
	"git.k6n.net/mats/go-cart-actor/pkg/cart"
	cart_messages "git.k6n.net/mats/go-cart-actor/proto/cart"
)

// Scanner finds RecoveryCandidates by scanning the per-pod event-log directory.
//
// The model:
//   - Each cart pod owns a shard of `<id>.events.log` files under `dir`.
//   - File mtime approximates "last successful mutation" cheaply — replay only
//     when something lands in the window.
//   - For each candidate file, replay the grain (a few kb, ms-scale) and build
//     a RecoveryCandidate iff it has items + a contact.
//
// Two important implementation details:
//
//  1. After LoadEvents we MUST call UpdateTotals(). Replaying event handlers
//     runs the per-mutation registered handlers (SetUserId, SetRecoveryContact,
//     AddItem, ...) but NOT the post-mutation processors — those run only on
//     the live mutation path (see cmd/cart/main.go::RegisterProcessor for the
//     "Totals and promotions" pipeline). Without this call, g.TotalPrice is
//     nil and TotalIncVat in the candidate is always 0.
//
//  2. Read-vs-write races: DiskStorage may be in the middle of appending while
//     Scan reads. The bufio scanner will fail JSON-unmarshal on a torn line and
//     we silently continue — LogEvents treats that as "skip", here we do too.
//     A persistent unreadable file is a separate problem runPurge handles.
//
// v0 deliberately does NOT track "already notified" within this scanner —
// the window sizing plus caller conventions give roughly-once delivery.
// Idempotency is the next pass.
type Scanner struct {
	dir     string
	storage *actor.DiskStorage[cart.CartGrain]
}

// NewScanner returns a Scanner over dir (the same CART_DIR used by the cart
// service). storage must share the same dir so LoadEvents reading paths
// match the scanning paths.
func NewScanner(dir string, storage *actor.DiskStorage[cart.CartGrain]) *Scanner {
	return &Scanner{dir: dir, storage: storage}
}

// Scan returns candidates whose event-log mtime falls in
// [now-delay-window, now-delay] AND whose grain has items AND (email OR
// push tokens). The half-open window semantics match the caller's intent:
//
//	"carts whose mtime is in this slice of the recent past"
//
// — younger than now-delay (still active) and older than now-delay-window
// (never recovery-eligible yet).
//
// Empty slice + nil error means "no candidates this pass"; an error means the
// pass itself failed and the caller should log + retry.
func (s *Scanner) Scan(ctx context.Context, delay, window time.Duration, now time.Time) ([]RecoveryCandidate, error) {
	if s == nil || s.storage == nil || s.dir == "" {
		return nil, fmt.Errorf("recovery: scanner not configured")
	}
	if delay <= 0 {
		return nil, nil
	}
	if window <= 0 {
		window = delay
	}

	lower := now.Add(-(delay + window))
	upper := now.Add(-delay)
	entries, err := os.ReadDir(s.dir)
	if err != nil {
		return nil, fmt.Errorf("recovery: read dir %s: %w", s.dir, err)
	}

	out := make([]RecoveryCandidate, 0)
	for _, e := range entries {
		if e.IsDir() {
			continue
		}
		name := e.Name()
		if !strings.HasSuffix(name, ".events.log") {
			continue
		}
		id, ok := fileIDFor(name)
		if !ok {
			continue
		}
		info, infoErr := e.Info()
		if infoErr != nil {
			continue
		}
		mtime := info.ModTime()
		// Closed window:
		//   mtime BEFORE lower  → never reached the abandonment threshold yet.
		//   mtime AFTER upper   → still too fresh (caller is actively using it).
		if mtime.Before(lower) || mtime.After(upper) {
			continue
		}

		grain := cart.NewCartGrain(id, mtime)
		if loadErr := s.storage.LoadEvents(ctx, id, grain); loadErr != nil {
			// Skip unreadable grains — they're purged separately by runPurge.
			continue
		}
		// Replay does NOT run the Totals-and-promotions processor (see the
		// package doc); recompute TotalPrice from the item list so the
		// candidate TotalIncVat matches what the live API would return.
		grain.UpdateTotals()
		if !eligible(grain) {
			continue
		}
		out = append(out, buildCandidate(id, grain, mtime))
	}
	return out, nil
}

// eligible mirrors "would we ever want to send a reminder?" — items present
// and a contact available. The grain's own UpdateTotals has already been
// run by every mutation's processor, so TotalPrice is reliable here.
//
// Wishlists / offers are excluded — a wishlist cart isn't forgotten, and an
// offer cart has its own delivery semantics.
func eligible(g *cart.CartGrain) bool {
	if g == nil || len(g.Items) == 0 {
		return false
	}
	switch g.Type {
	case cart_messages.CartType_WISHLIST, cart_messages.CartType_OFFER:
		return false
	}
	hasContact := strings.TrimSpace(g.Email) != "" || len(g.PushTokens) > 0
	return hasContact
}

// buildCandidate projects the replayed grain into a RecoveryCandidate. Email
// validation is intentionally lenient — we hand the value to the notifier
// and the notifier decides what to send; rejecting "looks-invalid" addresses
// here would just hide useful debugging signal from the dry-run logs.
func buildCandidate(id uint64, g *cart.CartGrain, mtime time.Time) RecoveryCandidate {
	var total int64
	if g.TotalPrice != nil {
		// tax.Price.IncVat is int64-backed via money.Cents. Reading the field
		// directly keeps the recovery layer decoupled from platform/money
		// internals.
		total = int64(g.TotalPrice.IncVat)
	}
	tokens := make([]cart.PushToken, len(g.PushTokens))
	copy(tokens, g.PushTokens)
	return RecoveryCandidate{
		CartID:         cart.CartId(id),
		UserID:         g.UserId,
		Email:          g.Email,
		PushTokens:     tokens,
		ItemCount:      len(g.Items),
		TotalIncVat:    total,
		Currency:       g.Currency,
		LastChangeUnix: mtime.Unix(),
	}
}

// IsValidEmail is exported as a convenience the notifier may use to decide
// whether to actually send (MailerSend will reject invalid); the scanner
// itself is permissive.
func IsValidEmail(s string) bool {
	_, err := mail.ParseAddress(s)
	return err == nil
}

// fileIDFor is a small helper that extracts the grain id from an .events.log
// filename. Returns 0/false if the name doesn't match the expected shape.
func fileIDFor(name string) (uint64, bool) {
	base := strings.TrimSuffix(name, ".events.log")
	id, err := strconv.ParseUint(base, 10, 64)
	if err != nil || id == 0 {
		return 0, false
	}
	return id, true
}