use crate::utils::clock::Instant;
#[cfg(feature = "sqlite")]
use anyhow::{Context, Result};
#[cfg(feature = "sqlite")]
use rusqlite::Connection;
#[cfg(feature = "sqlite")]
use std::collections::HashMap;
#[cfg(feature = "sqlite")]
use std::path::PathBuf;
#[cfg(feature = "sqlite")]
use std::sync::Mutex;
use crate::state::{ClientId, ServerId};
pub const MEMORY_DATABASE_PATH: &str = ":memory:";
pub const MAX_DATABASE_NAME_LEN: usize = 64;
pub const MAX_QUERY_ROWS: usize = 10_000;
pub const MAX_QUERY_RESULT_BYTES: usize = 8 * 1024 * 1024;
pub const MAX_QUERY_VALUE_BYTES: i32 = 8 * 1024 * 1024;
pub const MAX_QUERY_SQL_BYTES: i32 = 1024 * 1024;
pub const MAX_QUERY_VM_STEPS: usize = 10_000_000;
pub const MAX_QUERY_DURATION: std::time::Duration = std::time::Duration::from_secs(5);
#[cfg(feature = "sqlite")]
fn quoted_identifier(name: &str) -> String {
format!("\"{}\"", name.replace('"', "\"\""))
}
pub fn validate_database_name(name: &str) -> anyhow::Result<()> {
if name.is_empty() {
anyhow::bail!(
"Invalid database name: name must not be empty. \
Allowed: 1-{} characters from A-Z, a-z, 0-9, '_' and '-'.",
MAX_DATABASE_NAME_LEN
);
}
if name.len() > MAX_DATABASE_NAME_LEN {
anyhow::bail!(
"Invalid database name '{}': {} bytes exceeds the {}-byte limit.",
name,
name.len(),
MAX_DATABASE_NAME_LEN
);
}
if let Some(bad) = name
.chars()
.find(|c| !(c.is_ascii_alphanumeric() || *c == '_' || *c == '-'))
{
anyhow::bail!(
"Invalid database name '{}': character {:?} is not allowed. \
A database name may contain only A-Z, a-z, 0-9, '_' and '-' \
(it becomes the filename './netget_db_<name>.db').",
name,
bad
);
}
Ok(())
}
pub fn database_file_path(name: &str) -> anyhow::Result<String> {
validate_database_name(name)?;
Ok(format!("./netget_db_{}.db", name))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct DatabaseId(u32);
impl DatabaseId {
pub fn new(id: u32) -> Self {
Self(id)
}
pub fn as_u32(&self) -> u32 {
self.0
}
pub fn from_string(s: &str) -> Option<Self> {
let s = s.trim();
let id_str = s.strip_prefix("db-").unwrap_or(s);
id_str.parse::<u32>().ok().map(Self)
}
}
impl std::fmt::Display for DatabaseId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "db-{}", self.0)
}
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum DatabaseOwner {
Server(ServerId),
Client(ClientId),
Global,
}
impl std::fmt::Display for DatabaseOwner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Server(id) => write!(f, "Server {}", id),
Self::Client(id) => write!(f, "Client {}", id),
Self::Global => write!(f, "Global"),
}
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct TableSchema {
pub name: String,
pub columns: Vec<String>,
pub row_count: u64,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct DatabaseInstance {
pub id: DatabaseId,
pub name: String,
pub path: String,
pub owner: DatabaseOwner,
pub tables: Vec<TableSchema>,
#[serde(skip, default = "Instant::now")]
pub created_at: Instant,
#[serde(skip, default)]
pub last_query_at: Option<Instant>,
pub query_count: u64,
}
impl DatabaseInstance {
#[cfg(feature = "sqlite")]
pub fn new(id: DatabaseId, name: String, path: String, owner: DatabaseOwner) -> Self {
Self {
id,
name,
path,
owner,
tables: Vec::new(),
created_at: Instant::now(),
last_query_at: None,
query_count: 0,
}
}
pub fn is_memory(&self) -> bool {
self.path == ":memory:"
}
#[cfg(feature = "sqlite")]
pub fn refresh_schema(&mut self, conn: &Connection) -> Result<()> {
self.tables.clear();
let mut stmt = conn.prepare(
"SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
)?;
let table_names: Vec<String> = stmt
.query_map([], |row| row.get(0))?
.collect::<rusqlite::Result<_>>()?;
for table_name in table_names {
let mut columns = Vec::new();
let quoted_name = quoted_identifier(&table_name);
let mut stmt = conn.prepare(&format!("PRAGMA table_info({quoted_name})"))?;
let rows = stmt.query_map([], |row| {
let name: String = row.get(1)?;
let type_name: String = row.get(2)?;
let not_null: i32 = row.get(3)?;
let pk: i32 = row.get(5)?;
let mut col_def = format!("{} {}", name, type_name);
if pk > 0 {
col_def.push_str(" PRIMARY KEY");
} else if not_null != 0 {
col_def.push_str(" NOT NULL");
}
Ok(col_def)
})?;
for row in rows {
columns.push(row?);
}
let row_count: u64 =
conn.query_row(&format!("SELECT COUNT(*) FROM {quoted_name}"), [], |row| {
row.get(0)
})?;
self.tables.push(TableSchema {
name: table_name,
columns,
row_count,
});
}
Ok(())
}
pub fn schema_summary(&self) -> String {
if self.tables.is_empty() {
return format!("{} ({}): No tables", self.name, self.id);
}
let mut summary = format!("{} ({}):\n", self.name, self.id);
for table in &self.tables {
summary.push_str(&format!(" - {} ({} rows)\n", table.name, table.row_count));
for column in &table.columns {
summary.push_str(&format!(" {}\n", column));
}
}
summary
}
#[cfg(feature = "sqlite")]
pub fn record_query(&mut self) {
self.query_count += 1;
self.last_query_at = Some(Instant::now());
}
#[cfg(feature = "sqlite")]
pub fn update_row_counts(&mut self, conn: &Connection) -> Result<()> {
for table in &mut self.tables {
table.row_count = conn.query_row(
&format!("SELECT COUNT(*) FROM {}", quoted_identifier(&table.name)),
[],
|row| row.get(0),
)?;
}
Ok(())
}
}
#[cfg(feature = "sqlite")]
pub struct DatabaseConnection {
conn: Mutex<Connection>,
instance: DatabaseInstance,
}
#[cfg(feature = "sqlite")]
fn filesystem_authorizer(ctx: rusqlite::hooks::AuthContext<'_>) -> rusqlite::hooks::Authorization {
use rusqlite::hooks::{AuthAction, Authorization};
match ctx.action {
AuthAction::Attach { filename } => {
if filename.is_empty() || filename == MEMORY_DATABASE_PATH {
Authorization::Allow
} else {
Authorization::Deny
}
}
AuthAction::Unknown { code, .. } if code == rusqlite::ffi::SQLITE_ATTACH => {
Authorization::Deny
}
AuthAction::Function { function_name }
if function_name.eq_ignore_ascii_case("load_extension") =>
{
Authorization::Deny
}
_ => Authorization::Allow,
}
}
#[cfg(feature = "sqlite")]
impl DatabaseConnection {
pub fn new(instance: DatabaseInstance) -> Result<Self> {
let conn =
Connection::open(&instance.path).context("Failed to open database connection")?;
conn.authorizer(Some(filesystem_authorizer));
conn.set_limit(
rusqlite::limits::Limit::SQLITE_LIMIT_LENGTH,
MAX_QUERY_VALUE_BYTES,
)
.context("Failed to set SQLite value size limit")?;
conn.set_limit(
rusqlite::limits::Limit::SQLITE_LIMIT_SQL_LENGTH,
MAX_QUERY_SQL_BYTES,
)
.context("Failed to set SQLite SQL size limit")?;
Ok(Self {
conn: Mutex::new(conn),
instance,
})
}
pub fn instance(&self) -> &DatabaseInstance {
&self.instance
}
pub fn instance_mut(&mut self) -> &mut DatabaseInstance {
&mut self.instance
}
pub fn execute_query(&mut self, sql: &str) -> Result<QueryResult> {
let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
self.instance.record_query();
let started = std::time::Instant::now();
let mut steps = 0usize;
conn.progress_handler(
1000,
Some(move || {
steps += 1000;
steps >= MAX_QUERY_VM_STEPS || started.elapsed() >= MAX_QUERY_DURATION
}),
);
let result = (|| -> Result<QueryResult> {
use rusqlite::fallible_iterator::FallibleIterator;
let mut batch = rusqlite::Batch::new(&conn, sql);
let mut stmt = batch.next()?.context("query contains no SQL statement")?;
anyhow::ensure!(
batch.next()?.is_none(),
"execute_query accepts exactly one SQL statement"
);
let readonly = stmt.readonly();
let result = if stmt.column_count() > 0 {
let column_names: Vec<String> = stmt
.column_names()
.iter()
.map(|name| name.to_string())
.collect();
let mut bytes: usize = column_names
.iter()
.map(|name| name.len().saturating_mul(6).saturating_add(3))
.sum();
let mut rows = Vec::new();
let mut query_rows = stmt.query([])?;
while let Some(row) = query_rows.next()? {
anyhow::ensure!(rows.len() < MAX_QUERY_ROWS, "query result exceeds {} rows; use LIMIT (a modifying statement may already have executed)", MAX_QUERY_ROWS);
let mut values = Vec::with_capacity(column_names.len());
bytes = bytes.saturating_add(2);
for i in 0..column_names.len() {
use rusqlite::types::ValueRef;
let value = row.get_ref(i)?;
let cost = match value {
ValueRef::Text(text) => text.len().saturating_mul(6).saturating_add(3),
ValueRef::Blob(blob) => blob.len().saturating_mul(2).saturating_add(3),
_ => 32,
};
bytes = bytes.saturating_add(cost);
anyhow::ensure!(bytes <= MAX_QUERY_RESULT_BYTES, "query result exceeds {} bytes; select fewer/smaller values (a modifying statement may already have executed)", MAX_QUERY_RESULT_BYTES);
values.push(match value {
ValueRef::Null => serde_json::Value::Null,
ValueRef::Integer(value) => serde_json::json!(value),
ValueRef::Real(value) => serde_json::json!(value),
ValueRef::Text(value) => serde_json::Value::String(
String::from_utf8_lossy(value).into_owned(),
),
ValueRef::Blob(value) => serde_json::Value::String(hex::encode(value)),
});
}
rows.push(values);
}
QueryResult::Select {
columns: column_names,
rows,
}
} else {
QueryResult::Modified {
affected_rows: stmt.execute([])?,
}
};
drop(stmt);
if !readonly {
self.instance.refresh_schema(&conn)?;
}
Ok(result)
})();
conn.progress_handler(0, None::<fn() -> bool>);
result.map_err(|error| anyhow::anyhow!("SQLite query failed: {error:#}"))
}
pub fn refresh_schema(&mut self) -> Result<()> {
let conn = self.conn.lock().unwrap_or_else(|e| e.into_inner());
self.instance.refresh_schema(&conn)
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type")]
pub enum QueryResult {
Select {
columns: Vec<String>,
rows: Vec<Vec<serde_json::Value>>,
},
Modified {
affected_rows: usize,
},
}
impl QueryResult {
pub fn format(&self) -> String {
match self {
Self::Select { columns, rows } => {
if rows.is_empty() {
return format!("No rows returned. Columns: {}", columns.join(", "));
}
let mut output = String::new();
output.push_str(&columns.join(" | "));
output.push('\n');
output.push_str(&"-".repeat(output.len()));
output.push('\n');
for row in rows {
let row_str: Vec<String> = row
.iter()
.map(|v| match v {
serde_json::Value::Null => "NULL".to_string(),
serde_json::Value::String(s) => s.clone(),
other => other.to_string(),
})
.collect();
output.push_str(&row_str.join(" | "));
output.push('\n');
}
output
}
Self::Modified { affected_rows } => {
format!("{} row(s) affected", affected_rows)
}
}
}
pub fn row_count(&self) -> usize {
match self {
Self::Select { rows, .. } => rows.len(),
Self::Modified { affected_rows } => *affected_rows,
}
}
}
#[cfg(feature = "sqlite")]
pub struct DatabaseManager {
connections: HashMap<DatabaseId, DatabaseConnection>,
}
#[cfg(feature = "sqlite")]
impl DatabaseManager {
pub fn new() -> Self {
Self {
connections: HashMap::new(),
}
}
pub fn create_database(
&mut self,
id: DatabaseId,
name: String,
path: String,
owner: DatabaseOwner,
init_sql: Option<&str>,
) -> Result<()> {
validate_database_name(&name)?;
let instance = DatabaseInstance::new(id, name, path, owner);
let mut conn = DatabaseConnection::new(instance)?;
if let Some(sql) = init_sql {
let db_conn = conn.conn.lock().unwrap_or_else(|e| e.into_inner());
db_conn.execute_batch(sql)?;
drop(db_conn);
}
conn.refresh_schema()?;
self.connections.insert(id, conn);
Ok(())
}
pub fn get_instance(&self, id: DatabaseId) -> Option<&DatabaseInstance> {
self.connections.get(&id).map(|conn| conn.instance())
}
pub fn get_all_instances(&self) -> Vec<&DatabaseInstance> {
self.connections
.values()
.map(|conn| conn.instance())
.collect()
}
pub fn execute_query(&mut self, id: DatabaseId, sql: &str) -> Result<QueryResult> {
let conn = self
.connections
.get_mut(&id)
.context("Database not found")?;
conn.execute_query(sql)
}
pub fn delete_database(&mut self, id: DatabaseId) -> Result<()> {
let conn = self.connections.remove(&id).context("Database not found")?;
if !conn.instance().is_memory() {
let path = PathBuf::from(&conn.instance().path);
if path.exists() {
std::fs::remove_file(&path).context("Failed to delete database file")?;
}
}
Ok(())
}
pub fn get_databases_by_server(&self, server_id: ServerId) -> Vec<&DatabaseInstance> {
self.connections
.values()
.map(|conn| conn.instance())
.filter(|instance| instance.owner == DatabaseOwner::Server(server_id))
.collect()
}
pub fn get_databases_by_client(&self, client_id: ClientId) -> Vec<&DatabaseInstance> {
self.connections
.values()
.map(|conn| conn.instance())
.filter(|instance| instance.owner == DatabaseOwner::Client(client_id))
.collect()
}
pub fn delete_databases_by_server(&mut self, server_id: ServerId) -> Result<()> {
let db_ids: Vec<DatabaseId> = self
.connections
.values()
.filter(|conn| conn.instance().owner == DatabaseOwner::Server(server_id))
.map(|conn| conn.instance().id)
.collect();
for id in db_ids {
self.delete_database(id)?;
}
Ok(())
}
pub fn delete_databases_by_client(&mut self, client_id: ClientId) -> Result<()> {
let db_ids: Vec<DatabaseId> = self
.connections
.values()
.filter(|conn| conn.instance().owner == DatabaseOwner::Client(client_id))
.map(|conn| conn.instance().id)
.collect();
for id in db_ids {
self.delete_database(id)?;
}
Ok(())
}
}
#[cfg(not(feature = "sqlite"))]
pub struct DatabaseManager;
#[cfg(not(feature = "sqlite"))]
impl DatabaseManager {
pub fn new() -> Self {
Self
}
}