GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add the data observatory

Browse Source
This commit is contained in:
Mark Sinclair
2025-07-22 00:47:14 +01:00
parent 188fb5b970
commit b7c99f802d
58 changed files with 2864 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+339 -310
View File
File diff suppressed because it is too large Load Diff
Cargo.toml
+1
View File
@@ -126,6 +126,7 @@ members = [
"nym-credential-proxy/nym-credential-proxy",
"nym-credential-proxy/nym-credential-proxy-requests",
"nym-credential-proxy/vpn-api-lib-wasm",
"nym-data-observatory",
"nym-ip-packet-client",
"nym-network-monitor",
"nym-node",
common/nyxd-scraper-psql/sql_migrations/01_metadata.sql → common/nyxd-scraper-psql/sql_migrations/0001_metadata.sql
+1 -1
View File
@@ -6,5 +6,5 @@
CREATE TABLE METADATA
(
id INTEGER PRIMARY KEY CHECK (id = 0),
last_processed_height INTEGER NOT NULL
last_processed_height BIGINT NOT NULL
);
common/nyxd-scraper-psql/sql_migrations/02_cosmos.sql → common/nyxd-scraper-psql/sql_migrations/0002_cosmos.sql
+16 -12
View File
@@ -44,7 +44,7 @@ CREATE TABLE transaction
success BOOLEAN NOT NULL,
/* Body */
messages JSON NOT NULL DEFAULT '[]'::JSON,
messages JSONB NOT NULL DEFAULT '[]'::JSONB,
memo TEXT,
signatures TEXT[] NOT NULL,
@@ -63,31 +63,34 @@ CREATE TABLE transaction
CREATE INDEX transaction_hash_index ON transaction (hash);
CREATE INDEX transaction_height_index ON transaction (height);
CREATE TABLE message_type
CREATE TYPE COIN AS
(
type TEXT NOT NULL UNIQUE,
module TEXT NOT NULL,
label TEXT NOT NULL,
height BIGINT NOT NULL
denom TEXT,
amount TEXT
);
CREATE INDEX message_type_module_index ON message_type (module);
CREATE INDEX message_type_type_index ON message_type (type);
CREATE TABLE message
(
transaction_hash TEXT NOT NULL,
index BIGINT NOT NULL,
type TEXT NOT NULL REFERENCES message_type (type),
value JSON NOT NULL,
type TEXT NOT NULL,
value JSONB NOT NULL,
involved_accounts_addresses TEXT[] NOT NULL,
height BIGINT NOT NULL,
funds COIN[] DEFAULT '{}',
wasm_sender TEXT,
wasm_contract_address TEXT,
wasm_message_type TEXT,
FOREIGN KEY (transaction_hash) REFERENCES transaction (hash),
CONSTRAINT unique_message_per_tx UNIQUE (transaction_hash, index)
);
CREATE INDEX message_transaction_hash_index ON message (transaction_hash);
CREATE INDEX message_type_index ON message (type);
CREATE INDEX message_involved_accounts_index ON message USING GIN (involved_accounts_addresses);
CREATE INDEX message_wasm_contract_message_type_index ON message (wasm_message_type);
/**
* This function is used to find all the utils that involve any of the given addresses and have
@@ -124,4 +127,5 @@ $$ LANGUAGE sql STABLE;
CREATE TABLE pruning
(
last_pruned_height BIGINT NOT NULL
);
);
common/nyxd-scraper-psql/src/storage/manager.rs
+59 -22
View File
@@ -1,10 +1,11 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::models::Coin;
use crate::storage::models::{CommitSignature, Validator};
use nyxd_scraper_shared::storage::helpers::log_db_operation_time;
use sqlx::types::time::PrimitiveDateTime;
use sqlx::types::{Json, JsonValue};
use sqlx::types::JsonValue;
use sqlx::{Executor, Postgres};
use tokio::time::Instant;
use tracing::{instrument, trace};
@@ -200,7 +201,8 @@ impl StorageManager {
log_db_operation_time("get_last_processed_height", start);
if let Some(row) = maybe_record {
Ok(row.last_processed_height as i64)
#[allow(clippy::useless_conversion)]
Ok(row.last_processed_height.into())
} else {
Ok(-1)
}
@@ -392,6 +394,7 @@ where
Ok(())
}
#[allow(clippy::too_many_arguments)]
#[instrument(skip(executor))]
pub(crate) async fn insert_message<'a, E>(
transaction_hash: String,
@@ -400,6 +403,10 @@ pub(crate) async fn insert_message<'a, E>(
value: JsonValue,
involved_account_addresses: Vec<String>,
height: i64,
wasm_sender: Option<String>,
wasm_contract_address: Option<String>,
wasm_message_type: Option<String>,
funds: Option<Vec<Coin>>,
executor: E,
) -> Result<(), sqlx::Error>
where
@@ -408,25 +415,55 @@ where
trace!("insert_message");
let start = Instant::now();
sqlx::query!(
r#"
INSERT INTO message(transaction_hash, index, type, value, involved_accounts_addresses, height)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (transaction_hash, index) DO UPDATE
SET height = excluded.height,
type = excluded.type,
value = excluded.value,
involved_accounts_addresses = excluded.involved_accounts_addresses
"#,
transaction_hash,
index,
typ,
value,
&involved_account_addresses,
height
)
.execute(executor)
.await?;
// sqlx doesn't understand option types
if let Some(coins) = funds {
sqlx::query!(
r#"
INSERT INTO message(transaction_hash, index, type, value, involved_accounts_addresses, height, wasm_sender, wasm_contract_address, wasm_message_type, funds)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT (transaction_hash, index) DO UPDATE
SET height = excluded.height,
type = excluded.type,
value = excluded.value,
involved_accounts_addresses = excluded.involved_accounts_addresses
"#,
transaction_hash,
index,
typ,
value,
&involved_account_addresses,
height,
wasm_sender,
wasm_contract_address,
wasm_message_type,
&coins as _,
)
.execute(executor)
.await?;
} else {
sqlx::query!(
r#"
INSERT INTO message(transaction_hash, index, type, value, involved_accounts_addresses, height, wasm_sender, wasm_contract_address, wasm_message_type)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (transaction_hash, index) DO UPDATE
SET height = excluded.height,
type = excluded.type,
value = excluded.value,
involved_accounts_addresses = excluded.involved_accounts_addresses
"#,
transaction_hash,
index,
typ,
value,
&involved_account_addresses,
height,
wasm_sender,
wasm_contract_address,
wasm_message_type,
)
.execute(executor)
.await?;
}
log_db_operation_time("insert_message", start);
Ok(())
@@ -434,7 +471,7 @@ where
#[instrument(skip(executor))]
pub(crate) async fn update_last_processed<'a, E>(
height: i32,
height: i64,
executor: E,
) -> Result<(), sqlx::Error>
where
common/nyxd-scraper-psql/src/storage/models.rs
+9 -1
View File
@@ -1,8 +1,9 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use sqlx::FromRow;
use serde::{Deserialize, Serialize};
use sqlx::types::time::OffsetDateTime;
use sqlx::FromRow;
#[derive(Debug, Clone, Eq, PartialEq, Hash, FromRow)]
pub struct Validator {
@@ -28,3 +29,10 @@ pub struct CommitSignature {
pub proposer_priority: i64,
pub timestamp: OffsetDateTime,
}
#[derive(Debug, Serialize, Deserialize, sqlx::Type)]
#[sqlx(type_name = "coin")]
pub struct Coin {
pub amount: String,
pub denom: String,
}
common/nyxd-scraper-psql/src/storage/transaction.rs
+55 -4
View File
@@ -2,14 +2,18 @@
// SPDX-License-Identifier: Apache-2.0
use crate::error::PostgresScraperError;
use crate::storage::helpers::{parse_addresses_from_events, PlaceholderStruct};
use crate::models::Coin;
use crate::storage::helpers::parse_addresses_from_events;
use crate::storage::manager::{
insert_block, insert_message, insert_precommit, insert_transaction, insert_validator,
update_last_processed,
};
use async_trait::async_trait;
use base64::engine::general_purpose;
use base64::Engine as _;
use cosmrs::proto;
use cosmrs::proto::cosmwasm::wasm::v1::MsgExecuteContract;
use cosmrs::proto::prost::Message;
use nyxd_scraper_shared::helpers::{
validator_consensus_address, validator_info, validator_pubkey_to_bech32,
};
@@ -18,7 +22,7 @@ use nyxd_scraper_shared::storage::{
validators, Block, Commit, CommitSig, NyxdScraperStorageError, NyxdScraperTransaction,
};
use nyxd_scraper_shared::{Any, MessageRegistry, ParsedTransactionResponse};
use serde_json::json;
use serde_json::{json, Value};
use sqlx::types::time::{OffsetDateTime, PrimitiveDateTime};
use sqlx::{Postgres, Transaction};
use std::ops::{Deref, DerefMut};
@@ -211,13 +215,44 @@ impl PostgresStorageTransaction {
for chain_tx in txs {
let involved_addresses = parse_addresses_from_events(chain_tx);
for (index, msg) in chain_tx.tx.body.messages.iter().enumerate() {
let mut wasm_sender: Option<String> = None;
let mut wasm_contract_address: Option<String> = None;
let mut wasm_message_type: Option<String> = None;
let mut funds: Option<Vec<Coin>> = None;
let value = serde_json::to_value(self.decode_or_skip(msg))?;
if msg.type_url == "/cosmwasm.wasm.v1.MsgExecuteContract" {
if let Ok(wasm_execute) = MsgExecuteContract::decode(msg.value.as_ref()) {
wasm_sender = Some(wasm_execute.sender);
wasm_contract_address = Some(wasm_execute.contract);
if let Some(raw_msg) = value.get("msg") {
wasm_message_type = get_first_field_name(raw_msg);
}
funds = Some(
wasm_execute
.funds
.iter()
.map(|c| Coin {
amount: c.amount.to_string(),
denom: c.denom.clone(),
})
.collect(),
);
}
}
insert_message(
chain_tx.hash.to_string(),
index as i64,
msg.type_url.clone(),
serde_json::to_value(self.decode_or_skip(msg))?,
value,
involved_addresses.clone(),
chain_tx.height.into(),
wasm_sender,
wasm_contract_address,
wasm_message_type,
funds,
self.inner.as_mut(),
)
.await?
@@ -226,6 +261,20 @@ impl PostgresStorageTransaction {
Ok(())
}
async fn update_last_processed(&mut self, height: i64) -> Result<(), PostgresScraperError> {
debug!("update_last_processed");
update_last_processed(height, self.inner.as_mut()).await?;
Ok(())
}
}
fn get_first_field_name(value: &Value) -> Option<String> {
debug!("value:\n{value}");
match value.as_object() {
Some(map) => map.keys().next().cloned(),
None => None,
}
}
#[async_trait]
@@ -286,6 +335,8 @@ impl NyxdScraperTransaction for PostgresStorageTransaction {
}
async fn update_last_processed(&mut self, height: i64) -> Result<(), NyxdScraperStorageError> {
self.update_last_processed(height).await
self.update_last_processed(height)
.await
.map_err(NyxdScraperStorageError::from)
}
}
common/nyxd-scraper-sqlite/src/storage/transaction.rs
+7
View File
@@ -4,6 +4,7 @@
use crate::error::SqliteScraperError;
use crate::storage::manager::{
insert_block, insert_message, insert_precommit, insert_transaction, insert_validator,
update_last_processed,
};
use async_trait::async_trait;
use nyxd_scraper_shared::helpers::{
@@ -169,6 +170,12 @@ impl SqliteStorageTransaction {
Ok(())
}
async fn update_last_processed(&mut self, height: i64) -> Result<(), SqliteScraperError> {
debug!("update_last_processed");
update_last_processed(height, self.0.as_mut()).await?;
Ok(())
}
}
#[async_trait]
nym-data-observatory/.gitignore
+2
View File
@@ -0,0 +1,2 @@
0001_metadata.sql
0002_cosmos.sql
nym-data-observatory/.sqlx/query-140df23f816ff5d7501128682ce378d582b7da78c45bc0de934f92c1abe14bda.json
Generated
+19
View File
@@ -0,0 +1,19 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO price_history\n (timestamp, chf, usd, eur, gbp, btc)\n VALUES\n ($1, $2, $3, $4, $5, $6)\n ON CONFLICT(timestamp) DO UPDATE SET\n chf=excluded.chf,\n usd=excluded.usd,\n eur=excluded.eur,\n gbp=excluded.gbp,\n btc=excluded.btc;",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Float8",
"Float8",
"Float8",
"Float8",
"Float8"
]
},
"nullable": []
},
"hash": "140df23f816ff5d7501128682ce378d582b7da78c45bc0de934f92c1abe14bda"
}
nym-data-observatory/.sqlx/query-a57b74a049b33aee36b72741056d60df8ad35a747808d5d1d3d525a76bbf0618.json
Generated
+52
View File
@@ -0,0 +1,52 @@
{
"db_name": "PostgreSQL",
"query": "SELECT timestamp, chf, usd, eur, gbp, btc FROM price_history WHERE timestamp >= $1;",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "timestamp",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "chf",
"type_info": "Float8"
},
{
"ordinal": 2,
"name": "usd",
"type_info": "Float8"
},
{
"ordinal": 3,
"name": "eur",
"type_info": "Float8"
},
{
"ordinal": 4,
"name": "gbp",
"type_info": "Float8"
},
{
"ordinal": 5,
"name": "btc",
"type_info": "Float8"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false,
false,
false,
false,
false,
false
]
},
"hash": "a57b74a049b33aee36b72741056d60df8ad35a747808d5d1d3d525a76bbf0618"
}
nym-data-observatory/.sqlx/query-f81a3275a1c7cbeefb3fdf7904c677d46a284e0446b96a2fc5bd77630c62d4b8.json
Generated
+50
View File
@@ -0,0 +1,50 @@
{
"db_name": "PostgreSQL",
"query": "SELECT timestamp, chf, usd, eur, gbp, btc FROM price_history ORDER BY timestamp DESC LIMIT 1;",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "timestamp",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "chf",
"type_info": "Float8"
},
{
"ordinal": 2,
"name": "usd",
"type_info": "Float8"
},
{
"ordinal": 3,
"name": "eur",
"type_info": "Float8"
},
{
"ordinal": 4,
"name": "gbp",
"type_info": "Float8"
},
{
"ordinal": 5,
"name": "btc",
"type_info": "Float8"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
false,
false,
false,
false,
false
]
},
"hash": "f81a3275a1c7cbeefb3fdf7904c677d46a284e0446b96a2fc5bd77630c62d4b8"
}
nym-data-observatory/.sqlx/query-fbf7dc2d779476fffcefafaa0a1731dfc6affe6c672df121140a5c7141f71c63.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO watcher_execution(start_ts, end_ts, error_message)\n VALUES ($1, $2, $3)\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Timestamptz",
"Timestamptz",
"Text"
]
},
"nullable": []
},
"hash": "fbf7dc2d779476fffcefafaa0a1731dfc6affe6c672df121140a5c7141f71c63"
}
nym-data-observatory/Cargo.toml
+48
View File
@@ -0,0 +1,48 @@
# Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
# SPDX-License-Identifier: GPL-3.0-only
[package]
name = "nym-data-observatory"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license.workspace = true
rust-version.workspace = true
readme.workspace = true
[dependencies]
anyhow = { workspace = true }
async-trait.workspace = true
axum = { workspace = true, features = ["tokio"] }
chrono = { workspace = true }
clap = { workspace = true, features = ["cargo", "derive", "env"] }
nym-config = { path = "../common/config" }
nym-bin-common = { path = "../common/bin-common", features = ["output_format"] }
nym-network-defaults = { path = "../common/network-defaults" }
nym-task = { path = "../common/task" }
nym-validator-client = { path = "../common/client-libs/validator-client" }
nyxd-scraper-psql = { path = "../common/nyxd-scraper-psql" }
nyxd-scraper-shared = { path = "../common/nyxd-scraper-shared" }
reqwest = { workspace = true, features = ["rustls-tls"] }
schemars = { workspace = true }
serde = { workspace = true, features = ["derive"] }
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres", "time"] }
thiserror = { workspace = true }
time = { workspace = true }
tokio = { workspace = true, features = ["process", "rt-multi-thread"] }
tokio-util = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true, features = ["env-filter"] }
tower-http = { workspace = true, features = ["cors", "trace"] }
utoipa = { workspace = true, features = ["axum_extras", "time"] }
utoipa-swagger-ui = { workspace = true, features = ["axum"] }
utoipauto = { workspace = true }
[build-dependencies]
anyhow = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres"] }
nym-data-observatory/Dockerfile
+32
View File
@@ -0,0 +1,32 @@
# this will only work with VPN, otherwise remove the harbor part
FROM harbor.nymte.ch/dockerhub/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_DB_URL=postgres://nym_data_observatory:data-data-data@localhost/nym_data_observatory_data
#
# And optionally:
#
# NYM_DATA_OBSERVATORY_WEBHOOK_URL="https://webhook.site"
# NYM_DATA_OBSERVATORY_WEBHOOK_AUTH=1234
# NYX_CHAIN_WATCHER_CONFIG_ENV_FILE_ARG = /mnt/sandbox.env for sandbox environment
#
# see https://github.com/nymtech/nym/blob/develop/nym-data-observatory/src/cli/commands/run/args.rs for details
# and https://github.com/nymtech/nym/blob/develop/nym-data-observatory/src/env.rs for env vars
#-------------------------------------------------------------------
FROM harbor.nymte.ch/dockerhub/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", "run" ]
nym-data-observatory/Makefile
+104
View File
@@ -0,0 +1,104 @@
# Makefile for nyx_chain_scraper database management
# --- Configuration ---
TEST_DATABASE_URL := postgres://testuser:testpass@localhost:5433/nym_data_observatory_test
# Docker compose service names
DB_SERVICE_NAME := postgres-test
DB_CONTAINER_NAME := nym_data_observatory_test
# Default target
.PHONY: default
default: help
# --- Main Targets ---
.PHONY: prepare-pg
prepare-pg: test-db-up test-db-wait test-db-migrate test-db-prepare test-db-down ## Setup PostgreSQL and prepare SQLx offline cache
.PHONY: test-db
test-db: test-db-up test-db-wait test-db-migrate test-db-run test-db-down ## Run tests with PostgreSQL database
.PHONY: dev-db
dev-db: test-db-up test-db-wait test-db-migrate ## Start PostgreSQL for development (keeps running)
@echo "PostgreSQL is running on port 5433"
@echo "Connection string: $(TEST_DATABASE_URL)"
# --- Docker Compose Targets ---
.PHONY: test-db-up
test-db-up: ## Start the PostgreSQL test database in the background
@echo "Starting PostgreSQL test database..."
docker compose up -d $(DB_SERVICE_NAME)
.PHONY: test-db-wait
test-db-wait: ## Wait for the PostgreSQL database to be healthy
@echo "Waiting for PostgreSQL database..."
@while ! docker inspect --format='{{.State.Health.Status}}' $(DB_CONTAINER_NAME) 2>/dev/null | grep -q 'healthy'; do \
echo -n "."; \
sleep 1; \
done; \
echo " Database is healthy!"
.PHONY: test-db-down
test-db-down: ## Stop and remove the test database
@echo "Stopping PostgreSQL test database..."
docker compose down
# --- SQLx Targets ---
.PHONY: test-db-migrate
test-db-migrate: ## Run database migrations against PostgreSQL
@echo "Copying common PostgreSQL migrations..."
cp ../common/nyxd-scraper-psql/sql_migrations/* migrations
@echo "Running watcher PostgreSQL migrations..."
RUST_LOG=debug DATABASE_URL="$(TEST_DATABASE_URL)" sqlx migrate run --source migrations
.PHONY: test-db-prepare
test-db-prepare: ## Run sqlx prepare for compile-time query verification
@echo "Running sqlx prepare for PostgreSQL..."
DATABASE_URL="$(TEST_DATABASE_URL)" cargo sqlx prepare --
# --- Build and Test Targets ---
.PHONY: test-db-run
test-db-run: ## Run tests with PostgreSQL feature
@echo "Running tests with PostgreSQL..."
DATABASE_URL="$(TEST_DATABASE_URL)" cargo test --no-default-features
.PHONY: build-pg
build-pg: ## Build with PostgreSQL feature
@echo "Building with PostgreSQL feature..."
cargo build --no-default-features
.PHONY: check-pg
check-pg: ## Check code with PostgreSQL feature
@echo "Checking code with PostgreSQL feature..."
cargo check --no-default-features
.PHONY: clippy
clippy: clippy-pg
.PHONY: clippy-pg
clippy-pg: ## Run clippy with PostgreSQL feature
@echo "Running clippy with PostgreSQL feature..."
DATABASE_URL="$(TEST_DATABASE_URL)" cargo clippy --no-default-features -- -D warnings
# --- Cleanup Targets ---
.PHONY: clean
clean: ## Clean build artifacts and SQLx cache
cargo clean
rm -rf .sqlx
.PHONY: clean-db
clean-db: test-db-down ## Stop database and clean volumes
docker volume rm -f nym_data_observatory_test_data 2>/dev/null || true
# --- Utility Targets ---
.PHONY: sqlx-cli
sqlx-cli: ## Install sqlx-cli if not already installed
@command -v sqlx >/dev/null 2>&1 || cargo install sqlx-cli --features postgres
.PHONY: psql
psql: ## Connect to the running PostgreSQL database with psql
@docker exec -it $(DB_CONTAINER_NAME) psql -U testuser -d nym_data_observatory_test
.PHONY: help
help: ## Show help for Makefile targets
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-20s\033[0m %s\n", $$1, $$2}'
nym-data-observatory/README.md
+96
View File
@@ -0,0 +1,96 @@
# Nym Data Observatory
Collects data about the Nym network including:
- **Chain scraper** - that parses blocks, transactions and messages on the Nyx chain
- **Price scraper** - to get the NYM/USD token price from CoinGecko
- **Webhooks** - trigger on messages or all messages to call with details
## Running locally
### 1. Install Prerequisites
```bash
# Install sqlx-cli if not already installed
make sqlx-cli
```
### 2. Prepare PostgreSQL for Development
```bash
# This will:
# - Start PostgreSQL in Docker
# - Run migrations
# - Generate SQLx offline query cache
# - Stop the database
make prepare-pg
```
### 3. Build
```bash
make build-pg
```
### 4. Run with PostgreSQL
```bash
# Start PostgreSQL for development (keeps running)
make test-db-up
# In another terminal, run the application
NYM_DATA_OBSERVATORY_DB_URL=postgres://testuser:testpass@localhost:5433/nym_data_observatory_test \
NYM_DATA_OBSERVATORY_WEBHOOK_URL="https://webhook.site" \
NYM_DATA_OBSERVATORY_WEBHOOK_AUTH=1234 \
cargo run -- run
```
To start from a block add the env var: `NYXD_SCRAPER_START_HEIGHT=19266184`.
## Deploying
Connect with `psql` to your local database:
```sql
CREATE USER nym_data_observatory WITH PASSWORD 'data-data-data';
CREATE DATABASE nym_data_observatory_data;
GRANT ALL ON DATABASE nym_data_observatory_data TO nym_data_observatory;
```
Then run:
```
cargo run -- init --db_url postgres://nym_data_observatory:data-data-data@localhost/nym_data_observatory_data
```
and then:
```
NYM_DATA_OBSERVATORY_DB_URL=postgres://nym_data_observatory:data-data-data@localhost/nym_data_observatory_data \
NYM_DATA_OBSERVATORY_WEBHOOK_URL="https://webhook.site" \
NYM_DATA_OBSERVATORY_WEBHOOK_AUTH=1234 \
cargo run -- run
```
## Troubleshooting
### SQLx Offline Mode
If you see "no cached data for this query" errors:
1. Ensure PostgreSQL is running: `make dev-db`
2. Run: `make test-db-prepare`
Also see [README_SQLX.md](../nyx-chain-watcher/README_SQLX.md).
### Connection Refused
If you see "Connection refused" errors:
1. Check Docker is running: `docker ps`
2. Check PostgreSQL container: `docker ps | grep nym_data_observatory
3. Restart database: `make test-db-down && make dev-db`
nym-data-observatory/build.rs
+8
View File
@@ -0,0 +1,8 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
fn main() {
if let Ok(database_url) = std::env::var("DATABASE_URL") {
println!("cargo:rustc-env=DATABASE_URL={database_url}");
}
}
nym-data-observatory/docker-compose.yml
+21
View File
@@ -0,0 +1,21 @@
services:
postgres-test:
image: postgres:16-alpine
container_name: nym_data_observatory_test
environment:
POSTGRES_DB: nym_data_observatory_test
POSTGRES_USER: testuser
POSTGRES_PASSWORD: testpass
ports:
- '5433:5432' # Map to 5433 to avoid conflicts with default PostgreSQL
healthcheck:
test: ['CMD-SHELL', 'pg_isready -U testuser -d nyx_chain_scraper_test']
interval: 5s
timeout: 5s
retries: 5
# Optional: Add volume for persistent data during development
# volumes:
# - nym_data_observatory_test_data:/var/lib/postgresql/data
#volumes:
# nym_data_observatory_test_data:
nym-data-observatory/migrations/0101_price_data.sql
+8
View File
@@ -0,0 +1,8 @@
CREATE TABLE price_history (
timestamp bigint PRIMARY KEY,
chf double precision NOT NULL,
usd double precision NOT NULL,
eur double precision NOT NULL,
btc double precision NOT NULL,
gbp double precision NOT NULL
);
nym-data-observatory/migrations/0102_payment_transactions.sql
+10
View File
@@ -0,0 +1,10 @@
CREATE TABLE payments (
id INTEGER PRIMARY KEY,
transaction_hash TEXT NOT NULL UNIQUE,
sender_address TEXT NOT NULL,
receiver_address TEXT NOT NULL,
amount double precision NOT NULL,
timestamp bigint NOT NULL,
height bigint NOT NULL,
memo TEXT
);
nym-data-observatory/migrations/0103_create_transactions_table.sql
+12
View File
@@ -0,0 +1,12 @@
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY,
tx_hash TEXT NOT NULL,
height BIGINT NOT NULL,
message_index BIGINT NOT NULL,
sender TEXT NOT NULL,
recipient TEXT NOT NULL,
amount TEXT NOT NULL,
memo TEXT,
created_at DATE DEFAULT CURRENT_TIMESTAMP,
UNIQUE(tx_hash, message_index)
);
nym-data-observatory/migrations/0104_add_listener_failure_table.sql
+11
View File
@@ -0,0 +1,11 @@
/*
* Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
* SPDX-License-Identifier: GPL-3.0-only
*/
CREATE TABLE watcher_execution
(
start_ts TIMESTAMPTZ NOT NULL,
end_ts TIMESTAMPTZ NOT NULL,
error_message TEXT
);
nym-data-observatory/src/chain_scraper/mod.rs
+66
View File
@@ -0,0 +1,66 @@
use crate::db::DbPool;
use crate::env::vars::{
NYXD_SCRAPER_START_HEIGHT, NYXD_SCRAPER_UNSAFE_NUKE_DB,
NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT,
};
use nyxd_scraper_psql::{PostgresNyxdScraper, PruningOptions};
use std::fs;
use tracing::{info, warn};
pub(crate) mod webhook;
pub(crate) async fn run_chain_scraper(
config: &crate::config::Config,
_connection_pool: DbPool,
) -> anyhow::Result<PostgresNyxdScraper> {
let websocket_url = std::env::var("NYXD_WS").expect("NYXD_WS not defined");
let rpc_url = std::env::var("NYXD").expect("NYXD not defined");
let websocket_url = reqwest::Url::parse(&websocket_url)?;
let rpc_url = reqwest::Url::parse(&rpc_url)?;
// why are those not part of CLI? : (
let start_block_height = match std::env::var(NYXD_SCRAPER_START_HEIGHT).ok() {
None => None,
// blow up if passed malformed env value
Some(raw) => Some(raw.parse()?),
};
let use_best_effort_start_height =
match std::env::var(NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT).ok() {
None => false,
// blow up if passed malformed env value
Some(raw) => raw.parse()?,
};
let nuke_db: bool = match std::env::var(NYXD_SCRAPER_UNSAFE_NUKE_DB).ok() {
None => false,
// blow up if passed malformed env value
Some(raw) => raw.parse()?,
};
if nuke_db {
warn!("☢️☢️☢️ NUKING THE SCRAPER DATABASE");
fs::remove_file(config.chain_scraper_connection_string())?;
}
let scraper = PostgresNyxdScraper::builder(nyxd_scraper_psql::Config {
websocket_url,
rpc_url,
database_storage: config.chain_scraper_connection_string.clone(),
pruning_options: PruningOptions::nothing(),
store_precommits: false,
start_block: nyxd_scraper_psql::StartingBlockOpts {
start_block_height,
use_best_effort_start_height,
},
})
.with_msg_module(webhook::WebhookModule::new(config.clone())?);
let instance = scraper.build_and_start().await?;
info!("🚧 blocking until the chain has caught up...");
instance.wait_for_startup_sync().await;
Ok(instance)
}
nym-data-observatory/src/chain_scraper/webhook.rs
+147
View File
@@ -0,0 +1,147 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::data_observatory::{HttpAuthenticationOptions, WebhookConfig};
use crate::models::WebhookPayload;
use anyhow::Context;
use async_trait::async_trait;
use nym_validator_client::nyxd::{Any, Msg, MsgSend, Name};
use nyxd_scraper_psql::{
MsgModule, NyxdScraperTransaction, ParsedTransactionResponse, ScraperError,
};
use nyxd_scraper_shared::{default_message_registry, MessageRegistry};
use reqwest::{Client, Url};
use tracing::{error, info, warn};
use utoipa::gen::serde_json;
pub struct WebhookModule {
webhooks: Vec<Webhook>,
registry: MessageRegistry,
}
impl WebhookModule {
pub fn new(config: crate::config::Config) -> anyhow::Result<Self> {
let webhooks = config
.data_observatory_config
.webhooks
.iter()
.map(|watcher_cfg| Webhook::new(watcher_cfg.clone()))
.collect::<anyhow::Result<Vec<_>>>()?;
Ok(Self {
webhooks,
registry: default_message_registry(),
})
}
fn decode_or_skip(&self, msg: &Any) -> Option<serde_json::Value> {
match self.registry.try_decode(msg) {
Ok(decoded) => Some(decoded),
Err(err) => {
warn!("webhook processing failed {err}");
None
}
}
}
}
#[async_trait]
impl MsgModule for WebhookModule {
fn type_url(&self) -> String {
<MsgSend as Msg>::Proto::type_url()
}
async fn handle_msg(
&mut self,
index: usize,
msg: &Any,
tx: &ParsedTransactionResponse,
_storage_tx: &mut dyn NyxdScraperTransaction,
) -> Result<(), ScraperError> {
let message = serde_json::to_value(self.decode_or_skip(msg)).ok();
let payload = WebhookPayload {
height: tx.height.value(),
message_index: index as u64,
transaction_hash: tx.hash.to_string(),
message,
};
for webhook in self.webhooks.clone() {
let payload = payload.clone();
tokio::spawn(async move {
webhook
.invoke_webhook(&payload)
.await
.expect("webhook failed to process");
});
}
Ok(())
}
}
#[derive(Clone)]
pub(crate) struct Webhook {
webhook_url: Url,
config: WebhookConfig,
}
impl Webhook {
pub(crate) fn new(config: WebhookConfig) -> anyhow::Result<Self> {
Ok(Webhook {
webhook_url: config
.webhook_url
.as_str()
.parse()
.context("invalid config: provided webhook URL is malformed")?,
config,
})
}
pub(crate) fn id(&self) -> &str {
&self.config.id
}
pub(crate) async fn invoke_webhook(&self, payload: &WebhookPayload) -> anyhow::Result<()> {
let client = Client::builder()
.user_agent(format!(
"nym-data-observatory/{}/webhook-{}",
env!("CARGO_PKG_VERSION"),
self.id()
))
.build()
.context("failed to build reqwest client")?;
let mut request_builder = client.post(self.webhook_url.clone()).json(payload);
if let Some(auth) = &self.config.authentication {
match auth {
HttpAuthenticationOptions::AuthorizationBearerToken { token } => {
request_builder = request_builder.bearer_auth(token);
}
}
}
match request_builder.send().await {
Ok(res) => info!(
"[webhook = {}] ✅ Webhook {} {} - tx {}, index {}",
self.config.id,
res.status(),
res.url(),
payload.transaction_hash,
payload.message_index,
),
Err(err) => {
error!(
"[webhook = {}] ❌ Webhook {:?} {:?} error = {err}",
self.config.id,
err.status(),
err.url(),
);
return Err(err.into());
}
}
Ok(())
}
}
nym-data-observatory/src/cli/commands/build_info.rs
+17
View File
@@ -0,0 +1,17 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::NymDataObservatoryError;
use nym_bin_common::bin_info_owned;
use nym_bin_common::output_format::OutputFormat;
#[derive(clap::Args, Debug)]
pub(crate) struct Args {
#[clap(short, long, default_value_t = OutputFormat::default())]
output: OutputFormat,
}
pub(crate) fn execute(args: Args) -> Result<(), NymDataObservatoryError> {
println!("{}", args.output.format(&bin_info_owned!()));
Ok(())
}
nym-data-observatory/src/cli/commands/init.rs
+46
View File
@@ -0,0 +1,46 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::cli::DEFAULT_NYM_DATA_OBSERVATORY_ID;
use crate::config::data_observatory::HttpAuthenticationOptions::AuthorizationBearerToken;
use crate::config::data_observatory::WebhookConfig;
use crate::config::{default_config_filepath, Config, ConfigBuilder, DataObservatoryConfig};
use crate::env::vars::*;
use crate::error::NymDataObservatoryError;
use nym_config::save_unformatted_config_to_file;
#[derive(clap::Args, Debug)]
pub(crate) struct Args {
/// (Override) Postgres connection string for data storage
#[arg(long, env = NYM_DATA_OBSERVATORY_DB_URL, alias = "db_url")]
pub(crate) chain_history_db_connection_string: String,
}
pub(crate) async fn execute(args: Args) -> Result<(), NymDataObservatoryError> {
let config_path = default_config_filepath();
let data_dir = Config::default_data_directory(&config_path)?;
let builder = ConfigBuilder::new(
config_path.clone(),
data_dir,
args.chain_history_db_connection_string,
)
.with_data_observatory_config(DataObservatoryConfig {
webhooks: vec![WebhookConfig {
id: DEFAULT_NYM_DATA_OBSERVATORY_ID.to_string(),
webhook_url: "https://webhook.site".to_string(),
authentication: Some(AuthorizationBearerToken {
token: "1234".to_string(),
}),
description: None,
watch_for_chain_message_types: vec![
"/cosmos.bank.v1beta1.MsgSend".to_string(),
"/ibc.applications.transfer.v1.MsgTransfer".to_string(),
],
}],
});
let config = builder.build();
Ok(save_unformatted_config_to_file(&config, &config_path)?)
}
nym-data-observatory/src/cli/commands/mod.rs
+3
View File
@@ -0,0 +1,3 @@
pub(crate) mod build_info;
pub(crate) mod init;
pub(crate) mod run;
nym-data-observatory/src/cli/commands/run/args.rs
+33
View File
@@ -0,0 +1,33 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::env::vars::*;
#[derive(clap::Args, Debug)]
pub(crate) struct Args {
/// (Override) Postgres connection string for chain scraper history
#[arg(long, env = NYM_DATA_OBSERVATORY_DB_URL, alias = "db_url")]
pub(crate) db_connection_string: Option<String>,
/// (Override) Watch for chain messages of these types
#[clap(
long,
value_delimiter = ',',
env = NYM_DATA_OBSERVATORY_WATCH_CHAIN_MESSAGE_TYPES
)]
pub watch_for_chain_message_types: Vec<String>,
/// (Override) The webhook to call when we find something
#[clap(
long,
env = NYM_DATA_OBSERVATORY_WEBHOOK_URL
)]
pub webhook_url: Option<String>,
/// (Override) Optionally, authenticate with the webhook
#[clap(
long,
env = NYM_DATA_OBSERVATORY_WEBHOOK_AUTH
)]
pub webhook_auth: Option<String>,
}
nym-data-observatory/src/cli/commands/run/config.rs
+68
View File
@@ -0,0 +1,68 @@
use crate::cli::commands::run::args::Args;
use crate::cli::DEFAULT_NYM_DATA_OBSERVATORY_ID;
use crate::config::data_observatory::{HttpAuthenticationOptions, WebhookConfig};
use crate::config::{default_config_filepath, Config, ConfigBuilder, DataObservatoryConfig};
use crate::error::NymDataObservatoryError;
use tracing::{info, warn};
pub(crate) fn get_run_config(args: Args) -> Result<Config, NymDataObservatoryError> {
info!("{args:#?}");
let Args {
mut watch_for_chain_message_types,
webhook_auth,
webhook_url,
..
} = args;
// if there are no args set, then try load the config
if args.db_connection_string.is_none() {
info!("Loading default config file...");
return Config::read_from_toml_file_in_default_location();
}
// set default messages
if watch_for_chain_message_types.is_empty() {
watch_for_chain_message_types = vec!["/cosmos.bank.v1beta1.MsgSend".to_string()];
}
let config_path = default_config_filepath();
let data_dir = Config::default_data_directory(&config_path)?;
if args.db_connection_string.is_none() {
return Err(NymDataObservatoryError::DbConnectionStringMissing);
}
let mut builder = ConfigBuilder::new(
config_path,
data_dir,
args.db_connection_string
.expect("db connection string is required"),
);
if let Some(webhook_url) = webhook_url {
let authentication =
webhook_auth.map(|token| HttpAuthenticationOptions::AuthorizationBearerToken { token });
let watcher_config = DataObservatoryConfig {
webhooks: vec![WebhookConfig {
id: DEFAULT_NYM_DATA_OBSERVATORY_ID.to_string(),
description: None,
watch_for_chain_message_types,
webhook_url,
authentication,
}],
};
info!("Overriding watcher config with env vars");
builder = builder.with_data_observatory_config(watcher_config);
} else {
warn!(
"You did not specify a webhook in {}. Only database items will be stored.",
crate::env::vars::NYM_DATA_OBSERVATORY_WEBHOOK_URL
);
}
Ok(builder.build())
}
nym-data-observatory/src/cli/commands/run/mod.rs
+211
View File
@@ -0,0 +1,211 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::NymDataObservatoryError;
use anyhow::Context;
use std::time::Duration;
use time::OffsetDateTime;
use tokio::task::{JoinHandle, JoinSet};
use tokio_util::sync::CancellationToken;
use tracing::{error, info};
mod args;
mod config;
use crate::chain_scraper::run_chain_scraper;
use crate::db::DbPool;
use crate::http::state::PriceScraperState;
use crate::price_scraper::PriceScraper;
use crate::{db, http};
pub(crate) use args::Args;
use nym_task::signal::wait_for_signal;
async fn try_insert_watcher_execution_information(
db_pool: DbPool,
start: OffsetDateTime,
end: OffsetDateTime,
error_message: Option<String>,
) {
let _ = sqlx::query!(
r#"
INSERT INTO watcher_execution(start_ts, end_ts, error_message)
VALUES ($1, $2, $3)
"#,
start.into(),
end.into(),
error_message
)
.execute(&db_pool)
.await
.inspect_err(|err| error!("failed to insert run information: {err}"));
}
async fn wait_for_shutdown(
db_pool: DbPool,
start: OffsetDateTime,
main_cancellation_token: CancellationToken,
scraper_cancellation_token: CancellationToken,
mut tasks: JoinSet<Option<anyhow::Result<()>>>,
) {
async fn finalize_shutdown(
db_pool: DbPool,
start: OffsetDateTime,
main_cancellation_token: CancellationToken,
scraper_cancellation_token: CancellationToken,
mut tasks: JoinSet<Option<anyhow::Result<()>>>,
error_message: Option<String>,
) {
// cancel all tasks
main_cancellation_token.cancel();
scraper_cancellation_token.cancel();
// stupid nasty and hacky workaround to make sure all relevant tasks have finished before hard aborting them
// nasty stupid and hacky workaround
tokio::time::sleep(Duration::from_secs(1)).await;
tasks.abort_all();
// insert execution result into the db
try_insert_watcher_execution_information(
db_pool,
start,
OffsetDateTime::now_utc(),
error_message,
)
.await
}
tokio::select! {
// graceful shutdown
_ = wait_for_signal() => {
info!("received shutdown signal");
finalize_shutdown(db_pool, start, main_cancellation_token, scraper_cancellation_token, tasks, None).await;
}
_ = scraper_cancellation_token.cancelled() => {
info!("the scraper has issued cancellation");
finalize_shutdown(db_pool, start, main_cancellation_token, scraper_cancellation_token, tasks, Some("unexpected scraper task cancellation".into())).await;
}
_ = main_cancellation_token.cancelled() => {
info!("one of the tasks has cancelled the token");
finalize_shutdown(db_pool, start, main_cancellation_token, scraper_cancellation_token, tasks, Some("unexpected main task cancellation".into())).await;
}
task_result = tasks.join_next() => {
// the first unwrap is fine => join set was not empty
let error_message = match task_result.unwrap() {
Err(_join_err) => Some("unexpected join error".to_string()),
Ok(Some(Ok(_))) => None,
Ok(Some(Err(err))) => Some(err.to_string()),
Ok(None) => {
Some("unexpected task cancellation".to_string())
}
};
error!("unexpected task termination: {error_message:?}");
finalize_shutdown(db_pool, start, main_cancellation_token, scraper_cancellation_token, tasks, error_message).await;
}
}
}
pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NymDataObservatoryError> {
let start = OffsetDateTime::now_utc();
info!("passed arguments: {args:#?}");
let config = config::get_run_config(args)?;
let db_connection_string = config.chain_scraper_connection_string();
info!("Config is {config:#?}");
info!(
"Chain History Database path is {:?}",
std::path::Path::new(&config.chain_scraper_connection_string()).canonicalize()
);
let storage = db::Storage::init(db_connection_string).await?;
let watcher_pool = storage.pool_owned();
let mut tasks = JoinSet::new();
let cancellation_token = CancellationToken::new();
let scraper_pool = storage.pool_owned();
let shutdown_pool = storage.pool_owned();
// construct shared state
let price_scraper_shared_state = PriceScraperState::new();
// spawn all the tasks
// 1. chain scraper (note: this doesn't really spawn the full scraper on this task, but we don't want to be blocking waiting for its startup)
let scraper_token_handle: JoinHandle<anyhow::Result<CancellationToken>> = tokio::spawn({
let config = config.clone();
async move {
// this only blocks until startup sync is done; it then runs on its own set of tasks
let scraper = run_chain_scraper(&config, scraper_pool).await?;
Ok(scraper.cancel_token())
}
});
// 2. price scraper (note, this task never terminates on its own)
let price_scraper = PriceScraper::new(price_scraper_shared_state.clone(), watcher_pool);
{
let token = cancellation_token.clone();
tasks.spawn(async move {
token
.run_until_cancelled(async move {
price_scraper.run().await;
Ok(())
})
.await
});
}
// 3. http api
let http_server = http::server::build_http_api(
storage.pool_owned(),
&config,
http_port,
price_scraper_shared_state,
)
.await?;
{
let token = cancellation_token.clone();
tasks.spawn(async move {
info!("Starting HTTP server on port {http_port}",);
async move {
Some(
http_server
.run(token.cancelled_owned())
.await
.context("http server failure"),
)
}
.await
});
}
// 1. wait for either shutdown or scraper having finished startup
tokio::select! {
_ = wait_for_signal() => {
info!("received shutdown signal while waiting for scraper to finish its startup");
return Ok(())
}
scraper_token = scraper_token_handle => {
let scraper_token = match scraper_token {
Ok(Ok(token)) => token,
Ok(Err(startup_err)) => {
error!("failed to startup the chain scraper: {startup_err}");
return Err(startup_err.into());
}
Err(runtime_err) => {
error!("failed to finish the scraper startup task: {runtime_err}");
return Ok(())
}
};
wait_for_shutdown(shutdown_pool, start, cancellation_token, scraper_token, tasks).await
}
}
Ok(())
}
nym-data-observatory/src/cli/mod.rs
+67
View File
@@ -0,0 +1,67 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::cli::commands::{build_info, init, run};
use crate::env::vars::*;
use crate::error::NymDataObservatoryError;
use clap::{Parser, Subcommand};
use nym_bin_common::bin_info;
use std::sync::OnceLock;
mod commands;
pub const DEFAULT_NYM_DATA_OBSERVATORY_ID: &str = "default-nym-data-observatory";
// Helper for passing LONG_VERSION to clap
fn pretty_build_info_static() -> &'static str {
static PRETTY_BUILD_INFORMATION: OnceLock<String> = OnceLock::new();
PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print())
}
#[derive(Parser, Debug)]
#[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)]
pub(crate) struct Cli {
/// Path pointing to an env file that configures the nym-data-observatory and overrides any preconfigured values.
#[clap(
short,
long,
env = NYM_DATA_OBSERVATORY_CONFIG_ENV_FILE_ARG
)]
pub(crate) config_env_file: Option<std::path::PathBuf>,
/// Flag used for disabling the printed banner in tty.
#[clap(
long,
env = NYM_DATA_OBSERVATORY_NO_BANNER_ARG
)]
pub(crate) no_banner: bool,
/// Port to listen on
#[arg(long, default_value_t = 8000, env = "NYM_DATA_OBSERVATORY_HTTP_PORT")]
pub(crate) http_port: u16,
#[clap(subcommand)]
command: Commands,
}
impl Cli {
pub(crate) async fn execute(self) -> Result<(), NymDataObservatoryError> {
match self.command {
Commands::BuildInfo(args) => build_info::execute(args),
Commands::Run(args) => run::execute(*args, self.http_port).await,
Commands::Init(args) => init::execute(args).await,
}
}
}
#[derive(Subcommand, Debug)]
pub(crate) enum Commands {
/// Show build information of this binary
BuildInfo(build_info::Args),
/// Start this nym-chain-watcher
Run(Box<run::Args>),
/// Initialise config
Init(init::Args),
}
nym-data-observatory/src/config/data_observatory.rs
+20
View File
@@ -0,0 +1,20 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct DataObservatoryConfig {
pub webhooks: Vec<WebhookConfig>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WebhookConfig {
pub id: String,
pub description: Option<String>,
pub webhook_url: String,
pub watch_for_chain_message_types: Vec<String>,
pub authentication: Option<HttpAuthenticationOptions>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum HttpAuthenticationOptions {
AuthorizationBearerToken { token: String },
}
nym-data-observatory/src/config/mod.rs
+219
View File
@@ -0,0 +1,219 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::template::CONFIG_TEMPLATE;
use nym_bin_common::logging::LoggingSettings;
use nym_config::{
must_get_home, read_config_from_toml_file, save_unformatted_config_to_file, NymConfigTemplate,
DEFAULT_CONFIG_DIR, DEFAULT_CONFIG_FILENAME, DEFAULT_DATA_DIR, NYM_DIR,
};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use tracing::{debug, error};
pub(crate) mod data_observatory;
mod template;
pub use crate::config::data_observatory::DataObservatoryConfig;
use crate::error::NymDataObservatoryError;
const DEFAULT_NYM_DATA_OBSERVATORY_DIR: &str = "nym-data-observatory";
/// Derive default path to nym-data-observatory's config directory.
/// It should get resolved to `$HOME/.nym/nym-data-observatory/config`
pub fn default_config_directory() -> PathBuf {
must_get_home()
.join(NYM_DIR)
.join(DEFAULT_NYM_DATA_OBSERVATORY_DIR)
.join(DEFAULT_CONFIG_DIR)
}
/// Derive default path to nym-data-observatory's config file.
/// It should get resolved to `$HOME/.nym/nym-data-observatory/config/config.toml`
pub fn default_config_filepath() -> PathBuf {
default_config_directory().join(DEFAULT_CONFIG_FILENAME)
}
pub struct ConfigBuilder {
pub config_path: PathBuf,
pub data_dir: PathBuf,
pub chain_scraper_connection_string: String,
pub data_observatory_config: Option<DataObservatoryConfig>,
pub logging: Option<LoggingSettings>,
}
impl ConfigBuilder {
pub fn new(
config_path: PathBuf,
data_dir: PathBuf,
chain_scraper_connection_string: String,
) -> Self {
ConfigBuilder {
config_path,
data_dir,
data_observatory_config: None,
logging: None,
chain_scraper_connection_string,
}
}
#[allow(dead_code)]
pub fn with_data_observatory_config(
mut self,
data_observatory_config: impl Into<DataObservatoryConfig>,
) -> Self {
self.data_observatory_config = Some(data_observatory_config.into());
self
}
#[allow(dead_code)]
pub fn with_logging(mut self, section: impl Into<Option<LoggingSettings>>) -> Self {
self.logging = section.into();
self
}
pub fn build(self) -> Config {
Config {
logging: self.logging.unwrap_or_default(),
save_path: Some(self.config_path),
data_observatory_config: self.data_observatory_config.unwrap_or_default(),
data_dir: self.data_dir,
chain_scraper_connection_string: self.chain_scraper_connection_string,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Config {
// additional metadata holding on-disk location of this config file
#[serde(skip)]
pub(crate) save_path: Option<PathBuf>,
#[serde(skip)]
pub(crate) data_dir: PathBuf,
pub chain_scraper_connection_string: String,
#[serde(default)]
pub data_observatory_config: DataObservatoryConfig,
#[serde(default)]
pub logging: LoggingSettings,
}
impl NymConfigTemplate for Config {
fn template(&self) -> &'static str {
CONFIG_TEMPLATE
}
}
impl Config {
#[allow(unused)]
pub fn save(&self) -> Result<(), NymDataObservatoryError> {
let save_location = self.save_location();
debug!(
"attempting to save config file to '{}'",
save_location.display()
);
save_unformatted_config_to_file(self, &save_location).map_err(|source| {
NymDataObservatoryError::UnformattedConfigSaveFailure {
path: save_location,
source,
}
})
}
#[allow(unused)]
pub fn save_location(&self) -> PathBuf {
self.save_path
.clone()
.unwrap_or(self.default_save_location())
}
#[allow(unused)]
pub fn default_save_location(&self) -> PathBuf {
default_config_filepath()
}
pub fn default_data_directory<P: AsRef<Path>>(
config_path: P,
) -> Result<PathBuf, NymDataObservatoryError> {
let config_path = config_path.as_ref();
// we got a proper path to the .toml file
let Some(config_dir) = config_path.parent() else {
error!(
"'{}' does not have a parent directory. Have you pointed to the fs root?",
config_path.display()
);
return Err(NymDataObservatoryError::DataDirDerivationFailure);
};
let Some(config_dir_name) = config_dir.file_name() else {
error!(
"could not obtain parent directory name of '{}'. Have you used relative paths?",
config_path.display()
);
return Err(NymDataObservatoryError::DataDirDerivationFailure);
};
if config_dir_name != DEFAULT_CONFIG_DIR {
error!(
"the parent directory of '{}' ({}) is not {DEFAULT_CONFIG_DIR}. currently this is not supported",
config_path.display(), config_dir_name.to_str().unwrap_or("UNKNOWN")
);
return Err(NymDataObservatoryError::DataDirDerivationFailure);
}
let Some(node_dir) = config_dir.parent() else {
error!(
"'{}' does not have a parent directory. Have you pointed to the fs root?",
config_dir.display()
);
return Err(NymDataObservatoryError::DataDirDerivationFailure);
};
Ok(node_dir.join(DEFAULT_DATA_DIR))
}
pub fn chain_scraper_connection_string(&self) -> String {
self.chain_scraper_connection_string.clone()
}
// simple wrapper that reads config file and assigns path location
fn read_from_path<P: AsRef<Path>>(
path: P,
data_dir: P,
) -> Result<Self, NymDataObservatoryError> {
let path = path.as_ref();
let data_dir = data_dir.as_ref();
let mut loaded: Config = read_config_from_toml_file(path).map_err(|source| {
NymDataObservatoryError::ConfigLoadFailure {
path: path.to_path_buf(),
source,
}
})?;
loaded.data_dir = data_dir.to_path_buf();
loaded.save_path = Some(path.to_path_buf());
debug!("loaded config file from {}", path.display());
Ok(loaded)
}
#[allow(unused)]
pub fn read_from_toml_file<P: AsRef<Path>>(
path: P,
data_dir: P,
) -> Result<Self, NymDataObservatoryError> {
Self::read_from_path(path, data_dir)
}
pub fn read_from_toml_file_in_default_location() -> Result<Self, NymDataObservatoryError> {
let config_path = default_config_filepath();
let data_dir = Config::default_data_directory(&config_path)?;
Self::read_from_path(config_path, data_dir)
}
}
nym-data-observatory/src/config/template.rs
+29
View File
@@ -0,0 +1,29 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
// While using normal toml marshalling would have been way simpler with less overhead,
// I think it's useful to have comments attached to the saved config file to explain behaviour of
// particular fields.
// Note: any changes to the template must be reflected in the appropriate structs.
pub(crate) const CONFIG_TEMPLATE: &str = r#"
# This is a TOML config file.
# For more information, see https://github.com/toml-lang/toml
[data_observatory_config]
{{#each data_observatory_config.webhooks }}
[[webhooks]]
id={{this.id}}
description='{{this.description}}'
webhook_url='{{this.webhook_url}}'
{{/each}}
##### logging configuration options #####
[logging]
# TODO
"#;
nym-data-observatory/src/db/mod.rs
+38
View File
@@ -0,0 +1,38 @@
use anyhow::{anyhow, Result};
use sqlx::{migrate::Migrator, postgres::PgConnectOptions, Postgres};
use std::str::FromStr;
pub(crate) mod models;
pub(crate) mod queries {
pub mod price;
}
static _MIGRATOR: Migrator = sqlx::migrate!("./migrations");
pub(crate) type DbPool = sqlx::Pool<Postgres>;
pub(crate) struct Storage {
pool: DbPool,
}
impl Storage {
pub async fn init(connection_url: String) -> Result<Self> {
let connect_options = PgConnectOptions::from_str(&connection_url)?;
let pool = DbPool::connect_with(connect_options)
.await
.map_err(|err| anyhow!("Failed to connect to {}: {}", &connection_url, err))?;
// MIGRATOR
// .run(&pool)
// .await
// .map_err(|err| anyhow!("Failed to run migrations: {}", err))?;
Ok(Storage { pool })
}
/// Cloning pool is cheap, it's the same underlying set of connections
pub fn pool_owned(&self) -> DbPool {
self.pool.clone()
}
}
nym-data-observatory/src/db/models.rs
+58
View File
@@ -0,0 +1,58 @@
use serde::{Deserialize, Serialize};
use sqlx::FromRow;
use time::OffsetDateTime;
use utoipa::ToSchema;
#[derive(Clone, Serialize, Deserialize, Debug, ToSchema)]
pub(crate) struct CurrencyPrices {
pub(crate) chf: f64,
pub(crate) usd: f64,
pub(crate) eur: f64,
pub(crate) gbp: f64,
pub(crate) btc: f64,
}
// Struct to hold Coingecko response
#[derive(Clone, Serialize, 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 PriceHistory {
pub(crate) timestamp: i64,
pub(crate) chf: f64,
pub(crate) usd: f64,
pub(crate) eur: f64,
pub(crate) gbp: f64,
pub(crate) btc: f64,
}
#[derive(Serialize, Deserialize, Debug, ToSchema)]
pub(crate) struct PaymentRecord {
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,
}
#[derive(Serialize, Deserialize, Debug, FromRow)]
pub(crate) struct Transaction {
pub(crate) id: i64,
pub(crate) tx_hash: String,
pub(crate) height: i64,
pub(crate) message_index: i64,
pub(crate) sender: String,
pub(crate) recipient: String,
pub(crate) amount: String,
pub(crate) memo: Option<String>,
pub(crate) created_at: Option<OffsetDateTime>,
}
nym-data-observatory/src/db/queries/mod.rs
+5
View File
@@ -0,0 +1,5 @@
mod price;
// re-exporting allows us to access all queries via `queries::bla``
pub(crate) use payments::{get_last_checked_height, insert_payment};
pub(crate) use price::{get_latest_price, insert_nym_prices};
nym-data-observatory/src/db/queries/price.rs
+119
View File
@@ -0,0 +1,119 @@
use crate::db::models::{PriceHistory, PriceRecord};
use crate::db::DbPool;
use chrono::Local;
use std::ops::Sub;
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, gbp, btc)
VALUES
($1, $2, $3, $4, $5, $6)
ON CONFLICT(timestamp) DO UPDATE SET
chf=excluded.chf,
usd=excluded.usd,
eur=excluded.eur,
gbp=excluded.gbp,
btc=excluded.btc;",
timestamp,
price_data.nym.chf,
price_data.nym.usd,
price_data.nym.eur,
price_data.nym.gbp,
price_data.nym.btc,
)
.execute(&mut *conn)
.await?;
Ok(())
}
pub(crate) async fn get_latest_price(pool: &DbPool) -> anyhow::Result<PriceHistory> {
let result = sqlx::query!(
"SELECT timestamp, chf, usd, eur, gbp, btc FROM price_history ORDER BY timestamp DESC LIMIT 1;"
)
.fetch_one(pool)
.await?;
Ok(PriceHistory {
timestamp: result.timestamp,
chf: result.chf,
usd: result.usd,
eur: result.eur,
gbp: result.gbp,
btc: result.btc,
})
}
pub(crate) async fn get_average_price(pool: &DbPool) -> anyhow::Result<PriceHistory> {
// now less 1 day
let earliest_timestamp = Local::now().sub(chrono::Duration::days(1)).timestamp();
let result = sqlx::query!(
"SELECT timestamp, chf, usd, eur, gbp, btc FROM price_history WHERE timestamp >= $1;",
earliest_timestamp
)
.fetch_all(pool)
.await?;
let mut price = PriceHistory {
timestamp: Local::now().timestamp(),
chf: 0f64,
usd: 0f64,
eur: 0f64,
gbp: 0f64,
btc: 0f64,
};
let mut chf_count = 0;
let mut usd_count = 0;
let mut eur_count = 0;
let mut gbp_count = 0;
let mut btc_count = 0;
for p in &result {
if p.chf != 0f64 {
price.chf += p.chf;
chf_count += 1;
}
if p.usd != 0f64 {
price.usd += p.usd;
usd_count += 1;
}
if p.eur != 0f64 {
price.eur += p.eur;
eur_count += 1;
}
if p.gbp != 0f64 {
price.gbp += p.gbp;
gbp_count += 1;
}
if p.btc != 0f64 {
price.btc += p.btc;
btc_count += 1;
}
}
if chf_count > 0 {
price.chf /= chf_count as f64;
}
if usd_count > 0 {
price.usd /= usd_count as f64;
}
if eur_count > 0 {
price.eur /= eur_count as f64;
}
if gbp_count > 0 {
price.gbp /= gbp_count as f64;
}
if btc_count > 0 {
price.btc /= btc_count as f64;
}
Ok(price)
}
nym-data-observatory/src/env.rs
+27
View File
@@ -0,0 +1,27 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
#[allow(unused)]
pub mod vars {
pub const NYM_DATA_OBSERVATORY_NO_BANNER_ARG: &str = "NYM_DATA_OBSERVATORY_NO_BANNER";
pub const NYM_DATA_OBSERVATORY_CONFIG_ENV_FILE_ARG: &str =
"NYM_DATA_OBSERVATORY_CONFIG_ENV_FILE_ARG";
pub const NYM_DATA_OBSERVATORY_DB_URL: &str = "NYM_DATA_OBSERVATORY_DB_URL";
pub const NYXD_SCRAPER_START_HEIGHT: &str = "NYXD_SCRAPER_START_HEIGHT";
pub const NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT: &str =
"NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT";
pub const NYXD_SCRAPER_UNSAFE_NUKE_DB: &str = "NYXD_SCRAPER_UNSAFE_NUKE_DB";
pub const NYM_DATA_OBSERVATORY_ID_ARG: &str = "NYM_DATA_OBSERVATORY_ID";
pub const NYM_DATA_OBSERVATORY_OUTPUT_ARG: &str = "NYM_DATA_OBSERVATORY_OUTPUT";
pub const NYM_DATA_OBSERVATORY_CONFIG_PATH_ARG: &str = "NYM_DATA_OBSERVATORY_CONFIG";
pub const NYM_DATA_OBSERVATORY_WATCH_CHAIN_MESSAGE_TYPES: &str =
"NYM_DATA_OBSERVATORY_WATCH_CHAIN_MESSAGE_TYPES";
pub const NYM_DATA_OBSERVATORY_WEBHOOK_URL: &str = "NYM_DATA_OBSERVATORY_WEBHOOK_URL";
pub const NYM_DATA_OBSERVATORY_WEBHOOK_AUTH: &str = "NYM_DATA_OBSERVATORY_WEBHOOK_AUTH";
}
nym-data-observatory/src/error.rs
+43
View File
@@ -0,0 +1,43 @@
use std::io;
use std::path::PathBuf;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum NymDataObservatoryError {
// #[error("failed to save config file using path '{}'. detailed message: {source}", path.display())]
// ConfigSaveFailure {
// path: PathBuf,
// #[source]
// source: io::Error,
// },
#[error("failed to save config file using path '{}'. detailed message: {source}", path.display())]
UnformattedConfigSaveFailure {
path: PathBuf,
#[source]
source: nym_config::error::NymConfigTomlError,
},
#[error("could not derive path to data directory of this nyx chain watcher")]
DataDirDerivationFailure,
#[error("please provide a database connection string as an env var, cli argument or in a config file")]
DbConnectionStringMissing,
// #[error("could not derive path to config directory of this nyx chain watcher")]
// ConfigDirDerivationFailure,
#[error("failed to load config file using path '{}'. detailed message: {source}", path.display())]
ConfigLoadFailure {
path: PathBuf,
#[source]
source: io::Error,
},
#[error(transparent)]
FileIoFailure(#[from] io::Error),
#[error(transparent)]
AnyhowFailure(#[from] anyhow::Error),
#[error(transparent)]
NymConfigTomlE(#[from] nym_config::error::NymConfigTomlError),
}
nym-data-observatory/src/http/api/mod.rs
+79
View File
@@ -0,0 +1,79 @@
use anyhow::anyhow;
use axum::{response::Redirect, Router};
use tokio::net::ToSocketAddrs;
use tower_http::{cors::CorsLayer, trace::TraceLayer};
use utoipa::OpenApi;
use utoipa_swagger_ui::SwaggerUi;
use crate::http::{api_docs, server::HttpServer, state::AppState};
pub(crate) mod price;
pub(crate) mod status;
pub(crate) struct RouterBuilder {
unfinished_router: Router<AppState>,
}
impl RouterBuilder {
pub(crate) fn with_default_routes() -> Self {
let router = Router::new()
.merge(
SwaggerUi::new("/swagger")
.url("/api-docs/openapi.json", api_docs::ApiDoc::openapi()),
)
.route(
"/",
axum::routing::get(|| async { Redirect::permanent("/swagger") }),
)
.nest(
"/v1",
Router::new()
.nest("/status", status::routes())
.nest("/price", price::routes()),
);
Self {
unfinished_router: router,
}
}
pub(crate) fn with_state(self, state: AppState) -> RouterWithState {
RouterWithState {
router: self.finalize_routes().with_state(state),
}
}
fn finalize_routes(self) -> Router<AppState> {
// layers added later wrap earlier layers
self.unfinished_router
// CORS layer needs to wrap other API layers
.layer(setup_cors())
// logger should be outermost layer
.layer(TraceLayer::new_for_http())
}
}
pub(crate) struct RouterWithState {
router: Router,
}
impl RouterWithState {
pub(crate) async fn build_server<A: ToSocketAddrs>(
self,
bind_address: A,
) -> anyhow::Result<HttpServer> {
tokio::net::TcpListener::bind(bind_address)
.await
.map(|listener| HttpServer::new(self.router, listener))
.map_err(|err| anyhow!("Couldn't bind to address due to {}", err))
}
}
fn setup_cors() -> CorsLayer {
use axum::http::Method;
CorsLayer::new()
.allow_origin(tower_http::cors::Any)
.allow_methods([Method::POST, Method::GET, Method::PATCH, Method::OPTIONS])
.allow_headers(tower_http::cors::Any)
.allow_credentials(false)
}
nym-data-observatory/src/http/api/price.rs
+44
View File
@@ -0,0 +1,44 @@
use crate::db::models::PriceHistory;
use crate::db::queries::price::{get_average_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<AppState> {
Router::new()
.route("/", axum::routing::get(price))
.route("/average", axum::routing::get(average_price))
}
#[utoipa::path(
tag = "NYM Price",
get,
path = "/v1/price",
responses(
(status = 200, body = String)
)
)]
/// Fetch the latest price cached by this API
async fn price(State(state): State<AppState>) -> HttpResult<Json<PriceHistory>> {
get_latest_price(state.db_pool())
.await
.map(Json::from)
.map_err(|_| Error::internal())
}
#[utoipa::path(
tag = "NYM Price",
get,
path = "/v1/price/average",
responses(
(status = 200, body = String)
)
)]
/// Fetch the average price cached by this API
async fn average_price(State(state): State<AppState>) -> HttpResult<Json<PriceHistory>> {
get_average_price(state.db_pool())
.await
.map(Json::from)
.map_err(|_| Error::internal())
}
nym-data-observatory/src/http/api/status.rs
+79
View File
@@ -0,0 +1,79 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::http::models::status::{
ApiStatus, HealthResponse, PriceScraperLastError, PriceScraperLastSuccess,
PriceScraperStatusResponse,
};
use crate::http::state::{AppState, PriceScraperState, StatusState};
use axum::extract::State;
use axum::routing::get;
use axum::{Json, Router};
use nym_bin_common::build_information::BinaryBuildInformationOwned;
pub(crate) fn routes() -> Router<AppState> {
Router::new()
.route("/health", get(health))
.route("/build-information", get(build_information))
.route("/price-scraper", get(price_scraper_status))
}
#[utoipa::path(
tag = "Status",
get,
path = "/build-information",
context_path = "/v1/status",
responses(
(status = 200, body = BinaryBuildInformationOwned)
)
)]
async fn build_information(State(state): State<StatusState>) -> Json<BinaryBuildInformationOwned> {
Json(state.build_information.to_owned())
}
#[utoipa::path(
tag = "Status",
get,
path = "/health",
context_path = "/v1/status",
responses(
(status = 200, body = HealthResponse)
)
)]
async fn health(State(state): State<StatusState>) -> Json<HealthResponse> {
let uptime = state.startup_time.elapsed();
let health = HealthResponse {
status: ApiStatus::Up,
uptime: uptime.as_secs(),
};
Json(health)
}
#[utoipa::path(
tag = "Status",
get,
path = "/price-scraper",
context_path = "/v1/status",
responses(
(status = 200, body = PriceScraperStatusResponse)
)
)]
pub(crate) async fn price_scraper_status(
State(state): State<PriceScraperState>,
) -> Json<PriceScraperStatusResponse> {
let guard = state.inner.read().await;
Json(PriceScraperStatusResponse {
last_success: guard
.last_success
.as_ref()
.map(|s| PriceScraperLastSuccess {
timestamp: s.timestamp,
response: s.response.clone(),
}),
last_failure: guard.last_failure.as_ref().map(|f| PriceScraperLastError {
timestamp: f.timestamp,
message: f.message.clone(),
}),
})
}
nym-data-observatory/src/http/api_docs.rs
+14
View File
@@ -0,0 +1,14 @@
use utoipa::OpenApi;
use utoipauto::utoipauto;
// manually import external structs which are behind feature flags because they
// can't be automatically discovered
// https://github.com/ProbablyClem/utoipauto/issues/13#issuecomment-1974911829
#[utoipauto(paths = "./nym-data-observatory/src")]
#[derive(OpenApi)]
#[openapi(
info(title = "Nym Data Observatory API"),
tags(),
components(schemas())
)]
pub(super) struct ApiDoc;
nym-data-observatory/src/http/error.rs
+21
View File
@@ -0,0 +1,21 @@
pub(crate) type HttpResult<T> = Result<T, Error>;
pub(crate) struct Error {
message: String,
status: axum::http::StatusCode,
}
impl Error {
pub(crate) fn internal() -> Self {
Self {
message: String::from("Internal server error"),
status: axum::http::StatusCode::INTERNAL_SERVER_ERROR,
}
}
}
impl axum::response::IntoResponse for Error {
fn into_response(self) -> axum::response::Response {
(self.status, self.message).into_response()
}
}
nym-data-observatory/src/http/mod.rs
+6
View File
@@ -0,0 +1,6 @@
pub(crate) mod api;
pub(crate) mod api_docs;
pub(crate) mod error;
pub(crate) mod models;
pub(crate) mod server;
pub(crate) mod state;
nym-data-observatory/src/http/models.rs
+68
View File
@@ -0,0 +1,68 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
// if we ever create some sort of chain watcher client, those would need to be extracted
pub mod status {
use crate::config::data_observatory::WebhookConfig;
use crate::db::models::CoingeckoPriceResponse;
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
#[derive(Clone, Copy, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
#[serde(rename_all = "lowercase")]
pub enum ApiStatus {
Up,
}
#[derive(Clone, Copy, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
pub struct HealthResponse {
pub status: ApiStatus,
pub uptime: u64,
}
#[derive(Debug, Serialize, Deserialize, ToSchema)]
pub struct ActiveWebhooksResponse {
pub watchers: Vec<Webhook>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct Webhook {
pub id: String,
pub description: String,
pub webhook_url: String,
pub watched_message_types: Vec<String>,
}
impl From<&WebhookConfig> for Webhook {
fn from(value: &WebhookConfig) -> Self {
Webhook {
id: value.id.clone(),
description: value.description.clone().unwrap_or_default(),
webhook_url: value.webhook_url.clone(),
watched_message_types: value.watch_for_chain_message_types.clone(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub(crate) struct PriceScraperStatusResponse {
pub(crate) last_success: Option<PriceScraperLastSuccess>,
pub(crate) last_failure: Option<PriceScraperLastError>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub(crate) struct PriceScraperLastSuccess {
#[serde(with = "time::serde::rfc3339")]
pub(crate) timestamp: OffsetDateTime,
pub(crate) response: CoingeckoPriceResponse,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub(crate) struct PriceScraperLastError {
#[serde(with = "time::serde::rfc3339")]
pub(crate) timestamp: OffsetDateTime,
pub(crate) message: String,
}
}
nym-data-observatory/src/http/server.rs
+59
View File
@@ -0,0 +1,59 @@
use axum::Router;
use core::net::SocketAddr;
use tokio::net::TcpListener;
use tokio_util::sync::WaitForCancellationFutureOwned;
use crate::config::Config;
use crate::http::state::PriceScraperState;
use crate::{
db::DbPool,
http::{api::RouterBuilder, state::AppState},
};
pub(crate) async fn build_http_api(
db_pool: DbPool,
config: &Config,
http_port: u16,
price_scraper_state: PriceScraperState,
) -> anyhow::Result<HttpServer> {
let router_builder = RouterBuilder::with_default_routes();
let state = AppState::new(
db_pool,
config
.data_observatory_config
.webhooks
.iter()
.map(Into::into)
.collect(),
price_scraper_state,
);
let router = router_builder.with_state(state);
let bind_addr = format!("0.0.0.0:{http_port}");
let server = router.build_server(bind_addr).await?;
Ok(server)
}
pub(crate) struct HttpServer {
router: Router,
listener: TcpListener,
}
impl HttpServer {
pub(crate) fn new(router: Router, listener: TcpListener) -> Self {
Self { router, listener }
}
pub(crate) async fn run(self, receiver: WaitForCancellationFutureOwned) -> std::io::Result<()> {
// into_make_service_with_connect_info allows us to see client ip address
// in middleware, for logging, TLS, routing etc.
axum::serve(
self.listener,
self.router
.into_make_service_with_connect_info::<SocketAddr>(),
)
.with_graceful_shutdown(receiver)
.await
}
}
nym-data-observatory/src/http/state.rs
+124
View File
@@ -0,0 +1,124 @@
use crate::db::models::CoingeckoPriceResponse;
use crate::db::DbPool;
use crate::http::models::status::Webhook;
use axum::extract::FromRef;
use nym_bin_common::bin_info;
use nym_bin_common::build_information::BinaryBuildInformation;
use std::ops::Deref;
use std::sync::Arc;
use time::OffsetDateTime;
use tokio::sync::RwLock;
use tokio::time::Instant;
#[derive(Debug, Clone)]
pub(crate) struct AppState {
db_pool: DbPool,
#[allow(dead_code)]
pub(crate) registered_webhooks: Arc<Vec<Webhook>>,
pub(crate) status_state: StatusState,
pub(crate) price_scraper_state: PriceScraperState,
}
impl AppState {
pub(crate) fn new(
db_pool: DbPool,
registered_payment_watchers: Vec<Webhook>,
price_scraper_state: PriceScraperState,
) -> Self {
Self {
db_pool,
registered_webhooks: Arc::new(registered_payment_watchers),
status_state: Default::default(),
price_scraper_state,
}
}
pub(crate) fn db_pool(&self) -> &DbPool {
&self.db_pool
}
}
#[derive(Clone, Debug)]
pub(crate) struct StatusState {
inner: Arc<StatusStateInner>,
}
impl Default for StatusState {
fn default() -> Self {
StatusState {
inner: Arc::new(StatusStateInner {
startup_time: Instant::now(),
build_information: bin_info!(),
}),
}
}
}
impl Deref for StatusState {
type Target = StatusStateInner;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
#[derive(Debug)]
pub(crate) struct StatusStateInner {
pub(crate) startup_time: Instant,
pub(crate) build_information: BinaryBuildInformation,
}
#[derive(Debug, Clone)]
pub(crate) struct PriceScraperState {
pub(crate) inner: Arc<RwLock<PriceScraperStateInner>>,
}
impl PriceScraperState {
pub(crate) fn new() -> Self {
PriceScraperState {
inner: Arc::new(Default::default()),
}
}
pub(crate) async fn new_failure<S: Into<String>>(&self, error: S) {
self.inner.write().await.last_failure = Some(PriceScraperLastError {
timestamp: OffsetDateTime::now_utc(),
message: error.into(),
})
}
pub(crate) async fn new_success(&self, response: CoingeckoPriceResponse) {
self.inner.write().await.last_success = Some(PriceScraperLastSuccess {
timestamp: OffsetDateTime::now_utc(),
response,
})
}
}
#[derive(Debug, Default)]
pub(crate) struct PriceScraperStateInner {
pub(crate) last_success: Option<PriceScraperLastSuccess>,
pub(crate) last_failure: Option<PriceScraperLastError>,
}
#[derive(Debug)]
pub(crate) struct PriceScraperLastSuccess {
pub(crate) timestamp: OffsetDateTime,
pub(crate) response: CoingeckoPriceResponse,
}
#[derive(Debug)]
pub(crate) struct PriceScraperLastError {
pub(crate) timestamp: OffsetDateTime,
pub(crate) message: String,
}
impl FromRef<AppState> for StatusState {
fn from_ref(input: &AppState) -> Self {
input.status_state.clone()
}
}
impl FromRef<AppState> for PriceScraperState {
fn from_ref(input: &AppState) -> Self {
input.price_scraper_state.clone()
}
}
nym-data-observatory/src/logging.rs
+43
View File
@@ -0,0 +1,43 @@
use tracing::level_filters::LevelFilter;
use tracing_subscriber::{filter::Directive, EnvFilter};
pub(crate) fn setup_tracing_logger() {
fn directive_checked(directive: String) -> Directive {
directive.parse().expect("Failed to parse log directive")
}
let log_builder = tracing_subscriber::fmt()
// Use a more compact, abbreviated log format
.compact()
// Display source code file paths
.with_file(true)
// Display source code line numbers
.with_line_number(true)
// Don't display the event's target (module path)
.with_target(false);
let mut filter = EnvFilter::builder()
// if RUST_LOG isn't set, set default level
.with_default_directive(LevelFilter::INFO.into())
.from_env_lossy();
// these crates are more granularly filtered
let filter_crates = [
"nym_bin_common",
"nym_explorer_client",
"nym_network_defaults",
"nym_validator_client",
"reqwest",
"rustls",
"hyper",
"sqlx",
"h2",
"tendermint_rpc",
"tower_http",
"axum",
];
for crate_name in filter_crates {
filter = filter.add_directive(directive_checked(format!("{crate_name}=warn")));
}
log_builder.with_env_filter(filter).init();
}
nym-data-observatory/src/main.rs
+34
View File
@@ -0,0 +1,34 @@
use clap::{crate_name, crate_version, Parser};
use nym_bin_common::bin_info_owned;
use nym_bin_common::logging::maybe_print_banner;
use nym_network_defaults::setup_env;
use tracing::info;
mod chain_scraper;
mod cli;
mod config;
mod db;
mod env;
mod error;
mod http;
mod logging;
pub mod models;
mod price_scraper;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let cli = cli::Cli::parse();
setup_env(cli.config_env_file.as_ref());
logging::setup_tracing_logger();
if !cli.no_banner {
maybe_print_banner(crate_name!(), crate_version!());
}
let bin_info = bin_info_owned!();
info!("using the following version: {bin_info}");
cli.execute().await?;
Ok(())
}
nym-data-observatory/src/models.rs
+22
View File
@@ -0,0 +1,22 @@
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use utoipa::gen::serde_json;
use utoipa::ToSchema;
#[derive(Serialize, Deserialize, Clone, JsonSchema, ToSchema)]
pub struct WebhookPayload {
pub height: u64,
pub transaction_hash: String,
pub message_index: u64,
pub message: Option<serde_json::Value>,
}
pub mod openapi_schema {
use super::*;
#[derive(ToSchema)]
pub struct Coin {
pub denom: String,
pub amount: String,
}
}
nym-data-observatory/src/price_scraper/mod.rs
+76
View File
@@ -0,0 +1,76 @@
use crate::db::{
models::{CoingeckoPriceResponse, PriceRecord},
queries::price::insert_nym_prices,
};
use core::str;
use tokio::time::Duration;
use crate::db::DbPool;
use crate::http::state::PriceScraperState;
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,gbp,btc";
pub(crate) struct PriceScraper {
shared_state: PriceScraperState,
db_pool: DbPool,
}
impl PriceScraper {
pub(crate) fn new(shared_state: PriceScraperState, db_pool: DbPool) -> Self {
PriceScraper {
shared_state,
db_pool,
}
}
async fn get_coingecko_prices(&self) -> anyhow::Result<CoingeckoPriceResponse> {
tracing::info!("💰 Fetching CoinGecko prices from {COINGECKO_API_URL}");
let response = reqwest::get(COINGECKO_API_URL)
.await?
.json::<CoingeckoPriceResponse>()
.await;
tracing::info!("Got response {:?}", response);
match response {
Ok(resp) => {
let price_record = PriceRecord {
timestamp: time::OffsetDateTime::now_utc().unix_timestamp(),
nym: resp.nym.clone(),
};
insert_nym_prices(&self.db_pool, price_record).await?;
Ok(resp)
}
Err(err) => {
//tracing::info!("💰 CoinGecko price response: {:?}", response);
tracing::error!("Error sending request: {err}");
Err(err.into())
}
}
}
pub(crate) async fn run(&self) {
loop {
tracing::info!("Running in a loop 🏃");
match self.get_coingecko_prices().await {
Ok(coingecko_price_response) => {
self.shared_state
.new_success(coingecko_price_response)
.await;
tracing::info!("✅ Successfully fetched CoinGecko prices");
tokio::time::sleep(REFRESH_DELAY).await;
}
Err(err) => {
tracing::error!("❌ Failed to get CoinGecko prices: {err}");
tracing::info!("Retrying in {}s...", FAILURE_RETRY_DELAY.as_secs());
self.shared_state.new_failure(err.to_string()).await;
tokio::time::sleep(FAILURE_RETRY_DELAY).await;
}
}
}
}
}
nyx-chain-watcher/src/cli/commands/run/mod.rs
+1 -1
View File
@@ -15,7 +15,7 @@ mod config;
use crate::chain_scraper::run_chain_scraper;
use crate::db::DbPool;
use crate::http::state::{BankScraperModuleState, PaymentListenerState, PriceScraperState};
use crate::payment_listener::PaymentListener;
use crate::listener::PaymentListener;
use crate::price_scraper::PriceScraper;
use crate::{db, http};
pub(crate) use args::Args;
nyx-chain-watcher/src/payment_listener/mod.rs → nyx-chain-watcher/src/listener/mod.rs
+1 -1
View File
@@ -7,8 +7,8 @@ use crate::db::queries;
use crate::http::state::{
PaymentListenerFailureDetails, PaymentListenerState, ProcessedPayment, WatcherFailureDetails,
};
use crate::listener::watcher::PaymentWatcher;
use crate::models::WebhookPayload;
use crate::payment_listener::watcher::PaymentWatcher;
use anyhow::Context;
use sqlx::SqlitePool;
use tokio::time::{self, Duration};
nyx-chain-watcher/src/payment_listener/watcher.rs → nyx-chain-watcher/src/listener/watcher.rs
View File
nyx-chain-watcher/src/main.rs
+1 -1
View File
@@ -12,9 +12,9 @@ mod env;
mod error;
pub(crate) mod helpers;
mod http;
mod listener;
mod logging;
pub mod models;
mod payment_listener;
mod price_scraper;
#[tokio::main]