NFT Owners by Contract

CREATE TABLE recipe_nft_owners_by_contract.fact_nft_owners_by_contract
(
    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, kept as text
    contract_type     LowCardinality(String),   -- ERC721 | ERC1155
    wallet_address    String,
    amount            String,                   -- absolute post-transfer holding of this token_id
    leg               LowCardinality(String),   -- from | to
    sign              Int8
)
ENGINE = ReplicatedCollapsingMergeTree(
    '/clickhouse/tables/{database}/fact_nft_owners_by_contract', '{replica}', sign)
PARTITION BY (chain_id, toYYYYMM(event_ts))
ORDER BY (chain_id, token_address, token_id, wallet_address, block_number, log_index, vendor_event_id, leg);

CREATE TABLE public.nft_owner_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, kept as text
  contract_type    TEXT    NOT NULL,         -- ERC721 | ERC1155
  wallet_address   TEXT    NOT NULL,
  amount           NUMERIC(76, 0) NOT NULL,  -- absolute post-balance of this token_id
  leg              TEXT    NOT NULL,
  vendor_event_id  TEXT    NOT NULL
);

-- Recency index for the DISTINCT ON; leads with the holding key so
-- REFRESH MATERIALIZED VIEW CONCURRENTLY avoids a sort.
CREATE INDEX noo_owner_recency_idx
  ON public.nft_owner_observations
  (token_address, token_id, wallet_address, block_number DESC, log_index DESC);

-- Current-owner surface: latest observation per (token, id, wallet).
CREATE MATERIALIZED VIEW public.nft_owners AS
SELECT DISTINCT ON (token_address, token_id, wallet_address)
  token_address, token_id, wallet_address, contract_type,
  amount, block_number, log_index, block_timestamp
FROM public.nft_owner_observations
ORDER BY token_address, token_id, wallet_address, block_number DESC, log_index DESC;

CREATE UNIQUE INDEX nft_owners_pk
  ON public.nft_owners (token_address, token_id, wallet_address);

-- Primary access path: current owners of a collection.
CREATE INDEX nft_owners_by_contract_active_idx
  ON public.nft_owners (token_address, token_id, wallet_address) WHERE amount > 0;

-- OPTIONAL: sibling by-wallet access path — every NFT held by a wallet.
-- CREATE INDEX nft_owners_by_wallet_active_idx
--   ON public.nft_owners (wallet_address, token_address, token_id) WHERE amount > 0;