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

nautilus_blockchain/cache/
copy.rs
