pangolin/pay/internal/watcher/watcher.go

// 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)
			}
		}
	}
}