Files
nym/nym-node-status-api/nym-node-status-client/src/lib.rs
T
Drazen Urch af9f6e5ca0 Allow PG database backend (#5880)
* feat(db): add SQL query wrapper for PostgreSQL placeholder conversion

- Created query_wrapper module with functions to automatically convert
  SQLite ? placeholders to PostgreSQL $1, $2, ... format
- Updated build.rs to handle mutually exclusive feature flags
- Modified one query in mixnodes.rs as proof of concept
- Added type conversions for PostgreSQL compatibility (u32->i64, u16->i32)

This is a checkpoint commit before converting all queries to use the wrapper.

* feat(nym-node-status-api): add PostgreSQL database support via feature flags

Implement dual database support for SQLite and PostgreSQL through Cargo feature flags.
The implementation uses a query wrapper that automatically converts SQLite-style ?
placeholders to PostgreSQL-style $1, $2, ... placeholders at runtime.

Key changes:
- Add query wrapper functions that handle placeholder conversion
- Convert all sqlx::query\! macros to use wrapper functions
- Handle type conversions between databases (i64 vs i32)
- Add feature-gated implementations for database-specific SQL syntax
- Update Makefile with clippy targets for both database features
- Document database support in README

* feat(nym-node-status-agent): add multi-API support with random selection

Agents can now connect to multiple APIs and randomly select one for each testrun:
- Accept multiple --server arguments in format "address:port:auth_key"
- Randomly shuffle server list before attempting connections
- Try each server until a testrun is obtained
- Submit results back only to the API that provided the testrun
- Continue to next server if one is down or has no testruns available

* feat(nym-node-status): implement primary/secondary server architecture

- Agent now requests testruns only from primary server (first in list)
- Results are submitted to all configured servers in parallel
- Secondary servers accept external testruns via new v2 endpoint
- Added auto-creation of gateway and testrun records on secondary servers
- New database queries: get_or_create_gateway, insert_external_testrun
- Client library enhanced with submit_results_with_context method

* Bump Node status API version

* Fix build workdir

* Bump to 3.1.4

* Fix types and queries

* 3.1.6

* Fix gateway perf, bump 3.1.7

* NodeId -> i32, 3.1.8

* Bump agent version

* i64 -> i32

* Use image yq

* Migration and more types

* Update remaining JSONB columns

* Simplify server config

* Update build path

* Change delimiter

* bump agent

* Split up pg and sqlite builds

* More typing fixes, build-and-push script

* Fix Dockerfile-pg

* Bump node-status-api

* TYping

* Agent build script

* More logging around testruns

* Fail loudly on read errors

* Cleanup

* Debug get gateways query

* Fix get_gateways query

* Use pg cert, 3.1.16

* Submit regular results to primary server

* Bump freshenss cutoff

* Update Cargo.lock

* fix: resolve rebase conflicts and compilation errors

After rebasing onto develop, fixed several issues:
- Fixed borrowed data escapes error by using sqlx::query directly in transaction functions
- Removed unused imports and cleaned up code
- Maintained database-specific implementations for transaction functions

* fmt

* Make PG default to make lives easier

* Performance improvements for Explorer v2

* Fix sqlite build

* Fix PG migration

* Tests round 1

* DB tests

* More tests

* And some more tests

* And some more, more tests

* cargo fmt

* Fix some failing lints

* Fix lioness version problems

* Clippy in tests

---------

Co-authored-by: dynco-nym <173912580+dynco-nym@users.noreply.github.com>
2025-07-22 15:25:43 +02:00

137 lines
4.2 KiB
Rust

use crate::models::{get_testrun, submit_results, submit_results_v2, TestrunAssignment};
use anyhow::bail;
use api::ApiPaths;
use nym_crypto::asymmetric::ed25519::{PrivateKey, Signature};
use tracing::instrument;
mod api;
pub mod auth;
pub mod models;
pub struct NsApiClient {
api: ApiPaths,
client: reqwest::Client,
auth_key: PrivateKey,
}
impl NsApiClient {
pub fn new(server_ip: &str, server_port: u16, auth_key: PrivateKey) -> Self {
let server_address = format!("{server_ip}:{server_port}");
let api = ApiPaths::new(server_address);
let client = reqwest::Client::new();
Self {
api,
client,
auth_key,
}
}
#[instrument(level = "debug", skip_all)]
pub async fn request_testrun(&self) -> anyhow::Result<Option<TestrunAssignment>> {
let target_url = self.api.request_testrun();
let payload = get_testrun::Payload {
agent_public_key: self.auth_key.public_key(),
timestamp: time::UtcDateTime::now().unix_timestamp(),
};
let signature = self.sign_message(&payload)?;
let request = get_testrun::GetTestrunRequest { payload, signature };
let res = self.client.get(target_url).json(&request).send().await?;
let status = res.status();
let response_text = res.text().await?;
if status.is_client_error() {
bail!("{}: {}", status, response_text);
} else if status.is_server_error() {
if matches!(status, reqwest::StatusCode::SERVICE_UNAVAILABLE)
&& response_text.contains("No testruns available")
{
return Ok(None);
} else {
bail!("{}: {}", status, response_text);
}
}
serde_json::from_str(&response_text)
.map(|testrun| {
tracing::info!("Received testrun assignment: {:?}", testrun);
testrun
})
.map_err(|err| {
tracing::error!("err");
err.into()
})
}
#[instrument(level = "debug", skip(self, probe_result))]
pub async fn submit_results(
&self,
testrun_id: i64,
probe_result: String,
assigned_at_utc: i64,
) -> anyhow::Result<()> {
let target_url = self.api.submit_results(testrun_id);
let payload = submit_results::Payload {
probe_result,
agent_public_key: self.auth_key.public_key(),
assigned_at_utc,
};
let signature = self.sign_message(&payload)?;
let submit_results = submit_results::SubmitResults { payload, signature };
let res = self
.client
.post(target_url)
.json(&submit_results)
.send()
.await
.and_then(|response| response.error_for_status())?;
tracing::debug!("Submitted results: {})", res.status());
Ok(())
}
#[instrument(level = "debug", skip(self, probe_result))]
pub async fn submit_results_with_context(
&self,
testrun_id: i32,
probe_result: String,
assigned_at_utc: i64,
gateway_identity_key: String,
) -> anyhow::Result<()> {
let target_url = self.api.submit_results_v2(testrun_id);
let payload = submit_results_v2::Payload {
probe_result,
agent_public_key: self.auth_key.public_key(),
assigned_at_utc,
gateway_identity_key,
};
let signature = self.sign_message(&payload)?;
let submit_results = submit_results_v2::SubmitResultsV2 { payload, signature };
let res = self
.client
.post(target_url)
.json(&submit_results)
.send()
.await
.and_then(|response| response.error_for_status())?;
tracing::debug!("Submitted results with context: {})", res.status());
Ok(())
}
fn sign_message<T>(&self, message: &T) -> anyhow::Result<Signature>
where
T: serde::Serialize,
{
let serialized = bincode::serialize(message)?;
let signed = self.auth_key.sign(&serialized);
Ok(signed)
}
}