Hướng dẫn thu thập dữ liệu on-chain thông qua lắng nghe events trên Blockchain

Trong thực tế, chúng ta có rất nhiều nghiệp vụ thực tế đòi hỏi phải thu thập dữ liệu từ Blockchain như:

• Đồng bộ dữ liệu giữa giao diện DApp và blockchain. Và người dùng sẽ luôn muốn thấy các thông tin được cập nhật ngay lập tức sau khi các transaction của họ được thực thi xong. 
• Theo dõi các lệnh giao dịch, rút thanh khoản hay thêm thanh khoản trên các Pool thanh khoản
• Theo dõi tài sản của một ví
• …

Để thu thập dữ liệu, chúng ta có hai cách:

• C1: Request trực tiếp tới các RPC Node để lấy dữ liệu. Trong trường hợp này thường người ta hay dùng Multicall: Hợp đồng Multicall là gì? Tại sao lại sử dụng hợp đồng Multicall?
• C2: Là theo dõi các events trên blockchain qua đó cập nhật dữ liệu. Trong bài viết này, chúng ta đi sâu cách này.

Giới thiệu về events

Event thì là một thành phần cũng có thể được kế thừa như Function. Nó thì thường được sử dụng để broadcast ra cho phía client biết rằng hàm nào đó đã được gọi và thực thi. Nó hay được để ở cuối của Function và cũng có các tham số để phía client có thể hiểu được. Ví dụ như một token ERC20 thường emit ra một event Transfer với các thông số người gửi, người nhận và value bao nhiêu, để các clients có thể biết được là vừa có một giao dịch chuyển token.

...

// Định nghĩa event Transfer
event Transfer(address indexed from, address indexed to, uint256 value);

...

// Hàm transfer token
function transfer(address to, uint256 value) public returns (bool) {
    require(value <= _balances[msg.sender]);
    require(to != address(0));

    _balances[msg.sender] = _balances[msg.sender].sub(value);
    _balances[to] = _balances[to].add(value);
    emit Transfer(msg.sender, to, value);
    return true;
}

Kết quả trả về sẽ có dạng như sau:

 Transfer (index_topic_1 address from, index_topic_2 address to, uint256 value)
   address from
   0x8f7f6c18039776c8b7f952185c3e75649a25b9cb
   address to
   0x611abc072ee91c0cc19ffef97ac7e69a1a7a17ec
   uint256 value
   6538744793000000000000
   
   [topic0] 0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef
   [topic1] 0x0000000000000000000000008f7f6c18039776c8b7f952185c3e75649a25b9cb
   [topic2] 0x000000000000000000000000611abc072ee91c0cc19ffef97ac7e69a1a7a17ec

Môt số cách lắng nghe events của một Smart Contract trên Blockchain

Để lắng nghe các events của một Smart Contract trên Blockchain, bạn cần phải có:

• Websocket Provider: Trên Ethereum bạn có thể lấy thông tin này từ Infura. Chi tiết xem: Getting Started With Infura
• Contract Address: Địa chỉ của Contract mà bạn muốn theo dõi các events
• ABI của Contract: Bạn phải có ABI của contract mà bạn muốn theo dõi. Với các dự án uy tín, thông thường ABI và source code của Contract đều được public, bạn có thể lấy nó trên Etherscan.
• Thư viện Web3: Bạn cũng có thể chọn thư viện khác nếu muốn. Mình chọn thư viện này vì nó được sử dụng phổ biến hiện nay.

const Web3 = require('web3');
const ABI = require('./ABI-WETH.json');

const web3 = new Web3('wss://mainnet.infura.io/ws/v3/<PROJECT_ID_INFURA>');
const CONTRACT_ADDRESS = '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2';
const myContract = new web3.eth.Contract(ABI, CONTRACT_ADDRESS);

Chi tiết hơn bạn xem tài liệu: web3.eth.Contract

Bây giờ chúng ta sẽ đi sâu hơn vào một số cách lấy dữ liệu của các events.

C1: Lấy các dữ liệu các sự kiện trong quá khứ

Thực ra cách này thì giống với việc truy vấn dữ liệu từ blockchain. Do các events được lưu trữ trên blockchain nên ta có thể query nó bất cứ khi nào. Chúng ta sẽ sử dụng phương thức getPastEvents() có sẵn trong đối tượng Contract mà phía trên ta đã tạo để lấy về các events trong quá khứ. Ví dụ như ở đây chúng ta sẽ get về tất các event Transfer của contract ERC20 WETH từ block 12856158 đến block mới nhất:

async function example1() {
    let options = {
        filter: {
            value: ['1000', '1337']    // Only get events where transfer value was 1000 or 1337
        },
        fromBlock: 12856158,                  // Number || "earliest" || "pending" || "latest"
        toBlock: 'latest'
    };

    myContract.getPastEvents('Transfer', options)
        .then(result => console.log ('result', result))
        .catch(err => console.log ('error', err.message, err.stack));
}

Kết quả:

  ...

  {
    address: '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2',
    blockHash: '0x5d6592c6f8be4443dd79b27fafa35bb58fd8b71161a5d58715a8888a0b186601',
    blockNumber: 12856161,
    logIndex: 287,
    removed: false,
    transactionHash: '0x7c9d4b3655b89179b09bc025ecc12530eb38111c7cc79312c85786bd7d8a398a',
    transactionIndex: 180,
    id: 'log_980923fb',
    returnValues: Result {
      '0': '0x05f21E62952566CeFb77F5153Ec6B83C14FB6b1D',
      '1': '0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D',
      '2': '20568368904018071',
      src: '0x05f21E62952566CeFb77F5153Ec6B83C14FB6b1D',
      dst: '0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D',
      wad: '20568368904018071'
    },
    event: 'Transfer',
    signature: '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef',
    raw: {
      data: '0x000000000000000000000000000000000000000000000000004912d292221897',
      topics: [Array]
    }
  },
  ... 404 more items
]

Thực tế bạn có thể filter với nhiều điều kiện chi tiết hơn để lấy đúng dữ liệu mà mình cần.

C2: Theo dõi realtime các events của một Contract

Chúng ta sử dụng trực tiếp các phương thức tương ứng với tên sự kiện trong events của đối tượng Contract. Khi một đối tượng Contract được tạo ra, nó sẽ có tất cả các events được khai báo trong ABI của contract. Ví dụ code lắng nghe sự kiện Transfer như sau:

function example2() {
    let options = {
        filter: {
            value: [],
        },
        fromBlock: 12856518
    };

    myContract.events.Transfer(options)
        .on('data', event => console.log(event))
        .on('changed', changed => console.log(changed))
        .on('error', err => console.log ('error', err.message, err.stack))
        .on('connected', str => console.log(str))
}

Mỗi phương thức trong events sẽ trả về một EventEmitter. Event này sẽ lắng nghe các sự kiện như dưới đây.

• data – Nó sẽ được kích hoạt mỗi khi có một event Transfer gọi đến contract được emited
• changed – Nó sẽ được kích hoạt mỗi khi event bị xóa khỏi blockchain
• error – Sẽ được kích hoạt khi có lỗi trong quá trình subscription events
• connected – Sẽ được kích hoạt khi thiết lập subscription đã thành công. Nó sẽ trả về một subscription id và điều này chỉ xảy ra một lần duy nhất.

Kết quả như sau:

...

{
  address: '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2',
  blockHash: '0xee2da8ff4b39294545ed2af49ee40c17ddad9c12d21d0733554a2579b2ca006b',
  blockNumber: 12856552,
  logIndex: 252,
  removed: false,
  transactionHash: '0x1871d9dbe683796d4219ed34403495dd5bf8784e9f73e92602adcac9ebd887fe',
  transactionIndex: 147,
  id: 'log_190e2a37',
  returnValues: Result {
    '0': '0x11b815efB8f581194ae79006d24E0d814B7697F6',
    '1': '0xE592427A0AEce92De3Edee1F18E0157C05861564',
    '2': '1667655933682553825',
    src: '0x11b815efB8f581194ae79006d24E0d814B7697F6',
    dst: '0xE592427A0AEce92De3Edee1F18E0157C05861564',
    wad: '1667655933682553825'
  },
  event: 'Transfer',
  signature: '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef',
  raw: {
    data: '0x0000000000000000000000000000000000000000000000001724b43c6eb893e1',
    topics: [
      '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef',
      '0x00000000000000000000000011b815efb8f581194ae79006d24e0d814b7697f6',
      '0x000000000000000000000000e592427a0aece92de3edee1f18e0157c05861564'
    ]
  }
}

...

Một số cách lắng nghe events tổng quát trên blockchain

Tức là bạn không phải lắng nghe events từ một Contract cụ thể nào mà là của một hoặc nhiều contract bất kỳ hoặc toàn bộ events ở trên Blockchain, tùy thuộc vào điều kiện filter của bạn. Trong trường hợp này chúng ta đều dùng hàm web3.eth.subscribe mặc định hỗ trợ trong Web3.

Lắng nghe các event logs: subscribe(“logs”)

Ở đây chúng ta dùng hàm web3.eth.subscribe() có sẵn trong web3 và subscibe event logs. Kiểu này là một cách tổng quát nhất để có thể lắng tất cả các logs events đã được emited. Và nếu muốn lọc events thì bạn phải thiết lập trong phần options:

function example3() {
    let options = {
        fromBlock: 12856551,
        address: ['address-1', 'address-2'],    // Only get events from specific addresses
        topics: []                              // What topics to subscribe to
    };

    let subscription = web3.eth.subscribe('logs', options,(err,event) => {
        if (!err)
        console.log(event)
    });

    subscription.on('data', event => console.log(event))
    subscription.on('changed', changed => console.log(changed))
    subscription.on('error', err => console.log ('error', err.message, err.stack))
    subscription.on('connected', nr => console.log(nr))
}

Kiểu subscribe này sẽ trả lại một thể hiện Subscription nó cũng là một EventEmitter. Vì vậy các events cũng giống với cách số 2, ngoài ra nó cũng có thêm một vài properties khác như:

• id – Đây là id của subscription
• unsubscribe(callback) – Method này được sử dụng để unsubscribe
• subscribe(callback) – Method này được sử dụng để re-subscribe khởi tạo lại subscription với những parameters đã có trước đó
• arguments – Subscription arguments như ở trên thì được nó được sử dụng để re-subscribe.

Theo dõi các pending transactions: subscribe(“pendingTransactions”)

Chúng ta có thể lắng nghe sự kiện kể từ khi transaction còn đang ở trạng pending (Trạng thái đang chờ để được confirm). Nó phù hợp để sử dụng cho các trường hợp cần theo dõi các giao dịch lớn và tạo transaction với tốc độ nhanh giống như front-running.

function example4() {
    let subscription = web3.eth.subscribe('pendingTransactions', function(error, result){
        if (!error)
        console.log(result)
    });

    subscription.on('data', event => console.log(event))
    subscription.on('changed', changed => console.log(changed))
    subscription.on('error', err => console.log ('error', err.message, err.stack))
    subscription.on('connected', nr => console.log(nr))

    // unsubscribes the subscription
    subscription.unsubscribe(function(error, success){
        if(success)
            console.log('Successfully unsubscribed!');
    });
}

Kết quả ta sẽ thấy rất nhiều các mã hash của các pending transaction mới được tạo và đang ở trong mempool. Ở đây ta có thể lấy dữ liệu chi tiết của từng transaction sau đó giải mã giá trị trong trong params của transaction là ta có thể thấy được các tông tin cần thiết. Để có thể ngay lập tạo một transaction khác front-running.

Một số blockchain mới theo cơ chế mới không cho phép theo dõi mempool để lấy các pending transaction, do đó loại bỏ được việc tấn công front-running.

Theo dõi để lấy thông tin khối mới: subscribe(“newBlockHeaders”)

Bạn có thể subscribe để lấy thông tin khối mới. Điều này có thể được sử dụng làm bộ đếm thời gian để kiểm tra các thay đổi trên chuỗi khối.

var subscription = web3.eth.subscribe('newBlockHeaders', function(error, result){
    if (!error) {
        console.log(result);

        return;
    }

    console.error(error);
})
.on("connected", function(subscriptionId){
    console.log(subscriptionId);
})
.on("data", function(blockHeader){
    console.log(blockHeader);
})
.on("error", console.error);

Thực ra nếu bạn subcribe sự kiện newBlockHeaders này thì tương đương với việc bạn đồng bộ toàn bộ dữ liệu của blockchain tính từ thời điểm bạn bắt đầu subcribe. Thực sự lượng dữ liệu này rất lớn và rất tốn băng thông của mạng.

Một số vấn đề thực tế phát sinh khi theo dõi events trong thời gian thực

Dữ liệu các event trả về thường chỉ có Block Number mà không có Block Timestamp

Tất cả dữ liệu các event trả về đều chỉ có Block Number tương ứng với giao dịch tạo ra event đó mà không có Block Timestamp. Trong thực tế, các nghiệp vụ thu thập dữ liệu thường đòi hỏi dữ liệu mới nhất, như vậy thông tin Block Timestamp quan trọng hơn, để sự kiện hoặc dữ liệu đó đã cập nhật cách hiện tại bao lâu.

Như vậy để có được Block Timestamp chúng ta phải truy vấn thêm 1 lần nữa tới Blockchain thông qua RPC.

Không lấy được Block Timestamp của một Block Number của một sự kiện bắn ra

Đây là thực tế tôi gặp khi theo dõi sự kiện SWAP của các pool Uniswap V3 trên Arbitrum. Sau khi dữ liệu trả về, tôi lấy được Block Number, và request qua RPC để lấy Block Timestamp. Nhưng thực tế không phải lúc nào cũng lấy được Block Timestamp mà nhiều lúc không lấy được Timestamp, có thể event bắn đến trước nhưng dữ liệu node vẫn chưa đồng bộ được block mới nhất.

Việc này ảnh hưởng khá nhiều tới luồng cập nhật dữ liệu.

Theo dõi thông tin khối mới newBlockHeaders khá tốn tài nguyên

Tôi đã từng thử theo dõi để liên tục lấy thông tin khối mới thông qua subcribe sự kiện newBlockHeaders trên Arbitrum, thông qua đó để xác định tất cả các thay đổi liên quan. Tôi không ngờ việc này tốn tài nguyên kinh khủng.

Tôi đăng ký tài khoản Alchemy lấy link Websocket trên đó để thử. Hàng tháng trang này cho phép 300 triệu CCU miễn phí, nếu dùng RPC thì cũng khá thoải mái. Vậy mà khi tôi dùng để theo dõi khối mới qua sự kiên newBlockHeaders thì chỉ trong vòng 4h đồng hồ đã sử dụng hết luôn cả 300 triệu CCU. Thực sự cách này chỉ phù hợp nếu bạn có Node riêng, chứ mà thuê node thì cực kỳ tốn kém.

Cập nhật dữ liệu thanh khoản pool thông qua sự kiện swap

Như bạn đã biết thông tin quan trong nhất của một Pool V2 là reserve0 và reserve1. Chỉ cần thông tin này, ta có thể tính toán lượng đầu ra với đầu vào bất kỳ. Thay vì phải dùng RPC để lấy thông tin này liên tục, tôi có ý tưởng là sẽ theo dõi sự kiện SWAP, MINT, BURN của các pool này, và từ dữ liệu đó cập nhật lại reserve0 và reserve1.

Với sự kiện SWAP, ta sẽ biết được có bao nhiêu token0 / token1 vào hoặc ra khỏi pool, từ đó cộng trừ với reserve0 và reserve1. Nhưng dần dần tôi thấy dữ liệu theo cách này không chính xác vì:

• Nhiều pool khi swap thì một phần phí swap sẽ vào lại pool, khá khó để tính lượng phí này.
• Các events nhiều khi bị lặp
• Các events nhiều khi không có do node có vấn đề hoặc do đường truyền mạng.

Lắng nghe sự kiện SWAP để thu thập dữ liệu giao dịch liên quan tới Pool thì okie, nhưng để thu thập dữ liệu thanh khoản của Pool thì không được ổn lắm. Nên các sự kiện này chỉ để giúp chúng ta biết có sự thay đổi dữ liệu trong pool đó, ta vẫn phải gọi RPC để lấy dữ liệu chính xác.