Token Balances by Token

CREATE TABLE recipe_token_balances_by_token.fact_balances_by_token
(
    vendor_event_id   String,
    ingested_at       DateTime64(3),
    chain_id          UInt32,
    block_hash        String,
    block_number      UInt64,
    log_index         UInt32,
    event_ts          DateTime64(3),
    token_address     String,
    wallet_address    String,
    balance           String,          -- absolute balance after this transfer (raw uint256)
    leg               LowCardinality(String),   -- 'from' | 'to'
    sign              Int8
)
ENGINE = ReplicatedCollapsingMergeTree(
    '/clickhouse/tables/{database}/fact_balances_by_token', '{replica}', sign)
PARTITION BY (chain_id, toYYYYMM(event_ts))
ORDER BY (chain_id, token_address, wallet_address, block_number, log_index, vendor_event_id, leg);

-- Observations (sink target). Append-only, keyed on the block-level position.
CREATE TABLE public.token_balance_observations (
  position         BIGINT  NOT NULL,
  log_index        BIGINT  NOT NULL,
  block_number     BIGINT  NOT NULL,
  block_timestamp  BIGINT  NOT NULL,         -- unix seconds
  token_address    TEXT    NOT NULL,
  wallet_address   TEXT    NOT NULL,
  balance          NUMERIC(76, 0) NOT NULL,  -- absolute balance after the transfer
  leg              TEXT    NOT NULL,         -- 'from' | 'to'
  vendor_event_id  TEXT    NOT NULL
);

-- Speeds the DISTINCT ON (latest-per-key) the materialized view computes.
CREATE INDEX IF NOT EXISTS tbo_token_wallet_recency_idx
  ON public.token_balance_observations
  (token_address, wallet_address, block_number DESC, log_index DESC);

-- Current-balance projection: latest observation per (token, wallet).
CREATE MATERIALIZED VIEW public.token_balances AS
SELECT DISTINCT ON (token_address, wallet_address)
  token_address,
  wallet_address,
  balance,
  block_number,
  log_index,
  block_timestamp
FROM public.token_balance_observations
ORDER BY token_address, wallet_address, block_number DESC, log_index DESC;

-- Required by REFRESH MATERIALIZED VIEW CONCURRENTLY.
CREATE UNIQUE INDEX IF NOT EXISTS token_balances_pk
  ON public.token_balances (token_address, wallet_address);

-- Active holders of a token (this recipe's primary access path).
CREATE INDEX IF NOT EXISTS token_balances_by_token_active_idx
  ON public.token_balances (token_address, wallet_address)
  WHERE balance > 0;

-- Sibling access path: all non-zero balances held by a wallet.
CREATE INDEX IF NOT EXISTS token_balances_by_wallet_active_idx
  ON public.token_balances (wallet_address, token_address)
  WHERE balance > 0;

-- Cleanup index: find zeroed-out positions to prune.
CREATE INDEX IF NOT EXISTS token_balances_zero_cleanup_idx
  ON public.token_balances (token_address, wallet_address)
  WHERE balance = 0;