PendingTransactionFilter

type PendingTransactionFilterType = {
  readonly filterId: FilterIdType;
  readonly type: "pendingTransaction";
} & {
  readonly [brand]: "PendingTransactionFilter";
};

import * as PendingTransactionFilter from './primitives/PendingTransactionFilter/index.js';
import * as FilterId from './primitives/FilterId/index.js';

// Create filter on node
const filterIdStr = await rpc.eth_newPendingTransactionFilter();
const filterId = FilterId.from(filterIdStr);

// Wrap in PendingTransactionFilter type
const filter = PendingTransactionFilter.from(filterId);

// Returns filter ID
const filterIdStr = await rpc.eth_newPendingTransactionFilter();
const filterId = FilterId.from(filterIdStr);
const filter = PendingTransactionFilter.from(filterId);

// Returns array of new pending transaction hashes since last poll
const txHashes = await rpc.eth_getFilterChanges(filter.filterId);

for (const hash of txHashes) {
  console.log(`New pending tx: ${hash}`);
}

const success = await rpc.eth_uninstallFilter(filter.filterId);

import * as PendingTransactionFilter from './primitives/PendingTransactionFilter/index.js';
import * as FilterId from './primitives/FilterId/index.js';
import { Hash } from './primitives/Hash/index.js';

// Install pending transaction filter
const filterIdStr = await rpc.eth_newPendingTransactionFilter();
const filterId = FilterId.from(filterIdStr);
const filter = PendingTransactionFilter.from(filterId);

console.log(`Pending tx filter installed: ${filterId.toString()}`);

// Poll every 5 seconds
const interval = setInterval(async () => {
  try {
    const hashes = await rpc.eth_getFilterChanges(filter.filterId);

    if (hashes.length > 0) {
      console.log(`New pending txs: ${hashes.length}`);

      // Fetch transaction details
      for (const hashStr of hashes) {
        const hash = Hash.from(hashStr);
        const tx = await rpc.eth_getTransactionByHash(hash);

        console.log(`From: ${tx.from}, To: ${tx.to}, Value: ${tx.value}`);
      }
    }
  } catch (error) {
    console.error('Filter error:', error);
    clearInterval(interval);
  }
}, 5000);

// Cleanup on exit
process.on('SIGINT', async () => {
  await rpc.eth_uninstallFilter(filter.filterId);
  clearInterval(interval);
  process.exit();
});

import * as PendingTransactionFilter from './primitives/PendingTransactionFilter/index.js';
import * as FilterId from './primitives/FilterId/index.js';
import { Denomination } from './primitives/Denomination/index.js';

const filterId = FilterId.from(await rpc.eth_newPendingTransactionFilter());
const filter = PendingTransactionFilter.from(filterId);

// Alert threshold: 100 ETH
const threshold = Denomination.parseEther("100");

setInterval(async () => {
  const hashes = await rpc.eth_getFilterChanges(filter.filterId);

  for (const hash of hashes) {
    const tx = await rpc.eth_getTransactionByHash(hash);

    if (tx.value >= threshold) {
      console.log(`HIGH VALUE TX: ${tx.value} wei`);
      console.log(`From: ${tx.from} -> To: ${tx.to}`);
      console.log(`Hash: ${hash}`);

      // Send alert (email, webhook, etc.)
      await sendAlert(tx);
    }
  }
}, 5000);

import * as PendingTransactionFilter from './primitives/PendingTransactionFilter/index.js';
import * as FilterId from './primitives/FilterId/index.js';

const filterId = FilterId.from(await rpc.eth_newPendingTransactionFilter());
const filter = PendingTransactionFilter.from(filterId);

// Track pending swaps
const pendingSwaps = new Map(); // methodId -> txs

// Uniswap swapExactTokensForTokens selector
const swapSelector = "0x38ed1739";

setInterval(async () => {
  const hashes = await rpc.eth_getFilterChanges(filter.filterId);

  for (const hash of hashes) {
    const tx = await rpc.eth_getTransactionByHash(hash);

    // Check if swap transaction
    if (tx.input.startsWith(swapSelector)) {
      const gasPrice = tx.maxFeePerGas || tx.gasPrice;

      // Check for frontrunning (same method, higher gas)
      const existing = pendingSwaps.get(swapSelector) || [];
      const frontruns = existing.filter(ptx => ptx.gasPrice < gasPrice);

      if (frontruns.length > 0) {
        console.log(`POTENTIAL FRONTRUN DETECTED`);
        console.log(`Original: ${frontruns[0].hash}`);
        console.log(`Frontrun: ${hash}`);
      }

      pendingSwaps.set(swapSelector, [...existing, { hash, gasPrice }]);
    }
  }

  // Clean up old txs (keep last 100)
  for (const [key, txs] of pendingSwaps) {
    if (txs.length > 100) {
      pendingSwaps.set(key, txs.slice(-100));
    }
  }
}, 5000);

// HTTP polling
const filterId = FilterId.from(await rpc.eth_newPendingTransactionFilter());
setInterval(async () => {
  const hashes = await rpc.eth_getFilterChanges(filterId);
  // Process hashes...
}, 5000);

// WebSocket subscription
const subscription = await ws.eth_subscribe('newPendingTransactions');
subscription.on('data', (txHash) => {
  console.log(`New pending tx: ${txHash}`);
});

// Calculate expected load
const txPerSecond = 150;
const pollInterval = 5; // seconds
const expectedTxsPerPoll = txPerSecond * pollInterval; // ~750 txs

console.log(`Expect ~${expectedTxsPerPoll} txs per poll`);

// INEFFICIENT: Fetch all txs
const hashes = await rpc.eth_getFilterChanges(filterId);
for (const hash of hashes) {
  const tx = await rpc.eth_getTransactionByHash(hash); // 750 requests!
}

// EFFICIENT: Filter by hash pattern first
const hashes = await rpc.eth_getFilterChanges(filterId);
const relevantHashes = hashes.filter(h => {
  // Filter by some criteria before fetching
  return h.startsWith('0xa');
});

// Then fetch only relevant txs
for (const hash of relevantHashes) {
  const tx = await rpc.eth_getTransactionByHash(hash);
}

const hashes = await rpc.eth_getFilterChanges(filterId);

// Batch request for all txs
const batch = hashes.map(hash => ({
  method: 'eth_getTransactionByHash',
  params: [hash],
  id: hash
}));

const txs = await rpc.batch(batch);

async function pollPendingTxFilter(filterId: FilterIdType) {
  try {
    const hashes = await rpc.eth_getFilterChanges(filterId);
    return hashes;
  } catch (error) {
    if (error.message.includes('filter not found')) {
      // Recreate filter
      const newFilterId = FilterId.from(
        await rpc.eth_newPendingTransactionFilter()
      );
      return pollPendingTxFilter(newFilterId);
    }
    throw error;
  }
}

const filterId = FilterId.from(await rpc.eth_newPendingTransactionFilter());

// Track MEV patterns
const mevSignatures = [
  "0x38ed1739", // Uniswap swap
  "0x7ff36ab5", // swapExactETHForTokens
  "0x18cbafe5"  // swapExactTokensForETH
];

setInterval(async () => {
  const hashes = await rpc.eth_getFilterChanges(filterId);

  for (const hash of hashes) {
    const tx = await rpc.eth_getTransactionByHash(hash);

    const isMEV = mevSignatures.some(sig =>
      tx.input.startsWith(sig)
    );

    if (isMEV && tx.gasPrice > threshold) {
      console.log(`Potential MEV: ${hash}`);
      analyzeMEV(tx);
    }
  }
}, 5000);

async function broadcastAndMonitor(signedTx: string) {
  // Start monitoring before broadcast
  const filterId = FilterId.from(
    await rpc.eth_newPendingTransactionFilter()
  );

  // Broadcast transaction
  const txHash = await rpc.eth_sendRawTransaction(signedTx);

  // Monitor for inclusion
  const interval = setInterval(async () => {
    const hashes = await rpc.eth_getFilterChanges(filterId);

    if (hashes.includes(txHash)) {
      console.log(`Transaction ${txHash} seen in mempool`);
    }

    // Check if mined
    const receipt = await rpc.eth_getTransactionReceipt(txHash);
    if (receipt) {
      console.log(`Transaction mined in block ${receipt.blockNumber}`);
      clearInterval(interval);
      await rpc.eth_uninstallFilter(filterId);
    }
  }, 5000);
}

const filterId = FilterId.from(await rpc.eth_newPendingTransactionFilter());

// Track gas prices
const gasPrices = [];

setInterval(async () => {
  const hashes = await rpc.eth_getFilterChanges(filterId);

  // Sample first 50 txs
  const sample = hashes.slice(0, 50);

  for (const hash of sample) {
    const tx = await rpc.eth_getTransactionByHash(hash);
    const gasPrice = tx.maxFeePerGas || tx.gasPrice;
    gasPrices.push(gasPrice);
  }

  // Keep last 1000 samples
  if (gasPrices.length > 1000) {
    gasPrices.splice(0, gasPrices.length - 1000);
  }

  // Calculate percentiles
  const sorted = [...gasPrices].sort((a, b) => Number(a - b));
  const p50 = sorted[Math.floor(sorted.length * 0.5)];
  const p75 = sorted[Math.floor(sorted.length * 0.75)];
  const p90 = sorted[Math.floor(sorted.length * 0.9)];

  console.log(`Gas prices - P50: ${p50}, P75: ${p75}, P90: ${p90}`);
}, 5000);