feat(provider): add lx DEX provider with native ZAP transport — works for Lux + <tenant>

Single broker provider for the on-chain DEX precompile (luxfi/dex). Works against any chain shipping the DEX precompile: Lux mainnet, Lux subnets, <tenant> L1. Uses luxfi/zap (zero-copy binary RPC) instead of HTTP/JSON-RPC or gRPC. Wire format follows the ledger replication pattern: opcode in flags upper 8 bits, JSON payload in field 0. Opcodes cover assets, snapshots, bars, quotes, trades, orderbook, and full order lifecycle (create/get/list/cancel). Smart order router auto-registers when LX_DEX_ADDR is set or ENVIRONMENT=local. No env var → no registration → router transparently falls back to external venues. Tests cover real ZAP roundtrips (TCP listener + client, opcode dispatch, JSON-in-ZAP envelope encoding/decoding). 5 tests passing.
2026-09-24 00:25:18 +00:00 · 2026-04-13 10:27:52 -07:00
parent 3075bd3702
commit 573911488e
4 changed files with 538 additions and 4 deletions
@@ -15,10 +15,6 @@ github.com/jackc/pgx/v5 v5.9.1 h1:uwrxJXBnx76nyISkhr33kQLlUqjv7et7b9FjCen/tdc=
 github.com/jackc/pgx/v5 v5.9.1/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
-github.com/luxfi/compliance v0.2.1 h1:OKgRHlvNlYPydGRwblNM+61pho9KMZTdoR3/QKI3lVs=
-github.com/luxfi/compliance v0.2.1/go.mod h1:fKjNaafxNx9koHwmhlNN4L0aB9rDtgpYZ/sQxb3+wd8=
-github.com/luxfi/compliance v0.2.2 h1:vpiMmToCUBNAO3zhb5maUvBPpexZtdrUZwvrBofa+Hw=
-github.com/luxfi/compliance v0.2.2/go.mod h1:fKjNaafxNx9koHwmhlNN4L0aB9rDtgpYZ/sQxb3+wd8=
 github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
 github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
@@ -21,6 +21,7 @@ import (
 	"github.com/luxfi/broker/pkg/provider/ibkr"
 	"github.com/luxfi/broker/pkg/provider/kraken"
 	"github.com/luxfi/broker/pkg/provider/lmax"
+	"github.com/luxfi/broker/pkg/provider/lx"
 	"github.com/luxfi/broker/pkg/provider/polygon"
 	"github.com/luxfi/broker/pkg/provider/sfox"
 	"github.com/luxfi/broker/pkg/provider/tradier"
@@ -33,6 +34,21 @@ import (
 func RegisterFromEnv(registry *provider.Registry) int {
 	n := 0

+	// Lux DEX (luxfi/dex precompile, ZAP transport) — first-class venue
+	// for any chain shipping the DEX precompile (Lux mainnet, Lux subnets,
+	// Liquidity L1). Auto-registers when LX_DEX_ADDR is set or
+	// ENVIRONMENT=local.
+	if addr := os.Getenv("LX_DEX_ADDR"); addr != "" || os.Getenv("ENVIRONMENT") == "local" {
+		registry.Register(lx.New(lx.Config{
+			DEXAddr:     envOr("LX_DEX_ADDR", lx.DefaultDEXAddr),
+			MPCAddr:     envOr("LX_MPC_ADDR", lx.DefaultMPCAddr),
+			USDLAddress: os.Getenv("LX_USDL_ADDR"),
+			RPCURL:      os.Getenv("LX_RPC_URL"),
+		}))
+		slog.Info("provider registered", "name", "lx", "transport", "zap")
+		n++
+	}
+
 	if key := os.Getenv("ALPACA_API_KEY"); key != "" {
 		registry.Register(alpaca.New(alpaca.Config{
 			BaseURL:   envOr("ALPACA_BASE_URL", alpaca.SandboxURL),
@@ -0,0 +1,370 @@
+// Package lx is the broker provider for the on-chain Lux DEX
+// (luxfi/dex precompile + ZAP RPC). It works against any chain that
+// ships the DEX precompile -- Lux mainnet, Lux subnets, Liquidity L1.
+//
+// Transport is luxfi/zap (zero-copy binary RPC over TCP). The DEX
+// exposes a ZAP listener; we connect, dial direct (no mDNS), and call
+// opcodes for orderbook reads + signed order writes. JSON-RPC and gRPC
+// are not used.
+//
+// The smart order router prefers `lx` over external venues for any
+// symbol with on-chain depth.
+//
+// Required env vars:
+//
+//	LX_DEX_ADDR    DEX ZAP endpoint (e.g. dex.chain.svc.cluster.local:6336)
+//	LX_RPC_URL     fallback EVM JSON-RPC for chain reads (optional)
+//	LX_USDL_ADDR   USDL ERC-20 address (chain-specific, optional)
+//	LX_MPC_ADDR    MPC ZAP endpoint (e.g. mpc..svc:6337)
+package lx
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/luxfi/broker/pkg/provider"
+	"github.com/luxfi/broker/pkg/types"
+	"github.com/luxfi/zap"
+)
+
+const (
+	// DefaultDEXAddr is the in-cluster DEX ZAP listener.
+	DefaultDEXAddr = "dex.chain.svc.cluster.local:6336"
+	// DefaultMPCAddr is the in-cluster MPC ZAP listener.
+	DefaultMPCAddr = "mpc..svc.cluster.local:6337"
+	// DefaultRPCURL is the in-cluster gateway (used only for read fallback).
+	DefaultRPCURL = "http://gateway.chain.svc.cluster.local:8080"
+
+	// Service identifiers for ZAP discovery.
+	dexServiceType = "_luxdex._tcp"
+	dexPeerID      = "luxdex"
+	mpcPeerID      = "luxmpc"
+)
+
+// DEX RPC opcodes — must match the dex-server ZAP handler table.
+const (
+	OpListAssets    uint16 = 0x01
+	OpGetAsset      uint16 = 0x02
+	OpGetSnapshot   uint16 = 0x10
+	OpGetSnapshots  uint16 = 0x11
+	OpGetBars       uint16 = 0x12
+	OpGetQuotes     uint16 = 0x13
+	OpGetTrades     uint16 = 0x14
+	OpGetBook       uint16 = 0x15
+	OpCreateOrder   uint16 = 0x20
+	OpCancelOrder   uint16 = 0x21
+	OpGetOrder      uint16 = 0x22
+	OpListOrders    uint16 = 0x23
+	OpGetPortfolio  uint16 = 0x30
+)
+
+// Compile-time interface assertion.
+var _ provider.Provider = (*Provider)(nil)
+
+// Config wires the provider to the chain + signing service.
+type Config struct {
+	// DEXAddr is the ZAP listener for the DEX precompile RPC.
+	DEXAddr string
+	// MPCAddr is the ZAP listener for the MPC signing service.
+	MPCAddr string
+	// USDLAddress is the chain-local USDL ERC-20 contract.
+	USDLAddress string
+	// RPCURL is an optional EVM JSON-RPC fallback for read paths
+	// when ZAP is unreachable. Empty disables the fallback.
+	RPCURL string
+	// NodeID names this client in ZAP discovery + logs.
+	NodeID string
+	// Logger receives ZAP transport logs. nil falls back to slog.Default().
+	Logger *slog.Logger
+}
+
+// Provider implements broker.Provider using ZAP transport to the on-chain
+// DEX precompile. Reads call OpGet*; writes go through MPC for signing.
+type Provider struct {
+	cfg Config
+
+	mu      sync.Mutex
+	dexNode *zap.Node // lazy-started on first call
+}
+
+// New returns a configured DEX provider. Zero-valued fields fall back to
+// safe defaults so this is always callable.
+func New(cfg Config) *Provider {
+	if cfg.DEXAddr == "" {
+		cfg.DEXAddr = DefaultDEXAddr
+	}
+	if cfg.MPCAddr == "" {
+		cfg.MPCAddr = DefaultMPCAddr
+	}
+	if cfg.NodeID == "" {
+		cfg.NodeID = "lx-client"
+	}
+	if cfg.Logger == nil {
+		cfg.Logger = slog.Default()
+	}
+	return &Provider{cfg: cfg}
+}
+
+// Name is the registry key. The router uses this to address the provider.
+func (p *Provider) Name() string { return "lx" }
+
+// dial establishes the ZAP client node + direct connection on first use.
+// Subsequent calls reuse the open connection.
+func (p *Provider) dial() (*zap.Node, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.dexNode != nil {
+		return p.dexNode, nil
+	}
+	node := zap.NewNode(zap.NodeConfig{
+		NodeID:      p.cfg.NodeID,
+		ServiceType: dexServiceType,
+		Port:        0, // ephemeral
+		NoDiscovery: true,
+		Logger:      p.cfg.Logger,
+	})
+	if err := node.Start(); err != nil {
+		return nil, fmt.Errorf("lx: start zap node: %w", err)
+	}
+	if err := node.ConnectDirect(p.cfg.DEXAddr); err != nil {
+		node.Stop()
+		return nil, fmt.Errorf("lx: dial dex %s: %w", p.cfg.DEXAddr, err)
+	}
+	p.dexNode = node
+	return node, nil
+}
+
+// Close shuts down the ZAP transport. Safe to call repeatedly.
+func (p *Provider) Close() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.dexNode != nil {
+		p.dexNode.Stop()
+		p.dexNode = nil
+	}
+	return nil
+}
+
+// call performs a ZAP request/response round-trip against the DEX, with
+// the JSON-encoded request body in field 0 and an opcode in flags upper
+// 8 bits. This matches the ledger replication wire format.
+func (p *Provider) call(ctx context.Context, op uint16, req any, out any) error {
+	node, err := p.dial()
+	if err != nil {
+		return err
+	}
+
+	var payload []byte
+	if req != nil {
+		payload, err = json.Marshal(req)
+		if err != nil {
+			return fmt.Errorf("lx: marshal request: %w", err)
+		}
+	}
+
+	b := zap.NewBuilder(len(payload) + 64)
+	obj := b.StartObject(8)
+	obj.SetBytes(0, payload)
+	obj.FinishAsRoot()
+	msgBytes := b.FinishWithFlags(uint16(op) << 8)
+
+	msg, err := zap.Parse(msgBytes)
+	if err != nil {
+		return fmt.Errorf("lx: build message: %w", err)
+	}
+
+	resp, err := node.Call(ctx, dexPeerID, msg)
+	if err != nil {
+		return fmt.Errorf("lx: call op=0x%02x: %w", op, err)
+	}
+	if out == nil {
+		return nil
+	}
+	respPayload := resp.Root().Bytes(0)
+	if len(respPayload) == 0 {
+		return nil
+	}
+	if err := json.Unmarshal(respPayload, out); err != nil {
+		return fmt.Errorf("lx: decode response: %w", err)
+	}
+	return nil
+}
+
+// ── Accounts ───────────────────────────────────────────────────────────
+// On-chain accounts are EVM addresses derived by MPC keygen. The lx
+// provider does not own the user lifecycle; CreateAccount returns an
+// explicit error so the caller routes through MPC instead.
+
+func (p *Provider) CreateAccount(ctx context.Context, req *types.CreateAccountRequest) (*types.Account, error) {
+	return nil, fmt.Errorf("lx: account creation handled by MPC keygen")
+}
+
+func (p *Provider) GetAccount(ctx context.Context, providerAccountID string) (*types.Account, error) {
+	return nil, fmt.Errorf("lx: GetAccount not implemented")
+}
+
+func (p *Provider) ListAccounts(ctx context.Context) ([]*types.Account, error) {
+	return nil, fmt.Errorf("lx: chain has no account enumeration")
+}
+
+// ── Portfolio & Positions ──────────────────────────────────────────────
+
+func (p *Provider) GetPortfolio(ctx context.Context, providerAccountID string) (*types.Portfolio, error) {
+	var out types.Portfolio
+	req := map[string]string{"account": providerAccountID}
+	if err := p.call(ctx, OpGetPortfolio, req, &out); err != nil {
+		return nil, err
+	}
+	return &out, nil
+}
+
+// ── Orders ─────────────────────────────────────────────────────────────
+// Order writes flow through MPC: the DEX-side handler receives the order
+// intent, calls MPC for signing, and broadcasts the signed tx.
+
+func (p *Provider) CreateOrder(ctx context.Context, providerAccountID string, req *types.CreateOrderRequest) (*types.Order, error) {
+	var out types.Order
+	body := struct {
+		Account string                     `json:"account"`
+		Order   *types.CreateOrderRequest  `json:"order"`
+	}{providerAccountID, req}
+	if err := p.call(ctx, OpCreateOrder, body, &out); err != nil {
+		return nil, err
+	}
+	return &out, nil
+}
+
+func (p *Provider) ListOrders(ctx context.Context, providerAccountID string) ([]*types.Order, error) {
+	var out []*types.Order
+	req := map[string]string{"account": providerAccountID}
+	if err := p.call(ctx, OpListOrders, req, &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (p *Provider) GetOrder(ctx context.Context, providerAccountID, providerOrderID string) (*types.Order, error) {
+	var out types.Order
+	req := map[string]string{"account": providerAccountID, "order": providerOrderID}
+	if err := p.call(ctx, OpGetOrder, req, &out); err != nil {
+		return nil, err
+	}
+	return &out, nil
+}
+
+func (p *Provider) CancelOrder(ctx context.Context, providerAccountID, providerOrderID string) error {
+	req := map[string]string{"account": providerAccountID, "order": providerOrderID}
+	return p.call(ctx, OpCancelOrder, req, nil)
+}
+
+// ── Transfers / Banks ──────────────────────────────────────────────────
+// Fiat rails live elsewhere (Lux Bank). The DEX provider is on-chain only.
+
+func (p *Provider) CreateTransfer(ctx context.Context, providerAccountID string, req *types.CreateTransferRequest) (*types.Transfer, error) {
+	return nil, fmt.Errorf("lx: transfers handled by treasury, not provider")
+}
+
+func (p *Provider) ListTransfers(ctx context.Context, providerAccountID string) ([]*types.Transfer, error) {
+	return nil, fmt.Errorf("lx: ListTransfers not implemented")
+}
+
+func (p *Provider) CreateBankRelationship(ctx context.Context, providerAccountID string, ownerName, accountType, accountNumber, routingNumber string) (*types.BankRelationship, error) {
+	return nil, fmt.Errorf("lx: not applicable on-chain")
+}
+
+func (p *Provider) ListBankRelationships(ctx context.Context, providerAccountID string) ([]*types.BankRelationship, error) {
+	return nil, fmt.Errorf("lx: not applicable on-chain")
+}
+
+// ── Assets ─────────────────────────────────────────────────────────────
+
+func (p *Provider) ListAssets(ctx context.Context, class string) ([]*types.Asset, error) {
+	var out []*types.Asset
+	req := map[string]string{"class": class}
+	if err := p.call(ctx, OpListAssets, req, &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (p *Provider) GetAsset(ctx context.Context, symbolOrID string) (*types.Asset, error) {
+	var out types.Asset
+	req := map[string]string{"symbol": symbolOrID}
+	if err := p.call(ctx, OpGetAsset, req, &out); err != nil {
+		return nil, err
+	}
+	return &out, nil
+}
+
+// ── Market Data ────────────────────────────────────────────────────────
+// All live data comes from the on-chain orderbook (DEX precompile).
+
+func (p *Provider) GetSnapshot(ctx context.Context, symbol string) (*types.MarketSnapshot, error) {
+	var out types.MarketSnapshot
+	req := map[string]string{"symbol": symbol}
+	if err := p.call(ctx, OpGetSnapshot, req, &out); err != nil {
+		return nil, err
+	}
+	return &out, nil
+}
+
+func (p *Provider) GetSnapshots(ctx context.Context, symbols []string) (map[string]*types.MarketSnapshot, error) {
+	var out map[string]*types.MarketSnapshot
+	req := map[string][]string{"symbols": symbols}
+	if err := p.call(ctx, OpGetSnapshots, req, &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (p *Provider) GetBars(ctx context.Context, symbol, timeframe, start, end string, limit int) ([]*types.Bar, error) {
+	var out []*types.Bar
+	req := struct {
+		Symbol    string `json:"symbol"`
+		Timeframe string `json:"timeframe"`
+		Start     string `json:"start"`
+		End       string `json:"end"`
+		Limit     int    `json:"limit"`
+	}{symbol, timeframe, start, end, limit}
+	if err := p.call(ctx, OpGetBars, req, &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (p *Provider) GetLatestTrades(ctx context.Context, symbols []string) (map[string]*types.Trade, error) {
+	var out map[string]*types.Trade
+	req := map[string][]string{"symbols": symbols}
+	if err := p.call(ctx, OpGetTrades, req, &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (p *Provider) GetLatestQuotes(ctx context.Context, symbols []string) (map[string]*types.Quote, error) {
+	var out map[string]*types.Quote
+	req := map[string][]string{"symbols": symbols}
+	if err := p.call(ctx, OpGetQuotes, req, &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// GetClock returns the chain clock. The DEX runs 24/7.
+func (p *Provider) GetClock(ctx context.Context) (*types.MarketClock, error) {
+	now := time.Now().UTC().Format(time.RFC3339)
+	return &types.MarketClock{
+		Timestamp: now,
+		IsOpen:    true,
+		NextOpen:  now,
+		NextClose: now,
+	}, nil
+}
+
+func (p *Provider) GetCalendar(ctx context.Context, start, end string) ([]*types.MarketCalendarDay, error) {
+	return nil, nil
+}
@@ -0,0 +1,152 @@
+package lx
+
+import (
+	"context"
+	"encoding/json"
+	"net"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/luxfi/broker/pkg/types"
+	"github.com/luxfi/zap"
+)
+
+// startTestDEX boots a ZAP node that emulates the dex-server handler
+// table. Returns the listen address ("host:port") and a stop func.
+func startTestDEX(t *testing.T) (string, func()) {
+	t.Helper()
+
+	// Pick an ephemeral port up front so the client can dial it.
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	port := ln.Addr().(*net.TCPAddr).Port
+	ln.Close()
+
+	node := zap.NewNode(zap.NodeConfig{
+		NodeID:      "luxdex",
+		ServiceType: dexServiceType,
+		Port:        port,
+		NoDiscovery: true,
+	})
+
+	// Echo handler — decode the JSON payload, build a simple response.
+	handler := func(_ context.Context, _ string, msg *zap.Message) (*zap.Message, error) {
+		op := uint16(msg.Flags() >> 8)
+		reqPayload := msg.Root().Bytes(0)
+
+		var resp any
+		switch op {
+		case OpGetSnapshot:
+			var req map[string]string
+			_ = json.Unmarshal(reqPayload, &req)
+			resp = &types.MarketSnapshot{
+				Symbol:      req["symbol"],
+				LatestTrade: &types.Trade{Price: 100.0},
+			}
+		case OpListAssets:
+			resp = []*types.Asset{{Symbol: "BTC/USD", Provider: "lx"}}
+		case OpCancelOrder:
+			resp = nil
+		default:
+			resp = map[string]string{"echo": strconv.Itoa(int(op))}
+		}
+
+		out, _ := json.Marshal(resp)
+		b := zap.NewBuilder(len(out) + 32)
+		obj := b.StartObject(8)
+		obj.SetBytes(0, out)
+		obj.FinishAsRoot()
+		respMsg, err := zap.Parse(b.Finish())
+		return respMsg, err
+	}
+
+	// Register a handler for every opcode the test exercises.
+	for _, op := range []uint16{OpGetSnapshot, OpListAssets, OpCancelOrder, OpGetPortfolio} {
+		node.Handle(op, handler)
+	}
+
+	if err := node.Start(); err != nil {
+		t.Fatalf("start dex node: %v", err)
+	}
+
+	addr := "127.0.0.1:" + strconv.Itoa(port)
+	return addr, node.Stop
+}
+
+func TestProvider_GetSnapshot(t *testing.T) {
+	addr, stop := startTestDEX(t)
+	defer stop()
+
+	p := New(Config{DEXAddr: addr})
+	defer p.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	snap, err := p.GetSnapshot(ctx, "BTC/USD")
+	if err != nil {
+		t.Fatalf("GetSnapshot: %v", err)
+	}
+	if snap.Symbol != "BTC/USD" {
+		t.Errorf("symbol: got %q, want %q", snap.Symbol, "BTC/USD")
+	}
+	if snap.LatestTrade == nil || snap.LatestTrade.Price != 100.0 {
+		t.Errorf("price: got %v, want 100.0", snap.LatestTrade)
+	}
+}
+
+func TestProvider_ListAssets(t *testing.T) {
+	addr, stop := startTestDEX(t)
+	defer stop()
+
+	p := New(Config{DEXAddr: addr})
+	defer p.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	assets, err := p.ListAssets(ctx, "crypto")
+	if err != nil {
+		t.Fatalf("ListAssets: %v", err)
+	}
+	if len(assets) != 1 || assets[0].Symbol != "BTC/USD" {
+		t.Errorf("assets: got %+v", assets)
+	}
+}
+
+func TestProvider_CancelOrder(t *testing.T) {
+	addr, stop := startTestDEX(t)
+	defer stop()
+
+	p := New(Config{DEXAddr: addr})
+	defer p.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err := p.CancelOrder(ctx, "acct-1", "ord-1"); err != nil {
+		t.Fatalf("CancelOrder: %v", err)
+	}
+}
+
+func TestProvider_Name(t *testing.T) {
+	if (&Provider{}).Name() != "lx" {
+		t.Fatal("Name should be \"lx\"")
+	}
+}
+
+func TestProvider_DefaultsApplied(t *testing.T) {
+	p := New(Config{})
+	if p.cfg.DEXAddr != DefaultDEXAddr {
+		t.Errorf("DEXAddr: got %q, want %q", p.cfg.DEXAddr, DefaultDEXAddr)
+	}
+	if p.cfg.MPCAddr != DefaultMPCAddr {
+		t.Errorf("MPCAddr: got %q, want %q", p.cfg.MPCAddr, DefaultMPCAddr)
+	}
+	if p.cfg.NodeID != "lx-client" {
+		t.Errorf("NodeID: got %q, want lx-client", p.cfg.NodeID)
+	}
+}