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 `.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 }