Merge fix/native-dex-indexing: native 0x9999 V4/DEXFill indexing + dex/amm subgraphs + re-genesis self-heal

e2e/dex_native_test.go

@@ -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"])
+	}
+}

e2e/wasm_test.go

@@ -4,6 +4,7 @@ package e2e
 
 import (
 	"context"
+	"os"
 	"testing"
 
 	"github.com/luxfi/graph/wasm"

engine/amm.go

@@ -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

engine/dex_route_test.go

@@ -0,0 +1,107 @@
+package engine
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/luxfi/graph/storage"
+)
+
+// dex_route_test.go is the end-to-end proof for the exchange-api FIX D routing: the
+// FE's market/fill (CLOB) query shapes resolve against the `dex` schema and NOT the
+// `amm` schema. Red proved live that `amm` returns "unknown field: markets"; this
+// pins that markets/fills ARE dex-schema fields that resolve to the entities the
+// 0x9999 indexer writes (Market from the Initialize log, Fill from DEXFill), so the
+// exchange-api routing target (/v1/graph/cchain/dex/graphql) is correct.
+
+func newDexEngine(t *testing.T) (*Engine, *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() })
+	eng := New(s, &Config{MaxQueryDepth: 10, MaxResultSize: 1 << 20, QueryTimeoutMs: 30000})
+	if err := eng.LoadBuiltin("dex"); err != nil {
+		t.Fatalf("LoadBuiltin(dex): %v", err)
+	}
+	return eng, s
+}
+
+// TestDexSchema_ResolvesMarketsAndFills proves the FE's CLOB query shapes resolve on
+// the dex schema against the exact entities the native 0x9999 indexer produces.
+func TestDexSchema_ResolvesMarketsAndFills(t *testing.T) {
+	eng, s := newDexEngine(t)
+
+	// Seed exactly what the indexer writes: a rich Market (from the 0x9999 Initialize
+	// log) and a Fill (from a DEXFill at 0x9999).
+	poolID := "0x" + strings.Repeat("ab", 32)
+	s.SetEntity("Market", poolID, map[string]interface{}{
+		"id": poolID, "symbol": poolID, "baseToken": "0x11", "quoteToken": "0x22",
+		"feeTier": int64(3000), "volume24h": "400", "tradeCount": int64(1), "lastPrice": "0",
+	})
+	s.SetEntity("Fill", "0xtx#0", map[string]interface{}{
+		"id": "0xtx#0", "market": poolID, "taker": "0xaa", "amountOut": "400", "timestamp": int64(9),
+	})
+
+	// The FE markets query shape.
+	resp := eng.Execute(context.Background(), &Request{
+		Query: "{ markets(first: 5){ id baseToken quoteToken feeTier volume24h } }",
+	})
+	if len(resp.Errors) != 0 {
+		t.Fatalf("markets must resolve on dex schema, got errors: %+v", resp.Errors)
+	}
+	// Round-trip through JSON to assert the field actually carries the seeded data.
+	raw, _ := json.Marshal(resp.Data)
+	got := string(raw)
+	for _, want := range []string{`"markets"`, poolID, `"feeTier"`, `400`} {
+		if !strings.Contains(got, want) {
+			t.Errorf("markets response missing %q; got %s", want, got)
+		}
+	}
+
+	// The FE fills query shape.
+	resp = eng.Execute(context.Background(), &Request{
+		Query: "{ fills(first: 5){ id market taker amountOut } }",
+	})
+	if len(resp.Errors) != 0 {
+		t.Fatalf("fills must resolve on dex schema, got errors: %+v", resp.Errors)
+	}
+	raw, _ = json.Marshal(resp.Data)
+	if !strings.Contains(string(raw), `"fills"`) || !strings.Contains(string(raw), "0xtx#0") {
+		t.Errorf("fills response missing seeded fill; got %s", string(raw))
+	}
+}
+
+// TestAmmSchema_DoesNotResolveMarkets pins the negative half: the amm schema does NOT
+// know `markets` (it returns an unknown-field error) — which is WHY the exchange-api
+// must route CLOB queries to the dex endpoint. This mirrors Red's live finding.
+func TestAmmSchema_DoesNotResolveMarkets(t *testing.T) {
+	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() })
+	eng := New(s, &Config{MaxQueryDepth: 10, MaxResultSize: 1 << 20, QueryTimeoutMs: 30000})
+	if err := eng.LoadBuiltin("amm"); err != nil {
+		t.Fatalf("LoadBuiltin(amm): %v", err)
+	}
+
+	resp := eng.Execute(context.Background(), &Request{Query: "{ markets(first: 1){ id } }"})
+	if len(resp.Errors) == 0 {
+		t.Fatal("amm schema must NOT resolve `markets` (it is a dex-schema field) — routing depends on this")
+	}
+	// And amm DOES resolve its own root (pools), confirming the engine is live.
+	resp = eng.Execute(context.Background(), &Request{Query: "{ pools(first: 1){ id } }"})
+	if len(resp.Errors) != 0 {
+		t.Fatalf("amm schema must resolve `pools`, got: %+v", resp.Errors)
+	}
+}

indexer/dex_v4_test.go

@@ -0,0 +1,426 @@
+package indexer
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+	"testing"
+
+	"github.com/luxfi/graph/storage"
+)
+
+// fixedGenesis returns a genesisHashFn that always reports the same hash — the
+// signature of a canonical chain (no re-genesis), even when a backend is lagging.
+func fixedGenesis(hash string) func(context.Context) (string, error) {
+	return func(context.Context) (string, error) { return hash, nil }
+}
+
+// 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"])
+	}
+}
+
+// initV4Log builds an Initialize log in the EXACT shape the 0x9999 precompile's
+// emitInitializeEvent emits (verified by the producer test
+// TestInitializeEvent_EmittedAt9999WithPairAndFee): topics [sig, poolId, currency0,
+// currency1]; data = fee(uint24) | tickSpacing(int24) | hooks(address) |
+// sqrtPriceX96(uint160) | tick(int24). NOT a fabricated layout — this mirrors the
+// real producer so the consumer test exercises the genuine wire format.
+func initV4Log(poolID, cur0, cur1 string, fee, tickSpacing, sqrtPriceX96, tick int64, blockHex string) *logEntry {
+	return &logEntry{
+		Address: LXSettleAddress,
+		Topics:  []string{SigInitializeV4, poolID, addrTopic(cur0), addrTopic(cur1)},
+		Data: "0x" + word(big.NewInt(fee)) + word(big.NewInt(tickSpacing)) +
+			word(new(big.Int)) /*hooks=0*/ + word(big.NewInt(sqrtPriceX96)) + word(big.NewInt(tick)),
+		BlockNumber: blockHex,
+	}
+}
+
+// Initialize at 0x9999 creates an AMM pool, bumps the factory aggregate, and
+// creates a RICH DEX Market carrying the token pair + fee tier (FIX B(a)) — the
+// fields the native producer (emitInitializeEvent at 0x9999) now supplies.
+func TestHandleInitializeV4_PoolMarketFactory(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	idx := NewWithConfig(Config{RPC: "http://unused"}, s)
+
+	poolID := topic32("d00d")
+	cur0 := "0x0000000000000000000000000000000000000011"
+	cur1 := "0x0000000000000000000000000000000000000022"
+	idx.processLog(initV4Log(poolID, cur0, cur1, 3000, 60, 0, 0, "0x1"))
+
+	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)
+	}
+	mk, _ := s.GetByType("Market", poolID)
+	if mk == nil {
+		t.Fatal("expected DEX Market from InitializeV4 at 0x9999")
+	}
+	mm := mk.(map[string]interface{})
+	// The Market is RICH, not a poolID-only stub: token pair + fee tier present.
+	if fmt.Sprint(mm["baseToken"]) != cur0 {
+		t.Errorf("Market baseToken = %v, want %s", mm["baseToken"], cur0)
+	}
+	if fmt.Sprint(mm["quoteToken"]) != cur1 {
+		t.Errorf("Market quoteToken = %v, want %s", mm["quoteToken"], cur1)
+	}
+	if asInt64(mm["feeTier"]) != 3000 {
+		t.Errorf("Market feeTier = %v, want 3000", mm["feeTier"])
+	}
+}
+
+// A Fill that arrives BEFORE its market's Initialize creates a volume stub; the
+// later Initialize must MERGE — adding the rich token pair + fee tier WITHOUT
+// resetting the accumulated volume/tradeCount to zero (LOW: stub-vs-rich merge).
+func TestHandleInitializeV4_MergesStubVolume(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	idx := NewWithConfig(Config{RPC: "http://unused"}, s)
+
+	poolID := topic32("beef")
+	taker := "0x00000000000000000000000000000000000000aa"
+	// 1) A DEXFill lands first → writeFill creates a Market stub with volume=400.
+	idx.processLog(&logEntry{
+		Address: LXSettleAddress, Topics: []string{SigDEXFill, poolID, addrTopic(taker)},
+		Data: "0x" + word(big.NewInt(400)) + word(big.NewInt(7)), BlockNumber: "0x5", TransactionHash: "0xaa", LogIndex: "0x0",
+	})
+	// 2) Initialize arrives later → must enrich, not wipe.
+	cur0 := "0x0000000000000000000000000000000000000011"
+	cur1 := "0x0000000000000000000000000000000000000022"
+	idx.processLog(initV4Log(poolID, cur0, cur1, 500, 10, 0, 0, "0x6"))
+
+	mk, _ := s.GetByType("Market", poolID)
+	mm := mk.(map[string]interface{})
+	if mm["volume24h"] != "400" {
+		t.Errorf("Initialize must preserve accumulated volume, got %v want 400", mm["volume24h"])
+	}
+	if asInt64(mm["tradeCount"]) != 1 {
+		t.Errorf("Initialize must preserve tradeCount, got %v want 1", mm["tradeCount"])
+	}
+	if fmt.Sprint(mm["baseToken"]) != cur0 || asInt64(mm["feeTier"]) != 500 {
+		t.Errorf("Initialize must add rich fields: baseToken=%v feeTier=%v", mm["baseToken"], mm["feeTier"])
+	}
+}
+
+// bumpFactory must PRESERVE the USD aggregates across its read-modify-write — the
+// landing-page header's TVL/volume. SeedFactory is INSERT OR REPLACE, so writing
+// them empty on every pool-create/swap would clobber them (MEDIUM).
+func TestBumpFactory_PreservesTVLAndVolume(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	idx := NewWithConfig(Config{RPC: "http://unused"}, s)
+
+	// Seed the factory with USD aggregates (as the price-oracle path would).
+	s.SeedFactory("1", &storage.SeedFactoryData{
+		PoolCount: 2, TxCount: 5, TotalValueLockedUSD: "1000000", TotalVolumeUSD: "5000000",
+	})
+
+	// A pool-create bumps counts — must not zero the USD aggregates.
+	idx.bumpFactory(1)
+
+	f, _ := s.GetFactory(nil, "1")
+	fm := f.(map[string]interface{})
+	if asInt64(fm["poolCount"]) != 3 {
+		t.Errorf("poolCount = %v, want 3", fm["poolCount"])
+	}
+	if fmt.Sprint(fm["totalValueLockedUSD"]) != "1000000" {
+		t.Errorf("totalValueLockedUSD clobbered: got %v want 1000000", fm["totalValueLockedUSD"])
+	}
+	if fmt.Sprint(fm["totalVolumeUSD"]) != "5000000" {
+		t.Errorf("totalVolumeUSD clobbered: got %v want 5000000", fm["totalVolumeUSD"])
+	}
+}
+
+// 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 —
+// but only after the low head is CONFIRMED across regenesisConfirmations polls AND
+// the genesis hash has actually CHANGED (the deterministic re-genesis signal).
+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)
+	}
+	ctx := context.Background()
+	// Baseline: the OLD chain's genesis hash, captured before the relaunch.
+	idx.genesisHash = "0xold"
+	// The fresh chain reports a DIFFERENT genesis hash — the re-genesis signal.
+	idx.genesisHashFn = fixedGenesis("0xnew")
+
+	// A fresh chain reports head=3, far below cursor — sustained across the
+	// confirmation window. Only the FINAL confirming poll rewinds.
+	for i := 0; i < regenesisConfirmations-1; i++ {
+		idx.maybeResetForRegenesis(ctx, 3)
+		if idx.lastBlock != 5_000_000 {
+			t.Fatalf("must not rewind before %d confirmations (poll %d), lastBlock=%d", regenesisConfirmations, i+1, idx.lastBlock)
+		}
+	}
+	idx.maybeResetForRegenesis(ctx, 3) // the confirming poll
+	if idx.lastBlock != 0 {
+		t.Fatalf("expected rewind to StartBlock 0 after %d confirmations + changed genesis, got %d", regenesisConfirmations, idx.lastBlock)
+	}
+	if s.GetLastBlock() != 0 {
+		t.Fatalf("expected persisted cursor reset to 0, got %d", s.GetLastBlock())
+	}
+	if idx.genesisHash != "0xnew" {
+		t.Fatalf("expected baseline to advance to the new chain's genesis, got %s", idx.genesisHash)
+	}
+}
+
+// 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.genesisHash = "0xsame"
+	idx.genesisHashFn = func(context.Context) (string, error) {
+		t.Fatal("shallow reorg must not even probe the genesis hash")
+		return "", nil
+	}
+
+	idx.maybeResetForRegenesis(context.Background(), 995) // 5 blocks back — within reorgDepth
+	if idx.lastBlock != 1000 {
+		t.Fatalf("shallow reorg must not reset; lastBlock=%d", idx.lastBlock)
+	}
+}
+
+// An idle/lagging RPC that momentarily reports a low head and then RECOVERS must
+// NOT wipe the cursor (re-genesis hysteresis — the DoS-amplification case Red
+// flagged: a single low reading must not trigger a full re-index from genesis).
+func TestPoll_IdleLowHeadRecoversNoWipe(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	s.SetLastBlock(5_000_000)
+	idx := NewWithConfig(Config{RPC: "http://unused", StartBlock: 0}, s)
+	ctx := context.Background()
+	idx.genesisHash = "0xsame"
+	idx.genesisHashFn = func(context.Context) (string, error) {
+		t.Fatal("transient/recovering low head must not reach the genesis probe")
+		return "", nil
+	}
+
+	// A couple of transient low readings (idle "Building block" / lag) — fewer than
+	// the confirmation window — must not rewind.
+	idx.maybeResetForRegenesis(ctx, 0) // idle: eth_blockNumber=0x0
+	idx.maybeResetForRegenesis(ctx, 1)
+	if idx.lastBlock != 5_000_000 {
+		t.Fatalf("transient low head must not wipe; lastBlock=%d", idx.lastBlock)
+	}
+	// RPC recovers to the true height — the streak must clear, no wipe ever.
+	idx.maybeResetForRegenesis(ctx, 5_000_001)
+	if idx.lastBlock != 5_000_000 {
+		t.Fatalf("recovered head must not wipe; lastBlock=%d", idx.lastBlock)
+	}
+	if idx.lowHeadStreak != 0 {
+		t.Fatalf("recovery must clear the low-head streak, got %d", idx.lowHeadStreak)
+	}
+	// And a subsequent single low blip still must not wipe (streak restarts at 1).
+	idx.maybeResetForRegenesis(ctx, 2)
+	if idx.lastBlock != 5_000_000 {
+		t.Fatalf("post-recovery single blip must not wipe; lastBlock=%d", idx.lastBlock)
+	}
+}
+
+// TestPoll_DeeplyBehindButCanonicalNoWipe is the item-4 hardening: a deeply-stale
+// load-balanced RPC backend can report a head >reorgDepth below our cursor for the
+// FULL confirmation window — satisfying the poll-count pre-filter — yet the chain is
+// still canonical. The genesis hash is UNCHANGED, so the cursor must NOT be wiped.
+// Under the old poll-count-only gate this scenario wiped progress and re-indexed
+// from genesis (the false-positive Red flagged).
+func TestPoll_DeeplyBehindButCanonicalNoWipe(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	s.SetLastBlock(5_000_000)
+	idx := NewWithConfig(Config{RPC: "http://unused", StartBlock: 0}, s)
+	ctx := context.Background()
+	idx.genesisHash = "0xcanonical"
+	// The cursor block hash is still valid / genesis unchanged — same chain, just behind.
+	probes := 0
+	idx.genesisHashFn = func(context.Context) (string, error) {
+		probes++
+		return "0xcanonical", nil
+	}
+
+	// Hold the head deeply below the cursor for WELL past the confirmation window.
+	for i := 0; i < regenesisConfirmations*3; i++ {
+		idx.maybeResetForRegenesis(ctx, 3)
+		if idx.lastBlock != 5_000_000 {
+			t.Fatalf("deeply-behind-but-canonical RPC must NOT wipe (iter %d); lastBlock=%d", i, idx.lastBlock)
+		}
+	}
+	if s.GetLastBlock() != 5_000_000 {
+		t.Fatalf("persisted cursor must be intact; got %d", s.GetLastBlock())
+	}
+	if probes == 0 {
+		t.Fatal("expected at least one genesis-hash probe once the streak confirmed")
+	}
+}
+
+// TestPoll_RegenesisHashChangeWipes is the positive companion: a true re-genesis
+// (genesis hash GONE/changed) must wipe to StartBlock, but only after the streak
+// confirms. This is the "cursor hash gone ⇒ wipe" half of the deterministic gate.
+func TestPoll_RegenesisHashChangeWipes(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	s.SetLastBlock(5_000_000)
+	idx := NewWithConfig(Config{RPC: "http://unused", StartBlock: 0}, s)
+	ctx := context.Background()
+	idx.genesisHash = "0xchainA"
+	idx.genesisHashFn = fixedGenesis("0xchainB") // fresh chain — different genesis
+
+	// Pre-confirmation polls must not wipe even though the hash differs (hysteresis
+	// still gates the probe; we only probe once the streak completes).
+	for i := 0; i < regenesisConfirmations-1; i++ {
+		idx.maybeResetForRegenesis(ctx, 2)
+		if idx.lastBlock != 5_000_000 {
+			t.Fatalf("must not wipe before confirmation (iter %d)", i)
+		}
+	}
+	idx.maybeResetForRegenesis(ctx, 2) // confirming poll — genesis changed ⇒ wipe
+	if idx.lastBlock != 0 || s.GetLastBlock() != 0 {
+		t.Fatalf("changed genesis must wipe to StartBlock; lastBlock=%d persisted=%d", idx.lastBlock, s.GetLastBlock())
+	}
+}
+
+// TestPoll_RegenesisProbeFailsNoWipe: if the genesis probe errors at the confirming
+// poll, we cannot prove a reset, so we must NOT wipe.
+func TestPoll_RegenesisProbeFailsNoWipe(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	s.SetLastBlock(5_000_000)
+	idx := NewWithConfig(Config{RPC: "http://unused", StartBlock: 0}, s)
+	ctx := context.Background()
+	idx.genesisHash = "0xbaseline"
+	idx.genesisHashFn = func(context.Context) (string, error) {
+		return "", fmt.Errorf("backend unreachable")
+	}
+
+	for i := 0; i < regenesisConfirmations+2; i++ {
+		idx.maybeResetForRegenesis(ctx, 3)
+	}
+	if idx.lastBlock != 5_000_000 {
+		t.Fatalf("probe failure must not wipe; lastBlock=%d", idx.lastBlock)
+	}
+}

indexer/events.go

@@ -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"`

indexer/indexer.go

@@ -26,21 +26,83 @@ type Status struct {
 	IndexedEvents uint64 `json:"indexedEvents"`
 }
 
-// Indexer watches an EVM RPC and writes events to storage.
-type Indexer struct {
-	rpc    string
-	store  *storage.Store
-	client *http.Client
-
-	lastBlock uint64
-	status    Status
+// 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
 }
 
-// New creates an indexer connected to the given RPC endpoint.
+// 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
+
+// regenesisConfirmations is how many CONSECUTIVE polls must observe a head far
+// below our cursor before we even SPEND an RPC to confirm a re-genesis. A single
+// low reading is NOT enough: an idle chain ("Building block", head=0x0) or a briefly
+// lagging RPC backend can momentarily report a head below our cursor, and probing on
+// every blip would be wasteful. A genuine relaunch keeps the head low for many polls
+// as the fresh chain rebuilds; a transient dip recovers within one or two. The streak
+// is a CHEAP PRE-FILTER only — the rewind itself is gated on a deterministic signal
+// (a changed genesis hash), so even a head held low for the full window by a deeply
+// lagging load-balanced backend never wipes a still-canonical chain.
+const regenesisConfirmations = 3
+
+// genesisBlockTag is the block selector for the genesis (block 0) header. The
+// genesis hash is the chain-identity fingerprint we compare to decide a re-genesis:
+// every node of a chain has block 0 regardless of how far behind its head is, so an
+// equal genesis hash deterministically rules out a re-genesis even on a deeply-stale
+// backend — unlike a by-number/by-hash probe of the CURSOR block, which a backend
+// whose head is below the cursor cannot serve (it would null-out and false-trip).
+const genesisBlockTag = "0x0"
+
+// Indexer watches an EVM RPC and writes events to storage.
+type Indexer struct {
+	rpc         string
+	poolManager string // lower-cased 0x9999 settlement address
+	startBlock  uint64
+	store       *storage.Store
+	client      *http.Client
+
+	lastBlock     uint64
+	lowHeadStreak int    // consecutive polls observing a head far below the cursor (re-genesis pre-filter)
+	genesisHash   string // chain-identity fingerprint; a change ⇒ true re-genesis (lazily captured)
+	status        Status
+
+	// genesisHashFn fetches the chain's genesis (block 0) hash. It is a field so
+	// tests can drive the canonicality decision deterministically without a live
+	// RPC; in production it is fetchGenesisHash (eth_getBlockByNumber 0x0). Honors
+	// the no-eth_getCode rule — only a block header is read.
+	genesisHashFn func(ctx context.Context) (string, error)
+}
+
+// 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{
@@ -50,15 +112,127 @@ func New(rpc string, store *storage.Store) *Indexer {
 			},
 		},
 	}
+	idx.genesisHashFn = idx.fetchGenesisHash
 	idx.lastBlock = idx.store.GetLastBlock()
+	if idx.lastBlock < cfg.StartBlock {
+		idx.lastBlock = cfg.StartBlock
+	}
 	return idx
 }
 
+// fetchGenesisHash reads the chain's genesis (block 0) header hash via
+// eth_getBlockByNumber. Only the header is fetched (txs=false); this honors the
+// no-eth_getCode rule and is cheap. The genesis hash is the chain-identity
+// fingerprint used to distinguish a true re-genesis (hash changed) from a merely
+// lagging/idle backend on the same chain (hash unchanged).
+func (idx *Indexer) fetchGenesisHash(ctx context.Context) (string, error) {
+	raw, err := idx.rpcCall(ctx, "eth_getBlockByNumber", []interface{}{genesisBlockTag, false})
+	if err != nil {
+		return "", err
+	}
+	var head struct {
+		Hash string `json:"hash"`
+	}
+	if err := json.Unmarshal(raw, &head); err != nil {
+		return "", fmt.Errorf("parse genesis header: %w", err)
+	}
+	if head.Hash == "" {
+		return "", fmt.Errorf("genesis header missing hash")
+	}
+	return strings.ToLower(head.Hash), nil
+}
+
+// ensureGenesisHash lazily captures the current chain's genesis hash as the
+// baseline the first time it succeeds. Idempotent and a no-op once set (a string
+// check), so it costs one RPC only on the first successful poll. Capturing the
+// baseline before any reset decision guarantees maybeResetForRegenesis always has
+// something to compare against.
+func (idx *Indexer) ensureGenesisHash(ctx context.Context) {
+	if idx.genesisHash != "" {
+		return
+	}
+	h, err := idx.genesisHashFn(ctx)
+	if err != nil {
+		log.Printf("[indexer] genesis hash baseline unavailable (will retry): %v", err)
+		return
+	}
+	idx.genesisHash = h
+}
+
 // Status returns current indexer progress.
 func (idx *Indexer) Status() Status {
 	return idx.status
 }
 
+// maybeResetForRegenesis rewinds the cursor to StartBlock ONLY on a confirmed
+// re-genesis: the chain's genesis hash has changed. 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 hold forever and the new chain would never index.
+//
+// Two gates, in order:
+//
+//  1. CHEAP PRE-FILTER (hysteresis): the head must stay far below our cursor for
+//     regenesisConfirmations consecutive polls. A shallow reorg, an idle chain
+//     ("Building block", head=0x0), or a transient lag clears the streak before we
+//     spend any extra RPC. This keeps the common case free of probes.
+//
+//  2. DETERMINISTIC SIGNAL: only after the streak confirms do we fetch the current
+//     genesis hash and compare it to our captured baseline. A CHANGED hash ⇒ a
+//     different chain ⇒ true re-genesis ⇒ rewind. An UNCHANGED hash ⇒ the same chain
+//     behind a deeply-stale/idle/load-balanced backend ⇒ we must NOT wipe; we clear
+//     the streak (it is proven not a reset) and keep the cursor. Because every node
+//     of a chain serves block 0 regardless of how far behind its head is, this signal
+//     cannot be false-tripped by lag — unlike the poll count alone, which a backend
+//     held >reorgDepth behind for the full window would satisfy on a still-canonical
+//     chain.
+//
+// If the baseline genesis hash is not yet captured (first run, or the probe failed),
+// we capture the current one and decline to wipe this round — a reset cannot be
+// proven without a baseline to compare against.
+func (idx *Indexer) maybeResetForRegenesis(ctx context.Context, latest uint64) {
+	farBelow := idx.lastBlock > idx.startBlock+reorgDepth && latest+reorgDepth < idx.lastBlock
+	if !farBelow {
+		idx.lowHeadStreak = 0 // head recovered (or normal) — abandon any suspicion.
+		return
+	}
+	idx.lowHeadStreak++
+	if idx.lowHeadStreak < regenesisConfirmations {
+		log.Printf("[indexer] head %d << cursor %d (>%d behind) — suspected chain reset, awaiting confirmation %d/%d",
+			latest, idx.lastBlock, reorgDepth, idx.lowHeadStreak, regenesisConfirmations)
+		return
+	}
+
+	// Streak confirmed — spend one RPC for the deterministic check.
+	current, err := idx.genesisHashFn(ctx)
+	if err != nil {
+		log.Printf("[indexer] head %d << cursor %d for %d polls but genesis-hash probe failed (%v) — "+
+			"NOT wiping; cannot confirm a reset without a canonical signal", latest, idx.lastBlock, idx.lowHeadStreak, err)
+		return
+	}
+	if idx.genesisHash == "" {
+		// No baseline yet — adopt the current hash and decline to wipe this round.
+		idx.genesisHash = current
+		log.Printf("[indexer] head %d << cursor %d but no genesis baseline yet — captured %s, NOT wiping",
+			latest, idx.lastBlock, current)
+		return
+	}
+	if current == idx.genesisHash {
+		// Same chain, just deeply behind/idle. The poll-count gate alone would have
+		// wiped here; the genesis hash proves it canonical, so we hold the cursor.
+		log.Printf("[indexer] head %d << cursor %d for %d polls but genesis hash unchanged (%s) — "+
+			"canonical chain behind a stale backend, NOT wiping", latest, idx.lastBlock, idx.lowHeadStreak, current)
+		idx.lowHeadStreak = 0
+		return
+	}
+
+	log.Printf("[indexer] genesis hash changed %s -> %s (head %d << cursor %d) — re-genesis confirmed, rewinding to %d",
+		idx.genesisHash, current, latest, idx.lastBlock, idx.startBlock)
+	idx.lastBlock = idx.startBlock
+	idx.store.SetLastBlock(idx.startBlock)
+	idx.genesisHash = current
+	idx.lowHeadStreak = 0
+}
+
 // 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 +314,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 +348,11 @@ func (idx *Indexer) poll(ctx context.Context) error {
 		return fmt.Errorf("parse hex block: %w", err)
 	}
 
+	// Capture the genesis-hash baseline (once) before any reset decision, then
+	// run the re-genesis self-heal (see maybeResetForRegenesis).
+	idx.ensureGenesisHash(ctx)
+	idx.maybeResetForRegenesis(ctx, latest)
+
 	// Nothing new
 	if latest <= idx.lastBlock {
 		return nil
@@ -239,9 +420,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 ───────────────────────────────────────────────
@@ -311,6 +497,9 @@ func (idx *Indexer) processLog(l *logEntry) {
 }
 
 // decodeUint256 reads a 32-byte hex word from data at the given word index.
+// Malformed hex (a non-hex digit in the word) yields 0 rather than a silently
+// half-parsed value: SetString reports ok=false and leaves n unmodified, so we
+// reset to a clean zero. A bad word must not masquerade as a real (partial) amount.
 func decodeUint256(data string, wordIndex int) *big.Int {
 	data = strings.TrimPrefix(data, "0x")
 	start := wordIndex * 64
@@ -318,7 +507,9 @@ func decodeUint256(data string, wordIndex int) *big.Int {
 		return new(big.Int)
 	}
 	n := new(big.Int)
-	n.SetString(data[start:start+64], 16)
+	if _, ok := n.SetString(data[start:start+64], 16); !ok {
+		return new(big.Int)
+	}
 	return n
 }
 
@@ -394,6 +585,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 +611,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 +622,204 @@ 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
+	var tvl, vol string
+	if m, ok := f.(map[string]interface{}); ok {
+		pc = asInt64(m["poolCount"])
+		tc = asInt64(m["txCount"])
+		// Carry the accumulated USD aggregates through the read-modify-write.
+		// SeedFactory is INSERT OR REPLACE (full overwrite); writing them empty
+		// here would clobber the landing-page header's TVL/volume on every
+		// pool-create / swap. They are accumulated elsewhere (price oracle / day
+		// data); bumpFactory only owns the pool + tx counts and must preserve the rest.
+		tvl = asString(m["totalValueLockedUSD"])
+		vol = asString(m["totalVolumeUSD"])
+	}
+	idx.store.SeedFactory("1", &storage.SeedFactoryData{
+		PoolCount:           pc + deltaPools,
+		TxCount:             tc + 1,
+		TotalValueLockedUSD: tvl,
+		TotalVolumeUSD:      vol,
+	})
+}
+
+// asString coerces a JSON-decoded value to its string form, returning "" for nil
+// so a missing aggregate carries through a read-modify-write as empty (not "<nil>").
+func asString(v interface{}) string {
+	if v == nil {
+		return ""
+	}
+	if s, ok := v.(string); ok {
+		return s
+	}
+	return fmt.Sprint(v)
+}
+
+// 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) {
+		// MERGE with any stub a prior fill already accumulated (writeFill creates a
+		// volume24h/tradeCount stub when a Fill arrives before its Initialize). The
+		// Initialize supplies the RICH fields (token pair, fee tier); it must NOT
+		// reset the accumulated trade aggregates to zero. Defaults apply only when
+		// the field is absent (no prior stub).
+		mk := map[string]interface{}{
+			"id": poolID, "symbol": poolID, "baseToken": token0, "quoteToken": token1,
+			"feeTier": fee, "volume24h": "0", "tradeCount": int64(0), "lastPrice": "0",
+		}
+		if existing, _ := idx.store.GetByType("Market", poolID); existing != nil {
+			if em, ok := existing.(map[string]interface{}); ok {
+				for _, k := range []string{"volume24h", "tradeCount", "lastPrice", "lastUpdate"} {
+					if v, present := em[k]; present {
+						mk[k] = v
+					}
+				}
+			}
+		}
+		idx.store.SetEntity("Market", poolID, mk)
+	}
+}
+
+// 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
+}