Ocean: Fixes LW support (#3079)

* Add skip_serialize_none macro

* Use skip_serializing_none

* pp skip_serializing_none

* lint

* Plugin ocean_invalidate_block

* Ocean: api rpc controller (#3090)

* add rpc ctrl

* update ocean ci

* lint

* Ocean logging middleware

* Fix lint

* fmt cpp

---------

Co-authored-by: canonbrother <w.canonbrother@gmail.com>

.github/workflows/tests-ocean.yml

@@ -5,7 +5,7 @@ on:
   pull_request:
     branches:
       - master
-      - feature/ocean-archive # TODO(): remove before merge to master
+      - ocean-refinements # TODO(): remove before merge to master
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref || github.run_id }}

lib/Cargo.toml

@@ -74,7 +74,7 @@ hex-literal = "0.4"
 bincode = "1.3"
 serde_with = "3.0"
 heck = "0.4"
-tower-http = { version = "0.4.0", features = ["full"] }
+tower-http = { version = "0.4", features = ["full"] }
 tower = "0.4.13"
 hyper = "0.14.20"
 

lib/ain-ocean/src/api/address.rs

@@ -41,6 +41,7 @@ struct History {
     txno: u32,
 }
 
+#[skip_serializing_none]
 #[derive(Debug, Serialize)]
 struct AddressHistory {
     owner: String,
@@ -426,6 +427,7 @@ pub struct ScriptUnspentScriptResponse {
     pub hex: String,
 }
 
+#[skip_serializing_none]
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ScriptUnspentVoutResponse {

lib/ain-ocean/src/api/governance.rs

@@ -5,6 +5,7 @@ use axum::{routing::get, Extension, Router};
 use bitcoin::Txid;
 use defichain_rpc::{json::governance::*, GovernanceRPC};
 use serde::Deserialize;
+use serde_with::skip_serializing_none;
 
 use super::{
     path::Path,
@@ -14,6 +15,7 @@ use super::{
 };
 use crate::{error::ApiError, model::ApiProposalInfo, Result};
 
+#[skip_serializing_none]
 #[derive(Deserialize, Default)]
 pub struct GovernanceQuery {
     #[serde(flatten)]

lib/ain-ocean/src/api/loan.rs

@@ -18,6 +18,7 @@ use defichain_rpc::{
 use futures::future::try_join_all;
 use log::trace;
 use serde::{Serialize, Serializer};
+use serde_with::skip_serializing_none;
 use snafu::OptionExt;
 
 use super::{
@@ -95,6 +96,7 @@ async fn get_scheme(
     Ok(Response::new(scheme.into()))
 }
 
+#[skip_serializing_none]
 #[derive(Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct CollateralToken {
@@ -219,6 +221,7 @@ async fn get_collateral_token(
     )))
 }
 
+#[skip_serializing_none]
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct LoanToken {
@@ -571,13 +574,15 @@ pub struct VaultLiquidatedResponse {
     pub batches: Vec<VaultLiquidatedBatchResponse>,
 }
 
+#[skip_serializing_none]
 #[derive(Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct HighestBidResponse {
     pub owner: String,
     pub amount: Option<VaultTokenAmountResponse>,
 }
 
+#[skip_serializing_none]
 #[derive(Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct VaultLiquidatedBatchResponse {
@@ -588,6 +593,7 @@ pub struct VaultLiquidatedBatchResponse {
     froms: Vec<String>,
 }
 
+#[skip_serializing_none]
 #[derive(Serialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct VaultTokenAmountResponse {

lib/ain-ocean/src/api/mod.rs

@@ -1,13 +1,14 @@
-use std::{net::SocketAddr, str::FromStr, sync::Arc};
+use std::{net::SocketAddr, str::FromStr, sync::Arc, time::Instant};
 
 use axum::{
-    body::Body,
-    extract::{ConnectInfo, Request},
-    http::{HeaderValue, StatusCode},
-    middleware::{from_fn, Next},
+    body::{to_bytes, Body},
+    extract::{ConnectInfo, OriginalUri, Request},
+    http::{HeaderMap, HeaderValue, StatusCode},
+    middleware::{self, from_fn, Next},
     response::{IntoResponse, Response},
     Json, Router,
 };
+use log::{debug, log_enabled, trace, Level};
 
 mod address;
 mod block;
@@ -25,12 +26,14 @@ pub mod prices;
 mod query;
 mod rawtx;
 mod response;
+mod rpc;
 mod stats;
 mod tokens;
 mod transactions;
 
 use defichain_rpc::Client;
 use serde::{Deserialize, Serialize};
+use serde_json::json;
 
 use crate::{network::Network, Result, Services};
 
@@ -121,6 +124,7 @@ pub async fn ocean_router(
         .nest("/tokens", tokens::router(Arc::clone(&context)))
         .nest("/transactions", transactions::router(Arc::clone(&context)))
         .nest("/blocks", block::router(Arc::clone(&context)))
+        .nest("/rpc", rpc::router(Arc::clone(&context)))
         .fallback(not_found);
 
     let debug_router = Router::new()
@@ -132,7 +136,8 @@ pub async fn ocean_router(
             format!("/v0/{}", context.network).as_str(),
             main_router.merge(debug_router),
         )
-        .layer(from_fn(cors)))
+        .layer(from_fn(cors))
+        .layer(middleware::from_fn(logging_middleware)))
 }
 
 async fn localhost_only(
@@ -166,3 +171,69 @@ async fn localhost_only(
         Err(StatusCode::FORBIDDEN)
     }
 }
+
+const MAX_BODY_SIZE: usize = 1024 * 1024 * 16; // 16MB limit for body logging
+
+async fn logging_middleware(
+    OriginalUri(original_uri): OriginalUri,
+    req: Request<Body>,
+    next: Next,
+) -> std::result::Result<impl IntoResponse, StatusCode> {
+    let method = req.method().clone();
+    let path = req.uri().path().to_owned();
+    let query = req.uri().query().unwrap_or("").to_owned();
+
+    debug!("Request: {} {}", method, path);
+
+    if log_enabled!(Level::Trace) {
+        let headers = format_headers(req.headers());
+        let request_log = json!({
+            "method": method.as_str(),
+            "path": path,
+            "query": query,
+            "headers": headers,
+            "original_uri": original_uri.to_string(),
+        });
+        if let Ok(json) = serde_json::to_string(&request_log) {
+            trace!("Request: {json}");
+        }
+    }
+
+    let start = Instant::now();
+    let res = next.run(req).await;
+    let latency = start.elapsed();
+
+    debug!("Response: {} {} {:?}", method, path, latency);
+
+    if log_enabled!(Level::Trace) {
+        let (parts, body) = res.into_parts();
+        let bytes = to_bytes(body, MAX_BODY_SIZE).await.unwrap_or_default();
+        let body_str = String::from_utf8_lossy(&bytes);
+
+        let response_log = json!({
+            "status": parts.status.as_u16(),
+            "headers": format_headers(&parts.headers),
+            "body": body_str,
+        });
+        if let Ok(json) = serde_json::to_string(&response_log) {
+            trace!("Response: {json}",);
+        }
+
+        Ok(Response::from_parts(parts, Body::from(bytes)))
+    } else {
+        Ok(res)
+    }
+}
+
+fn format_headers(headers: &HeaderMap) -> serde_json::Value {
+    let mut map = serde_json::Map::new();
+    for (key, value) in headers.iter() {
+        if let Ok(v) = value.to_str() {
+            map.insert(
+                key.as_str().to_owned(),
+                serde_json::Value::String(v.to_owned()),
+            );
+        }
+    }
+    serde_json::Value::Object(map)
+}

lib/ain-ocean/src/api/pool_pair/mod.rs

@@ -128,6 +128,7 @@ impl From<PoolSwap> for PoolSwapResponse {
     }
 }
 
+#[skip_serializing_none]
 #[derive(Serialize, Debug, Clone, Default)]
 #[serde(rename_all = "camelCase")]
 struct PoolPairFeeResponse {
@@ -136,6 +137,7 @@ struct PoolPairFeeResponse {
     out_pct: Option<String>,
 }
 
+#[skip_serializing_none]
 #[derive(Serialize, Debug, Clone, Default)]
 #[serde(rename_all = "camelCase")]
 struct PoolPairTokenResponse {
@@ -173,6 +175,7 @@ pub struct PoolPairAprResponse {
     pub commission: Decimal,
 }
 
+#[skip_serializing_none]
 #[derive(Serialize, Debug, Clone, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct PoolPairResponse {

lib/ain-ocean/src/api/prices.rs

@@ -11,6 +11,7 @@ use bitcoin::{hashes::Hash, Txid};
 use indexmap::IndexSet;
 use rust_decimal::{prelude::ToPrimitive, Decimal};
 use serde::{Deserialize, Serialize};
+use serde_with::skip_serializing_none;
 use snafu::OptionExt;
 
 use super::{
@@ -352,6 +353,7 @@ async fn get_feed_with_interval(
     }))
 }
 
+#[skip_serializing_none]
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PriceOracleResponse {

lib/ain-ocean/src/api/rpc.rs

@@ -0,0 +1,67 @@
+use std::sync::Arc;
+
+use ain_macros::ocean_endpoint;
+use axum::{routing::post, Extension, Json, Router};
+use defichain_rpc::RpcApi;
+use serde::{Deserialize, Serialize};
+
+use super::{response::Response, AppContext};
+use crate::{
+    error::{ApiError, Error},
+    Result,
+};
+
+#[derive(Serialize, Deserialize, Default, Clone)]
+#[serde(rename_all = "camelCase")]
+struct RpcDto {
+    method: String,
+    params: Vec<serde_json::Value>,
+}
+
+fn method_whitelist(method: &str) -> Result<()> {
+    let methods = [
+        "getblockchaininfo",
+        "getblockhash",
+        "getblockcount",
+        "getblock",
+        "getblockstats",
+        "getgov",
+        "validateaddress",
+        "listcommunitybalances",
+        "getaccounthistory",
+        "getfutureswapblock",
+        "getpendingfutureswaps",
+        "sendrawtransaction",
+        "getrawtransaction",
+        "getgovproposal",
+        "listgovproposals",
+        "listgovproposalvotes",
+        "vmmap",
+        "gettxout",
+    ];
+
+    if !methods.contains(&method) {
+        log::debug!("forbidden");
+        return Err(Error::Forbidden {
+            method: method.to_owned(),
+        });
+    }
+
+    Ok(())
+}
+
+#[ocean_endpoint]
+async fn rpc(
+    Extension(ctx): Extension<Arc<AppContext>>,
+    Json(body): Json<RpcDto>,
+) -> Result<Response<serde_json::Value>> {
+    method_whitelist(&body.method)?;
+
+    let res: serde_json::Value = ctx.client.call(&body.method, &body.params).await?;
+
+    Ok(Response::new(res))
+}
+
+pub fn router(ctx: Arc<AppContext>) -> Router {
+    Router::new().route("/", post(rpc)).layer(Extension(ctx))
+}

lib/ain-ocean/src/api/tokens.rs

@@ -8,7 +8,7 @@ use defichain_rpc::{
 };
 use serde::Serialize;
 use serde_json::json;
-use serde_with::{serde_as, DisplayFromStr};
+use serde_with::{serde_as, skip_serializing_none, DisplayFromStr};
 
 use super::{
     common::parse_display_symbol,
@@ -28,6 +28,7 @@ pub struct TxHeight {
     height: i64,
 }
 
+#[skip_serializing_none]
 #[serde_as]
 #[derive(Serialize, Debug, Clone, Default)]
 #[serde(rename_all = "camelCase")]

lib/ain-ocean/src/api/transactions.rs

@@ -4,6 +4,7 @@ use ain_macros::ocean_endpoint;
 use axum::{extract::Query, routing::get, Extension, Router};
 use bitcoin::Txid;
 use serde::{Deserialize, Serialize};
+use serde_with::skip_serializing_none;
 
 use super::{path::Path, query::PaginationQuery, response::ApiPagedResponse, AppContext};
 use crate::{
@@ -33,6 +34,7 @@ async fn get_transaction(
     Ok(Response::new(transactions))
 }
 
+#[skip_serializing_none]
 #[derive(Debug, Serialize)]
 struct TransactionVinResponse {
     pub id: String,
@@ -96,6 +98,7 @@ async fn get_vins(
     }))
 }
 
+#[skip_serializing_none]
 #[derive(Debug, Serialize)]
 struct TransactionVoutResponse {
     pub id: String,

lib/ain-ocean/src/error.rs

@@ -108,6 +108,10 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+    #[snafu(display("Rpc {} method is not whitelisted", method))]
+    Forbidden {
+        method: String,
+    },
     #[snafu(context(false))]
     IOError {
         #[snafu(source)]
@@ -312,6 +316,7 @@ impl Error {
             } => (StatusCode::NOT_FOUND, e.message.to_string()),
             Self::NotFound { kind: _ } => (StatusCode::NOT_FOUND, format!("{self}")),
             Self::NotFoundMessage { msg } => (StatusCode::NOT_FOUND, msg.clone()),
+            Self::Forbidden { method: _ } => (StatusCode::FORBIDDEN, format!("{self}")),
             Self::BadRequest { msg } => (StatusCode::BAD_REQUEST, msg.clone()),
             Self::Other { msg } => (StatusCode::INTERNAL_SERVER_ERROR, msg.clone()),
             _ => (StatusCode::INTERNAL_SERVER_ERROR, self.to_string()),