reifydb-client 0.9.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

use std::{collections::HashMap, error::Error, fmt::Write, sync::Arc};

#[cfg(not(reifydb_single_threaded))]
use reifydb::{Database, RuntimeConfig, WithSubsystem, server};
use reifydb_client::{Frame, Params, Value};
use reifydb_testing::testscript::command::Command;
#[cfg(not(reifydb_single_threaded))]
use tokio::runtime::Runtime;

#[cfg(not(reifydb_single_threaded))]
pub fn create_server_instance(_runtime: &Arc<Runtime>) -> Database {
	// Cargo unifies reqwest's `rustls-no-provider` feature across the workspace, so reqwest
	// expects a process-wide provider that only uptime's own `main` installs. Install ring
	// once; an error on later calls just means it is already installed.
	let _ = rustls::crypto::ring::default_provider().install_default();
	server::memory()
		.with_runtime_config(RuntimeConfig::default().seeded(0))
		.with_flow(|f| f)
		.with_grpc(|grpc| grpc.admin_bind_addr("[::1]:0"))
		.with_http(|http| http.admin_bind_addr("::1:0"))
		.with_ws(|ws| ws.admin_bind_addr("::1:0"))
		.build()
		.unwrap()
}

/// Throttle and linger gates compare against `clock.now()`; the default seeded clock is frozen
/// at 0, so a window never elapses on its own. Handing in a clock the test also holds lets it
/// step time forward on demand.
#[allow(dead_code)]
#[cfg(not(reifydb_single_threaded))]
pub fn create_server_instance_with_clock(
	_runtime: &Arc<Runtime>,
	clock: reifydb::runtime::context::clock::MockClock,
) -> Database {
	let mut config = RuntimeConfig::default().seeded(0);
	config.clock = reifydb::runtime::context::clock::Clock::Mock(clock);
	server::memory()
		.with_runtime_config(config)
		.with_flow(|f| f)
		.with_grpc(|grpc| grpc.admin_bind_addr("[::1]:0"))
		.with_http(|http| http.admin_bind_addr("::1:0"))
		.with_ws(|ws| ws.admin_bind_addr("::1:0"))
		.build()
		.unwrap()
}

#[allow(dead_code)]
#[cfg(not(reifydb_single_threaded))]
pub fn start_server_and_get_ws_port(_runtime: &Arc<Runtime>, server: &mut Database) -> Result<u16, Box<dyn Error>> {
	server.admin_as_root(
		"CREATE AUTHENTICATION FOR root { method: token; token: 'mysecrettoken' }",
		reifydb_value::params::Params::None,
	)
	.unwrap();
	Ok(server.sub_server_ws().unwrap().admin_port().unwrap())
}

#[allow(dead_code)]
#[cfg(not(reifydb_single_threaded))]
pub fn start_server_and_get_grpc_port(_runtime: &Arc<Runtime>, server: &mut Database) -> Result<u16, Box<dyn Error>> {
	server.admin_as_root(
		"CREATE AUTHENTICATION FOR root { method: token; token: 'mysecrettoken' }",
		reifydb_value::params::Params::None,
	)
	.unwrap();
	Ok(server.sub_server_grpc().unwrap().admin_port().unwrap())
}

#[allow(dead_code)]
#[cfg(not(reifydb_single_threaded))]
pub fn start_server_and_get_http_port(_runtime: &Arc<Runtime>, server: &mut Database) -> Result<u16, Box<dyn Error>> {
	server.admin_as_root(
		"CREATE AUTHENTICATION FOR root { method: token; token: 'mysecrettoken' }",
		reifydb_value::params::Params::None,
	)
	.unwrap();
	Ok(server.sub_server_http().unwrap().admin_port().unwrap())
}

#[cfg(not(reifydb_single_threaded))]
pub fn cleanup_server(mut server: Option<Database>) {
	if let Some(mut srv) = server.take() {
		let _ = srv.stop();
		drop(srv);
	}
}

#[cfg(reifydb_single_threaded)]
use reifydb::{Database, RuntimeConfig, embedded};
#[cfg(reifydb_single_threaded)]
use reifydb_client::DstClient;
#[cfg(reifydb_single_threaded)]
use reifydb_core::actors::server::{ServerAuthResponse, ServerMessage, ServerResponse};
#[cfg(reifydb_single_threaded)]
use reifydb_runtime::actor::system::{ActorHandle, ActorSystem};
#[cfg(reifydb_single_threaded)]
use reifydb_sub_server::actor::ServerActor;
#[cfg(reifydb_single_threaded)]
use reifydb_value::value::identity::IdentityId;

#[cfg(reifydb_single_threaded)]
pub struct DstTestContext {
	pub db: Database,
	#[allow(dead_code)]
	pub system: ActorSystem,
	pub identity: IdentityId,
	_handle: ActorHandle<ServerMessage>,
	pub client: DstClient,
}

#[cfg(reifydb_single_threaded)]
impl DstTestContext {
	pub fn new() -> Self {
		let db = embedded::memory().with_runtime_config(RuntimeConfig::default().seeded(0)).build().unwrap();

		db.admin_as_root(
			"CREATE AUTHENTICATION FOR root { method: token; token: 'mysecrettoken' }",
			reifydb_value::params::Params::None,
		)
		.unwrap();

		let engine = db.engine().clone();
		let auth_service = db.auth_service().clone();
		let system = db.engine().spawner().system();
		let clock = db.clock().clone();

		let handle = system.spawn_ephemeral("server", ServerActor::new(engine, auth_service, clock));
		let client = DstClient::new(handle.actor_ref().clone(), system.clone());

		let auth_response = client.authenticate(
			"token".to_string(),
			HashMap::from([("token".to_string(), "mysecrettoken".to_string())]),
		);
		let identity = match auth_response {
			ServerAuthResponse::Authenticated {
				identity,
				..
			} => identity,
			ServerAuthResponse::Failed {
				reason,
			} => panic!("authentication failed: {}", reason),
			ServerAuthResponse::Error(e) => panic!("authentication error: {}", e),
			ServerAuthResponse::Challenge {
				..
			} => panic!("unexpected challenge response"),
		};

		Self {
			db,
			system,
			identity,
			_handle: handle,
			client,
		}
	}
}

#[cfg(reifydb_single_threaded)]
#[allow(dead_code)]
pub fn dst_response_to_result(response: ServerResponse) -> Result<Vec<Frame>, Box<dyn Error>> {
	match response {
		ServerResponse::Success {
			frames,
			..
		} => Ok(frames),
		ServerResponse::EngineError {
			diagnostic,
			..
		} => {
			let err = reifydb_value::error::Error(diagnostic);
			Err(err.to_string().into())
		}
	}
}

#[allow(dead_code)]
pub fn parse_rql(command: &Command) -> String {
	command.args.iter().map(|a| a.value.as_str()).collect::<Vec<_>>().join(" ")
}

/// First argument is the RQL, the rest are positional parameters.
#[allow(dead_code)]
pub fn parse_positional_params(command: &Command) -> (String, Params) {
	let args: Vec<&str> = command.args.iter().map(|a| a.value.as_str()).collect();

	if args.is_empty() {
		return (String::new(), Params::Positional(Arc::new(vec![])));
	}

	let rql = args[0].to_string();
	let params: Vec<_> = args[1..].iter().map(|s| parse_param_value(s)).collect();

	(rql, Params::Positional(Arc::new(params)))
}

/// First argument is the RQL, the rest are `name=value` pairs.
#[allow(dead_code)]
pub fn parse_named_params(command: &Command) -> (String, Params) {
	let args: Vec<&str> = command.args.iter().map(|a| a.value.as_str()).collect();

	if args.is_empty() {
		return (String::new(), Params::Named(Arc::new(HashMap::new())));
	}

	let rql = args[0].to_string();
	let mut params = HashMap::new();

	for arg in &args[1..] {
		if let Some((name, value)) = arg.split_once('=') {
			params.insert(name.to_string(), parse_param_value(value));
		}
	}

	(rql, Params::Named(Arc::new(params)))
}

#[allow(dead_code)]
fn parse_param_value(s: &str) -> Value {
	if let Ok(i) = s.parse::<i32>() {
		return Value::Int4(i);
	}
	if let Ok(i) = s.parse::<i64>() {
		return Value::Int8(i);
	}
	if let Ok(f) = s.parse::<f64>() {
		if let Ok(ordered) = reifydb_client::OrderedF64::try_from(f) {
			return Value::Float8(ordered);
		}
	}

	if s == "true" {
		return Value::Boolean(true);
	}
	if s == "false" {
		return Value::Boolean(false);
	}

	if s.starts_with('\'') && s.ends_with('\'') && s.len() > 1 {
		return Value::Utf8(s[1..s.len() - 1].to_string());
	}
	if s.starts_with('"') && s.ends_with('"') && s.len() > 1 {
		return Value::Utf8(s[1..s.len() - 1].to_string());
	}

	Value::Utf8(s.to_string())
}

#[allow(dead_code)]
pub fn write_frames(frames: Vec<Frame>) -> Result<String, Box<dyn Error>> {
	let mut output = String::new();
	for frame in frames {
		writeln!(output, "{}", frame).unwrap();
	}
	Ok(output)
}