Indexer

An EVM log indexer for the CEX-style wallet in contract/. It ingests Deposit, Transfer, and Withdraw logs emitted by CEXWallet and serves them back over a read-only HTTP API.

The goal is maximum concurrency under load, not the contract. The contract is only there to generate realistic logs against a local anvil node.

Architecture

                 logs                                read-only HTTP
  anvil ────────▶ indexer ─────────▶ Turso ──────────▶ axum ──────▶ clients
 (forge deploy)  (3 concurrent      (3 shared         (GET routers)
                  log listeners)     tables + registry)

One process, one Turso connection, four concurrent tasks joined with futures::try_join! in src/main.rs:

• Ingestion — three independent listeners (deposit_listener, withdraw_listener, transfer_listener in src/events.rs), one per event type, so the three types contend independently. Each builds its own alloy provider, subscribes with watch_logs, decodes via the sol!-generated bindings against the forge artifact, and inserts on arrival.
• Storage is a single Turso database with three shared tables, one per event type, plus a contract registry. No table-per-address.
• Serving is a read-only axum HTTP API on 127.0.0.1:3000: GET endpoints only, gzip-compressed responses.

At startup main applies every file in db/migrations in filename order, then runs SELECT 1 FROM contracts to warm the connection before serving.

De-duplication

Every insert is ON CONFLICT(tx_hash, log_index) DO NOTHING. Since watch_logs can replay a log across polls, this makes ingestion idempotent without a read-before-write.

Configuration

Copy .env.example to .env. Variables actually read at runtime:

Variable	Read by	Notes
RPC_URL	src/events.rs, /rpc_health	Required by the listeners. /rpc_health falls back to http://localhost:8545.
DEFAULT_CONTRACT_ADDRESS	src/events.rs, Filters::default	The wallet to index, and the default contract query filter.
PINGPONG_CONTRACT_ADDRESS	ping_listener / pong_listener	Only needed if those (currently unwired) listeners are enabled.
RUST_LOG	src/log.rs	Tracing env filter; defaults to info.

The database path is currently hardcoded to db/data.db in src/main.rs — DATABASE_URL in .env.example is not read yet.

Running

just recipes wrap the whole loop:

just start        # anvil -> deploy -> cargo run -> python3 script.py (load)
just stop         # kill the anvil / indexer background processes
just anvil        # local node only
just deploy       # forge script script/DeployAll.s.sol --broadcast
just run-indexer  # cargo run (applies migrations, serves 127.0.0.1:3000)
just load         # python3 script.py — random deposit/transfer/withdraw, 2 tx/s
just test         # forge test
just fmt          # forge fmt + cargo fmt

script.py is the load generator: it deploys TestToken from the forge artifact at startup, then fires one random deposit / transfer / withdraw against CEXWallet every 0.5s using the deterministic anvil accounts. Requires pip install web3 and a prior forge build.

Logging

src/log.rs installs two tracing layers over one EnvFilter:

• stdout — ANSI, human-readable, with thread ids/names, target, file, line.
• file — JSON, no ANSI, same fields, written through a non-blocking daily-rolling appender to log/indexer.log.YYYY-MM-DD.

init_tracing() returns a WorkerGuard that main holds for the process lifetime so buffered logs flush on exit. Every IndexerError logs itself on the way out (warn! for 4xx-class, error! for 5xx-class) — see IndexerError::error_out.

Database Schema (Turso / SQLite)

Schema lives in db/migrations/001_init.sql. All logs carry a (tx_hash, log_index) pair which is unique per emitted event; it is used for de-duplication. Storage is 128-bit-safe for amounts via TEXT (we store integer strings, avoiding lossy float conversion).

-- Registry of indexed CEXWallet deployments. One row per parent contract.
CREATE TABLE IF NOT EXISTS contracts (
    address     TEXT PRIMARY KEY,      -- parent wallet address
    created_at  INTEGER NOT NULL       -- unix epoch ms when registered
);

-- One deposit event per row.
CREATE TABLE IF NOT EXISTS deposits (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    contract     TEXT NOT NULL REFERENCES contracts(address),
    user         TEXT NOT NULL,                  -- indexed topic (address)
    token        TEXT NOT NULL,                  -- indexed topic (address)
    amount       TEXT NOT NULL,                  -- unindexed data (uint256)
    tx_hash      TEXT NOT NULL,
    log_index    INTEGER NOT NULL,
    block_number INTEGER NOT NULL,
    timestamp    INTEGER NOT NULL,               -- block timestamp
    UNIQUE (tx_hash, log_index)
);
CREATE INDEX IF NOT EXISTS idx_deposits_user   ON deposits(user);
CREATE INDEX IF NOT EXISTS idx_deposits_token  ON deposits(token);
CREATE INDEX IF NOT EXISTS idx_deposits_block  ON deposits(block_number);

-- One transfer event per row (internal bookkeeping between users).
CREATE TABLE IF NOT EXISTS transfers (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    contract     TEXT NOT NULL REFERENCES contracts(address),
    from_addr    TEXT NOT NULL,                  -- indexed topic (address)
    to_addr      TEXT NOT NULL,                  -- indexed topic (address)
    token        TEXT NOT NULL,                  -- indexed topic (address)
    amount       TEXT NOT NULL,                  -- unindexed data (uint256)
    tx_hash      TEXT NOT NULL,
    log_index    INTEGER NOT NULL,
    block_number INTEGER NOT NULL,
    timestamp    INTEGER NOT NULL,
    UNIQUE (tx_hash, log_index)
);
CREATE INDEX IF NOT EXISTS idx_transfers_from  ON transfers(from_addr);
CREATE INDEX IF NOT EXISTS idx_transfers_to    ON transfers(to_addr);
CREATE INDEX IF NOT EXISTS idx_transfers_token ON transfers(token);
CREATE INDEX IF NOT EXISTS idx_transfers_block ON transfers(block_number);

-- One withdraw event per row.
CREATE TABLE IF NOT EXISTS withdrawals (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    contract     TEXT NOT NULL REFERENCES contracts(address),
    user         TEXT NOT NULL,                  -- indexed topic (address)
    token        TEXT NOT NULL,                  -- indexed topic (address)
    amount       TEXT NOT NULL,                  -- unindexed data (uint256)
    destination  TEXT NOT NULL,                  -- unindexed data (address `to`)
    tx_hash      TEXT NOT NULL,
    log_index    INTEGER NOT NULL,
    block_number INTEGER NOT NULL,
    timestamp    INTEGER NOT NULL,
    UNIQUE (tx_hash, log_index)
);
CREATE INDEX IF NOT EXISTS idx_withdrawals_user   ON withdrawals(user);
CREATE INDEX IF NOT EXISTS idx_withdrawals_token  ON withdrawals(token);
CREATE INDEX IF NOT EXISTS idx_withdrawals_block  ON withdrawals(block_number);

Column mapping to the contract events

Event	Indexed topics (no decoding)	Unindexed data (ABI-decoded)
Deposit(user, token, amount)	user, token	amount
Transfer(from, to, token, amount)	from, to, token	amount
Withdraw(user, token, amount, to)	user, token	amount, to

Address columns store the hex 0x-prefixed string; amount columns store the raw uint256 as a decimal string.

Endpoints

All endpoints are read-only (GET). The parent wallet is not a path segment — it is the contract query filter, which defaults to DEFAULT_CONTRACT_ADDRESS. Responses are JSON arrays of the raw row records, newest first (ORDER BY id DESC), gzip-compressed when the client offers it.

Health

• GET /health — 200 {"status":"ok"}. Liveness only, no DB round-trip.
• GET /rpc_health — {"status":"ok","block_number":N} from the configured RPC_URL, or {"status":"error","message":…} if the node is unreachable.

Contracts

• GET /contracts — list all registered CEXWallet parent addresses.
• GET /contracts/{contract} — confirm a parent address is indexed.

Deposits

• GET /deposits — deposits, narrowed by the query filters below.
• GET /deposits/user/{user} — deposits by depositor.
• GET /deposits/token/{token} — deposits by token.

Transfers

• GET /transfers — all transfers.
• GET /transfers/from/{addr} — transfers out of {addr}.
• GET /transfers/to/{addr} — transfers into {addr}.
• GET /transfers/token/{token} — transfers by token.
• GET /transfers/between/{a}/{b} — transfers between two addresses in either direction.

Withdrawals

• GET /withdrawals — all withdrawals.
• GET /withdrawals/user/{user} — withdrawals by user.
• GET /withdrawals/token/{token} — withdrawals by token.

Anything else falls through to not_found_handler, which returns 404 naming the requested path.

Query filters

Every list endpoint takes the same filter set through the ValidatedFilters extractor, which parses, merges over the defaults, and validates before the handler runs.

Param	Type	Default
contract	Address	DEFAULT_CONTRACT_ADDRESS
token	Address	—
tx_hash	FixedBytes<32> (66-char 0x…)	—
from_block	u64	—
to_block	u64	—
limit	u64, max 1000	10
offset	u64	0

Filters are typed, so a malformed address, a bad tx hash, or a negative / non-numeric block or page value is rejected at extraction rather than reaching SQL. Filters is #[serde(deny_unknown_fields)], so an unrecognised query parameter is an error too — a typo'd filter fails loudly instead of being silently ignored and widening the result set.

Validated constraints (validate_filters):

• limit ≤ MAX_LIMIT (1000)
• from_block ≤ to_block
• to_block - from_block ≤ MAX_BLOCK_RANGE (1000)

Path parameters are typed as Address too, so /deposits/user/notanaddress is a 400, never a query.

Amounts are returned as decimal strings (amount), never floats.

Errors

Handlers return Result<Json<…>, IndexerError> (src/error.rs). IndexerError implements IntoResponse, so each variant maps to a status code and a JSON body {"error": "…"}, and logs itself on the way out.

Variant	Status
Validation	400 — carries the specific message (limit cannot exceed 1000, Invalid query filters, …)
Decode, InvalidInput, AddressDecode	400
NotFound, RouteNotFound	404
Database, Rpc, MissingEnvVar, Unknown	500

Note that an empty result set is a 404, not an empty 200 array: query_rows returns NotFound when a query matches no rows. The client-facing message stays generic; the query and params are recorded in the log.

Contracts

contract/ is a Foundry project. CEXWallet.sol is the indexed target; TestToken.sol / MockERC20.sol back the load script; PingPong.sol and SimpleAuction.sol are extra log sources for experiments (the ping_listener / pong_listener in src/events.rs are written but not joined in main).

The sol! macro reads forge output from contract/out/…, so forge build must run before cargo build. Those artifact paths are currently absolute — they need editing to build the indexer outside this checkout.

Known gaps

• Nothing writes to the contracts registry — the listeners insert events with a contract value but never register the address, so /contracts and /contracts/{contract} return 404 until a row is inserted by hand.
• GET /transfers/between/{a}/{b} builds its WHERE clause inline instead of going through select(), and defaults limit to 1000 rather than 10.
• Ingestion starts from the current head via watch_logs; there is no historical backfill and no persisted cursor across restarts.
• DATABASE_URL is unused (see Configuration).
• No Rust-side tests yet; just test runs the Solidity suite only.