diff --git a/.env.example b/.env.example index b90ebc4..17ceee8 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,8 @@ # Database +# Local file (redb driver): DATABASE_URL=sqlite:wallet.db +# Or PostgreSQL (postgres driver): +# DATABASE_URL=postgres://user:password@localhost:5432/evmhot # Blockchain Provider (choose one) # For HTTP: diff --git a/.gitignore b/.gitignore index dd37d4a..4f095d6 100644 --- a/.gitignore +++ b/.gitignore @@ -6,9 +6,7 @@ # Database files *.db *.db-shm -*.db-wal - -# Docker +*.db-wal# Docker docker-compose.override.yml # Backups diff --git a/Cargo.lock b/Cargo.lock index 97d81d0..cf5ecbb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -779,7 +779,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "81953c529336010edd6d8e358f886d9581267795c61b19475b71314bffa46d35" dependencies = [ "concurrent-queue", - "event-listener", + "event-listener 2.5.3", "futures-core", ] @@ -827,6 +827,15 @@ dependencies = [ "rustc_version 0.4.1", ] +[[package]] +name = "atoi" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528" +dependencies = [ + "num-traits", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -1225,6 +1234,30 @@ dependencies = [ "libc", ] +[[package]] +name = "crc" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" +dependencies = [ + "crc-catalog", +] + +[[package]] +name = "crc-catalog" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" + +[[package]] +name = "crossbeam-queue" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "803d13fb3b09d88be9f4dbc29062c66b19bf7170867ceb746d2a8689bf6c7a26" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -1442,6 +1475,9 @@ name = "either" version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" +dependencies = [ + "serde", +] [[package]] name = "elliptic-curve" @@ -1507,12 +1543,34 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "etcetera" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "136d1b5283a1ab77bd9257427ffd09d8667ced0570b6f938942bc7568ed5b943" +dependencies = [ + "cfg-if", + "home", + "windows-sys 0.48.0", +] + [[package]] name = "event-listener" version = "2.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" +[[package]] +name = "event-listener" +version = "5.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab" +dependencies = [ + "concurrent-queue", + "parking", + "pin-project-lite", +] + [[package]] name = "evm_hot_wallet" version = "0.1.0" @@ -1529,6 +1587,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sqlx", "tempfile", "thiserror 1.0.69", "tokio", @@ -1686,6 +1745,17 @@ dependencies = [ "futures-util", ] +[[package]] +name = "futures-intrusive" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f" +dependencies = [ + "futures-core", + "lock_api", + "parking_lot", +] + [[package]] name = "futures-io" version = "0.3.31" @@ -1885,6 +1955,15 @@ version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +[[package]] +name = "hashlink" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" +dependencies = [ + "hashbrown 0.15.5", +] + [[package]] name = "heck" version = "0.5.0" @@ -1915,6 +1994,15 @@ version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6fe2267d4ed49bc07b63801559be28c718ea06c4738b7a03c94df7386d2cde46" +[[package]] +name = "hkdf" +version = "0.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7" +dependencies = [ + "hmac", +] + [[package]] name = "hmac" version = "0.12.1" @@ -1924,6 +2012,15 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "home" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "http" version = "0.2.12" @@ -2395,6 +2492,18 @@ version = "0.2.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9fbbcab51052fe104eb5e5d351cf728d30a5be1fe14d9be8a3b097481fb97de" +[[package]] +name = "libredox" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c943259e342f1e06ff2da7a83eabdfe7f92ce10262688dbf1895ff0b3e6e4652" +dependencies = [ + "bitflags", + "libc", + "plain", + "redox_syscall 0.9.0", +] + [[package]] name = "linux-raw-sys" version = "0.11.0" @@ -2446,6 +2555,16 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" +[[package]] +name = "md-5" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf" +dependencies = [ + "cfg-if", + "digest 0.10.7", +] + [[package]] name = "memchr" version = "2.7.6" @@ -2663,7 +2782,7 @@ checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" dependencies = [ "cfg-if", "libc", - "redox_syscall", + "redox_syscall 0.5.18", "smallvec", "windows-link", ] @@ -2768,6 +2887,12 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" +[[package]] +name = "plain" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" + [[package]] name = "potential_utf" version = "0.1.4" @@ -3024,6 +3149,15 @@ dependencies = [ "bitflags", ] +[[package]] +name = "redox_syscall" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c5102a6aaa05aa011a238e178e6bca86d2cb56fc9f586d37cb80f5bca6e07759" +dependencies = [ + "bitflags", +] + [[package]] name = "regex" version = "1.12.2" @@ -3529,6 +3663,9 @@ name = "smallvec" version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +dependencies = [ + "serde", +] [[package]] name = "socket2" @@ -3560,6 +3697,126 @@ dependencies = [ "der", ] +[[package]] +name = "sqlx" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fefb893899429669dcdd979aff487bd78f4064e5e7907e4269081e0ef7d97dc" +dependencies = [ + "sqlx-core", + "sqlx-macros", + "sqlx-postgres", +] + +[[package]] +name = "sqlx-core" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6" +dependencies = [ + "base64 0.22.1", + "bytes", + "crc", + "crossbeam-queue", + "either", + "event-listener 5.4.1", + "futures-core", + "futures-intrusive", + "futures-io", + "futures-util", + "hashbrown 0.15.5", + "hashlink", + "indexmap", + "log", + "memchr", + "once_cell", + "percent-encoding", + "rustls", + "serde", + "serde_json", + "sha2", + "smallvec", + "thiserror 2.0.17", + "tokio", + "tokio-stream", + "tracing", + "url", + "webpki-roots 0.26.11", +] + +[[package]] +name = "sqlx-macros" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2d452988ccaacfbf5e0bdbc348fb91d7c8af5bee192173ac3636b5fb6e6715d" +dependencies = [ + "proc-macro2", + "quote", + "sqlx-core", + "sqlx-macros-core", + "syn 2.0.111", +] + +[[package]] +name = "sqlx-macros-core" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b" +dependencies = [ + "dotenvy", + "either", + "heck", + "hex", + "once_cell", + "proc-macro2", + "quote", + "serde", + "serde_json", + "sha2", + "sqlx-core", + "sqlx-postgres", + "syn 2.0.111", + "tokio", + "url", +] + +[[package]] +name = "sqlx-postgres" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46" +dependencies = [ + "atoi", + "base64 0.22.1", + "bitflags", + "byteorder", + "crc", + "dotenvy", + "etcetera", + "futures-channel", + "futures-core", + "futures-util", + "hex", + "hkdf", + "hmac", + "home", + "itoa", + "log", + "md-5", + "memchr", + "once_cell", + "rand 0.8.5", + "serde", + "serde_json", + "sha2", + "smallvec", + "sqlx-core", + "stringprep", + "thiserror 2.0.17", + "tracing", + "whoami", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -3572,6 +3829,17 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "stringprep" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1" +dependencies = [ + "unicode-bidi", + "unicode-normalization", + "unicode-properties", +] + [[package]] name = "strum" version = "0.27.2" @@ -4127,6 +4395,12 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94" +[[package]] +name = "unicode-bidi" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" + [[package]] name = "unicode-ident" version = "1.0.22" @@ -4142,6 +4416,12 @@ dependencies = [ "tinyvec", ] +[[package]] +name = "unicode-properties" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" + [[package]] name = "unicode-xid" version = "0.2.6" @@ -4241,6 +4521,12 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "wasite" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b" + [[package]] name = "wasm-bindgen" version = "0.2.105" @@ -4327,6 +4613,16 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "whoami" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d4a4db5077702ca3015d3d02d74974948aba2ad9e12ab7df718ee64ccd7e97d" +dependencies = [ + "libredox", + "wasite", +] + [[package]] name = "widestring" version = "1.2.1" @@ -4368,6 +4664,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" +dependencies = [ + "windows-targets 0.48.5", +] + [[package]] name = "windows-sys" version = "0.52.0" @@ -4395,6 +4700,21 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c" +dependencies = [ + "windows_aarch64_gnullvm 0.48.5", + "windows_aarch64_msvc 0.48.5", + "windows_i686_gnu 0.48.5", + "windows_i686_msvc 0.48.5", + "windows_x86_64_gnu 0.48.5", + "windows_x86_64_gnullvm 0.48.5", + "windows_x86_64_msvc 0.48.5", +] + [[package]] name = "windows-targets" version = "0.52.6" @@ -4428,6 +4748,12 @@ dependencies = [ "windows_x86_64_msvc 0.53.1", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8" + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" @@ -4440,6 +4766,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" +[[package]] +name = "windows_aarch64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" @@ -4452,6 +4784,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" +[[package]] +name = "windows_i686_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e" + [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -4476,6 +4814,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" +[[package]] +name = "windows_i686_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406" + [[package]] name = "windows_i686_msvc" version = "0.52.6" @@ -4488,6 +4832,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" +[[package]] +name = "windows_x86_64_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" @@ -4500,6 +4850,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc" + [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" @@ -4512,6 +4868,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" +[[package]] +name = "windows_x86_64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index e8cefc6..a541136 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,7 @@ hex = "0.4" thiserror = "1.0" futures = "0.3.31" async-trait = "0.1.89" +sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "tls-rustls", "postgres"] } [dev-dependencies] tempfile = "3.23.0" diff --git a/README.md b/README.md index 6cff406..b63391c 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,7 @@ The service is configured via environment variables. Create a `.env` file or set | Variable | Description | Default | |----------|-------------|---------| -| `DATABASE_URL` | Path to the database file | `sqlite:wallet.db` | +| `DATABASE_URL` | Storage backend: a `postgres://` / `postgresql://` URL uses the PostgreSQL driver; any other value is treated as a local database file path (redb driver). Tables are created automatically on startup with either driver. | `sqlite:wallet.db` | | `PORT` | API server port | `3000` | | `POLL_INTERVAL` | Block polling interval in seconds (HTTP mode only) | `10` | | `BLOCK_OFFSET_FROM_HEAD` | Number of blocks to stay behind chain head for confirmation safety | `20` | diff --git a/src/api.rs b/src/api.rs index 95e96f1..a78f50a 100644 --- a/src/api.rs +++ b/src/api.rs @@ -10,7 +10,18 @@ use evm_hot_wallet::{ HotWalletService, RegisterRequest, RegisterResponse, VerifyTransferRequest, VerifyTransferResponse, }; +use serde::{Deserialize, Serialize}; use std::sync::Arc; + +#[derive(Deserialize)] +pub struct SetBlockNumberRequest { + pub block_number: u64, +} + +#[derive(Serialize)] +pub struct BlockNumberResponse { + pub block_number: u64, +} use tokio::net::TcpListener; #[derive(Clone)] @@ -33,6 +44,8 @@ where .route("/health", get(health::)) .route("/register", post(register::)) .route("/verify_transfer", post(verify_transfer::)) + .route("/block_number", get(get_block_number::)) + .route("/block_number", post(set_block_number::)) .with_state(state); let addr = format!("0.0.0.0:{}", port); @@ -80,6 +93,37 @@ where } } +async fn get_block_number( + State(state): State>, +) -> Result, ApiError> +where + T: Transport + Clone + Send + Sync + 'static, +{ + let block_number = state + .service + .get_block_number() + .await + .map_err(|e| ApiError::Internal(e.to_string()))?; + Ok(Json(BlockNumberResponse { block_number })) +} + +async fn set_block_number( + State(state): State>, + Json(payload): Json, +) -> Result, ApiError> +where + T: Transport + Clone + Send + Sync + 'static, +{ + state + .service + .set_block_number(payload.block_number) + .await + .map_err(|e| ApiError::Internal(e.to_string()))?; + Ok(Json(BlockNumberResponse { + block_number: payload.block_number, + })) +} + // Error handling for the API #[derive(Debug)] enum ApiError { diff --git a/src/config.rs b/src/config.rs index e522d4a..63c2db6 100644 --- a/src/config.rs +++ b/src/config.rs @@ -22,6 +22,8 @@ pub struct Config { pub block_offset_from_head: u64, pub get_logs_max_retries: u32, pub get_logs_delay_ms: u64, + /// Optional JWT token for webhook authorization + pub webhook_jwt_token: Option, } impl Config { @@ -60,6 +62,7 @@ impl Config { let get_logs_delay_ms = env::var("GET_LOGS_DELAY_MS") .unwrap_or_else(|_| "50".to_string()) .parse()?; + let webhook_jwt_token = env::var("WEBHOOK_JWT_TOKEN").ok(); Ok(Self { database_url, @@ -74,6 +77,7 @@ impl Config { block_offset_from_head, get_logs_max_retries, get_logs_delay_ms, + webhook_jwt_token, }) } } diff --git a/src/db.rs b/src/db.rs deleted file mode 100644 index c988608..0000000 --- a/src/db.rs +++ /dev/null @@ -1,287 +0,0 @@ -use anyhow::Result; -use redb::{Database, ReadableTable, TableDefinition}; -use std::sync::Arc; - -const ACCOUNTS: TableDefinition<&str, (u32, &str, &str)> = TableDefinition::new("accounts"); // account_id -> (index, address, webhook_url) -const ADDRESS_TO_ID: TableDefinition<&str, &str> = TableDefinition::new("address_to_id"); -const DEPOSITS: TableDefinition<&str, (&str, &str, &str)> = TableDefinition::new("deposits"); // tx_hash -> (account_id, amount, status) -const STATE: TableDefinition<&str, &str> = TableDefinition::new("state"); -const TOKEN_METADATA: TableDefinition<&str, (&str, u64, &str)> = - TableDefinition::new("token_metadata"); // token_address -> (symbol, decimals, name) -const ERC20_DEPOSITS: TableDefinition<&str, (&str, &str, &str, &str, &str)> = - TableDefinition::new("erc20_deposits"); // tx_hash:log_index -> (account_id, amount, token_address, token_symbol, status) - -#[derive(Clone, Debug)] -pub struct Erc20Deposit { - pub key: String, - pub account_id: String, - pub amount: String, - pub token_address: String, - pub token_symbol: String, -} - -#[derive(Clone)] -pub struct Db { - db: Arc, -} - -impl Db { - pub fn new(path: &str) -> Result { - let db = Database::create(path)?; - - // Initialize tables - let write_txn = db.begin_write()?; - { - let _ = write_txn.open_table(ACCOUNTS)?; - let _ = write_txn.open_table(ADDRESS_TO_ID)?; - let _ = write_txn.open_table(DEPOSITS)?; - let _ = write_txn.open_table(STATE)?; - let _ = write_txn.open_table(TOKEN_METADATA)?; - let _ = write_txn.open_table(ERC20_DEPOSITS)?; - } - write_txn.commit()?; - - Ok(Self { db: Arc::new(db) }) - } - - #[allow(dead_code)] - pub fn get_next_derivation_index(&self) -> Result { - let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(ACCOUNTS)?; - // This is inefficient O(N) but fine for MVP. - // Better: Store a counter in STATE table. - let last = table.iter()?.next_back(); - - match last { - Some(Ok((_, v))) => Ok(v.value().0 + 1), - _ => Ok(0), - } - } - - pub fn register_account( - &self, - id: &str, - index: u32, - address: &str, - webhook_url: &str, - ) -> Result<()> { - let write_txn = self.db.begin_write()?; - { - let mut accounts = write_txn.open_table(ACCOUNTS)?; - accounts.insert(id, (index, address, webhook_url))?; - - let mut addr_map = write_txn.open_table(ADDRESS_TO_ID)?; - addr_map.insert(address, id)?; - } - write_txn.commit()?; - Ok(()) - } - - pub fn get_account_by_address(&self, address: &str) -> Result> { - let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(ADDRESS_TO_ID)?; - let result = table.get(address)?; - Ok(result.map(|v| v.value().to_string())) - } - - pub fn get_account_by_id(&self, id: &str) -> Result> { - let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(ACCOUNTS)?; - let result = table.get(id)?; - Ok(result.map(|v| { - let val = v.value(); - (val.0, val.1.to_string(), val.2.to_string()) - })) - } - - pub fn get_webhook_url(&self, account_id: &str) -> Result> { - let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(ACCOUNTS)?; - let result = table.get(account_id)?; - Ok(result.map(|v| v.value().2.to_string())) - } - - /// Record a deposit and return true if it was newly recorded, false if it was a duplicate - pub fn record_deposit(&self, tx_hash: &str, account_id: &str, amount: &str) -> Result { - let write_txn = self.db.begin_write()?; - let is_new = { - let mut deposits = write_txn.open_table(DEPOSITS)?; - // Check if exists to avoid overwrite and duplicates - if deposits.get(tx_hash)?.is_none() { - deposits.insert(tx_hash, (account_id, amount, "detected"))?; - true - } else { - false - } - }; - write_txn.commit()?; - Ok(is_new) - } - - pub fn mark_deposit_swept(&self, tx_hash: &str) -> Result<()> { - let write_txn = self.db.begin_write()?; - { - let mut deposits = write_txn.open_table(DEPOSITS)?; - let (account_id, amount) = { - let current_val = deposits.get(tx_hash)?; - if let Some(v) = current_val { - let val = v.value(); - (val.0.to_string(), val.1.to_string()) - } else { - return Ok(()); - } - }; - - deposits.insert(tx_hash, (account_id.as_str(), amount.as_str(), "swept"))?; - } - write_txn.commit()?; - Ok(()) - } - - pub fn get_detected_deposits(&self) -> Result> { - let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(DEPOSITS)?; - let mut results = Vec::new(); - for item in table.iter()? { - let (tx_hash, value) = item?; - let (account_id, amount, status) = value.value(); - if status == "detected" { - results.push(( - tx_hash.value().to_string(), - account_id.to_string(), - amount.to_string(), - )); - } - } - Ok(results) - } - - pub fn get_last_processed_block(&self) -> Result { - let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(STATE)?; - let result = table.get("last_block")?; - Ok(result.map(|v| v.value().parse().unwrap_or(0)).unwrap_or(0)) - } - - pub fn set_last_processed_block(&self, block: u64) -> Result<()> { - let write_txn = self.db.begin_write()?; - { - let mut state = write_txn.open_table(STATE)?; - state.insert("last_block", block.to_string().as_str())?; - } - write_txn.commit()?; - Ok(()) - } - - // ========== ERC20 Token Metadata ========== - - pub fn store_token_metadata( - &self, - address: &str, - symbol: &str, - decimals: u8, - name: &str, - ) -> Result<()> { - let write_txn = self.db.begin_write()?; - { - let mut metadata = write_txn.open_table(TOKEN_METADATA)?; - metadata.insert(address, (symbol, decimals as u64, name))?; - } - write_txn.commit()?; - Ok(()) - } - - pub fn get_token_metadata(&self, address: &str) -> Result> { - let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(TOKEN_METADATA)?; - let result = table.get(address)?; - Ok(result.map(|v| { - let val = v.value(); - (val.0.to_string(), val.1 as u8, val.2.to_string()) - })) - } - - // ========== ERC20 Deposits ========== - - /// Record an ERC20 deposit and return true if it was newly recorded, false if it was a duplicate - pub fn record_erc20_deposit( - &self, - tx_hash: &str, - log_index: u64, - account_id: &str, - amount: &str, - token_address: &str, - token_symbol: &str, - ) -> Result { - let write_txn = self.db.begin_write()?; - let is_new = { - let mut deposits = write_txn.open_table(ERC20_DEPOSITS)?; - let key = format!("{}:{}", tx_hash, log_index); - if deposits.get(key.as_str())?.is_none() { - deposits.insert( - key.as_str(), - (account_id, amount, token_address, token_symbol, "detected"), - )?; - true - } else { - false - } - }; - write_txn.commit()?; - Ok(is_new) - } - - pub fn get_detected_erc20_deposits(&self) -> Result> { - let read_txn = self.db.begin_read()?; - let table = read_txn.open_table(ERC20_DEPOSITS)?; - let mut results = Vec::new(); - for item in table.iter()? { - let (key, value) = item?; - let (account_id, amount, token_address, token_symbol, status) = value.value(); - if status == "detected" { - results.push(Erc20Deposit { - key: key.value().to_string(), // tx_hash:log_index - account_id: account_id.to_string(), - amount: amount.to_string(), - token_address: token_address.to_string(), - token_symbol: token_symbol.to_string(), - }); - } - } - Ok(results) - } - - pub fn mark_erc20_deposit_swept(&self, key: &str) -> Result<()> { - let write_txn = self.db.begin_write()?; - { - let mut deposits = write_txn.open_table(ERC20_DEPOSITS)?; - let (account_id, amount, token_address, token_symbol) = { - let current_val = deposits.get(key)?; - if let Some(v) = current_val { - let val = v.value(); - ( - val.0.to_string(), - val.1.to_string(), - val.2.to_string(), - val.3.to_string(), - ) - } else { - return Ok(()); - } - }; - - deposits.insert( - key, - ( - account_id.as_str(), - amount.as_str(), - token_address.as_str(), - token_symbol.as_str(), - "swept", - ), - )?; - } - write_txn.commit()?; - Ok(()) - } -} diff --git a/src/db/mod.rs b/src/db/mod.rs new file mode 100644 index 0000000..e4a631d --- /dev/null +++ b/src/db/mod.rs @@ -0,0 +1,145 @@ +mod pg; +mod redb; + +use anyhow::Result; +use async_trait::async_trait; +use std::sync::Arc; + +pub use pg::PostgresStorage; +pub use redb::RedbStorage; + +#[derive(Clone, Debug)] +pub struct Erc20Deposit { + pub key: String, + pub account_id: String, + pub amount: String, + pub token_address: String, + pub token_symbol: String, +} + +/// Storage backend for accounts, deposits and sweep bookkeeping. +/// +/// Implemented by [`RedbStorage`] (embedded file database) and +/// [`PostgresStorage`] (PostgreSQL server). +#[async_trait] +pub trait Storage: Send + Sync { + #[allow(dead_code)] + async fn get_next_derivation_index(&self) -> Result; + + async fn register_account( + &self, + id: &str, + index: u32, + address: &str, + webhook_url: &str, + ) -> Result<()>; + + async fn get_registration_id_by_address(&self, address: &str) -> Result>; + + async fn get_account_by_address(&self, address: &str) -> Result>; + + async fn get_account_by_id(&self, id: &str) -> Result>; + + async fn get_webhook_url(&self, account_id: &str) -> Result>; + + /// Record a deposit and return true if it was newly recorded, false if it was a duplicate + async fn record_deposit(&self, tx_hash: &str, account_id: &str, amount: &str) -> Result; + + async fn mark_deposit_swept(&self, tx_hash: &str) -> Result<()>; + + async fn get_detected_deposits(&self) -> Result>; + + async fn get_last_processed_block(&self) -> Result; + + async fn set_last_processed_block(&self, block: u64) -> Result<()>; + + async fn store_token_metadata( + &self, + address: &str, + symbol: &str, + decimals: u8, + name: &str, + ) -> Result<()>; + + async fn get_token_metadata(&self, address: &str) -> Result>; + + /// Record an ERC20 deposit and return true if it was newly recorded, false if it was a duplicate + async fn record_erc20_deposit( + &self, + tx_hash: &str, + log_index: u64, + account_id: &str, + amount: &str, + token_address: &str, + token_symbol: &str, + ) -> Result; + + async fn get_detected_erc20_deposits(&self) -> Result>; + + async fn mark_erc20_deposit_swept(&self, key: &str) -> Result<()>; + + /// Mark all detected ERC20 deposits for a given (account_id, token_address) as swept. + /// Returns the list of deposit keys that were marked. + async fn mark_erc20_deposits_swept_for_account_token( + &self, + account_id: &str, + token_address: &str, + ) -> Result>; + + /// Increment zero-balance retry count for a deposit. Returns the new count. + async fn increment_zero_balance_count(&self, key: &str) -> Result; + + /// Store the on-chain sweep tx hash for a single deposit key. + #[allow(dead_code)] + async fn set_sweep_tx_hash(&self, key: &str, tx_hash: &str) -> Result<()>; + + /// Store the on-chain sweep tx hash for multiple deposit keys in one transaction. + async fn set_sweep_tx_hash_for_keys(&self, keys: &[String], tx_hash: &str) -> Result<()>; + + /// Read sweep metadata for a deposit key. + #[allow(dead_code)] + async fn get_sweep_meta(&self, key: &str) -> Result>; + + /// Increment the sweep failure count for a deposit. Returns the new count. + async fn increment_sweep_failure_count(&self, key: &str) -> Result; + + /// Mark a single ERC20 deposit as permanently failed. + #[allow(dead_code)] + async fn mark_erc20_deposit_failed(&self, key: &str) -> Result<()>; + + /// Mark all detected ERC20 deposits for a given (account_id, token_address) as permanently failed. + /// Returns the list of deposit keys that were marked. + async fn mark_erc20_deposits_failed_for_account_token( + &self, + account_id: &str, + token_address: &str, + ) -> Result>; +} + +/// Handle to the configured storage backend. +/// +/// The driver is selected from the database URL scheme: +/// `postgres://` / `postgresql://` use PostgreSQL, anything else is treated +/// as a redb file path. +#[derive(Clone)] +pub struct Db(Arc); + +impl Db { + pub async fn new(database_url: &str) -> Result { + if database_url.starts_with("postgres://") || database_url.starts_with("postgresql://") { + Ok(Self(Arc::new( + PostgresStorage::connect(database_url).await?, + ))) + } else { + Ok(Self(Arc::new(RedbStorage::new(database_url)?))) + } + } +} + +impl std::ops::Deref for Db { + type Target = dyn Storage; + + fn deref(&self) -> &Self::Target { + &*self.0 + } +} diff --git a/src/db/pg.rs b/src/db/pg.rs new file mode 100644 index 0000000..b8f7a0b --- /dev/null +++ b/src/db/pg.rs @@ -0,0 +1,405 @@ +use anyhow::Result; +use async_trait::async_trait; +use sqlx::postgres::{PgPool, PgPoolOptions}; +use sqlx::Row; + +use super::{Erc20Deposit, Storage}; + +const SCHEMA: &str = r#" +CREATE TABLE IF NOT EXISTS accounts ( + id TEXT PRIMARY KEY, + derivation_index BIGINT NOT NULL, + address TEXT NOT NULL, + webhook_url TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS accounts_address_idx ON accounts (address); + +CREATE TABLE IF NOT EXISTS deposits ( + tx_hash TEXT PRIMARY KEY, + account_id TEXT NOT NULL, + amount TEXT NOT NULL, + status TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS deposits_status_idx ON deposits (status); + +CREATE TABLE IF NOT EXISTS state ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS token_metadata ( + address TEXT PRIMARY KEY, + symbol TEXT NOT NULL, + decimals BIGINT NOT NULL, + name TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS erc20_deposits ( + key TEXT PRIMARY KEY, + account_id TEXT NOT NULL, + amount TEXT NOT NULL, + token_address TEXT NOT NULL, + token_symbol TEXT NOT NULL, + status TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS erc20_deposits_status_idx ON erc20_deposits (status); + +CREATE TABLE IF NOT EXISTS sweep_meta ( + key TEXT PRIMARY KEY, + sweep_tx_hash TEXT NOT NULL DEFAULT '', + zero_balance_retry_count BIGINT NOT NULL DEFAULT 0 +); + +CREATE TABLE IF NOT EXISTS sweep_failures ( + key TEXT PRIMARY KEY, + count BIGINT NOT NULL DEFAULT 0 +); +"#; + +/// PostgreSQL storage driver. +#[derive(Clone)] +pub struct PostgresStorage { + pool: PgPool, +} + +impl PostgresStorage { + pub async fn connect(database_url: &str) -> Result { + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(database_url) + .await?; + + sqlx::raw_sql(SCHEMA).execute(&pool).await?; + + Ok(Self { pool }) + } +} + +#[async_trait] +impl Storage for PostgresStorage { + async fn get_next_derivation_index(&self) -> Result { + let row = + sqlx::query("SELECT COALESCE(MAX(derivation_index), -1) AS max_index FROM accounts") + .fetch_one(&self.pool) + .await?; + let max_index: i64 = row.get("max_index"); + Ok((max_index + 1) as u32) + } + + async fn register_account( + &self, + id: &str, + index: u32, + address: &str, + webhook_url: &str, + ) -> Result<()> { + sqlx::query( + "INSERT INTO accounts (id, derivation_index, address, webhook_url) + VALUES ($1, $2, $3, $4) + ON CONFLICT (id) DO UPDATE + SET derivation_index = EXCLUDED.derivation_index, + address = EXCLUDED.address, + webhook_url = EXCLUDED.webhook_url", + ) + .bind(id) + .bind(index as i64) + .bind(address) + .bind(webhook_url) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn get_registration_id_by_address(&self, address: &str) -> Result> { + let row = sqlx::query("SELECT id FROM accounts WHERE address = $1 LIMIT 1") + .bind(address) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| r.get("id"))) + } + + async fn get_account_by_address(&self, address: &str) -> Result> { + self.get_registration_id_by_address(address).await + } + + async fn get_account_by_id(&self, id: &str) -> Result> { + let row = sqlx::query( + "SELECT derivation_index, address, webhook_url FROM accounts WHERE id = $1", + ) + .bind(id) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| { + let index: i64 = r.get("derivation_index"); + (index as u32, r.get("address"), r.get("webhook_url")) + })) + } + + async fn get_webhook_url(&self, account_id: &str) -> Result> { + let row = sqlx::query("SELECT webhook_url FROM accounts WHERE id = $1") + .bind(account_id) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| r.get("webhook_url"))) + } + + async fn record_deposit(&self, tx_hash: &str, account_id: &str, amount: &str) -> Result { + let result = sqlx::query( + "INSERT INTO deposits (tx_hash, account_id, amount, status) + VALUES ($1, $2, $3, 'detected') + ON CONFLICT (tx_hash) DO NOTHING", + ) + .bind(tx_hash) + .bind(account_id) + .bind(amount) + .execute(&self.pool) + .await?; + Ok(result.rows_affected() == 1) + } + + async fn mark_deposit_swept(&self, tx_hash: &str) -> Result<()> { + sqlx::query("UPDATE deposits SET status = 'swept' WHERE tx_hash = $1") + .bind(tx_hash) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn get_detected_deposits(&self) -> Result> { + let rows = sqlx::query( + "SELECT tx_hash, account_id, amount FROM deposits WHERE status = 'detected' ORDER BY tx_hash", + ) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|r| (r.get("tx_hash"), r.get("account_id"), r.get("amount"))) + .collect()) + } + + async fn get_last_processed_block(&self) -> Result { + let row = sqlx::query("SELECT value FROM state WHERE key = 'last_block'") + .fetch_optional(&self.pool) + .await?; + Ok(row + .map(|r| r.get::("value").parse().unwrap_or(0)) + .unwrap_or(0)) + } + + async fn set_last_processed_block(&self, block: u64) -> Result<()> { + sqlx::query( + "INSERT INTO state (key, value) VALUES ('last_block', $1) + ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value", + ) + .bind(block.to_string()) + .execute(&self.pool) + .await?; + Ok(()) + } + + // ========== ERC20 Token Metadata ========== + + async fn store_token_metadata( + &self, + address: &str, + symbol: &str, + decimals: u8, + name: &str, + ) -> Result<()> { + sqlx::query( + "INSERT INTO token_metadata (address, symbol, decimals, name) + VALUES ($1, $2, $3, $4) + ON CONFLICT (address) DO UPDATE + SET symbol = EXCLUDED.symbol, + decimals = EXCLUDED.decimals, + name = EXCLUDED.name", + ) + .bind(address) + .bind(symbol) + .bind(decimals as i64) + .bind(name) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn get_token_metadata(&self, address: &str) -> Result> { + let row = + sqlx::query("SELECT symbol, decimals, name FROM token_metadata WHERE address = $1") + .bind(address) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| { + let decimals: i64 = r.get("decimals"); + (r.get("symbol"), decimals as u8, r.get("name")) + })) + } + + // ========== ERC20 Deposits ========== + + async fn record_erc20_deposit( + &self, + tx_hash: &str, + log_index: u64, + account_id: &str, + amount: &str, + token_address: &str, + token_symbol: &str, + ) -> Result { + let key = format!("{}:{}", tx_hash, log_index); + let result = sqlx::query( + "INSERT INTO erc20_deposits (key, account_id, amount, token_address, token_symbol, status) + VALUES ($1, $2, $3, $4, $5, 'detected') + ON CONFLICT (key) DO NOTHING", + ) + .bind(&key) + .bind(account_id) + .bind(amount) + .bind(token_address) + .bind(token_symbol) + .execute(&self.pool) + .await?; + Ok(result.rows_affected() == 1) + } + + async fn get_detected_erc20_deposits(&self) -> Result> { + let rows = sqlx::query( + "SELECT key, account_id, amount, token_address, token_symbol + FROM erc20_deposits WHERE status = 'detected' ORDER BY key", + ) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|r| Erc20Deposit { + key: r.get("key"), + account_id: r.get("account_id"), + amount: r.get("amount"), + token_address: r.get("token_address"), + token_symbol: r.get("token_symbol"), + }) + .collect()) + } + + async fn mark_erc20_deposit_swept(&self, key: &str) -> Result<()> { + sqlx::query("UPDATE erc20_deposits SET status = 'swept' WHERE key = $1") + .bind(key) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn mark_erc20_deposits_swept_for_account_token( + &self, + account_id: &str, + token_address: &str, + ) -> Result> { + let rows = sqlx::query( + "UPDATE erc20_deposits SET status = 'swept' + WHERE status = 'detected' AND account_id = $1 AND token_address = $2 + RETURNING key", + ) + .bind(account_id) + .bind(token_address) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(|r| r.get("key")).collect()) + } + + // ========== Sweep Metadata ========== + + async fn increment_zero_balance_count(&self, key: &str) -> Result { + let row = sqlx::query( + "INSERT INTO sweep_meta (key, zero_balance_retry_count) VALUES ($1, 1) + ON CONFLICT (key) DO UPDATE + SET zero_balance_retry_count = sweep_meta.zero_balance_retry_count + 1 + RETURNING zero_balance_retry_count", + ) + .bind(key) + .fetch_one(&self.pool) + .await?; + let count: i64 = row.get("zero_balance_retry_count"); + Ok(count as u64) + } + + async fn set_sweep_tx_hash(&self, key: &str, tx_hash: &str) -> Result<()> { + sqlx::query( + "INSERT INTO sweep_meta (key, sweep_tx_hash) VALUES ($1, $2) + ON CONFLICT (key) DO UPDATE SET sweep_tx_hash = EXCLUDED.sweep_tx_hash", + ) + .bind(key) + .bind(tx_hash) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn set_sweep_tx_hash_for_keys(&self, keys: &[String], tx_hash: &str) -> Result<()> { + let mut txn = self.pool.begin().await?; + for key in keys { + sqlx::query( + "INSERT INTO sweep_meta (key, sweep_tx_hash) VALUES ($1, $2) + ON CONFLICT (key) DO UPDATE SET sweep_tx_hash = EXCLUDED.sweep_tx_hash", + ) + .bind(key) + .bind(tx_hash) + .execute(&mut *txn) + .await?; + } + txn.commit().await?; + Ok(()) + } + + async fn get_sweep_meta(&self, key: &str) -> Result> { + let row = sqlx::query( + "SELECT sweep_tx_hash, zero_balance_retry_count FROM sweep_meta WHERE key = $1", + ) + .bind(key) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| { + let count: i64 = r.get("zero_balance_retry_count"); + (r.get("sweep_tx_hash"), count as u64) + })) + } + + // ========== Sweep Failure Tracking ========== + + async fn increment_sweep_failure_count(&self, key: &str) -> Result { + let row = sqlx::query( + "INSERT INTO sweep_failures (key, count) VALUES ($1, 1) + ON CONFLICT (key) DO UPDATE SET count = sweep_failures.count + 1 + RETURNING count", + ) + .bind(key) + .fetch_one(&self.pool) + .await?; + let count: i64 = row.get("count"); + Ok(count as u64) + } + + async fn mark_erc20_deposit_failed(&self, key: &str) -> Result<()> { + sqlx::query("UPDATE erc20_deposits SET status = 'failed' WHERE key = $1") + .bind(key) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn mark_erc20_deposits_failed_for_account_token( + &self, + account_id: &str, + token_address: &str, + ) -> Result> { + let rows = sqlx::query( + "UPDATE erc20_deposits SET status = 'failed' + WHERE status = 'detected' AND account_id = $1 AND token_address = $2 + RETURNING key", + ) + .bind(account_id) + .bind(token_address) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(|r| r.get("key")).collect()) + } +} diff --git a/src/db/redb.rs b/src/db/redb.rs new file mode 100644 index 0000000..222d62a --- /dev/null +++ b/src/db/redb.rs @@ -0,0 +1,498 @@ +use anyhow::Result; +use async_trait::async_trait; +use redb::{Database, ReadableTable, TableDefinition}; +use std::sync::Arc; + +use super::{Erc20Deposit, Storage}; + +const ACCOUNTS: TableDefinition<&str, (u32, &str, &str)> = TableDefinition::new("accounts"); // account_id -> (index, address, webhook_url) +const ADDRESS_TO_ID: TableDefinition<&str, &str> = TableDefinition::new("address_to_id"); +const DEPOSITS: TableDefinition<&str, (&str, &str, &str)> = TableDefinition::new("deposits"); // tx_hash -> (account_id, amount, status) +const STATE: TableDefinition<&str, &str> = TableDefinition::new("state"); +const TOKEN_METADATA: TableDefinition<&str, (&str, u64, &str)> = + TableDefinition::new("token_metadata"); // token_address -> (symbol, decimals, name) +const ERC20_DEPOSITS: TableDefinition<&str, (&str, &str, &str, &str, &str)> = + TableDefinition::new("erc20_deposits"); // tx_hash:log_index -> (account_id, amount, token_address, token_symbol, status) +const SWEEP_META: TableDefinition<&str, (&str, u64)> = TableDefinition::new("sweep_meta"); // deposit_key -> (sweep_tx_hash, zero_balance_retry_count) +const SWEEP_FAILURES: TableDefinition<&str, u64> = TableDefinition::new("sweep_failures"); // deposit_key -> consecutive_failure_count + +/// Embedded file-based storage driver backed by redb. +#[derive(Clone)] +pub struct RedbStorage { + db: Arc, +} + +impl RedbStorage { + pub fn new(path: &str) -> Result { + let db = Database::create(path)?; + + // Initialize tables + let write_txn = db.begin_write()?; + { + let _ = write_txn.open_table(ACCOUNTS)?; + let _ = write_txn.open_table(ADDRESS_TO_ID)?; + let _ = write_txn.open_table(DEPOSITS)?; + let _ = write_txn.open_table(STATE)?; + let _ = write_txn.open_table(TOKEN_METADATA)?; + let _ = write_txn.open_table(ERC20_DEPOSITS)?; + let _ = write_txn.open_table(SWEEP_META)?; + let _ = write_txn.open_table(SWEEP_FAILURES)?; + } + write_txn.commit()?; + + Ok(Self { db: Arc::new(db) }) + } +} + +#[async_trait] +impl Storage for RedbStorage { + async fn get_next_derivation_index(&self) -> Result { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(ACCOUNTS)?; + // This is inefficient O(N) but fine for MVP. + // Better: Store a counter in STATE table. + let last = table.iter()?.next_back(); + + match last { + Some(Ok((_, v))) => Ok(v.value().0 + 1), + _ => Ok(0), + } + } + + async fn register_account( + &self, + id: &str, + index: u32, + address: &str, + webhook_url: &str, + ) -> Result<()> { + let write_txn = self.db.begin_write()?; + { + let mut accounts = write_txn.open_table(ACCOUNTS)?; + accounts.insert(id, (index, address, webhook_url))?; + + let mut addr_map = write_txn.open_table(ADDRESS_TO_ID)?; + addr_map.insert(address, id)?; + } + write_txn.commit()?; + Ok(()) + } + + async fn get_registration_id_by_address(&self, address: &str) -> Result> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(ADDRESS_TO_ID)?; + let result = table.get(address)?; + Ok(result.map(|v| v.value().to_string())) + } + + async fn get_account_by_address(&self, address: &str) -> Result> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(ADDRESS_TO_ID)?; + let result = table.get(address)?; + Ok(result.map(|v| v.value().to_string())) + } + + async fn get_account_by_id(&self, id: &str) -> Result> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(ACCOUNTS)?; + let result = table.get(id)?; + Ok(result.map(|v| { + let val = v.value(); + (val.0, val.1.to_string(), val.2.to_string()) + })) + } + + async fn get_webhook_url(&self, account_id: &str) -> Result> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(ACCOUNTS)?; + let result = table.get(account_id)?; + Ok(result.map(|v| v.value().2.to_string())) + } + + async fn record_deposit(&self, tx_hash: &str, account_id: &str, amount: &str) -> Result { + let write_txn = self.db.begin_write()?; + let is_new = { + let mut deposits = write_txn.open_table(DEPOSITS)?; + // Check if exists to avoid overwrite and duplicates + if deposits.get(tx_hash)?.is_none() { + deposits.insert(tx_hash, (account_id, amount, "detected"))?; + true + } else { + false + } + }; + write_txn.commit()?; + Ok(is_new) + } + + async fn mark_deposit_swept(&self, tx_hash: &str) -> Result<()> { + let write_txn = self.db.begin_write()?; + { + let mut deposits = write_txn.open_table(DEPOSITS)?; + let (account_id, amount) = { + let current_val = deposits.get(tx_hash)?; + if let Some(v) = current_val { + let val = v.value(); + (val.0.to_string(), val.1.to_string()) + } else { + return Ok(()); + } + }; + + deposits.insert(tx_hash, (account_id.as_str(), amount.as_str(), "swept"))?; + } + write_txn.commit()?; + Ok(()) + } + + async fn get_detected_deposits(&self) -> Result> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(DEPOSITS)?; + let mut results = Vec::new(); + for item in table.iter()? { + let (tx_hash, value) = item?; + let (account_id, amount, status) = value.value(); + if status == "detected" { + results.push(( + tx_hash.value().to_string(), + account_id.to_string(), + amount.to_string(), + )); + } + } + Ok(results) + } + + async fn get_last_processed_block(&self) -> Result { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(STATE)?; + let result = table.get("last_block")?; + Ok(result.map(|v| v.value().parse().unwrap_or(0)).unwrap_or(0)) + } + + async fn set_last_processed_block(&self, block: u64) -> Result<()> { + let write_txn = self.db.begin_write()?; + { + let mut state = write_txn.open_table(STATE)?; + state.insert("last_block", block.to_string().as_str())?; + } + write_txn.commit()?; + Ok(()) + } + + // ========== ERC20 Token Metadata ========== + + async fn store_token_metadata( + &self, + address: &str, + symbol: &str, + decimals: u8, + name: &str, + ) -> Result<()> { + let write_txn = self.db.begin_write()?; + { + let mut metadata = write_txn.open_table(TOKEN_METADATA)?; + metadata.insert(address, (symbol, decimals as u64, name))?; + } + write_txn.commit()?; + Ok(()) + } + + async fn get_token_metadata(&self, address: &str) -> Result> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(TOKEN_METADATA)?; + let result = table.get(address)?; + Ok(result.map(|v| { + let val = v.value(); + (val.0.to_string(), val.1 as u8, val.2.to_string()) + })) + } + + // ========== ERC20 Deposits ========== + + async fn record_erc20_deposit( + &self, + tx_hash: &str, + log_index: u64, + account_id: &str, + amount: &str, + token_address: &str, + token_symbol: &str, + ) -> Result { + let write_txn = self.db.begin_write()?; + let is_new = { + let mut deposits = write_txn.open_table(ERC20_DEPOSITS)?; + let key = format!("{}:{}", tx_hash, log_index); + if deposits.get(key.as_str())?.is_none() { + deposits.insert( + key.as_str(), + (account_id, amount, token_address, token_symbol, "detected"), + )?; + true + } else { + false + } + }; + write_txn.commit()?; + Ok(is_new) + } + + async fn get_detected_erc20_deposits(&self) -> Result> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(ERC20_DEPOSITS)?; + let mut results = Vec::new(); + for item in table.iter()? { + let (key, value) = item?; + let (account_id, amount, token_address, token_symbol, status) = value.value(); + if status == "detected" { + results.push(Erc20Deposit { + key: key.value().to_string(), // tx_hash:log_index + account_id: account_id.to_string(), + amount: amount.to_string(), + token_address: token_address.to_string(), + token_symbol: token_symbol.to_string(), + }); + } + } + Ok(results) + } + + async fn mark_erc20_deposit_swept(&self, key: &str) -> Result<()> { + let write_txn = self.db.begin_write()?; + { + let mut deposits = write_txn.open_table(ERC20_DEPOSITS)?; + let (account_id, amount, token_address, token_symbol) = { + let current_val = deposits.get(key)?; + if let Some(v) = current_val { + let val = v.value(); + ( + val.0.to_string(), + val.1.to_string(), + val.2.to_string(), + val.3.to_string(), + ) + } else { + return Ok(()); + } + }; + + deposits.insert( + key, + ( + account_id.as_str(), + amount.as_str(), + token_address.as_str(), + token_symbol.as_str(), + "swept", + ), + )?; + } + write_txn.commit()?; + Ok(()) + } + + async fn mark_erc20_deposits_swept_for_account_token( + &self, + account_id: &str, + token_address: &str, + ) -> Result> { + let write_txn = self.db.begin_write()?; + let mut marked_keys = Vec::new(); + { + let mut deposits = write_txn.open_table(ERC20_DEPOSITS)?; + + // First pass: collect keys that need updating + let keys_to_update: Vec<(String, String, String, String)> = { + let mut to_update = Vec::new(); + for item in deposits.iter()? { + let (key, value) = item?; + let (acc_id, amount, tok_addr, tok_symbol, status) = value.value(); + if status == "detected" && acc_id == account_id && tok_addr == token_address { + to_update.push(( + key.value().to_string(), + amount.to_string(), + tok_symbol.to_string(), + acc_id.to_string(), + )); + } + } + to_update + }; + + // Second pass: update the entries + for (key, amount, tok_symbol, acc_id) in &keys_to_update { + deposits.insert( + key.as_str(), + ( + acc_id.as_str(), + amount.as_str(), + token_address, + tok_symbol.as_str(), + "swept", + ), + )?; + marked_keys.push(key.clone()); + } + } + write_txn.commit()?; + Ok(marked_keys) + } + + // ========== Sweep Metadata (new table, existing schemas unchanged) ========== + + async fn increment_zero_balance_count(&self, key: &str) -> Result { + let write_txn = self.db.begin_write()?; + let new_count = { + let mut meta = write_txn.open_table(SWEEP_META)?; + let (sweep_tx_hash, count) = match meta.get(key)? { + Some(v) => { + let val = v.value(); + (val.0.to_string(), val.1) + } + None => (String::new(), 0), + }; + let new_count = count + 1; + meta.insert(key, (sweep_tx_hash.as_str(), new_count))?; + new_count + }; + write_txn.commit()?; + Ok(new_count) + } + + async fn set_sweep_tx_hash(&self, key: &str, tx_hash: &str) -> Result<()> { + let write_txn = self.db.begin_write()?; + { + let mut meta = write_txn.open_table(SWEEP_META)?; + let count = match meta.get(key)? { + Some(v) => v.value().1, + None => 0, + }; + meta.insert(key, (tx_hash, count))?; + } + write_txn.commit()?; + Ok(()) + } + + async fn set_sweep_tx_hash_for_keys(&self, keys: &[String], tx_hash: &str) -> Result<()> { + let write_txn = self.db.begin_write()?; + { + let mut meta = write_txn.open_table(SWEEP_META)?; + for key in keys { + let count = match meta.get(key.as_str())? { + Some(v) => v.value().1, + None => 0, + }; + meta.insert(key.as_str(), (tx_hash, count))?; + } + } + write_txn.commit()?; + Ok(()) + } + + async fn get_sweep_meta(&self, key: &str) -> Result> { + let read_txn = self.db.begin_read()?; + let table = read_txn.open_table(SWEEP_META)?; + let result = table.get(key)?; + Ok(result.map(|v| { + let val = v.value(); + (val.0.to_string(), val.1) + })) + } + + // ========== Sweep Failure Tracking ========== + + async fn increment_sweep_failure_count(&self, key: &str) -> Result { + let write_txn = self.db.begin_write()?; + let new_count = { + let mut failures = write_txn.open_table(SWEEP_FAILURES)?; + let count = match failures.get(key)? { + Some(v) => v.value(), + None => 0, + }; + let new_count = count + 1; + failures.insert(key, new_count)?; + new_count + }; + write_txn.commit()?; + Ok(new_count) + } + + async fn mark_erc20_deposit_failed(&self, key: &str) -> Result<()> { + let write_txn = self.db.begin_write()?; + { + let mut deposits = write_txn.open_table(ERC20_DEPOSITS)?; + let (account_id, amount, token_address, token_symbol) = { + let current_val = deposits.get(key)?; + if let Some(v) = current_val { + let val = v.value(); + ( + val.0.to_string(), + val.1.to_string(), + val.2.to_string(), + val.3.to_string(), + ) + } else { + return Ok(()); + } + }; + + deposits.insert( + key, + ( + account_id.as_str(), + amount.as_str(), + token_address.as_str(), + token_symbol.as_str(), + "failed", + ), + )?; + } + write_txn.commit()?; + Ok(()) + } + + async fn mark_erc20_deposits_failed_for_account_token( + &self, + account_id: &str, + token_address: &str, + ) -> Result> { + let write_txn = self.db.begin_write()?; + let mut marked_keys = Vec::new(); + { + let mut deposits = write_txn.open_table(ERC20_DEPOSITS)?; + + let keys_to_update: Vec<(String, String, String, String)> = { + let mut to_update = Vec::new(); + for item in deposits.iter()? { + let (key, value) = item?; + let (acc_id, amount, tok_addr, tok_symbol, status) = value.value(); + if status == "detected" && acc_id == account_id && tok_addr == token_address { + to_update.push(( + key.value().to_string(), + amount.to_string(), + tok_symbol.to_string(), + acc_id.to_string(), + )); + } + } + to_update + }; + + for (key, amount, tok_symbol, acc_id) in &keys_to_update { + deposits.insert( + key.as_str(), + ( + acc_id.as_str(), + amount.as_str(), + token_address, + tok_symbol.as_str(), + "failed", + ), + )?; + marked_keys.push(key.clone()); + } + } + write_txn.commit()?; + Ok(marked_keys) + } +} diff --git a/src/e2e_tests.rs b/src/e2e_tests.rs index 839ffa2..00130c5 100644 --- a/src/e2e_tests.rs +++ b/src/e2e_tests.rs @@ -1,11 +1,13 @@ use crate::config::{Config, ProviderUrl}; use crate::db::Db; +use crate::faucet::Faucet; use crate::monitor::Monitor; use crate::sweeper::Sweeper; use crate::traits::Service; use crate::wallet::Wallet; use alloy::providers::ProviderBuilder; use serde_json::json; +use std::sync::Arc; use std::time::Duration; use tempfile::NamedTempFile; use tokio::time::sleep; @@ -37,10 +39,11 @@ async fn test_e2e_deposit_sweep_flow() { block_offset_from_head: 0, // Use 0 for tests to avoid underflow with low block numbers get_logs_max_retries: 30, get_logs_delay_ms: 50, + webhook_jwt_token: None, }; let wallet = Wallet::new(config.mnemonic.clone()); - let db = Db::new(&config.database_url).unwrap(); + let db = Db::new(&config.database_url).await.unwrap(); // 3. Register Users // User 1 -> Index 0 @@ -48,12 +51,14 @@ async fn test_e2e_deposit_sweep_flow() { let addr1_str = addr1.to_string(); let webhook_url = webhook_server.uri(); db.register_account("user_1", 0, &addr1_str, &webhook_url) + .await .unwrap(); // User 2 -> Index 1 let addr2 = wallet.derive_address(1).unwrap(); let addr2_str = addr2.to_string(); db.register_account("user_2", 1, &addr2_str, &webhook_url) + .await .unwrap(); // 4. Initialize Provider @@ -151,6 +156,33 @@ async fn test_e2e_deposit_sweep_flow() { .mount(&rpc_server) .await; + // eth_feeHistory (for EIP-1559 fee estimation) + Mock::given(method("POST")) + .and(body_json_contains("eth_feeHistory")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": 1, + "result": { + "baseFeePerGas": ["0x3B9ACA00", "0x3B9ACA00"], // 1 Gwei + "gasUsedRatio": [0.5], + "oldestBlock": "0x9", + "reward": [["0x3B9ACA00"]] // 1 Gwei priority fee + } + }))) + .mount(&rpc_server) + .await; + + // eth_maxPriorityFeePerGas (fallback for EIP-1559) + Mock::given(method("POST")) + .and(body_json_contains("eth_maxPriorityFeePerGas")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": 1, + "result": "0x3B9ACA00" // 1 Gwei + }))) + .mount(&rpc_server) + .await; + // eth_getTransactionCount (Nonce) Mock::given(method("POST")) .and(body_json_contains("eth_getTransactionCount")) @@ -232,7 +264,21 @@ async fn test_e2e_deposit_sweep_flow() { // 5. Run Monitor & Sweeper let monitor = Monitor::new(config.clone(), db.clone(), provider.clone()); - let sweeper = Sweeper::new(config.clone(), db.clone(), wallet.clone(), provider.clone()); + let faucet = Arc::new( + Faucet::new( + config.faucet_mnemonic.clone(), + provider.clone(), + &config.existential_deposit, + ) + .unwrap(), + ); + let sweeper = Sweeper::new( + config.clone(), + db.clone(), + wallet.clone(), + provider.clone(), + faucet, + ); // Run monitor once (manually or spawn short lived) // We can't easily "run once" with the loop, but we can spawn and wait a bit. @@ -250,7 +296,7 @@ async fn test_e2e_deposit_sweep_flow() { // Wait for deposit detection let mut detected = false; for _ in 0..10 { - let deposits = db.get_detected_deposits().unwrap(); + let deposits = db.get_detected_deposits().await.unwrap(); if !deposits.is_empty() { detected = true; break; @@ -265,7 +311,7 @@ async fn test_e2e_deposit_sweep_flow() { // We don't have a direct "get_swept_deposits" but we can check if detected list is empty // assuming we only had one. Or check DB directly if we exposed a method. // Let's check if detected becomes empty. - let deposits = db.get_detected_deposits().unwrap(); + let deposits = db.get_detected_deposits().await.unwrap(); if deposits.is_empty() { swept = true; break; diff --git a/src/lib.rs b/src/lib.rs index 445a931..595ff3d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,7 +1,7 @@ // Library modules pub mod config; pub mod db; -mod faucet; +pub(crate) mod faucet; mod monitor; mod sweeper; pub mod traits; @@ -127,6 +127,16 @@ where Ok("OK".to_string()) } + /// Set the last processed block number manually + pub async fn set_block_number(&self, block_number: u64) -> anyhow::Result<()> { + self.db.set_last_processed_block(block_number).await + } + + /// Get the current last processed block number + pub async fn get_block_number(&self) -> anyhow::Result { + self.db.get_last_processed_block().await + } + /// Verify if a transaction contains a transfer matching the expected criteria pub async fn verify_transfer( &self, @@ -379,7 +389,7 @@ where // Check if account already exists if let Ok(Some((_index, existing_address, _webhook))) = - self.db.get_account_by_id(&request.id) + self.db.get_account_by_id(&request.id).await { info!( "Account {} already exists with address {}", @@ -403,7 +413,8 @@ where // Save to DB with webhook URL self.db - .register_account(&request.id, index, &address_str, &request.webhook_url)?; + .register_account(&request.id, index, &address_str, &request.webhook_url) + .await?; info!( "Registered account {} with address {} (index: {})", @@ -415,6 +426,7 @@ where let db = self.db.clone(); let account_id = request.id.clone(); let address_for_funding = address_str.clone(); + let webhook_jwt_token = self.config.webhook_jwt_token.clone(); tokio::spawn(async move { info!( @@ -437,6 +449,7 @@ where &tx_hash, true, None, + webhook_jwt_token.as_deref(), ) .await { @@ -457,6 +470,7 @@ where "", false, Some(&e.to_string()), + webhook_jwt_token.as_deref(), ) .await { @@ -480,7 +494,7 @@ where impl HotWalletService> { /// Create a new HotWalletService with HTTP provider from configuration pub async fn new_http(config: Config) -> anyhow::Result { - let db = Db::new(&config.database_url)?; + let db = Db::new(&config.database_url).await?; let wallet = Wallet::new(config.mnemonic.clone()); let url = match &config.provider_url { @@ -526,15 +540,25 @@ impl HotWalletService> { } }); + // Create faucet for sweeper + let sweeper_faucet = Arc::new(Faucet::new( + self.config.faucet_mnemonic.clone(), + provider.clone(), + &self.config.existential_deposit, + )?); + // Spawn Sweeper tokio::spawn({ let config = self.config.clone(); let db = self.db.clone(); let wallet = self.wallet.clone(); let provider = provider.clone(); + let faucet = sweeper_faucet; async move { tracing::info!("Starting Sweeper in Polling mode"); - Sweeper::new(config, db, wallet, provider).run().await; + Sweeper::new(config, db, wallet, provider, faucet) + .run() + .await; } }); @@ -546,7 +570,7 @@ impl HotWalletService> { impl HotWalletService { /// Create a new HotWalletService with WebSocket provider from configuration pub async fn new_ws(config: Config) -> anyhow::Result { - let db = Db::new(&config.database_url)?; + let db = Db::new(&config.database_url).await?; let wallet = Wallet::new(config.mnemonic.clone()); let url = match &config.provider_url { @@ -591,15 +615,25 @@ impl HotWalletService { } }); + // Create faucet for sweeper + let sweeper_faucet = Arc::new(Faucet::new( + self.config.faucet_mnemonic.clone(), + provider.clone(), + &self.config.existential_deposit, + )?); + // Spawn Sweeper tokio::spawn({ let config = self.config.clone(); let db = self.db.clone(); let wallet = self.wallet.clone(); let provider = provider.clone(); + let faucet = sweeper_faucet; async move { tracing::info!("Starting Sweeper in Streaming mode"); - Sweeper::new(config, db, wallet, provider).run().await; + Sweeper::new(config, db, wallet, provider, faucet) + .run() + .await; } }); @@ -608,32 +642,37 @@ impl HotWalletService { } /// Send webhook notification for faucet funding event +/// registration_id: The original id used when registering the account +/// address: The Polygon address (account_id in webhook) +/// jwt_token: Optional JWT token for authorization header async fn send_faucet_funding_webhook( db: &Db, - account_id: &str, + registration_id: &str, address: &str, tx_hash: &str, success: bool, error_message: Option<&str>, + jwt_token: Option<&str>, ) -> anyhow::Result<()> { use tracing::{error, info}; - // Get the webhook URL for this account - let webhook_url = match db.get_webhook_url(account_id)? { - Some(url) => url, - None => { - error!("No webhook URL found for account: {}", account_id); - return Ok(()); - } + // Get the webhook URL using registration_id (the key in ACCOUNTS table) + let Some(webhook_url) = db.get_webhook_url(registration_id).await? else { + error!( + "No webhook URL found for registration_id: {}", + registration_id + ); + return Ok(()); }; let client = reqwest::Client::new(); let mut payload = serde_json::json!({ "event": "faucet_funding", - "account_id": account_id, - "address": address, + "account_id": address, + "registration_id": registration_id, "success": success, + "id": format!("{}:funding", registration_id) }); // Add tx_hash if funding was successful @@ -646,13 +685,21 @@ async fn send_faucet_funding_webhook( payload["error"] = serde_json::json!(error); } - let res = client.post(&webhook_url).json(&payload).send().await; + let mut request = client.post(&webhook_url).json(&payload); + + // Add JWT authorization header if provided + if let Some(token) = jwt_token { + request = request.header("Authorization", format!("Bearer {}", token)); + } + + let res = request.send().await; match res { Ok(r) => info!( - "Faucet funding webhook sent to {}: status={}", + "Faucet funding webhook sent to {}: status={}, registration_id={}", webhook_url, - r.status() + r.status(), + registration_id ), Err(e) => error!( "Failed to send faucet funding webhook to {}: {:?}", diff --git a/src/main.rs b/src/main.rs index 32feebb..9fca72e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,10 +13,7 @@ async fn main() -> anyhow::Result<()> { use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; tracing_subscriber::registry() - .with( - EnvFilter::try_from_default_env() - .unwrap_or_else(|_| "evm_hot_wallet=info,alloy=trace".into()), - ) + .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| "evm_hot_wallet=info".into())) .with(tracing_subscriber::fmt::layer()) .init(); @@ -43,6 +40,14 @@ async fn main() -> anyhow::Result<()> { config.block_offset_from_head ); tracing::info!("🌐 API Port: {}", config.port); + tracing::info!( + "🔐 Webhook JWT Auth: {}", + if config.webhook_jwt_token.is_some() { + "Enabled" + } else { + "Disabled" + } + ); let port = config.port; diff --git a/src/monitor.rs b/src/monitor.rs index 3c9a988..1f3d653 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -8,7 +8,8 @@ use tracing::{error, info, warn}; /// Information about a detected deposit for webhook notification struct DepositInfo<'a> { id: &'a str, - account_id: &'a str, + account_id: &'a str, // Polygon address + registration_id: &'a str, // Original id used when registering tx_hash: &'a str, amount: &'a str, token_type: &'a str, @@ -40,7 +41,7 @@ where // Use saturating_sub to prevent underflow if block_offset_from_head > latest_block let current_block = latest_block.saturating_sub(self.config.block_offset_from_head); - let last_processed = self.db.get_last_processed_block()?; + let last_processed = self.db.get_last_processed_block().await?; let start_block = if last_processed == 0 { current_block // Start from now if fresh @@ -75,7 +76,7 @@ where } async fn process_single_block(&self, block_num: u64) -> Result<()> { - info!("Processing block {}", block_num); + info!("🔍 Processing block {}", block_num); if let Some(block) = self .provider @@ -84,8 +85,6 @@ where { // Process native ETH transfers if let Some(txs) = block.transactions.as_transactions() { - info!("Transactions: {:?}", txs.len()); - for tx in txs { if let Some(to) = tx.to { let to_address_str = to.to_string(); @@ -100,27 +99,36 @@ where continue; } - if let Some(account_id) = self.db.get_account_by_address(&to_address_str)? { + if let Some(registration_id) = self + .db + .get_registration_id_by_address(&to_address_str) + .await? + { info!( - "Native ETH deposit detected! Tx: {:?}, Account: {}", - tx.hash, account_id + "Native ETH deposit detected! Tx: {:?}, Address: {}, Registration ID: {}", + tx.hash, to_address_str, registration_id ); // Only send webhook if this is a new deposit (not a duplicate) let tx_hash_str = tx.hash.to_string(); - let is_new_deposit = self.db.record_deposit( - &tx_hash_str, - &account_id, - &tx.value.to_string(), - )?; + let is_new_deposit = self + .db + .record_deposit( + &tx_hash_str, + ®istration_id, + &tx.value.to_string(), + ) + .await?; // Send webhook notification for deposit detection only if it's new if is_new_deposit { + let amount_str = tx.value.to_string(); let deposit_info = DepositInfo { id: &tx_hash_str, - account_id: &account_id, + account_id: &to_address_str, + registration_id: ®istration_id, tx_hash: &tx_hash_str, - amount: &tx.value.to_string(), + amount: &amount_str, token_type: "native", token_symbol: None, token_address: None, @@ -141,7 +149,8 @@ where self.process_erc20_transfers(block_num).await?; } - self.db.set_last_processed_block(block_num)?; + // info!("Processing D {}", block_num); + self.db.set_last_processed_block(block_num).await?; Ok(()) } @@ -158,8 +167,6 @@ where .to_block(block_num) .event_signature(transfer_signature); - info!("Filter: {:?}", filter); - let logs = self .get_logs_with_retry( &filter, @@ -168,12 +175,6 @@ where ) .await?; - info!( - "Found {} Transfer events in block {}", - logs.len(), - block_num - ); - for log in logs { // Decode Transfer event: topic[0] = signature, topic[1] = from, topic[2] = to if log.topics().len() >= 3 { @@ -194,22 +195,45 @@ where } // Check if this is one of our monitored addresses - if let Some(account_id) = self.db.get_account_by_address(&to_address_str)? { - // Decode the amount from data field - let amount = if !log.data().data.is_empty() { + if let Some(registration_id) = self + .db + .get_registration_id_by_address(&to_address_str) + .await? + { + // Decode the amount from data field (ABI-encoded uint256 is 32 bytes) + let amount = if log.data().data.len() >= 32 { + // Standard case: take first 32 bytes (ABI-encoded uint256) + let amount_bytes: [u8; 32] = log.data().data[..32] + .try_into() + .expect("slice length is 32"); + alloy::primitives::U256::from_be_bytes(amount_bytes) + } else if !log.data().data.is_empty() { + // Short data (non-standard, but handle gracefully) alloy::primitives::U256::from_be_slice(&log.data().data) } else { alloy::primitives::U256::ZERO }; - // log::info("Detected ERC20 deposit: Token: {}, To: {}, From: {}, Amount: {}", token_address, to_address_str, from_address_str, log.data().data); + info!( + "Detected ERC20 deposit: Token: {}, To: {}, From: {}, Amount: {}", + token_address, to_address_str, from_address_str, amount + ); // Fetch token metadata (symbol, decimals, name) let token_info = self.get_or_fetch_token_metadata(token_address).await?; + // Skip tokens with symbol longer than 5 characters + if token_info.symbol.len() > 5 { + info!( + "Skipping ERC20 deposit: token symbol '{}' exceeds 5 characters", + token_info.symbol + ); + continue; + } + info!( - "ERC20 deposit detected! Token: {} ({}), Amount: {}, Account: {}, Tx: {:?}", - token_info.symbol, token_address, amount, account_id, log.transaction_hash + "ERC20 deposit detected! Token: {} ({}), Amount: {}, Address: {}, Registration ID: {}, Tx: {:?}", + token_info.symbol, token_address, amount, to_address_str, registration_id, log.transaction_hash ); // Store ERC20 deposit @@ -219,14 +243,17 @@ where let deposit_id = format!("{}:{}", tx_hash_str, log_index); // Only send webhook if this is a new deposit (not a duplicate) - let is_new_deposit = self.db.record_erc20_deposit( - &tx_hash_str, - log_index, - &account_id, - &amount.to_string(), - &token_address.to_string(), - &token_info.symbol, - )?; + let is_new_deposit = self + .db + .record_erc20_deposit( + &tx_hash_str, + log_index, + ®istration_id, + &amount.to_string(), + &token_address.to_string(), + &token_info.symbol, + ) + .await?; // Send webhook notification for ERC20 deposit detection only if it's new if is_new_deposit { @@ -234,7 +261,8 @@ where let amount_str = amount.to_string(); let deposit_info = DepositInfo { id: &deposit_id, - account_id: &account_id, + account_id: &to_address_str, + registration_id: ®istration_id, tx_hash: &tx_hash_str, amount: &amount_str, token_type: "erc20", @@ -302,7 +330,9 @@ where let token_address_str = token_address.to_string(); // Check cache first - if let Some((symbol, decimals, name)) = self.db.get_token_metadata(&token_address_str)? { + if let Some((symbol, decimals, name)) = + self.db.get_token_metadata(&token_address_str).await? + { return Ok(TokenInfo { address: token_address_str, symbol, @@ -315,12 +345,14 @@ where match get_token_info(&self.provider, token_address).await { Ok(token_info) => { // Cache it - self.db.store_token_metadata( - &token_address_str, - &token_info.symbol, - token_info.decimals, - &token_info.name, - )?; + self.db + .store_token_metadata( + &token_address_str, + &token_info.symbol, + token_info.decimals, + &token_info.name, + ) + .await?; Ok(token_info) } Err(e) => { @@ -340,13 +372,13 @@ where } async fn send_deposit_detected_webhook(&self, info: &DepositInfo<'_>) -> Result<()> { - // Get the webhook URL for this account - let webhook_url = match self.db.get_webhook_url(info.account_id)? { - Some(url) => url, - None => { - error!("No webhook URL found for account: {}", info.account_id); - return Ok(()); - } + // Get the webhook URL for this account using registration_id + let Some(webhook_url) = self.db.get_webhook_url(info.registration_id).await? else { + error!( + "No webhook URL found for registration_id: {}", + info.registration_id + ); + return Ok(()); }; let client = reqwest::Client::new(); @@ -355,6 +387,7 @@ where "id": info.id, "event": "deposit_detected", "account_id": info.account_id, + "registration_id": info.registration_id, "tx_hash": info.tx_hash, "amount": info.amount, "token_type": info.token_type @@ -371,13 +404,21 @@ where payload["token_decimals"] = serde_json::json!(decimals); } - let res = client.post(&webhook_url).json(&payload).send().await; + let mut request = client.post(&webhook_url).json(&payload); + + // Add JWT authorization header if configured + if let Some(ref token) = self.config.webhook_jwt_token { + request = request.header("Authorization", format!("Bearer {}", token)); + } + + let res = request.send().await; match res { Ok(r) => info!( - "Deposit detected webhook sent to {}: status={}", + "Deposit detected webhook sent to {}: status={}, registration_id={}", webhook_url, - r.status() + r.status(), + info.registration_id ), Err(e) => error!( "Failed to send deposit detected webhook to {}: {:?}", diff --git a/src/sweeper.rs b/src/sweeper.rs index b76e391..f31ce72 100644 --- a/src/sweeper.rs +++ b/src/sweeper.rs @@ -1,6 +1,7 @@ use crate::{ config::Config, db::{Db, Erc20Deposit}, + faucet::Faucet, wallet::Wallet, }; use alloy::network::TransactionBuilder; @@ -9,7 +10,9 @@ use alloy::providers::Provider; use alloy::rpc::types::TransactionRequest; use alloy::sol_types::SolCall; use anyhow::Result; +use std::collections::HashSet; use std::str::FromStr; +use std::sync::Arc; use std::time::Duration; use tokio::time::sleep; use tracing::{error, info}; @@ -17,12 +20,14 @@ use tracing::{error, info}; /// Information about an ERC20 deposit for webhook notification struct Erc20WebhookInfo<'a> { id: &'a str, - account_id: &'a str, + account_id: &'a str, // Polygon address + registration_id: &'a str, // Original id used when registering deposit_key: &'a str, amount: &'a str, token_symbol: &'a str, token_address: &'a str, token_decimals: Option, + sweep_tx_hash: &'a str, // On-chain tx hash of the sweep (idempotency key for consumers) } pub struct Sweeper

{ @@ -30,6 +35,7 @@ pub struct Sweeper

{ db: Db, wallet: Wallet, provider: P, + faucet: Arc>, } use crate::traits::Service; @@ -50,6 +56,14 @@ where } } +/// After this many consecutive zero-balance checks, a deposit is assumed to have been +/// swept as part of a consolidated sweep and is marked as swept to avoid infinite retries. +const MAX_ZERO_BALANCE_RETRIES: u64 = 10; + +/// After this many consecutive sweep failures (e.g. "buffer overrun while deserializing"), +/// a deposit is marked as permanently failed to stop wasting RPC credits on deterministic errors. +const MAX_SWEEP_RETRIES: u64 = 5; + impl Sweeper> where T: alloy::transports::Transport + Clone, @@ -59,29 +73,32 @@ where db: Db, wallet: Wallet, provider: alloy::providers::RootProvider, + faucet: Arc>>, ) -> Self { Self { config, db, wallet, provider, + faucet, } } async fn process_deposits(&self) -> Result<()> { // Process native ETH deposits - let deposits = self.db.get_detected_deposits()?; + let deposits = self.db.get_detected_deposits().await?; - for (tx_hash, account_id, amount_str) in deposits { + for (tx_hash, registration_id, amount_str) in deposits { info!( - "Processing native ETH deposit: tx_hash={}, account={}, amount={}", - tx_hash, account_id, amount_str + "Processing native ETH deposit: tx_hash={}, registration_id={}, amount={}", + tx_hash, registration_id, amount_str ); - // Get account details to derive key + // Get account details to derive key (registration_id is the key in ACCOUNTS table) let (derivation_index, address_str, _webhook_url) = self .db - .get_account_by_id(&account_id)? + .get_account_by_id(®istration_id) + .await? .ok_or_else(|| anyhow::anyhow!("Account not found"))?; let signer = self.wallet.get_signer(derivation_index)?; @@ -92,26 +109,42 @@ where .wallet(wallet) .on_provider(&self.provider); - self.sweep_deposit( - &sweep_provider, - &address_str, - &tx_hash, - &account_id, - &amount_str, - ) - .await?; + match self + .sweep_deposit( + &sweep_provider, + &address_str, + &tx_hash, + ®istration_id, + &amount_str, + ) + .await + { + Ok(_) => info!("Successfully swept native ETH deposit: {}", tx_hash), + Err(e) => { + error!("Failed to sweep native ETH deposit {}: {:?}", tx_hash, e); + } + } } // Process ERC20 deposits - let erc20_deposits = self.db.get_detected_erc20_deposits()?; + let erc20_deposits = self.db.get_detected_erc20_deposits().await?; + + // Track (address, token) pairs already swept in this cycle to avoid redundant attempts. + // After sweeping the full token_balance for one deposit, all other deposits for the same + // address+token are already marked as swept by the bulk mark method. Any remaining ones + // would see zero balance and harmlessly skip, but we can avoid the RPC call entirely. + let mut swept_pairs: HashSet<(String, String)> = HashSet::new(); for deposit in erc20_deposits { + // deposit.account_id is actually the registration_id (original id from registration) + let registration_id = &deposit.account_id; + info!( - "Processing ERC20 deposit: key={}, token={} ({}), account={}, amount={}", + "Processing ERC20 deposit: key={}, token={} ({}), registration_id={}, amount={}", deposit.key, deposit.token_symbol, deposit.token_address, - deposit.account_id, + registration_id, deposit.amount ); @@ -120,16 +153,27 @@ where "Skipping ERC20 deposit token symbol for deposit: {}", deposit.key ); - self.db.mark_erc20_deposit_swept(&deposit.key)?; + self.db.mark_erc20_deposit_swept(&deposit.key).await?; continue; } - // Get account details to derive key + // Get account details to derive key (registration_id is the key in ACCOUNTS table) let (derivation_index, address_str, _webhook_url) = self .db - .get_account_by_id(&deposit.account_id)? + .get_account_by_id(registration_id) + .await? .ok_or_else(|| anyhow::anyhow!("Account not found"))?; + // Skip if we already swept this (address, token) pair in this cycle + let pair_key = (address_str.clone(), deposit.token_address.clone()); + if swept_pairs.contains(&pair_key) { + info!( + "Skipping ERC20 deposit {} - already swept address {} for token {} in this cycle", + deposit.key, address_str, deposit.token_symbol + ); + continue; + } + let signer = self.wallet.get_signer(derivation_index)?; info!("Signer address: {}", signer.address()); @@ -146,11 +190,36 @@ where .sweep_erc20_deposit(&sweep_provider, &address_str, &deposit) .await { - Ok(_) => info!("Successfully swept ERC20 deposit: {}", deposit.key), + Ok(_) => { + info!("Successfully swept ERC20 deposit: {}", deposit.key); + swept_pairs.insert(pair_key); + } Err(e) => { error!("Failed to sweep ERC20 deposit {}: {:?}", deposit.key, e); - // Don't return error - continue processing other deposits - // This deposit will be retried in the next sweep cycle + if let Ok(failures) = self.db.increment_sweep_failure_count(&deposit.key).await + { + if failures >= MAX_SWEEP_RETRIES { + let registration_id = &deposit.account_id; + match self + .db + .mark_erc20_deposits_failed_for_account_token( + registration_id, + &deposit.token_address, + ) + .await + { + Ok(failed_keys) => { + error!( + "Permanently marked {} deposit(s) as failed for account={}, token={} after {} attempts: {:?}", + failed_keys.len(), registration_id, deposit.token_symbol, failures, e + ); + } + Err(db_err) => { + error!("Failed to mark deposits as failed: {:?}", db_err); + } + } + } + } } } } @@ -163,7 +232,7 @@ where provider: &SP, from_address_str: &str, tx_hash: &str, - account_id: &str, + registration_id: &str, amount_str: &str, ) -> Result<()> where @@ -173,25 +242,74 @@ where let to_address = Address::from_str(&self.config.treasury_address)?; // Check balance again to be sure (and to calculate gas) - let balance = provider.get_balance(from_address).await?; + let mut balance = provider.get_balance(from_address).await?; - // Simple gas estimation/reservation (leaving some dust for gas) - let gas_price = provider.get_gas_price().await?; - let gas_limit = 21000; // Standard transfer - let gas_cost = U256::from(gas_limit) * U256::from(gas_price); + // Standard ETH transfer gas limit + let gas_limit: u128 = 21000; - if balance <= gas_cost { - info!("Balance too low to sweep: {} <= {}", balance, gas_cost); - return Ok(()); + // Get current fee estimates (EIP-1559 compatible) + let fee_estimate = provider.estimate_eip1559_fees(None).await?; + let max_fee_per_gas = fee_estimate.max_fee_per_gas; + + // Calculate gas cost with 50% buffer for price fluctuations + let gas_cost = U256::from(gas_limit) * U256::from(max_fee_per_gas); + let gas_cost_with_buffer = gas_cost + (gas_cost / U256::from(10)); + + info!( + "Gas estimation for native ETH transfer: gas_limit={}, max_fee_per_gas={}, gas_cost={} wei (with 50% buffer: {} wei)", + gas_limit, max_fee_per_gas, gas_cost, gas_cost_with_buffer + ); + + // If balance is too low to cover gas, try to fund via faucet + if balance <= gas_cost_with_buffer { + info!( + "Balance too low to sweep: {} <= {}. Attempting to fund via faucet...", + balance, gas_cost_with_buffer + ); + + // Fund the address via faucet + match self.faucet.fund_new_address(from_address_str).await { + Ok(tx_hash) => { + info!( + "Successfully funded address {} via faucet with tx: {}. Waiting for balance update...", + from_address_str, tx_hash + ); + + // Wait a bit for the transaction to be processed and balance to update + sleep(Duration::from_secs(2)).await; + + // Re-check the balance after funding + balance = provider.get_balance(from_address).await?; + info!( + "Updated balance after faucet funding: {} wei for address {}", + balance, from_address_str + ); + + // Final check - if still not enough, return error + if balance <= gas_cost_with_buffer { + return Err(anyhow::anyhow!( + "Still insufficient balance after faucet funding. Address: {}, Balance: {} wei, Gas cost: {} wei", + from_address_str, balance, gas_cost_with_buffer + )); + } + } + Err(e) => { + return Err(anyhow::anyhow!( + "Failed to fund address {} via faucet: {}", + from_address_str, + e + )); + } + } } - let value_to_send = balance - gas_cost; + // Use actual gas cost (without buffer) for value calculation to maximize sweep amount + let value_to_send = balance - gas_cost_with_buffer; let tx = TransactionRequest::default() .with_to(to_address) .with_value(value_to_send) - .with_gas_limit(gas_limit) - .with_gas_price(gas_price); + .with_gas_limit(gas_limit); let pending_tx = provider.send_transaction(tx).await?; let receipt = pending_tx.get_receipt().await?; @@ -199,11 +317,18 @@ where info!("Swept funds! Tx hash: {:?}", receipt.transaction_hash); // Update DB - self.db.mark_deposit_swept(tx_hash)?; + self.db.mark_deposit_swept(tx_hash).await?; // Send Webhook (for native deposits, id = tx_hash) - self.send_webhook(tx_hash, account_id, tx_hash, amount_str) - .await?; + // account_id = Polygon address, registration_id = original id from registration + self.send_webhook( + tx_hash, + from_address_str, + registration_id, + tx_hash, + amount_str, + ) + .await?; Ok(()) } @@ -221,59 +346,139 @@ where let to_address = Address::from_str(&self.config.treasury_address)?; let token_address = Address::from_str(&deposit.token_address)?; - // Check native balance first (need gas for ERC20 transfer) - let native_balance = provider.get_balance(from_address).await?; - - info!( - "Native balance: {} wei for address {}", - native_balance, from_address_str - ); + // Check token balance first + let token_balance = get_token_balance(&self.provider, token_address, from_address).await?; - if native_balance.is_zero() { + if deposit.token_symbol.len() > 5 { error!( - "Cannot sweep ERC20 tokens from {}: no native balance for gas. Address needs to be funded first.", - from_address_str + "Skipping ERC20 deposit token symbol '{}' exceeds 5 characters for deposit: {}", + deposit.token_symbol, deposit.key ); + self.db.mark_erc20_deposit_swept(&deposit.key).await?; + return Ok(()); + } - return Err(anyhow::anyhow!( - "Insufficient native balance for gas. Address: {}, Balance: 0", - from_address_str - )); + if token_balance.is_zero() { + let retry_count = self.db.increment_zero_balance_count(&deposit.key).await?; + if retry_count >= MAX_ZERO_BALANCE_RETRIES { + self.db.mark_erc20_deposit_swept(&deposit.key).await?; + info!( + "Marking deposit {} as swept after {} zero-balance retries (funds likely consolidated in a prior sweep)", + deposit.key, retry_count + ); + } else { + info!( + "ERC20 balance is zero for {} at {}, retry {}/{} (will retry next cycle)", + deposit.token_symbol, from_address_str, retry_count, MAX_ZERO_BALANCE_RETRIES + ); + } + return Ok(()); } - // Estimate gas cost - let gas_price = provider.get_gas_price().await?; - let gas_limit = 100000u128; - let estimated_gas_cost = U256::from(gas_limit) * U256::from(gas_price); + // Sweep the full on-chain token balance to ensure all funds are moved to treasury, + // regardless of how many individual deposits contributed to this balance. + let amount = token_balance; - if native_balance < estimated_gas_cost { - error!( - "Insufficient native balance for gas. Address: {}, Balance: {} wei, Estimated gas cost: {} wei", - from_address_str, native_balance, estimated_gas_cost - ); - return Err(anyhow::anyhow!( - "Insufficient native balance for gas. Need at least {} wei, but only have {} wei", - estimated_gas_cost, - native_balance - )); - } + // Build ERC20 transfer call data for gas estimation + let transfer_call = IERC20::transferCall { + to: to_address, + amount, + }; + + let call_data = transfer_call.abi_encode(); + + // Build transaction request for gas estimation + let tx_for_estimate = TransactionRequest::default() + .with_from(from_address) + .with_to(token_address) + .with_input(call_data.clone()); + + // Estimate actual gas needed for this specific transaction + let estimated_gas = provider.estimate_gas(&tx_for_estimate).await?; + + let gas_limit_with_buffer = estimated_gas + (estimated_gas / 10); + + // Get current fee estimates (EIP-1559 compatible) + let fee_estimate = provider.estimate_eip1559_fees(None).await?; + let max_fee_per_gas = fee_estimate.max_fee_per_gas; + + // Calculate worst-case gas cost with safety buffer + // Add extra 10% buffer on top for gas price fluctuations + let estimated_gas_cost = U256::from(gas_limit_with_buffer) * U256::from(max_fee_per_gas); + let estimated_gas_cost_with_buffer = + estimated_gas_cost + (estimated_gas_cost / U256::from(10)); info!( - "Native balance check passed: {} wei (gas estimate: {} wei)", - native_balance, estimated_gas_cost + "Gas estimation for ERC20 transfer: gas={}, max_fee_per_gas={}, estimated_cost={} wei (with 50% buffer: {} wei)", + gas_limit_with_buffer, max_fee_per_gas, estimated_gas_cost, estimated_gas_cost_with_buffer ); - // Check token balance - let token_balance = get_token_balance(&self.provider, token_address, from_address).await?; + // Check native balance (need gas for ERC20 transfer) + let mut native_balance = provider.get_balance(from_address).await?; - if token_balance.is_zero() { + info!( + "Native balance: {} wei for address {}", + native_balance, from_address_str + ); + + // If insufficient balance for gas, try to fund via faucet + if native_balance < estimated_gas_cost_with_buffer { info!( - "ERC20 balance is zero for {} token at {}, skipping sweep", - deposit.token_symbol, from_address_str + "Insufficient native balance for gas. Address: {}, Balance: {} wei, Estimated gas cost: {} wei. Attempting to fund via faucet...", + from_address_str, native_balance, estimated_gas_cost_with_buffer ); - return Ok(()); + + // Fund the address via faucet + match self.faucet.fund_new_address(from_address_str).await { + Ok(tx_hash) => { + info!( + "Successfully funded address {} via faucet with tx: {}. Waiting for balance update...", + from_address_str, tx_hash + ); + + // Wait a bit for the transaction to be processed and balance to update + sleep(Duration::from_secs(2)).await; + + // Re-check the balance after funding + native_balance = provider.get_balance(from_address).await?; + info!( + "Updated native balance after faucet funding: {} wei for address {}", + native_balance, from_address_str + ); + + // Final check - if still not enough, error out + if native_balance < estimated_gas_cost_with_buffer { + error!( + "Still insufficient balance after faucet funding. Address: {}, Balance: {} wei, Required: {} wei", + from_address_str, native_balance, estimated_gas_cost_with_buffer + ); + return Err(anyhow::anyhow!( + "Insufficient native balance for gas even after faucet funding. Need at least {} wei, but only have {} wei", + estimated_gas_cost_with_buffer, + native_balance + )); + } + } + Err(e) => { + error!( + "Failed to fund address {} via faucet: {:?}", + from_address_str, e + ); + return Err(anyhow::anyhow!( + "Insufficient native balance for gas and faucet funding failed. Address: {}, Balance: {} wei, Error: {}", + from_address_str, + native_balance, + e + )); + } + } } + info!( + "Native balance check passed: {} wei (gas estimate with buffer: {} wei)", + native_balance, estimated_gas_cost_with_buffer + ); + info!( "Sweeping {} {} tokens (raw: {}) from {} to {} (native balance: {} wei)", token_balance, @@ -284,45 +489,67 @@ where native_balance ); - // Build ERC20 transfer call data - let transfer_call = IERC20::transferCall { - to: to_address, - amount: token_balance, - }; - - let call_data = transfer_call.abi_encode(); - + // Build final transaction with estimated gas limit let tx = TransactionRequest::default() .with_to(token_address) .with_input(call_data) - .with_gas_limit(gas_limit); + .with_gas_limit(gas_limit_with_buffer); + + info!("++++++++++++++++"); + info!("Transaction request: {:?}", tx); + info!("++++++++++++++++"); let pending_tx = provider.send_transaction(tx).await?; + info!("++++++++++++++++"); + info!("Pending transaction: {:?}", pending_tx.tx_hash()); + info!("++++++++++++++++"); let receipt = pending_tx.get_receipt().await?; + info!("++++++++++++++++"); + info!("Receipt: {:?}", receipt.transaction_hash); + info!("++++++++++++++++"); + + let sweep_tx_hash = receipt.transaction_hash.to_string(); + + // Mark ALL detected deposits for this account+token as swept (consolidates multi-deposit sweeps) + let registration_id = &deposit.account_id; + let marked_keys = self + .db + .mark_erc20_deposits_swept_for_account_token(registration_id, &deposit.token_address) + .await?; + + // Store sweep tx hash for all marked deposits (audit trail + webhook idempotency key) + self.db + .set_sweep_tx_hash_for_keys(&marked_keys, &sweep_tx_hash) + .await?; info!( - "Swept ERC20 tokens! Tx hash: {:?}", - receipt.transaction_hash + "Marked {} ERC20 deposit(s) as swept for account={}, token={}, sweep_tx={}: {:?}", + marked_keys.len(), + registration_id, + deposit.token_symbol, + sweep_tx_hash, + marked_keys ); - // Update DB - self.db.mark_erc20_deposit_swept(&deposit.key)?; - // Fetch token decimals from DB let token_decimals = self .db - .get_token_metadata(&deposit.token_address)? + .get_token_metadata(&deposit.token_address) + .await? .map(|(_, decimals, _)| decimals); - // Send Webhook (for ERC20 deposits, id = deposit.key which is tx_hash:log_index) + // Send Webhook with the actual swept amount and the sweep tx hash for consumer deduplication + let swept_amount_str = amount.to_string(); let webhook_info = Erc20WebhookInfo { id: &deposit.key, - account_id: &deposit.account_id, + account_id: from_address_str, + registration_id, deposit_key: &deposit.key, - amount: &deposit.amount, + amount: &swept_amount_str, token_symbol: &deposit.token_symbol, token_address: &deposit.token_address, token_decimals, + sweep_tx_hash: &sweep_tx_hash, }; self.send_erc20_webhook(&webhook_info).await?; @@ -333,16 +560,17 @@ where &self, id: &str, account_id: &str, + registration_id: &str, tx_hash: &str, amount: &str, ) -> Result<()> { - // Get the webhook URL for this account - let webhook_url = match self.db.get_webhook_url(account_id)? { - Some(url) => url, - None => { - error!("No webhook URL found for account: {}", account_id); - return Ok(()); - } + // Get the webhook URL using registration_id (the key in ACCOUNTS table) + let Some(webhook_url) = self.db.get_webhook_url(registration_id).await? else { + error!( + "No webhook URL found for registration_id: {}", + registration_id + ); + return Ok(()); }; let client = reqwest::Client::new(); @@ -350,15 +578,33 @@ where "id": id, "event": "deposit_swept", "account_id": account_id, + "registration_id": registration_id, "original_tx_hash": tx_hash, "amount": amount, "token_type": "native" }); - let res = client.post(&webhook_url).json(&payload).send().await; + info!("++++++++++++++++"); + info!("Webhook URL: {}", webhook_url); + info!("Sending webhook: {:?}", payload); + info!("++++++++++++++++"); + + let mut request = client.post(&webhook_url).json(&payload); + + // Add JWT authorization header if configured + if let Some(ref token) = self.config.webhook_jwt_token { + request = request.header("Authorization", format!("Bearer {}", token)); + } + + let res = request.send().await; match res { - Ok(r) => info!("Webhook sent to {}: status={}", webhook_url, r.status()), + Ok(r) => info!( + "Webhook sent to {}: status={}, registration_id={}", + webhook_url, + r.status(), + registration_id + ), Err(e) => error!("Failed to send webhook to {}: {:?}", webhook_url, e), } @@ -366,13 +612,13 @@ where } async fn send_erc20_webhook(&self, info: &Erc20WebhookInfo<'_>) -> Result<()> { - // Get the webhook URL for this account - let webhook_url = match self.db.get_webhook_url(info.account_id)? { - Some(url) => url, - None => { - error!("No webhook URL found for account: {}", info.account_id); - return Ok(()); - } + // Get the webhook URL using registration_id (the key in ACCOUNTS table) + let Some(webhook_url) = self.db.get_webhook_url(info.registration_id).await? else { + error!( + "No webhook URL found for registration_id: {}", + info.registration_id + ); + return Ok(()); }; let client = reqwest::Client::new(); @@ -380,25 +626,40 @@ where "id": info.id, "event": "deposit_swept", "account_id": info.account_id, - "original_tx_hash": info.deposit_key, + "registration_id": info.registration_id, + "original_tx_hash": info.deposit_key.split(':').nth(0).unwrap(), "amount": info.amount, "token_type": "erc20", "token_symbol": info.token_symbol, - "token_address": info.token_address + "token_address": info.token_address, + "sweep_tx_hash": info.sweep_tx_hash }); + info!("++++++++++++++++"); + info!("Webhook URL: {}", webhook_url); + info!("Sending webhook: {:?}", payload); + info!("++++++++++++++++"); + // Add decimals if available if let Some(decimals) = info.token_decimals { payload["token_decimals"] = serde_json::json!(decimals); } - let res = client.post(&webhook_url).json(&payload).send().await; + let mut request = client.post(&webhook_url).json(&payload); + + // Add JWT authorization header if configured + if let Some(ref token) = self.config.webhook_jwt_token { + request = request.header("Authorization", format!("Bearer {}", token)); + } + + let res = request.send().await; match res { Ok(r) => info!( - "ERC20 Webhook sent to {}: status={}", + "ERC20 Webhook sent to {}: status={}, registration_id={}", webhook_url, - r.status() + r.status(), + info.registration_id ), Err(e) => error!("Failed to send ERC20 webhook to {}: {:?}", webhook_url, e), } diff --git a/src/tests.rs b/src/tests.rs index 7b81c87..8eef90a 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -1,11 +1,13 @@ use crate::config::{Config, ProviderUrl}; use crate::db::Db; +use crate::faucet::Faucet; use crate::monitor::Monitor; use crate::sweeper::Sweeper; use crate::wallet::Wallet; use crate::{HotWalletService, VerifyTransferRequest, VerifyTransferResponse}; use alloy::providers::ProviderBuilder; use serde_json::json; +use std::sync::Arc; use tempfile::NamedTempFile; use wiremock::matchers::method; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -26,11 +28,11 @@ fn test_wallet_derivation() { ); } -#[test] -fn test_db_operations() { +#[tokio::test] +async fn test_db_operations() { let tmp_file = NamedTempFile::new().unwrap(); let db_path = tmp_file.path().to_str().unwrap(); - let db = Db::new(db_path).unwrap(); + let db = Db::new(db_path).await.unwrap(); // Test Account Registration let id = "user_1"; @@ -38,31 +40,32 @@ fn test_db_operations() { let address = "0x123"; db.register_account(id, index, address, "https://webhook.example.com") + .await .unwrap(); - let fetched_addr = db.get_account_by_id(id).unwrap().unwrap().1; + let fetched_addr = db.get_account_by_id(id).await.unwrap().unwrap().1; assert_eq!(fetched_addr, address); - let fetched_id = db.get_account_by_address(address).unwrap().unwrap(); + let fetched_id = db.get_account_by_address(address).await.unwrap().unwrap(); assert_eq!(fetched_id, id); // Test Index Increment - let next_idx = db.get_next_derivation_index().unwrap(); + let next_idx = db.get_next_derivation_index().await.unwrap(); assert_eq!(next_idx, 1); // Test Deposits let tx_hash = "0xabc"; let amount = "100"; - db.record_deposit(tx_hash, id, amount).unwrap(); + db.record_deposit(tx_hash, id, amount).await.unwrap(); - let deposits = db.get_detected_deposits().unwrap(); + let deposits = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits.len(), 1); assert_eq!(deposits[0].0, tx_hash); assert_eq!(deposits[0].2, amount); // Test Sweep Mark - db.mark_deposit_swept(tx_hash).unwrap(); - let deposits_after = db.get_detected_deposits().unwrap(); + db.mark_deposit_swept(tx_hash).await.unwrap(); + let deposits_after = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits_after.len(), 0); } @@ -72,7 +75,7 @@ fn test_db_operations() { async fn test_monitor_creation_with_http_provider() { // Test that Monitor can be created with an HTTP provider let db_file = NamedTempFile::new().unwrap(); - let db = Db::new(db_file.path().to_str().unwrap()).unwrap(); + let db = Db::new(db_file.path().to_str().unwrap()).await.unwrap(); let config = Config { database_url: db_file.path().to_str().unwrap().to_string(), @@ -87,6 +90,7 @@ async fn test_monitor_creation_with_http_provider() { faucet_address: "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266".to_string(), get_logs_max_retries: 30, get_logs_delay_ms: 50, + webhook_jwt_token: None, }; // Create provider and monitor (no actual connection needed for this test) @@ -94,11 +98,11 @@ async fn test_monitor_creation_with_http_provider() { let _monitor = Monitor::new(config, db.clone(), provider); } -#[test] -fn test_monitor_db_operations() { +#[tokio::test] +async fn test_monitor_db_operations() { // Test Monitor's interaction with DB for deposit tracking let db_file = NamedTempFile::new().unwrap(); - let db = Db::new(db_file.path().to_str().unwrap()).unwrap(); + let db = Db::new(db_file.path().to_str().unwrap()).await.unwrap(); let wallet = Wallet::new("test test test test test test test test test test test junk".to_string()); @@ -106,32 +110,34 @@ fn test_monitor_db_operations() { // Register account db.register_account("test_user", 0, &user_address, "https://webhook.example.com") + .await .unwrap(); // Verify no deposits initially - let deposits_before = db.get_detected_deposits().unwrap(); + let deposits_before = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits_before.len(), 0); // Simulate Monitor recording a deposit db.record_deposit("0xtxhash", "test_user", "1000000000000000000") + .await .unwrap(); - let deposits_after = db.get_detected_deposits().unwrap(); + let deposits_after = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits_after.len(), 1); assert_eq!(deposits_after[0].0, "0xtxhash"); assert_eq!(deposits_after[0].1, "test_user"); assert_eq!(deposits_after[0].2, "1000000000000000000"); // Test block tracking - db.set_last_processed_block(100).unwrap(); - assert_eq!(db.get_last_processed_block().unwrap(), 100); + db.set_last_processed_block(100).await.unwrap(); + assert_eq!(db.get_last_processed_block().await.unwrap(), 100); } -#[test] -fn test_monitor_address_lookup() { +#[tokio::test] +async fn test_monitor_address_lookup() { // Test that only registered addresses are trackable let db_file = NamedTempFile::new().unwrap(); - let db = Db::new(db_file.path().to_str().unwrap()).unwrap(); + let db = Db::new(db_file.path().to_str().unwrap()).await.unwrap(); let wallet = Wallet::new("test test test test test test test test test test test junk".to_string()); @@ -140,15 +146,16 @@ fn test_monitor_address_lookup() { // Register only addr1 db.register_account("user1", 0, &addr1, "https://webhook.example.com") + .await .unwrap(); // Check addr1 is registered - let account = db.get_account_by_address(&addr1).unwrap(); + let account = db.get_account_by_address(&addr1).await.unwrap(); assert!(account.is_some()); assert_eq!(account.unwrap(), "user1"); // Check addr2 is not registered - let account2 = db.get_account_by_address(&addr2).unwrap(); + let account2 = db.get_account_by_address(&addr2).await.unwrap(); assert!(account2.is_none()); } @@ -158,7 +165,7 @@ fn test_monitor_address_lookup() { async fn test_sweeper_creation() { // Test that Sweeper can be created with proper dependencies let db_file = NamedTempFile::new().unwrap(); - let db = Db::new(db_file.path().to_str().unwrap()).unwrap(); + let db = Db::new(db_file.path().to_str().unwrap()).await.unwrap(); let config = Config { database_url: db_file.path().to_str().unwrap().to_string(), @@ -173,19 +180,28 @@ async fn test_sweeper_creation() { faucet_address: "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266".to_string(), get_logs_max_retries: 30, get_logs_delay_ms: 50, + webhook_jwt_token: None, }; let wallet = Wallet::new(config.mnemonic.clone()); let provider = ProviderBuilder::new().on_http("http://localhost:8545".parse().unwrap()); + let faucet = Arc::new( + Faucet::new( + config.faucet_mnemonic.clone(), + provider.clone(), + &config.existential_deposit, + ) + .unwrap(), + ); - Sweeper::new(config, db, wallet, provider); + Sweeper::new(config, db, wallet, provider, faucet); } -#[test] -fn test_sweeper_deposit_workflow() { +#[tokio::test] +async fn test_sweeper_deposit_workflow() { // Test the full deposit workflow through the DB let db_file = NamedTempFile::new().unwrap(); - let db = Db::new(db_file.path().to_str().unwrap()).unwrap(); + let db = Db::new(db_file.path().to_str().unwrap()).await.unwrap(); let wallet = Wallet::new("test test test test test test test test test test test junk".to_string()); @@ -193,23 +209,25 @@ fn test_sweeper_deposit_workflow() { // Register account and create a deposit db.register_account("test_user", 0, &user_address, "https://webhook.example.com") + .await .unwrap(); db.record_deposit("0xtx123", "test_user", "1000000000000000000") + .await .unwrap(); // Verify deposit exists - let deposits_before = db.get_detected_deposits().unwrap(); + let deposits_before = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits_before.len(), 1); assert_eq!(deposits_before[0].0, "0xtx123"); assert_eq!(deposits_before[0].1, "test_user"); // Simulate sweep completion - db.mark_deposit_swept("0xtx123").unwrap(); - let deposits_after = db.get_detected_deposits().unwrap(); + db.mark_deposit_swept("0xtx123").await.unwrap(); + let deposits_after = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits_after.len(), 0); // Verify the account details are correct for deriving keys - let account_details = db.get_account_by_id("test_user").unwrap().unwrap(); + let account_details = db.get_account_by_id("test_user").await.unwrap().unwrap(); assert_eq!(account_details.0, 0); // derivation index assert_eq!(account_details.1, user_address); } @@ -238,11 +256,11 @@ fn test_sweeper_wallet_integration() { assert_eq!(signer1.address(), addr1); } -#[test] -fn test_sweeper_multiple_deposits() { +#[tokio::test] +async fn test_sweeper_multiple_deposits() { // Test handling multiple deposits for different users let db_file = NamedTempFile::new().unwrap(); - let db = Db::new(db_file.path().to_str().unwrap()).unwrap(); + let db = Db::new(db_file.path().to_str().unwrap()).await.unwrap(); let wallet = Wallet::new("test test test test test test test test test test test junk".to_string()); @@ -253,35 +271,41 @@ fn test_sweeper_multiple_deposits() { let addr2 = wallet.derive_address(2).unwrap().to_string(); db.register_account("user_0", 0, &addr0, "https://webhook.example.com") + .await .unwrap(); db.register_account("user_1", 1, &addr1, "https://webhook.example.com") + .await .unwrap(); db.register_account("user_2", 2, &addr2, "https://webhook.example.com") + .await .unwrap(); // Record deposits for each db.record_deposit("0xtx1", "user_0", "1000000000000000000") + .await .unwrap(); db.record_deposit("0xtx2", "user_1", "2000000000000000000") + .await .unwrap(); db.record_deposit("0xtx3", "user_2", "3000000000000000000") + .await .unwrap(); // Verify all deposits are tracked - let deposits = db.get_detected_deposits().unwrap(); + let deposits = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits.len(), 3); // Process one deposit at a time - db.mark_deposit_swept("0xtx1").unwrap(); - let deposits_after_1 = db.get_detected_deposits().unwrap(); + db.mark_deposit_swept("0xtx1").await.unwrap(); + let deposits_after_1 = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits_after_1.len(), 2); - db.mark_deposit_swept("0xtx2").unwrap(); - let deposits_after_2 = db.get_detected_deposits().unwrap(); + db.mark_deposit_swept("0xtx2").await.unwrap(); + let deposits_after_2 = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits_after_2.len(), 1); - db.mark_deposit_swept("0xtx3").unwrap(); - let deposits_after_3 = db.get_detected_deposits().unwrap(); + db.mark_deposit_swept("0xtx3").await.unwrap(); + let deposits_after_3 = db.get_detected_deposits().await.unwrap(); assert_eq!(deposits_after_3.len(), 0); } @@ -321,6 +345,7 @@ async fn test_verify_native_transfer_success() { faucet_address: "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266".to_string(), get_logs_max_retries: 30, get_logs_delay_ms: 50, + webhook_jwt_token: None, }; let to_address = "0x742d35Cc6634C0532925a3b844Bc454e4438f44e"; @@ -436,6 +461,7 @@ async fn test_verify_native_transfer_amount_mismatch() { faucet_address: "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266".to_string(), get_logs_max_retries: 30, get_logs_delay_ms: 50, + webhook_jwt_token: None, }; let to_address = "0x742d35Cc6634C0532925a3b844Bc454e4438f44e"; @@ -543,6 +569,7 @@ async fn test_verify_erc20_transfer_success() { faucet_address: "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266".to_string(), get_logs_max_retries: 30, get_logs_delay_ms: 50, + webhook_jwt_token: None, }; let to_address = "0x742d35Cc6634C0532925a3b844Bc454e4438f44e"; @@ -670,6 +697,7 @@ async fn test_verify_erc20_transfer_symbol_mismatch() { faucet_address: "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266".to_string(), get_logs_max_retries: 30, get_logs_delay_ms: 50, + webhook_jwt_token: None, }; let to_address = "0x742d35Cc6634C0532925a3b844Bc454e4438f44e"; @@ -764,6 +792,7 @@ async fn test_verify_transfer_reverted_transaction() { faucet_address: "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266".to_string(), get_logs_max_retries: 30, get_logs_delay_ms: 50, + webhook_jwt_token: None, }; let to_address = "0x742d35Cc6634C0532925a3b844Bc454e4438f44e"; @@ -845,3 +874,54 @@ async fn test_verify_transfer_reverted_transaction() { } } } + +// ========== Sweep Failure Tracking Tests ========== + +#[tokio::test] +async fn test_erc20_deposit_failure_tracking() { + let db_file = NamedTempFile::new().unwrap(); + let db = Db::new(db_file.path().to_str().unwrap()).await.unwrap(); + + db.record_erc20_deposit("0xabc", 1, "user_1", "1000000", "0xtoken", "USDC") + .await + .unwrap(); + + assert_eq!(db.get_detected_erc20_deposits().await.unwrap().len(), 1); + + for i in 1..=5 { + let count = db.increment_sweep_failure_count("0xabc:1").await.unwrap(); + assert_eq!(count, i); + } + + db.mark_erc20_deposit_failed("0xabc:1").await.unwrap(); + + assert_eq!(db.get_detected_erc20_deposits().await.unwrap().len(), 0); +} + +#[tokio::test] +async fn test_erc20_bulk_mark_failed_for_account_token() { + let db_file = NamedTempFile::new().unwrap(); + let db = Db::new(db_file.path().to_str().unwrap()).await.unwrap(); + + db.record_erc20_deposit("0xaaa", 1, "user_1", "1000000", "0xtoken_a", "USDC") + .await + .unwrap(); + db.record_erc20_deposit("0xbbb", 2, "user_1", "2000000", "0xtoken_a", "USDC") + .await + .unwrap(); + db.record_erc20_deposit("0xccc", 3, "user_1", "3000000", "0xtoken_b", "USDT") + .await + .unwrap(); + + assert_eq!(db.get_detected_erc20_deposits().await.unwrap().len(), 3); + + let failed = db + .mark_erc20_deposits_failed_for_account_token("user_1", "0xtoken_a") + .await + .unwrap(); + assert_eq!(failed.len(), 2); + + let remaining = db.get_detected_erc20_deposits().await.unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].token_symbol, "USDT"); +}