solwatch 0.1.18

//! Chain access layer. `SolClient` holds one reqwest client + the endpoints and
//! exposes typed Solana JSON-RPC (jsonParsed account reads, largest token
//! accounts, signature history, parsed transactions) plus the DexScreener and
//! Jupiter HTTP APIs. Pure reqwest — no solana-sdk — so it builds into a small
//! static musl binary that drops into the Rokha sandbox.

pub mod dexscreener;
pub mod jupiter;

use crate::config;
use anyhow::{anyhow, Result};
use serde::Deserialize;
use serde_json::{json, Value};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

static RPC_ID: AtomicU64 = AtomicU64::new(0);

pub struct SolClient {
    pub http: reqwest::Client,
    pub rpc: String,
    pub jupiter: String,
    /// Global RPC concurrency gate — public endpoints throttle bursts hard,
    /// and throttle-induced data loss must never be silent.
    limiter: tokio::sync::Semaphore,
    /// Count of RPC calls that hit a throttle (429 / in-band rate limit) at
    /// least once. Surfaced in audit warnings + cluster coverage.
    pub throttled: AtomicU64,
    /// Count of RPC calls that FAILED after all retries.
    pub failed: AtomicU64,
    /// Wall-clock start + budget. Once `started.elapsed() > deadline`, `rpc()`
    /// short-circuits so every pass winds down to a labeled partial instead of
    /// being killed mid-flight by the sandbox timeout. Zero budget = unbounded.
    started: Instant,
    deadline: Duration,
    /// Count of RPC calls skipped because the wall-clock budget was spent.
    pub past_deadline: AtomicU64,
}

#[derive(Debug, Clone, Deserialize)]
pub struct SigInfo {
    pub signature: String,
    pub slot: u64,
    #[serde(rename = "blockTime")]
    pub block_time: Option<i64>,
    pub err: Option<Value>,
}

/// A parsed mint account: authorities + supply + (token-2022) extensions.
#[derive(Debug, Clone, Default)]
pub struct MintInfo {
    pub program: String,
    pub decimals: u32,
    pub supply: u128,
    pub supply_raw: String,
    pub mint_authority: Option<String>,
    pub freeze_authority: Option<String>,
    pub extensions: Vec<Value>,
}

#[derive(Debug, Clone)]
pub struct LargeAccount {
    pub token_account: String,
    pub amount: u128,
}

#[derive(Debug, Clone, Default)]
pub struct MetaplexMeta {
    pub name: String,
    pub symbol: String,
    pub uri: String,
    pub update_authority: String,
    pub is_mutable: bool,
}

impl Default for SolClient {
    fn default() -> Self {
        Self::new()
    }
}

impl SolClient {
    pub fn new() -> Self {
        let http = reqwest::Client::builder()
            .timeout(Duration::from_secs(25))
            .user_agent("solwatch/0.1")
            .build()
            .expect("http client");
        Self {
            http,
            rpc: config::rpc_url(),
            jupiter: config::jupiter_base(),
            limiter: tokio::sync::Semaphore::new(config::rpc_concurrency()),
            throttled: AtomicU64::new(0),
            failed: AtomicU64::new(0),
            started: Instant::now(),
            deadline: Duration::from_millis(config::deadline_ms()),
            past_deadline: AtomicU64::new(0),
        }
    }

    /// True once the wall-clock budget is spent (never true when disabled).
    pub fn past_deadline(&self) -> bool {
        !self.deadline.is_zero() && self.started.elapsed() > self.deadline
    }

    /// Human summary of throttle/failure pressure this run. None when clean
    /// OR when every throttled call recovered — a retry that succeeded lost
    /// nothing, so it must not read as a coverage warning (observed
    /// 2026-07-15: "20 throttled, 0 failed" rendered under the scary
    /// "checks did NOT run" banner when in fact everything ran).
    pub fn rpc_pressure(&self) -> Option<String> {
        let t = self.throttled.load(Ordering::Relaxed);
        let f = self.failed.load(Ordering::Relaxed);
        let d = self.past_deadline.load(Ordering::Relaxed);
        if d > 0 {
            // The audit ran out of wall-clock and stopped early — the dominant
            // story is "partial", not "throttled". Say so plainly.
            return Some(format!(
                "Audit hit its time budget and stopped early — {d} read(s) were skipped, so this scan is partial. Point SOLWATCH_RPC_URL at a dedicated RPC (or set HELIUS_API_KEY) for a faster, complete scan."
            ));
        }
        if f == 0 {
            return None;
        }
        Some(format!(
            "RPC throttling hit {t} call(s) and {f} failed even after retries — those reads are missing"
        ))
    }

    /// Raw JSON-RPC with a GLOBAL concurrency gate + exponential backoff on
    /// transient errors (429s are common on the public mainnet endpoint). A
    /// deterministic JSON-RPC error is returned as Err without retrying.
    pub async fn rpc(&self, method: &str, params: Value) -> Result<Value> {
        // Wall-clock guard: once the budget is spent, refuse new work up front so
        // the caller degrades to a partial instead of the sandbox killing us.
        if self.past_deadline() {
            self.past_deadline.fetch_add(1, Ordering::Relaxed);
            return Err(anyhow!("{method}: skipped (audit time budget spent)"));
        }
        let _permit = self.limiter.acquire().await;
        let id = RPC_ID.fetch_add(1, Ordering::Relaxed);
        let body = json!({"jsonrpc":"2.0","id":id,"method":method,"params":params});
        let mut last = anyhow!("rpc {method} failed");
        let mut saw_throttle = false;
        for attempt in 0..5u32 {
            if attempt > 0 {
                // Don't burn the remaining budget sleeping to retry a throttled
                // call — give up now so other passes still get a turn.
                if self.past_deadline() {
                    self.past_deadline.fetch_add(1, Ordering::Relaxed);
                    return Err(anyhow!(
                        "{method}: gave up after throttle (time budget spent)"
                    ));
                }
                // Exponential: 400ms, 800ms, 1.6s, 3.2s (+ light jitter).
                let base = 400u64 << (attempt - 1);
                let jitter = id.wrapping_mul(2654435761) % 250;
                tokio::time::sleep(Duration::from_millis(base + jitter)).await;
            }
            match self.http.post(&self.rpc).json(&body).send().await {
                Ok(resp) => {
                    let status = resp.status();
                    if status.as_u16() == 429 || status.is_server_error() {
                        saw_throttle = true;
                        last = anyhow!("{method} HTTP {status}");
                        continue;
                    }
                    let v: Value = match resp.json().await {
                        Ok(v) => v,
                        Err(e) => {
                            last = anyhow!("{method} decode: {e}");
                            continue;
                        }
                    };
                    if let Some(err) = v.get("error") {
                        let code = err.get("code").and_then(|c| c.as_i64()).unwrap_or(0);
                        let msg = err
                            .get("message")
                            .and_then(|m| m.as_str())
                            .unwrap_or("error")
                            .to_string();
                        // The public endpoint throttles some calls with an
                        // in-band 429 error object — retry those too.
                        if code == 429 || msg.contains("Too many requests") {
                            saw_throttle = true;
                            last = anyhow!("{method}: {msg}");
                            continue;
                        }
                        if saw_throttle {
                            self.throttled.fetch_add(1, Ordering::Relaxed);
                        }
                        return Err(anyhow!("{method}: {msg}"));
                    }
                    if saw_throttle {
                        self.throttled.fetch_add(1, Ordering::Relaxed);
                    }
                    return Ok(v.get("result").cloned().unwrap_or(Value::Null));
                }
                Err(e) => last = anyhow!("{method}: {e}"),
            }
        }
        if saw_throttle {
            self.throttled.fetch_add(1, Ordering::Relaxed);
        }
        self.failed.fetch_add(1, Ordering::Relaxed);
        Err(last)
    }

    pub async fn get_slot(&self) -> Result<u64> {
        let r = self.rpc("getSlot", json!([])).await?;
        r.as_u64().ok_or_else(|| anyhow!("getSlot: bad result"))
    }

    /// jsonParsed account info; Null result value means the account does not exist.
    pub async fn account_info(&self, address: &str) -> Result<Value> {
        let r = self
            .rpc(
                "getAccountInfo",
                json!([address, {"encoding": "jsonParsed"}]),
            )
            .await?;
        Ok(r.get("value").cloned().unwrap_or(Value::Null))
    }

    /// Parse a mint account (spl-token or token-2022, incl. extensions).
    pub async fn mint_info(&self, mint: &str) -> Result<Option<MintInfo>> {
        let v = self.account_info(mint).await?;
        if v.is_null() {
            return Ok(None);
        }
        let program = v
            .get("owner")
            .and_then(|o| o.as_str())
            .unwrap_or_default()
            .to_string();
        let parsed = v.pointer("/data/parsed").cloned().unwrap_or(Value::Null);
        if parsed.pointer("/type").and_then(|t| t.as_str()) != Some("mint") {
            return Err(anyhow!("{mint} is not a token mint account"));
        }
        let info = parsed.get("info").cloned().unwrap_or(Value::Null);
        let supply_raw = info
            .get("supply")
            .and_then(|s| s.as_str())
            .unwrap_or("0")
            .to_string();
        Ok(Some(MintInfo {
            program,
            decimals: info.get("decimals").and_then(|d| d.as_u64()).unwrap_or(0) as u32,
            supply: supply_raw.parse().unwrap_or(0),
            supply_raw,
            mint_authority: info
                .get("mintAuthority")
                .and_then(|a| a.as_str())
                .map(String::from),
            freeze_authority: info
                .get("freezeAuthority")
                .and_then(|a| a.as_str())
                .map(String::from),
            extensions: info
                .get("extensions")
                .and_then(|e| e.as_array())
                .cloned()
                .unwrap_or_default(),
        }))
    }

    /// Top-20 largest token accounts for a mint. This call has its own, much
    /// stricter throttle bucket on the public endpoint, so it gets a patient
    /// outer retry on top of the standard one.
    pub async fn largest_accounts(&self, mint: &str) -> Result<Vec<LargeAccount>> {
        let mut r = self.rpc("getTokenLargestAccounts", json!([mint])).await;
        if r.is_err() {
            // One patient extra round — some endpoints throttle per-method.
            // (The default public RPC hard-blocks this method entirely; the
            // caller falls back to Jupiter's indexed number in that case.)
            tokio::time::sleep(Duration::from_millis(2500)).await;
            r = self.rpc("getTokenLargestAccounts", json!([mint])).await;
        }
        let r = r?;
        let list = r
            .get("value")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();
        Ok(list
            .iter()
            .filter_map(|e| {
                Some(LargeAccount {
                    token_account: e.get("address")?.as_str()?.to_string(),
                    amount: e.get("amount")?.as_str()?.parse().ok()?,
                })
            })
            .collect())
    }

    /// getMultipleAccounts, base64 — raw account data bytes aligned to input
    /// order (None for a missing account or one with no data). Used to read
    /// pool accounts and scan them for their reserve-vault token-account keys,
    /// so an AMM/launchpad reserve is never mistaken for a whale.
    pub async fn accounts_raw(&self, addrs: &[String]) -> Result<Vec<Option<Vec<u8>>>> {
        if addrs.is_empty() {
            return Ok(vec![]);
        }
        use base64::Engine;
        let r = self
            .rpc(
                "getMultipleAccounts",
                json!([addrs, {"encoding": "base64"}]),
            )
            .await?;
        Ok(r.get("value")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default()
            .iter()
            .map(|a| {
                a.pointer("/data/0")
                    .and_then(|d| d.as_str())
                    .and_then(|s| base64::engine::general_purpose::STANDARD.decode(s).ok())
            })
            .collect())
    }

    /// getMultipleAccounts, jsonParsed. Result aligns with the input order
    /// (missing accounts are Null).
    pub async fn multiple_accounts(&self, addrs: &[String]) -> Result<Vec<Value>> {
        if addrs.is_empty() {
            return Ok(vec![]);
        }
        let r = self
            .rpc(
                "getMultipleAccounts",
                json!([addrs, {"encoding": "jsonParsed"}]),
            )
            .await?;
        Ok(r.get("value")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default())
    }

    pub async fn signatures(
        &self,
        address: &str,
        limit: usize,
        before: Option<&str>,
    ) -> Result<Vec<SigInfo>> {
        let mut opts = json!({"limit": limit});
        if let Some(b) = before {
            opts["before"] = json!(b);
        }
        let r = self
            .rpc("getSignaturesForAddress", json!([address, opts]))
            .await?;
        Ok(serde_json::from_value(r).unwrap_or_default())
    }

    /// Full parsed transaction (v0-compatible).
    pub async fn transaction(&self, signature: &str) -> Result<Value> {
        self.rpc(
            "getTransaction",
            json!([signature, {"encoding": "jsonParsed", "maxSupportedTransactionVersion": 0}]),
        )
        .await
    }

    /// Sum of an owner's balance for one mint, in raw base units.
    pub async fn owner_mint_balance(&self, owner: &str, mint: &str) -> Result<u128> {
        let r = self
            .rpc(
                "getTokenAccountsByOwner",
                json!([owner, {"mint": mint}, {"encoding": "jsonParsed"}]),
            )
            .await?;
        let list = r
            .get("value")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();
        Ok(list
            .iter()
            .filter_map(|a| {
                a.pointer("/account/data/parsed/info/tokenAmount/amount")?
                    .as_str()?
                    .parse::<u128>()
                    .ok()
            })
            .sum())
    }

    /// Walk an address's signature history back to the OLDEST page (bounded).
    /// Returns `(oldest_page, truncated)` — `truncated=true` means the walk hit
    /// `max_pages` and the returned page is NOT proven to be the account's
    /// beginning. Pages are newest-first; the last element of the returned page
    /// is the oldest signature seen.
    pub async fn oldest_signatures(
        &self,
        address: &str,
        max_pages: usize,
    ) -> Result<(Vec<SigInfo>, bool)> {
        let mut before: Option<String> = None;
        let mut oldest_page: Vec<SigInfo> = vec![];
        let mut pages = 0usize;
        loop {
            let page = self.signatures(address, 1000, before.as_deref()).await?;
            pages += 1;
            let n = page.len();
            if n > 0 {
                before = Some(page[n - 1].signature.clone());
                oldest_page = page;
            }
            if n < 1000 {
                return Ok((oldest_page, false));
            }
            if pages >= max_pages {
                return Ok((oldest_page, true));
            }
        }
    }

    /// The wallet's FIRST inbound SOL funding: walks to its oldest signature,
    /// fetches that transaction, and parses the system-transfer source that
    /// paid the wallet. Returns `(funder, wallet_total_sig_floor)` — the floor
    /// is the observed signature count (>=1000 means "busy account", a hub
    /// signal). None funder when no inbound transfer is identifiable.
    pub async fn first_funding(&self, wallet: &str) -> Result<(Option<String>, usize)> {
        let (page, truncated) = self.oldest_signatures(wallet, 3).await?;
        let count_floor = if truncated { 3000 } else { page.len() };
        if truncated {
            // >=3k signatures: too busy to attribute a "first funder" — and a
            // wallet this active is not a fresh sybil anyway.
            return Ok((None, count_floor));
        }
        let Some(first) = page.last() else {
            return Ok((None, 0));
        };
        let tx = self.transaction(&first.signature).await?;
        Ok((first_sol_source(&tx, wallet), count_floor))
    }

    /// Jito bundle-proof lookup for one transaction signature. Returns the
    /// bundle_id when the tx landed in a Jito bundle, None when it didn't or
    /// the (third-party) API is unavailable — the caller falls back to the tip
    /// heuristic; this must never be fatal.
    pub async fn jito_bundle_id(&self, signature: &str) -> Option<String> {
        let url = format!("{}/{}", config::JITO_BUNDLES_API, signature);
        let resp = self
            .http
            .get(&url)
            .timeout(Duration::from_secs(8))
            .send()
            .await
            .ok()?;
        if !resp.status().is_success() {
            return None;
        }
        let v: Value = resp.json().await.ok()?;
        v.pointer("/0/bundle_id")
            .and_then(|b| b.as_str())
            .map(String::from)
    }

    /// Fetch + parse the Metaplex metadata account for a mint (classic SPL
    /// tokens; token-2022 metadata rides the mint's extensions instead).
    pub async fn metaplex_metadata(&self, mint: &str) -> Result<Option<MetaplexMeta>> {
        let Some(pda) = metadata_pda(mint) else {
            return Ok(None);
        };
        let r = self
            .rpc("getAccountInfo", json!([pda, {"encoding": "base64"}]))
            .await?;
        let Some(data_b64) = r.pointer("/value/data/0").and_then(|d| d.as_str()) else {
            return Ok(None);
        };
        use base64::Engine;
        let bytes = base64::engine::general_purpose::STANDARD
            .decode(data_b64)
            .unwrap_or_default();
        Ok(parse_metaplex(&bytes))
    }
}

/// Borsh-walk the Metaplex Metadata account. Returns None on any malformed
/// layout — the RPC is env-overridable, so bad data must not panic us.
pub fn parse_metaplex(b: &[u8]) -> Option<MetaplexMeta> {
    let mut i = 0usize;
    let take = |i: &mut usize, n: usize| -> Option<&[u8]> {
        let s = b.get(*i..*i + n)?;
        *i += n;
        Some(s)
    };
    let _key = take(&mut i, 1)?;
    let update_authority = bs58::encode(take(&mut i, 32)?).into_string();
    let _mint = take(&mut i, 32)?;
    let string = |i: &mut usize| -> Option<String> {
        let len = u32::from_le_bytes(take(i, 4)?.try_into().ok()?) as usize;
        if len > 4096 {
            return None;
        }
        let raw = take(i, len)?;
        Some(
            String::from_utf8_lossy(raw)
                .trim_end_matches('\0')
                .to_string(),
        )
    };
    let name = string(&mut i)?;
    let symbol = string(&mut i)?;
    let uri = string(&mut i)?;
    let _sfbp = take(&mut i, 2)?;
    let has_creators = take(&mut i, 1)?[0];
    if has_creators == 1 {
        let n = u32::from_le_bytes(take(&mut i, 4)?.try_into().ok()?) as usize;
        if n > 16 {
            return None;
        }
        take(&mut i, n * 34)?;
    }
    let _primary_sale = take(&mut i, 1)?;
    let is_mutable = take(&mut i, 1)?[0] == 1;
    Some(MetaplexMeta {
        name,
        symbol,
        uri,
        update_authority,
        is_mutable,
    })
}

/// Parse the inbound SOL system-transfer source that paid `wallet` in one
/// jsonParsed transaction (top-level + inner instructions). Covers `transfer`,
/// `transferWithSeed` and `createAccount` — the three ways a fresh wallet gets
/// its first lamports.
pub fn first_sol_source(tx: &Value, wallet: &str) -> Option<String> {
    let check = |ins: &Value| -> Option<String> {
        let parsed = ins.get("parsed")?;
        let typ = parsed.get("type").and_then(|t| t.as_str())?;
        let info = parsed.get("info")?;
        let dest_key = match typ {
            "transfer" | "transferWithSeed" => "destination",
            "createAccount" | "createAccountWithSeed" => "newAccount",
            _ => return None,
        };
        if info.get(dest_key).and_then(|d| d.as_str()) != Some(wallet) {
            return None;
        }
        info.get("source")
            .and_then(|s| s.as_str())
            .filter(|s| *s != wallet)
            .map(String::from)
    };
    if let Some(list) = tx
        .pointer("/transaction/message/instructions")
        .and_then(|v| v.as_array())
    {
        for ins in list {
            if let Some(src) = check(ins) {
                return Some(src);
            }
        }
    }
    if let Some(groups) = tx
        .pointer("/meta/innerInstructions")
        .and_then(|v| v.as_array())
    {
        for g in groups {
            if let Some(list) = g.get("instructions").and_then(|v| v.as_array()) {
                for ins in list {
                    if let Some(src) = check(ins) {
                        return Some(src);
                    }
                }
            }
        }
    }
    None
}

/// Generic find_program_address — pure sha256 + off-curve check, no
/// solana-sdk. Seeds are raw byte slices; program is a base58 program id.
pub fn find_pda(seeds: &[&[u8]], program: &str) -> Option<String> {
    use sha2::{Digest, Sha256};
    let program_b = bs58::decode(program).into_vec().ok()?;
    if program_b.len() != 32 {
        return None;
    }
    for bump in (0u8..=255).rev() {
        let mut h = Sha256::new();
        for s in seeds {
            h.update(s);
        }
        h.update([bump]);
        h.update(&program_b);
        h.update(b"ProgramDerivedAddress");
        let out: [u8; 32] = h.finalize().into();
        if !is_on_curve(&out) {
            return Some(bs58::encode(out).into_string());
        }
    }
    None
}

/// find_program_address(["metadata", metadata_program, mint], metadata_program)
pub fn metadata_pda(mint: &str) -> Option<String> {
    let program = bs58::decode(config::METAPLEX_METADATA).into_vec().ok()?;
    let mint_b = bs58::decode(mint).into_vec().ok()?;
    if mint_b.len() != 32 {
        return None;
    }
    find_pda(&[b"metadata", &program, &mint_b], config::METAPLEX_METADATA)
}

/// The pump.fun bonding-curve PDA for a mint:
/// find_program_address(["bonding-curve", mint], PUMPFUN_PROGRAM).
/// Derived OFFLINE — the curve's ~793M initial allocation must be excluded
/// from buyer/holder math even when no API has indexed the pair yet.
pub fn pumpfun_bonding_curve_pda(mint: &str) -> Option<String> {
    let mint_b = bs58::decode(mint).into_vec().ok()?;
    if mint_b.len() != 32 {
        return None;
    }
    find_pda(&[b"bonding-curve", &mint_b], config::PUMPFUN_PROGRAM)
}

/// The associated token account of `owner` for `mint` under `token_program`:
/// find_program_address([owner, token_program, mint], ATA_PROGRAM).
pub fn ata_pda(owner: &str, mint: &str, token_program: &str) -> Option<String> {
    let owner_b = bs58::decode(owner).into_vec().ok()?;
    let mint_b = bs58::decode(mint).into_vec().ok()?;
    let prog_b = bs58::decode(token_program).into_vec().ok()?;
    if owner_b.len() != 32 || mint_b.len() != 32 || prog_b.len() != 32 {
        return None;
    }
    find_pda(&[&owner_b, &prog_b, &mint_b], config::ASSOCIATED_TOKEN)
}

fn is_on_curve(bytes: &[u8; 32]) -> bool {
    curve25519_dalek::edwards::CompressedEdwardsY(*bytes)
        .decompress()
        .is_some()
}

/// Loose base58 pubkey check (32 bytes decoded).
pub fn is_pubkey(s: &str) -> bool {
    (32..=44).contains(&s.len())
        && bs58::decode(s)
            .into_vec()
            .map(|v| v.len() == 32)
            .unwrap_or(false)
}