/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2025 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::{io::ErrorKind, str::FromStr};
use futures::{select, FutureExt};
use libc::{fcntl, F_GETFL, F_SETFL, O_NONBLOCK};
use linenoise_rs::{
linenoise_history_add, linenoise_history_load, linenoise_history_save,
linenoise_set_completion_callback, linenoise_set_hints_callback, LinenoiseState,
};
use smol::channel::{unbounded, Receiver, Sender};
use url::Url;
use darkfi::{
cli_desc,
system::{msleep, ExecutorPtr, StoppableTask, StoppableTaskPtr},
util::path::expand_path,
Error,
};
use crate::{
cli_util::{generate_completions, kaching},
rpc::subscribe_blocks,
DrkPtr,
};
// TODO:
// 1. add rest commands handling, along with their completions, hints and help message.
// 2. add input definitions, so you input from files not just stdin.
// 3. add output definitions, so you can output to files not just stdout.
// 4. create a transactions cache in the wallet db, so you can use it to handle them.
/// Auxiliary function to print the help message.
fn help() {
println!("{}", cli_desc!());
println!("Commands:");
println!("\thelp: Prints the help message");
println!("\tkaching: Fun");
println!("\tping: Send a ping request to the darkfid RPC endpoint");
println!("\tcompletions: Generate a SHELL completion script and print to stdout");
println!(
"\tsubscribe: Perform a scan and then subscribe to darkfid to listen for incoming blocks"
);
println!("\tunsubscribe: Stops the background subscription, if its active");
println!("\tsnooze: Disables the background subscription messages printing");
println!("\tunsnooze: Enables the background subscription messages printing");
println!("\tscan: Scan the blockchain and parse relevant transactions");
}
/// Auxiliary function to define the interactive shell completions.
fn completion(buf: &str, lc: &mut Vec) {
// First we define the specific commands prefixes
if buf.starts_with("h") {
lc.push("help".to_string());
return
}
if buf.starts_with("k") {
lc.push("kaching".to_string());
return
}
if buf.starts_with("p") {
lc.push("ping".to_string());
return
}
if buf.starts_with("c") {
lc.push("completions".to_string());
return
}
if buf.starts_with("su") {
lc.push("subscribe".to_string());
return
}
if buf.starts_with("unsu") {
lc.push("unsubscribe".to_string());
return
}
if buf.starts_with("sn") {
lc.push("snooze".to_string());
return
}
if buf.starts_with("unsn") {
lc.push("unsnooze".to_string());
return
}
if buf.starts_with("sc") {
lc.push("scan".to_string());
return
}
// Now the catch alls
if buf.starts_with("s") {
lc.push("subscribe".to_string());
lc.push("snooze".to_string());
lc.push("scan".to_string());
return
}
if buf.starts_with("u") {
lc.push("unsubscribe".to_string());
lc.push("unsnooze".to_string());
}
}
/// Auxiliary function to define the interactive shell hints.
fn hints(buf: &str) -> Option<(String, i32, bool)> {
match buf {
"completions " => Some(("{shell}".to_string(), 35, false)), // 35 = magenta
"scan " => Some(("--reset {height}".to_string(), 35, false)), // 35 = magenta
_ => None,
}
}
/// Auxiliary function to start provided Drk as an interactive shell.
/// Only sane/linenoise terminals are suported.
pub async fn interactive(drk: &DrkPtr, endpoint: &Url, history_path: &str, ex: &ExecutorPtr) {
// Expand the history file path
let history_path = match expand_path(history_path) {
Ok(p) => p,
Err(e) => {
eprintln!("Error while expanding history file path: {e}");
return
}
};
let history_path = history_path.into_os_string();
let history_file = history_path.to_str().unwrap();
// Set the completion callback. This will be called every time the
// user uses the key.
linenoise_set_completion_callback(completion);
// Set the shell hints
linenoise_set_hints_callback(hints);
// Load history from file.The history file is just a plain text file
// where entries are separated by newlines.
let _ = linenoise_history_load(history_file);
// Create a detached task to use for block subscription
let mut subscription_active = false;
let mut snooze_active = false;
let subscription_task = StoppableTask::new();
// Create an unbounded smol channel, so we can have a printing
// queue the background task can submit messages to the shell.
let (shell_sender, shell_receiver) = unbounded();
// Start the interactive shell
loop {
// Wait for next line to process
let line = listen_for_line(&snooze_active, &shell_receiver).await;
// Grab input or end if Ctrl-D or Ctrl-C was pressed
let Some(line) = line else { break };
// Check if line is empty
if line.is_empty() {
continue
}
// Add line to history
linenoise_history_add(&line);
// Parse command parts
let parts: Vec<&str> = line.split_whitespace().collect();
if parts.is_empty() {
continue
}
// Handle command
match parts[0] {
"help" => help(),
"kaching" => kaching().await,
"ping" => handle_ping(drk).await,
"completions" => handle_completions(&parts),
"subscribe" => {
handle_subscribe(
drk,
endpoint,
&mut subscription_active,
&subscription_task,
&shell_sender,
ex,
)
.await
}
"unsubscribe" => handle_unsubscribe(&mut subscription_active, &subscription_task).await,
"snooze" => snooze_active = true,
"unsnooze" => snooze_active = false,
"scan" => handle_scan(drk, &subscription_active, &parts).await,
_ => println!("Unreconized command: {}", parts[0]),
}
}
// Stop the subscription task if its active
if subscription_active {
subscription_task.stop().await;
}
// Write history file
let _ = linenoise_history_save(history_file);
}
/// Auxiliary function to listen for linenoise input line and handle
/// background task messages.
async fn listen_for_line(
snooze_active: &bool,
shell_receiver: &Receiver>,
) -> Option {
// Generate the linoise state structure
let mut state = match LinenoiseState::edit_start(-1, -1, "drk> ") {
Ok(s) => s,
Err(e) => {
eprintln!("Error while generating linenoise state: {e}");
return None
}
};
// Set stdin to non-blocking mode
let fd = state.get_fd();
unsafe {
let flags = fcntl(fd, F_GETFL, 0);
fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}
// Read until we get a line to process
let mut line = None;
loop {
// Future that polls stdin for input
let input_future = async {
loop {
match state.edit_feed() {
Ok(Some(l)) => {
line = Some(l);
break
}
Ok(None) => break,
Err(e) if e.kind() == ErrorKind::Interrupted => break,
Err(e) if e.kind() == ErrorKind::WouldBlock => {
// No data available, yield and retry
msleep(10).await;
continue
}
Err(e) => {
eprintln!("Error while reading linenoise feed: {e}");
break
}
}
}
};
// Future that polls the channel
let channel_future = async {
loop {
if !shell_receiver.is_empty() {
break
}
msleep(1000).await;
}
};
// Manage the futures
select! {
// When input is ready we break out the loop
_ = input_future.fuse() => break,
// Manage filled channel
_ = channel_future.fuse() => {
while !shell_receiver.is_empty() {
match shell_receiver.recv().await {
Ok(msg) => {
// We only print if snooze is inactive,
// but have to consume the message regardless,
// so the queue gets empty.
if *snooze_active {
continue
}
// Hide prompt, print output, show prompt again
let _ = state.hide();
for line in msg {
println!("{}\r", line.replace("\n", "\n\r"));
}
let _ = state.show();
}
Err(e) => {
eprintln!("Error while reading shell receiver channel: {e}");
break
}
}
}
}
}
}
// Restore blocking mode
unsafe {
let flags = fcntl(fd, F_GETFL, 0);
fcntl(fd, F_SETFL, flags & !O_NONBLOCK);
}
let _ = state.edit_stop();
line
}
/// Auxiliary function to define the ping command handling.
async fn handle_ping(drk: &DrkPtr) {
if let Err(e) = drk.read().await.ping().await {
println!("Error while executing ping command: {e}")
}
}
/// Auxiliary function to define the completions command handling.
fn handle_completions(parts: &[&str]) {
if parts.len() != 2 {
println!("Malformed `completions` command");
println!("Usage: completions {{shell}}");
return
}
if let Err(e) = generate_completions(parts[1]) {
println!("Error while executing completions command: {e}")
}
}
/// Auxiliary function to define the subscribe command handling.
async fn handle_subscribe(
drk: &DrkPtr,
endpoint: &Url,
subscription_active: &mut bool,
subscription_task: &StoppableTaskPtr,
shell_sender: &Sender>,
ex: &ExecutorPtr,
) {
if *subscription_active {
println!("Subscription is already active!");
return
}
if let Err(e) = drk.read().await.scan_blocks().await {
println!("Failed during scanning: {e:?}");
return
}
println!("Finished scanning blockchain");
// Start the subcristion task
let drk_ = drk.clone();
let endpoint_ = endpoint.clone();
let shell_sender_ = shell_sender.clone();
let ex_ = ex.clone();
subscription_task.clone().start(
async move { subscribe_blocks(&drk_, shell_sender_, endpoint_, &ex_).await },
|res| async {
match res {
Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
Err(e) => println!("Failed starting subscription task: {e}"),
}
},
Error::DetachedTaskStopped,
ex.clone(),
);
*subscription_active = true;
}
/// Auxiliary function to define the unsubscribe command handling.
async fn handle_unsubscribe(subscription_active: &mut bool, subscription_task: &StoppableTaskPtr) {
if !*subscription_active {
println!("Subscription is already inactive!");
return
}
subscription_task.stop().await;
*subscription_active = false;
}
/// Auxiliary function to define the scan command handling.
async fn handle_scan(drk: &DrkPtr, subscription_active: &bool, parts: &[&str]) {
if *subscription_active {
println!("Subscription is already active!");
return
}
// Check correct command structure
if parts.len() != 1 && parts.len() != 3 {
println!("Malformed `scan` command");
return
}
// Check if reset was requested
let lock = drk.read().await;
if parts.len() == 3 {
if parts[1] != "--reset" {
println!("Malformed `scan` command");
println!("Usage: scan --reset {{height}}");
return
}
let height = match u32::from_str(parts[2]) {
Ok(h) => h,
Err(e) => {
println!("Invalid reset height: {e:?}");
return
}
};
if let Err(e) = lock.reset_to_height(height) {
println!("Failed during wallet reset: {e:?}");
return
}
}
if let Err(e) = lock.scan_blocks().await {
println!("Failed during scanning: {e:?}");
return
}
println!("Finished scanning blockchain");
}