nautilus_blockchain/cache/
copy.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
16//! PostgreSQL COPY BINARY operations for high-performance bulk data loading.
17//!
18//! This module provides utilities for using PostgreSQL's COPY command with binary format,
19//! which offers significantly better performance than standard INSERT operations for bulk data loading.
20
21use std::fmt::Display;
22
23use nautilus_model::defi::{
24    Block, Pool, PoolLiquidityUpdate, PoolSwap, Token, data::PoolFeeCollect,
25};
26use sqlx::{PgPool, postgres::PgPoolCopyExt};
27
28/// Handles PostgreSQL COPY BINARY operations for blockchain data.
29#[derive(Debug)]
30pub struct PostgresCopyHandler<'a> {
31    pool: &'a PgPool,
32}
33
34impl<'a> PostgresCopyHandler<'a> {
35    /// Creates a new COPY handler with a reference to the database pool.
36    #[must_use]
37    pub const fn new(pool: &'a PgPool) -> Self {
38        Self { pool }
39    }
40
41    /// Inserts blocks using PostgreSQL COPY BINARY for maximum performance.
42    ///
43    /// This method is significantly faster than INSERT for bulk operations as it bypasses
44    /// SQL parsing and uses PostgreSQL's native binary protocol.
45    ///
46    /// # Errors
47    ///
48    /// Returns an error if the COPY operation fails.
49    pub async fn copy_blocks(&self, chain_id: u32, blocks: &[Block]) -> anyhow::Result<()> {
50        if blocks.is_empty() {
51            return Ok(());
52        }
53
54        let copy_statement = "
55            COPY block (
56                chain_id, number, hash, parent_hash, miner, gas_limit, gas_used, timestamp,
57                base_fee_per_gas, blob_gas_used, excess_blob_gas,
58                l1_gas_price, l1_gas_used, l1_fee_scalar
59            ) FROM STDIN WITH (FORMAT BINARY)";
60
61        let mut copy_in = self
62            .pool
63            .copy_in_raw(copy_statement)
64            .await
65            .map_err(|e| anyhow::anyhow!("Failed to start COPY operation: {e}"))?;
66
67        // Write binary header
68        self.write_copy_header(&mut copy_in).await?;
69
70        // Write each block as binary data
71        for block in blocks {
72            self.write_block_binary(&mut copy_in, chain_id, block)
73                .await?;
74        }
75
76        // Write binary trailer
77        self.write_copy_trailer(&mut copy_in).await?;
78
79        // Finish the COPY operation
80        copy_in
81            .finish()
82            .await
83            .map_err(|e| anyhow::anyhow!("Failed to finish COPY operation: {e}"))?;
84
85        Ok(())
86    }
87
88    /// Inserts tokens using PostgreSQL COPY BINARY for maximum performance.
89    ///
90    /// # Errors
91    ///
92    /// Returns an error if the COPY operation fails.
93    pub async fn copy_tokens(&self, chain_id: u32, tokens: &[Token]) -> anyhow::Result<()> {
94        if tokens.is_empty() {
95            return Ok(());
96        }
97
98        let copy_statement = "
99            COPY token (
100                chain_id, address, name, symbol, decimals
101            ) FROM STDIN WITH (FORMAT BINARY)";
102
103        let mut copy_in = self
104            .pool
105            .copy_in_raw(copy_statement)
106            .await
107            .map_err(|e| anyhow::anyhow!("Failed to start COPY operation: {e}"))?;
108
109        self.write_copy_header(&mut copy_in).await?;
110        for token in tokens {
111            self.write_token_binary(&mut copy_in, chain_id, token)
112                .await?;
113        }
114        self.write_copy_trailer(&mut copy_in).await?;
115        copy_in
116            .finish()
117            .await
118            .map_err(|e| anyhow::anyhow!("Failed to finish COPY operation: {e}"))?;
119
120        Ok(())
121    }
122
123    /// Inserts pools using PostgreSQL COPY BINARY for maximum performance.
124    ///
125    /// # Errors
126    ///
127    /// Returns an error if the COPY operation fails.
128    pub async fn copy_pools(&self, chain_id: u32, pools: &[Pool]) -> anyhow::Result<()> {
129        if pools.is_empty() {
130            return Ok(());
131        }
132
133        let copy_statement = "
134            COPY pool (
135                chain_id, dex_name, address, pool_identifier, creation_block,
136                token0_chain, token0_address, token1_chain, token1_address,
137                fee, tick_spacing, initial_tick, initial_sqrt_price_x96, hook_address
138            ) FROM STDIN WITH (FORMAT BINARY)";
139
140        let mut copy_in = self
141            .pool
142            .copy_in_raw(copy_statement)
143            .await
144            .map_err(|e| anyhow::anyhow!("Failed to start COPY operation: {e}"))?;
145
146        self.write_copy_header(&mut copy_in).await?;
147        for pool in pools {
148            self.write_pool_binary(&mut copy_in, chain_id, pool).await?;
149        }
150        self.write_copy_trailer(&mut copy_in).await?;
151        copy_in
152            .finish()
153            .await
154            .map_err(|e| anyhow::anyhow!("Failed to finish COPY operation: {e}"))?;
155
156        Ok(())
157    }
158
159    /// Inserts pool swaps using PostgreSQL COPY BINARY for maximum performance.
160    ///
161    /// # Errors
162    ///
163    /// Returns an error if the COPY operation fails.
164    pub async fn copy_pool_swaps(&self, chain_id: u32, swaps: &[PoolSwap]) -> anyhow::Result<()> {
165        if swaps.is_empty() {
166            return Ok(());
167        }
168
169        let copy_statement = "
170            COPY pool_swap_event (
171                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
172                log_index, sender, recipient, sqrt_price_x96, liquidity, tick, amount0, amount1,
173                order_side, base_quantity, quote_quantity, spot_price, execution_price
174            ) FROM STDIN WITH (FORMAT BINARY)";
175
176        let mut copy_in = self
177            .pool
178            .copy_in_raw(copy_statement)
179            .await
180            .map_err(|e| anyhow::anyhow!("Failed to start COPY operation: {e}"))?;
181
182        // Write binary header
183        self.write_copy_header(&mut copy_in).await?;
184
185        // Write each swap as binary data
186        for swap in swaps {
187            self.write_pool_swap_binary(&mut copy_in, chain_id, swap)
188                .await?;
189        }
190
191        // Write binary trailer
192        self.write_copy_trailer(&mut copy_in).await?;
193
194        // Finish the COPY operation
195        copy_in.finish().await.map_err(|e| {
196            // Emit a single error per failed COPY so one failure is one shutdown-on-error trigger
197            let mut detail = format!(
198                "COPY operation failed for pool_swap batch: chain_id={chain_id}, batch_size={}",
199                swaps.len()
200            );
201
202            if !swaps.is_empty() {
203                detail.push_str(&format!(
204                    ", block_range={} to {}",
205                    swaps.iter().map(|s| s.block).min().unwrap_or(0),
206                    swaps.iter().map(|s| s.block).max().unwrap_or(0)
207                ));
208            }
209
210            for (i, swap) in swaps.iter().take(5).enumerate() {
211                detail.push_str(&format!(
212                    "\n  Swap[{i}]: tx={} log_idx={} block={} pool={}",
213                    swap.transaction_hash, swap.log_index, swap.block, swap.instrument_id
214                ));
215            }
216
217            if swaps.len() > 5 {
218                detail.push_str(&format!("\n  ... and {} more swaps", swaps.len() - 5));
219            }
220
221            log::error!("{detail}");
222
223            anyhow::anyhow!("Failed to finish COPY operation: {e}")
224        })?;
225
226        Ok(())
227    }
228
229    /// Inserts pool liquidity updates using PostgreSQL COPY BINARY for maximum performance.
230    ///
231    /// # Errors
232    ///
233    /// Returns an error if the COPY operation fails.
234    pub async fn copy_pool_liquidity_updates(
235        &self,
236        chain_id: u32,
237        updates: &[PoolLiquidityUpdate],
238    ) -> anyhow::Result<()> {
239        if updates.is_empty() {
240            return Ok(());
241        }
242
243        let copy_statement = "
244            COPY pool_liquidity_event (
245                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
246                log_index, event_type, sender, owner, position_liquidity,
247                amount0, amount1, tick_lower, tick_upper
248            ) FROM STDIN WITH (FORMAT BINARY)";
249
250        let mut copy_in = self
251            .pool
252            .copy_in_raw(copy_statement)
253            .await
254            .map_err(|e| anyhow::anyhow!("Failed to start COPY operation: {e}"))?;
255
256        // Write binary header
257        self.write_copy_header(&mut copy_in).await?;
258
259        // Write each liquidity update as binary data
260        for update in updates {
261            self.write_pool_liquidity_update_binary(&mut copy_in, chain_id, update)
262                .await?;
263        }
264
265        // Write binary trailer
266        self.write_copy_trailer(&mut copy_in).await?;
267
268        // Finish the COPY operation
269        copy_in.finish().await.map_err(|e| {
270            // Emit a single error per failed COPY so one failure is one shutdown-on-error trigger
271            let mut detail = format!(
272                "COPY operation failed for pool_liquidity batch: chain_id={chain_id}, batch_size={}",
273                updates.len()
274            );
275
276            if !updates.is_empty() {
277                detail.push_str(&format!(
278                    ", block_range={} to {}",
279                    updates.iter().map(|u| u.block).min().unwrap_or(0),
280                    updates.iter().map(|u| u.block).max().unwrap_or(0)
281                ));
282            }
283
284            for (i, update) in updates.iter().take(5).enumerate() {
285                detail.push_str(&format!(
286                    "\n  Update[{i}]: tx={} log_idx={} block={} pool={} type={}",
287                    update.transaction_hash,
288                    update.log_index,
289                    update.block,
290                    update.pool_identifier,
291                    update.kind
292                ));
293            }
294
295            if updates.len() > 5 {
296                detail.push_str(&format!("\n  ... and {} more updates", updates.len() - 5));
297            }
298
299            log::error!("{detail}");
300
301            anyhow::anyhow!("Failed to finish COPY operation: {e}")
302        })?;
303
304        Ok(())
305    }
306
307    /// Writes the PostgreSQL COPY binary format header.
308    ///
309    /// The header consists of:
310    /// - 11-byte signature: "PGCOPY\n\xff\r\n\0"
311    /// - 4-byte flags field (all zeros)
312    /// - 4-byte header extension length (all zeros)
313    async fn write_copy_header(
314        &self,
315        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,
316    ) -> anyhow::Result<()> {
317        use std::io::Write;
318        let mut header = Vec::new();
319
320        // PostgreSQL binary copy header
321        header.write_all(b"PGCOPY\n\xff\r\n\0")?; // Signature
322        header.write_all(&[0, 0, 0, 0])?; // Flags field
323        header.write_all(&[0, 0, 0, 0])?; // Header extension length
324
325        copy_in
326            .send(header)
327            .await
328            .map_err(|e| anyhow::anyhow!("Failed to write COPY header: {e}"))?;
329        Ok(())
330    }
331
332    /// Writes a single block in PostgreSQL binary format.
333    ///
334    /// Each row in binary format consists of:
335    /// - 2-byte field count
336    /// - For each field: 4-byte length followed by data (or -1 for NULL)
337    async fn write_block_binary(
338        &self,
339        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,
340        chain_id: u32,
341        block: &Block,
342    ) -> anyhow::Result<()> {
343        use std::io::Write;
344        let mut row_data = Vec::new();
345
346        // Number of fields (14)
347        row_data.write_all(&14u16.to_be_bytes())?;
348
349        let chain_id_bytes = (chain_id as i32).to_be_bytes();
350        row_data.write_all(&(chain_id_bytes.len() as i32).to_be_bytes())?;
351        row_data.write_all(&chain_id_bytes)?;
352
353        let number_bytes = (block.number as i64).to_be_bytes();
354        row_data.write_all(&(number_bytes.len() as i32).to_be_bytes())?;
355        row_data.write_all(&number_bytes)?;
356
357        let hash_bytes = block.hash.as_bytes();
358        row_data.write_all(&(hash_bytes.len() as i32).to_be_bytes())?;
359        row_data.write_all(hash_bytes)?;
360
361        let parent_hash_bytes = block.parent_hash.as_bytes();
362        row_data.write_all(&(parent_hash_bytes.len() as i32).to_be_bytes())?;
363        row_data.write_all(parent_hash_bytes)?;
364
365        let miner_bytes = block.miner.to_string().as_bytes().to_vec();
366        row_data.write_all(&(miner_bytes.len() as i32).to_be_bytes())?;
367        row_data.write_all(&miner_bytes)?;
368
369        let gas_limit_bytes = (block.gas_limit as i64).to_be_bytes();
370        row_data.write_all(&(gas_limit_bytes.len() as i32).to_be_bytes())?;
371        row_data.write_all(&gas_limit_bytes)?;
372
373        let gas_used_bytes = (block.gas_used as i64).to_be_bytes();
374        row_data.write_all(&(gas_used_bytes.len() as i32).to_be_bytes())?;
375        row_data.write_all(&gas_used_bytes)?;
376
377        let timestamp_bytes = block.timestamp.to_string().as_bytes().to_vec();
378        row_data.write_all(&(timestamp_bytes.len() as i32).to_be_bytes())?;
379        row_data.write_all(&timestamp_bytes)?;
380
381        if let Some(ref base_fee) = block.base_fee_per_gas {
382            let base_fee_bytes = base_fee.to_string().as_bytes().to_vec();
383            row_data.write_all(&(base_fee_bytes.len() as i32).to_be_bytes())?;
384            row_data.write_all(&base_fee_bytes)?;
385        } else {
386            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL value
387        }
388
389        if let Some(ref blob_gas) = block.blob_gas_used {
390            let blob_gas_bytes = blob_gas.to_string().as_bytes().to_vec();
391            row_data.write_all(&(blob_gas_bytes.len() as i32).to_be_bytes())?;
392            row_data.write_all(&blob_gas_bytes)?;
393        } else {
394            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL value
395        }
396
397        if let Some(ref excess_blob) = block.excess_blob_gas {
398            let excess_blob_bytes = excess_blob.to_string().as_bytes().to_vec();
399            row_data.write_all(&(excess_blob_bytes.len() as i32).to_be_bytes())?;
400            row_data.write_all(&excess_blob_bytes)?;
401        } else {
402            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL value
403        }
404
405        if let Some(ref l1_gas_price) = block.l1_gas_price {
406            let l1_gas_price_bytes = l1_gas_price.to_string().as_bytes().to_vec();
407            row_data.write_all(&(l1_gas_price_bytes.len() as i32).to_be_bytes())?;
408            row_data.write_all(&l1_gas_price_bytes)?;
409        } else {
410            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL value
411        }
412
413        if let Some(l1_gas_used) = block.l1_gas_used {
414            let l1_gas_used_bytes = (l1_gas_used as i64).to_be_bytes();
415            row_data.write_all(&(l1_gas_used_bytes.len() as i32).to_be_bytes())?;
416            row_data.write_all(&l1_gas_used_bytes)?;
417        } else {
418            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL value
419        }
420
421        if let Some(l1_fee_scalar) = block.l1_fee_scalar {
422            let l1_fee_scalar_bytes = (l1_fee_scalar as i64).to_be_bytes();
423            row_data.write_all(&(l1_fee_scalar_bytes.len() as i32).to_be_bytes())?;
424            row_data.write_all(&l1_fee_scalar_bytes)?;
425        } else {
426            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL value
427        }
428
429        copy_in
430            .send(row_data)
431            .await
432            .map_err(|e| anyhow::anyhow!("Failed to write block data: {e}"))?;
433        Ok(())
434    }
435
436    /// Writes a single pool swap in PostgreSQL binary format.
437    ///
438    /// Each row in binary format consists of:
439    /// - 2-byte field count
440    /// - For each field: 4-byte length followed by data (or -1 for NULL)
441    async fn write_pool_swap_binary(
442        &self,
443        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,
444        chain_id: u32,
445        swap: &PoolSwap,
446    ) -> anyhow::Result<()> {
447        use std::io::Write;
448        let mut row_data = Vec::new();
449
450        row_data.write_all(&19u16.to_be_bytes())?;
451
452        let chain_id_bytes = (chain_id as i32).to_be_bytes();
453        row_data.write_all(&(chain_id_bytes.len() as i32).to_be_bytes())?;
454        row_data.write_all(&chain_id_bytes)?;
455
456        let dex_name_bytes = swap.dex.name.to_string().as_bytes().to_vec();
457        row_data.write_all(&(dex_name_bytes.len() as i32).to_be_bytes())?;
458        row_data.write_all(&dex_name_bytes)?;
459
460        let pool_identifier = swap.instrument_id.to_string();
461        let pool_identifier_bytes = pool_identifier.as_bytes();
462        row_data.write_all(&(pool_identifier_bytes.len() as i32).to_be_bytes())?;
463        row_data.write_all(pool_identifier_bytes)?;
464
465        let block_bytes = (swap.block as i64).to_be_bytes();
466        row_data.write_all(&(block_bytes.len() as i32).to_be_bytes())?;
467        row_data.write_all(&block_bytes)?;
468
469        let tx_hash_bytes = swap.transaction_hash.as_bytes();
470        row_data.write_all(&(tx_hash_bytes.len() as i32).to_be_bytes())?;
471        row_data.write_all(tx_hash_bytes)?;
472
473        let tx_index_bytes = (swap.transaction_index as i32).to_be_bytes();
474        row_data.write_all(&(tx_index_bytes.len() as i32).to_be_bytes())?;
475        row_data.write_all(&tx_index_bytes)?;
476
477        let log_index_bytes = (swap.log_index as i32).to_be_bytes();
478        row_data.write_all(&(log_index_bytes.len() as i32).to_be_bytes())?;
479        row_data.write_all(&log_index_bytes)?;
480
481        let sender_bytes = swap.sender.to_string().as_bytes().to_vec();
482        row_data.write_all(&(sender_bytes.len() as i32).to_be_bytes())?;
483        row_data.write_all(&sender_bytes)?;
484
485        let recipient_bytes = swap.recipient.to_string().as_bytes().to_vec();
486        row_data.write_all(&(recipient_bytes.len() as i32).to_be_bytes())?;
487        row_data.write_all(&recipient_bytes)?;
488
489        write_copy_numeric(&mut row_data, swap.sqrt_price_x96);
490        write_copy_numeric(&mut row_data, swap.liquidity);
491
492        let tick_bytes = swap.tick.to_be_bytes();
493        row_data.write_all(&(tick_bytes.len() as i32).to_be_bytes())?;
494        row_data.write_all(&tick_bytes)?;
495
496        write_copy_numeric(&mut row_data, swap.amount0);
497        write_copy_numeric(&mut row_data, swap.amount1);
498
499        if let Some(trade_info) = &swap.trade_info {
500            let side_bytes = trade_info.order_side.to_string().as_bytes().to_vec();
501            row_data.write_all(&(side_bytes.len() as i32).to_be_bytes())?;
502            row_data.write_all(&side_bytes)?;
503
504            let base_qty_decimal = trade_info.quantity_base.as_decimal();
505            let base_qty_str = base_qty_decimal.to_string();
506            let base_qty_bytes = base_qty_str.as_bytes();
507            row_data.write_all(&(base_qty_bytes.len() as i32).to_be_bytes())?;
508            row_data.write_all(base_qty_bytes)?;
509
510            let quote_qty_decimal = trade_info.quantity_quote.as_decimal();
511            let quote_qty_str = quote_qty_decimal.to_string();
512            let quote_qty_bytes = quote_qty_str.as_bytes();
513            row_data.write_all(&(quote_qty_bytes.len() as i32).to_be_bytes())?;
514            row_data.write_all(quote_qty_bytes)?;
515
516            let spot_price_decimal = trade_info.spot_price.as_decimal();
517            let spot_price_str = spot_price_decimal.to_string();
518            let spot_price_bytes = spot_price_str.as_bytes();
519            row_data.write_all(&(spot_price_bytes.len() as i32).to_be_bytes())?;
520            row_data.write_all(spot_price_bytes)?;
521
522            let exec_price_decimal = trade_info.execution_price.as_decimal();
523            let exec_price_str = exec_price_decimal.to_string();
524            let exec_price_bytes = exec_price_str.as_bytes();
525            row_data.write_all(&(exec_price_bytes.len() as i32).to_be_bytes())?;
526            row_data.write_all(exec_price_bytes)?;
527        } else {
528            row_data.write_all(&(-1i32).to_be_bytes())?;
529            row_data.write_all(&(-1i32).to_be_bytes())?;
530            row_data.write_all(&(-1i32).to_be_bytes())?;
531            row_data.write_all(&(-1i32).to_be_bytes())?;
532            row_data.write_all(&(-1i32).to_be_bytes())?;
533        }
534
535        copy_in
536            .send(row_data)
537            .await
538            .map_err(|e| anyhow::anyhow!("Failed to write pool swap data: {e}"))?;
539        Ok(())
540    }
541
542    /// Writes a single pool liquidity update in PostgreSQL binary format.
543    ///
544    /// Each row in binary format consists of:
545    /// - 2-byte field count
546    /// - For each field: 4-byte length followed by data (or -1 for NULL)
547    async fn write_pool_liquidity_update_binary(
548        &self,
549        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,
550        chain_id: u32,
551        update: &PoolLiquidityUpdate,
552    ) -> anyhow::Result<()> {
553        use std::io::Write;
554        let mut row_data = Vec::new();
555
556        row_data.write_all(&15u16.to_be_bytes())?;
557
558        let chain_id_bytes = (chain_id as i32).to_be_bytes();
559        row_data.write_all(&(chain_id_bytes.len() as i32).to_be_bytes())?;
560        row_data.write_all(&chain_id_bytes)?;
561
562        let dex_name_bytes = update.dex.name.to_string().as_bytes().to_vec();
563        row_data.write_all(&(dex_name_bytes.len() as i32).to_be_bytes())?;
564        row_data.write_all(&dex_name_bytes)?;
565
566        let pool_identifier = update.instrument_id.to_string();
567        let pool_identifier_bytes = pool_identifier.as_bytes();
568        row_data.write_all(&(pool_identifier_bytes.len() as i32).to_be_bytes())?;
569        row_data.write_all(pool_identifier_bytes)?;
570
571        let block_bytes = (update.block as i64).to_be_bytes();
572        row_data.write_all(&(block_bytes.len() as i32).to_be_bytes())?;
573        row_data.write_all(&block_bytes)?;
574
575        let tx_hash_bytes = update.transaction_hash.as_bytes();
576        row_data.write_all(&(tx_hash_bytes.len() as i32).to_be_bytes())?;
577        row_data.write_all(tx_hash_bytes)?;
578
579        let tx_index_bytes = (update.transaction_index as i32).to_be_bytes();
580        row_data.write_all(&(tx_index_bytes.len() as i32).to_be_bytes())?;
581        row_data.write_all(&tx_index_bytes)?;
582
583        let log_index_bytes = (update.log_index as i32).to_be_bytes();
584        row_data.write_all(&(log_index_bytes.len() as i32).to_be_bytes())?;
585        row_data.write_all(&log_index_bytes)?;
586
587        let event_type_bytes = update.kind.to_string().as_bytes().to_vec();
588        row_data.write_all(&(event_type_bytes.len() as i32).to_be_bytes())?;
589        row_data.write_all(&event_type_bytes)?;
590
591        if let Some(sender) = update.sender {
592            let sender_bytes = sender.to_string().as_bytes().to_vec();
593            row_data.write_all(&(sender_bytes.len() as i32).to_be_bytes())?;
594            row_data.write_all(&sender_bytes)?;
595        } else {
596            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL value
597        }
598
599        let owner_bytes = update.owner.to_string().as_bytes().to_vec();
600        row_data.write_all(&(owner_bytes.len() as i32).to_be_bytes())?;
601        row_data.write_all(&owner_bytes)?;
602
603        write_copy_numeric(&mut row_data, update.position_liquidity);
604        write_copy_numeric(&mut row_data, update.amount0);
605        write_copy_numeric(&mut row_data, update.amount1);
606
607        let tick_lower_bytes = update.tick_lower.to_be_bytes();
608        row_data.write_all(&(tick_lower_bytes.len() as i32).to_be_bytes())?;
609        row_data.write_all(&tick_lower_bytes)?;
610
611        let tick_upper_bytes = update.tick_upper.to_be_bytes();
612        row_data.write_all(&(tick_upper_bytes.len() as i32).to_be_bytes())?;
613        row_data.write_all(&tick_upper_bytes)?;
614
615        copy_in
616            .send(row_data)
617            .await
618            .map_err(|e| anyhow::anyhow!("Failed to write pool liquidity update data: {e}"))?;
619        Ok(())
620    }
621
622    /// Inserts pool fee collect events using PostgreSQL COPY BINARY for maximum performance.
623    ///
624    /// # Errors
625    ///
626    /// Returns an error if the COPY operation fails.
627    pub async fn copy_pool_collects(
628        &self,
629        chain_id: u32,
630        collects: &[PoolFeeCollect],
631    ) -> anyhow::Result<()> {
632        if collects.is_empty() {
633            return Ok(());
634        }
635
636        let copy_statement = "
637            COPY pool_collect_event (
638                chain_id, dex_name, pool_identifier, block, transaction_hash, transaction_index,
639                log_index, owner, amount0, amount1, tick_lower, tick_upper
640            ) FROM STDIN WITH (FORMAT BINARY)";
641
642        let mut copy_in = self
643            .pool
644            .copy_in_raw(copy_statement)
645            .await
646            .map_err(|e| anyhow::anyhow!("Failed to start COPY operation: {e}"))?;
647
648        // Write binary header
649        self.write_copy_header(&mut copy_in).await?;
650
651        // Write each collect event as binary data
652        for collect in collects {
653            self.write_pool_fee_collect_binary(&mut copy_in, chain_id, collect)
654                .await?;
655        }
656
657        // Write binary trailer
658        self.write_copy_trailer(&mut copy_in).await?;
659
660        // Finish the COPY operation
661        copy_in.finish().await.map_err(|e| {
662            // Emit a single error per failed COPY so one failure is one shutdown-on-error trigger
663            let mut detail = format!(
664                "COPY operation failed for temp_pool_collect batch: chain_id={chain_id}, batch_size={}",
665                collects.len()
666            );
667
668            if !collects.is_empty() {
669                detail.push_str(&format!(
670                    ", block_range={} to {}",
671                    collects.iter().map(|c| c.block).min().unwrap_or(0),
672                    collects.iter().map(|c| c.block).max().unwrap_or(0)
673                ));
674            }
675
676            for (i, collect) in collects.iter().take(5).enumerate() {
677                detail.push_str(&format!(
678                    "\n  Collect[{i}]: tx={} log_idx={} block={} pool={} owner={}",
679                    collect.transaction_hash,
680                    collect.log_index,
681                    collect.block,
682                    collect.pool_identifier,
683                    collect.owner
684                ));
685            }
686
687            if collects.len() > 5 {
688                detail.push_str(&format!("\n  ... and {} more collects", collects.len() - 5));
689            }
690
691            log::error!("{detail}");
692
693            anyhow::anyhow!("Failed to finish COPY operation: {e}")
694        })?;
695
696        Ok(())
697    }
698
699    /// Writes a single pool fee collect in PostgreSQL binary format.
700    ///
701    /// Each row in binary format consists of:
702    /// - 2-byte field count
703    /// - For each field: 4-byte length followed by data (or -1 for NULL)
704    async fn write_pool_fee_collect_binary(
705        &self,
706        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,
707        chain_id: u32,
708        collect: &PoolFeeCollect,
709    ) -> anyhow::Result<()> {
710        use std::io::Write;
711        let mut row_data = Vec::new();
712
713        row_data.write_all(&12u16.to_be_bytes())?;
714
715        let chain_id_bytes = (chain_id as i32).to_be_bytes();
716        row_data.write_all(&(chain_id_bytes.len() as i32).to_be_bytes())?;
717        row_data.write_all(&chain_id_bytes)?;
718
719        let dex_name_bytes = collect.dex.name.to_string().as_bytes().to_vec();
720        row_data.write_all(&(dex_name_bytes.len() as i32).to_be_bytes())?;
721        row_data.write_all(&dex_name_bytes)?;
722
723        let pool_identifier = collect.instrument_id.to_string();
724        let pool_identifier_bytes = pool_identifier.as_bytes();
725        row_data.write_all(&(pool_identifier_bytes.len() as i32).to_be_bytes())?;
726        row_data.write_all(pool_identifier_bytes)?;
727
728        let block_bytes = (collect.block as i64).to_be_bytes();
729        row_data.write_all(&(block_bytes.len() as i32).to_be_bytes())?;
730        row_data.write_all(&block_bytes)?;
731
732        let tx_hash_bytes = collect.transaction_hash.as_bytes();
733        row_data.write_all(&(tx_hash_bytes.len() as i32).to_be_bytes())?;
734        row_data.write_all(tx_hash_bytes)?;
735
736        let tx_index_bytes = (collect.transaction_index as i32).to_be_bytes();
737        row_data.write_all(&(tx_index_bytes.len() as i32).to_be_bytes())?;
738        row_data.write_all(&tx_index_bytes)?;
739
740        let log_index_bytes = (collect.log_index as i32).to_be_bytes();
741        row_data.write_all(&(log_index_bytes.len() as i32).to_be_bytes())?;
742        row_data.write_all(&log_index_bytes)?;
743
744        let owner_bytes = collect.owner.to_string().as_bytes().to_vec();
745        row_data.write_all(&(owner_bytes.len() as i32).to_be_bytes())?;
746        row_data.write_all(&owner_bytes)?;
747
748        write_copy_numeric(&mut row_data, collect.amount0);
749        write_copy_numeric(&mut row_data, collect.amount1);
750
751        let tick_lower_bytes = collect.tick_lower.to_be_bytes();
752        row_data.write_all(&(tick_lower_bytes.len() as i32).to_be_bytes())?;
753        row_data.write_all(&tick_lower_bytes)?;
754
755        let tick_upper_bytes = collect.tick_upper.to_be_bytes();
756        row_data.write_all(&(tick_upper_bytes.len() as i32).to_be_bytes())?;
757        row_data.write_all(&tick_upper_bytes)?;
758
759        copy_in
760            .send(row_data)
761            .await
762            .map_err(|e| anyhow::anyhow!("Failed to write pool fee collect data: {e}"))?;
763        Ok(())
764    }
765
766    /// Writes a single token in PostgreSQL binary format.
767    ///
768    /// Each row in binary format consists of:
769    /// - 2-byte field count
770    /// - For each field: 4-byte length followed by data (or -1 for NULL)
771    async fn write_token_binary(
772        &self,
773        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,
774        chain_id: u32,
775        token: &Token,
776    ) -> anyhow::Result<()> {
777        use std::io::Write;
778        let mut row_data = Vec::new();
779
780        row_data.write_all(&5u16.to_be_bytes())?;
781
782        let chain_id_bytes = (chain_id as i32).to_be_bytes();
783        row_data.write_all(&(chain_id_bytes.len() as i32).to_be_bytes())?;
784        row_data.write_all(&chain_id_bytes)?;
785
786        let address_bytes = token.address.to_string().as_bytes().to_vec();
787        row_data.write_all(&(address_bytes.len() as i32).to_be_bytes())?;
788        row_data.write_all(&address_bytes)?;
789
790        let name_bytes = token.name.as_bytes();
791        row_data.write_all(&(name_bytes.len() as i32).to_be_bytes())?;
792        row_data.write_all(name_bytes)?;
793
794        let symbol_bytes = token.symbol.as_bytes();
795        row_data.write_all(&(symbol_bytes.len() as i32).to_be_bytes())?;
796        row_data.write_all(symbol_bytes)?;
797
798        let decimals_bytes = (i32::from(token.decimals)).to_be_bytes();
799        row_data.write_all(&(decimals_bytes.len() as i32).to_be_bytes())?;
800        row_data.write_all(&decimals_bytes)?;
801
802        copy_in
803            .send(row_data)
804            .await
805            .map_err(|e| anyhow::anyhow!("Failed to write token data: {e}"))?;
806        Ok(())
807    }
808
809    /// Writes a single pool in PostgreSQL binary format.
810    ///
811    /// Each row in binary format consists of:
812    /// - 2-byte field count
813    /// - For each field: 4-byte length followed by data (or -1 for NULL)
814    async fn write_pool_binary(
815        &self,
816        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,
817        chain_id: u32,
818        pool: &Pool,
819    ) -> anyhow::Result<()> {
820        use std::io::Write;
821        let mut row_data = Vec::new();
822
823        row_data.write_all(&14u16.to_be_bytes())?;
824
825        let chain_id_bytes = (chain_id as i32).to_be_bytes();
826        row_data.write_all(&(chain_id_bytes.len() as i32).to_be_bytes())?;
827        row_data.write_all(&chain_id_bytes)?;
828
829        let dex_name_bytes = pool.dex.name.to_string().as_bytes().to_vec();
830        row_data.write_all(&(dex_name_bytes.len() as i32).to_be_bytes())?;
831        row_data.write_all(&dex_name_bytes)?;
832
833        let address_bytes = pool.address.to_string().as_bytes().to_vec();
834        row_data.write_all(&(address_bytes.len() as i32).to_be_bytes())?;
835        row_data.write_all(&address_bytes)?;
836
837        let pool_identifier_bytes = pool.pool_identifier.as_str().as_bytes();
838        row_data.write_all(&(pool_identifier_bytes.len() as i32).to_be_bytes())?;
839        row_data.write_all(pool_identifier_bytes)?;
840
841        let creation_block_bytes = (pool.creation_block as i64).to_be_bytes();
842        row_data.write_all(&(creation_block_bytes.len() as i32).to_be_bytes())?;
843        row_data.write_all(&creation_block_bytes)?;
844
845        let token0_chain_bytes = (pool.token0.chain.chain_id as i32).to_be_bytes();
846        row_data.write_all(&(token0_chain_bytes.len() as i32).to_be_bytes())?;
847        row_data.write_all(&token0_chain_bytes)?;
848
849        let token0_address_bytes = pool.token0.address.to_string().as_bytes().to_vec();
850        row_data.write_all(&(token0_address_bytes.len() as i32).to_be_bytes())?;
851        row_data.write_all(&token0_address_bytes)?;
852
853        let token1_chain_bytes = (pool.token1.chain.chain_id as i32).to_be_bytes();
854        row_data.write_all(&(token1_chain_bytes.len() as i32).to_be_bytes())?;
855        row_data.write_all(&token1_chain_bytes)?;
856
857        let token1_address_bytes = pool.token1.address.to_string().as_bytes().to_vec();
858        row_data.write_all(&(token1_address_bytes.len() as i32).to_be_bytes())?;
859        row_data.write_all(&token1_address_bytes)?;
860
861        if let Some(fee) = pool.fee {
862            let fee_bytes = (fee as i32).to_be_bytes();
863            row_data.write_all(&(fee_bytes.len() as i32).to_be_bytes())?;
864            row_data.write_all(&fee_bytes)?;
865        } else {
866            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL
867        }
868
869        if let Some(tick_spacing) = pool.tick_spacing {
870            let tick_spacing_bytes = (tick_spacing as i32).to_be_bytes();
871            row_data.write_all(&(tick_spacing_bytes.len() as i32).to_be_bytes())?;
872            row_data.write_all(&tick_spacing_bytes)?;
873        } else {
874            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL
875        }
876
877        if let Some(initial_tick) = pool.initial_tick {
878            let initial_tick_bytes = initial_tick.to_be_bytes();
879            row_data.write_all(&(initial_tick_bytes.len() as i32).to_be_bytes())?;
880            row_data.write_all(&initial_tick_bytes)?;
881        } else {
882            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL
883        }
884
885        if let Some(ref initial_sqrt_price) = pool.initial_sqrt_price_x96 {
886            write_copy_numeric(&mut row_data, initial_sqrt_price);
887        } else {
888            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL
889        }
890
891        if let Some(ref hooks) = pool.hooks {
892            let hooks_bytes = hooks.to_string().as_bytes().to_vec();
893            row_data.write_all(&(hooks_bytes.len() as i32).to_be_bytes())?;
894            row_data.write_all(&hooks_bytes)?;
895        } else {
896            row_data.write_all(&(-1i32).to_be_bytes())?; // NULL
897        }
898
899        copy_in
900            .send(row_data)
901            .await
902            .map_err(|e| anyhow::anyhow!("Failed to write pool data: {e}"))?;
903        Ok(())
904    }
905
906    /// Writes the PostgreSQL COPY binary format trailer.
907    ///
908    /// The trailer is a 2-byte value of -1 to indicate end of data.
909    async fn write_copy_trailer(
910        &self,
911        copy_in: &mut sqlx::postgres::PgCopyIn<sqlx::pool::PoolConnection<sqlx::Postgres>>,
912    ) -> anyhow::Result<()> {
913        // Binary trailer: -1 as i16 to indicate end of data
914        let trailer = (-1i16).to_be_bytes();
915        copy_in
916            .send(trailer.to_vec())
917            .await
918            .map_err(|e| anyhow::anyhow!("Failed to write COPY trailer: {e}"))?;
919        Ok(())
920    }
921}
922
923fn write_copy_numeric(row: &mut Vec<u8>, value: impl Display) {
924    let value = value.to_string();
925    row.extend_from_slice(&(value.len() as i32).to_be_bytes());
926    row.extend_from_slice(value.as_bytes());
927}
nautilus_blockchain/cache/copy.rs

nautilus_blockchain/cache/
copy.rs
