NFTs by Wallet

CREATE TABLE recipe_wallet_nfts.fact_wallet_nfts
(
    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,
    token_id          String,                   -- uint256 id → String, not numeric
    contract_type     LowCardinality(String),   -- ERC721 | ERC1155
    wallet_address    String,
    balance           String,                   -- absolute post-transfer holding
    leg               LowCardinality(String),   -- from | to
    sign              Int8
)
ENGINE = ReplicatedCollapsingMergeTree(
    '/clickhouse/tables/{database}/fact_wallet_nfts', '{replica}', sign)
PARTITION BY (chain_id, toYYYYMM(event_ts))
ORDER BY (chain_id, wallet_address, token_address, token_id, block_number, log_index, vendor_event_id, leg);

-- 1. Observations (sink target). Latest per key wins as the current holding.
CREATE TABLE public.wallet_nft_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,
  token_id         TEXT    NOT NULL,         -- uint256 id → TEXT, not numeric
  contract_type    TEXT    NOT NULL,         -- ERC721 | ERC1155
  wallet_address   TEXT    NOT NULL,
  balance          NUMERIC(76, 0) NOT NULL,  -- absolute post-balance of the id
  leg              TEXT    NOT NULL,
  vendor_event_id  TEXT    NOT NULL
);

-- Recency index: leads with the DISTINCT ON keys so REFRESH avoids a sort.
CREATE INDEX wno_wallet_token_id_recency_idx
  ON public.wallet_nft_observations
  (wallet_address, token_address, token_id, block_number DESC, log_index DESC);

-- 2. Current-holding view: latest observation per (wallet, token, token_id).
CREATE MATERIALIZED VIEW public.wallet_nfts AS
SELECT DISTINCT ON (wallet_address, token_address, token_id)
  wallet_address, token_address, token_id, contract_type,
  balance, block_number, log_index, block_timestamp
FROM public.wallet_nft_observations
ORDER BY wallet_address, token_address, token_id, block_number DESC, log_index DESC;

CREATE UNIQUE INDEX wallet_nfts_pk
  ON public.wallet_nfts (wallet_address, token_address, token_id);

-- Primary access path: NFTs currently held by a wallet.
CREATE INDEX wallet_nfts_active_idx
  ON public.wallet_nfts (wallet_address, token_address, token_id)
  WHERE balance > 0;

-- Sibling access path: all current holders of a given NFT id.
CREATE INDEX wallet_nfts_by_token_active_idx
  ON public.wallet_nfts (token_address, token_id, wallet_address)
  WHERE balance > 0;

-- Cleanup index for zeroed (no-longer-held) positions.
CREATE INDEX wallet_nfts_zero_cleanup_idx
  ON public.wallet_nfts (wallet_address, token_address, token_id)
  WHERE balance = 0;