Deposit Processing

import { JsonRpcProvider, getAddress, id, zeroPadValue } from "ethers";

const provider = new JsonRpcProvider("https://public-evm-rpc.c8ntinuum.com");
const chainId = 8;
const transferTopic = id("Transfer(address,address,uint256)");

const monitored = new Set([
  getAddress("0x1111111111111111111111111111111111111111"),
  getAddress("0x2222222222222222222222222222222222222222"),
]);

async function scanNativeBlock(height) {
  const block = await provider.send("eth_getBlockByNumber", [
    "0x" + height.toString(16),
    true,
  ]);

  for (const tx of block.transactions) {
    if (!tx.to || tx.value === "0x0") continue;
    const to = getAddress(tx.to);
    if (!monitored.has(to)) continue;

    const receipt = await provider.getTransactionReceipt(tx.hash);
    if (!receipt || receipt.status !== 1) continue;

    await upsertDeposit({
      idempotencyKey: `${chainId}:${tx.hash}:native`,
      blockNumber: Number(receipt.blockNumber),
      blockHash: receipt.blockHash,
      txHash: tx.hash,
      from: getAddress(tx.from),
      to,
      asset: "CTM",
      amountWei: BigInt(tx.value).toString(),
    });
  }
}

async function scanErc20(tokenAddress, fromBlock, toBlock) {
  for (const address of monitored) {
    const recipientTopic = zeroPadValue(address, 32);
    const logs = await provider.getLogs({
      address: tokenAddress,
      topics: [transferTopic, null, recipientTopic],
      fromBlock,
      toBlock,
    });

    for (const log of logs) {
      const receipt = await provider.getTransactionReceipt(log.transactionHash);
      if (!receipt || receipt.status !== 1) continue;

      await upsertDeposit({
        idempotencyKey: `${chainId}:${log.transactionHash}:${log.index}:${tokenAddress}`,
        blockNumber: log.blockNumber,
        blockHash: log.blockHash,
        txHash: log.transactionHash,
        token: getAddress(tokenAddress),
        to: address,
        amountRaw: BigInt(log.data).toString(),
      });
    }
  }
}

async function upsertDeposit(deposit) {
  // Store first, then credit from a separate idempotent ledger step.
  console.log(deposit);
}

import java.math.BigInteger;
import java.util.List;
import java.util.Set;
import org.web3j.crypto.Keys;
import org.web3j.protocol.Web3j;
import org.web3j.protocol.core.DefaultBlockParameterNumber;
import org.web3j.protocol.core.methods.request.EthFilter;
import org.web3j.protocol.core.methods.response.EthBlock;
import org.web3j.protocol.core.methods.response.EthGetTransactionReceipt;
import org.web3j.protocol.core.methods.response.EthLog;
import org.web3j.protocol.http.HttpService;
import org.web3j.utils.Numeric;

Web3j web3j = Web3j.build(new HttpService("https://public-evm-rpc.c8ntinuum.com"));
long chainId = 8L;
String transferTopic = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef";
Set<String> monitored = Set.of(Keys.toChecksumAddress("0x1111111111111111111111111111111111111111"));

void scanNativeBlock(BigInteger height) throws Exception {
  EthBlock block = web3j
      .ethGetBlockByNumber(new DefaultBlockParameterNumber(height), true)
      .send();

  for (EthBlock.TransactionResult<?> result : block.getBlock().getTransactions()) {
    EthBlock.TransactionObject tx = (EthBlock.TransactionObject) result.get();
    if (tx.getTo() == null || tx.getValue().signum() == 0) continue;

    String to = Keys.toChecksumAddress(tx.getTo());
    if (!monitored.contains(to)) continue;

    EthGetTransactionReceipt receiptResponse =
        web3j.ethGetTransactionReceipt(tx.getHash()).send();
    if (receiptResponse.getTransactionReceipt().isEmpty()) continue;
    var receipt = receiptResponse.getTransactionReceipt().get();
    if (!receipt.isStatusOK()) continue;

    upsertDeposit(chainId + ":" + tx.getHash() + ":native", tx.getHash(), to, tx.getValue());
  }
}

void scanErc20(String token, BigInteger fromBlock, BigInteger toBlock) throws Exception {
  for (String address : monitored) {
    String topicTo = "0x" + Numeric.cleanHexPrefix(address).toLowerCase();
    topicTo = "0x" + "0".repeat(24) + Numeric.cleanHexPrefix(topicTo);

    EthFilter filter = new EthFilter(
        new DefaultBlockParameterNumber(fromBlock),
        new DefaultBlockParameterNumber(toBlock),
        token);
    filter.addSingleTopic(transferTopic);
    filter.addOptionalTopics(null);
    filter.addSingleTopic(topicTo);

    EthLog logs = web3j.ethGetLogs(filter).send();
    for (EthLog.LogResult<?> item : logs.getLogs()) {
      EthLog.LogObject log = (EthLog.LogObject) item.get();
      EthGetTransactionReceipt receiptResponse =
          web3j.ethGetTransactionReceipt(log.getTransactionHash()).send();
      if (receiptResponse.getTransactionReceipt().isEmpty()) continue;
      if (!receiptResponse.getTransactionReceipt().get().isStatusOK()) continue;

      String key = chainId + ":" + log.getTransactionHash() + ":" + log.getLogIndex() + ":" + token;
      upsertDeposit(key, log.getTransactionHash(), address, Numeric.toBigInt(log.getData()));
    }
  }
}

void upsertDeposit(String idempotencyKey, String txHash, String to, BigInteger amount) {
  System.out.println(idempotencyKey + " " + txHash + " " + to + " " + amount);
}

package scanner

import (
	"context"
	"fmt"
	"math/big"

	"github.com/ethereum/go-ethereum"
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/ethereum/go-ethereum/ethclient"
	"github.com/ethereum/go-ethereum/crypto"
)

var transferTopic = crypto.Keccak256Hash([]byte("Transfer(address,address,uint256)"))

func ScanNativeBlock(ctx context.Context, client *ethclient.Client, chainID int64, height uint64, monitored map[common.Address]bool) error {
	block, err := client.BlockByNumber(ctx, new(big.Int).SetUint64(height))
	if err != nil {
		return err
	}

	for _, tx := range block.Transactions() {
		if tx.To() == nil || tx.Value().Sign() == 0 || !monitored[*tx.To()] {
			continue
		}

		receipt, err := client.TransactionReceipt(ctx, tx.Hash())
		if err != nil || receipt.Status != types.ReceiptStatusSuccessful {
			continue
		}

		key := fmt.Sprintf("%d:%s:native", chainID, tx.Hash())
		if err := UpsertDeposit(key, tx.Hash(), *tx.To(), tx.Value()); err != nil {
			return err
		}
	}
	return nil
}

func ScanERC20(ctx context.Context, client *ethclient.Client, chainID int64, token common.Address, fromBlock, toBlock uint64, recipients []common.Address) error {
	for _, recipient := range recipients {
		query := ethereum.FilterQuery{
			FromBlock: new(big.Int).SetUint64(fromBlock),
			ToBlock:   new(big.Int).SetUint64(toBlock),
			Addresses: []common.Address{token},
			Topics: [][]common.Hash{
				{transferTopic},
				nil,
				{common.BytesToHash(recipient.Bytes())},
			},
		}

		logs, err := client.FilterLogs(ctx, query)
		if err != nil {
			return err
		}

		for _, log := range logs {
			receipt, err := client.TransactionReceipt(ctx, log.TxHash)
			if err != nil || receipt.Status != types.ReceiptStatusSuccessful {
				continue
			}
			amount := new(big.Int).SetBytes(log.Data)
			key := fmt.Sprintf("%d:%s:%d:%s", chainID, log.TxHash, log.Index, token)
			if err := UpsertDeposit(key, log.TxHash, recipient, amount); err != nil {
				return err
			}
		}
	}
	return nil
}

func UpsertDeposit(key string, txHash common.Hash, to common.Address, amount *big.Int) error {
	return nil
}

use alloy::primitives::{keccak256, Address, B256, U256};
use alloy::providers::{Provider, ProviderBuilder};
use alloy::rpc::types::{BlockNumberOrTag, Filter};
use std::collections::HashSet;

const CHAIN_ID: u64 = 8;

async fn scan_native_block<P: Provider>(
    provider: &P,
    height: u64,
    monitored: &HashSet<Address>,
) -> eyre::Result<()> {
    let block = provider
        .get_block_by_number(BlockNumberOrTag::Number(height), true.into())
        .await?
        .ok_or_else(|| eyre::eyre!("missing block"))?;

    for tx in block.transactions.into_transactions() {
        let Some(to) = tx.to else { continue };
        if tx.value == U256::ZERO || !monitored.contains(&to) {
            continue;
        }

        let Some(receipt) = provider.get_transaction_receipt(tx.hash).await? else {
            continue;
        };
        if !receipt.status() {
            continue;
        }

        let key = format!("{CHAIN_ID}:{}:native", tx.hash);
        upsert_deposit(&key, tx.hash, to, tx.value).await?;
    }

    Ok(())
}

async fn scan_erc20<P: Provider>(
    provider: &P,
    token: Address,
    from_block: u64,
    to_block: u64,
    recipients: &[Address],
) -> eyre::Result<()> {
    let transfer_topic: B256 = keccak256("Transfer(address,address,uint256)");

    for recipient in recipients {
        let filter = Filter::new()
            .address(token)
            .event_signature(transfer_topic)
            .topic2(B256::left_padding_from(recipient.as_slice()))
            .from_block(from_block)
            .to_block(to_block);

        for log in provider.get_logs(&filter).await? {
            let Some(receipt) = provider.get_transaction_receipt(log.transaction_hash.unwrap()).await? else {
                continue;
            };
            if !receipt.status() {
                continue;
            }

            let amount = U256::from_be_slice(&log.data().data);
            let key = format!(
                "{CHAIN_ID}:{}:{}:{}",
                log.transaction_hash.unwrap(),
                log.log_index.unwrap_or_default(),
                token
            );
            upsert_deposit(&key, log.transaction_hash.unwrap(), *recipient, amount).await?;
        }
    }

    Ok(())
}

async fn upsert_deposit(key: &str, tx_hash: B256, to: Address, amount: U256) -> eyre::Result<()> {
    println!("{key} {tx_hash} {to} {amount}");
    Ok(())
}

#[tokio::main]
async fn main() -> eyre::Result<()> {
    let provider = ProviderBuilder::new().on_http(
        "https://public-evm-rpc.c8ntinuum.com".parse()?,
    );
    let monitored = HashSet::new();
    scan_native_block(&provider, 123456, &monitored).await?;
    Ok(())
}