nautilus_blockchain/exchanges/parsing/uniswap_v3/
fee_protocol_collect.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use alloy::{dyn_abi::SolType, primitives::Address, sol};
17use nautilus_model::defi::{PoolIdentifier, SharedDex, rpc::RpcLog};
18use ustr::Ustr;
19
20use crate::{
21    events::fee_protocol_collect::FeeProtocolCollectEvent,
22    hypersync::{
23        HypersyncLog,
24        log::{
25            extract_address_from_topic, extract_block_number, extract_log_index,
26            extract_transaction_hash, extract_transaction_index, validate_event_signature_hash,
27        },
28    },
29    rpc::log as rpc_log,
30};
31
32const FEE_PROTOCOL_COLLECT_EVENT_SIGNATURE_HASH: &str =
33    "596b573906218d3411850b26a6b437d6c4522fdb43d2d2386263f86d50b8b151";
34
35// Define sol macro for parsing CollectProtocol event data.
36// sender and recipient are indexed (in topics); amount0 and amount1 are non-indexed (in data):
37// amount0 (uint128), amount1 (uint128)
38sol! {
39    struct FeeProtocolCollectEventData {
40        uint128 amount0;
41        uint128 amount1;
42    }
43}
44
45/// Parses a `CollectProtocol` event from a Uniswap V3 HyperSync log.
46///
47/// # Errors
48///
49/// Returns an error if the log parsing fails or if the event data is invalid.
50///
51/// # Panics
52///
53/// Panics if the contract address is not set in the log.
54pub fn parse_fee_protocol_collect_event_hypersync(
55    dex: SharedDex,
56    log: &HypersyncLog,
57) -> anyhow::Result<FeeProtocolCollectEvent> {
58    validate_event_signature_hash(
59        "CollectProtocol",
60        FEE_PROTOCOL_COLLECT_EVENT_SIGNATURE_HASH,
61        log,
62    )?;
63
64    let sender = extract_address_from_topic(log, 1, "sender")?;
65    let recipient = extract_address_from_topic(log, 2, "recipient")?;
66
67    if let Some(data) = &log.data {
68        let data_bytes = data.as_ref();
69
70        // Validate the data contains 2 parameters of 32 bytes each
71        if data_bytes.len() < 2 * 32 {
72            anyhow::bail!("CollectProtocol event data is too short");
73        }
74
75        let decoded = match <FeeProtocolCollectEventData as SolType>::abi_decode(data_bytes) {
76            Ok(decoded) => decoded,
77            Err(e) => anyhow::bail!("Failed to decode CollectProtocol event data: {e}"),
78        };
79
80        let pool_address = Address::from_slice(
81            log.address
82                .clone()
83                .expect("Contract address should be set in logs")
84                .as_ref(),
85        );
86        let pool_identifier = PoolIdentifier::Address(Ustr::from(&pool_address.to_string()));
87
88        Ok(FeeProtocolCollectEvent::new(
89            dex,
90            pool_identifier,
91            extract_block_number(log)?,
92            extract_transaction_hash(log)?,
93            extract_transaction_index(log)?,
94            extract_log_index(log)?,
95            sender,
96            recipient,
97            decoded.amount0,
98            decoded.amount1,
99        ))
100    } else {
101        anyhow::bail!("Missing data in CollectProtocol event log");
102    }
103}
104
105/// Parses a `CollectProtocol` event from an RPC log.
106///
107/// # Errors
108///
109/// Returns an error if the log parsing fails or if the event data is invalid.
110pub fn parse_fee_protocol_collect_event_rpc(
111    dex: SharedDex,
112    log: &RpcLog,
113) -> anyhow::Result<FeeProtocolCollectEvent> {
114    rpc_log::validate_event_signature(
115        log,
116        FEE_PROTOCOL_COLLECT_EVENT_SIGNATURE_HASH,
117        "CollectProtocol",
118    )?;
119
120    let sender = rpc_log::extract_address_from_topic(log, 1, "sender")?;
121    let recipient = rpc_log::extract_address_from_topic(log, 2, "recipient")?;
122
123    let data_bytes = rpc_log::extract_data_bytes(log)?;
124
125    // Validate the data contains 2 parameters of 32 bytes each
126    if data_bytes.len() < 2 * 32 {
127        anyhow::bail!("CollectProtocol event data is too short");
128    }
129
130    let decoded = match <FeeProtocolCollectEventData as SolType>::abi_decode(&data_bytes) {
131        Ok(decoded) => decoded,
132        Err(e) => anyhow::bail!("Failed to decode CollectProtocol event data: {e}"),
133    };
134
135    let pool_address = rpc_log::extract_address(log)?;
136    let pool_identifier = PoolIdentifier::Address(Ustr::from(&pool_address.to_string()));
137    Ok(FeeProtocolCollectEvent::new(
138        dex,
139        pool_identifier,
140        rpc_log::extract_block_number(log)?,
141        rpc_log::extract_transaction_hash(log)?,
142        rpc_log::extract_transaction_index(log)?,
143        rpc_log::extract_log_index(log)?,
144        sender,
145        recipient,
146        decoded.amount0,
147        decoded.amount1,
148    ))
149}
150
151#[cfg(test)]
152mod tests {
153    use std::sync::Arc;
154
155    use rstest::*;
156    use serde_json::json;
157
158    use super::*;
159    use crate::exchanges::arbitrum;
160
161    // CollectProtocol log at block 0x175a6484 (391799940), pool
162    // 0xd13040d4fe917EE704158CfCB3338dCd2838B245.
163    // sender:    0xc36442b4a4522e871399cd717abdd847ab11fe88
164    // recipient: 0xa61da382c18d9d5beb905ea192bae25e4c15d512
165    // Asymmetric amounts (amount0 != amount1) catch a token0/token1 column swap.
166    const EXPECTED_AMOUNT0: u128 = 0x1234_5678_90ab_cdef;
167    const EXPECTED_AMOUNT1: u128 = 0x0fed_cba9_8765_4321;
168    const DATA: &str = "0x0000000000000000000000000000000000000000000000001234567890abcdef0000000000000000000000000000000000000000000000000fedcba987654321";
169
170    #[fixture]
171    fn hypersync_log() -> HypersyncLog {
172        let log_json = json!({
173            "removed": null,
174            "log_index": "0x11",
175            "transaction_index": "0x5",
176            "transaction_hash": "0x0c70f6d6bcf8508ba620b9d1250c95ad67108e35707c5d7456349ea207051bae",
177            "block_hash": null,
178            "block_number": "0x175a6484",
179            "address": "0xd13040d4fe917EE704158CfCB3338dCd2838B245",
180            "data": DATA,
181            "topics": [
182                "0x596b573906218d3411850b26a6b437d6c4522fdb43d2d2386263f86d50b8b151",
183                "0x000000000000000000000000c36442b4a4522e871399cd717abdd847ab11fe88",
184                "0x000000000000000000000000a61da382c18d9d5beb905ea192bae25e4c15d512"
185            ]
186        });
187        serde_json::from_value(log_json).expect("Failed to deserialize HyperSync log")
188    }
189
190    #[fixture]
191    fn rpc_log() -> RpcLog {
192        let log_json = json!({
193            "removed": false,
194            "logIndex": "0x11",
195            "transactionIndex": "0x5",
196            "transactionHash": "0x0c70f6d6bcf8508ba620b9d1250c95ad67108e35707c5d7456349ea207051bae",
197            "blockHash": "0xe925eaa1f5178ceedfa24043a974edb928ddab7195600b6b99ff5403fbf13c8b",
198            "blockNumber": "0x175a6484",
199            "address": "0xd13040d4fe917EE704158CfCB3338dCd2838B245",
200            "data": DATA,
201            "topics": [
202                "0x596b573906218d3411850b26a6b437d6c4522fdb43d2d2386263f86d50b8b151",
203                "0x000000000000000000000000c36442b4a4522e871399cd717abdd847ab11fe88",
204                "0x000000000000000000000000a61da382c18d9d5beb905ea192bae25e4c15d512"
205            ]
206        });
207        serde_json::from_value(log_json).expect("Failed to deserialize RPC log")
208    }
209
210    #[rstest]
211    fn test_parse_fee_protocol_collect_event_hypersync(hypersync_log: HypersyncLog) {
212        let dex = Arc::clone(&arbitrum::UNISWAP_V3.dex);
213        let event = parse_fee_protocol_collect_event_hypersync(dex, &hypersync_log).unwrap();
214
215        assert_eq!(
216            event.pool_identifier.to_string(),
217            "0xd13040d4fe917EE704158CfCB3338dCd2838B245"
218        );
219        assert_eq!(
220            event.sender.to_string().to_lowercase(),
221            "0xc36442b4a4522e871399cd717abdd847ab11fe88"
222        );
223        assert_eq!(
224            event.recipient.to_string().to_lowercase(),
225            "0xa61da382c18d9d5beb905ea192bae25e4c15d512"
226        );
227        assert_eq!(event.amount0, EXPECTED_AMOUNT0);
228        assert_eq!(event.amount1, EXPECTED_AMOUNT1);
229        assert_eq!(event.block_number, 391_799_940);
230        assert_eq!(event.transaction_index, 5);
231        assert_eq!(event.log_index, 17);
232    }
233
234    #[rstest]
235    fn test_parse_fee_protocol_collect_event_rpc(rpc_log: RpcLog) {
236        let dex = Arc::clone(&arbitrum::UNISWAP_V3.dex);
237        let event = parse_fee_protocol_collect_event_rpc(dex, &rpc_log).unwrap();
238
239        assert_eq!(
240            event.pool_identifier.to_string(),
241            "0xd13040d4fe917EE704158CfCB3338dCd2838B245"
242        );
243        assert_eq!(
244            event.sender.to_string().to_lowercase(),
245            "0xc36442b4a4522e871399cd717abdd847ab11fe88"
246        );
247        assert_eq!(
248            event.recipient.to_string().to_lowercase(),
249            "0xa61da382c18d9d5beb905ea192bae25e4c15d512"
250        );
251        assert_eq!(event.amount0, EXPECTED_AMOUNT0);
252        assert_eq!(event.amount1, EXPECTED_AMOUNT1);
253        assert_eq!(event.block_number, 391_799_940);
254    }
255
256    #[rstest]
257    fn test_hypersync_rpc_match(hypersync_log: HypersyncLog, rpc_log: RpcLog) {
258        let dex = Arc::clone(&arbitrum::UNISWAP_V3.dex);
259        let event_hypersync =
260            parse_fee_protocol_collect_event_hypersync(Arc::clone(&dex), &hypersync_log).unwrap();
261        let event_rpc = parse_fee_protocol_collect_event_rpc(dex, &rpc_log).unwrap();
262
263        assert_eq!(event_hypersync.pool_identifier, event_rpc.pool_identifier);
264        assert_eq!(event_hypersync.sender, event_rpc.sender);
265        assert_eq!(event_hypersync.recipient, event_rpc.recipient);
266        assert_eq!(event_hypersync.amount0, event_rpc.amount0);
267        assert_eq!(event_hypersync.amount1, event_rpc.amount1);
268        assert_eq!(event_hypersync.block_number, event_rpc.block_number);
269        assert_eq!(event_hypersync.transaction_hash, event_rpc.transaction_hash);
270        assert_eq!(event_hypersync.log_index, event_rpc.log_index);
271    }
272}
nautilus_blockchain/exchanges/parsing/uniswap_v3/fee_protocol_collect.rs

nautilus_blockchain/exchanges/parsing/uniswap_v3/
fee_protocol_collect.rs
