graph/indexer/blocks_test.go

package indexer

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"github.com/luxfi/graph/storage"
)

// A swap's time comes from its block's header, and nothing else has one. These
// assert the two ways a row gets dated — as it arrives, and after the fact —
// because a day series is exactly as good as the clock underneath it, and the
// clock underneath it used to be a block number.

// chainAt serves a fake node: one PoolCreated, one Swap, and headers whose time
// is `mined`. It answers single calls and batches, as a node does.
func chainAt(t *testing.T, mined int64, logs []*logEntry) *httptest.Server {
	t.Helper()
	answer := func(method string, params json.RawMessage) interface{} {
		switch method {
		case "eth_blockNumber":
			return "0x9"
		case "eth_getBlockByNumber":
			var p []interface{}
			json.Unmarshal(params, &p)
			if len(p) > 0 && p[0] == "0x0" {
				return map[string]interface{}{"hash": "0xgenesis"}
			}
			return map[string]interface{}{"timestamp": hexUint(mined)}
		case "eth_getLogs":
			out := make([]interface{}, 0, len(logs))
			for _, l := range logs {
				out = append(out, map[string]interface{}{
					"address": l.Address, "topics": l.Topics, "data": l.Data,
					"blockNumber": l.BlockNumber, "transactionHash": l.TransactionHash,
					"logIndex": l.LogIndex,
				})
			}
			return out
		}
		return "0x"
	}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var raw json.RawMessage
		json.NewDecoder(r.Body).Decode(&raw)
		type call struct {
			ID     int             `json:"id"`
			Method string          `json:"method"`
			Params json.RawMessage `json:"params"`
		}
		reply := func(c call) map[string]interface{} {
			return map[string]interface{}{"jsonrpc": "2.0", "id": c.ID, "result": answer(c.Method, c.Params)}
		}
		if len(raw) > 0 && raw[0] == '[' {
			var batch []call
			json.Unmarshal(raw, &batch)
			out := make([]map[string]interface{}, 0, len(batch))
			for _, c := range batch {
				out = append(out, reply(c))
			}
			json.NewEncoder(w).Encode(out)
			return
		}
		var c call
		json.Unmarshal(raw, &c)
		json.NewEncoder(w).Encode(reply(c))
	}))
	t.Cleanup(srv.Close)
	return srv
}

func hexUint(n int64) string {
	const hex = "0123456789abcdef"
	if n == 0 {
		return "0x0"
	}
	var b []byte
	for n > 0 {
		b = append([]byte{hex[n&0xf]}, b...)
		n >>= 4
	}
	return "0x" + string(b)
}

// A swap indexed off the chain is dated by its block, not numbered by it. This
// drives the real poll loop, so it fails if the poll stops reading headers as
// surely as if the reader itself broke.
func TestPollDatesSwapsFromTheirBlock(t *testing.T) {
	const (
		token0 = "0x0000000000000000000000000000000000000011"
		token1 = "0x0000000000000000000000000000000000000022"
		pool   = "0x00000000000000000000000000000000000000c3"
		mined  = 1767225600 + 3600 // an hour into 2026-01-01
	)
	s := newMemSQLiteStore(t)
	srv := chainAt(t, mined, []*logEntry{
		poolCreatedLog(testFactoryV3, token0, token1, 3000, pool),
		swapV3Log(pool, token0, "0xreal", 1000, -2500),
	})
	idx := NewWithConfig(Config{RPC: srv.URL, FactoryV3: testFactoryV3}, s)

	if err := idx.poll(context.Background()); err != nil {
		t.Fatalf("poll: %v", err)
	}

	raw := s.RecentSwapsRaw(10)
	if len(raw) != 1 {
		t.Fatalf("expected one swap, got %d", len(raw))
	}
	for _, sw := range raw {
		if sw.Timestamp != mined {
			t.Errorf("timestamp = %d, want %d — a swap's time is its block's time, not its number",
				sw.Timestamp, mined)
		}
		if sw.Block != 2 {
			t.Errorf("block = %d, want 2", sw.Block)
		}
		if !sw.Dated() {
			t.Error("a swap read from a live chain must be dated")
		}
	}
}

// A block that will not answer for its header is not indexed. Advancing past it
// would leave a permanent hole in the series that nothing revisits.
func TestPollWillNotAdvancePastAnUnreadableBlock(t *testing.T) {
	const pool = "0x00000000000000000000000000000000000000c3"
	s := newMemSQLiteStore(t)
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var raw json.RawMessage
		json.NewDecoder(r.Body).Decode(&raw)
		if len(raw) > 0 && raw[0] == '[' { // the header batch
			json.NewEncoder(w).Encode([]map[string]interface{}{})
			return
		}
		var c struct {
			ID     int    `json:"id"`
			Method string `json:"method"`
		}
		json.Unmarshal(raw, &c)
		var result interface{} = "0x"
		switch c.Method {
		case "eth_blockNumber":
			result = "0x9"
		case "eth_getBlockByNumber":
			result = map[string]interface{}{"hash": "0xgenesis"}
		case "eth_getLogs":
			result = []interface{}{map[string]interface{}{
				"address": pool, "topics": []string{SigSwapV3, addrTopic(pool), addrTopic(pool)},
				"data": "0x", "blockNumber": "0x2", "transactionHash": "0xtx", "logIndex": "0x0",
			}}
		}
		json.NewEncoder(w).Encode(map[string]interface{}{"jsonrpc": "2.0", "id": c.ID, "result": result})
	}))
	defer srv.Close()

	idx := NewWithConfig(Config{RPC: srv.URL, FactoryV3: testFactoryV3}, s)
	err := idx.poll(context.Background())
	if err == nil {
		t.Fatal("poll reported success for a block it could not date")
	}
	if !strings.Contains(err.Error(), "block times") {
		t.Errorf("error = %v, want it to name the unread block times", err)
	}
	if got := s.GetLastBlock(); got != 0 {
		t.Errorf("cursor advanced to %d past a block that was not fully read", got)
	}
}

// Rows an older build wrote hold a block NUMBER where the time belongs. They are
// the entire indexed history, so a series that cannot date them starts from
// today — which is not a chart.
func TestHealDatesTheHistory(t *testing.T) {
	const mined = 1767225600 + 7200
	s := newMemSQLiteStore(t)
	s.SeedSwap("0xold#0x1", &storage.SeedSwapData{
		Timestamp: 1136227, Pool: "0xpool", Amount0: "1", Amount1: "-2", AmountUSD: "0",
	})
	s.SeedSwap("0xnew#0x1", &storage.SeedSwapData{
		Timestamp: mined, Block: 99, Pool: "0xpool", Amount0: "1", Amount1: "-2", AmountUSD: "0",
	})
	idx := NewWithConfig(Config{RPC: chainAt(t, mined, nil).URL}, s)

	found, healed, err := idx.healSwapTimes(context.Background())
	if err != nil || found != healed {
		t.Fatalf("heal dated %d of %d rows found: %v", healed, found, err)
	}
	got := s.RecentSwapsRaw(10)
	old := got["0xold#0x1"]
	if old == nil {
		t.Fatal("the healed row vanished")
	}
	if old.Timestamp != mined || old.Block != 1136227 {
		t.Errorf("healed row = {ts %d, block %d}, want {ts %d, block 1136227}", old.Timestamp, old.Block, mined)
	}
	if n := got["0xnew#0x1"]; n.Timestamp != mined || n.Block != 99 {
		t.Errorf("an already-dated row was rewritten: %+v", n)
	}
	// Idempotent: a second run has nothing to do and must not re-date anything.
	if found, _, err := idx.healSwapTimes(context.Background()); found != 0 || err != nil {
		t.Errorf("second heal found %d undated rows: %v", found, err)
	}
}

// The rows needing repair are the ones a newest-first window never contains.
//
// An undated row holds a block number where a time belongs, and a block number
// is small beside a unix second, so ordering by timestamp sorts every broken row
// behind every correct one. A heal that read the newest rows therefore looked
// straight past its own subject the moment a full window of dated rows existed:
// it found nothing stale, reported success and stopped, and a million blocks of
// trade history stayed dated to 1970 — indexed, and invisible on every
// newest-first list the exchange draws.
func TestHealFindsRowsBuriedUnderDatedOnes(t *testing.T) {
	const mined = 1767225600 + 7200
	s := newMemSQLiteStore(t)
	// One undated row, then more dated rows than a single window would hold, so
	// the undated one is only reachable by asking for undated rows.
	s.SeedSwap("0xburied#0x1", &storage.SeedSwapData{
		Timestamp: 1136227, Pool: "0xpool", Amount0: "1", Amount1: "-2", AmountUSD: "0",
	})
	for i := 0; i < maxValuedSwaps+16; i++ {
		s.SeedSwap(fmt.Sprintf("0xdated#%d", i), &storage.SeedSwapData{
			Timestamp: mined + int64(i), Block: uint64(100 + i),
			Pool: "0xpool", Amount0: "1", Amount1: "-2", AmountUSD: "0",
		})
	}
	idx := NewWithConfig(Config{RPC: chainAt(t, mined, nil).URL}, s)

	found, healed, err := idx.healSwapTimes(context.Background())
	if err != nil || found != 1 || healed != 1 {
		t.Fatalf("heal found %d and dated %d (%v) — the buried row was not found", found, healed, err)
	}
	got := s.UndatedSwapsRaw(10)
	if len(got) != 0 {
		t.Errorf("%d rows still hold a block number where a time belongs", len(got))
	}
}

// The day a trade belongs to is the UTC day containing it.
func TestDayStartIsUTCMidnight(t *testing.T) {
	const midnight = 1767225600 // 2026-01-01 00:00:00 UTC
	for _, c := range []struct {
		ts   int64
		want int64
	}{
		{midnight, midnight},
		{midnight + 1, midnight},
		{midnight + 86399, midnight},
		{midnight + 86400, midnight + 86400},
		{midnight - 1, midnight - 86400},
	} {
		if got := dayStart(c.ts); got != c.want {
			t.Errorf("dayStart(%d) = %d, want %d", c.ts, got, c.want)
		}
	}
	if got := dayID("0xABC", midnight); got != "0xabc-20454" {
		t.Errorf("dayID = %q, want 0xabc-20454", got)
	}
}

// A store holds more undated rows than one pass can window, and every one of
// them has to end up dated.
//
// The heal reads a bounded window, so a long history takes several passes. The
// loop that repeats it must stop when the STORE has nothing left, not when a
// window does — a window that heals completely has no failures in it, and
// reading that as "finished" stops after the first 50,000 and leaves the rest
// exactly as invisible as before.
func TestHealDrainsEveryWindow(t *testing.T) {
	const mined = 1767225600 + 7200
	s := newMemSQLiteStore(t)
	const rows = maxValuedSwaps + 2500
	for i := 0; i < rows; i++ {
		s.SeedSwap(fmt.Sprintf("0xold#%d", i), &storage.SeedSwapData{
			Timestamp: int64(1 + i%9000), // a block number, as an older build wrote it
			Pool:      "0xpool", Amount0: "1", Amount1: "-2", AmountUSD: "0",
		})
	}
	idx := NewWithConfig(Config{RPC: chainAt(t, mined, nil).URL}, s)

	keepSwapTimesDone := make(chan struct{})
	go func() { idx.keepSwapTimes(context.Background()); close(keepSwapTimesDone) }()
	select {
	case <-keepSwapTimesDone:
	case <-time.After(90 * time.Second):
		t.Fatal("heal never finished draining")
	}

	if left := len(s.UndatedSwapsRaw(10)); left != 0 {
		t.Errorf("%d rows still hold a block number after the heal stopped", left)
	}
}

// A public JSON-RPC door caps a batch at 100 calls and refuses a larger one
// whole, answering one error object instead of an array. Every header over a
// long range still has to be read through such a door.
func TestBlockTimesFitThroughABatchCap(t *testing.T) {
	const cap = 100
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var batch []struct {
			ID     int           `json:"id"`
			Params []interface{} `json:"params"`
		}
		if err := json.NewDecoder(r.Body).Decode(&batch); err != nil || len(batch) > cap {
			json.NewEncoder(w).Encode(map[string]interface{}{
				"jsonrpc": "2.0", "id": nil,
				"error": map[string]interface{}{"code": -32600, "message": "batch size"},
			})
			return
		}
		out := make([]map[string]interface{}, 0, len(batch))
		for _, c := range batch {
			out = append(out, map[string]interface{}{
				"jsonrpc": "2.0", "id": c.ID,
				"result": map[string]interface{}{"timestamp": c.Params[0]},
			})
		}
		json.NewEncoder(w).Encode(out)
	}))
	defer srv.Close()

	idx := NewWithConfig(Config{RPC: srv.URL, FactoryV3: testFactoryV3}, newMemSQLiteStore(t))
	blocks := make([]uint64, 0, 3*cap+7)
	for b := uint64(1); b <= 3*cap+7; b++ {
		blocks = append(blocks, b)
	}
	got := idx.blockTimes(context.Background(), blocks)
	if len(got) != len(blocks) {
		t.Fatalf("read %d of %d block times through a %d-call batch cap", len(got), len(blocks), cap)
	}
	for _, b := range blocks {
		if got[b] != int64(b) {
			t.Fatalf("block %d: time %d, want %d", b, got[b], b)
		}
	}
}