feat(clickhouse): realtime point-in-time DEX fill state for richer swap filters

.changelog/dex-enriched-fills.md

@@ -0,0 +1,5 @@
+---
+tidx: minor
+---
+
+Added `dex_order_events` and `dex_fills_enriched`, two ClickHouse objects that resolve each DEX fill's point-in-time order state. `dex_order_events` decodes the union of `OrderPlaced` and `OrderFlipped` into one positioned state stream (the existing `dex_orders` captures `OrderPlaced` only, so it is stale for flip orders). `dex_fills_enriched` is a plain view that ASOF-joins each fill to the latest order-state event strictly before it, exposing each fill book-natively: `token`/`quote_token`, `isBid`, `tick`, `at_peg`, `price`, and the base/quote amounts (`amountFilled`/`quote_amount`). Resolving the join at query time over the already-decoded, sort-keyed source tables keeps it realtime (no refresh lag) and correct for same-block flips and reorgs, letting the swaps feed restore its at-peg and token filters as plain SQL predicates instead of replaying the raw `logs` order-state stream per request. Taker source→destination orientation is left to swap assembly in the API, since it is a swap-level (route) notion rather than a per-fill property.

db/clickhouse/dex_fills_enriched.sql

@@ -0,0 +1,100 @@
+-- Point-in-time enriched `OrderFilled` stream, resolved at query time.
+--
+-- `OrderFilled` carries no `(token, isBid, tick)` — those are properties of the
+-- resting order at the moment of the fill, and for flip orders they change over
+-- the order's life (see `dex_order_events`). Consumers (the swaps feed, at-peg /
+-- token filters) otherwise had to replay the raw `logs` order-state stream per
+-- request and re-derive each fill's price in memory.
+--
+-- This view resolves that state in ClickHouse: each fill is ASOF-joined to the
+-- latest `dex_order_events` row strictly before its `(block_num, log_idx)`
+-- position, exactly mirroring the app's "latest state event before the fill"
+-- rule. It exposes each fill the way a fill is natively represented — one taker
+-- filling one resting maker order on one book — point-in-time and flip-correct:
+--   * `token` / `quote_token` — the book's base and quote tokens
+--   * `isBid` — order side: a maker bid pays quote for base, an ask sells base
+--   * `tick` — the order tick at fill time
+--   * `at_peg` — 1 iff the fill executed at tick 0 (rate == 1)
+--   * `price` — quote-per-base implied by the tick: (priceScale + tick) /
+--               priceScale, side-independent (priceScale = 100000)
+--   * `amountFilled` / `quote_amount` — the base-side and quote-side amounts of
+--               the fill (quote = intDiv(amountFilled * (priceScale + tick),
+--               priceScale))
+--
+-- Taker source -> destination orientation is intentionally NOT materialized
+-- here: it is a swap-level (route) notion, not a fill property. `getSwaps`
+-- groups these fills by `(tx, taker)`, orders by `logIndex`, and derives each
+-- hop's source/destination from `isBid` (bid: taker sends base, receives quote;
+-- ask: reversed) plus the swap's route ends and rate during assembly.
+--
+-- Quote resolution joins each book base to its `dex_pairs` row. Genesis system
+-- books predate the indexer's `PairCreated` decode and have no `dex_pairs` row,
+-- so their `quote_token` resolves to '' — best-effort, matching that those
+-- books are absent from `dex_pairs`.
+--
+-- Why a plain view (query-time join) rather than a materialized view:
+--   * Realtime. A fill is filterable the instant it lands; there is no refresh
+--     lag and no rolling-window cutoff, because the view reads the live tables.
+--   * Correct. The join sees the complete, ordered, reorg-corrected state at
+--     query time, so same-block flips and late/out-of-order events resolve
+--     right. An insert-time MV can't: sibling MVs (`dex_fills_mv`,
+--     `dex_order_events_mv`) off one `logs` insert have no guaranteed execution
+--     order, so a flip emitted in the same block as a later fill could be
+--     joined before it exists. A refreshable MV is correct but only as fresh as
+--     its last recompute.
+--   * Cheap enough. The heavy raw-log decode is already materialized and
+--     sort-keyed in `dex_order_events` (by `orderId`) and `dex_fills`, so the
+--     per-query cost is an ASOF join over decoded, indexed rows — far below the
+--     app's current per-request raw-`logs` replay. Callers scope cost with the
+--     usual time / token / pagination predicates, applied in SQL before LIMIT.
+--
+-- `FINAL` collapses the `ReplacingMergeTree` sources so reorg-duplicated
+-- pre-merge parts can't double-count.
+CREATE VIEW IF NOT EXISTS dex_fills_enriched AS
+SELECT
+    f.block_num AS block_num,
+    f.block_timestamp AS block_timestamp,
+    f.tx_idx AS tx_idx,
+    f.log_idx AS log_idx,
+    f.tx_hash AS tx_hash,
+    f.orderId AS orderId,
+    f.maker AS maker,
+    f.taker AS taker,
+    f.amountFilled AS amountFilled,
+    f.partialFill AS partialFill,
+    e.token AS token,
+    p.quote AS quote_token,
+    e.isBid AS isBid,
+    e.tick AS tick,
+    toUInt8(e.tick = 0) AS at_peg,
+    (toFloat64(100000) + e.tick) / 100000 AS price,
+    intDiv(f.amountFilled * toUInt256(toInt64(100000) + e.tick), 100000) AS quote_amount
+FROM
+(
+    SELECT
+        block_num,
+        block_timestamp,
+        tx_idx,
+        log_idx,
+        tx_hash,
+        orderId,
+        maker,
+        taker,
+        amountFilled,
+        partialFill,
+        toUInt64(block_num) * 4294967296 + toUInt64(log_idx) AS pos
+    FROM dex_fills FINAL
+) AS f
+ASOF INNER JOIN
+(
+    SELECT
+        orderId,
+        token,
+        isBid,
+        tick,
+        toUInt64(block_num) * 4294967296 + toUInt64(log_idx) AS pos
+    FROM dex_order_events FINAL
+) AS e
+    ON f.orderId = e.orderId AND f.pos > e.pos
+LEFT JOIN dex_pairs AS p FINAL ON p.base = e.token
+SETTINGS join_use_nulls = 0

db/clickhouse/dex_order_events.sql

@@ -0,0 +1,30 @@
+CREATE TABLE IF NOT EXISTS dex_order_events (
+    block_num       Int64,
+    block_timestamp DateTime64(3, 'UTC'),
+    tx_idx          Int32,
+    log_idx         Int32,
+    tx_hash         String,
+    address         String,
+    orderId         UInt256,
+    maker           String,
+    token           String,
+    amount          UInt256,
+    isBid           UInt8,
+    tick            Int16,
+    eventType       String,
+
+    INDEX idx_order_id orderId TYPE bloom_filter GRANULARITY 1,
+    INDEX idx_maker    maker   TYPE bloom_filter GRANULARITY 1
+) ENGINE = ReplacingMergeTree()
+PARTITION BY toYYYYMM(block_timestamp)
+ORDER BY (orderId, block_num, log_idx)
+-- Point-in-time order-state stream: the union of `OrderPlaced` and
+-- `OrderFlipped` events decoded from the raw `logs` stream, one positioned
+-- `(orderId, block_num, log_idx) -> (token, isBid, tick)` row per state change.
+--
+-- The decoded `dex_orders` table captures `OrderPlaced` only, so it is stale
+-- for flip orders: a T5+ flip order keeps its `orderId` but mutates
+-- `(isBid, tick)` on every fill and emits only `OrderFlipped` (no second
+-- `OrderPlaced`). Resolving a fill's true price therefore requires the latest
+-- state event *before* the fill, which this table makes available as a plain
+-- ASOF join (see `dex_fills_enriched`) instead of an app-side replay of `logs`.

db/clickhouse/dex_order_events_select.sql

@@ -0,0 +1,46 @@
+-- Decodes the union of `OrderPlaced(uint128 indexed orderId,
+-- address indexed maker, address indexed token, uint128 amount, bool isBid,
+-- int16 tick, bool isFlipOrder, int16 flipTick)` (selector 0xc200d837…) and
+-- `OrderFlipped(uint128 indexed orderId, address indexed maker,
+-- address indexed token, uint128 amount, bool isBid, int16 tick,
+-- int16 flipTick)` (selector 0x37a42d10…) from the raw `logs` stream.
+--
+-- Both events share the layout of the words read here: orderId/maker/token are
+-- indexed topics; `data` word0 is `amount` (low 16 bytes, hex 35..66), word1 is
+-- `isBid` (last byte, hex 129..130), word2 is `tick` (int16, sign-extended,
+-- last 2 bytes, hex 191..194). int16s read the trailing 2 bytes little-endian
+-- so two's-complement negatives decode correctly; the `OrderPlaced`-only tail
+-- words (`isFlipOrder`, `flipTick`) are not read, so the shorter `OrderFlipped`
+-- payload (4 data words) decodes through the same projection as `OrderPlaced`.
+SELECT
+    block_num,
+    block_timestamp,
+    tx_idx,
+    log_idx,
+    tx_hash,
+    address,
+    reinterpretAsUInt256(reverse(unhex(substring(topic1, 3, 64)))) AS orderId,
+    concat('0x', lower(substring(topic2, 27))) AS maker,
+    concat('0x', lower(substring(topic3, 27))) AS token,
+    reinterpretAsUInt256(reverse(unhex(substring(data, 3, 64)))) AS amount,
+    reinterpretAsUInt8(unhex(substring(data, 129, 2))) AS isBid,
+    reinterpretAsInt16(reverse(unhex(substring(data, 191, 4)))) AS tick,
+    if(
+        selector = '0xc200d837816d02c5ee9bf081cba1a32ab1482de7a738b41c0b357186b0b998cd',
+        'placed',
+        'flipped'
+    ) AS eventType
+FROM logs
+WHERE
+    selector IN (
+        '0xc200d837816d02c5ee9bf081cba1a32ab1482de7a738b41c0b357186b0b998cd',
+        '0x37a42d10bbce3e94e109a6a44e4479f0ee45dd6ecc6ca902168ea58e01ba32fe'
+    )
+    AND address = '0xdec0000000000000000000000000000000000000'
+    AND topic1 IS NOT NULL
+    AND topic2 IS NOT NULL
+    AND topic3 IS NOT NULL
+    AND length(topic1) >= 66
+    AND length(topic2) >= 66
+    AND length(topic3) >= 66
+    AND length(data) >= 194

src/clickhouse_schema/dex.rs

@@ -6,6 +6,10 @@ const DEX_ORDERS_SCHEMA: &str = include_str!("../../db/clickhouse/dex_orders.sql
 const DEX_ORDERS_SELECT: &str = include_str!("../../db/clickhouse/dex_orders_select.sql");
 const DEX_FILLS_SCHEMA: &str = include_str!("../../db/clickhouse/dex_fills.sql");
 const DEX_FILLS_SELECT: &str = include_str!("../../db/clickhouse/dex_fills_select.sql");
+const DEX_ORDER_EVENTS_SCHEMA: &str = include_str!("../../db/clickhouse/dex_order_events.sql");
+const DEX_ORDER_EVENTS_SELECT: &str =
+    include_str!("../../db/clickhouse/dex_order_events_select.sql");
+const DEX_FILLS_ENRICHED: &str = include_str!("../../db/clickhouse/dex_fills_enriched.sql");
 const DEX_OHLC_1M: &str = include_str!("../../db/clickhouse/dex_ohlc_1m.sql");
 const DEX_PAIR_LIQUIDITY: &str = include_str!("../../db/clickhouse/dex_pair_liquidity.sql");
 
@@ -82,6 +86,27 @@ pub const OBJECTS: &[ClickHouseObject] = &[
         block_column: None,
         backfill: None,
     },
+    ClickHouseObject {
+        name: "dex_order_events",
+        kind: ClickHouseObjectKind::Table(DEX_ORDER_EVENTS_SCHEMA),
+        depends_on: &["logs"],
+        public_query: true,
+        block_column: Some("block_num"),
+        backfill: Some(BackfillPolicy::Ranged {
+            select_sql: DEX_ORDER_EVENTS_SELECT,
+        }),
+    },
+    ClickHouseObject {
+        name: "dex_order_events_mv",
+        kind: ClickHouseObjectKind::MaterializedView {
+            target_table: "dex_order_events",
+            select_sql: DEX_ORDER_EVENTS_SELECT,
+        },
+        depends_on: &["logs", "dex_order_events"],
+        public_query: false,
+        block_column: None,
+        backfill: None,
+    },
     ClickHouseObject {
         name: "dex_pair_liquidity",
         kind: ClickHouseObjectKind::View(DEX_PAIR_LIQUIDITY),
@@ -101,6 +126,18 @@ pub const OBJECTS: &[ClickHouseObject] = &[
         block_column: None,
         backfill: None,
     },
+    ClickHouseObject {
+        name: "dex_fills_enriched",
+        kind: ClickHouseObjectKind::View(DEX_FILLS_ENRICHED),
+        depends_on: &["dex_fills", "dex_order_events", "dex_pairs"],
+        public_query: true,
+        // Plain view: ASOF-joins each fill to its point-in-time order state at
+        // query time over the already-decoded, sort-keyed source tables. Stores
+        // nothing, so it's realtime (no refresh lag), reorg-correct (reads live
+        // tables) and carries no block_column.
+        block_column: None,
+        backfill: None,
+    },
 ];
 
 #[cfg(test)]
@@ -113,7 +150,7 @@ mod tests {
 
     #[test]
     fn decoded_tables_are_block_scoped_and_public() {
-        for name in ["dex_pairs", "dex_orders", "dex_fills"] {
+        for name in ["dex_pairs", "dex_orders", "dex_fills", "dex_order_events"] {
             let table = object(name);
             assert!(table.is_table(), "{name} should be a table");
             assert!(table.public_query, "{name} should be public");
@@ -187,6 +224,78 @@ mod tests {
         );
     }
 
+    #[test]
+    fn order_events_decode_placed_and_flipped_from_logs() {
+        let mv = object("dex_order_events_mv");
+        let ddl = mv.ddl();
+        assert!(ddl.starts_with("CREATE MATERIALIZED VIEW IF NOT EXISTS dex_order_events_mv"));
+        assert!(ddl.contains("TO dex_order_events AS\n"));
+        assert!(ddl.contains("FROM logs"));
+        assert!(
+            ddl.contains("address = '0xdec0000000000000000000000000000000000000'"),
+            "should only decode DEX precompile logs"
+        );
+        // Unions OrderPlaced (0xc200d837…) and OrderFlipped (0x37a42d10…) so
+        // flip orders' mutated (isBid, tick) are captured — `dex_orders` misses
+        // them. Asserted here so an accidental edit can't drop a selector.
+        for selector in [
+            "0xc200d837816d02c5ee9bf081cba1a32ab1482de7a738b41c0b357186b0b998cd",
+            "0x37a42d10bbce3e94e109a6a44e4479f0ee45dd6ecc6ca902168ea58e01ba32fe",
+        ] {
+            assert!(ddl.contains(selector), "should decode selector {selector}");
+        }
+        // Both events share the data-word layout read here (isBid word1, tick
+        // word2 as a sign-extended int16).
+        assert!(ddl.contains("reinterpretAsUInt8(unhex(substring(data, 129, 2))) AS isBid"));
+        assert!(
+            ddl.contains("reinterpretAsInt16(reverse(unhex(substring(data, 191, 4)))) AS tick")
+        );
+    }
+
+    #[test]
+    fn enriched_fills_are_realtime_point_in_time_join() {
+        let enriched = object("dex_fills_enriched");
+        // Plain view, not a materialized view: the join runs at query time so a
+        // fill is filterable the instant it lands, and the join always sees the
+        // complete, reorg-corrected source state (no insert-time ordering hazard,
+        // no refresh lag).
+        assert!(enriched.is_view());
+        // Public so Cadent reads flip-correct, book-resolved fills instead of
+        // replaying the raw order-state stream; stores nothing, so no block_column.
+        assert!(enriched.public_query);
+        assert!(enriched.block_column.is_none());
+
+        let ddl = enriched.ddl();
+        assert!(ddl.contains("CREATE VIEW IF NOT EXISTS dex_fills_enriched"));
+        // A plain view stores nothing, so it must not carry materialization or
+        // refresh clauses.
+        assert!(!ddl.contains("MATERIALIZED VIEW"));
+        assert!(!ddl.contains("REFRESH"));
+        // ASOF join resolves each fill's latest state strictly before its
+        // position, against the placed+flipped events stream.
+        assert!(ddl.contains("ASOF INNER JOIN"));
+        assert!(ddl.contains("FROM dex_fills FINAL"));
+        assert!(ddl.contains("dex_order_events FINAL"));
+        assert!(ddl.contains("f.orderId = e.orderId AND f.pos > e.pos"));
+        // Quote token resolved from dex_pairs; genesis books fall back to ''.
+        assert!(ddl.contains("LEFT JOIN dex_pairs AS p FINAL"));
+        // Book-native fill columns: base/quote tokens, side, at-peg, price and
+        // the quote-side amount become column reads. Taker source/destination is
+        // a swap-level (route) notion derived during assembly, not stored here.
+        assert!(ddl.contains("e.token AS token"));
+        assert!(ddl.contains("p.quote AS quote_token"));
+        assert!(ddl.contains("e.isBid AS isBid"));
+        assert!(ddl.contains("toUInt8(e.tick = 0) AS at_peg"));
+        assert!(ddl.contains("AS quote_amount"));
+        assert!(!ddl.contains("source_token"));
+        assert!(!ddl.contains("destination_token"));
+        assert!(ddl.contains("100000"));
+        assert_eq!(
+            enriched.drop_sql().as_deref(),
+            Some("DROP VIEW IF EXISTS dex_fills_enriched")
+        );
+    }
+
     #[test]
     fn pair_liquidity_joins_pairs_to_dex_escrow_balances() {
         let view = object("dex_pair_liquidity");