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        helpers::{
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::helpers as rpc_helpers,
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_helpers::validate_event_signature(
115        log,
116        FEE_PROTOCOL_COLLECT_EVENT_SIGNATURE_HASH,
117        "CollectProtocol",
118    )?;
119
120    let sender = rpc_helpers::extract_address_from_topic(log, 1, "sender")?;
121    let recipient = rpc_helpers::extract_address_from_topic(log, 2, "recipient")?;
122
123    let data_bytes = rpc_helpers::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_helpers::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_helpers::extract_block_number(log)?,
141        rpc_helpers::extract_transaction_hash(log)?,
142        rpc_helpers::extract_transaction_index(log)?,
143        rpc_helpers::extract_log_index(log)?,
144        sender,
145        recipient,
146        decoded.amount0,
147        decoded.amount1,
148    ))
149}
150
151#[cfg(test)]
152mod tests {
153    use rstest::*;
154    use serde_json::json;
155
156    use super::*;
157    use crate::exchanges::arbitrum;
158
159    // CollectProtocol log at block 0x175a6484 (391799940), pool
160    // 0xd13040d4fe917EE704158CfCB3338dCd2838B245.
161    // sender:    0xc36442b4a4522e871399cd717abdd847ab11fe88
162    // recipient: 0xa61da382c18d9d5beb905ea192bae25e4c15d512
163    // Asymmetric amounts (amount0 != amount1) catch a token0/token1 column swap.
164    const EXPECTED_AMOUNT0: u128 = 0x1234_5678_90ab_cdef;
165    const EXPECTED_AMOUNT1: u128 = 0x0fed_cba9_8765_4321;
166    const DATA: &str = "0x0000000000000000000000000000000000000000000000001234567890abcdef0000000000000000000000000000000000000000000000000fedcba987654321";
167
168    #[fixture]
169    fn hypersync_log() -> HypersyncLog {
170        let log_json = json!({
171            "removed": null,
172            "log_index": "0x11",
173            "transaction_index": "0x5",
174            "transaction_hash": "0x0c70f6d6bcf8508ba620b9d1250c95ad67108e35707c5d7456349ea207051bae",
175            "block_hash": null,
176            "block_number": "0x175a6484",
177            "address": "0xd13040d4fe917EE704158CfCB3338dCd2838B245",
178            "data": DATA,
179            "topics": [
180                "0x596b573906218d3411850b26a6b437d6c4522fdb43d2d2386263f86d50b8b151",
181                "0x000000000000000000000000c36442b4a4522e871399cd717abdd847ab11fe88",
182                "0x000000000000000000000000a61da382c18d9d5beb905ea192bae25e4c15d512"
183            ]
184        });
185        serde_json::from_value(log_json).expect("Failed to deserialize HyperSync log")
186    }
187
188    #[fixture]
189    fn rpc_log() -> RpcLog {
190        let log_json = json!({
191            "removed": false,
192            "logIndex": "0x11",
193            "transactionIndex": "0x5",
194            "transactionHash": "0x0c70f6d6bcf8508ba620b9d1250c95ad67108e35707c5d7456349ea207051bae",
195            "blockHash": "0xe925eaa1f5178ceedfa24043a974edb928ddab7195600b6b99ff5403fbf13c8b",
196            "blockNumber": "0x175a6484",
197            "address": "0xd13040d4fe917EE704158CfCB3338dCd2838B245",
198            "data": DATA,
199            "topics": [
200                "0x596b573906218d3411850b26a6b437d6c4522fdb43d2d2386263f86d50b8b151",
201                "0x000000000000000000000000c36442b4a4522e871399cd717abdd847ab11fe88",
202                "0x000000000000000000000000a61da382c18d9d5beb905ea192bae25e4c15d512"
203            ]
204        });
205        serde_json::from_value(log_json).expect("Failed to deserialize RPC log")
206    }
207
208    #[rstest]
209    fn test_parse_fee_protocol_collect_event_hypersync(hypersync_log: HypersyncLog) {
210        let dex = arbitrum::UNISWAP_V3.dex.clone();
211        let event = parse_fee_protocol_collect_event_hypersync(dex, &hypersync_log).unwrap();
212
213        assert_eq!(
214            event.pool_identifier.to_string(),
215            "0xd13040d4fe917EE704158CfCB3338dCd2838B245"
216        );
217        assert_eq!(
218            event.sender.to_string().to_lowercase(),
219            "0xc36442b4a4522e871399cd717abdd847ab11fe88"
220        );
221        assert_eq!(
222            event.recipient.to_string().to_lowercase(),
223            "0xa61da382c18d9d5beb905ea192bae25e4c15d512"
224        );
225        assert_eq!(event.amount0, EXPECTED_AMOUNT0);
226        assert_eq!(event.amount1, EXPECTED_AMOUNT1);
227        assert_eq!(event.block_number, 391_799_940);
228        assert_eq!(event.transaction_index, 5);
229        assert_eq!(event.log_index, 17);
230    }
231
232    #[rstest]
233    fn test_parse_fee_protocol_collect_event_rpc(rpc_log: RpcLog) {
234        let dex = arbitrum::UNISWAP_V3.dex.clone();
235        let event = parse_fee_protocol_collect_event_rpc(dex, &rpc_log).unwrap();
236
237        assert_eq!(
238            event.pool_identifier.to_string(),
239            "0xd13040d4fe917EE704158CfCB3338dCd2838B245"
240        );
241        assert_eq!(
242            event.sender.to_string().to_lowercase(),
243            "0xc36442b4a4522e871399cd717abdd847ab11fe88"
244        );
245        assert_eq!(
246            event.recipient.to_string().to_lowercase(),
247            "0xa61da382c18d9d5beb905ea192bae25e4c15d512"
248        );
249        assert_eq!(event.amount0, EXPECTED_AMOUNT0);
250        assert_eq!(event.amount1, EXPECTED_AMOUNT1);
251        assert_eq!(event.block_number, 391_799_940);
252    }
253
254    #[rstest]
255    fn test_hypersync_rpc_match(hypersync_log: HypersyncLog, rpc_log: RpcLog) {
256        let dex = arbitrum::UNISWAP_V3.dex.clone();
257        let event_hypersync =
258            parse_fee_protocol_collect_event_hypersync(dex.clone(), &hypersync_log).unwrap();
259        let event_rpc = parse_fee_protocol_collect_event_rpc(dex, &rpc_log).unwrap();
260
261        assert_eq!(event_hypersync.pool_identifier, event_rpc.pool_identifier);
262        assert_eq!(event_hypersync.sender, event_rpc.sender);
263        assert_eq!(event_hypersync.recipient, event_rpc.recipient);
264        assert_eq!(event_hypersync.amount0, event_rpc.amount0);
265        assert_eq!(event_hypersync.amount1, event_rpc.amount1);
266        assert_eq!(event_hypersync.block_number, event_rpc.block_number);
267        assert_eq!(event_hypersync.transaction_hash, event_rpc.transaction_hash);
268        assert_eq!(event_hypersync.log_index, event_rpc.log_index);
269    }
270}
nautilus_blockchain/exchanges/parsing/uniswap_v3/fee_protocol_collect.rs

nautilus_blockchain/exchanges/parsing/uniswap_v3/
fee_protocol_collect.rs
