Tokenized equities & RWA data layer

name: tokenized-stock-holders
resource_size: s

sources:
  rh_transfers:
    type: dataset
    dataset_name: robinhood_testnet.erc20_transfers
    version: 1.0.0
    start_at: earliest

transforms:
  # Every transfer moves a balance: CREDIT the recipient (+amount) and
  # DEBIT the sender (-amount). Emitting both rows is what makes the
  # running balance net correctly - a credit-only stream overstates holdings.
  balance_deltas:
    type: sql
    primary_key: id
    sql: |
      -- Credit: tokens received (also covers mints, where the sender is the zero address)
      SELECT
        CONCAT(id, '-in')            AS id,
        lower(recipient)             AS holder,
        lower(address)               AS token,
        CAST(amount AS DECIMAL(38, 0)) AS delta,
        _gs_op
      FROM rh_transfers
      WHERE recipient <> '0x0000000000000000000000000000000000000000'

      UNION ALL

      -- Debit: tokens sent (also covers burns, where the recipient is the zero address)
      SELECT
        CONCAT(id, '-out')           AS id,
        lower(sender)                AS holder,
        lower(address)               AS token,
        -CAST(amount AS DECIMAL(38, 0)) AS delta,
        _gs_op
      FROM rh_transfers
      WHERE sender <> '0x0000000000000000000000000000000000000000'

sinks:
  # The aggregate sink sums delta per (holder, token) via a DB trigger,
  # maintaining the current balance incrementally.
  holdings:
    type: postgres_aggregate
    from: balance_deltas
    schema: registry
    landing_table: holdings_log
    agg_table: holdings
    primary_key: id
    secret_name: MY_POSTGRES
    group_by:
      holder:
        type: text
      token:
        type: text
    aggregate:
      balance:
        from: delta
        fn: sum
        type: numeric(38,0)

import type { TaskContext, Chain } from "compose";

// Dividends must go out on the SAME network the stock token lives on - the
// chain this guide indexes holders from. Robinhood Chain (an Arbitrum Orbit L2)
// isn't a built-in Edge RPC chain yet, so define it as a custom chain and fill
// in its id/RPC. Swap for evm.chains.<name> once it's available on Edge.
const STOCK_CHAIN: Chain = {
  id: 0, // Robinhood Chain testnet chain ID
  name: "Robinhood Chain Testnet",
  testnet: true,
  nativeCurrency: { name: "Ether", symbol: "ETH", decimals: 18 },
  rpcUrls: {
    default: { http: ["https://<robinhood-chain-rpc>"] },
    public: { http: ["https://<robinhood-chain-rpc>"] },
  },
  blockExplorers: { default: { name: "Explorer", url: "https://<explorer>" } },
};

export async function main(
  context: TaskContext,
  payload: { token: string; dividendPerToken: string; recordBlock: number },
) {
  const { fetch, evm } = context;

  // Holders as of the record-date block, from your registry service
  const holders = await fetch<{ address: string; balance: string }[]>(
    `https://registry.example.com/holders?token=${payload.token}&block=${payload.recordBlock}`,
  );

  const wallet = await evm.wallet({ name: "dividend-payer", sponsorGas: true });

  const results = await Promise.allSettled(
    holders.map((h) => {
      const amount = (BigInt(h.balance) * BigInt(payload.dividendPerToken)) / 10n ** 18n;
      // Robinhood Chain stablecoins are ERC-20 - pay with the standard transfer.
      return wallet.writeContract(
        STOCK_CHAIN,
        "0xDIVIDEND_STABLECOIN",
        "transfer(address,uint256)",
        [h.address, amount],
      );
    }),
  );

  const paid = results.filter((r) => r.status === "fulfilled").length;
  return { holders: holders.length, paid };
}