maker/seeder.go

package maker

import (
	"bytes"
	"context"
	"fmt"
	"log/slog"
	"math"
	"sort"
	"sync"
	"time"
)

// roundTo8 rounds a price/size to the 8-decimal DEX grid.
func roundTo8(f float64) float64 { return math.Round(f*1e8) / 1e8 }

// market submits orders to a DEX orderbook for a single symbol.
// DEXClient satisfies it; tests can substitute an in-process implementation.
type market interface {
	PlaceLimit(ctx context.Context, symbol, side string, price, size float64) (uint64, error)
	Cancel(ctx context.Context, orderID uint64) error
	BestBid(ctx context.Context, symbol string) (price, size float64, err error)
	BestAsk(ctx context.Context, symbol string) (price, size float64, err error)
	// OpenMarket binds the symbol's (base, quote) asset ids (dex_open_market).
	OpenMarket(ctx context.Context, symbol string, base, quote [32]byte) error
	// Deposit credits the maker's available balance of asset (dex_deposit).
	Deposit(ctx context.Context, asset [32]byte, amount uint64) error
	// Submit crosses the book with a taker order and returns the fills the
	// matcher produced. Callers that need to know what actually traded read
	// these rather than scanning chain logs after the fact.
	Submit(ctx context.Context, symbol, side string, isMarket bool, limitPrice, size float64) ([]Fill, error)
}

// poolBinder is the optional capability to pin a symbol's venue poolId.
// *DEXClient satisfies it; seedPair binds a pair's explicit PoolID through it so
// the maker seeds the exact on-chain market a deployed V4 PoolKey resolves to.
type poolBinder interface {
	BindPool(symbol string, poolID [32]byte)
}

// restingCanceller is the optional capability to cancel the maker's own resting
// orders on a symbol (startup reconcile). *DEXClient satisfies it; an in-process
// test market need not.
type restingCanceller interface {
	CancelRestingForSymbol(ctx context.Context, symbol string) (int, error)
}

// MidProvider supplies the live EXTERNAL reference mid for a symbol — the price
// this market should track. *Aggregator satisfies it. It is orthogonal to the DEX
// venue: the venue is WHERE the maker quotes, the MidProvider is WHAT it tracks
// (the native AMM for LUX/*, a CEX for listed majors). Composed, not braided.
type MidProvider interface {
	GetMidPrice(ctx context.Context, symbol string) (float64, error)
}

// Seeder creates markets on the DEX and maintains a 2-sided orderbook for a
// configured set of pairs (e.g. LUX/LUSD, LUX/LETH). It is the entity that
// seeds initial liquidity: on Seed it places the first 2-sided quotes; on
// each Maintain cycle it cancels the prior quotes and re-places fresh ones
// around the current mid.
type Seeder struct {
	cfg      Config
	dex      market
	strategy Strategy
	feed     MidProvider // optional live external reference; nil = track DEX book / reference mid
	logger   *slog.Logger

	mu     sync.Mutex
	live   map[string][]uint64 // symbol -> active order ids
	stopCh chan struct{}
}

// UseFeed installs the live external price reference. With a feed, every seed and
// requote tracks the feed's mid for each symbol (the maker keeps the book pinned
// to whatever is live); without one, it tracks the DEX book mid then the static
// reference. Call once at setup, before Run. Matches DEXClient.UseSigner.
func (s *Seeder) UseFeed(f MidProvider) { s.feed = f }

// NewSeeder builds a seeder. strategy may be nil, in which case a symmetric
// MidSpreadStrategy is used.
func NewSeeder(cfg Config, dex market, strategy Strategy, logger *slog.Logger) *Seeder {
	if strategy == nil {
		strategy = &MidSpreadStrategy{}
	}
	return &Seeder{
		cfg:      cfg,
		dex:      dex,
		strategy: strategy,
		logger:   logger,
		live:     make(map[string][]uint64),
		stopCh:   make(chan struct{}),
	}
}

// Seed opens and funds every configured market, then places the initial 2-sided
// quotes. Funding happens FIRST and is aggregated PER ASSET across all pairs (a
// deposit is keyed by (user, asset, amount), so two markets that share the native
// LUX base are funded by one summed LUX deposit — depositing per-market-leg would
// make the second leg's byte-identical deposit dedup to a no-op and starve it).
// Returns the number of orders placed across all pairs.
func (s *Seeder) Seed(ctx context.Context) (int, error) {
	if len(s.cfg.Pairs) == 0 {
		return 0, fmt.Errorf("no pairs configured")
	}
	// Reconcile first: cancel any orders this account left resting from a prior run
	// (the in-memory live set is empty after a restart), so the restart reseeds a
	// clean book instead of leaking stale quotes — crucial when the requote price
	// has moved (e.g. tracking a live feed), so old far-from-market orders can't be
	// picked off. Refunds their locks too. No-op on a transport without an orders
	// read (ZAP / pre-read venue).
	if rc, ok := s.dex.(restingCanceller); ok {
		for _, p := range s.cfg.Pairs {
			if n, err := rc.CancelRestingForSymbol(ctx, p.Symbol); err != nil {
				s.logger.Warn("reconcile failed", "symbol", p.Symbol, "error", err)
			} else if n > 0 {
				s.logger.Info("reconciled stale orders", "symbol", p.Symbol, "cancelled", n)
			}
		}
	}
	// Resolve each pair's mid to the live reference up front, so the very first
	// quote tracks the live price (not the static config mid).
	pairs := s.resolveMids(ctx, s.cfg.Pairs)
	if err := s.fundAll(ctx, pairs); err != nil {
		return 0, err
	}
	total := 0
	for _, p := range pairs {
		n, err := s.seedPair(ctx, p)
		if err != nil {
			return total, fmt.Errorf("seed %s: %w", p.Symbol, err)
		}
		total += n
	}
	return total, nil
}

// resolveMids returns a copy of pairs with each Mid set to the current live mid
// (feed → DEX book → configured reference), preserving every other field. The one
// place both Seed and Maintain turn "what to track" into a concrete requote mid.
func (s *Seeder) resolveMids(ctx context.Context, pairs []Pair) []Pair {
	out := make([]Pair, len(pairs))
	for i, p := range pairs {
		rp := p
		rp.Mid = s.currentMid(ctx, p)
		out[i] = rp
	}
	return out
}

// seedPair opens the pair's market (binding its assets when custody) and places
// its 2-sided quotes, recording the live order ids. Funding is done up front by
// fundAll, so by the time a custody pair places, the maker's balance already
// covers every resting order's lock.
func (s *Seeder) seedPair(ctx context.Context, p Pair) (int, error) {
	if p.Mid <= 0 {
		return 0, fmt.Errorf("invalid mid %f", p.Mid)
	}

	// Pin an explicit venue poolId if the pair carries one, so this pair seeds
	// the exact on-chain book a deployed V4 PoolKey resolves to. Otherwise the
	// pair routes through SymbolPoolID — the venue convention every native rail
	// uses — so it lands in the SAME book the FIX/WS/precompile rails address.
	if p.hasPoolID() {
		if b, ok := s.dex.(poolBinder); ok {
			b.BindPool(p.Symbol, p.PoolID)
		}
	}

	// Bind the market's asset ids so an order locks/settles the right ledger row.
	// Idempotent on the venue; the maker's funds were deposited by fundAll.
	if p.Custody {
		if err := s.dex.OpenMarket(ctx, p.Symbol, p.Base, p.Quote); err != nil {
			return 0, fmt.Errorf("open market %s: %w", p.Symbol, err)
		}
	}

	book := &AggregatedBook{Symbol: p.Symbol, MidPrice: p.Mid}
	bids, asks, err := s.strategy.Quote(ctx, book, 0, s.cfg)
	if err != nil {
		return 0, fmt.Errorf("strategy: %w", err)
	}

	var ids []uint64
	place := func(orders []Order) error {
		for _, o := range orders {
			id, err := s.dex.PlaceLimit(ctx, p.Symbol, o.Side, roundTo8(o.Price), roundTo8(o.Qty))
			if err != nil {
				return err
			}
			ids = append(ids, id)
		}
		return nil
	}
	if err := place(bids); err != nil {
		return len(ids), err
	}
	if err := place(asks); err != nil {
		return len(ids), err
	}

	s.mu.Lock()
	s.live[p.Symbol] = ids
	s.mu.Unlock()

	s.logger.Info("seeded market", "symbol", p.Symbol, "mid", p.Mid,
		"bids", len(bids), "asks", len(asks), "orders", len(ids))
	return len(ids), nil
}

// fundAll deposits exactly enough of each asset to cover the locks of every
// resting order across all custody pairs, BEFORE any order is placed. It
// aggregates PER ASSET (not per pair): a SELL locks base = floor(size), a BUY
// locks quote = ceil(size*price), sized with lockUnits — the same fixed-point
// math the venue applies in orderLock — over the SAME roundTo8 prices/sizes
// seedPair places. Aggregating per asset is required for correctness AND DRY:
// markets that share the native LUX base sum into ONE LUX deposit, so the venue
// can fund every order while the maker deposits exactly what the books lock.
func (s *Seeder) fundAll(ctx context.Context, pairs []Pair) error {
	need := map[[32]byte]uint64{} // asset id -> total balance the maker must hold
	for _, p := range pairs {
		if !p.Custody {
			continue
		}
		if p.Mid <= 0 {
			return fmt.Errorf("invalid mid %f for %s", p.Mid, p.Symbol)
		}
		book := &AggregatedBook{Symbol: p.Symbol, MidPrice: p.Mid}
		bids, asks, err := s.strategy.Quote(ctx, book, 0, s.cfg)
		if err != nil {
			return fmt.Errorf("strategy %s: %w", p.Symbol, err)
		}
		for _, o := range asks { // sells lock base
			need[p.Base] += lockUnits(sideSell, roundTo8(o.Price), roundTo8(o.Qty))
		}
		for _, o := range bids { // buys lock quote
			need[p.Quote] += lockUnits(sideBuy, roundTo8(o.Price), roundTo8(o.Qty))
		}
	}
	// Deposit once per distinct asset, in a deterministic order so the deposit
	// refs (and any logs) are stable run to run.
	for _, asset := range sortedAssets(need) {
		if err := s.dex.Deposit(ctx, asset, need[asset]); err != nil {
			return fmt.Errorf("deposit %x: %w", asset, err)
		}
	}
	return nil
}

// sortedAssets returns the asset ids of m in ascending byte order, for
// deterministic deposit sequencing.
func sortedAssets(m map[[32]byte]uint64) [][32]byte {
	out := make([][32]byte, 0, len(m))
	for a := range m {
		out = append(out, a)
	}
	sort.Slice(out, func(i, j int) bool { return bytes.Compare(out[i][:], out[j][:]) < 0 })
	return out
}

// Maintain refreshes the book for every pair once: cancel prior quotes and
// re-seed around the current mid. Returns total orders placed.
func (s *Seeder) Maintain(ctx context.Context) (int, error) {
	// Cancel every pair's live quotes first; cancelling refunds each order's lock
	// back to the maker's available balance.
	for _, p := range s.cfg.Pairs {
		if err := s.cancelPair(ctx, p.Symbol); err != nil {
			s.logger.Warn("cancel failed", "symbol", p.Symbol, "error", err)
		}
	}

	// Requote around the current mid while preserving each pair's identity
	// (PoolID, Base/Quote, Custody) — only Mid changes.
	requoted := s.resolveMids(ctx, s.cfg.Pairs)

	// Re-fund per asset before re-placing. The deposit refs are content-derived
	// from (user, asset, amount); when the requote mid (hence the locks) is
	// unchanged the deposit dedups to a no-op and the cancels' refunds cover the
	// re-placement, and when the mid moved the new amount is a genuinely distinct
	// deposit that tops the balance up. Either way every re-placed order is funded.
	if err := s.fundAll(ctx, requoted); err != nil {
		return 0, err
	}

	total := 0
	for _, rp := range requoted {
		n, err := s.seedPair(ctx, rp)
		if err != nil {
			return total, fmt.Errorf("maintain %s: %w", rp.Symbol, err)
		}
		total += n
	}
	return total, nil
}

// currentMid derives the requote mid, in priority order:
//  1. the live EXTERNAL reference (feed) — the price this market should track;
//  2. the DEX book's own midpoint when both sides exist (external flow moving our
//     book, when no feed is configured);
//  3. the configured reference mid (cold start / no live signal).
//
// So with a feed the book stays pinned to the live AMM/CEX price; without one the
// prior behavior is preserved exactly.
func (s *Seeder) currentMid(ctx context.Context, p Pair) float64 {
	if s.feed != nil {
		if mid, err := s.feed.GetMidPrice(ctx, p.Symbol); err == nil && mid > 0 {
			return roundTo8(mid)
		}
	}
	bid, _, errB := s.dex.BestBid(ctx, p.Symbol)
	ask, _, errA := s.dex.BestAsk(ctx, p.Symbol)
	if errB == nil && errA == nil && bid > 0 && ask > 0 {
		return roundTo8((bid + ask) / 2)
	}
	return p.Mid
}

// cancelPair cancels all live orders for a symbol.
func (s *Seeder) cancelPair(ctx context.Context, symbol string) error {
	s.mu.Lock()
	ids := s.live[symbol]
	delete(s.live, symbol)
	s.mu.Unlock()

	var firstErr error
	for _, id := range ids {
		if err := s.dex.Cancel(ctx, id); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

// Run seeds once, then maintains the book every RefreshInterval until ctx is
// cancelled or Stop is called.
func (s *Seeder) Run(ctx context.Context) error {
	if _, err := s.Seed(ctx); err != nil {
		return err
	}

	interval := s.cfg.RefreshInterval
	if interval <= 0 {
		interval = 5 * time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-s.stopCh:
			return nil
		case <-ticker.C:
			if _, err := s.Maintain(ctx); err != nil {
				s.logger.Error("maintain failed", "error", err)
			}
		}
	}
}

// Stop halts Run.
func (s *Seeder) Stop() {
	select {
	case <-s.stopCh:
	default:
		close(s.stopCh)
	}
}

// LiveOrders returns the active order ids for a symbol (copy).
func (s *Seeder) LiveOrders(symbol string) []uint64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	ids := s.live[symbol]
	out := make([]uint64, len(ids))
	copy(out, ids)
	return out
}