mirror of
https://github.com/luxfi/graph.git
synced 2026-09-24 00:26:07 +00:00
fix(dex): wire native 0x9999 V4/DEXFill indexing + uniswapFactories alias + re-genesis self-heal
- indexer: handleSwapV4 (AMM swap/pool), handleInitializeV4 (pool/Market/factory), handleDEXFill (sole CLOB Fill source, scoped to 0x9999), bumpFactory, signed int128 decode - indexer: poll() re-genesis self-heal (head<<cursor by >reorgDepth => rewind to StartBlock) - indexer: correct SigInitializeV4/SigModifyLiquidity hashes; add SigDEXFill + LXSettleAddress - engine: uniswapFactory(ies) alias resolvers (single-source schema-delta fix) - tests: dex_v4_test (V4/Fill/regenesis/scoping), e2e dex_native_test (dex schema + alias) - fix pre-existing e2e/wasm_test.go missing os import
This commit is contained in:
@@ -0,0 +1,115 @@
|
||||
// Native-DEX (CLOB) + uniswap-alias e2e.
|
||||
//
|
||||
// Proves the explorer's `dex` graph schema serves the Fill/Market entities the
|
||||
// 0x9999 indexer writes, and that the uniswap-v2 `uniswapFactories` spelling
|
||||
// resolves against the same factory rows as `factories`.
|
||||
//
|
||||
// Run: go test -v ./e2e/ -tags=e2e
|
||||
//
|
||||
//go:build e2e
|
||||
|
||||
package e2e
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/luxfi/graph/engine"
|
||||
"github.com/luxfi/graph/storage"
|
||||
)
|
||||
|
||||
// TestUniswapFactoriesAlias proves the uniswap-v2 spelling resolves to the same
|
||||
// data as `factories` (the schema delta the exchange-api used to hit as
|
||||
// "unknown field: uniswapFactories").
|
||||
func TestUniswapFactoriesAlias(t *testing.T) {
|
||||
store, err := storage.New(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.Init(nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store.SeedFactory("1", &storage.SeedFactoryData{
|
||||
PoolCount: 7, TxCount: 70, TotalVolumeUSD: "100.0", TotalValueLockedUSD: "200.0",
|
||||
})
|
||||
|
||||
eng := engine.New(store, nil)
|
||||
if err := eng.LoadBuiltin("amm"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
resp := eng.Execute(nil, &engine.Request{Query: `{
|
||||
uniswapFactories(first: 1) {
|
||||
poolCount
|
||||
totalVolumeUSD
|
||||
totalValueLockedUSD
|
||||
}
|
||||
}`})
|
||||
if len(resp.Errors) > 0 {
|
||||
t.Fatalf("uniswapFactories errors: %v", resp.Errors)
|
||||
}
|
||||
data := resp.Data.(map[string]interface{})
|
||||
arr, ok := data["uniswapFactories"].([]interface{})
|
||||
if !ok || len(arr) == 0 {
|
||||
t.Fatalf("expected uniswapFactories array, got %#v", data["uniswapFactories"])
|
||||
}
|
||||
f := arr[0].(map[string]interface{})
|
||||
if f["poolCount"] == nil {
|
||||
t.Error("missing poolCount via uniswapFactories alias")
|
||||
}
|
||||
|
||||
// Singular alias too.
|
||||
resp = eng.Execute(nil, &engine.Request{Query: `{ uniswapFactory(id: "1") { poolCount } }`})
|
||||
if len(resp.Errors) > 0 {
|
||||
t.Fatalf("uniswapFactory errors: %v", resp.Errors)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNativeDexSchema proves the `dex` (CLOB) schema serves Order/Market/Fill
|
||||
// entities — the shapes the 0x9999 indexer writes via SetEntity.
|
||||
func TestNativeDexSchema(t *testing.T) {
|
||||
store, err := storage.New(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.Init(nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Seed the entity shapes the indexer's writeFill / handleInitializeV4 emit.
|
||||
store.SetEntity("Market", "0xpool", map[string]interface{}{
|
||||
"id": "0xpool", "symbol": "LUX/LUSD", "volume24h": "2500", "tradeCount": int64(1),
|
||||
})
|
||||
store.SetEntity("Fill", "0xtx#0", map[string]interface{}{
|
||||
"id": "0xtx#0", "market": "0xpool", "taker": "0xaa", "amountOut": "2500", "timestamp": int64(16),
|
||||
})
|
||||
|
||||
eng := engine.New(store, nil)
|
||||
if err := eng.LoadBuiltin("dex"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
resp := eng.Execute(nil, &engine.Request{Query: `{
|
||||
markets(first: 10) { id symbol volume24h tradeCount }
|
||||
fills(first: 10) { id market taker amountOut }
|
||||
}`})
|
||||
if len(resp.Errors) > 0 {
|
||||
t.Fatalf("dex schema errors: %v", resp.Errors)
|
||||
}
|
||||
data := resp.Data.(map[string]interface{})
|
||||
|
||||
markets, ok := data["markets"].([]interface{})
|
||||
if !ok || len(markets) != 1 {
|
||||
t.Fatalf("expected 1 market, got %#v", data["markets"])
|
||||
}
|
||||
if markets[0].(map[string]interface{})["symbol"] != "LUX/LUSD" {
|
||||
t.Errorf("market symbol = %v", markets[0].(map[string]interface{})["symbol"])
|
||||
}
|
||||
|
||||
fills, ok := data["fills"].([]interface{})
|
||||
if !ok || len(fills) != 1 {
|
||||
t.Fatalf("expected 1 fill, got %#v", data["fills"])
|
||||
}
|
||||
if fills[0].(map[string]interface{})["amountOut"] != "2500" {
|
||||
t.Errorf("fill amountOut = %v", fills[0].(map[string]interface{})["amountOut"])
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,7 @@ package e2e
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/luxfi/graph/wasm"
|
||||
|
||||
@@ -18,6 +18,12 @@ func (e *Engine) registerAMMResolvers() {
|
||||
// v2 + v3 unified
|
||||
e.resolvers["factory"] = e.resolveFactory
|
||||
e.resolvers["factories"] = e.resolveFactories
|
||||
// The Graph's uniswap-v2 subgraph names the factory entity `uniswapFactory(ies)`
|
||||
// (the v3 subgraph uses `factory(ies)`). The exchange frontend and exchange-api
|
||||
// issue both spellings depending on protocol version. They resolve to the same
|
||||
// underlying factory rows, so alias them here — one source, every consumer.
|
||||
e.resolvers["uniswapFactory"] = e.resolveFactory
|
||||
e.resolvers["uniswapFactories"] = e.resolveFactories
|
||||
e.resolvers["bundle"] = e.resolveBundle
|
||||
e.resolvers["bundles"] = e.resolveBundles
|
||||
e.resolvers["pool"] = e.resolvePool
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
package indexer
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/big"
|
||||
"testing"
|
||||
|
||||
"github.com/luxfi/graph/storage"
|
||||
)
|
||||
|
||||
// word renders a non-negative big.Int as a 64-hex-char ABI word.
|
||||
func word(n *big.Int) string { return fmt.Sprintf("%064x", n) }
|
||||
|
||||
// signedWord renders a (possibly negative) big.Int as a two's-complement
|
||||
// 64-hex-char ABI word — the on-wire form for V4 int128 amounts.
|
||||
func signedWord(n *big.Int) string {
|
||||
if n.Sign() >= 0 {
|
||||
return word(n)
|
||||
}
|
||||
twos := new(big.Int).Add(new(big.Int).Lsh(big.NewInt(1), 256), n)
|
||||
return fmt.Sprintf("%064x", twos)
|
||||
}
|
||||
|
||||
// topic32 left-pads a hex value to a 32-byte (66-char 0x) topic.
|
||||
func topic32(hexNo0x string) string { return "0x" + fmt.Sprintf("%064s", hexNo0x) }
|
||||
|
||||
// addrTopic encodes an address as an indexed topic (left-padded to 32 bytes).
|
||||
func addrTopic(addr string) string {
|
||||
return "0x" + fmt.Sprintf("%064s", addr[2:])
|
||||
}
|
||||
|
||||
func newMemSQLiteStore(t *testing.T) *storage.Store {
|
||||
t.Helper()
|
||||
s, err := storage.New(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.Init(nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { s.Close() })
|
||||
return s
|
||||
}
|
||||
|
||||
func TestDecodeInt256_SignedV4Amounts(t *testing.T) {
|
||||
cases := []*big.Int{
|
||||
big.NewInt(0),
|
||||
big.NewInt(1),
|
||||
big.NewInt(251250000),
|
||||
big.NewInt(-1),
|
||||
big.NewInt(-251250000),
|
||||
}
|
||||
for _, want := range cases {
|
||||
data := "0x" + signedWord(want)
|
||||
got := decodeInt256(data, 0)
|
||||
if got.Cmp(want) != 0 {
|
||||
t.Errorf("decodeInt256(%s) = %s, want %s", want, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A V4 swap records ONLY an AMM swap (with signed legs), never a CLOB Fill — the
|
||||
// Fill has a single source (the DEXFill event, see TestHandleDEXFill_*). This
|
||||
// holds whether or not the emitter is the 0x9999 settlement precompile: the
|
||||
// AMM-side view and the CLOB-side view come from distinct events.
|
||||
func TestHandleSwapV4_WritesAmmSwapOnly(t *testing.T) {
|
||||
s := newMemSQLiteStore(t)
|
||||
idx := NewWithConfig(Config{RPC: "http://unused", PoolManager: LXSettleAddress}, s)
|
||||
|
||||
poolID := topic32("abcd")
|
||||
taker := "0x00000000000000000000000000000000000000aa"
|
||||
amount0 := big.NewInt(1000) // taker pays in
|
||||
amount1 := big.NewInt(-2500) // taker receives out (negative to pool)
|
||||
|
||||
l := &logEntry{
|
||||
Address: LXSettleAddress, // even at 0x9999 — still AMM-only here
|
||||
Topics: []string{SigSwapV4, poolID, addrTopic(taker)},
|
||||
Data: "0x" + signedWord(amount0) + signedWord(amount1) + word(big.NewInt(0)) + word(big.NewInt(0)) + word(big.NewInt(0)) + word(big.NewInt(3000)),
|
||||
BlockNumber: "0x10",
|
||||
TransactionHash: "0xdeadbeef",
|
||||
LogIndex: "0x0",
|
||||
}
|
||||
idx.processLog(l)
|
||||
|
||||
// AMM swap present with correctly signed legs.
|
||||
sw, err := s.GetSwaps(nil, 10, "timestamp", "desc", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
swaps := sw.([]interface{})
|
||||
if len(swaps) != 1 {
|
||||
t.Fatalf("want 1 amm swap, got %d", len(swaps))
|
||||
}
|
||||
if swaps[0].(map[string]interface{})["amount1"] != "-2500" {
|
||||
t.Errorf("amount1 = %v, want -2500 (signed)", swaps[0].(map[string]interface{})["amount1"])
|
||||
}
|
||||
|
||||
// No Fill — a V4 Swap is never a CLOB fill (that comes from DEXFill only).
|
||||
fills, _ := s.ListByType("Fill", 10)
|
||||
if fills != nil && len(fills.([]interface{})) != 0 {
|
||||
t.Fatalf("V4 swap must not create a fill (DEXFill is the only source), got %d", len(fills.([]interface{})))
|
||||
}
|
||||
}
|
||||
|
||||
// An explicit DEXFill at 0x9999 creates a Fill and rolls market volume; a
|
||||
// spoofed DEXFill from another address is ignored.
|
||||
func TestHandleDEXFill_ScopedToPoolManager(t *testing.T) {
|
||||
s := newMemSQLiteStore(t)
|
||||
idx := NewWithConfig(Config{RPC: "http://unused"}, s) // default 0x9999
|
||||
|
||||
poolID := topic32("cafe")
|
||||
taker := "0x00000000000000000000000000000000000000cc"
|
||||
mkFill := func(addr, tx string, out int64) *logEntry {
|
||||
return &logEntry{
|
||||
Address: addr,
|
||||
Topics: []string{SigDEXFill, poolID, addrTopic(taker)},
|
||||
Data: "0x" + word(big.NewInt(out)) + word(big.NewInt(99)),
|
||||
BlockNumber: "0x30",
|
||||
TransactionHash: tx,
|
||||
LogIndex: "0x0",
|
||||
}
|
||||
}
|
||||
// Legit fill at 0x9999.
|
||||
idx.processLog(mkFill(LXSettleAddress, "0xaa", 400))
|
||||
// Spoofed fill from a rogue address — must be dropped.
|
||||
idx.processLog(mkFill("0x0000000000000000000000000000000000009998", "0xbb", 999999))
|
||||
|
||||
fills, _ := s.ListByType("Fill", 10)
|
||||
fl := fills.([]interface{})
|
||||
if len(fl) != 1 {
|
||||
t.Fatalf("want exactly 1 fill (spoof dropped), got %d", len(fl))
|
||||
}
|
||||
mk, _ := s.GetByType("Market", poolID)
|
||||
mm := mk.(map[string]interface{})
|
||||
if mm["volume24h"] != "400" {
|
||||
t.Errorf("market volume24h = %v, want 400 (spoof excluded)", mm["volume24h"])
|
||||
}
|
||||
}
|
||||
|
||||
// Initialize at 0x9999 creates an AMM pool, bumps the factory aggregate, and
|
||||
// creates a DEX Market.
|
||||
func TestHandleInitializeV4_PoolMarketFactory(t *testing.T) {
|
||||
s := newMemSQLiteStore(t)
|
||||
idx := NewWithConfig(Config{RPC: "http://unused"}, s)
|
||||
|
||||
poolID := topic32("d00d")
|
||||
cur0 := "0x0000000000000000000000000000000000000011"
|
||||
cur1 := "0x0000000000000000000000000000000000000022"
|
||||
l := &logEntry{
|
||||
Address: LXSettleAddress,
|
||||
Topics: []string{SigInitializeV4, poolID, addrTopic(cur0), addrTopic(cur1)},
|
||||
Data: "0x" + word(big.NewInt(3000)) + word(big.NewInt(60)) + word(big.NewInt(0)) + word(big.NewInt(0)) + word(big.NewInt(0)),
|
||||
BlockNumber: "0x1",
|
||||
}
|
||||
idx.processLog(l)
|
||||
|
||||
if p, _ := s.GetPool(nil, poolID); p == nil {
|
||||
t.Fatal("expected pool from InitializeV4")
|
||||
}
|
||||
f, _ := s.GetFactory(nil, "1")
|
||||
if f == nil || asInt64(f.(map[string]interface{})["poolCount"]) != 1 {
|
||||
t.Fatalf("expected factory poolCount=1, got %v", f)
|
||||
}
|
||||
if mk, _ := s.GetByType("Market", poolID); mk == nil {
|
||||
t.Fatal("expected DEX Market from InitializeV4 at 0x9999")
|
||||
}
|
||||
}
|
||||
|
||||
// The re-genesis self-heal: a persisted cursor far above a freshly-relaunched
|
||||
// chain's head must rewind to StartBlock so the new chain indexes from genesis.
|
||||
func TestPoll_RegenesisSelfHeal(t *testing.T) {
|
||||
s := newMemSQLiteStore(t)
|
||||
s.SetLastBlock(5_000_000) // stale cursor from the previous chain
|
||||
idx := NewWithConfig(Config{RPC: "http://unused", StartBlock: 0}, s)
|
||||
if idx.lastBlock != 5_000_000 {
|
||||
t.Fatalf("precondition: lastBlock=%d", idx.lastBlock)
|
||||
}
|
||||
|
||||
// Simulate the head as observed on a fresh chain (head=3, far below cursor).
|
||||
idx.maybeResetForRegenesis(3)
|
||||
if idx.lastBlock != 0 {
|
||||
t.Fatalf("expected rewind to StartBlock 0, got %d", idx.lastBlock)
|
||||
}
|
||||
if s.GetLastBlock() != 0 {
|
||||
t.Fatalf("expected persisted cursor reset to 0, got %d", s.GetLastBlock())
|
||||
}
|
||||
}
|
||||
|
||||
// A shallow reorg (head dips a few blocks) must NOT trigger a reset.
|
||||
func TestPoll_ShallowReorgNoReset(t *testing.T) {
|
||||
s := newMemSQLiteStore(t)
|
||||
s.SetLastBlock(1000)
|
||||
idx := NewWithConfig(Config{RPC: "http://unused", StartBlock: 0}, s)
|
||||
|
||||
idx.maybeResetForRegenesis(995) // 5 blocks back — within reorgDepth
|
||||
if idx.lastBlock != 1000 {
|
||||
t.Fatalf("shallow reorg must not reset; lastBlock=%d", idx.lastBlock)
|
||||
}
|
||||
}
|
||||
+20
-4
@@ -156,17 +156,27 @@ func SecuritiesTopics() []string {
|
||||
}
|
||||
}
|
||||
|
||||
// V4 events (PoolManager singleton pattern)
|
||||
// V4 events — Lux LXPool / 0x9999 PoolManager.
|
||||
//
|
||||
// These topic0 hashes are keccak256 of the EXACT signatures the Lux precompile
|
||||
// emits (see ~/work/lux/precompile/dex/events.go). The Lux Swap signature is
|
||||
// NOT the upstream Uniswap-v4 one: it carries a trailing uint24 `fee`. Indexing
|
||||
// the upstream hash would match nothing.
|
||||
const (
|
||||
// Initialize(bytes32 indexed id, address indexed currency0, address indexed currency1, uint24 fee, int24 tickSpacing, address hooks, uint160 sqrtPriceX96, int24 tick)
|
||||
SigInitializeV4 = "0x344560c924012b32cbed54ad0e83e24c2cf3e723de46e06c72e6ab3463a7a8c0"
|
||||
SigInitializeV4 = "0xdd466e674ea557f56295e2d0218a125ea4b4f0f6f3307b95f85e6110838d6438"
|
||||
|
||||
// ModifyLiquidity(bytes32 indexed id, address indexed sender, int24 tickLower, int24 tickUpper, int256 liquidityDelta, bytes32 salt)
|
||||
SigModifyLiquidity = "0xf208f4912782fd87d4a358e8291c46e9fefb649e838a67dcaa5db23503a4f53b"
|
||||
SigModifyLiquidity = "0xf208f4912782fd25c7f114ca3723a2d5dd6f3bcc3ac8db5af63baa85f711d5ec"
|
||||
|
||||
// Swap(bytes32 indexed id, address indexed sender, int128 amount0, int128 amount1, uint160 sqrtPriceX96After, uint128 liquidityAfter, int24 tickAfter)
|
||||
// Swap(bytes32 indexed id, address indexed sender, int128 amount0, int128 amount1, uint160 sqrtPriceX96After, uint128 liquidityAfter, int24 tickAfter, uint24 fee)
|
||||
SigSwapV4 = "0x40e9cecb9f5f1f1c5b9c97dec2917b7ee92e57ba5563708daca94dd84ad7112f"
|
||||
|
||||
// DEXFill(bytes32 indexed poolId, address indexed taker, uint256 amountOut, uint256 blockNumber)
|
||||
// Emitted by the 0x9999 settlement precompile on a Phase-B credit (a settled
|
||||
// native-CLOB fill). This is the indexable signal for the DEX (CLOB) graph.
|
||||
SigDEXFill = "0x6f744d074efc4fa8512636853f8a4c67842f230d4279421c7599dc5c5daf9874"
|
||||
|
||||
// Subscription(uint256 indexed tokenId, address indexed subscriber)
|
||||
SigSubscription = "0x572f161235911da04685a68c6b07422b4ba12dab20405b8b230c2420f49f1266"
|
||||
|
||||
@@ -174,6 +184,12 @@ const (
|
||||
SigUnsubscription = "0x6112cc683e05f5c607f9d5f0f57a0f0be2ca872a8f94ed4b6e2d09da2f1d1882"
|
||||
)
|
||||
|
||||
// LXSettleAddress is the Lux DEX settlement precompile (LP-9999) — the V4
|
||||
// PoolManager money path. Settled native-DEX fills originate here. Used to
|
||||
// distinguish a native-CLOB settlement from a vanilla AMM-v4 swap so the DEX
|
||||
// (CLOB) graph schema only ingests genuine fills.
|
||||
const LXSettleAddress = "0x0000000000000000000000000000000000009999"
|
||||
|
||||
// Known factory addresses per network.
|
||||
type NetworkConfig struct {
|
||||
Name string `json:"name" yaml:"name"`
|
||||
|
||||
+238
-8
@@ -26,21 +26,56 @@ type Status struct {
|
||||
IndexedEvents uint64 `json:"indexedEvents"`
|
||||
}
|
||||
|
||||
// Config tunes an Indexer. The zero value is valid: PoolManager defaults to the
|
||||
// canonical 0x9999 settlement address and StartBlock to 0 (index from genesis).
|
||||
type Config struct {
|
||||
RPC string
|
||||
// PoolManager is the DEX settlement precompile address (0x9999). Logs from
|
||||
// this address are additionally derived into the DEX (CLOB) graph schema as
|
||||
// Fill/Market entities. Lower-cased on construction for case-insensitive
|
||||
// compare against eth_getLogs `address` fields.
|
||||
PoolManager string
|
||||
// StartBlock is the genesis-relative block to (re)index from. On a clean
|
||||
// chain relaunch the indexer rewinds to this value (see poll's reorg guard).
|
||||
StartBlock uint64
|
||||
}
|
||||
|
||||
// reorgDepth is how far the chain head may legitimately move backwards (deep
|
||||
// reorg) before we treat a lower head as a chain RESET (re-genesis) rather than
|
||||
// a reorg. A re-genesised chain reports a head far below our persisted cursor;
|
||||
// a normal reorg is shallow. This avoids wiping progress on a transient blip
|
||||
// while still self-healing across a clean relaunch.
|
||||
const reorgDepth = 128
|
||||
|
||||
// Indexer watches an EVM RPC and writes events to storage.
|
||||
type Indexer struct {
|
||||
rpc string
|
||||
store *storage.Store
|
||||
client *http.Client
|
||||
rpc string
|
||||
poolManager string // lower-cased 0x9999 settlement address
|
||||
startBlock uint64
|
||||
store *storage.Store
|
||||
client *http.Client
|
||||
|
||||
lastBlock uint64
|
||||
status Status
|
||||
}
|
||||
|
||||
// New creates an indexer connected to the given RPC endpoint.
|
||||
// New creates an indexer connected to the given RPC endpoint with default
|
||||
// settlement address and a genesis (block 0) start.
|
||||
func New(rpc string, store *storage.Store) *Indexer {
|
||||
return NewWithConfig(Config{RPC: rpc}, store)
|
||||
}
|
||||
|
||||
// NewWithConfig creates an indexer from an explicit Config.
|
||||
func NewWithConfig(cfg Config, store *storage.Store) *Indexer {
|
||||
pm := cfg.PoolManager
|
||||
if pm == "" {
|
||||
pm = LXSettleAddress
|
||||
}
|
||||
idx := &Indexer{
|
||||
rpc: rpc,
|
||||
store: store,
|
||||
rpc: cfg.RPC,
|
||||
poolManager: strings.ToLower(pm),
|
||||
startBlock: cfg.StartBlock,
|
||||
store: store,
|
||||
client: &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
Transport: &http.Transport{
|
||||
@@ -51,6 +86,9 @@ func New(rpc string, store *storage.Store) *Indexer {
|
||||
},
|
||||
}
|
||||
idx.lastBlock = idx.store.GetLastBlock()
|
||||
if idx.lastBlock < cfg.StartBlock {
|
||||
idx.lastBlock = cfg.StartBlock
|
||||
}
|
||||
return idx
|
||||
}
|
||||
|
||||
@@ -59,6 +97,22 @@ func (idx *Indexer) Status() Status {
|
||||
return idx.status
|
||||
}
|
||||
|
||||
// maybeResetForRegenesis rewinds the cursor to StartBlock when the observed
|
||||
// chain head is far enough below our persisted cursor to indicate a clean
|
||||
// relaunch (fresh genesis) rather than a shallow reorg. A re-genesised chain
|
||||
// reports a head near zero while our store still holds the old chain's height;
|
||||
// without this rewind, `latest <= lastBlock` would be true forever and the new
|
||||
// chain would never be indexed. A normal reorg is shallow (< reorgDepth) and is
|
||||
// ignored here so transient head dips don't wipe progress.
|
||||
func (idx *Indexer) maybeResetForRegenesis(latest uint64) {
|
||||
if idx.lastBlock > idx.startBlock+reorgDepth && latest+reorgDepth < idx.lastBlock {
|
||||
log.Printf("[indexer] head %d << cursor %d (>%d behind) — chain reset detected, rewinding to %d",
|
||||
latest, idx.lastBlock, reorgDepth, idx.startBlock)
|
||||
idx.lastBlock = idx.startBlock
|
||||
idx.store.SetLastBlock(idx.startBlock)
|
||||
}
|
||||
}
|
||||
|
||||
// Run starts the indexer loop. Blocks until ctx is cancelled.
|
||||
func (idx *Indexer) Run(ctx context.Context) error {
|
||||
log.Printf("[indexer] starting — rpc=%s lastBlock=%d", idx.rpc, idx.lastBlock)
|
||||
@@ -140,6 +194,8 @@ func knownTopics() []string {
|
||||
SigPoolCreated, SigInitialize, SigSwapV3, SigMintV3, SigBurnV3,
|
||||
SigCollect, SigFlash,
|
||||
SigInitializeV4, SigModifyLiquidity, SigSwapV4,
|
||||
// Native DEX (CLOB) settlement at 0x9999.
|
||||
SigDEXFill,
|
||||
}
|
||||
// Securities — full ERC-3643 + ONCHAINID surface.
|
||||
return append(topics, SecuritiesTopics()...)
|
||||
@@ -172,6 +228,9 @@ func (idx *Indexer) poll(ctx context.Context) error {
|
||||
return fmt.Errorf("parse hex block: %w", err)
|
||||
}
|
||||
|
||||
// Re-genesis self-heal (see maybeResetForRegenesis).
|
||||
idx.maybeResetForRegenesis(latest)
|
||||
|
||||
// Nothing new
|
||||
if latest <= idx.lastBlock {
|
||||
return nil
|
||||
@@ -239,9 +298,14 @@ func (idx *Indexer) processLog(l *logEntry) {
|
||||
idx.handlePoolCreated(l)
|
||||
case SigTransfer:
|
||||
idx.handleTransfer(l, txHash, logIdx)
|
||||
case SigInitializeV4:
|
||||
idx.handleInitializeV4(l)
|
||||
case SigSwapV4:
|
||||
idx.handleSwapV4(l, blockNum, txHash, logIdx)
|
||||
case SigDEXFill:
|
||||
idx.handleDEXFill(l, blockNum, txHash, logIdx)
|
||||
case SigMintV2, SigMintV3, SigBurnV2, SigBurnV3, SigSync,
|
||||
SigCollect, SigFlash, SigInitialize,
|
||||
SigInitializeV4, SigModifyLiquidity, SigSwapV4:
|
||||
SigCollect, SigFlash, SigInitialize, SigModifyLiquidity:
|
||||
// Recognized but storage for these types not yet wired
|
||||
|
||||
// ── ERC-3643 IToken ───────────────────────────────────────────────
|
||||
@@ -394,6 +458,7 @@ func (idx *Indexer) handlePairCreated(l *logEntry) {
|
||||
})
|
||||
idx.store.SeedToken(token0, &storage.SeedTokenData{Symbol: token0[:8], Name: token0, Decimals: 18})
|
||||
idx.store.SeedToken(token1, &storage.SeedTokenData{Symbol: token1[:8], Name: token1, Decimals: 18})
|
||||
idx.bumpFactory(1)
|
||||
}
|
||||
|
||||
func (idx *Indexer) handlePoolCreated(l *logEntry) {
|
||||
@@ -419,6 +484,7 @@ func (idx *Indexer) handlePoolCreated(l *logEntry) {
|
||||
})
|
||||
idx.store.SeedToken(token0, &storage.SeedTokenData{Symbol: token0[:8], Name: token0, Decimals: 18})
|
||||
idx.store.SeedToken(token1, &storage.SeedTokenData{Symbol: token1[:8], Name: token1, Decimals: 18})
|
||||
idx.bumpFactory(1)
|
||||
}
|
||||
|
||||
func (idx *Indexer) handleTransfer(l *logEntry, txHash, logIdx string) {
|
||||
@@ -429,3 +495,167 @@ func (idx *Indexer) handleTransfer(l *logEntry, txHash, logIdx string) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// decodeInt256 reads a 32-byte word as a signed two's-complement big.Int. V4
|
||||
// emits amount0/amount1 as int128 (sign-extended into a 32-byte word), so a
|
||||
// straight unsigned read would misreport negative legs as astronomically large.
|
||||
func decodeInt256(data string, wordIndex int) *big.Int {
|
||||
n := decodeUint256(data, wordIndex)
|
||||
// Words with the high bit set are negative: subtract 2^256.
|
||||
if n.Bit(255) == 1 {
|
||||
twoExp256 := new(big.Int).Lsh(big.NewInt(1), 256)
|
||||
n.Sub(n, twoExp256)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// bumpFactory increments the singleton factory ("1") aggregate so the
|
||||
// dashboard `factory`/`factories`/`uniswapFactories` query returns non-zero
|
||||
// pool/tx counts. One place, called from every pool/pair creation handler —
|
||||
// the factory aggregate was previously only ever seeded by tests, leaving the
|
||||
// exchange landing page's TVL/volume header empty even with live pools.
|
||||
func (idx *Indexer) bumpFactory(deltaPools int64) {
|
||||
f, _ := idx.store.GetFactory(nil, "1")
|
||||
var pc, tc int64
|
||||
if m, ok := f.(map[string]interface{}); ok {
|
||||
pc = asInt64(m["poolCount"])
|
||||
tc = asInt64(m["txCount"])
|
||||
}
|
||||
idx.store.SeedFactory("1", &storage.SeedFactoryData{
|
||||
PoolCount: pc + deltaPools,
|
||||
TxCount: tc + 1,
|
||||
})
|
||||
}
|
||||
|
||||
// asInt64 best-effort coerces a JSON-decoded numeric (float64/json.Number/int)
|
||||
// to int64 for aggregate accumulation.
|
||||
func asInt64(v interface{}) int64 {
|
||||
switch n := v.(type) {
|
||||
case int64:
|
||||
return n
|
||||
case int:
|
||||
return int64(n)
|
||||
case float64:
|
||||
return int64(n)
|
||||
default:
|
||||
var out int64
|
||||
fmt.Sscanf(fmt.Sprint(v), "%d", &out)
|
||||
return out
|
||||
}
|
||||
}
|
||||
|
||||
// handleInitializeV4 records a V4 pool (Initialize at the PoolManager). The
|
||||
// pool id is the indexed bytes32 `id`; token0/token1 are the indexed
|
||||
// currency0/currency1. Fee is the first data word. Pools created by the 0x9999
|
||||
// settlement manager are ALSO surfaced as DEX (CLOB) Market entities.
|
||||
func (idx *Indexer) handleInitializeV4(l *logEntry) {
|
||||
if len(l.Topics) < 4 {
|
||||
return
|
||||
}
|
||||
poolID := l.Topics[1] // full 32-byte id
|
||||
token0 := topicAddr(l.Topics[2])
|
||||
token1 := topicAddr(l.Topics[3])
|
||||
fee := decodeUint256(l.Data, 0).Int64()
|
||||
|
||||
idx.store.SeedPool(poolID, &storage.SeedPoolData{
|
||||
Token0: token0,
|
||||
Token1: token1,
|
||||
FeeTier: fee,
|
||||
})
|
||||
idx.store.SeedToken(token0, &storage.SeedTokenData{Symbol: shortAddr(token0), Name: token0, Decimals: 18})
|
||||
idx.store.SeedToken(token1, &storage.SeedTokenData{Symbol: shortAddr(token1), Name: token1, Decimals: 18})
|
||||
idx.bumpFactory(1)
|
||||
|
||||
if idx.isPoolManager(l.Address) {
|
||||
idx.store.SetEntity("Market", poolID, map[string]interface{}{
|
||||
"id": poolID, "symbol": poolID, "baseToken": token0, "quoteToken": token1,
|
||||
"feeTier": fee, "volume24h": "0", "tradeCount": 0, "lastPrice": "0",
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// handleSwapV4 records a V4 swap (Swap at any V4 PoolManager) as an AMM swap —
|
||||
// amount0/amount1 are signed int128. This is the AMM-side view only. The DEX
|
||||
// (CLOB) Fill is NOT derived here: a settled native fill is recorded from its
|
||||
// own authoritative DEXFill event (handleDEXFill), so the Fill has exactly one
|
||||
// source. A vanilla (non-0x9999) V4 pool emits only this AMM swap, never a fill.
|
||||
func (idx *Indexer) handleSwapV4(l *logEntry, blockNum uint64, txHash, logIdx string) {
|
||||
if len(l.Topics) < 3 {
|
||||
return
|
||||
}
|
||||
id := fmt.Sprintf("%s#%s", txHash, logIdx)
|
||||
poolID := l.Topics[1]
|
||||
sender := topicAddr(l.Topics[2])
|
||||
amount0 := decodeInt256(l.Data, 0)
|
||||
amount1 := decodeInt256(l.Data, 1)
|
||||
|
||||
idx.store.SeedSwap(id, &storage.SeedSwapData{
|
||||
Timestamp: int64(blockNum),
|
||||
Pool: poolID,
|
||||
Amount0: amount0.String(),
|
||||
Amount1: amount1.String(),
|
||||
AmountUSD: "0",
|
||||
Sender: sender,
|
||||
})
|
||||
idx.bumpFactory(0)
|
||||
}
|
||||
|
||||
// handleDEXFill records a native-CLOB settlement (DEXFill at 0x9999). This is
|
||||
// the authoritative settled-fill signal: poolId + taker indexed, amountOut +
|
||||
// blockNumber in data. Only logs from the configured PoolManager are honored —
|
||||
// a spoofed DEXFill from another address is ignored.
|
||||
func (idx *Indexer) handleDEXFill(l *logEntry, blockNum uint64, txHash, logIdx string) {
|
||||
if len(l.Topics) < 3 || !idx.isPoolManager(l.Address) {
|
||||
return
|
||||
}
|
||||
id := fmt.Sprintf("%s#%s", txHash, logIdx)
|
||||
poolID := l.Topics[1]
|
||||
taker := topicAddr(l.Topics[2])
|
||||
amountOut := decodeUint256(l.Data, 0)
|
||||
idx.writeFill(id, poolID, taker, amountOut, int64(blockNum), txHash)
|
||||
}
|
||||
|
||||
// writeFill persists a DEX (CLOB) Fill entity and rolls its volume into the
|
||||
// market aggregate. Shared by the V4-swap-at-0x9999 derivation and the explicit
|
||||
// DEXFill handler so there is one fill-shape and one market-rollup path.
|
||||
func (idx *Indexer) writeFill(id, poolID, taker string, amountOut *big.Int, ts int64, txHash string) {
|
||||
idx.store.SetEntity("Fill", id, map[string]interface{}{
|
||||
"id": id, "market": poolID, "taker": taker,
|
||||
"amountOut": amountOut.String(), "timestamp": ts, "txHash": txHash,
|
||||
})
|
||||
|
||||
// Market rollup — accumulate volume + trade count, keyed by pool id.
|
||||
var vol = new(big.Int)
|
||||
var tc int64
|
||||
if m, _ := idx.store.GetByType("Market", poolID); m != nil {
|
||||
if mm, ok := m.(map[string]interface{}); ok {
|
||||
vol.SetString(fmt.Sprint(mm["volume24h"]), 10)
|
||||
tc = asInt64(mm["tradeCount"])
|
||||
vol.Add(vol, amountOut)
|
||||
mm["volume24h"] = vol.String()
|
||||
mm["tradeCount"] = tc + 1
|
||||
mm["lastUpdate"] = ts
|
||||
idx.store.SetEntity("Market", poolID, mm)
|
||||
return
|
||||
}
|
||||
}
|
||||
// First fill for a market we never saw an Initialize for — create a stub.
|
||||
idx.store.SetEntity("Market", poolID, map[string]interface{}{
|
||||
"id": poolID, "symbol": poolID, "volume24h": amountOut.String(),
|
||||
"tradeCount": int64(1), "lastUpdate": ts,
|
||||
})
|
||||
}
|
||||
|
||||
// isPoolManager reports whether a log address is the configured 0x9999
|
||||
// settlement precompile (case-insensitive).
|
||||
func (idx *Indexer) isPoolManager(addr string) bool {
|
||||
return strings.EqualFold(addr, idx.poolManager)
|
||||
}
|
||||
|
||||
// shortAddr returns a stable short symbol for an unknown token address.
|
||||
func shortAddr(addr string) string {
|
||||
if len(addr) >= 8 {
|
||||
return addr[:8]
|
||||
}
|
||||
return addr
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user