From 868d7439ec371bb5c64422386613391d5a58ea25 Mon Sep 17 00:00:00 2001 From: Sachin Kamath Date: Mon, 11 Nov 2024 20:34:52 +0700 Subject: [PATCH] observatory 0.1 --- common/nyxd-scraper/src/storage/manager.rs | 52 ++++++++ common/nyxd-scraper/src/storage/mod.rs | 12 ++ ...c41d20896db543eed74a6f320c041bcbb723d.json | 16 --- ...2f1482df00de2dc1d2b4debbb2e12553d997b.json | 34 ------ ...50ad69686cf6a109e37d6c3f0623c3e9f91d0.json | 32 ----- nym-data-observatory/Cargo.toml | 8 +- nym-data-observatory/Dockerfile | 27 ----- nym-data-observatory/build.rs | 65 +++++----- .../migrations/001_price_data.sql | 7 ++ .../migrations/002_payment_transactions.sql | 10 ++ nym-data-observatory/pg_up.sh | 13 -- .../src/background_task/mod.rs | 61 ---------- nym-data-observatory/src/chain_scraper/mod.rs | 22 ++++ nym-data-observatory/src/db/mod.rs | 16 ++- nym-data-observatory/src/db/models.rs | 47 +++++--- nym-data-observatory/src/db/queries/joke.rs | 39 ------ nym-data-observatory/src/db/queries/mod.rs | 7 +- .../src/db/queries/payments.rs | 41 +++++++ nym-data-observatory/src/db/queries/price.rs | 46 +++++++ nym-data-observatory/src/http/api/jokes.rs | 78 ------------ nym-data-observatory/src/http/api/mod.rs | 7 +- nym-data-observatory/src/http/api/price.rs | 27 +++++ nym-data-observatory/src/http/error.rs | 7 -- nym-data-observatory/src/main.rs | 78 +++++++----- .../src/payment_listener/mod.rs | 114 ++++++++++++++++++ nym-data-observatory/src/price_scraper/mod.rs | 55 +++++++++ 26 files changed, 518 insertions(+), 403 deletions(-) delete mode 100644 nym-data-observatory/.sqlx/query-249faa11b88b749f50342bb5c9cc41d20896db543eed74a6f320c041bcbb723d.json delete mode 100644 nym-data-observatory/.sqlx/query-aff7fbd06728004d2f2226d20c32f1482df00de2dc1d2b4debbb2e12553d997b.json delete mode 100644 nym-data-observatory/.sqlx/query-e53f479f8cead3dc8aa1875e5d450ad69686cf6a109e37d6c3f0623c3e9f91d0.json delete mode 100644 nym-data-observatory/Dockerfile create mode 100644 nym-data-observatory/migrations/001_price_data.sql create mode 100644 nym-data-observatory/migrations/002_payment_transactions.sql delete mode 100755 nym-data-observatory/pg_up.sh delete mode 100644 nym-data-observatory/src/background_task/mod.rs create mode 100644 nym-data-observatory/src/chain_scraper/mod.rs delete mode 100644 nym-data-observatory/src/db/queries/joke.rs create mode 100644 nym-data-observatory/src/db/queries/payments.rs create mode 100644 nym-data-observatory/src/db/queries/price.rs delete mode 100644 nym-data-observatory/src/http/api/jokes.rs create mode 100644 nym-data-observatory/src/http/api/price.rs create mode 100644 nym-data-observatory/src/payment_listener/mod.rs create mode 100644 nym-data-observatory/src/price_scraper/mod.rs diff --git a/common/nyxd-scraper/src/storage/manager.rs b/common/nyxd-scraper/src/storage/manager.rs index fb40a065b8..5475f5d355 100644 --- a/common/nyxd-scraper/src/storage/manager.rs +++ b/common/nyxd-scraper/src/storage/manager.rs @@ -237,6 +237,58 @@ impl StorageManager { Ok(-1) } } + + #[allow(dead_code)] + pub async fn get_transactions_after_height( + &self, + min_height: i64, + message_type: Option<&str>, + ) -> Result, sqlx::Error> { + match message_type { + Some(msg_type) => { + sqlx::query_as!( + TransactionWithBlock, + r#" + SELECT t.hash, t.height, t.memo, t.raw_log + FROM message m + JOIN "transaction" t ON m.transaction_hash = t.hash + JOIN block b ON t.height = b.height + WHERE t.height > ? + AND m.type = ? + ORDER BY t.height ASC + "#, + min_height, + msg_type + ) + .fetch_all(&self.connection_pool) + .await + } + None => { + sqlx::query_as!( + TransactionWithBlock, + r#" + SELECT t.hash, t.height, t.memo, t.raw_log + FROM message m + JOIN "transaction" t ON m.transaction_hash = t.hash + JOIN block b ON t.height = b.height + WHERE t.height > ? + ORDER BY t.height ASC + "#, + min_height + ) + .fetch_all(&self.connection_pool) + .await + } + } + } +} + +#[derive(Debug, sqlx::FromRow)] +pub struct TransactionWithBlock { + pub hash: String, + pub height: i64, + pub memo: Option, + pub raw_log: Option, } // make those generic over executor so that they could be performed over connection pool and a tx diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index f2d0921ee5..39a2842fa8 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -13,6 +13,7 @@ use crate::{ models::{CommitSignature, Validator}, }, }; +use manager::TransactionWithBlock; use sqlx::{ sqlite::{SqliteAutoVacuum, SqliteSynchronous}, types::time::OffsetDateTime, @@ -214,6 +215,17 @@ impl ScraperStorage { pub async fn get_pruned_height(&self) -> Result { Ok(self.manager.get_pruned_height().await?) } + + pub async fn get_transactions_after_height( + &self, + min_height: i64, + message_type: Option<&str>, + ) -> Result, ScraperError> { + Ok(self + .manager + .get_transactions_after_height(min_height, message_type) + .await?) + } } pub async fn persist_block( diff --git a/nym-data-observatory/.sqlx/query-249faa11b88b749f50342bb5c9cc41d20896db543eed74a6f320c041bcbb723d.json b/nym-data-observatory/.sqlx/query-249faa11b88b749f50342bb5c9cc41d20896db543eed74a6f320c041bcbb723d.json deleted file mode 100644 index a42f7b7e32..0000000000 --- a/nym-data-observatory/.sqlx/query-249faa11b88b749f50342bb5c9cc41d20896db543eed74a6f320c041bcbb723d.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO responses\n (joke_id, joke, date_created)\n VALUES\n ($1, $2, $3)\n ON CONFLICT(joke_id) DO UPDATE SET\n joke=excluded.joke,\n date_created=excluded.date_created;", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Text", - "Int4" - ] - }, - "nullable": [] - }, - "hash": "249faa11b88b749f50342bb5c9cc41d20896db543eed74a6f320c041bcbb723d" -} diff --git a/nym-data-observatory/.sqlx/query-aff7fbd06728004d2f2226d20c32f1482df00de2dc1d2b4debbb2e12553d997b.json b/nym-data-observatory/.sqlx/query-aff7fbd06728004d2f2226d20c32f1482df00de2dc1d2b4debbb2e12553d997b.json deleted file mode 100644 index 88e19ca1b9..0000000000 --- a/nym-data-observatory/.sqlx/query-aff7fbd06728004d2f2226d20c32f1482df00de2dc1d2b4debbb2e12553d997b.json +++ /dev/null @@ -1,34 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT joke_id, joke, date_created FROM responses WHERE joke_id = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "joke_id", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "joke", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "date_created", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false, - false, - false - ] - }, - "hash": "aff7fbd06728004d2f2226d20c32f1482df00de2dc1d2b4debbb2e12553d997b" -} diff --git a/nym-data-observatory/.sqlx/query-e53f479f8cead3dc8aa1875e5d450ad69686cf6a109e37d6c3f0623c3e9f91d0.json b/nym-data-observatory/.sqlx/query-e53f479f8cead3dc8aa1875e5d450ad69686cf6a109e37d6c3f0623c3e9f91d0.json deleted file mode 100644 index e770f72289..0000000000 --- a/nym-data-observatory/.sqlx/query-e53f479f8cead3dc8aa1875e5d450ad69686cf6a109e37d6c3f0623c3e9f91d0.json +++ /dev/null @@ -1,32 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT joke_id, joke, date_created FROM responses", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "joke_id", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "joke", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "date_created", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - false, - false, - false - ] - }, - "hash": "e53f479f8cead3dc8aa1875e5d450ad69686cf6a109e37d6c3f0623c3e9f91d0" -} diff --git a/nym-data-observatory/Cargo.toml b/nym-data-observatory/Cargo.toml index 220449bfb1..0270057477 100644 --- a/nym-data-observatory/Cargo.toml +++ b/nym-data-observatory/Cargo.toml @@ -24,9 +24,13 @@ nym-task = { path = "../common/task" } nym-node-requests = { path = "../nym-node/nym-node-requests", features = [ "openapi", ] } +nyxd-scraper = {path = "../common/nyxd-scraper"} +reqwest = {workspace= true, features = ["rustls-tls"]} +rocket = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } -sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres"] } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "time"] } +time = {version = "0.3.36"} tokio = { workspace = true, features = ["process", "rt-multi-thread"] } tokio-util = { workspace = true } tracing = { workspace = true } @@ -40,4 +44,4 @@ utoipauto = { workspace = true } [build-dependencies] anyhow = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } -sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres"] } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite"] } diff --git a/nym-data-observatory/Dockerfile b/nym-data-observatory/Dockerfile deleted file mode 100644 index d36a910faf..0000000000 --- a/nym-data-observatory/Dockerfile +++ /dev/null @@ -1,27 +0,0 @@ -FROM rust:latest AS builder - -COPY ./ /usr/src/nym -WORKDIR /usr/src/nym/nym-data-observatory - -RUN cargo build --release - -#------------------------------------------------------------------- -# The following environment variables are required at runtime: -# -# NYM_DATA_OBSERVATORY_CONNECTION_URL -# -# And optionally: -# -# NYM_DATA_OBSERVATORY_HTTP_PORT -# -# see https://github.com/nymtech/nym/blob/develop/nym-data-observatory/src/main.rs for details -#------------------------------------------------------------------- - -FROM ubuntu:24.04 - -RUN apt update && apt install -yy curl ca-certificates - -WORKDIR /nym - -COPY --from=builder /usr/src/nym/target/release/nym-data-observatory ./ -ENTRYPOINT [ "/nym/nym-data-observatory" ] diff --git a/nym-data-observatory/build.rs b/nym-data-observatory/build.rs index ba795cf1d0..faedd8b4c7 100644 --- a/nym-data-observatory/build.rs +++ b/nym-data-observatory/build.rs @@ -1,58 +1,51 @@ use anyhow::Result; -use sqlx::{Connection, PgConnection}; +use sqlx::{sqlite::SqliteConnectOptions, Connection, SqliteConnection}; use std::io::Write; -use std::{collections::HashMap, fs::File}; +use std::{collections::HashMap, fs::File, path::PathBuf, str::FromStr}; -const POSTGRES_USER: &str = "nym"; -const POSTGRES_PASSWORD: &str = "password123"; -const POSTGRES_DB: &str = "data_obs_db"; - -/// if schema changes, rerun `cargo sqlx prepare` with a running DB -/// https://github.com/launchbadge/sqlx/blob/main/sqlx-cli/README.md#enable-building-in-offline-mode-with-query #[tokio::main] async fn main() -> Result<()> { - let db_url = - format!("postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@localhost:5432/{POSTGRES_DB}"); + let db_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("data_observatory.sqlite"); - export_db_variables(&db_url)?; - // if a live DB is reachable, use that - if PgConnection::connect(&db_url).await.is_ok() { - println!("cargo::rustc-env=SQLX_OFFLINE=false"); - run_migrations(&db_url).await?; - } else { - // by default, run in offline mode - println!("cargo::rustc-env=SQLX_OFFLINE=true"); + // Create the database directory if it doesn't exist + if let Some(parent) = db_path.parent() { + std::fs::create_dir_all(parent)?; } - rerun_if_changed(); + let db_url = format!("sqlite:{}", db_path.display()); + + // Ensure database file is created with proper permissions + let connect_options = SqliteConnectOptions::from_str(&db_url)? + .create_if_missing(true) + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .foreign_keys(true); + + // Create initial connection to ensure database exists + let mut conn = SqliteConnection::connect_with(&connect_options).await?; + + export_db_variables(&db_url)?; + println!("cargo:rustc-env=SQLX_OFFLINE=false"); + + // Run migrations after ensuring database exists + sqlx::migrate!("./migrations").run(&mut conn).await?; + + // Add rerun-if-changed directives + println!("cargo:rerun-if-changed=migrations"); + println!("cargo:rerun-if-changed=build.rs"); + println!("cargo:rerun-if-changed=src"); Ok(()) } fn export_db_variables(db_url: &str) -> Result<()> { let mut map = HashMap::new(); - map.insert("POSTGRES_USER", POSTGRES_USER); - map.insert("POSTGRES_PASSWORD", POSTGRES_PASSWORD); - map.insert("POSTGRES_DB", POSTGRES_DB); map.insert("DATABASE_URL", db_url); let mut file = File::create(".env")?; for (var, value) in map.iter() { - println!("cargo::rustc-env={}={}", var, value); - writeln!(file, "{}={}", var, value).expect("Failed to write to dotenv file"); + println!("cargo:rustc-env={}={}", var, value); + writeln!(file, "{}={}", var, value)?; } Ok(()) } - -async fn run_migrations(db_url: &str) -> Result<()> { - let mut conn = PgConnection::connect(db_url).await?; - sqlx::migrate!("./migrations").run(&mut conn).await?; - - Ok(()) -} - -fn rerun_if_changed() { - println!("cargo::rerun-if-changed=migrations"); - println!("cargo::rerun-if-changed=src/db/queries"); -} diff --git a/nym-data-observatory/migrations/001_price_data.sql b/nym-data-observatory/migrations/001_price_data.sql new file mode 100644 index 0000000000..f50f6299b0 --- /dev/null +++ b/nym-data-observatory/migrations/001_price_data.sql @@ -0,0 +1,7 @@ +CREATE TABLE price_history ( + timestamp INTEGER PRIMARY KEY, + chf REAL NOT NULL, + usd REAL NOT NULL, + eur REAL NOT NULL, + btc REAL NOT NULL +); \ No newline at end of file diff --git a/nym-data-observatory/migrations/002_payment_transactions.sql b/nym-data-observatory/migrations/002_payment_transactions.sql new file mode 100644 index 0000000000..7ba3b0c8e2 --- /dev/null +++ b/nym-data-observatory/migrations/002_payment_transactions.sql @@ -0,0 +1,10 @@ +CREATE TABLE transactions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + transaction_hash TEXT NOT NULL UNIQUE, + sender_address TEXT NOT NULL, + receiver_address TEXT NOT NULL, + amount REAL NOT NULL, + timestamp INTEGER NOT NULL, + height INTEGER NOT NULL, + memo TEXT +); diff --git a/nym-data-observatory/pg_up.sh b/nym-data-observatory/pg_up.sh deleted file mode 100755 index ca4e1f7e51..0000000000 --- a/nym-data-observatory/pg_up.sh +++ /dev/null @@ -1,13 +0,0 @@ -#!/bin/bash - -# .env is generated in build.rs -source .env - -# Launching a container in such a way that it's destroyed after you detach from the terminal: -docker compose up - -# docker exec -it nym-data-observatory-pg /bin/bash -# psql -U youruser -d yourdb - -echo "Tearing down containers to have a clean slate" -docker compose down -v diff --git a/nym-data-observatory/src/background_task/mod.rs b/nym-data-observatory/src/background_task/mod.rs deleted file mode 100644 index 8799b818b4..0000000000 --- a/nym-data-observatory/src/background_task/mod.rs +++ /dev/null @@ -1,61 +0,0 @@ -use core::str; -use serde::Deserialize; -use tokio::process::Command; -use tokio::task::JoinHandle; -use tokio::time::Duration; - -use crate::db::{self, DbPool}; - -const REFRESH_DELAY: Duration = Duration::from_secs(15); -const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60 * 2); - -pub(crate) async fn spawn_in_background(db_pool: DbPool) -> JoinHandle<()> { - loop { - tracing::info!("Running in a loop 🏃"); - - if let Err(e) = some_network_action(&db_pool).await { - tracing::error!( - "❌ Run failed: {e}, retrying in {}s...", - FAILURE_RETRY_DELAY.as_secs() - ); - tokio::time::sleep(FAILURE_RETRY_DELAY).await; - } else { - tracing::info!( - "✅ Run successful, sleeping for {}s...", - REFRESH_DELAY.as_secs() - ); - tokio::time::sleep(REFRESH_DELAY).await; - } - } -} - -#[derive(Deserialize, Debug)] -pub(crate) struct Response { - #[serde(rename(deserialize = "id"))] - pub(crate) joke_id: String, - pub(crate) joke: String, - #[serde(rename(deserialize = "status"))] - pub(crate) _status: u16, -} - -async fn some_network_action(pool: &DbPool) -> anyhow::Result<()> { - // for demonstration purposes only. You should use reqwest if you need it - let output = Command::new("curl") - .arg("-H") - .arg("Accept: application/json") - .arg("https://icanhazdadjoke.com/") - .output() - .await?; - - if !output.status.success() { - anyhow::bail!("Curl command failed with status: {}", output.status); - } - - let response_str = str::from_utf8(&output.stdout)?; - let joke_response: Response = serde_json::from_str(response_str)?; - - tracing::info!("{:?}", joke_response.joke); - db::queries::insert_joke(pool, joke_response.into()).await?; - - Ok(()) -} diff --git a/nym-data-observatory/src/chain_scraper/mod.rs b/nym-data-observatory/src/chain_scraper/mod.rs new file mode 100644 index 0000000000..157dbd28bb --- /dev/null +++ b/nym-data-observatory/src/chain_scraper/mod.rs @@ -0,0 +1,22 @@ +use nyxd_scraper::{storage::ScraperStorage, Config, NyxdScraper, PruningOptions}; + +pub(crate) async fn run_chain_scraper() -> anyhow::Result { + let websocket_url = + std::env::var("NYXD_WEBSOCKET_URL").expect("NYXD_WEBSOCKET_URL not defined"); + + let rpc_url = std::env::var("NYXD_RPC_URL").expect("NYXD_RPC_URL not defined"); + let websocket_url = reqwest::Url::parse(&websocket_url)?; + let rpc_url = reqwest::Url::parse(&rpc_url)?; + + let scraper = NyxdScraper::builder(Config { + websocket_url, + rpc_url, + database_path: "chain_history.sqlite".into(), + pruning_options: PruningOptions::nothing(), + store_precommits: false, + }); + + let storage = scraper.build_and_start().await?; + + Ok(storage.storage) +} diff --git a/nym-data-observatory/src/db/mod.rs b/nym-data-observatory/src/db/mod.rs index 6e8938a578..74339ca575 100644 --- a/nym-data-observatory/src/db/mod.rs +++ b/nym-data-observatory/src/db/mod.rs @@ -1,13 +1,16 @@ use anyhow::{anyhow, Result}; -use sqlx::{migrate::Migrator, postgres::PgConnectOptions, ConnectOptions, PgPool}; +use sqlx::{migrate::Migrator, sqlite::SqliteConnectOptions, SqlitePool}; use std::str::FromStr; pub(crate) mod models; -pub(crate) mod queries; +pub(crate) mod queries { + pub mod payments; + pub mod price; +} static MIGRATOR: Migrator = sqlx::migrate!("./migrations"); -pub(crate) type DbPool = PgPool; +pub(crate) type DbPool = SqlitePool; pub(crate) struct Storage { pool: DbPool, @@ -16,13 +19,16 @@ pub(crate) struct Storage { impl Storage { pub async fn init(connection_url: String) -> Result { let connect_options = - PgConnectOptions::from_str(&connection_url)?.disable_statement_logging(); + SqliteConnectOptions::from_str(&connection_url)?.create_if_missing(true); let pool = DbPool::connect_with(connect_options) .await .map_err(|err| anyhow!("Failed to connect to {}: {}", &connection_url, err))?; - MIGRATOR.run(&pool).await?; + MIGRATOR + .run(&pool) + .await + .map_err(|err| anyhow!("Failed to run migrations: {}", err))?; Ok(Storage { pool }) } diff --git a/nym-data-observatory/src/db/models.rs b/nym-data-observatory/src/db/models.rs index e1e9682235..24d04de5f1 100644 --- a/nym-data-observatory/src/db/models.rs +++ b/nym-data-observatory/src/db/models.rs @@ -1,22 +1,41 @@ use serde::{Deserialize, Serialize}; use utoipa::ToSchema; -use crate::background_task::Response; +#[derive(Clone, Deserialize, Debug)] +pub(crate) struct CurrencyPrices { + pub(crate) chf: f32, + pub(crate) usd: f32, + pub(crate) eur: f32, + pub(crate) btc: f32, +} + +// Struct to hold Coingecko response +#[derive(Clone, Deserialize, Debug, ToSchema)] +pub(crate) struct CoingeckoPriceResponse { + pub(crate) nym: CurrencyPrices, +} + +#[derive(Clone, Deserialize, Debug, ToSchema)] +pub(crate) struct PriceRecord { + pub(crate) timestamp: i64, + pub(crate) nym: CurrencyPrices, +} #[derive(Serialize, Deserialize, Debug, ToSchema)] -pub(crate) struct JokeDto { - pub(crate) joke_id: String, - pub(crate) joke: String, - pub(crate) date_created: i32, +pub(crate) struct PriceHistory { + pub(crate) timestamp: i64, + pub(crate) chf: f32, + pub(crate) usd: f32, + pub(crate) eur: f32, + pub(crate) btc: f32, } -impl From for JokeDto { - fn from(value: Response) -> Self { - Self { - joke_id: value.joke_id, - joke: value.joke, - // casting not smart, can implicitly panic, don't do this in prod - date_created: chrono::offset::Utc::now().timestamp() as i32, - } - } +#[derive(Serialize, Deserialize, Debug, ToSchema)] +pub(crate) struct PaymentTransaction { + pub(crate) transaction_hash: String, + pub(crate) sender_address: String, + pub(crate) receiver_address: String, + pub(crate) amount: f64, + pub(crate) timestamp: i64, + pub(crate) height: i64, } diff --git a/nym-data-observatory/src/db/queries/joke.rs b/nym-data-observatory/src/db/queries/joke.rs deleted file mode 100644 index 649564a86c..0000000000 --- a/nym-data-observatory/src/db/queries/joke.rs +++ /dev/null @@ -1,39 +0,0 @@ -use crate::db::{models::JokeDto, DbPool}; - -pub(crate) async fn insert_joke(pool: &DbPool, joke: JokeDto) -> anyhow::Result<()> { - let mut conn = pool.acquire().await?; - sqlx::query!( - "INSERT INTO responses - (joke_id, joke, date_created) - VALUES - ($1, $2, $3) - ON CONFLICT(joke_id) DO UPDATE SET - joke=excluded.joke, - date_created=excluded.date_created;", - joke.joke_id, - joke.joke, - joke.date_created as i32, - ) - .execute(&mut *conn) - .await?; - - Ok(()) -} - -pub(crate) async fn select_joke_by_id(pool: &DbPool, joke_id: &str) -> anyhow::Result { - sqlx::query_as!( - JokeDto, - "SELECT joke_id, joke, date_created FROM responses WHERE joke_id = $1", - joke_id - ) - .fetch_one(pool) - .await - .map_err(anyhow::Error::from) -} - -pub(crate) async fn select_all(pool: &DbPool) -> anyhow::Result> { - sqlx::query_as!(JokeDto, "SELECT joke_id, joke, date_created FROM responses",) - .fetch_all(pool) - .await - .map_err(anyhow::Error::from) -} diff --git a/nym-data-observatory/src/db/queries/mod.rs b/nym-data-observatory/src/db/queries/mod.rs index b5723730b4..36867a6e7d 100644 --- a/nym-data-observatory/src/db/queries/mod.rs +++ b/nym-data-observatory/src/db/queries/mod.rs @@ -1,5 +1,6 @@ -// group queries in files by theme -mod joke; +mod payments; +mod price; // re-exporting allows us to access all queries via `queries::bla`` -pub(crate) use joke::{insert_joke, select_all, select_joke_by_id}; +pub(crate) use payments::{get_last_checked_height, insert_payment}; +pub(crate) use price::{get_latest_price, insert_nym_prices}; diff --git a/nym-data-observatory/src/db/queries/payments.rs b/nym-data-observatory/src/db/queries/payments.rs new file mode 100644 index 0000000000..cb9fae1f7f --- /dev/null +++ b/nym-data-observatory/src/db/queries/payments.rs @@ -0,0 +1,41 @@ +use crate::db::DbPool; +use anyhow::Result; + +pub async fn get_last_checked_height(pool: &DbPool) -> Result { + let result = sqlx::query_scalar!("SELECT MAX(height) FROM transactions") + .fetch_one(pool) + .await?; + Ok(result.unwrap_or(0)) +} + +pub async fn insert_payment( + pool: &DbPool, + transaction_hash: String, + sender_address: String, + receiver_address: String, + amount: f64, + height: i64, + memo: Option, +) -> Result<()> { + let timestamp = chrono::Utc::now().timestamp(); + + sqlx::query!( + r#" + INSERT INTO transactions ( + transaction_hash, sender_address, receiver_address, + amount, height, timestamp, memo + ) VALUES (?, ?, ?, ?, ?, ?, ?) + "#, + transaction_hash, + sender_address, + receiver_address, + amount, + height, + timestamp, + memo, + ) + .execute(pool) + .await?; + + Ok(()) +} diff --git a/nym-data-observatory/src/db/queries/price.rs b/nym-data-observatory/src/db/queries/price.rs new file mode 100644 index 0000000000..7c69863744 --- /dev/null +++ b/nym-data-observatory/src/db/queries/price.rs @@ -0,0 +1,46 @@ +use crate::db::models::{PriceHistory, PriceRecord}; +use crate::db::DbPool; + +pub(crate) async fn insert_nym_prices( + pool: &DbPool, + price_data: PriceRecord, +) -> anyhow::Result<()> { + let mut conn = pool.acquire().await?; + let timestamp = price_data.timestamp; + sqlx::query!( + "INSERT INTO price_history + (timestamp, chf, usd, eur, btc) + VALUES + ($1, $2, $3, $4, $5) + ON CONFLICT(timestamp) DO UPDATE SET + chf=excluded.chf, + usd=excluded.usd, + eur=excluded.eur, + btc=excluded.btc;", + timestamp, + price_data.nym.chf, + price_data.nym.usd, + price_data.nym.eur, + price_data.nym.btc, + ) + .execute(&mut *conn) + .await?; + + Ok(()) +} + +pub(crate) async fn get_latest_price(pool: &DbPool) -> anyhow::Result { + let result = sqlx::query!( + "SELECT timestamp, chf, usd, eur, btc FROM price_history ORDER BY timestamp DESC LIMIT 1;" + ) + .fetch_one(pool) + .await?; + + Ok(PriceHistory { + timestamp: result.timestamp, + chf: result.chf as f32, + usd: result.usd as f32, + eur: result.eur as f32, + btc: result.btc as f32, + }) +} diff --git a/nym-data-observatory/src/http/api/jokes.rs b/nym-data-observatory/src/http/api/jokes.rs deleted file mode 100644 index 6d429798e6..0000000000 --- a/nym-data-observatory/src/http/api/jokes.rs +++ /dev/null @@ -1,78 +0,0 @@ -use axum::{ - extract::{Path, State}, - Json, Router, -}; -use serde::Deserialize; -use utoipa::IntoParams; - -use crate::{ - db::{ - models::JokeDto, - queries::{self, select_joke_by_id}, - }, - http::{ - error::{Error, HttpResult}, - state::AppState, - }, -}; - -pub(crate) fn routes() -> Router { - Router::new() - .route("/", axum::routing::get(jokes)) - .route("/:joke_id", axum::routing::get(joke_by_id)) - .route("/fetch_another", axum::routing::get(fetch_another)) -} - -#[utoipa::path( - tag = "Dad Jokes", - get, - path = "/v1/jokes", - responses( - (status = 200, body = Vec) - ) -)] -async fn jokes(State(state): State) -> HttpResult>> { - queries::select_all(state.db_pool()) - .await - .map(Json::from) - .map_err(|_| Error::internal()) -} - -#[derive(Deserialize, IntoParams)] -#[into_params(parameter_in = Path)] -struct JokeIdParam { - joke_id: String, -} - -#[utoipa::path( - tag = "Dad Jokes", - get, - params( - JokeIdParam - ), - path = "/v1/jokes/{joke_id}", - responses( - (status = 200, body = JokeDto) - ) -)] -async fn joke_by_id( - Path(JokeIdParam { joke_id }): Path, - State(state): State, -) -> HttpResult> { - select_joke_by_id(state.db_pool(), &joke_id) - .await - .map(Json::from) - .map_err(|_| Error::not_found(joke_id)) -} - -#[utoipa::path( - tag = "Dad Jokes", - get, - path = "/v1/jokes/fetch_another", - responses( - (status = 200, body = String) - ) -)] -async fn fetch_another(State(_state): State) -> HttpResult> { - Ok(Json(String::from("Done boss, check the DB"))) -} diff --git a/nym-data-observatory/src/http/api/mod.rs b/nym-data-observatory/src/http/api/mod.rs index eeac99445d..4ca198a729 100644 --- a/nym-data-observatory/src/http/api/mod.rs +++ b/nym-data-observatory/src/http/api/mod.rs @@ -7,8 +7,8 @@ use utoipa_swagger_ui::SwaggerUi; use crate::http::{api_docs, server::HttpServer, state::AppState}; -pub(crate) mod jokes; pub(crate) mod mixnodes; +pub(crate) mod price; pub(crate) struct RouterBuilder { unfinished_router: Router, @@ -28,8 +28,9 @@ impl RouterBuilder { .nest( "/v1", Router::new() - .nest("/jokes", jokes::routes()) - .nest("/mixnodes", mixnodes::routes()), + //.nest("/jokes", jokes::routes()) + .nest("/mixnodes", mixnodes::routes()) + .nest("/price", price::routes()), ); Self { diff --git a/nym-data-observatory/src/http/api/price.rs b/nym-data-observatory/src/http/api/price.rs new file mode 100644 index 0000000000..1a239fcf54 --- /dev/null +++ b/nym-data-observatory/src/http/api/price.rs @@ -0,0 +1,27 @@ +use crate::db::models::PriceHistory; +use crate::db::queries::price::get_latest_price; +use crate::http::error::Error; +use crate::http::error::HttpResult; +use crate::http::state::AppState; +use axum::{extract::State, Json, Router}; + +pub(crate) fn routes() -> Router { + Router::new().route("/", axum::routing::get(price)) +} + +#[utoipa::path( + tag = "Nym Price", + get, + path = "/v1/price", + responses( + (status = 200, body = String) + ) +)] + +/// Fetch the latest price cached by the data observatory +async fn price(State(state): State) -> HttpResult> { + get_latest_price(state.db_pool()) + .await + .map(Json::from) + .map_err(|_| Error::internal()) +} diff --git a/nym-data-observatory/src/http/error.rs b/nym-data-observatory/src/http/error.rs index 30ba8a3c9d..fa6b274b69 100644 --- a/nym-data-observatory/src/http/error.rs +++ b/nym-data-observatory/src/http/error.rs @@ -6,13 +6,6 @@ pub(crate) struct Error { } impl Error { - pub(crate) fn not_found(message: String) -> Self { - Self { - message, - status: axum::http::StatusCode::NOT_FOUND, - } - } - pub(crate) fn internal() -> Self { Self { message: String::from("Internal server error"), diff --git a/nym-data-observatory/src/main.rs b/nym-data-observatory/src/main.rs index da8f53d4f9..4373f8850f 100644 --- a/nym-data-observatory/src/main.rs +++ b/nym-data-observatory/src/main.rs @@ -1,11 +1,15 @@ +use chain_scraper::run_chain_scraper; use clap::Parser; use nym_network_defaults::setup_env; use nym_task::signal::wait_for_signal; +use tokio::join; -mod background_task; +mod chain_scraper; mod db; mod http; mod logging; +mod payment_listener; +mod price_scraper; #[derive(Parser, Debug)] #[command(version, about, long_about = None)] @@ -18,25 +22,13 @@ struct Args { #[arg(short, long, default_value = None, env = "NYM_DATA_OBSERVATORY_ENV_FILE")] env_file: Option, - /// DB connection username - #[arg(long, default_value = None, env = "NYM_DATA_OBSERVATORY_CONNECTION_USERNAME")] - connection_username: String, - - /// DB connection password - #[arg(long, default_value = None, env = "NYM_DATA_OBSERVATORY_CONNECTION_PASSWORD")] - connection_password: String, - - /// DB connection host - #[arg(long, default_value = None, env = "NYM_DATA_OBSERVATORY_CONNECTION_HOST")] - connection_host: String, - - /// DB connection port - #[arg(long, default_value = None, env = "NYM_DATA_OBSERVATORY_CONNECTION_PORT")] - connection_port: String, - - /// DB connection database name - #[arg(long, default_value = None, env = "NYM_DATA_OBSERVATORY_CONNECTION_DB")] - connection_db: String, + /// SQLite database file path + #[arg( + long, + default_value = "data_observatory.sqlite", + env = "NYM_DATA_OBSERVATORY_DB_PATH" + )] + db_path: String, } #[tokio::main] @@ -44,30 +36,50 @@ async fn main() -> anyhow::Result<()> { logging::setup_tracing_logger(); let args = Args::parse(); - setup_env(args.env_file); // Defaults to mainnet if empty - let connection_url = format!( - "postgres://{}:{}@{}:{}/{}", - args.connection_username, - args.connection_password, - args.connection_host, - args.connection_port, - args.connection_db - ); + let db_path = args.db_path; + // Ensure parent directory exists + if let Some(parent) = std::path::Path::new(&db_path).parent() { + std::fs::create_dir_all(parent)?; + } + let connection_url = format!("sqlite://{}?mode=rwc", db_path); let storage = db::Storage::init(connection_url).await?; - let db_pool = storage.pool_owned().await; - tokio::spawn(async move { - background_task::spawn_in_background(db_pool).await; - tracing::info!("Started task"); + let observatory_pool = storage.pool_owned().await; + + // Spawn the chain scraper and get its storage + + // Spawn the payment listener task + let payment_listener_handle = tokio::spawn({ + let obs_pool = observatory_pool.clone(); + let chain_storage = run_chain_scraper().await?; + + async move { + if let Err(e) = payment_listener::run_payment_listener(obs_pool, chain_storage).await { + tracing::error!("Payment listener error: {}", e); + } + Ok::<_, anyhow::Error>(()) + } + }); + + // Clone pool for each task that needs it + //let background_pool = db_pool.clone(); + + let price_scraper_handle = tokio::spawn(async move { + price_scraper::run_price_scraper(&observatory_pool).await; }); let shutdown_handles = http::server::start_http_api(storage.pool_owned().await, args.http_port) .await .expect("Failed to start server"); + tracing::info!("Started HTTP server on port {}", args.http_port); + // Wait for the short-lived tasks to complete + let _ = join!(price_scraper_handle, payment_listener_handle); + + // Wait for a signal to terminate the long-running task wait_for_signal().await; if let Err(err) = shutdown_handles.shutdown().await { diff --git a/nym-data-observatory/src/payment_listener/mod.rs b/nym-data-observatory/src/payment_listener/mod.rs new file mode 100644 index 0000000000..4158e65a0d --- /dev/null +++ b/nym-data-observatory/src/payment_listener/mod.rs @@ -0,0 +1,114 @@ +use crate::db::queries; +use nyxd_scraper::storage::ScraperStorage; +use reqwest::Client; +use serde_json::{json, Value}; +use sqlx::SqlitePool; +use std::env; +use tokio::time::{self, Duration}; + +#[derive(Debug)] +struct TransferEvent { + recipient: String, + sender: String, + amount: String, +} + +pub(crate) async fn run_payment_listener( + observatory_pool: SqlitePool, + chain_storage: ScraperStorage, +) -> anyhow::Result<()> { + let payment_receive_address = env::var("PAYMENT_RECEIVE_ADDRESS").map_err(|_| { + anyhow::anyhow!("Environment variable `PAYMENT_RECEIVE_ADDRESS` not defined") + })?; + let webhook_url = env::var("WEBHOOK_URL") + .map_err(|_| anyhow::anyhow!("Environment variable `WEBHOOK_URL` not defined"))?; + + let client = Client::new(); + loop { + let last_checked_height = + queries::payments::get_last_checked_height(&observatory_pool).await?; + tracing::info!("Last checked height: {}", last_checked_height); + + let transactions = chain_storage + .get_transactions_after_height( + last_checked_height, + Some("/cosmos.bank.v1beta1.MsgSend"), + ) + .await?; + + for tx in transactions { + println!("Processing transaction: {}", tx.hash); + if let Some(raw_log) = tx.raw_log.as_deref() { + if let Some(transfer) = parse_transfer_from_raw_log(raw_log)? { + if transfer.recipient == payment_receive_address { + let amount: f64 = parse_unym_amount(&transfer.amount)?; + + queries::payments::insert_payment( + &observatory_pool, + tx.hash.clone(), + transfer.sender.clone(), + transfer.recipient.clone(), + amount, + tx.height, + tx.memo.clone(), + ) + .await?; + + let webhook_data = json!({ + "transaction_hash": tx.hash, + "sender_address": transfer.sender, + "receiver_address": transfer.recipient, + "amount": amount, + "height": tx.height, + "memo": tx.memo, + }); + let _ = client.post(&webhook_url).json(&webhook_data).send().await; + } + } + } + } + + time::sleep(Duration::from_secs(10)).await; + } +} + +fn parse_transfer_from_raw_log(raw_log: &str) -> anyhow::Result> { + let log_value: Value = serde_json::from_str(raw_log)?; + + if let Some(events) = log_value[0]["events"].as_array() { + if let Some(transfer_event) = events.iter().find(|e| e["type"] == "transfer") { + if let Some(attrs) = transfer_event["attributes"].as_array() { + let mut transfer = TransferEvent { + recipient: String::new(), + sender: String::new(), + amount: String::new(), + }; + + for attr in attrs { + match attr["key"].as_str() { + Some("recipient") => { + transfer.recipient = attr["value"].as_str().unwrap_or("").to_string() + } + Some("sender") => { + transfer.sender = attr["value"].as_str().unwrap_or("").to_string() + } + Some("amount") => { + transfer.amount = attr["value"].as_str().unwrap_or("").to_string() + } + _ => continue, + } + } + + return Ok(Some(transfer)); + } + } + } + + Ok(None) +} + +fn parse_unym_amount(amount: &str) -> anyhow::Result { + let amount = amount.trim_end_matches("unym"); + let parsed: f64 = amount.parse()?; + Ok(parsed / 1_000_000.0) +} diff --git a/nym-data-observatory/src/price_scraper/mod.rs b/nym-data-observatory/src/price_scraper/mod.rs new file mode 100644 index 0000000000..938d292b3e --- /dev/null +++ b/nym-data-observatory/src/price_scraper/mod.rs @@ -0,0 +1,55 @@ +use crate::db::{ + models::{CoingeckoPriceResponse, PriceRecord}, + queries::price::insert_nym_prices, +}; +use core::str; +use tokio::task::JoinHandle; +use tokio::time::Duration; + +use crate::db::DbPool; + +const REFRESH_DELAY: Duration = Duration::from_secs(300); +const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60 * 2); +const COINGECKO_API_URL: &str = + "https://api.coingecko.com/api/v3/simple/price?ids=nym&vs_currencies=chf,usd,eur,btc"; + +pub(crate) async fn run_price_scraper(db_pool: &DbPool) -> JoinHandle<()> { + loop { + tracing::info!("Running in a loop 🏃"); + if let Err(e) = get_coingecko_prices(db_pool).await { + tracing::error!("❌ Failed to get CoinGecko prices: {e}"); + tracing::info!("Retrying in {}s...", FAILURE_RETRY_DELAY.as_secs()); + tokio::time::sleep(FAILURE_RETRY_DELAY).await; + } else { + tracing::info!("✅ Successfully fetched CoinGecko prices"); + tokio::time::sleep(REFRESH_DELAY).await; + } + } +} + +async fn get_coingecko_prices(pool: &DbPool) -> anyhow::Result<()> { + tracing::info!("💰 Fetching CoinGecko prices from {}", COINGECKO_API_URL); + + let response = reqwest::get(COINGECKO_API_URL) + .await? + .json::() + .await; + + tracing::info!("Got response {:?}", response); + match response { + Ok(resp) => { + let price_record = PriceRecord { + timestamp: time::OffsetDateTime::now_utc().unix_timestamp(), + nym: resp.nym, + }; + + insert_nym_prices(pool, price_record).await?; + } + Err(e) => { + //tracing::info!("💰 CoinGecko price response: {:?}", response); + tracing::error!("Error sending request: {}", e); + } + } + + Ok(()) +}