/* 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;
use async_trait::async_trait;
use darkfi::{
event_graph::{util::recreate_from_replayer_log, Event},
net::P2pPtr,
rpc::{
jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
p2p_method::HandlerP2p,
server::RequestHandler,
util::{json_map, JsonValue},
},
system::StoppableTaskPtr,
};
use darkfi_serial::deserialize_async_partial;
use smol::lock::MutexGuard;
use tracing::debug;
use super::DarkIrc;
use crate::Privmsg;
#[async_trait]
impl RequestHandler<()> for DarkIrc {
async fn handle_request(&self, req: JsonRequest) -> JsonResult {
debug!(target: "darkirc::rpc", "--> {}", req.stringify().unwrap());
match req.method.as_str() {
"ping" => self.pong(req.id, req.params).await,
"dnet.switch" => self.dnet_switch(req.id, req.params).await,
"dnet.subscribe_events" => self.dnet_subscribe_events(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,
"eventgraph.replay" => self.eg_rep_info(req.id, req.params).await,
"gource.subscribe_events" => self.gource_subscribe_events(req.id, req.params).await,
_ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
}
}
async fn connections_mut(&self) -> MutexGuard<'life0, HashSet> {
self.rpc_connections.lock().await
}
}
impl DarkIrc {
// 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 Some(params) = params.get::>() else {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
};
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 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 Some(params) = params.get::>() else {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
};
if !params.is_empty() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
self.dnet_sub.clone().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 Some(params) = params.get::>() else {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
};
if !params.is_empty() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
self.deg_sub.clone().into()
}
// RPCAPI:
// Initializes a subscription to the Gource visualization feed.
// Once a subscription is established, every rotating-DAG event
// that successfully decodes as a Privmsg is projected to a
// Gource-shaped record and forwarded to the subscriber.
//
// To feed Gource directly, reformat to the pipe-delimited custom
// log format and pipe it in:
// ```
// ... | jq -r '.params[0]
// | "\(.timestamp)|\(.user)|\(.action)|\(.path)"' \
// | gource --log-format custom -
// ```
//
// --> {"jsonrpc": "2.0", "method": "gource.subscribe_events", "params": [], "id": 1}
// <-- {"jsonrpc": "2.0", "method": "gource.subscribe_events", "params": [`event`]}
pub async fn gource_subscribe_events(&self, id: i64, params: JsonValue) -> JsonResult {
let Some(params) = params.get::>() else {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
};
if !params.is_empty() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
self.gource_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 Some(params) = params.get::>() else {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
};
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();
} else {
self.event_graph.deg_disable();
}
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 Some(params_) = params.get::>() else {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
};
if !params_.is_empty() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
self.event_graph.eventgraph_info(id, params).await
}
// RPCAPI:
// Get replayed EVENTGRAPH info.
//
// --> {"jsonrpc": "2.0", "method": "eventgraph.replay", "params": ..., "id": 42}
// <-- {"jsonrpc": "2.0", "result": true, "id": 42}
async fn eg_rep_info(&self, id: i64, params: JsonValue) -> JsonResult {
let Some(params_) = params.get::>() else {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
};
if !params_.is_empty() {
return JsonError::new(ErrorCode::InvalidParams, None, id).into()
}
recreate_from_replayer_log(&self.replay_datastore).await
}
}
impl HandlerP2p for DarkIrc {
fn p2p(&self) -> P2pPtr {
self.p2p.clone()
}
}
/// Project a single rotating-DAG event to a Gource-shaped record.
///
/// Returns `None` if the event content isn't a [`Privmsg`] or the
/// privmsg's channel field is empty (in which case there's nothing
/// useful to visualize).
pub async fn privmsg_event_to_gource(event: &Event) -> Option {
let privmsg: Privmsg = match deserialize_async_partial(event.content()).await {
Ok((v, _)) => v,
Err(_) => return None,
};
if privmsg.channel.is_empty() {
return None
}
let path = if let Some(name) = privmsg.channel.strip_prefix('#') {
format!("channels/{name}")
} else {
format!("dms/{}", privmsg.channel)
};
// Gource's custom log expects Unix seconds, not millis.
let unix_secs = event.header.timestamp / 1_000;
Some(json_map([
("timestamp", JsonValue::String(unix_secs.to_string())),
("user", JsonValue::String(privmsg.nick.clone())),
// "M" = modify. We always emit "M" because tracking
// first-touch (which would justify "A") would need
// cross-event state and gource creates the file on first
// reference automatically anyway.
("action", JsonValue::String("M".into())),
("path", JsonValue::String(path)),
]))
}