Pair Reserves

CREATE TABLE recipe_pair_reserves.fact_pair_reserves
(
    vendor_event_id   String,
    ingested_at       DateTime64(3),
    chain_id          UInt32,
    block_hash        String,
    block_number      UInt64,
    log_index         UInt32,
    event_ts          DateTime64(3),
    pair_address      String,
    token0_address    String,
    token1_address    String,
    reserve0          String,          -- absolute pool reserve0 after this swap
    reserve1          String,
    sign              Int8
)
ENGINE = ReplicatedCollapsingMergeTree(
    '/clickhouse/tables/{database}/fact_pair_reserves', '{replica}', sign)
PARTITION BY (chain_id, toYYYYMM(event_ts))
ORDER BY (chain_id, pair_address, block_number, log_index, vendor_event_id);

-- 1. Observations (sink target): one raw row per swap.
CREATE TABLE public.pair_reserve_observations (
  position         BIGINT  NOT NULL,
  log_index        BIGINT  NOT NULL,
  block_number     BIGINT  NOT NULL,
  block_timestamp  BIGINT  NOT NULL,         -- unix seconds
  pair_address     TEXT    NOT NULL,
  token0_address   TEXT    NOT NULL,
  token1_address   TEXT    NOT NULL,
  reserve0         NUMERIC(76, 0) NOT NULL,  -- absolute pool reserve0 post-swap
  reserve1         NUMERIC(76, 0) NOT NULL,
  vendor_event_id  TEXT    NOT NULL
);

-- Speeds the DISTINCT ON (latest-per-pair) the materialized view computes.
CREATE INDEX IF NOT EXISTS pro_pair_recency_idx
  ON public.pair_reserve_observations
  (pair_address, block_number DESC, log_index DESC);

-- 2. Current-reserves materialized view: latest observation per pair.
CREATE MATERIALIZED VIEW public.pair_reserves AS
SELECT DISTINCT ON (pair_address)
  pair_address,
  token0_address,
  token1_address,
  reserve0,
  reserve1,
  block_number,
  log_index,
  block_timestamp
FROM public.pair_reserve_observations
ORDER BY pair_address, block_number DESC, log_index DESC;

-- Required by REFRESH MATERIALIZED VIEW CONCURRENTLY and the by-pair lookup.
CREATE UNIQUE INDEX IF NOT EXISTS pair_reserves_pk
  ON public.pair_reserves (pair_address);