maker/reboot.go

package maker

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"math"
	"math/rand/v2"
	"net/http"
	"strconv"
	"strings"
	"time"
)

// reboot.go is the lux.exchange trading-reboot path: bring every configured
// market back up at its LAST on-chain price, paint a continuous display history
// over the dark gap (June -> now), then run the existing custody-backed seeder.
// It is pure orchestration ON TOP of the maker — it reuses Seeder (fund before
// place, reject underfunded), MidSpreadStrategy (two-sided quotes), and DEXClient
// (now signing) verbatim; it adds only (a) a last-price source, (b) a
// display-only candle backfill, and (c) the orchestrator that wires them.
//
// MONEY SAFETY: the backfill is DISPLAY ONLY — it never places an order and never
// moves funds. The only path that moves value is Seeder.Seed/Maintain, whose
// fund-before-place + reject-underfunded invariants are unchanged. A non-positive
// mid is never seeded or backfilled.

// rebootCandleInterval is the gap-fill candle width. 1h matches the marketdata
// aggregator's Interval1h, so the painted history slots into a real candle store.
const rebootCandleInterval = time.Hour

// PriceSource yields the last traded/spot price for a symbol. The reboot resolves
// each market's seed mid from it, falling back to the configured Pair.Mid on error.
type PriceSource interface {
	LastPrice(ctx context.Context, symbol string) (mid float64, source string, err error)
}

// --- (a) AMM price source ----------------------------------------------------

// AMMPriceSource reads the last on-chain price from the native AMM subgraph
// (luxfi/graph, uniswap-v2-compatible) — the same `amm` schema exchange-api's
// getSubgraphPairs queries. It is the source of truth for "the price the market
// was last at" when the CLOB book is empty on reboot.
type AMMPriceSource struct {
	URL    string               // subgraph GraphQL endpoint
	HTTP   *http.Client         // optional; a 10s-timeout client is used when nil
	Tokens map[string][2]string // symbol -> {baseAddr, quoteAddr}, lowercased 0x addresses
}

var _ PriceSource = (*AMMPriceSource)(nil)

// ammPairsQuery asks for the pools plus each pool's newest swap in one round trip,
// so amm-price / amm-reserve / amm-swap are all derivable without a second call.
// Field shape mirrors exchange-api getSubgraphPairs (id, token0/1{id symbol},
// reserve0/1, token0Price/token1Price) plus the newest swap for the fallback.
const ammPairsQuery = `{
  pairs(first: 1000) {
    id
    token0 { id symbol }
    token1 { id symbol }
    reserve0
    reserve1
    token0Price
    token1Price
    swaps(first: 1, orderBy: timestamp, orderDirection: desc) {
      amount0In
      amount0Out
      amount1In
      amount1Out
    }
  }
}`

type ammToken struct {
	ID     string `json:"id"`
	Symbol string `json:"symbol"`
}

type ammSwap struct {
	Amount0In  string `json:"amount0In"`
	Amount0Out string `json:"amount0Out"`
	Amount1In  string `json:"amount1In"`
	Amount1Out string `json:"amount1Out"`
}

type ammPair struct {
	ID          string    `json:"id"`
	Token0      ammToken  `json:"token0"`
	Token1      ammToken  `json:"token1"`
	Reserve0    string    `json:"reserve0"`
	Reserve1    string    `json:"reserve1"`
	Token0Price string    `json:"token0Price"`
	Token1Price string    `json:"token1Price"`
	Swaps       []ammSwap `json:"swaps"`
}

type ammResponse struct {
	Data struct {
		Pairs []ammPair `json:"pairs"`
	} `json:"data"`
	Errors []struct {
		Message string `json:"message"`
	} `json:"errors"`
}

// LastPrice resolves the implied mid for a symbol from the AMM pool of its two
// tokens.
//
// ORIENTATION. A symbol is BASE/QUOTE and the mid is QUOTE per BASE (e.g.
// LUX/LUSD -> LUSD per LUX). The subgraph's token0Price is token0-per-token1 and
// token1Price is token1-per-token0 (= reserve1/reserve0). So:
//   - base == token0 (quote == token1): mid = token1Price = reserve1/reserve0;
//   - base == token1 (quote == token0): mid = token0Price = reserve0/reserve1.
//
// Preference: the indexer's quoted price ("amm-price"), else the reserve ratio
// ("amm-reserve"), else the newest swap's executed price ("amm-swap"). A missing
// pool or unusable reserves is a clear error — the orchestrator then keeps the
// configured Pair.Mid.
func (a *AMMPriceSource) LastPrice(ctx context.Context, symbol string) (float64, string, error) {
	pair, ok := a.Tokens[symbol]
	if !ok {
		return 0, "", fmt.Errorf("amm: no token mapping for symbol %q", symbol)
	}
	base, quote := strings.ToLower(pair[0]), strings.ToLower(pair[1])
	if base == "" || quote == "" {
		return 0, "", fmt.Errorf("amm: empty token address for symbol %q", symbol)
	}

	pairs, err := a.queryPairs(ctx)
	if err != nil {
		return 0, "", err
	}
	p := matchPair(pairs, base, quote)
	if p == nil {
		return 0, "", fmt.Errorf("amm: no pool for %s (%s / %s)", symbol, base, quote)
	}
	baseIsToken0 := strings.EqualFold(p.Token0.ID, base)

	// (1) Indexer-quoted price.
	priceStr := p.Token1Price
	if !baseIsToken0 {
		priceStr = p.Token0Price
	}
	if v, ok := parseFloatPos(priceStr); ok {
		return v, "amm-price", nil
	}

	// (2) Reserve ratio.
	r0, ok0 := parseFloatPos(p.Reserve0)
	r1, ok1 := parseFloatPos(p.Reserve1)
	if ok0 && ok1 {
		if baseIsToken0 {
			return r1 / r0, "amm-reserve", nil
		}
		return r0 / r1, "amm-reserve", nil
	}

	// (3) Newest swap's executed price.
	if mid, ok := lastSwapMid(p, baseIsToken0); ok {
		return mid, "amm-swap", nil
	}

	return 0, "", fmt.Errorf("amm: pool for %s has no usable price, reserves, or swap", symbol)
}

// queryPairs POSTs the GraphQL query and returns the decoded pools. std-lib only.
func (a *AMMPriceSource) queryPairs(ctx context.Context) ([]ammPair, error) {
	if a.URL == "" {
		return nil, fmt.Errorf("amm: empty subgraph URL")
	}
	reqBody, err := json.Marshal(map[string]string{"query": ammPairsQuery})
	if err != nil {
		return nil, fmt.Errorf("amm: marshal query: %w", err)
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.URL, bytes.NewReader(reqBody))
	if err != nil {
		return nil, fmt.Errorf("amm: build request: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")

	hc := a.HTTP
	if hc == nil {
		hc = &http.Client{Timeout: 10 * time.Second}
	}
	resp, err := hc.Do(req)
	if err != nil {
		return nil, fmt.Errorf("amm: post %s: %w", a.URL, err)
	}
	defer resp.Body.Close()

	raw, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("amm: read response: %w", err)
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("amm: http %d: %s", resp.StatusCode, strings.TrimSpace(string(raw)))
	}
	var out ammResponse
	if err := json.Unmarshal(raw, &out); err != nil {
		return nil, fmt.Errorf("amm: decode response: %w", err)
	}
	if len(out.Errors) > 0 {
		return nil, fmt.Errorf("amm: graphql error: %s", out.Errors[0].Message)
	}
	return out.Data.Pairs, nil
}

// matchPair finds the pool whose two token addresses are exactly {base, quote},
// in either ordering. Addresses are compared lowercased.
func matchPair(pairs []ammPair, base, quote string) *ammPair {
	for i := range pairs {
		t0 := strings.ToLower(pairs[i].Token0.ID)
		t1 := strings.ToLower(pairs[i].Token1.ID)
		if (t0 == base && t1 == quote) || (t0 == quote && t1 == base) {
			return &pairs[i]
		}
	}
	return nil
}

// lastSwapMid derives quote-per-base from the newest swap's net token magnitudes
// (token1 per token0 = |Δtoken1| / |Δtoken0|), oriented to base/quote.
func lastSwapMid(p *ammPair, baseIsToken0 bool) (float64, bool) {
	if len(p.Swaps) == 0 {
		return 0, false
	}
	s := p.Swaps[0]
	t0 := absFloat(s.Amount0In) + absFloat(s.Amount0Out)
	t1 := absFloat(s.Amount1In) + absFloat(s.Amount1Out)
	if !(t0 > 0) || !(t1 > 0) {
		return 0, false
	}
	if baseIsToken0 {
		return t1 / t0, true
	}
	return t0 / t1, true
}

// parseFloatPos parses s into a finite, strictly-positive float, reporting ok.
func parseFloatPos(s string) (float64, bool) {
	v, err := strconv.ParseFloat(strings.TrimSpace(s), 64)
	if err != nil || math.IsInf(v, 0) || math.IsNaN(v) || !(v > 0) {
		return 0, false
	}
	return v, true
}

// absFloat parses s and returns its magnitude (0 on parse failure).
func absFloat(s string) float64 {
	v, err := strconv.ParseFloat(strings.TrimSpace(s), 64)
	if err != nil {
		return 0
	}
	return math.Abs(v)
}

// --- (b) gap candle backfill (DISPLAY ONLY) ----------------------------------

// Candle is one OHLCV bar. Fields and JSON tags mirror marketdata.Candle (the
// gap-fill artifact uses int64 unix seconds for the times and a string interval).
type Candle struct {
	Symbol    string  `json:"symbol"`
	Interval  string  `json:"interval"`
	OpenTime  int64   `json:"openTime"`
	CloseTime int64   `json:"closeTime"`
	Open      float64 `json:"open"`
	High      float64 `json:"high"`
	Low       float64 `json:"low"`
	Close     float64 `json:"close"`
	Volume    float64 `json:"volume"`
}

// CandleSink consumes generated candles. Implementations must never place orders
// or move funds — a sink is display/history only.
type CandleSink interface {
	WriteCandle(c Candle) error
}

// KVCandleSink writes candles into a key-value store under the EXACT key the
// marketdata aggregator's storeCandle uses (candle:<symbol>:<interval>:<openUnix>),
// so a real marketdata store serves the painted history unchanged.
type KVCandleSink struct {
	Store interface {
		Put(key, val []byte) error
	}
}

func (s KVCandleSink) WriteCandle(c Candle) error {
	if s.Store == nil {
		return fmt.Errorf("candle sink: nil store")
	}
	key := fmt.Sprintf("candle:%s:%s:%d", c.Symbol, c.Interval, c.OpenTime)
	val, err := json.Marshal(c)
	if err != nil {
		return fmt.Errorf("candle sink: marshal %s: %w", key, err)
	}
	return s.Store.Put([]byte(key), val)
}

// JSONLinesSink writes one JSON candle per line — the dry-run proof artifact.
type JSONLinesSink struct {
	W io.Writer
}

func (s JSONLinesSink) WriteCandle(c Candle) error {
	if s.W == nil {
		return fmt.Errorf("candle sink: nil writer")
	}
	b, err := json.Marshal(c)
	if err != nil {
		return fmt.Errorf("candle sink: marshal: %w", err)
	}
	if _, err := s.W.Write(append(b, '\n')); err != nil {
		return fmt.Errorf("candle sink: write: %w", err)
	}
	return nil
}

// pcgStream is the fixed second PCG seed lane so a single int64 seed fully
// determines the walk (golden-ratio constant — arbitrary but stable).
const pcgStream uint64 = 0x9e3779b97f4a7c15

// GapBackfill paints a continuous, DETERMINISTIC OHLCV history from `from` to `to`
// at `interval`, as a bounded random walk around `anchor`. Invariants the display
// layer depends on:
//   - count == (to-from)/interval;
//   - Open[i] == Close[i-1] (continuity);
//   - every Close is within ±driftBps of anchor;
//   - the FINAL Close == anchor EXACTLY (history joins the live seed at the
//     last-on-chain price);
//   - High >= max(Open,Close), Low <= min(Open,Close), Volume > 0.
//
// Same seed => identical output. This is the June->now gap fill; it is display
// only and never touches funds or orders.
func GapBackfill(symbol string, anchor float64, from, to int64, interval time.Duration, driftBps float64, seed int64) []Candle {
	stepSecs := int64(interval.Seconds())
	if anchor <= 0 || stepSecs <= 0 || to <= from {
		return nil
	}
	n := int((to - from) / stepSecs)
	if n <= 0 {
		return nil
	}

	band := anchor * driftBps / 10000 // half-width of the allowed close band
	lo, hi := anchor-band, anchor+band
	rng := rand.New(rand.NewPCG(uint64(seed), pcgStream))

	out := make([]Candle, 0, n)
	prevClose := anchor // first open = anchor; the walk stays within [lo, hi]
	for i := range n {
		openPx := prevClose
		var closePx float64
		if i == n-1 {
			closePx = anchor // pin the history's tail to the live seed price
		} else {
			closePx = openPx + (rng.Float64()*2-1)*band*0.5 // mild step, then clamp
			if closePx < lo {
				closePx = lo
			}
			if closePx > hi {
				closePx = hi
			}
		}
		hiVal, loVal := math.Max(openPx, closePx), math.Min(openPx, closePx)
		wick := rng.Float64() * band * 0.25
		high := hiVal + wick
		low := loVal - wick
		if low <= 0 { // guard a tiny anchor: never emit a non-positive low
			low = loVal
		}
		openTime := from + int64(i)*stepSecs
		out = append(out, Candle{
			Symbol:    symbol,
			Interval:  intervalLabel(interval),
			OpenTime:  openTime,
			CloseTime: openTime + stepSecs,
			Open:      openPx,
			High:      high,
			Low:       low,
			Close:     closePx,
			Volume:    1 + rng.Float64()*9,
		})
		prevClose = closePx
	}
	return out
}

// intervalLabel maps a candle width to the marketdata.Interval label string, so a
// painted candle's interval field matches what a real aggregator stores/serves.
// Inverse of marketdata.Interval.Duration; unknown widths fall back to d.String().
func intervalLabel(d time.Duration) string {
	switch d {
	case time.Second:
		return "1s"
	case 5 * time.Second:
		return "5s"
	case 15 * time.Second:
		return "15s"
	case 30 * time.Second:
		return "30s"
	case time.Minute:
		return "1m"
	case 3 * time.Minute:
		return "3m"
	case 5 * time.Minute:
		return "5m"
	case 15 * time.Minute:
		return "15m"
	case 30 * time.Minute:
		return "30m"
	case time.Hour:
		return "1h"
	case 2 * time.Hour:
		return "2h"
	case 4 * time.Hour:
		return "4h"
	case 6 * time.Hour:
		return "6h"
	case 8 * time.Hour:
		return "8h"
	case 12 * time.Hour:
		return "12h"
	case 24 * time.Hour:
		return "1d"
	case 3 * 24 * time.Hour:
		return "3d"
	case 7 * 24 * time.Hour:
		return "1w"
	default:
		return d.String()
	}
}

// --- (c) orchestrator --------------------------------------------------------

// Reboot brings the configured markets back online: resolve each pair's last
// on-chain price, custody-fund + place a two-sided book at it, paint the dark-gap
// display history, then run the maintain loop. The Seeder MUST be constructed from
// the same Config value as Cfg (they share the Pairs backing array), so a mid
// ResolveMids writes into Cfg.Pairs[i] is the mid the Seeder seeds.
type Reboot struct {
	Seeder   *Seeder
	Prices   PriceSource
	Sink     CandleSink
	Cfg      Config
	GapFrom  int64         // unix seconds the painted history starts at
	Interval time.Duration // maintain-loop requote interval
	DriftBps float64       // bound on the painted walk around the seed price
	Logger   *slog.Logger
}

// ResolveMids sets each pair's Mid to its last on-chain price from Prices. On any
// error it keeps the configured Mid (and logs a fallback). A non-positive resolved
// price is never applied.
func (r *Reboot) ResolveMids(ctx context.Context) {
	for i := range r.Cfg.Pairs {
		sym := r.Cfg.Pairs[i].Symbol
		mid, source, err := r.Prices.LastPrice(ctx, sym)
		if err != nil || mid <= 0 {
			r.log().Warn("price resolve fallback to configured mid",
				"symbol", sym, "mid", r.Cfg.Pairs[i].Mid, "error", err)
			continue
		}
		r.Cfg.Pairs[i].Mid = mid
		r.log().Info("resolved last on-chain mid", "symbol", sym, "mid", mid, "source", source)
	}
}

// Backfill paints the gap [GapFrom, now] for every pair into Sink at the 1h candle
// width, anchored to the pair's (resolved) Mid. It is display only. Returns the
// total candles written. A pair with a non-positive Mid is skipped (never seeded
// or painted).
func (r *Reboot) Backfill(now int64) (int, error) {
	total := 0
	for i := range r.Cfg.Pairs {
		p := r.Cfg.Pairs[i]
		if p.Mid <= 0 {
			r.log().Warn("skip backfill: non-positive mid", "symbol", p.Symbol)
			continue
		}
		candles := GapBackfill(p.Symbol, p.Mid, r.GapFrom, now, rebootCandleInterval, r.DriftBps, backfillSeed(p.Symbol, r.GapFrom))
		for _, c := range candles {
			if err := r.Sink.WriteCandle(c); err != nil {
				return total, fmt.Errorf("backfill %s: %w", p.Symbol, err)
			}
			total++
		}
		r.log().Info("painted gap history", "symbol", p.Symbol, "candles", len(candles), "anchor", p.Mid)
	}
	return total, nil
}

// Run executes the reboot: resolve mids, seed the custody-backed two-sided books
// at the resolved prices (fund before place), paint the gap history, then maintain
// the books on Interval until ctx is done.
func (r *Reboot) Run(ctx context.Context) error {
	r.ResolveMids(ctx)

	placed, err := r.Seeder.Seed(ctx)
	if err != nil {
		return fmt.Errorf("reboot seed: %w", err)
	}
	r.log().Info("reboot seeded books", "orders", placed)

	now := time.Now().Unix()
	n, err := r.Backfill(now)
	if err != nil {
		return fmt.Errorf("reboot backfill: %w", err)
	}
	r.log().Info("reboot gap backfill complete", "candles", n, "from", r.GapFrom, "to", now)

	interval := r.Interval
	if interval <= 0 {
		interval = r.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 <-ticker.C:
			if _, err := r.Seeder.Maintain(ctx); err != nil {
				r.log().Error("reboot maintain failed", "error", err)
			}
		}
	}
}

// backfillSeed derives a deterministic per-(symbol, gap-start) seed so each
// market's painted history is distinct yet reproducible run to run.
func backfillSeed(symbol string, gapFrom int64) int64 {
	const (
		offset64 = 1469598103934665603
		prime64  = 1099511628211
	)
	h := uint64(offset64)
	for i := 0; i < len(symbol); i++ {
		h = (h ^ uint64(symbol[i])) * prime64
	}
	return int64(h ^ uint64(gapFrom))
}

// log returns the configured logger or a discarding one.
func (r *Reboot) log() *slog.Logger {
	if r.Logger != nil {
		return r.Logger
	}
	return slog.New(slog.NewTextHandler(io.Discard, nil))
}