/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2026 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
use std::{collections::HashSet, slice};
use async_trait::async_trait;
use smol::lock::{Mutex, MutexGuard};
use tinyjson::JsonValue;
use tracing::{debug, error};
use darkfi::{
event_graph::{proto::EventPut, Event, EventGraphPtr},
net,
rpc::{
jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult, JsonSubscriber},
p2p_method::HandlerP2p,
server::RequestHandler,
},
system::StoppableTaskPtr,
util::encoding::base64,
};
use darkfi_serial::{deserialize, deserialize_async_partial, serialize_async};
use genevd::GenEvent;
pub struct JsonRpcInterface {
_nickname: String,
event_graph: EventGraphPtr,
p2p: net::P2pPtr,
rpc_connections: Mutex>,
dnet_sub: JsonSubscriber,
deg_sub: JsonSubscriber,
}
#[async_trait]
impl RequestHandler<()> for JsonRpcInterface {
async fn handle_request(&self, req: JsonRequest) -> JsonResult {
match req.method.as_str() {
"add" => self.add(req.id, req.params).await,
"list" => self.list(req.id, req.params).await,
"ping" => self.pong(req.id, req.params).await,
"dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
"dnet.switch" => self.dnet_switch(req.id, req.params).await,
"p2p.get_info" => self.p2p_get_info(req.id, req.params).await,
"deg.switch" => self.deg_switch(req.id, req.params).await,
"deg.subscribe_events" => self.deg_subscribe_events(req.id, req.params).await,
"eventgraph.get_info" => self.eg_get_info(req.id, req.params).await,
_ => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
}
}
async fn connections_mut(&self) -> MutexGuard<'life0, HashSet> {
self.rpc_connections.lock().await
}
}
impl HandlerP2p for JsonRpcInterface {
fn p2p(&self) -> net::P2pPtr {
self.p2p.clone()
}
}
impl JsonRpcInterface {
pub fn new(
_nickname: String,
event_graph: EventGraphPtr,
p2p: net::P2pPtr,
dnet_sub: JsonSubscriber,
deg_sub: JsonSubscriber,
) -> Self {
Self {
_nickname,
event_graph,
p2p,
rpc_connections: Mutex::new(HashSet::new()),
dnet_sub,
deg_sub,
}
}
// RPCAPI:
// Initializes a subscription to p2p dnet events.
// Once a subscription is established, `darkirc` will send JSON-RPC notifications of
// new network events to the subscriber.
//
// --> {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [], "id": 1}
// <-- {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [`event`]}
pub async fn dnet_subscribe_events(&self, id: i64, params: JsonValue) -> JsonResult {
let params = params.get::>().unwrap();
if !params.is_empty() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
self.dnet_sub.clone().into()
}
// RPCAPI:
// Activate or deactivate dnet in the P2P stack.
// By sending `true`, dnet will be activated, and by sending `false` dnet
// will be deactivated. Returns `true` on success.
//
// --> {"jsonrpc": "2.0", "method": "dnet_switch", "params": [true], "id": 42}
// <-- {"jsonrpc": "2.0", "result": true, "id": 42}
async fn dnet_switch(&self, id: i64, params: JsonValue) -> JsonResult {
let params = params.get::>().unwrap();
if params.len() != 1 || !params[0].is_bool() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
let switch = params[0].get::().unwrap();
if *switch {
self.p2p.dnet_enable();
} else {
self.p2p.dnet_disable();
}
JsonResponse::new(JsonValue::Boolean(true), id).into()
}
// RPCAPI:
// Initializes a subscription to deg events.
// Once a subscription is established, apps using eventgraph will send JSON-RPC notifications of
// new eventgraph events to the subscriber.
//
// --> {"jsonrpc": "2.0", "method": "deg.subscribe_events", "params": [], "id": 1}
// <-- {"jsonrpc": "2.0", "method": "deg.subscribe_events", "params": [`event`]}
pub async fn deg_subscribe_events(&self, id: i64, params: JsonValue) -> JsonResult {
let params = params.get::>().unwrap();
if !params.is_empty() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
self.deg_sub.clone().into()
}
// RPCAPI:
// Activate or deactivate deg in the EVENTGRAPH.
// By sending `true`, deg will be activated, and by sending `false` deg
// will be deactivated. Returns `true` on success.
//
// --> {"jsonrpc": "2.0", "method": "deg.switch", "params": [true], "id": 42}
// <-- {"jsonrpc": "2.0", "result": true, "id": 42}
async fn deg_switch(&self, id: i64, params: JsonValue) -> JsonResult {
let params = params.get::>().unwrap();
if params.len() != 1 || !params[0].is_bool() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
let switch = params[0].get::().unwrap();
if *switch {
self.event_graph.deg_enable().await;
} else {
self.event_graph.deg_disable().await;
}
JsonResponse::new(JsonValue::Boolean(true), id).into()
}
// RPCAPI:
// Get EVENTGRAPH info.
//
// --> {"jsonrpc": "2.0", "method": "deg.switch", "params": [true], "id": 42}
// <-- {"jsonrpc": "2.0", "result": true, "id": 42}
async fn eg_get_info(&self, id: i64, params: JsonValue) -> JsonResult {
let params_ = params.get::>().unwrap();
if !params_.is_empty() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
self.event_graph.eventgraph_info(id, params).await
}
// RPCAPI:
// Add a new event
// --> {"jsonrpc": "2.0", "method": "add", "params": [], "id": 1}
// <-- {"jsonrpc": "2.0", "result": [nickname, ...], "id": 1}
async fn add(&self, id: i64, params: JsonValue) -> JsonResult {
let params = params.get::>().unwrap();
if params.len() != 1 || !params[0].is_string() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
let b64 = params[0].get::().unwrap();
let dec = base64::decode(b64).unwrap();
let genevent: GenEvent = deserialize(&dec).unwrap();
// Build a DAG event and return it.
let event = Event::new(serialize_async(&genevent).await, &self.event_graph).await;
let current_genesis = self.event_graph.current_genesis.read().await;
let dag_name = current_genesis.header.timestamp.to_string();
if let Err(e) =
self.event_graph.header_dag_insert(vec![event.header.clone()], &dag_name).await
{
error!("Failed inserting new header to Header DAG: {}", e);
}
if let Err(e) = self.event_graph.dag_insert(slice::from_ref(&event), &dag_name).await {
error!("Failed inserting new event to DAG: {e}");
} else {
// Otherwise, broadcast it
self.p2p.broadcast(&EventPut(event, vec![])).await;
}
let json = JsonValue::Boolean(true);
JsonResponse::new(json, id).into()
}
// RPCAPI:
// List events
// --> {"jsonrpc": "2.0", "method": "list", "params": [], "id": 1}
// <-- {"jsonrpc": "2.0", "result": [task_id, ...], "id": 1}
async fn list(&self, id: i64, _params: JsonValue) -> JsonResult {
debug!("Fetching all events");
let mut seen_events = vec![];
let dag_events = self.event_graph.order_events().await;
for event in dag_events.iter() {
let event_id = event.header.id();
// Try to deserialize it. (Here we skip errors)
let genevent: GenEvent = match deserialize_async_partial(event.content()).await {
Ok((v, _)) => v,
Err(e) => {
error!("Failed deserializing incoming event: {e}");
continue
}
};
debug!("Marking event {event_id} as seen");
seen_events.push(genevent);
}
let ser = darkfi_serial::serialize(&seen_events);
let enc = JsonValue::String(base64::encode(&ser));
JsonResponse::new(enc, id).into()
}
}