// Package watcher polls TronGrid for incoming USDT to the single receiving // address and matches each confirmed payment to a pending order by exact amount // + block time. It only reads the chain and flips order state — it never holds // keys or moves funds (sweeping is a separate offline step). package watcher import ( "context" "log/slog" "time" "github.com/wangjia/pangolin/pay/internal/store" "github.com/wangjia/pangolin/pay/internal/tron" ) type Watcher struct { st *store.Store tron tron.Fetcher address string log *slog.Logger now func() time.Time } func New(st *store.Store, f tron.Fetcher, address string, log *slog.Logger) *Watcher { if log == nil { log = slog.Default() } return &Watcher{st: st, tron: f, address: address, log: log, now: time.Now} } // Tick: // 1. expire overdue pending orders; // 2. fetch confirmed incoming USDT transfers to the single receiving address; // 3. match each transfer to a pending order by **exact amount** and **block time // after the order was created**; a confirmed transfer that matches no active // order (wrong amount / late after reuse) is recorded as an orphan. // // Idempotent: a tx already matched to an order or recorded as orphan is skipped. func (w *Watcher) Tick(ctx context.Context) error { if n, err := w.st.MarkExpired(ctx, w.now()); err != nil { return err } else if n > 0 { w.log.Info("orders expired", "count", n) } transfers, err := w.tron.IncomingTransfers(ctx, w.address) if err != nil { w.log.Warn("fetch transfers failed", "err", err) return nil // transient (rate limit / network); retried next tick } if len(transfers) == 0 { return nil } pending, err := w.st.ListPending(ctx) if err != nil { return err } // Index pending orders by their unique expect amount. byAmount := make(map[int64]*store.Order, len(pending)) for _, o := range pending { byAmount[o.ExpectAmount] = o } for _, t := range transfers { handled, err := w.st.TxHandled(ctx, t.TxID) if err != nil { w.log.Error("tx handled check", "tx", t.TxID, "err", err) continue } if handled { continue // already matched or already an orphan } if o := byAmount[t.Value]; o != nil && t.BlockTs > o.CreatedAt.Unix() { ok, err := w.st.MarkPaid(ctx, o.OrderNo, t.TxID) if err != nil { w.log.Error("mark paid", "order", o.OrderNo, "err", err) continue } if ok { w.log.Info("order paid", "order", o.OrderNo, "tx", t.TxID, "value", t.Value) delete(byAmount, t.Value) // a second transfer of the same amount can't reuse this order } continue } // Confirmed payment matching no active order -> orphan (needs reconciliation). if err := w.st.RecordOrphan(ctx, t.TxID, w.address, t.Value, t.BlockTs, w.now()); err != nil { w.log.Error("record orphan", "tx", t.TxID, "err", err) continue } w.log.Warn("orphan payment", "tx", t.TxID, "value", t.Value, "block_ts", t.BlockTs) } return nil } // Loop runs Tick every interval until ctx is cancelled. func (w *Watcher) Loop(ctx context.Context, interval time.Duration) { t := time.NewTicker(interval) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: if err := w.Tick(ctx); err != nil { w.log.Error("watcher tick", "err", err) } } } }