hello i need your assistance i need to setup a webhook for helius and i need to integrade it into this script here use anyhow::{anyhow, Result};
use bs58;
use config::Config;
use futures_util::{SinkExt, StreamExt};
use reqwest::Client;
use serde::{Deserialize, Serialize};
use serde_json::json;
use solana_client::rpc_client::RpcClient;
use solana_sdk::signer::keypair::Keypair;
use solana_sdk::signer::Signer;
use solana_sdk::transaction::Transaction;
use solana_sdk::hash::Hash;
use std::collections::HashMap;
use std::fs::{File, OpenOptions};
use std::io::Write;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::Mutex;
use tokio::time::interval;
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message as WsMessage};
use tracing::{debug, error, info, warn};
use tracing_subscriber::{self, fmt::format::FmtSpan, EnvFilter};
use tracing_appender;
use ctrlc;

const CONFIG_PATH: &str = "C:\\solana_rust_bot\\config.toml";
const WSOL_MINT: &str = "So11111111111111111111111111111111111111112";
const RAYDIUM_AMM_PROGRAM_ID: &str = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8";
const MAX_BUY_ATTEMPTS: u32 = 3;
const BUY_ATTEMPT_DELAY: Duration = Duration::from_millis(200);
const STATE_FILE: &str = "bot_state.json";
const MIN_LIQUIDITY: f64 = 10000000000.0;
const MAX_LIQUIDITY: f64 = 45000000000.0;
const MAX_ACTIVE_TRADES: usize = 5;

#[derive(Clone, Serialize, Deserialize)]
struct Trade {
    token_address: String,
    amount: u64,
    buy_price: f64,
    take_profit_price: f64,
    stop_loss_price: f64,
    timestamp: u64,
}

#[derive(Serialize, Deserialize)]
struct BotState {
    active_trades: HashMap<String, Trade>,
    last_save_time: SystemTime,
}

struct PriceCache {
    prices: HashMap<String, (f64, Instant)>,
    ttl: Duration,
}

impl PriceCache {
    fn new(ttl: u64) -> Self {
        Self {
            prices: HashMap::new(),
            ttl: Duration::from_secs(ttl),
        }
    }

    fn get(&self, token_address: &str) -> Option<f64> {
        self.prices.get(token_address).and_then(|(price, timestamp)| {
            if timestamp.elapsed() < self.ttl {
                Some(*price)
            } else {
                None
            }
        })
    }

    fn set(&mut self, token_address: String, price: f64) {
        self.prices.insert(token_address.clone(), (price, Instant::now()));
        debug!("Price for {} set to {}.", token_address, price);
    }
}

struct Bot {
    config: Config,
    rpc_client: RpcClient,
    keypair: Keypair,
    price_cache: Arc<Mutex<PriceCache>>,
    active_trades: Arc<Mutex<HashMap<String, Trade>>>,
    last_blockhash: Arc<Mutex<(Hash, Instant)>>,
    metadata_limiter: Arc<Mutex<()>>,
    swap_limiter: Arc<Mutex<()>>,
    http_client: Client,
    is_tracking_prices: Arc<AtomicBool>,
    shutdown_signal: Arc<AtomicBool>,
}

impl Bot {
    async fn new() -> Result<Self> {
        let config = Config::builder()
            .add_source(config::File::with_name(CONFIG_PATH))
            .build()?;
        
        info!("Loading configuration from {}", CONFIG_PATH);

        let rpc_url = config.get_string("rpc_url")?;
        let rpc_client = RpcClient::new(rpc_url.clone());
        info!("Connected to RPC URL: {}", rpc_url);

        let keypair = Self::load_keypair(&config)?;
        info!("Loaded keypair: {:?}", keypair.pubkey());

        let mut bot = Self {
            config,
            rpc_client,
            keypair,
            price_cache: Arc::new(Mutex::new(PriceCache::new(60))),
            active_trades: Arc::new(Mutex::new(HashMap::new())),
            last_blockhash: Arc::new(Mutex::new((Hash::default(), Instant::now()))),
            metadata_limiter: Arc::new(Mutex::new(())),
            swap_limiter: Arc::new(Mutex::new(())),
            http_client: Client::new(),
            is_tracking_prices: Arc::new(AtomicBool::new(false)),
            shutdown_signal: Arc::new(AtomicBool::new(false)),
        };

        bot.load_state().await?;
        info!("Bot initialized successfully.");
        Ok(bot)
    }

    fn load_keypair(config: &Config) -> Result<Keypair> {
        if let Ok(keypair_58) = config.get_string("keypair_58") {
            let bytes = bs58::decode(keypair_58)
                .into_vec()
                .map_err(|e| anyhow!("Base58 decode error: {}", e))?;
            Keypair::from_bytes(&bytes)
                .map_err(|e| anyhow!("Invalid keypair bytes: {}", e))
        } else if let Ok(keypair_json_path) = config.get_string("keypair_json") {
            let json_content = std::fs::read_to_string(keypair_json_path)?;
            let secret_key: Vec<u8> = serde_json::from_str(&json_content)?;
            Keypair::from_bytes(&secret_key)
                .map_err(|e| anyhow!("Invalid JSON keypair bytes: {}", e))
        } else if let Ok(keypair_64) = config.get_string("keypair_64") {
            let bytes: Vec<u8> = keypair_64
                .split(',')
                .map(|s| s.trim().parse::<u8>().map_err(|e| anyhow!("Failed to parse byte '{}': {}", s.trim(), e)))
                .collect::<Result<Vec<u8>>>()?;
            let bytes_array: [u8; 64] = bytes[..].try_into().map_err(|_| {
                anyhow!("Keypair must be exactly 64 bytes long, but got {}", bytes.len())
            })?;
            Keypair::from_bytes(&bytes_array).map_err(|e| anyhow!("Invalid 64-byte keypair: {}", e))
        } else {
            Err(anyhow!("No valid keypair found in config"))
        }
    }

    async fn fetch_and_cache_token_price(&self, token_address: &str) -> Result<f64> {
        if let Some(price) = self.price_cache.lock().await.get(token_address) {
            return Ok(price);
        }

        let _guard = self.metadata_limiter.lock().await;
        let price = self.fetch_token_price(token_address).await?;
        self.price_cache.lock().await.set(token_address.to_string(), price);

        Ok(price)
    }

    async fn fetch_token_price(&self, token_address: &str) -> Result<f64> {
        let metadata = self.fetch_metadata(token_address).await?;
        metadata
            .get("price")
            .and_then(serde_json::Value::as_f64)
            .ok_or_else(|| anyhow!("No price found for token {}", token_address))
    }

    async fn fetch_metadata(&self, token_address: &str) -> Result<serde_json::Value> {
        let url = format!(
            "https://api.helius.xyz/v0/token-metadata?api-key={}",
            self.config.get_string("helius_api_key")?
        );

        let response = self
            .http_client
            .post(&url)
            .json(&json!({
                "mintAccounts": [token_address],
                "includeOffChain": true,
                "disableCache": false
            }))
            .send()
            .await?
            .json::<Vec<serde_json::Value>>()
            .await?;

        response
            .into_iter()
            .next()
            .ok_or_else(|| anyhow!("No metadata found for token {}", token_address))
    }

    pub async fn run(&mut self, shutdown_signal: Arc<AtomicBool>) -> Result<()> {
        self.load_state().await?;
        let mut save_state_interval = interval(Duration::from_secs(60));
        let mut status_report_interval = interval(Duration::from_secs(300));

        info!("Entering main bot loop. Listening for events...");

        while !shutdown_signal.load(Ordering::SeqCst) {
            tokio::select! {
                result = self.connect_and_listen() => {
                    if let Err(ref e) = result {
                        error!("WebSocket connection error: {}", e);
                    }
                },
                _ = save_state_interval.tick() => {
                    if let Err(ref e) = self.save_state().await {
                        error!("Failed to save bot state: {}", e);
                    }
                },
                _ = status_report_interval.tick() => {
                    if let Err(ref e) = self.print_status_report().await {
                        error!("Failed to print status report: {}", e);
                    }
                },
                _ = tokio::time::sleep(Duration::from_millis(500)), if self.is_tracking_prices.load(Ordering::SeqCst) => {
                    if let Err(ref e) = self.check_active_trades().await {
                        error!("Error checking active trades: {}", e);
                    }
                    
                    if self.active_trades.lock().await.is_empty() {
                        self.is_tracking_prices.store(false, Ordering::SeqCst);
                        info!("All trades completed. Stopping price tracking.");
                    }
                }
            }
        }

        info!("Shutdown signal received. Performing graceful shutdown...");
        self.save_state().await?;
        info!("Bot state saved. Shutdown complete.");
        Ok(())
    }

    async fn connect_and_listen(&self) -> Result<()> {
        info!("Establishing WebSocket connection...");
        let (ws_stream, _) = connect_async(&self.config.get_string("helius_websocket_url")?).await?;
        let (mut write, mut read) = ws_stream.split();

        write
            .send(WsMessage::Text(
                json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "method": "subscribePoolCreations",
                    "params": []
                })
                .to_string(),
            ))
            .await?;

        info!("Connected to Helius WebSocket. Subscribed to pool creations.");

        while let Some(msg) = read.next().await {
            if self.shutdown_signal.load(Ordering::SeqCst) {
                break;
            }

            match msg {
                Ok(WsMessage::Text(text)) => {
                    if let Err(ref e) = self.handle_event(&text).await {
                        error!("Error handling event: {}", e);
                    }
                }
                Ok(WsMessage::Close(_)) => {
                    info!("WebSocket connection closed by server");
                    break;
                }
                Err(ref e) => {
                    error!("WebSocket error: {}", e);
                    break;
                }
                _ => {}
            }
        }

        Ok(())
    }

    async fn handle_event(&self, text: &str) -> Result<()> {
        debug!("Received WebSocket event: {}", text);
        let event: serde_json::Value = serde_json::from_str(text)?;
        if let Some(params) = event.get("params") {
            if let Some(token_address) = params.get("tokenAddress").and_then(serde_json::Value::as_str) {
                info!("Token {} detected from WebSocket event.", token_address);
                if self.is_valid_pool(token_address).await? {
                    self.attempt_buy(token_address).await?;
                } else {
                    info!("Token {} failed validation checks. Skipping.", token_address);
                }
            }
        }
        Ok(())
    }

    async fn is_valid_pool(&self, token_address: &str) -> Result<bool> {
        debug!("Validating pool for token: {}", token_address);
        let metadata = self.fetch_metadata(token_address).await?;

        let is_raydium_pool = metadata
            .get("updateAuthority")
            .and_then(serde_json::Value::as_str)
            .map_or(false, |auth| auth == RAYDIUM_AMM_PROGRAM_ID);

        if !is_raydium_pool {
            return Ok(false);
        }

        let is_spl_token = metadata.get("tokenStandard").and_then(serde_json::Value::as_str) == Some("SPL");

        if !is_spl_token {
            return Ok(false);
        }

        let liquidity = metadata
            .get("liquidity")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0);
        if liquidity < MIN_LIQUIDITY || liquidity > MAX_LIQUIDITY {
            return Ok(false);
        }

        Ok(true)
    }

    async fn attempt_buy(&self, token_address: &str) -> Result<()> {
        let active_trade_count = self.active_trades.lock().await.len();
        if active_trade_count >= MAX_ACTIVE_TRADES {
            info!("Maximum number of active trades ({}) reached. Skipping buy for {}.", MAX_ACTIVE_TRADES, token_address);
            return Ok(());
        }

        self.is_tracking_prices.store(true, Ordering::SeqCst);
        info!("Attempting to buy token: {}", token_address);

        for attempt in 0..MAX_BUY_ATTEMPTS {
            if let Err(ref e) = self.execute_buy(token_address).await {
                warn!("Failed to buy token {}. Attempt {} of {}. Error: {}", token_address, attempt + 1, MAX_BUY_ATTEMPTS, e);
                tokio::time::sleep(BUY_ATTEMPT_DELAY).await;
            } else {
                info!("Successfully bought token {} on attempt {}.", token_address, attempt + 1);
                return Ok(());
            }
        }

        error!("Failed to buy token {} after {} attempts.", token_address, MAX_BUY_ATTEMPTS);

        if self.active_trades.lock().await.is_empty() {
            self.is_tracking_prices.store(false, Ordering::SeqCst);
        }

        Ok(())
    }

    async fn execute_buy(&self, token_address: &str) -> Result<()> {
        let _guard = self.swap_limiter.lock().await;
        let amount = self.config.get_int("snipe_amount")? as u64;
        let slippage = self.config.get_float("max_slippage")?;

        let swap_response = self.http_client
            .post("https://api.raydium.io/v2/main/swap")
            .json(&json!({
                "inToken": WSOL_MINT,
                "outToken": token_address,
                "amount": amount,
                "fixedSide": "in",
                "slippage": slippage
            }))
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let transaction_data = base64::decode(swap_response.get("swapTransaction").ok_or_else(|| anyhow!("Missing swap transaction in response"))?.as_str().ok_or_else(|| anyhow!("Swap transaction is not a string"))?)?;
        let transaction: Transaction = bincode::deserialize(&transaction_data)?;

        let message = transaction.message.clone();
        let mut new_transaction = Transaction::new_unsigned(message);
        let recent_blockhash = self.get_recent_blockhash().await?;
        new_transaction.sign(&[&self.keypair], recent_blockhash);

        let signature = self.rpc_client.send_and_confirm_transaction_with_spinner(&new_transaction)?;
        info!("Buy transaction sent for token {}. Signature: {}", token_address, signature);

        let buy_price = self.fetch_and_cache_token_price(token_address).await?;
        let take_profit_price = buy_price * (1.0 + self.config.get_float("profit_target_percent")? / 100.0);
        let stop_loss_price = buy_price * (1.0 - self.config.get_float("stop_loss_percent")? / 100.0);

        let trade = Trade {
            token_address: token_address.to_string(),
            amount,
            buy_price,
            take_profit_price,
            stop_loss_price,
            timestamp: SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs(),
        };

        self.active_trades.lock().await.insert(token_address.to_string(), trade);
        self.save_state().await?;

        info!("Trade added for token {}. Buy price: {}, Take profit: {}, Stop loss: {}", 
              token_address, buy_price, take_profit_price, stop_loss_price);

        Ok(())
    }

    async fn check_active_trades(&self) -> Result<()> {
        let mut trades_to_remove = Vec::new();
        let mut active_trades = self.active_trades.lock().await;

        for (token_address, trade) in active_trades.iter() {
            let current_price = self.fetch_and_cache_token_price(token_address).await?;
            let profit_loss = (current_price - trade.buy_price) / trade.buy_price * 100.0;
            
            info!("Trade status for {}: Current price: {}, Buy price: {}, P/L: {:.2}%", 
                  token_address, current_price, trade.buy_price, profit_loss);

            if current_price >= trade.take_profit_price || current_price <= trade.stop_loss_price {
                info!("Executing sell for {} (Take profit: {}, Stop loss: {})", 
                      token_address, trade.take_profit_price, trade.stop_loss_price);
                if self.execute_sell(token_address, trade.amount).await.is_ok() {
                    trades_to_remove.push(token_address.clone());
                    info!("Successfully sold token {}", token_address);
                }
            }
        }

        for token_address in trades_to_remove {
            active_trades.remove(&token_address);
        }

        self.save_state().await?;
        Ok(())
    }

    async fn execute_sell(&self, token_address: &str, amount: u64) -> Result<()> {
        let _guard = self.swap_limiter.lock().await;
        let slippage = self.config.get_float("max_slippage")?;

        let swap_response = self.http_client
            .post("https://api.raydium.io/v2/main/swap")
            .json(&json!({
                "inToken": token_address,
                "outToken": WSOL_MINT,
                "amount": amount,
                "fixedSide": "in",
                "slippage": slippage
            }))
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        let transaction_data = base64::decode(swap_response.get("swapTransaction").ok_or_else(|| anyhow!("Missing swap transaction in response"))?.as_str().ok_or_else(|| anyhow!("Swap transaction is not a string"))?)?;
        let transaction: Transaction = bincode::deserialize(&transaction_data)?;

        let message = transaction.message.clone();
        let mut new_transaction = Transaction::new_unsigned(message);
        let recent_blockhash = self.get_recent_blockhash().await?;
        new_transaction.sign(&[&self.keypair], recent_blockhash);

        let signature = self.rpc_client.send_and_confirm_transaction_with_spinner(&new_transaction)?;
        info!("Sell transaction sent for token {}. Signature: {}", token_address, signature);

        Ok(())
    }

    async fn get_recent_blockhash(&self) -> Result<Hash> {
        let mut last_blockhash = self.last_blockhash.lock().await;
        if last_blockhash.1.elapsed() > Duration::from_secs(120) {
            *last_blockhash = (
                self.rpc_client.get_latest_blockhash()?,
                Instant::now(),
            );
        }
        Ok(last_blockhash.0)
    }

    async fn save_state(&self) -> Result<()> {
        let active_trades = self.active_trades.lock().await;
        let state = BotState {
            active_trades: active_trades.clone(),
            last_save_time: SystemTime::now(),
        };
        let serialized = serde_json::to_string(&state)?;
        let mut file = OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .open(STATE_FILE)?;
        file.write_all(serialized.as_bytes())?;
        info!("Bot state saved successfully.");
        Ok(())
    }

    async fn load_state(&mut self) -> Result<()> {
        match File::open(STATE_FILE) {
            Ok(file) => {
                if let Ok(state) = serde_json::from_reader::<_, BotState>(file) {
                    *self.active_trades.lock().await = state.active_trades;
                    info!("State loaded from {}", STATE_FILE);
                } else {
                    warn!("Failed to parse state file: {}. Starting with empty state.", STATE_FILE);
                }
            },
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                info!("No previous state file found. Starting with a clean state.");
            },
            Err(e) => {
                warn!("Error opening state file: {}. Starting with empty state.", e);
            }
        }

        self.verify_loaded_trades().await?;
        Ok(())
    }

    async fn verify_loaded_trades(&self) -> Result<()> {
        let mut trades_to_remove = Vec::new();
        let mut active_trades = self.active_trades.lock().await;

        for (token_address, _) in active_trades.iter() {
            if self.fetch_and_cache_token_price(token_address).await.is_err() {
                trades_to_remove.push(token_address.clone());
                warn!("Removing unverifiable trade for {}", token_address);
            }
        }

        for token_address in trades_to_remove {
            active_trades.remove(&token_address);
        }

        Ok(())
    }

    async fn print_status_report(&self) -> Result<()> {
        let active_trades = self.active_trades.lock().await;
        let total_trades = active_trades.len();
        let mut total_investment = 0.0;
        let mut total_current_value = 0.0;

        for (token_address, trade) in active_trades.iter() {
            let current_price = self.fetch_and_cache_token_price(token_address).await?;
            total_investment += trade.amount as f64 * trade.buy_price;
            total_current_value += trade.amount as f64 * current_price;
        }

        let overall_profit_loss = if total_investment > 0.0 {
            (total_current_value - total_investment) / total_investment * 100.0
        } else {
            0.0
        };

        info!("=== Bot Status Report ===");
        info!("Active Trades: {}", total_trades);
        info!("Total Investment: {:.2} SOL", total_investment);
        info!("Current Portfolio Value: {:.2} SOL", total_current_value);
        info!("Overall P&L: {:.2}%", overall_profit_loss);
        info!("========================");

        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    setup_logging()?;
    info!("🚀 Solana Sniper Bot v1.0 - Launching! 🚀");

    let shutdown_signal = Arc::new(AtomicBool::new(false));
    let mut bot = Bot::new().await?;
    
    let shutdown_signal_clone = shutdown_signal.clone();
    ctrlc::set_handler(move || {
        info!("Received shutdown signal. Exiting...");
        shutdown_signal_clone.store(true, Ordering::SeqCst);
    })?;

    bot.run(shutdown_signal).await?;
    Ok(())
}

fn setup_logging() -> Result<()> {
    let file_appender = tracing_appender::rolling::daily("log", "bot.log");
    let (non_blocking, _guard) = tracing_appender::non_blocking(file_appender);

    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::try_from_default_env().or_else(
                // Specify anyhow::Error as the error type
                |_| Ok::<EnvFilter, anyhow::Error>(EnvFilter::new("info"))
            )?
        )
        .with_writer(non_blocking)
        .with_span_events(FmtSpan::CLOSE)
        .init();

    info!("Logging initialized.");
    Ok(())
}