//! Portable workspace lifecycle for local structured-data workflows.
//!
//! A workspace is a directory with a versioned manifest and one Basalt
//! database. Import and export deliberately use common text formats so a
//! workspace is inspectable and recoverable without Basalt-specific tooling.
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::ffi::OsStr;
use std::fmt;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Read, Write};
use std::path::{Component, Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use csv::{ReaderBuilder, StringRecord, Writer};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value as JsonValue};
use sha2::{Digest, Sha256};
use crate::database::Database;
use crate::db::{Column, DbError, StatementResult};
use crate::engine::{ExecutionBudget, MCP_EXECUTION_WORK_LIMIT};
use crate::sql::ast::Statement;
use crate::sql::parser::parse;
use crate::storage;
use crate::types::{ColumnType, Value};
const MANIFEST_FILE: &str = "workspace.json";
const DATABASE_FILE: &str = "data.basalt";
const WORKSPACE_LOCK_FILE: &str = ".workspace.lock";
const FORMAT_VERSION: u32 = 1;
const MAX_IMPORT_BYTES: u64 = 64 * 1024 * 1024;
const MAX_MANIFEST_BYTES: u64 = 64 * 1024;
const MAX_HISTORY_METADATA_BYTES: u64 = 4 * 1024 * 1024;
pub(crate) const MAX_MCP_IMPORT_BYTES: usize = 16 * 1024 * 1024;
const MAX_MCP_IMPORT_ROWS: usize = 10_000;
const MAX_MCP_IMPORT_COLUMNS: usize = 256;
const MAX_MCP_IMPORT_CELLS: usize = 1_000_000;
const IMPORT_BATCH_SIZE: usize = 256;
const MAX_PREVIEW_BYTES: usize = 1024 * 1024;
const MAX_PREVIEW_STATEMENTS: usize = 64;
const MAX_PREVIEW_MUTATIONS: usize = 32;
const MAX_PREVIEW_ROWS: usize = 10_000;
const MAX_MCP_MUTATION_ROWS: usize = 10_000;
const MAX_MCP_DIFF_ROWS: usize = 10_000;
const MAX_MCP_EXPORT_ROWS: usize = 10_000;
const MAX_MCP_HISTORY_ENTRIES: usize = 10_000;
const MAX_MCP_HISTORY_METADATA_BYTES: u64 = 1024 * 1024;
const HISTORY_DIR: &str = "history";
const PLANS_DIR: &str = "plans";
const CHANGES_DIR: &str = "changes";
const SNAPSHOTS_DIR: &str = "snapshots";
static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
pub const HELP: &str = "Basalt workspace — local, portable SQL workspaces\n\n\
Usage:\n basalt workspace <COMMAND> [OPTIONS]\n\n\
Commands:\n init PATH Create a workspace\n inspect [--json] PATH Show workspace metadata and schema\n query [OPTIONS] PATH SQL Run a read-only query\n preview [--json] PATH SQL Preview a write and save its plan\n plan [--json] PATH PLAN_ID Load a saved preview plan\n apply [--json] PATH PLAN_ID Apply one exact preview plan\n history [--json] PATH List applied and recoverable changes\n diff [--json] PATH [CHANGE_ID] Compare a change recovery point\n undo [--json] PATH CHANGE_ID Undo the latest change safely\n import [OPTIONS] WORKSPACE SOURCE Import CSV, JSON, JSONL, or SQL\n export [OPTIONS] WORKSPACE TABLE OUTPUT\n Export CSV, JSONL, or SQL\n\n\
Import options:\n --table NAME Table name (required for stdin)\n --format csv|json|jsonl|sql Override format inference\n --json Emit a machine-readable import report\n\n\
Export options:\n --format csv|jsonl|sql Override format inference\n --json Emit a machine-readable export report\n\n\
Query options:\n --output table|csv|json Result format (table by default)\n\n\
State options:\n --json Emit machine-readable JSON\n\n\
SOURCE and OUTPUT may be '-' for stdin/stdout. File extensions infer formats.\n\
Imports are atomic. Workspace data stays local and uses a versioned manifest.\n";
#[derive(Debug)]
pub enum WorkspaceError {
Usage(String),
Invalid(String),
Io(io::Error),
Database(DbError),
Json(serde_json::Error),
Csv(csv::Error),
}
impl WorkspaceError {
pub fn exit_code(&self) -> i32 {
match self {
WorkspaceError::Usage(_) => 2,
WorkspaceError::Invalid(_)
| WorkspaceError::Io(_)
| WorkspaceError::Database(_)
| WorkspaceError::Json(_)
| WorkspaceError::Csv(_) => 1,
}
}
}
impl fmt::Display for WorkspaceError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
WorkspaceError::Usage(message) | WorkspaceError::Invalid(message) => {
write!(f, "{message}")
}
WorkspaceError::Io(error) => write!(f, "{error}"),
WorkspaceError::Database(error) => write!(f, "{error}"),
WorkspaceError::Json(error) => write!(f, "{error}"),
WorkspaceError::Csv(error) => write!(f, "{error}"),
}
}
}
impl std::error::Error for WorkspaceError {}
impl From<io::Error> for WorkspaceError {
fn from(error: io::Error) -> Self {
WorkspaceError::Io(error)
}
}
impl From<DbError> for WorkspaceError {
fn from(error: DbError) -> Self {
WorkspaceError::Database(error)
}
}
impl From<serde_json::Error> for WorkspaceError {
fn from(error: serde_json::Error) -> Self {
WorkspaceError::Json(error)
}
}
impl From<csv::Error> for WorkspaceError {
fn from(error: csv::Error) -> Self {
WorkspaceError::Csv(error)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkspaceManifest {
pub format_version: u32,
pub database: String,
}
#[derive(Debug, Clone)]
pub struct Workspace {
root: PathBuf,
manifest: WorkspaceManifest,
_lock: Arc<File>,
}
impl Workspace {
pub fn init(path: impl AsRef<Path>) -> Result<Workspace, WorkspaceError> {
let root = path.as_ref().to_path_buf();
let root_preexisting = root.exists();
if root.as_os_str().is_empty() {
return Err(WorkspaceError::Invalid(
"workspace path cannot be empty".to_string(),
));
}
if path_is_symlink(&root)? {
return Err(WorkspaceError::Invalid(
"workspace path cannot be a symbolic link".to_string(),
));
}
if root.exists() && !root.is_dir() {
return Err(WorkspaceError::Invalid(format!(
"workspace path is not a directory: {}",
root.display()
)));
}
fs::create_dir_all(&root)?;
let workspace_lock_path = root.join(WORKSPACE_LOCK_FILE);
let lock_preexisting = match fs::symlink_metadata(&workspace_lock_path) {
Ok(_) => true,
Err(error) if error.kind() == io::ErrorKind::NotFound => false,
Err(error) => return Err(WorkspaceError::Io(error)),
};
let lock = acquire_workspace_lock(&root)?;
let manifest_path = root.join(MANIFEST_FILE);
let database_path = root.join(DATABASE_FILE);
let manifest = WorkspaceManifest {
format_version: FORMAT_VERSION,
database: DATABASE_FILE.to_string(),
};
let mut manifest_created = false;
let initialization = (|| {
if path_is_symlink(&manifest_path)? || manifest_path.exists() {
return Err(WorkspaceError::Invalid(format!(
"workspace already exists: {}",
root.display()
)));
}
if path_is_symlink(&database_path)? || database_path.exists() {
return Err(WorkspaceError::Invalid(format!(
"reserved database path already exists: {}",
database_path.display()
)));
}
for suffix in [".wal", ".lock", ".tmp"] {
let path = sidecar_path(&database_path, suffix);
if path_is_symlink(&path)? || path.exists() {
return Err(WorkspaceError::Invalid(format!(
"reserved database sidecar already exists: {}",
path.display()
)));
}
}
write_new_file(&manifest_path, &manifest_bytes(&manifest)?)?;
manifest_created = true;
let database = Database::open_in_workspace(&database_path)?;
database.checkpoint()?;
Ok::<(), WorkspaceError>(())
})();
if let Err(error) = initialization {
drop(lock);
if let Err(cleanup_error) = cleanup_failed_init(
&root,
manifest_created,
!lock_preexisting,
!root_preexisting,
) {
return Err(WorkspaceError::Invalid(format!(
"workspace initialization failed: {error}; cleanup also failed: {cleanup_error}"
)));
}
return Err(error);
}
Ok(Workspace {
root,
manifest,
_lock: lock,
})
}
pub fn open(path: impl AsRef<Path>) -> Result<Workspace, WorkspaceError> {
let root = path.as_ref().to_path_buf();
if path_is_symlink(&root)? {
return Err(WorkspaceError::Invalid(
"workspace path cannot be a symbolic link".to_string(),
));
}
if !root.is_dir() {
return Err(WorkspaceError::Invalid(format!(
"workspace directory does not exist: {}",
root.display()
)));
}
let manifest_path = root.join(MANIFEST_FILE);
if path_is_symlink(&manifest_path)? {
return Err(WorkspaceError::Invalid(
"workspace manifest cannot be a symbolic link".to_string(),
));
}
let bytes = read_file_limited(&manifest_path, MAX_MANIFEST_BYTES, "workspace manifest")
.map_err(|error| {
if matches!(&error, WorkspaceError::Io(error) if error.kind() == io::ErrorKind::NotFound)
{
WorkspaceError::Invalid(format!(
"not a Basalt workspace: missing {}",
manifest_path.display()
))
} else {
error
}
})?;
let manifest: WorkspaceManifest = serde_json::from_slice(&bytes).map_err(|error| {
WorkspaceError::Invalid(format!(
"invalid workspace manifest {}: {error}",
manifest_path.display()
))
})?;
if manifest.format_version != FORMAT_VERSION {
return Err(WorkspaceError::Invalid(format!(
"unsupported workspace format version {}; expected {}",
manifest.format_version, FORMAT_VERSION
)));
}
if manifest.database != DATABASE_FILE {
return Err(WorkspaceError::Invalid(format!(
"workspace database must be {DATABASE_FILE:?}, got {:?}",
manifest.database
)));
}
let lock = acquire_workspace_lock(&root)?;
if let Err(error) = validate_database_paths(&root) {
drop(lock);
return Err(error);
}
Ok(Workspace {
root,
manifest,
_lock: lock,
})
}
/// Open a workspace, creating it only when the requested path is missing.
pub fn open_or_init(path: impl AsRef<Path>) -> Result<Workspace, WorkspaceError> {
let path = path.as_ref().to_path_buf();
match Self::open(&path) {
Ok(workspace) => Ok(workspace),
Err(_error) if !path.exists() => Self::init(path),
Err(error) => Err(error),
}
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn manifest(&self) -> &WorkspaceManifest {
&self.manifest
}
pub fn database_path(&self) -> PathBuf {
self.root.join(DATABASE_FILE)
}
pub fn database(&self) -> Result<Database, WorkspaceError> {
let path = self.database_path();
validate_database_paths(&self.root)?;
Ok(Database::open_in_workspace(path)?)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DataFormat {
Csv,
Json,
JsonLines,
Sql,
}
impl DataFormat {
fn parse(value: &str) -> Result<DataFormat, WorkspaceError> {
match value.to_ascii_lowercase().as_str() {
"csv" => Ok(DataFormat::Csv),
"json" => Ok(DataFormat::Json),
"jsonl" | "ndjson" => Ok(DataFormat::JsonLines),
"sql" => Ok(DataFormat::Sql),
_ => Err(WorkspaceError::Usage(format!(
"unknown format {value:?}; expected csv, json, jsonl, or sql"
))),
}
}
fn from_path(path: &Path) -> Option<DataFormat> {
let extension = path
.extension()
.and_then(OsStr::to_str)?
.to_ascii_lowercase();
match extension.as_str() {
"csv" => Some(DataFormat::Csv),
"json" => Some(DataFormat::Json),
"jsonl" | "ndjson" => Some(DataFormat::JsonLines),
"sql" => Some(DataFormat::Sql),
_ => None,
}
}
fn name(self) -> &'static str {
match self {
DataFormat::Csv => "csv",
DataFormat::Json => "json",
DataFormat::JsonLines => "jsonl",
DataFormat::Sql => "sql",
}
}
}
#[derive(Debug)]
enum Command {
Help,
Init(PathBuf),
Inspect {
json: bool,
workspace: PathBuf,
},
Query {
workspace: PathBuf,
sql: String,
output: crate::cli::OutputMode,
},
Preview {
workspace: PathBuf,
sql: String,
json: bool,
},
Plan {
workspace: PathBuf,
plan_id: String,
json: bool,
},
Apply {
workspace: PathBuf,
plan_id: String,
json: bool,
},
History {
workspace: PathBuf,
json: bool,
},
Diff {
workspace: PathBuf,
change_id: Option<String>,
json: bool,
},
Undo {
workspace: PathBuf,
change_id: String,
json: bool,
},
Import {
workspace: PathBuf,
source: PathBuf,
table: Option<String>,
format: Option<DataFormat>,
json: bool,
},
Export {
workspace: PathBuf,
table: String,
output: PathBuf,
format: Option<DataFormat>,
json: bool,
},
}
pub fn run<R: Read, W: Write>(
args: &[String],
input: &mut R,
output: &mut W,
) -> Result<(), WorkspaceError> {
if args.is_empty() || args.iter().any(|arg| arg == "--help") {
output.write_all(HELP.as_bytes())?;
return Ok(());
}
let command = parse_command(args)?;
match command {
Command::Help => output.write_all(HELP.as_bytes())?,
Command::Init(path) => {
let workspace = Workspace::init(path)?;
writeln!(output, "created workspace {}", workspace.root().display())?;
}
Command::Inspect { json, workspace } => {
let workspace = Workspace::open(workspace)?;
let database = workspace.database()?;
let report = inspect(&workspace, &database)?;
if json {
serde_json::to_writer_pretty(&mut *output, &report)?;
output.write_all(b"\n")?;
} else {
render_inspect(&report, output)?;
}
}
Command::Query {
workspace,
sql,
output: output_mode,
} => {
let workspace = Workspace::open(workspace)?;
let statements = parse(&sql).map_err(|error| {
WorkspaceError::Invalid(format!(
"query parse error at byte {}: {}",
error.offset, error.message
))
})?;
if statements.is_empty() || statements.iter().any(|statement| !is_read_only(statement))
{
return Err(WorkspaceError::Invalid(
"workspace query only accepts SELECT and EXPLAIN SELECT".to_string(),
));
}
let options = crate::cli::CliOptions {
database: workspace.database_path().display().to_string(),
actions: vec![crate::cli::InputAction::Command(sql)],
output: output_mode,
headers: true,
quiet: false,
help: false,
version: false,
};
let database = workspace.database()?;
let mut empty_input = io::Cursor::new(Vec::<u8>::new());
crate::cli::run(&options, database, &mut empty_input, output)
.map_err(|error| WorkspaceError::Invalid(error.to_string()))?;
}
Command::Preview {
workspace,
sql,
json,
} => {
let workspace = Workspace::open(workspace)?;
let plan = preview_plan(&workspace, &sql)?;
if json {
let report = PlanReport::from(&plan);
serde_json::to_writer_pretty(&mut *output, &report)?;
output.write_all(b"\n")?;
} else {
render_plan(&workspace, &plan, output)?;
}
}
Command::Plan {
workspace,
plan_id,
json,
} => {
let workspace = Workspace::open(workspace)?;
let plan = load_plan(&workspace, &plan_id)?;
if json {
let report = PlanReport::from(&plan);
serde_json::to_writer_pretty(&mut *output, &report)?;
output.write_all(b"\n")?;
} else {
render_plan(&workspace, &plan, output)?;
}
}
Command::Apply {
workspace,
plan_id,
json,
} => {
let workspace = Workspace::open(workspace)?;
let report = apply_plan(&workspace, &plan_id, None, None)?;
if json {
serde_json::to_writer_pretty(&mut *output, &report)?;
output.write_all(b"\n")?;
} else {
render_apply(&report, output)?;
}
}
Command::History { workspace, json } => {
let workspace = Workspace::open(workspace)?;
let entries = history(&workspace)?;
if json {
serde_json::to_writer_pretty(&mut *output, &entries)?;
output.write_all(b"\n")?;
} else {
render_history(&entries, output)?;
}
}
Command::Diff {
workspace,
change_id,
json,
} => {
let workspace = Workspace::open(workspace)?;
let report = diff(&workspace, change_id.as_deref())?;
if json {
serde_json::to_writer_pretty(&mut *output, &report)?;
output.write_all(b"\n")?;
} else {
render_diff(&report, output)?;
}
}
Command::Undo {
workspace,
change_id,
json,
} => {
let workspace = Workspace::open(workspace)?;
let report = undo(&workspace, &change_id, None)?;
if json {
serde_json::to_writer_pretty(&mut *output, &report)?;
output.write_all(b"\n")?;
} else {
render_undo(&report, output)?;
}
}
Command::Import {
workspace,
source,
table,
format,
json,
} => {
let workspace = Workspace::open(workspace)?;
let format = format
.or_else(|| DataFormat::from_path(&source))
.ok_or_else(|| {
WorkspaceError::Usage(
"cannot infer import format; provide --format or use a supported extension"
.to_string(),
)
})?;
let bytes = read_source(&source, input)?;
let table_name = if format == DataFormat::Sql {
None
} else {
table.or_else(|| inferred_table_name(&source))
};
let imported = import_with_recovery(
&workspace,
table_name.as_deref(),
format,
&bytes,
ImportLimits::unbounded(),
)?;
if json {
let report = CliImportReport {
operation: "import",
workspace: workspace.root.display().to_string(),
source: source.display().to_string(),
format: imported.format.clone(),
table: imported.table.clone(),
bytes: bytes.len(),
change_id: imported.change_id.clone(),
summary: imported.summary.clone(),
};
serde_json::to_writer_pretty(&mut *output, &report)?;
output.write_all(b"\n")?;
} else {
writeln!(
output,
"imported {}: {} (change {})",
imported.format, imported.summary, imported.change_id
)?;
}
}
Command::Export {
workspace,
table,
output: destination,
format,
json,
} => {
let workspace = Workspace::open(workspace)?;
let format = format
.or_else(|| DataFormat::from_path(&destination))
.ok_or_else(|| {
WorkspaceError::Usage(
"cannot infer export format; provide --format or use a supported extension"
.to_string(),
)
})?;
if format == DataFormat::Json {
return Err(WorkspaceError::Usage(
"JSON export is JSON Lines; use --format jsonl or a .jsonl file".to_string(),
));
}
if json && destination.as_os_str() == OsStr::new("-") {
return Err(WorkspaceError::Usage(
"--json cannot be combined with stdout export; write the export to a file"
.to_string(),
));
}
let database = workspace.database()?;
let (columns, rows) = select_table(&database, &table)?;
let bytes = match format {
DataFormat::Csv => export_csv(&columns, &rows)?,
DataFormat::JsonLines => export_json_lines(&columns, &rows)?,
DataFormat::Sql => export_sql(&database, &table, &rows)?,
DataFormat::Json => unreachable!(),
};
write_output(&workspace, &destination, &bytes, output)?;
if json {
let report = CliExportReport {
operation: "export",
workspace: workspace.root.display().to_string(),
table,
format: format.name().to_string(),
output: destination.display().to_string(),
rows: rows.len(),
bytes: bytes.len(),
};
serde_json::to_writer_pretty(&mut *output, &report)?;
output.write_all(b"\n")?;
} else if destination.as_os_str() == OsStr::new("-") {
output.flush()?;
} else {
writeln!(
output,
"exported {} rows to {}",
rows.len(),
destination.display()
)?;
}
}
}
output.flush()?;
Ok(())
}
fn parse_command(args: &[String]) -> Result<Command, WorkspaceError> {
match args.first().map(String::as_str) {
Some("init") => parse_init(&args[1..]),
Some("inspect") => parse_inspect(&args[1..]),
Some("query") => parse_query(&args[1..]),
Some("preview") => parse_preview(&args[1..]),
Some("plan") => parse_plan(&args[1..]),
Some("apply") => parse_apply(&args[1..]),
Some("history") => parse_history(&args[1..]),
Some("diff") => parse_diff(&args[1..]),
Some("undo") => parse_undo(&args[1..]),
Some("import") => parse_import(&args[1..]),
Some("export") => parse_export(&args[1..]),
Some("--help") => Ok(Command::Help),
Some(command) => Err(WorkspaceError::Usage(format!(
"unknown workspace command {command:?}; run `basalt workspace --help`"
))),
None => Err(WorkspaceError::Usage(
"missing workspace command".to_string(),
)),
}
}
fn parse_init(args: &[String]) -> Result<Command, WorkspaceError> {
if args.len() == 1 && args[0] != "--help" {
return Ok(Command::Init(PathBuf::from(&args[0])));
}
Err(WorkspaceError::Usage(
"usage: basalt workspace init PATH".to_string(),
))
}
fn parse_inspect(args: &[String]) -> Result<Command, WorkspaceError> {
let mut json = false;
let mut positional = Vec::new();
let mut options = true;
for arg in args {
if options && arg == "--" {
options = false;
} else if options && arg == "--json" {
json = true;
} else if options && arg == "--help" {
return Err(WorkspaceError::Usage(HELP.to_string()));
} else if options && arg.starts_with('-') {
return Err(WorkspaceError::Usage(format!(
"unknown inspect option {arg:?}"
)));
} else {
positional.push(arg);
}
}
if positional.len() != 1 {
return Err(WorkspaceError::Usage(
"usage: basalt workspace inspect [--json] PATH".to_string(),
));
}
Ok(Command::Inspect {
json,
workspace: PathBuf::from(positional[0]),
})
}
fn parse_query(args: &[String]) -> Result<Command, WorkspaceError> {
let mut output = crate::cli::OutputMode::Table;
let mut positional = Vec::new();
let mut index = 0;
let mut options = true;
while index < args.len() {
let arg = &args[index];
if options && arg == "--" {
options = false;
} else if options && arg == "--json" {
output = crate::cli::OutputMode::Json;
} else if options && arg == "--csv" {
output = crate::cli::OutputMode::Csv;
} else if options && arg == "--table" {
output = crate::cli::OutputMode::Table;
} else if options && arg == "--output" {
index += 1;
output = parse_query_output(&option_value(args, index, "--output")?)?;
} else if options && arg == "--help" {
return Err(WorkspaceError::Usage(HELP.to_string()));
} else if options && arg.starts_with('-') {
return Err(WorkspaceError::Usage(format!(
"unknown query option {arg:?}"
)));
} else {
positional.push(arg.clone());
}
index += 1;
}
if positional.len() != 2 {
return Err(WorkspaceError::Usage(
"usage: basalt workspace query [OPTIONS] PATH SQL".to_string(),
));
}
Ok(Command::Query {
workspace: PathBuf::from(&positional[0]),
sql: positional[1].clone(),
output,
})
}
fn parse_query_output(value: &str) -> Result<crate::cli::OutputMode, WorkspaceError> {
match value.to_ascii_lowercase().as_str() {
"table" => Ok(crate::cli::OutputMode::Table),
"csv" => Ok(crate::cli::OutputMode::Csv),
"json" | "jsonl" | "ndjson" => Ok(crate::cli::OutputMode::Json),
_ => Err(WorkspaceError::Usage(format!(
"unknown query output format {value:?}; expected table, csv, or json"
))),
}
}
fn parse_json_flagged_command(
args: &[String],
usage: &str,
positionals: std::ops::RangeInclusive<usize>,
) -> Result<(bool, Vec<String>), WorkspaceError> {
let mut json = false;
let mut positional = Vec::new();
let mut options = true;
for arg in args {
if options && arg == "--" {
options = false;
} else if options && arg == "--json" {
json = true;
} else if options && arg == "--help" {
return Err(WorkspaceError::Usage(HELP.to_string()));
} else if options && arg.starts_with('-') {
return Err(WorkspaceError::Usage(format!("unknown option {arg:?}")));
} else {
positional.push(arg.clone());
}
}
if !positionals.contains(&positional.len()) {
return Err(WorkspaceError::Usage(usage.to_string()));
}
Ok((json, positional))
}
fn parse_preview(args: &[String]) -> Result<Command, WorkspaceError> {
let (json, positional) = parse_json_flagged_command(
args,
"usage: basalt workspace preview [--json] PATH SQL",
2..=2,
)?;
Ok(Command::Preview {
workspace: PathBuf::from(&positional[0]),
sql: positional[1].clone(),
json,
})
}
fn parse_apply(args: &[String]) -> Result<Command, WorkspaceError> {
let (json, positional) = parse_json_flagged_command(
args,
"usage: basalt workspace apply [--json] PATH PLAN_ID",
2..=2,
)?;
Ok(Command::Apply {
workspace: PathBuf::from(&positional[0]),
plan_id: positional[1].clone(),
json,
})
}
fn parse_plan(args: &[String]) -> Result<Command, WorkspaceError> {
let (json, positional) = parse_json_flagged_command(
args,
"usage: basalt workspace plan [--json] PATH PLAN_ID",
2..=2,
)?;
Ok(Command::Plan {
workspace: PathBuf::from(&positional[0]),
plan_id: positional[1].clone(),
json,
})
}
fn parse_history(args: &[String]) -> Result<Command, WorkspaceError> {
let (json, positional) =
parse_json_flagged_command(args, "usage: basalt workspace history [--json] PATH", 1..=1)?;
Ok(Command::History {
workspace: PathBuf::from(&positional[0]),
json,
})
}
fn parse_diff(args: &[String]) -> Result<Command, WorkspaceError> {
let (json, positional) = parse_json_flagged_command(
args,
"usage: basalt workspace diff [--json] PATH [CHANGE_ID]",
1..=2,
)?;
Ok(Command::Diff {
workspace: PathBuf::from(&positional[0]),
change_id: positional.get(1).cloned(),
json,
})
}
fn parse_undo(args: &[String]) -> Result<Command, WorkspaceError> {
let (json, positional) = parse_json_flagged_command(
args,
"usage: basalt workspace undo [--json] PATH CHANGE_ID",
2..=2,
)?;
Ok(Command::Undo {
workspace: PathBuf::from(&positional[0]),
change_id: positional[1].clone(),
json,
})
}
fn parse_import(args: &[String]) -> Result<Command, WorkspaceError> {
let mut table = None;
let mut format = None;
let mut json = false;
let mut positional = Vec::new();
let mut index = 0;
let mut options = true;
while index < args.len() {
let arg = &args[index];
if options && arg == "--" {
options = false;
} else if options && arg == "--table" {
index += 1;
table = Some(option_value(args, index, "--table")?);
} else if options && arg == "--format" {
index += 1;
format = Some(DataFormat::parse(&option_value(args, index, "--format")?)?);
} else if options && arg == "--json" {
json = true;
} else if options && arg == "--help" {
return Err(WorkspaceError::Usage(HELP.to_string()));
} else if options && arg.starts_with('-') && arg != "-" {
return Err(WorkspaceError::Usage(format!(
"unknown import option {arg:?}"
)));
} else {
positional.push(arg.clone());
}
index += 1;
}
if positional.len() != 2 {
return Err(WorkspaceError::Usage(
"usage: basalt workspace import [OPTIONS] WORKSPACE SOURCE".to_string(),
));
}
let source = PathBuf::from(&positional[1]);
if source.as_os_str() == OsStr::new("-") && table.is_none() && format != Some(DataFormat::Sql) {
return Err(WorkspaceError::Usage(
"stdin imports require --table NAME".to_string(),
));
}
if format == Some(DataFormat::Sql) && table.is_some() {
return Err(WorkspaceError::Usage(
"--table is not valid for SQL imports".to_string(),
));
}
Ok(Command::Import {
workspace: PathBuf::from(&positional[0]),
source,
table,
format,
json,
})
}
fn parse_export(args: &[String]) -> Result<Command, WorkspaceError> {
let mut format = None;
let mut json = false;
let mut positional = Vec::new();
let mut index = 0;
let mut options = true;
while index < args.len() {
let arg = &args[index];
if options && arg == "--" {
options = false;
} else if options && arg == "--format" {
index += 1;
format = Some(DataFormat::parse(&option_value(args, index, "--format")?)?);
} else if options && arg == "--json" {
json = true;
} else if options && arg == "--help" {
return Err(WorkspaceError::Usage(HELP.to_string()));
} else if options && arg.starts_with('-') && arg != "-" {
return Err(WorkspaceError::Usage(format!(
"unknown export option {arg:?}"
)));
} else {
positional.push(arg.clone());
}
index += 1;
}
if positional.len() != 3 {
return Err(WorkspaceError::Usage(
"usage: basalt workspace export [OPTIONS] WORKSPACE TABLE OUTPUT".to_string(),
));
}
let output = PathBuf::from(&positional[2]);
if output.as_os_str() == OsStr::new("-") && format.is_none() {
return Err(WorkspaceError::Usage(
"stdout exports require --format FORMAT".to_string(),
));
}
Ok(Command::Export {
workspace: PathBuf::from(&positional[0]),
table: positional[1].clone(),
output,
format,
json,
})
}
fn option_value(args: &[String], index: usize, option: &str) -> Result<String, WorkspaceError> {
args.get(index)
.cloned()
.filter(|value| !value.starts_with('-') || value == "-")
.ok_or_else(|| WorkspaceError::Usage(format!("{option} requires a value")))
}
fn read_source<R: Read>(source: &Path, input: &mut R) -> Result<Vec<u8>, WorkspaceError> {
if source.as_os_str() == OsStr::new("-") {
return read_limited(input);
}
let mut file = File::open(source)?;
read_limited(&mut file)
}
fn read_file_limited(path: &Path, max_bytes: u64, label: &str) -> Result<Vec<u8>, WorkspaceError> {
let metadata = fs::symlink_metadata(path)?;
if metadata.file_type().is_symlink() {
return Err(WorkspaceError::Invalid(format!(
"{label} cannot be a symbolic link: {}",
path.display()
)));
}
if !metadata.is_file() {
return Err(WorkspaceError::Invalid(format!(
"{label} is not a regular file: {}",
path.display()
)));
}
if metadata.len() > max_bytes {
return Err(WorkspaceError::Invalid(format!(
"{label} exceeds the {max_bytes}-byte limit: {}",
path.display()
)));
}
let file = File::open(path)?;
let mut bytes = Vec::new();
file.take(max_bytes.saturating_add(1))
.read_to_end(&mut bytes)?;
if bytes.len() as u64 > max_bytes {
return Err(WorkspaceError::Invalid(format!(
"{label} exceeds the {max_bytes}-byte limit: {}",
path.display()
)));
}
Ok(bytes)
}
fn read_limited<R: Read>(reader: &mut R) -> Result<Vec<u8>, WorkspaceError> {
let mut limited = reader.take(MAX_IMPORT_BYTES + 1);
let mut bytes = Vec::new();
limited.read_to_end(&mut bytes)?;
if bytes.len() as u64 > MAX_IMPORT_BYTES {
return Err(WorkspaceError::Invalid(format!(
"input exceeds the {} MiB limit",
MAX_IMPORT_BYTES / (1024 * 1024)
)));
}
Ok(bytes)
}
fn inferred_table_name(source: &Path) -> Option<String> {
if source.as_os_str() == OsStr::new("-") {
return None;
}
source
.file_stem()
.and_then(OsStr::to_str)
.map(str::to_string)
}
#[derive(Debug, Clone)]
enum ImportedCell {
Null,
Empty,
Integer(i64),
Real(f64),
Boolean(bool),
Text(String),
}
#[derive(Debug, Clone)]
struct ImportedRows {
table: String,
columns: Vec<String>,
types: Vec<ColumnType>,
rows: Vec<Vec<ImportedCell>>,
}
#[derive(Debug, Clone, Copy)]
struct ImportLimits {
max_rows: Option<usize>,
max_columns: Option<usize>,
max_cells: Option<usize>,
max_work: Option<usize>,
}
impl ImportLimits {
fn unbounded() -> Self {
Self {
max_rows: None,
max_columns: None,
max_cells: None,
max_work: None,
}
}
fn mcp() -> Self {
Self {
max_rows: Some(MAX_MCP_IMPORT_ROWS),
max_columns: Some(MAX_MCP_IMPORT_COLUMNS),
max_cells: Some(MAX_MCP_IMPORT_CELLS),
max_work: Some(MCP_EXECUTION_WORK_LIMIT),
}
}
}
#[derive(Debug, Serialize)]
struct CliImportReport {
operation: &'static str,
workspace: String,
source: String,
format: String,
table: Option<String>,
bytes: usize,
change_id: String,
summary: String,
}
#[derive(Debug, Serialize)]
struct CliExportReport {
operation: &'static str,
workspace: String,
table: String,
format: String,
output: String,
rows: usize,
bytes: usize,
}
fn import_csv(
database: &Database,
table: Option<&str>,
bytes: &[u8],
limits: ImportLimits,
) -> Result<Option<String>, WorkspaceError> {
let table = required_table(table)?;
let mut reader = ReaderBuilder::new()
.has_headers(true)
.flexible(false)
.from_reader(bytes);
let headers = reader.headers()?.clone();
let columns = validate_headers(&headers)?;
enforce_import_limits(0, columns.len(), limits)?;
let mut rows = Vec::new();
for record in reader.records() {
if let Some(max_rows) = limits.max_rows
&& rows.len() >= max_rows
{
return Err(WorkspaceError::Invalid(format!(
"MCP import is limited to {max_rows} rows; use the CLI for larger imports"
)));
}
let record = record?;
enforce_import_limits(rows.len().saturating_add(1), columns.len(), limits)?;
rows.push(record.iter().map(parse_csv_cell).collect());
}
let imported = ImportedRows {
table: table.to_string(),
types: infer_types(&rows, columns.len()),
columns,
rows,
};
let summary = imported.summary();
import_rows(database, &imported, limits.max_work)?;
Ok(Some(summary))
}
fn import_json(
database: &Database,
table: Option<&str>,
bytes: &[u8],
limits: ImportLimits,
) -> Result<Option<String>, WorkspaceError> {
let value: JsonValue = serde_json::from_slice(bytes)?;
let objects = match value {
JsonValue::Array(values) => values,
JsonValue::Object(object) => vec![JsonValue::Object(object)],
_ => {
return Err(WorkspaceError::Invalid(
"JSON import expects an object or an array of objects".to_string(),
));
}
};
import_json_objects(database, table, objects, limits)
}
fn import_json_lines(
database: &Database,
table: Option<&str>,
bytes: &[u8],
limits: ImportLimits,
) -> Result<Option<String>, WorkspaceError> {
let text = std::str::from_utf8(bytes).map_err(|error| {
WorkspaceError::Invalid(format!("JSON Lines input is not UTF-8: {error}"))
})?;
let mut objects = Vec::new();
for (line_number, line) in text.lines().enumerate() {
if line.trim().is_empty() {
continue;
}
if let Some(max_rows) = limits.max_rows
&& objects.len() >= max_rows
{
return Err(WorkspaceError::Invalid(format!(
"MCP import is limited to {max_rows} rows; use the CLI for larger imports"
)));
}
let value: JsonValue = serde_json::from_str(line).map_err(|error| {
WorkspaceError::Invalid(format!("invalid JSON on line {}: {error}", line_number + 1))
})?;
if !value.is_object() {
return Err(WorkspaceError::Invalid(format!(
"JSON Lines line {} is not an object",
line_number + 1
)));
}
objects.push(value);
}
import_json_objects(database, table, objects, limits)
}
fn import_json_objects(
database: &Database,
table: Option<&str>,
objects: Vec<JsonValue>,
limits: ImportLimits,
) -> Result<Option<String>, WorkspaceError> {
let table = required_table(table)?;
let object_count = objects.len();
if objects.is_empty() {
return Err(WorkspaceError::Invalid(
"JSON import contains no objects; a table schema cannot be inferred".to_string(),
));
}
enforce_import_limits(object_count, 0, limits)?;
let mut names = BTreeSet::new();
let mut parsed_objects = Vec::with_capacity(objects.len());
for value in objects {
let JsonValue::Object(object) = value else {
return Err(WorkspaceError::Invalid(
"JSON import expects every row to be an object".to_string(),
));
};
for name in object.keys() {
validate_name(name, "JSON key")?;
names.insert(name.clone());
}
parsed_objects.push(object);
}
if names.is_empty() {
return Err(WorkspaceError::Invalid(
"JSON objects contain no fields; a table schema cannot be inferred".to_string(),
));
}
let columns: Vec<String> = names.into_iter().collect();
enforce_import_limits(object_count, columns.len(), limits)?;
let rows = parsed_objects
.iter()
.map(|object| {
columns
.iter()
.map(|name| {
object
.get(name)
.map(json_cell)
.unwrap_or(ImportedCell::Null)
})
.collect()
})
.collect::<Vec<Vec<ImportedCell>>>();
let imported = ImportedRows {
table: table.to_string(),
types: infer_types(&rows, columns.len()),
columns,
rows,
};
let summary = imported.summary();
import_rows(database, &imported, limits.max_work)?;
Ok(Some(summary))
}
fn enforce_import_limits(
rows: usize,
columns: usize,
limits: ImportLimits,
) -> Result<(), WorkspaceError> {
if let Some(max_rows) = limits.max_rows
&& rows > max_rows
{
return Err(WorkspaceError::Invalid(format!(
"MCP import is limited to {max_rows} rows; use the CLI for larger imports"
)));
}
if let Some(max_columns) = limits.max_columns
&& columns > max_columns
{
return Err(WorkspaceError::Invalid(format!(
"MCP import is limited to {max_columns} columns; use the CLI for wider imports"
)));
}
if let Some(max_cells) = limits.max_cells {
let cells = rows.saturating_mul(columns);
if cells > max_cells {
return Err(WorkspaceError::Invalid(format!(
"MCP import is limited to {max_cells} cells; use the CLI for larger imports"
)));
}
}
Ok(())
}
fn parse_sql_import(bytes: &[u8]) -> Result<(&str, Vec<Statement>), WorkspaceError> {
let sql = std::str::from_utf8(bytes)
.map_err(|error| WorkspaceError::Invalid(format!("SQL input is not UTF-8: {error}")))?;
let statements = parse(sql).map_err(|error| {
WorkspaceError::Invalid(format!(
"SQL import parse error at byte {}: {}",
error.offset, error.message
))
})?;
if statements.is_empty() {
return Err(WorkspaceError::Invalid(
"SQL import contains no statements".to_string(),
));
}
if statements.iter().any(statement_contains_control) {
return Err(WorkspaceError::Invalid(
"SQL imports must not contain BEGIN, COMMIT, ROLLBACK, or CHECKPOINT".to_string(),
));
}
Ok((sql, statements))
}
fn import_sql(database: &Database, bytes: &[u8]) -> Result<Option<String>, WorkspaceError> {
let (sql, _statements) = parse_sql_import(bytes)?;
let mut connection = database.connect();
connection.execute_sql("BEGIN")?;
let result = connection.execute_sql(sql);
match result {
Ok(results) => {
connection.execute_sql("COMMIT")?;
let mutations = results
.iter()
.filter(|result| {
matches!(
result,
StatementResult::Insert { .. }
| StatementResult::Update { .. }
| StatementResult::Delete { .. }
| StatementResult::CreateTable { .. }
| StatementResult::DropTable { .. }
| StatementResult::CreateIndex { .. }
| StatementResult::DropIndex { .. }
)
})
.count();
Ok(Some(format!("{mutations} statements from SQL")))
}
Err(error) => {
let _ = connection.execute_sql("ROLLBACK");
Err(error.into())
}
}
}
fn statement_contains_control(statement: &Statement) -> bool {
match statement {
Statement::Begin | Statement::Commit | Statement::Rollback | Statement::Checkpoint => true,
Statement::Explain(inner) => statement_contains_control(inner),
_ => false,
}
}
fn is_read_only(statement: &Statement) -> bool {
match statement {
Statement::Select { .. } => true,
Statement::Explain(inner) => is_read_only(inner),
_ => false,
}
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
enum ChangeKind {
#[serde(rename = "apply")]
Apply,
#[serde(rename = "undo")]
Undo,
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
enum ChangeStatus {
#[serde(rename = "prepared")]
Prepared,
#[serde(rename = "committed")]
Committed,
#[serde(rename = "recovered")]
Recovered,
#[serde(rename = "failed")]
Failed,
#[serde(rename = "unresolved")]
Unresolved,
}
impl ChangeStatus {
fn is_committed(&self) -> bool {
matches!(self, ChangeStatus::Committed | ChangeStatus::Recovered)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
struct PreviewItem {
statement: usize,
kind: String,
mutating: bool,
rows_affected: Option<usize>,
rows_returned: Option<usize>,
object: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct PlanRecord {
format_version: u32,
plan_id: String,
base_generation: u64,
base_state: String,
sql: String,
statements: Vec<PreviewItem>,
}
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct PlanReport {
plan_id: String,
sql: String,
base_generation: u64,
base_state: String,
statement_count: usize,
mutating_statements: usize,
statements: Vec<PreviewItem>,
}
impl From<&PlanRecord> for PlanReport {
fn from(plan: &PlanRecord) -> Self {
Self {
plan_id: plan.plan_id.clone(),
sql: plan.sql.clone(),
base_generation: plan.base_generation,
base_state: plan.base_state.clone(),
statement_count: plan.statements.len(),
mutating_statements: plan.statements.iter().filter(|item| item.mutating).count(),
statements: plan.statements.clone(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ImportMetadata {
request_key: String,
format: String,
#[serde(default)]
table: Option<String>,
bytes: usize,
summary: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ChangeRecord {
format_version: u32,
sequence: u64,
change_id: String,
kind: ChangeKind,
plan_id: Option<String>,
target_change_id: Option<String>,
base_generation: u64,
base_state: String,
expected_state: Option<String>,
snapshot_id: String,
sql: Option<String>,
status: ChangeStatus,
committed_generation: Option<u64>,
after_state: Option<String>,
error: Option<String>,
#[serde(default)]
import: Option<ImportMetadata>,
}
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct ApplyReport {
change_id: String,
plan_id: String,
base_state: String,
after_state: String,
generation: u64,
}
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct HistoryEntry {
sequence: u64,
change_id: String,
kind: ChangeKind,
status: ChangeStatus,
plan_id: Option<String>,
target_change_id: Option<String>,
base_state: String,
after_state: Option<String>,
committed_generation: Option<u64>,
error: Option<String>,
import: Option<HistoryImport>,
}
#[derive(Debug, Serialize, JsonSchema)]
struct HistoryImport {
format: String,
table: Option<String>,
bytes: usize,
summary: String,
}
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct DiffReport {
change_id: String,
kind: ChangeKind,
precision: &'static str,
before_state: String,
current_state: String,
state_changed: bool,
tables: Vec<TableDiff>,
}
#[derive(Debug, Serialize, JsonSchema)]
struct TableDiff {
table: String,
before_rows: Option<usize>,
after_rows: Option<usize>,
added_rows: usize,
removed_rows: usize,
schema_changed: bool,
data_changed: bool,
}
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct UndoReport {
change_id: String,
undone_change_id: String,
restored_state: String,
generation: u64,
}
#[derive(Debug, Clone, PartialEq)]
struct TableSnapshot {
columns: Vec<String>,
rows: Vec<Vec<Value>>,
}
fn preview_plan(workspace: &Workspace, sql: &str) -> Result<PlanRecord, WorkspaceError> {
preview_plan_with_output_limit(workspace, sql, None, None, None)
}
fn preview_plan_with_output_limit(
workspace: &Workspace,
sql: &str,
max_output_bytes: Option<usize>,
max_work: Option<usize>,
max_mutation_rows: Option<usize>,
) -> Result<PlanRecord, WorkspaceError> {
if sql.len() > MAX_PREVIEW_BYTES {
return Err(WorkspaceError::Invalid(format!(
"SQL exceeds the {} MiB preview limit",
MAX_PREVIEW_BYTES / (1024 * 1024)
)));
}
let statements = parse(sql).map_err(|error| {
WorkspaceError::Invalid(format!(
"preview parse error at byte {}: {}",
error.offset, error.message
))
})?;
if statements.is_empty() {
return Err(WorkspaceError::Invalid(
"preview requires at least one SQL statement".to_string(),
));
}
if statements.len() > MAX_PREVIEW_STATEMENTS {
return Err(WorkspaceError::Invalid(format!(
"preview accepts at most {MAX_PREVIEW_STATEMENTS} statements"
)));
}
let mutating_statements = statements
.iter()
.filter(|statement| is_mutation_statement(statement))
.count();
if mutating_statements > MAX_PREVIEW_MUTATIONS {
return Err(WorkspaceError::Invalid(format!(
"preview accepts at most {MAX_PREVIEW_MUTATIONS} mutating statements"
)));
}
if statements.iter().any(statement_contains_control) {
return Err(WorkspaceError::Invalid(
"preview SQL must not contain BEGIN, COMMIT, ROLLBACK, or CHECKPOINT".to_string(),
));
}
if !statements.iter().any(is_mutation_statement) {
return Err(WorkspaceError::Invalid(
"preview requires at least one mutating statement".to_string(),
));
}
let database = workspace.database()?;
let mut budget = max_work
.map(ExecutionBudget::bounded)
.unwrap_or_else(ExecutionBudget::unlimited);
database.checkpoint_with_budget(&mut budget)?;
let base_generation = database.generation();
let base_state = state_fingerprint(&workspace.database_path())?;
let mut connection = database.connect();
connection.execute_with_budget(&Statement::Begin, &mut budget)?;
let result = (|| {
let mut items = Vec::with_capacity(statements.len());
let mut mutation_rows = 0;
for (index, statement) in statements.iter().enumerate() {
let result = connection.execute_with_budget(statement, &mut budget)?;
enforce_mutation_row_limit(&mut mutation_rows, &result, max_mutation_rows)?;
if let StatementResult::Select { rows, .. } = &result
&& rows.len() > MAX_PREVIEW_ROWS
{
return Err(WorkspaceError::Invalid(format!(
"preview query result exceeds the {MAX_PREVIEW_ROWS}-row limit"
)));
}
items.push(preview_item(index + 1, &result));
}
Ok::<Vec<PreviewItem>, WorkspaceError>(items)
})();
let rollback = connection.execute_sql("ROLLBACK");
let preview_items = match result {
Ok(items) => {
rollback?;
items
}
Err(error) => {
let _ = rollback;
return Err(error);
}
};
let plan_id = plan_id_for(&base_state, sql);
let plan = PlanRecord {
format_version: FORMAT_VERSION,
plan_id,
base_generation,
base_state,
sql: sql.to_string(),
statements: preview_items,
};
if let Some(max_output_bytes) = max_output_bytes {
let report = PlanReport::from(&plan);
let output_size = serde_json::to_vec(&report)?.len();
if output_size > max_output_bytes {
return Err(WorkspaceError::Invalid(format!(
"workspace preview is {output_size} bytes; response limit is {max_output_bytes} bytes"
)));
}
}
// Keep the database handle live through persistence so another process
// cannot change the state between the captured fingerprint and plan write.
persist_plan(workspace, &plan)?;
Ok(plan)
}
fn persist_plan(workspace: &Workspace, plan: &PlanRecord) -> Result<(), WorkspaceError> {
ensure_history_dirs(workspace)?;
let path = plan_path(workspace, &plan.plan_id);
if path.exists() {
let existing: PlanRecord = read_json(&path)?;
if existing != *plan {
return Err(WorkspaceError::Invalid(
"plan identifier collision; refusing to replace an existing plan".to_string(),
));
}
} else {
write_new_json(&path, plan)?;
}
Ok(())
}
fn preview_item(statement: usize, result: &StatementResult) -> PreviewItem {
let (kind, mutating, rows_affected, rows_returned, object) = match result {
StatementResult::Select { rows, .. } => ("select", false, None, Some(rows.len()), None),
StatementResult::Insert { rows_affected } => {
("insert", true, Some(*rows_affected), None, None)
}
StatementResult::Update { rows_affected } => {
("update", true, Some(*rows_affected), None, None)
}
StatementResult::Delete { rows_affected } => {
("delete", true, Some(*rows_affected), None, None)
}
StatementResult::CreateTable { name } => {
("create_table", true, None, None, Some(name.clone()))
}
StatementResult::DropTable { name } => ("drop_table", true, None, None, Some(name.clone())),
StatementResult::CreateIndex { name, .. } => {
("create_index", true, None, None, Some(name.clone()))
}
StatementResult::DropIndex { name } => ("drop_index", true, None, None, Some(name.clone())),
StatementResult::Explain(_) => ("explain", false, None, None, None),
StatementResult::Begin => ("begin", false, None, None, None),
StatementResult::Commit => ("commit", false, None, None, None),
StatementResult::Rollback => ("rollback", false, None, None, None),
StatementResult::Checkpoint => ("checkpoint", false, None, None, None),
StatementResult::Echo(_) => ("echo", false, None, None, None),
};
PreviewItem {
statement,
kind: kind.to_string(),
mutating,
rows_affected,
rows_returned,
object,
}
}
fn enforce_mutation_row_limit(
total_rows: &mut usize,
result: &StatementResult,
max_rows: Option<usize>,
) -> Result<(), WorkspaceError> {
let rows = match result {
StatementResult::Insert { rows_affected }
| StatementResult::Update { rows_affected }
| StatementResult::Delete { rows_affected } => *rows_affected,
_ => return Ok(()),
};
*total_rows = total_rows.saturating_add(rows);
if let Some(max_rows) = max_rows
&& *total_rows > max_rows
{
return Err(WorkspaceError::Invalid(format!(
"MCP workspace mutations are limited to {max_rows} affected rows per plan; split the operation into smaller reviewed plans"
)));
}
Ok(())
}
fn is_mutation_statement(statement: &Statement) -> bool {
matches!(
statement,
Statement::CreateTable { .. }
| Statement::DropTable { .. }
| Statement::CreateIndex { .. }
| Statement::DropIndex { .. }
| Statement::Insert { .. }
| Statement::InsertSelect { .. }
| Statement::Update { .. }
| Statement::Delete { .. }
)
}
fn ensure_history_dirs(workspace: &Workspace) -> Result<(), WorkspaceError> {
let directories = history_directories(workspace);
validate_history_dirs(workspace)?;
for directory in directories {
fs::create_dir_all(&directory)?;
sync_parent(&directory)?;
}
validate_history_dirs(workspace)?;
Ok(())
}
fn history_directories(workspace: &Workspace) -> [PathBuf; 4] {
let history = workspace.root.join(HISTORY_DIR);
[
history.clone(),
history.join(PLANS_DIR),
history.join(CHANGES_DIR),
history.join(SNAPSHOTS_DIR),
]
}
fn validate_history_dirs(workspace: &Workspace) -> Result<(), WorkspaceError> {
for directory in history_directories(workspace) {
if path_is_symlink(&directory)? {
return Err(WorkspaceError::Invalid(format!(
"workspace history directory cannot be a symbolic link: {}",
directory.display()
)));
}
if directory.exists() && !directory.is_dir() {
return Err(WorkspaceError::Invalid(format!(
"workspace history path is not a directory: {}",
directory.display()
)));
}
}
Ok(())
}
fn valid_id(id: &str) -> Result<(), WorkspaceError> {
if id.len() != 64 || !id.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(WorkspaceError::Invalid(
"identifier must be a 64-character hexadecimal value".to_string(),
));
}
Ok(())
}
fn plan_path(workspace: &Workspace, plan_id: &str) -> PathBuf {
workspace
.root
.join(HISTORY_DIR)
.join(PLANS_DIR)
.join(format!("{plan_id}.json"))
}
fn change_path(workspace: &Workspace, change_id: &str) -> PathBuf {
workspace
.root
.join(HISTORY_DIR)
.join(CHANGES_DIR)
.join(format!("{change_id}.json"))
}
fn snapshot_path(workspace: &Workspace, snapshot_id: &str) -> PathBuf {
workspace
.root
.join(HISTORY_DIR)
.join(SNAPSHOTS_DIR)
.join(format!("{snapshot_id}.basalt"))
}
fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<T, WorkspaceError> {
if path_is_symlink(path)? {
return Err(WorkspaceError::Invalid(format!(
"workspace metadata file cannot be a symbolic link: {}",
path.display()
)));
}
let bytes = read_file_limited(path, MAX_HISTORY_METADATA_BYTES, "workspace metadata")?;
Ok(serde_json::from_slice(&bytes)?)
}
fn write_new_json<T: Serialize>(path: &Path, value: &T) -> Result<(), WorkspaceError> {
let mut bytes = serde_json::to_vec_pretty(value)?;
bytes.push(b'\n');
write_new_file(path, &bytes)
}
fn write_atomic_json<T: Serialize>(path: &Path, value: &T) -> Result<(), WorkspaceError> {
let mut bytes = serde_json::to_vec_pretty(value)?;
bytes.push(b'\n');
atomic_write_file(path, &bytes)
}
fn state_fingerprint(path: &Path) -> Result<String, WorkspaceError> {
if path_is_symlink(path)? {
return Err(WorkspaceError::Invalid(format!(
"workspace state file cannot be a symbolic link: {}",
path.display()
)));
}
Ok(format!(
"sha256:{}",
sha256_bytes(&read_file_limited(
path,
storage::MAX_SNAPSHOT_BYTES as u64,
"workspace state",
)?)
))
}
fn logical_state_fingerprint(path: &Path) -> Result<String, WorkspaceError> {
let (state, _) = storage::read_snapshot(path)?;
Ok(format!("sha256:{}", sha256_bytes(&state.encode())))
}
fn sha256_bytes(bytes: &[u8]) -> String {
let digest = Sha256::digest(bytes);
digest.iter().map(|byte| format!("{byte:02x}")).collect()
}
fn plan_id_for(base_state: &str, sql: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(b"basalt-plan-v1\0");
hasher.update(base_state.as_bytes());
hasher.update(b"\0");
hasher.update(sql.as_bytes());
hasher
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
fn import_change_id(base_state: &str, format: DataFormat, table: &str, content: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(b"basalt-import-v1\0");
hasher.update(base_state.as_bytes());
hasher.update(b"\0");
hasher.update(format.name().as_bytes());
hasher.update(b"\0");
hasher.update(table.as_bytes());
hasher.update(b"\0");
hasher.update(content);
hasher
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
fn import_request_key(format: DataFormat, table: &str, content: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(b"basalt-import-request-v1\0");
hasher.update(format.name().as_bytes());
hasher.update(b"\0");
hasher.update(table.as_bytes());
hasher.update(b"\0");
hasher.update(content);
hasher
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
fn apply_change_id(plan_id: &str, base_state: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(b"basalt-apply-v1\0");
hasher.update(plan_id.as_bytes());
hasher.update(b"\0");
hasher.update(base_state.as_bytes());
hasher
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
fn undo_change_id(change_id: &str, base_state: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(b"basalt-undo-v1\0");
hasher.update(change_id.as_bytes());
hasher.update(b"\0");
hasher.update(base_state.as_bytes());
hasher
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
fn next_sequence(changes: &[ChangeRecord]) -> Result<u64, WorkspaceError> {
changes
.iter()
.map(|change| change.sequence)
.max()
.unwrap_or(0)
.checked_add(1)
.ok_or_else(|| WorkspaceError::Invalid("change sequence exhausted".to_string()))
}
fn copy_atomic(source: &Path, destination: &Path) -> Result<(), WorkspaceError> {
if path_is_symlink(source)? || path_is_symlink(destination)? {
return Err(WorkspaceError::Invalid(
"workspace recovery files cannot be symbolic links".to_string(),
));
}
let bytes = read_file_limited(
source,
storage::MAX_SNAPSHOT_BYTES as u64,
"workspace recovery snapshot",
)?;
atomic_write_file(destination, &bytes)
}
fn load_plan(workspace: &Workspace, plan_id: &str) -> Result<PlanRecord, WorkspaceError> {
valid_id(plan_id)?;
validate_history_dirs(workspace)?;
let path = plan_path(workspace, plan_id);
if !path.is_file() {
return Err(WorkspaceError::Invalid(format!(
"plan does not exist: {plan_id}"
)));
}
let plan: PlanRecord = read_json(&path)?;
if plan.format_version != FORMAT_VERSION
|| plan.plan_id != plan_id
|| plan_id_for(&plan.base_state, &plan.sql) != plan.plan_id
{
return Err(WorkspaceError::Invalid(format!(
"plan is invalid or has been modified: {plan_id}"
)));
}
validate_plan_record(&plan, plan_id)?;
Ok(plan)
}
fn load_changes(workspace: &Workspace) -> Result<Vec<ChangeRecord>, WorkspaceError> {
load_changes_with_limits(workspace, None, None)
}
fn load_changes_with_limits(
workspace: &Workspace,
max_entries: Option<usize>,
max_metadata_bytes: Option<u64>,
) -> Result<Vec<ChangeRecord>, WorkspaceError> {
validate_history_dirs(workspace)?;
let directory = workspace.root.join(HISTORY_DIR).join(CHANGES_DIR);
if !directory.exists() {
return Ok(Vec::new());
}
let mut changes = Vec::new();
let mut metadata_bytes = 0u64;
let mut directory_entries = 0usize;
for entry in fs::read_dir(&directory)? {
directory_entries = directory_entries.checked_add(1).ok_or_else(|| {
WorkspaceError::Invalid(
"workspace history directory entry count overflowed".to_string(),
)
})?;
if let Some(max_entries) = max_entries
&& directory_entries > max_entries
{
return Err(WorkspaceError::Invalid(format!(
"MCP workspace history directory exceeds the {max_entries}-entry limit"
)));
}
let path = entry?.path();
if path_is_symlink(&path)? {
return Err(WorkspaceError::Invalid(format!(
"workspace history record cannot be a symbolic link: {}",
path.display()
)));
}
if path.extension() != Some(OsStr::new("json")) {
continue;
}
if !path.is_file() {
return Err(WorkspaceError::Invalid(format!(
"workspace history record is not a file: {}",
path.display()
)));
}
if let Some(max_metadata_bytes) = max_metadata_bytes {
let bytes = fs::symlink_metadata(&path)?.len();
metadata_bytes = metadata_bytes.checked_add(bytes).ok_or_else(|| {
WorkspaceError::Invalid("MCP workspace history metadata is too large".to_string())
})?;
if metadata_bytes > max_metadata_bytes {
return Err(WorkspaceError::Invalid(format!(
"MCP workspace history metadata exceeds the {max_metadata_bytes}-byte limit"
)));
}
}
let change: ChangeRecord = read_json(&path)?;
valid_id(&change.change_id)?;
if path.file_stem() != Some(OsStr::new(&change.change_id)) {
return Err(WorkspaceError::Invalid(format!(
"change filename does not match its identifier: {}",
path.display()
)));
}
if change.format_version != FORMAT_VERSION {
return Err(WorkspaceError::Invalid(format!(
"unsupported change format in {}",
path.display()
)));
}
validate_change_record(&change, &path)?;
changes.push(change);
}
changes.sort_by_key(|change| change.sequence);
if changes
.windows(2)
.any(|pair| pair[0].sequence == pair[1].sequence)
{
return Err(WorkspaceError::Invalid(
"workspace history contains duplicate change sequence numbers".to_string(),
));
}
Ok(changes)
}
fn validate_plan_record(plan: &PlanRecord, plan_id: &str) -> Result<(), WorkspaceError> {
if !valid_state_fingerprint(&plan.base_state) {
return Err(WorkspaceError::Invalid(format!(
"plan {plan_id} has an invalid base-state fingerprint"
)));
}
let statements = parse(&plan.sql).map_err(|error| {
WorkspaceError::Invalid(format!(
"plan {plan_id} contains invalid SQL at byte {}: {}",
error.offset, error.message
))
})?;
if statements.is_empty()
|| statements.len() > MAX_PREVIEW_STATEMENTS
|| statements.iter().any(statement_contains_control)
|| statements
.iter()
.filter(|statement| is_mutation_statement(statement))
.count()
> MAX_PREVIEW_MUTATIONS
|| !statements.iter().any(is_mutation_statement)
{
return Err(WorkspaceError::Invalid(format!(
"plan {plan_id} contains an invalid preview statement sequence"
)));
}
if plan.statements.len() != statements.len() {
return Err(WorkspaceError::Invalid(format!(
"plan {plan_id} statement metadata does not match its SQL"
)));
}
for (index, (statement, item)) in statements.iter().zip(&plan.statements).enumerate() {
let (kind, mutating) = statement_metadata(statement);
if item.statement != index + 1 || item.kind != kind || item.mutating != mutating {
return Err(WorkspaceError::Invalid(format!(
"plan {plan_id} statement metadata does not match statement {}",
index + 1
)));
}
if item
.rows_returned
.is_some_and(|rows| rows > MAX_PREVIEW_ROWS)
{
return Err(WorkspaceError::Invalid(format!(
"plan {plan_id} contains an oversized preview result"
)));
}
}
Ok(())
}
fn validate_change_record(change: &ChangeRecord, path: &Path) -> Result<(), WorkspaceError> {
let invalid = |message: &str| {
Err(WorkspaceError::Invalid(format!(
"invalid workspace change {}: {message}",
path.display()
)))
};
if change.sequence == 0 {
return invalid("sequence must be greater than zero");
}
if !valid_id(&change.change_id).is_ok() || change.snapshot_id != change.change_id {
return invalid("change and recovery-point identifiers are invalid or do not match");
}
if !valid_state_fingerprint(&change.base_state)
|| change
.expected_state
.as_deref()
.is_some_and(|state| !valid_state_fingerprint(state))
|| change
.after_state
.as_deref()
.is_some_and(|state| !valid_state_fingerprint(state))
{
return invalid("state fingerprint is invalid");
}
match change.kind {
ChangeKind::Apply => {
if let Some(plan_id) = &change.plan_id {
if !valid_id(plan_id).is_ok()
|| change.import.is_some()
|| change.sql.as_deref().is_none_or(str::is_empty)
|| change
.sql
.as_deref()
.is_none_or(|sql| plan_id_for(&change.base_state, sql) != *plan_id)
|| apply_change_id(plan_id, &change.base_state) != change.change_id
{
return invalid("apply plan metadata is inconsistent");
}
} else {
let Some(import) = &change.import else {
return invalid("apply record is missing its plan or import metadata");
};
if change.sql.is_some()
|| !valid_id(&import.request_key).is_ok()
|| u64::try_from(import.bytes).unwrap_or(u64::MAX) > MAX_IMPORT_BYTES
{
return invalid("import metadata is inconsistent");
}
let format = match DataFormat::parse(&import.format) {
Ok(format) => format,
Err(_) => return invalid("import format is invalid"),
};
if format == DataFormat::Sql {
if import.table.is_some() {
return invalid("SQL import must not have a table");
}
} else if import
.table
.as_deref()
.is_none_or(|table| validate_name(table, "table name").is_err())
{
return invalid("row import table is invalid");
}
}
if change.target_change_id.is_some() || change.expected_state.is_some() {
return invalid("apply record contains undo metadata");
}
}
ChangeKind::Undo => {
if change.plan_id.is_some()
|| change.sql.is_some()
|| change.import.is_some()
|| change
.target_change_id
.as_deref()
.is_none_or(|id| !valid_id(id).is_ok())
|| change.expected_state.is_none()
|| change.target_change_id.as_deref().is_some_and(|target| {
undo_change_id(target, &change.base_state) != change.change_id
})
{
return invalid("undo metadata is inconsistent");
}
}
}
match change.status {
ChangeStatus::Prepared | ChangeStatus::Failed | ChangeStatus::Unresolved => {
if change.committed_generation.is_some() || change.after_state.is_some() {
return invalid("uncommitted record contains a committed receipt");
}
}
ChangeStatus::Committed | ChangeStatus::Recovered => {
if change.committed_generation.is_none() || change.after_state.is_none() {
return invalid("committed record is missing its receipt");
}
}
}
Ok(())
}
fn statement_metadata(statement: &Statement) -> (&'static str, bool) {
match statement {
Statement::Select { .. } => ("select", false),
Statement::Explain(_) => ("explain", false),
Statement::CreateTable { .. } => ("create_table", true),
Statement::DropTable { .. } => ("drop_table", true),
Statement::CreateIndex { .. } => ("create_index", true),
Statement::DropIndex { .. } => ("drop_index", true),
Statement::Insert { .. } | Statement::InsertSelect { .. } => ("insert", true),
Statement::Update { .. } => ("update", true),
Statement::Delete { .. } => ("delete", true),
Statement::Begin => ("begin", false),
Statement::Commit => ("commit", false),
Statement::Rollback => ("rollback", false),
Statement::Checkpoint => ("checkpoint", false),
}
}
fn valid_state_fingerprint(value: &str) -> bool {
let Some(hex) = value.strip_prefix("sha256:") else {
return false;
};
hex.len() == 64 && hex.bytes().all(|byte| byte.is_ascii_hexdigit())
}
fn load_changes_for_operation(
workspace: &Workspace,
bounded: bool,
) -> Result<Vec<ChangeRecord>, WorkspaceError> {
if bounded {
load_changes_with_limits(
workspace,
Some(MAX_MCP_HISTORY_ENTRIES),
Some(MAX_MCP_HISTORY_METADATA_BYTES),
)
} else {
load_changes(workspace)
}
}
fn reconcile_change(
change: &mut ChangeRecord,
current_state: &str,
current_logical_state: Option<&str>,
current_generation: u64,
) -> bool {
if change.status != ChangeStatus::Prepared {
return false;
}
match change.kind {
ChangeKind::Apply => {
if current_generation == change.base_generation.saturating_add(1)
&& current_state != change.base_state
{
change.status = ChangeStatus::Recovered;
change.committed_generation = Some(current_generation);
change.after_state = Some(current_state.to_string());
change.error =
Some("commit completed before the history record was finalized".to_string());
} else if current_generation == change.base_generation
&& current_state == change.base_state
{
change.status = ChangeStatus::Failed;
change.error =
Some("operation was not observed as committed after interruption".to_string());
} else {
change.status = ChangeStatus::Unresolved;
change.error = Some(
"workspace state does not match either side of the prepared operation"
.to_string(),
);
}
}
ChangeKind::Undo => {
if change.expected_state.as_deref() == Some(current_state)
|| change.expected_state.as_deref() == current_logical_state
{
change.status = ChangeStatus::Recovered;
change.committed_generation = Some(current_generation);
change.after_state = Some(current_state.to_string());
change.error =
Some("restore completed before the history record was finalized".to_string());
} else if current_generation == change.base_generation
&& current_state == change.base_state
{
change.status = ChangeStatus::Failed;
change.error =
Some("restore was not observed as completed after interruption".to_string());
} else {
change.status = ChangeStatus::Unresolved;
change.error = Some(
"workspace state does not match either side of the prepared restore"
.to_string(),
);
}
}
}
true
}
fn reconcile_changes(
workspace: &Workspace,
changes: &mut [ChangeRecord],
current_state: &str,
current_generation: u64,
) -> Result<(), WorkspaceError> {
let current_logical_state = changes
.iter()
.any(|change| change.kind == ChangeKind::Undo && change.status == ChangeStatus::Prepared)
.then(|| logical_state_fingerprint(&workspace.database_path()))
.transpose()?;
for change in changes {
if reconcile_change(
change,
current_state,
current_logical_state.as_deref(),
current_generation,
) {
write_atomic_json(&change_path(workspace, &change.change_id), change)?;
}
}
Ok(())
}
fn history(workspace: &Workspace) -> Result<Vec<HistoryEntry>, WorkspaceError> {
history_with_limits(workspace, None, None)
}
fn history_with_limits(
workspace: &Workspace,
max_entries: Option<usize>,
max_metadata_bytes: Option<u64>,
) -> Result<Vec<HistoryEntry>, WorkspaceError> {
let database = workspace.database()?;
let current_state = state_fingerprint(&workspace.database_path())?;
let current_generation = database.generation();
let mut changes = load_changes_with_limits(workspace, max_entries, max_metadata_bytes)?;
reconcile_changes(workspace, &mut changes, ¤t_state, current_generation)?;
Ok(changes
.into_iter()
.map(|change| {
let import = change.import.map(|import| HistoryImport {
format: import.format,
table: import.table,
bytes: import.bytes,
summary: import.summary,
});
HistoryEntry {
sequence: change.sequence,
change_id: change.change_id,
kind: change.kind,
status: change.status,
plan_id: change.plan_id,
target_change_id: change.target_change_id,
base_state: change.base_state,
after_state: change.after_state,
committed_generation: change.committed_generation,
error: change.error,
import,
}
})
.collect())
}
fn latest_committed(changes: &[ChangeRecord]) -> Option<&ChangeRecord> {
changes
.iter()
.filter(|change| change.status.is_committed())
.max_by_key(|change| change.sequence)
}
fn apply_plan(
workspace: &Workspace,
requested_plan_id: &str,
max_work: Option<usize>,
max_mutation_rows: Option<usize>,
) -> Result<ApplyReport, WorkspaceError> {
let plan = load_plan(workspace, requested_plan_id)?;
if !plan.statements.iter().any(|item| item.mutating) {
return Err(WorkspaceError::Invalid(
"plan does not contain a mutating statement".to_string(),
));
}
ensure_history_dirs(workspace)?;
let database = workspace.database()?;
let mut budget = max_work
.map(ExecutionBudget::bounded)
.unwrap_or_else(ExecutionBudget::unlimited);
database.checkpoint_with_budget(&mut budget)?;
let current_state = state_fingerprint(&workspace.database_path())?;
let change_id = apply_change_id(&plan.plan_id, &plan.base_state);
let change_file = change_path(workspace, &change_id);
let mut changes = load_changes_for_operation(workspace, max_work.is_some())?;
if let Some(existing) = changes
.iter_mut()
.find(|change| change.change_id == change_id)
{
if reconcile_change(existing, ¤t_state, None, database.generation()) {
write_atomic_json(&change_file, existing)?;
}
if existing.status.is_committed() {
if existing.after_state.as_deref() == Some(current_state.as_str()) {
return Ok(ApplyReport {
change_id,
plan_id: plan.plan_id,
base_state: existing.base_state.clone(),
after_state: existing.after_state.clone().ok_or_else(|| {
WorkspaceError::Invalid(
"committed plan is missing its after-state".to_string(),
)
})?,
generation: existing.committed_generation.ok_or_else(|| {
WorkspaceError::Invalid(
"committed plan is missing its generation".to_string(),
)
})?,
});
}
return Err(WorkspaceError::Invalid(format!(
"plan has already been applied as change {change_id}; workspace state moved, so it will not be replayed"
)));
}
if existing.status == ChangeStatus::Unresolved {
return Err(WorkspaceError::Invalid(format!(
"change {change_id} is unresolved; inspect its recovery point before continuing"
)));
}
}
if current_state != plan.base_state {
return Err(WorkspaceError::Invalid(
"plan is stale; preview the operation again against the current workspace".to_string(),
));
}
let snapshot = snapshot_path(workspace, &change_id);
if snapshot.exists() {
if state_fingerprint(&snapshot)? != plan.base_state {
return Err(WorkspaceError::Invalid(format!(
"recovery point for change {change_id} does not match the plan"
)));
}
} else {
copy_atomic(&workspace.database_path(), &snapshot)?;
}
let sequence = changes
.iter()
.find(|change| change.change_id == change_id)
.map(|change| change.sequence)
.unwrap_or(next_sequence(&changes)?);
let mut change = ChangeRecord {
format_version: FORMAT_VERSION,
sequence,
change_id: change_id.clone(),
kind: ChangeKind::Apply,
plan_id: Some(plan.plan_id.clone()),
target_change_id: None,
base_generation: plan.base_generation,
base_state: plan.base_state.clone(),
expected_state: None,
snapshot_id: change_id.clone(),
sql: Some(plan.sql.clone()),
status: ChangeStatus::Prepared,
committed_generation: None,
after_state: None,
error: None,
import: None,
};
if change_file.exists() {
write_atomic_json(&change_file, &change)?;
} else {
write_new_json(&change_file, &change)?;
}
let mut connection = database.connect();
if let Err(error) = connection.execute_with_budget(&Statement::Begin, &mut budget) {
change.status = ChangeStatus::Failed;
change.error = Some(error.to_string());
write_atomic_json(&change_file, &change)?;
return Err(error.into());
}
let execution = if max_work.is_some() {
connection.execute_sql_using_budget(&plan.sql, &mut budget)
} else {
connection.execute_sql(&plan.sql)
};
let results = match execution {
Ok(results) => results,
Err(error) => {
let _ = connection.execute_sql("ROLLBACK");
change.status = ChangeStatus::Failed;
change.error = Some(error.to_string());
write_atomic_json(&change_file, &change)?;
return Err(error.into());
}
};
let mut mutation_rows = 0;
for result in &results {
if let Err(error) =
enforce_mutation_row_limit(&mut mutation_rows, result, max_mutation_rows)
{
let _ = connection.execute_sql("ROLLBACK");
change.status = ChangeStatus::Failed;
change.error = Some(error.to_string());
write_atomic_json(&change_file, &change)?;
return Err(error);
}
}
if let Err(error) = connection.execute_with_budget(&Statement::Commit, &mut budget) {
change.status = ChangeStatus::Unresolved;
change.error = Some(error.to_string());
write_atomic_json(&change_file, &change)?;
return Err(error.into());
}
drop(connection);
if let Err(error) = database.checkpoint_with_budget(&mut budget) {
change.status = ChangeStatus::Unresolved;
change.error = Some(format!(
"operation committed but checkpoint failed: {error}"
));
write_atomic_json(&change_file, &change)?;
return Err(error.into());
}
let after_state = state_fingerprint(&workspace.database_path())?;
let generation = database.generation();
if std::env::var_os("BASALT_CRASH_TEST_AFTER_APPLY_CHECKPOINT").is_some() {
std::process::abort();
}
change.status = ChangeStatus::Committed;
change.committed_generation = Some(generation);
change.after_state = Some(after_state.clone());
write_atomic_json(&change_file, &change)?;
Ok(ApplyReport {
change_id,
plan_id: plan.plan_id,
base_state: plan.base_state,
after_state,
generation,
})
}
fn diff(
workspace: &Workspace,
requested_change_id: Option<&str>,
) -> Result<DiffReport, WorkspaceError> {
diff_with_row_limit(workspace, requested_change_id, None, None)
}
fn diff_with_row_limit(
workspace: &Workspace,
requested_change_id: Option<&str>,
max_total_rows: Option<usize>,
max_work: Option<usize>,
) -> Result<DiffReport, WorkspaceError> {
let mut budget = max_work
.map(ExecutionBudget::bounded)
.unwrap_or_else(ExecutionBudget::unlimited);
let mut changes = load_changes_for_operation(workspace, max_work.is_some())?;
let database = workspace.database()?;
let current_state = state_fingerprint(&workspace.database_path())?;
let current_generation = database.generation();
reconcile_changes(workspace, &mut changes, ¤t_state, current_generation)?;
let change = match requested_change_id {
Some(change_id) => {
valid_id(change_id)?;
changes
.iter()
.find(|change| change.change_id == change_id)
.cloned()
.ok_or_else(|| {
WorkspaceError::Invalid(format!("change does not exist: {change_id}"))
})?
}
None => latest_committed(&changes)
.cloned()
.ok_or_else(|| WorkspaceError::Invalid("no committed changes to diff".to_string()))?,
};
if !change.status.is_committed() {
return Err(WorkspaceError::Invalid(format!(
"change {} is not committed; status is {:?}",
change.change_id, change.status
)));
}
let snapshot = snapshot_path(workspace, &change.snapshot_id);
if !snapshot.is_file() {
return Err(WorkspaceError::Invalid(format!(
"recovery point is missing for change {}",
change.change_id
)));
}
let before_database = Database::open(&snapshot)?;
let before = logical_snapshot(&before_database, max_total_rows, &mut budget)?;
let after = logical_snapshot(&database, max_total_rows, &mut budget)?;
let mut names = BTreeSet::new();
names.extend(before.keys().cloned());
names.extend(after.keys().cloned());
let tables = names
.into_iter()
.filter_map(|name| {
let before_table = before.get(&name);
let after_table = after.get(&name);
let schema_changed = match (before_table, after_table) {
(Some(before), Some(after)) => before.columns != after.columns,
(None, None) => false,
_ => true,
};
let (added_rows, removed_rows) = match (before_table, after_table) {
(Some(before), Some(after)) => row_delta_counts(&before.rows, &after.rows),
(None, Some(after)) => (after.rows.len(), 0),
(Some(before), None) => (0, before.rows.len()),
(None, None) => (0, 0),
};
let data_changed = added_rows > 0 || removed_rows > 0;
(schema_changed || data_changed).then(|| TableDiff {
table: name,
before_rows: before_table.map(|table| table.rows.len()),
after_rows: after_table.map(|table| table.rows.len()),
added_rows,
removed_rows,
schema_changed,
data_changed,
})
})
.collect::<Vec<_>>();
Ok(DiffReport {
change_id: change.change_id,
kind: change.kind,
precision: "table schema and row-multiset comparison",
before_state: change.base_state,
current_state,
state_changed: !tables.is_empty(),
tables,
})
}
fn row_delta_counts(before: &[Vec<Value>], after: &[Vec<Value>]) -> (usize, usize) {
let mut before_counts = HashMap::<Vec<u8>, usize>::new();
for row in before {
let key = row_key(row);
*before_counts.entry(key).or_default() += 1;
}
let mut after_counts = HashMap::<Vec<u8>, usize>::new();
for row in after {
let key = row_key(row);
*after_counts.entry(key).or_default() += 1;
}
let removed = before_counts.iter().fold(0usize, |total, (key, count)| {
total.saturating_add(count.saturating_sub(after_counts.get(key).copied().unwrap_or(0)))
});
let added = after_counts.iter().fold(0usize, |total, (key, count)| {
total.saturating_add(count.saturating_sub(before_counts.get(key).copied().unwrap_or(0)))
});
(added, removed)
}
fn row_key(row: &[Value]) -> Vec<u8> {
let mut key = Vec::new();
key.extend_from_slice(b"basalt-row-v1\0");
for value in row {
match value {
Value::Null => key.push(0),
Value::Integer(value) => {
key.push(1);
key.extend_from_slice(&value.to_le_bytes());
}
Value::Real(value) => {
key.push(2);
key.extend_from_slice(&value.to_bits().to_le_bytes());
}
Value::Text(value) => {
key.push(3);
key.extend_from_slice(&(value.len() as u64).to_le_bytes());
key.extend_from_slice(value.as_bytes());
}
Value::Boolean(value) => {
key.push(4);
key.push(*value as u8);
}
}
}
key
}
fn logical_snapshot(
database: &Database,
max_total_rows: Option<usize>,
budget: &mut ExecutionBudget,
) -> Result<BTreeMap<String, TableSnapshot>, WorkspaceError> {
let mut tables = BTreeMap::new();
let mut total_rows = 0usize;
for table in database.table_names()? {
let row_count = database.row_count(&table)?;
if let Some(max_total_rows) = max_total_rows {
total_rows = total_rows.saturating_add(row_count);
if total_rows > max_total_rows {
return Err(WorkspaceError::Invalid(format!(
"MCP diff is limited to {max_total_rows} rows across a compared database; use the CLI diff for larger workspaces"
)));
}
}
let (columns, rows) = select_table_with_budget(database, &table, Some(budget))?;
debug_assert_eq!(rows.len(), row_count);
tables.insert(table, TableSnapshot { columns, rows });
}
Ok(tables)
}
fn undo(
workspace: &Workspace,
requested_change_id: &str,
max_work: Option<usize>,
) -> Result<UndoReport, WorkspaceError> {
valid_id(requested_change_id)?;
ensure_history_dirs(workspace)?;
let mut budget = max_work
.map(ExecutionBudget::bounded)
.unwrap_or_else(ExecutionBudget::unlimited);
let mut changes = load_changes_for_operation(workspace, max_work.is_some())?;
let database = workspace.database()?;
database.checkpoint_with_budget(&mut budget)?;
let current_state = state_fingerprint(&workspace.database_path())?;
let current_generation = database.generation();
reconcile_changes(workspace, &mut changes, ¤t_state, current_generation)?;
if let Some(existing) = changes.iter().find(|change| {
change.kind == ChangeKind::Undo
&& change.target_change_id.as_deref() == Some(requested_change_id)
&& change.status.is_committed()
}) {
if existing.after_state.as_deref() == Some(current_state.as_str()) {
return Ok(UndoReport {
change_id: existing.change_id.clone(),
undone_change_id: requested_change_id.to_string(),
restored_state: existing.after_state.clone().ok_or_else(|| {
WorkspaceError::Invalid(
"committed undo is missing its restored state".to_string(),
)
})?,
generation: existing.committed_generation.ok_or_else(|| {
WorkspaceError::Invalid("committed undo is missing its generation".to_string())
})?,
});
}
return Err(WorkspaceError::Invalid(format!(
"change {requested_change_id} has already been undone as {}; workspace state moved, so it will not be replayed",
existing.change_id
)));
}
let target = changes
.iter()
.find(|change| change.change_id == requested_change_id)
.cloned()
.ok_or_else(|| {
WorkspaceError::Invalid(format!("change does not exist: {requested_change_id}"))
})?;
if !target.status.is_committed() {
return Err(WorkspaceError::Invalid(format!(
"change {} is not committed; status is {:?}",
target.change_id, target.status
)));
}
let latest = latest_committed(&changes).ok_or_else(|| {
WorkspaceError::Invalid("there are no committed changes to undo".to_string())
})?;
if latest.change_id != target.change_id {
return Err(WorkspaceError::Invalid(
"only the latest committed change can be undone; undo later changes first".to_string(),
));
}
if target.after_state.as_deref() != Some(current_state.as_str()) {
return Err(WorkspaceError::Invalid(
"workspace state moved after this change; refusing to discard later work".to_string(),
));
}
let target_snapshot = snapshot_path(workspace, &target.snapshot_id);
if !target_snapshot.is_file() {
return Err(WorkspaceError::Invalid(format!(
"recovery point is missing for change {}",
target.change_id
)));
}
if state_fingerprint(&target_snapshot)? != target.base_state {
return Err(WorkspaceError::Invalid(format!(
"recovery point for change {} failed integrity verification",
target.change_id
)));
}
let expected_state = logical_state_fingerprint(&target_snapshot)?;
let undo_id = undo_change_id(&target.change_id, ¤t_state);
let undo_file = change_path(workspace, &undo_id);
if let Some(existing) = changes.iter().find(|change| change.change_id == undo_id) {
if existing.status.is_committed() {
return Err(WorkspaceError::Invalid(format!(
"change has already been undone as {undo_id}"
)));
}
if existing.status == ChangeStatus::Unresolved {
return Err(WorkspaceError::Invalid(format!(
"undo change {undo_id} is unresolved; inspect its recovery point first"
)));
}
}
let undo_snapshot = snapshot_path(workspace, &undo_id);
if undo_snapshot.exists() {
if state_fingerprint(&undo_snapshot)? != current_state {
return Err(WorkspaceError::Invalid(format!(
"recovery point for undo {undo_id} does not match the current state"
)));
}
} else {
copy_atomic(&workspace.database_path(), &undo_snapshot)?;
}
let sequence = changes
.iter()
.find(|change| change.change_id == undo_id)
.map(|change| change.sequence)
.unwrap_or(next_sequence(&changes)?);
let mut undo_record = ChangeRecord {
format_version: FORMAT_VERSION,
sequence,
change_id: undo_id.clone(),
kind: ChangeKind::Undo,
plan_id: None,
target_change_id: Some(target.change_id.clone()),
base_generation: current_generation,
base_state: current_state.clone(),
expected_state: Some(expected_state),
snapshot_id: undo_id.clone(),
sql: None,
status: ChangeStatus::Prepared,
committed_generation: None,
after_state: None,
error: None,
import: None,
};
if undo_file.exists() {
write_atomic_json(&undo_file, &undo_record)?;
} else {
write_new_json(&undo_file, &undo_record)?;
}
let database_path = workspace.database_path();
if let Err(error) = database.restore_snapshot_with_budget(&target_snapshot, &mut budget) {
undo_record.status = ChangeStatus::Unresolved;
undo_record.error = Some(error.to_string());
write_atomic_json(&undo_file, &undo_record)?;
return Err(error.into());
}
if let Err(error) = database.checkpoint_with_budget(&mut budget) {
undo_record.status = ChangeStatus::Unresolved;
undo_record.error = Some(error.to_string());
write_atomic_json(&undo_file, &undo_record)?;
return Err(error.into());
}
let restored_state = state_fingerprint(&database_path)?;
let restored_generation = database.generation();
if std::env::var_os("BASALT_CRASH_TEST_AFTER_UNDO_RESTORE").is_some() {
std::process::abort();
}
undo_record.status = ChangeStatus::Committed;
undo_record.committed_generation = Some(restored_generation);
undo_record.after_state = Some(restored_state.clone());
write_atomic_json(&undo_file, &undo_record)?;
Ok(UndoReport {
change_id: undo_id,
undone_change_id: target.change_id,
restored_state,
generation: restored_generation,
})
}
pub(crate) fn mcp_preview(
workspace: &Workspace,
sql: &str,
max_output_bytes: usize,
) -> Result<PlanReport, WorkspaceError> {
let plan = preview_plan_with_output_limit(
workspace,
sql,
Some(max_output_bytes),
Some(MCP_EXECUTION_WORK_LIMIT),
Some(MAX_MCP_MUTATION_ROWS),
)?;
Ok(PlanReport::from(&plan))
}
pub(crate) fn mcp_plan(
workspace: &Workspace,
plan_id: &str,
max_output_bytes: usize,
) -> Result<PlanReport, WorkspaceError> {
let plan = load_plan(workspace, plan_id)?;
let report = PlanReport::from(&plan);
let output_size = serde_json::to_vec(&report)?.len();
if output_size > max_output_bytes {
return Err(WorkspaceError::Invalid(format!(
"workspace plan is {output_size} bytes; response limit is {max_output_bytes} bytes"
)));
}
Ok(report)
}
pub(crate) fn mcp_apply(
workspace: &Workspace,
plan_id: &str,
) -> Result<ApplyReport, WorkspaceError> {
apply_plan(
workspace,
plan_id,
Some(MCP_EXECUTION_WORK_LIMIT),
Some(MAX_MCP_MUTATION_ROWS),
)
}
pub(crate) fn mcp_history(workspace: &Workspace) -> Result<Vec<HistoryEntry>, WorkspaceError> {
history_with_limits(
workspace,
Some(MAX_MCP_HISTORY_ENTRIES),
Some(MAX_MCP_HISTORY_METADATA_BYTES),
)
}
pub(crate) fn mcp_diff(
workspace: &Workspace,
change_id: Option<&str>,
) -> Result<DiffReport, WorkspaceError> {
diff_with_row_limit(
workspace,
change_id,
Some(MAX_MCP_DIFF_ROWS),
Some(MCP_EXECUTION_WORK_LIMIT),
)
}
pub(crate) fn mcp_undo(
workspace: &Workspace,
change_id: &str,
) -> Result<UndoReport, WorkspaceError> {
undo(workspace, change_id, Some(MCP_EXECUTION_WORK_LIMIT))
}
fn import_with_recovery(
workspace: &Workspace,
requested_table: Option<&str>,
format: DataFormat,
bytes: &[u8],
limits: ImportLimits,
) -> Result<ImportReport, WorkspaceError> {
let table = match format {
DataFormat::Sql => {
if requested_table.is_some() {
return Err(WorkspaceError::Usage(
"--table is not valid for SQL imports".to_string(),
));
}
None
}
DataFormat::Csv | DataFormat::Json | DataFormat::JsonLines => {
Some(required_table(requested_table)?.to_string())
}
};
let table_name = table.as_deref();
let validated_summary = match format {
DataFormat::Csv | DataFormat::Json | DataFormat::JsonLines => validate_import(
format,
table_name.ok_or_else(|| {
WorkspaceError::Invalid("row import is missing its table name".to_string())
})?,
bytes,
limits,
)?,
DataFormat::Sql => {
let (_, statements) = parse_sql_import(bytes)?;
format!("{} statements from SQL", statements.len())
}
};
let database = workspace.database()?;
let mut preflight_budget = limits
.max_work
.map(ExecutionBudget::bounded)
.unwrap_or_else(ExecutionBudget::unlimited);
database.checkpoint_with_budget(&mut preflight_budget)?;
let base_state = state_fingerprint(&workspace.database_path())?;
let base_generation = database.generation();
ensure_history_dirs(workspace)?;
let mut changes = load_changes_for_operation(workspace, limits.max_work.is_some())?;
reconcile_changes(workspace, &mut changes, &base_state, base_generation)?;
let request_table = table_name.unwrap_or("");
let request_key = import_request_key(format, request_table, bytes);
if let Some(existing) = changes.iter().rev().find(|change| {
change
.import
.as_ref()
.is_some_and(|import| import.request_key == request_key)
}) && existing.status.is_committed()
{
if existing.after_state.as_deref() != Some(base_state.as_str()) {
return Err(WorkspaceError::Invalid(format!(
"import has already been committed as {}; workspace state moved, so it will not be replayed",
existing.change_id
)));
}
let import = existing.import.as_ref().ok_or_else(|| {
WorkspaceError::Invalid("committed import is missing its metadata".to_string())
})?;
let after_state = existing.after_state.clone().ok_or_else(|| {
WorkspaceError::Invalid("committed import is missing its after-state".to_string())
})?;
let generation = existing.committed_generation.ok_or_else(|| {
WorkspaceError::Invalid("committed import is missing its generation".to_string())
})?;
return Ok(ImportReport {
change_id: existing.change_id.clone(),
format: import.format.clone(),
table: import.table.clone(),
bytes: import.bytes,
summary: import.summary.clone(),
base_state: existing.base_state.clone(),
after_state,
generation,
});
}
let change_id = import_change_id(&base_state, format, request_table, bytes);
let retry_sequence = if let Some(existing) =
changes.iter().find(|change| change.change_id == change_id)
{
if existing.status.is_committed() {
return Err(WorkspaceError::Invalid(format!(
"import has already been committed as change {change_id}"
)));
}
if existing.status == ChangeStatus::Failed
&& existing.base_state == base_state
&& existing.after_state.is_none()
&& existing
.import
.as_ref()
.is_some_and(|import| import.request_key == request_key)
{
Some(existing.sequence)
} else {
return Err(WorkspaceError::Invalid(format!(
"import already has a history record with status {:?}; inspect workspace history before retrying",
existing.status
)));
}
} else {
None
};
let snapshot = snapshot_path(workspace, &change_id);
if snapshot.exists() {
if state_fingerprint(&snapshot)? != base_state {
return Err(WorkspaceError::Invalid(format!(
"recovery point for import {change_id} does not match the workspace"
)));
}
} else {
copy_atomic(&workspace.database_path(), &snapshot)?;
}
let change_file = change_path(workspace, &change_id);
let mut change = ChangeRecord {
format_version: FORMAT_VERSION,
sequence: match retry_sequence {
Some(sequence) => sequence,
None => next_sequence(&changes)?,
},
change_id: change_id.clone(),
kind: ChangeKind::Apply,
plan_id: None,
target_change_id: None,
base_generation,
base_state: base_state.clone(),
expected_state: None,
snapshot_id: change_id.clone(),
sql: None,
status: ChangeStatus::Prepared,
committed_generation: None,
after_state: None,
error: None,
import: Some(ImportMetadata {
request_key,
format: format.name().to_string(),
table: table.clone(),
bytes: bytes.len(),
summary: validated_summary,
}),
};
if change_file.exists() {
write_atomic_json(&change_file, &change)?;
} else {
write_new_json(&change_file, &change)?;
}
let imported = match format {
DataFormat::Csv => import_csv(&database, table_name, bytes, limits),
DataFormat::Json => import_json(&database, table_name, bytes, limits),
DataFormat::JsonLines => import_json_lines(&database, table_name, bytes, limits),
DataFormat::Sql => import_sql(&database, bytes),
};
let summary = match imported {
Ok(summary) => summary.unwrap_or_else(|| "import completed".to_string()),
Err(error) => {
change.status = ChangeStatus::Failed;
change.error = Some(error.to_string());
write_atomic_json(&change_file, &change)?;
return Err(error);
}
};
if let Some(import) = change.import.as_mut() {
import.summary = summary.clone();
}
if let Err(error) = database.checkpoint_with_budget(&mut preflight_budget) {
change.status = ChangeStatus::Unresolved;
change.error = Some(format!("import committed but checkpoint failed: {error}"));
write_atomic_json(&change_file, &change)?;
return Err(error.into());
}
let after_state = state_fingerprint(&workspace.database_path())?;
let generation = database.generation();
if std::env::var_os("BASALT_CRASH_TEST_AFTER_IMPORT_CHECKPOINT").is_some() {
std::process::abort();
}
change.status = ChangeStatus::Committed;
change.committed_generation = Some(generation);
change.after_state = Some(after_state.clone());
write_atomic_json(&change_file, &change)?;
Ok(ImportReport {
change_id,
format: format.name().to_string(),
table,
bytes: bytes.len(),
summary,
base_state,
after_state,
generation,
})
}
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct ImportReport {
change_id: String,
format: String,
table: Option<String>,
bytes: usize,
summary: String,
base_state: String,
after_state: String,
generation: u64,
}
pub(crate) fn mcp_import(
workspace: &Workspace,
table: Option<&str>,
format: &str,
content: &str,
) -> Result<ImportReport, WorkspaceError> {
if content.len() > MAX_MCP_IMPORT_BYTES {
return Err(WorkspaceError::Invalid(format!(
"MCP import content exceeds the {} MiB limit",
MAX_MCP_IMPORT_BYTES / (1024 * 1024)
)));
}
let format = DataFormat::parse(format)?;
if format == DataFormat::Sql {
return Err(WorkspaceError::Usage(
"workspace_import accepts csv, json, or jsonl; use the CLI for SQL dump imports"
.to_string(),
));
}
let table = required_mcp_table(table)?;
import_with_recovery(
workspace,
Some(table),
format,
content.as_bytes(),
ImportLimits::mcp(),
)
}
fn required_mcp_table(table: Option<&str>) -> Result<&str, WorkspaceError> {
let table = table.ok_or_else(|| {
WorkspaceError::Usage("workspace_import requires an explicit table name".to_string())
})?;
validate_name(table, "table name")?;
Ok(table)
}
fn validate_import(
format: DataFormat,
table: &str,
bytes: &[u8],
limits: ImportLimits,
) -> Result<String, WorkspaceError> {
let database = Database::in_memory();
match format {
DataFormat::Csv => import_csv(&database, Some(table), bytes, limits),
DataFormat::Json => import_json(&database, Some(table), bytes, limits),
DataFormat::JsonLines => import_json_lines(&database, Some(table), bytes, limits),
DataFormat::Sql => unreachable!(),
}
.map(|summary| summary.unwrap_or_else(|| "import completed".to_string()))
}
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct ExportReport {
table: String,
format: String,
content: String,
bytes: usize,
}
pub(crate) fn mcp_export(
workspace: &Workspace,
table: &str,
format: &str,
max_content_bytes: usize,
) -> Result<ExportReport, WorkspaceError> {
let format = DataFormat::parse(format)?;
if format == DataFormat::Json {
return Err(WorkspaceError::Usage(
"JSON export is JSON Lines; use jsonl".to_string(),
));
}
let database = workspace.database()?;
let row_count = database.row_count(table)?;
if row_count > MAX_MCP_EXPORT_ROWS {
return Err(WorkspaceError::Invalid(format!(
"MCP export is limited to {MAX_MCP_EXPORT_ROWS} rows; use the CLI export for larger tables"
)));
}
let mut budget = ExecutionBudget::bounded(MCP_EXECUTION_WORK_LIMIT);
let (columns, rows) = select_table_with_budget(&database, table, Some(&mut budget))?;
debug_assert_eq!(rows.len(), row_count);
let mut output = LimitedBuffer::new(max_content_bytes);
let result = match format {
DataFormat::Csv => write_csv(&columns, &rows, &mut output),
DataFormat::JsonLines => write_json_lines(&columns, &rows, &mut output),
DataFormat::Sql => write_sql(&database, table, &rows, &mut output),
DataFormat::Json => unreachable!(),
};
if output.exceeded {
return Err(WorkspaceError::Invalid(format!(
"MCP export exceeds the {max_content_bytes}-byte content limit"
)));
}
result?;
let content = String::from_utf8(output.into_inner())
.map_err(|error| WorkspaceError::Invalid(format!("export is not valid UTF-8: {error}")))?;
let bytes = content.len();
Ok(ExportReport {
table: table.to_string(),
format: format.name().to_string(),
bytes,
content,
})
}
pub(crate) fn mcp_inspect(workspace: &Workspace) -> Result<InspectReport, WorkspaceError> {
let database = workspace.database()?;
inspect(workspace, &database)
}
fn required_table(table: Option<&str>) -> Result<&str, WorkspaceError> {
let table = table.ok_or_else(|| {
WorkspaceError::Usage("row imports require --table NAME or a source filename".to_string())
})?;
validate_name(table, "table name")?;
Ok(table)
}
fn validate_headers(headers: &StringRecord) -> Result<Vec<String>, WorkspaceError> {
let mut seen = HashMap::new();
let mut columns = Vec::with_capacity(headers.len());
for header in headers {
validate_name(header, "CSV header")?;
let key = header.to_ascii_lowercase();
if seen.insert(key, ()).is_some() {
return Err(WorkspaceError::Invalid(format!(
"duplicate CSV header: {header:?}"
)));
}
columns.push(header.to_string());
}
if columns.is_empty() {
return Err(WorkspaceError::Invalid(
"CSV input must contain a header row".to_string(),
));
}
Ok(columns)
}
fn validate_name(name: &str, label: &str) -> Result<(), WorkspaceError> {
if name.trim().is_empty() {
return Err(WorkspaceError::Invalid(format!("{label} cannot be empty")));
}
if name.contains('\0') {
return Err(WorkspaceError::Invalid(format!(
"{label} cannot contain NUL bytes"
)));
}
Ok(())
}
fn parse_csv_cell(value: &str) -> ImportedCell {
if value.is_empty() {
return ImportedCell::Empty;
}
if value.eq_ignore_ascii_case("true") {
return ImportedCell::Boolean(true);
}
if value.eq_ignore_ascii_case("false") {
return ImportedCell::Boolean(false);
}
if let Ok(integer) = value.parse::<i64>() {
return ImportedCell::Integer(integer);
}
if let Ok(real) = value.parse::<f64>()
&& real.is_finite()
{
return ImportedCell::Real(real);
}
ImportedCell::Text(value.to_string())
}
fn json_cell(value: &JsonValue) -> ImportedCell {
match value {
JsonValue::Null => ImportedCell::Null,
JsonValue::Bool(value) => ImportedCell::Boolean(*value),
JsonValue::Number(value) => value
.as_i64()
.map(ImportedCell::Integer)
.or_else(|| {
value
.as_f64()
.filter(|value| value.is_finite())
.map(ImportedCell::Real)
})
.unwrap_or_else(|| ImportedCell::Text(value.to_string())),
JsonValue::String(value) => ImportedCell::Text(value.clone()),
JsonValue::Array(_) | JsonValue::Object(_) => ImportedCell::Text(value.to_string()),
}
}
fn infer_types(rows: &[Vec<ImportedCell>], width: usize) -> Vec<ColumnType> {
(0..width)
.map(|column| {
let mut inferred = None;
for row in rows {
let kind = match row.get(column) {
Some(ImportedCell::Null | ImportedCell::Empty) | None => None,
Some(ImportedCell::Integer(_)) => Some(ColumnType::Integer),
Some(ImportedCell::Real(_)) => Some(ColumnType::Real),
Some(ImportedCell::Boolean(_)) => Some(ColumnType::Boolean),
Some(ImportedCell::Text(_)) => Some(ColumnType::Text),
};
inferred = merge_inferred(inferred, kind);
if inferred == Some(ColumnType::Text) {
break;
}
}
inferred.unwrap_or(ColumnType::Text)
})
.collect()
}
fn merge_inferred(current: Option<ColumnType>, next: Option<ColumnType>) -> Option<ColumnType> {
match (current, next) {
(None, value) | (value, None) => value,
(Some(ColumnType::Integer), Some(ColumnType::Integer)) => Some(ColumnType::Integer),
(Some(ColumnType::Integer), Some(ColumnType::Real))
| (Some(ColumnType::Real), Some(ColumnType::Integer))
| (Some(ColumnType::Real), Some(ColumnType::Real)) => Some(ColumnType::Real),
(Some(ColumnType::Boolean), Some(ColumnType::Boolean)) => Some(ColumnType::Boolean),
_ => Some(ColumnType::Text),
}
}
impl ImportedRows {
fn summary(&self) -> String {
format!(
"table {} ({} rows, {} columns)",
self.table,
self.rows.len(),
self.columns.len()
)
}
}
fn import_rows(
database: &Database,
imported: &ImportedRows,
max_work: Option<usize>,
) -> Result<(), WorkspaceError> {
validate_name(&imported.table, "table name")?;
if imported.columns.is_empty() {
return Err(WorkspaceError::Invalid(
"cannot import a table with no columns".to_string(),
));
}
if imported
.rows
.iter()
.any(|row| row.len() != imported.columns.len())
{
return Err(WorkspaceError::Invalid(
"imported row width does not match the header".to_string(),
));
}
let definitions = imported
.columns
.iter()
.zip(&imported.types)
.map(|(name, ty)| format!("{} {}", quote_identifier(name), column_type_name(ty)))
.collect::<Vec<_>>();
let create = format!(
"CREATE TABLE {} ({})",
quote_identifier(&imported.table),
definitions.join(", ")
);
let mut budget = max_work
.map(ExecutionBudget::bounded)
.unwrap_or_else(ExecutionBudget::unlimited);
let mut connection = database.connect();
connection.execute_with_budget(&Statement::Begin, &mut budget)?;
let result = (|| {
connection.execute_sql_using_budget(&create, &mut budget)?;
for chunk in imported.rows.chunks(IMPORT_BATCH_SIZE) {
if chunk.is_empty() {
continue;
}
let rows = chunk
.iter()
.map(|row| {
row.iter()
.zip(&imported.types)
.map(|(cell, ty)| cell_to_value(cell, ty).map(|value| sql_literal(&value)))
.collect::<Result<Vec<_>, WorkspaceError>>()
.map(|values| format!("({})", values.join(", ")))
})
.collect::<Result<Vec<_>, WorkspaceError>>()?;
let insert = format!(
"INSERT INTO {} VALUES {}",
quote_identifier(&imported.table),
rows.join(", ")
);
connection.execute_sql_using_budget(&insert, &mut budget)?;
}
connection.execute_with_budget(&Statement::Commit, &mut budget)?;
Ok::<(), WorkspaceError>(())
})();
if let Err(error) = result {
let _ = connection.execute_sql("ROLLBACK");
return Err(error);
}
Ok(())
}
fn cell_to_value(cell: &ImportedCell, ty: &ColumnType) -> Result<Value, WorkspaceError> {
let value = match (cell, ty) {
(ImportedCell::Null, _) => Value::Null,
(ImportedCell::Empty, ColumnType::Text) => Value::Text(String::new()),
(ImportedCell::Empty, _) => Value::Null,
(ImportedCell::Integer(value), ColumnType::Integer) => Value::Integer(*value),
(ImportedCell::Integer(value), ColumnType::Real) => Value::Real(*value as f64),
(ImportedCell::Integer(value), ColumnType::Boolean) => Value::Boolean(*value != 0),
(ImportedCell::Integer(value), ColumnType::Text) => Value::Text(value.to_string()),
(ImportedCell::Real(value), ColumnType::Real) => Value::Real(*value),
(ImportedCell::Real(value), ColumnType::Text) => Value::Text(value.to_string()),
(ImportedCell::Real(value), ColumnType::Integer) => {
return Err(WorkspaceError::Invalid(format!(
"cannot store REAL value {value} in inferred INTEGER column"
)));
}
(ImportedCell::Real(value), ColumnType::Boolean) => {
return Err(WorkspaceError::Invalid(format!(
"cannot store REAL value {value} in inferred BOOLEAN column"
)));
}
(ImportedCell::Boolean(value), ColumnType::Boolean) => Value::Boolean(*value),
(ImportedCell::Boolean(value), ColumnType::Text) => Value::Text(value.to_string()),
(ImportedCell::Boolean(value), ColumnType::Integer) => Value::Integer(*value as i64),
(ImportedCell::Boolean(value), ColumnType::Real) => Value::Real(*value as u8 as f64),
(ImportedCell::Text(value), ColumnType::Text) => Value::Text(value.clone()),
(ImportedCell::Text(value), _) => {
return Err(WorkspaceError::Invalid(format!(
"text value {value:?} conflicts with inferred {:?} column",
ty
)));
}
(_, ColumnType::Any | ColumnType::Null) => Value::Null,
};
Ok(value)
}
fn select_table(
database: &Database,
table: &str,
) -> Result<(Vec<String>, Vec<Vec<Value>>), WorkspaceError> {
select_table_with_budget(database, table, None)
}
fn select_table_with_budget(
database: &Database,
table: &str,
budget: Option<&mut ExecutionBudget>,
) -> Result<(Vec<String>, Vec<Vec<Value>>), WorkspaceError> {
validate_name(table, "table name")?;
let sql = format!("SELECT * FROM {}", quote_identifier(table));
let results = match budget {
Some(budget) => {
let mut connection = database.connect();
connection.execute_sql_using_budget(&sql, budget)?
}
None => database.execute_sql(&sql)?,
};
let Some(StatementResult::Select { columns, rows }) = results.into_iter().next() else {
return Err(WorkspaceError::Invalid(
"table query did not return rows".to_string(),
));
};
Ok((columns, rows))
}
fn export_csv(columns: &[String], rows: &[Vec<Value>]) -> Result<Vec<u8>, WorkspaceError> {
let mut output = Vec::new();
write_csv(columns, rows, &mut output)?;
Ok(output)
}
fn write_csv<W: Write>(
columns: &[String],
rows: &[Vec<Value>],
output: &mut W,
) -> Result<(), WorkspaceError> {
let mut writer = Writer::from_writer(output);
writer.write_record(columns)?;
for row in rows {
ensure_row_width(columns, row)?;
let values = row.iter().map(csv_field).collect::<Vec<_>>();
writer.write_record(values)?;
}
writer.flush()?;
Ok(())
}
fn export_json_lines(columns: &[String], rows: &[Vec<Value>]) -> Result<Vec<u8>, WorkspaceError> {
let mut output = Vec::new();
write_json_lines(columns, rows, &mut output)?;
Ok(output)
}
fn write_json_lines<W: Write>(
columns: &[String],
rows: &[Vec<Value>],
output: &mut W,
) -> Result<(), WorkspaceError> {
for row in rows {
ensure_row_width(columns, row)?;
let mut object = Map::new();
for (column, value) in columns.iter().zip(row) {
object.insert(column.clone(), value_to_json(value)?);
}
serde_json::to_writer(&mut *output, &JsonValue::Object(object))?;
output.write_all(b"\n")?;
}
Ok(())
}
fn export_sql(
database: &Database,
table: &str,
rows: &[Vec<Value>],
) -> Result<Vec<u8>, WorkspaceError> {
let mut output = Vec::new();
write_sql(database, table, rows, &mut output)?;
Ok(output)
}
fn write_sql<W: Write>(
database: &Database,
table: &str,
rows: &[Vec<Value>],
output: &mut W,
) -> Result<(), WorkspaceError> {
let columns = database.columns(table)?;
write!(output, "CREATE TABLE {} (", quote_identifier(table))?;
for (index, column) in columns.iter().enumerate() {
if index > 0 {
output.write_all(b", ")?;
}
output.write_all(column_definition(column).as_bytes())?;
}
output.write_all(b");\n")?;
for row in rows {
ensure_row_width(
&columns
.iter()
.map(|column| column.name.clone())
.collect::<Vec<_>>(),
row,
)?;
write!(output, "INSERT INTO {} VALUES (", quote_identifier(table))?;
for (index, value) in row.iter().enumerate() {
if index > 0 {
output.write_all(b", ")?;
}
output.write_all(sql_literal(value).as_bytes())?;
}
output.write_all(b");\n")?;
}
Ok(())
}
struct LimitedBuffer {
bytes: Vec<u8>,
limit: usize,
exceeded: bool,
}
impl LimitedBuffer {
fn new(limit: usize) -> Self {
Self {
bytes: Vec::new(),
limit,
exceeded: false,
}
}
fn into_inner(self) -> Vec<u8> {
self.bytes
}
}
impl Write for LimitedBuffer {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
if bytes.len() > self.limit.saturating_sub(self.bytes.len()) {
self.exceeded = true;
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"export content limit exceeded",
));
}
self.bytes.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
fn column_definition(column: &Column) -> String {
let mut definition = format!(
"{} {}",
quote_identifier(&column.name),
column_type_name(&column.ty)
);
if column.primary_key {
definition.push_str(" PRIMARY KEY");
} else if column.not_null {
definition.push_str(" NOT NULL");
}
if column.unique && !column.primary_key {
definition.push_str(" UNIQUE");
}
definition
}
fn ensure_row_width(columns: &[String], row: &[Value]) -> Result<(), WorkspaceError> {
if columns.len() != row.len() {
return Err(WorkspaceError::Invalid(format!(
"row width mismatch: expected {}, got {}",
columns.len(),
row.len()
)));
}
Ok(())
}
fn csv_field(value: &Value) -> String {
match value {
Value::Null => String::new(),
Value::Integer(value) => value.to_string(),
Value::Real(value) => value.to_string(),
Value::Text(value) => value.clone(),
Value::Boolean(value) => value.to_string(),
}
}
fn value_to_json(value: &Value) -> Result<JsonValue, WorkspaceError> {
match value {
Value::Null => Ok(JsonValue::Null),
Value::Integer(value) => Ok(JsonValue::Number((*value).into())),
Value::Real(value) => serde_json::Number::from_f64(*value)
.map(JsonValue::Number)
.ok_or_else(|| WorkspaceError::Invalid("cannot export non-finite REAL".to_string())),
Value::Text(value) => Ok(JsonValue::String(value.clone())),
Value::Boolean(value) => Ok(JsonValue::Bool(*value)),
}
}
fn quote_identifier(value: &str) -> String {
format!("\"{}\"", value.replace('"', "\"\""))
}
fn column_type_name(ty: &ColumnType) -> &'static str {
match ty {
ColumnType::Integer => "INTEGER",
ColumnType::Real => "REAL",
ColumnType::Text => "TEXT",
ColumnType::Boolean => "BOOLEAN",
ColumnType::Any => "TEXT",
ColumnType::Null => "TEXT",
}
}
fn sql_literal(value: &Value) -> String {
match value {
Value::Null => "NULL".to_string(),
Value::Integer(value) => value.to_string(),
Value::Real(value) => value.to_string(),
Value::Text(value) => format!("'{}'", value.replace('\'', "''")),
Value::Boolean(value) => value.to_string(),
}
}
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct InspectReport {
path: String,
format_version: u32,
database: String,
tables: Vec<InspectTable>,
}
#[derive(Debug, Serialize, JsonSchema)]
struct InspectTable {
name: String,
rows: usize,
columns: Vec<InspectColumn>,
}
#[derive(Debug, Serialize, JsonSchema)]
struct InspectColumn {
name: String,
data_type: &'static str,
not_null: bool,
unique: bool,
primary_key: bool,
}
fn inspect(workspace: &Workspace, database: &Database) -> Result<InspectReport, WorkspaceError> {
let mut tables = Vec::new();
for table_name in database.table_names()? {
let columns = database.columns(&table_name)?;
let rows = table_row_count(database, &table_name)?;
tables.push(InspectTable {
name: table_name,
rows,
columns: columns
.iter()
.map(|column| InspectColumn {
name: column.name.clone(),
data_type: column_type_name(&column.ty),
not_null: column.not_null,
unique: column.unique,
primary_key: column.primary_key,
})
.collect(),
});
}
Ok(InspectReport {
path: workspace.root.display().to_string(),
format_version: workspace.manifest.format_version,
database: workspace.manifest.database.clone(),
tables,
})
}
fn table_row_count(database: &Database, table: &str) -> Result<usize, WorkspaceError> {
Ok(database.row_count(table)?)
}
fn render_inspect(report: &InspectReport, output: &mut dyn Write) -> Result<(), WorkspaceError> {
writeln!(output, "Workspace: {}", report.path)?;
writeln!(output, "Workspace format: {}", report.format_version)?;
writeln!(output, "Database: {}", report.database)?;
if report.tables.is_empty() {
writeln!(output, "Tables: none")?;
return Ok(());
}
writeln!(output, "Tables: {}", report.tables.len())?;
for table in &report.tables {
let columns = table
.columns
.iter()
.map(|column| format!("{} {}", column.name, column.data_type))
.collect::<Vec<_>>()
.join(", ");
writeln!(
output,
"- {} ({} rows): {}",
table.name, table.rows, columns
)?;
}
Ok(())
}
fn render_plan(
workspace: &Workspace,
plan: &PlanRecord,
output: &mut dyn Write,
) -> Result<(), WorkspaceError> {
writeln!(output, "Plan: {}", plan.plan_id)?;
writeln!(output, "Workspace: {}", workspace.root.display())?;
writeln!(output, "Base state: {}", plan.base_state)?;
writeln!(output, "Statements: {}", plan.statements.len())?;
for item in &plan.statements {
let detail = item
.rows_affected
.map(|rows| format!("{rows} row(s) affected"))
.or_else(|| {
item.rows_returned
.map(|rows| format!("{rows} row(s) returned"))
})
.or_else(|| item.object.clone())
.unwrap_or_default();
if detail.is_empty() {
writeln!(output, "- {}: {}", item.statement, item.kind)?;
} else {
writeln!(output, "- {}: {} ({detail})", item.statement, item.kind)?;
}
}
writeln!(
output,
"Apply: basalt workspace apply {} {}",
workspace.root.display(),
plan.plan_id
)?;
Ok(())
}
fn render_apply(report: &ApplyReport, output: &mut dyn Write) -> Result<(), WorkspaceError> {
writeln!(output, "Applied change {}", report.change_id)?;
writeln!(output, "Plan: {}", report.plan_id)?;
writeln!(output, "State: {}", report.after_state)?;
writeln!(output, "Generation: {}", report.generation)?;
Ok(())
}
fn render_history(entries: &[HistoryEntry], output: &mut dyn Write) -> Result<(), WorkspaceError> {
if entries.is_empty() {
writeln!(output, "No changes.")?;
return Ok(());
}
for entry in entries {
writeln!(
output,
"#{} {} {} {}",
entry.sequence,
entry.change_id,
change_kind_name(&entry.kind),
change_status_name(&entry.status)
)?;
if let Some(error) = &entry.error {
writeln!(output, " {error}")?;
}
}
Ok(())
}
fn render_diff(report: &DiffReport, output: &mut dyn Write) -> Result<(), WorkspaceError> {
writeln!(output, "Diff for change {}", report.change_id)?;
writeln!(output, "Precision: {}", report.precision)?;
writeln!(output, "Before: {}", report.before_state)?;
writeln!(output, "Current: {}", report.current_state)?;
if report.tables.is_empty() {
writeln!(output, "No logical table changes.")?;
return Ok(());
}
for table in &report.tables {
let before = table
.before_rows
.map(|rows| rows.to_string())
.unwrap_or_else(|| "absent".to_string());
let after = table
.after_rows
.map(|rows| rows.to_string())
.unwrap_or_else(|| "absent".to_string());
writeln!(
output,
"- {}: {} -> {} row(s), schema_changed={}, data_changed={}",
table.table, before, after, table.schema_changed, table.data_changed
)?;
writeln!(
output,
" rows added: {}, rows removed: {}",
table.added_rows, table.removed_rows
)?;
}
Ok(())
}
fn render_undo(report: &UndoReport, output: &mut dyn Write) -> Result<(), WorkspaceError> {
writeln!(
output,
"Undid change {} as {}",
report.undone_change_id, report.change_id
)?;
writeln!(output, "Restored state: {}", report.restored_state)?;
writeln!(output, "Generation: {}", report.generation)?;
Ok(())
}
fn change_kind_name(kind: &ChangeKind) -> &'static str {
match kind {
ChangeKind::Apply => "apply",
ChangeKind::Undo => "undo",
}
}
fn change_status_name(status: &ChangeStatus) -> &'static str {
match status {
ChangeStatus::Prepared => "prepared",
ChangeStatus::Committed => "committed",
ChangeStatus::Recovered => "recovered",
ChangeStatus::Failed => "failed",
ChangeStatus::Unresolved => "unresolved",
}
}
fn write_output(
workspace: &Workspace,
destination: &Path,
bytes: &[u8],
stdout: &mut dyn Write,
) -> Result<(), WorkspaceError> {
if destination.as_os_str() == OsStr::new("-") {
stdout.write_all(bytes)?;
return Ok(());
}
if is_protected_workspace_path(workspace, destination) {
return Err(WorkspaceError::Invalid(
"refusing to overwrite workspace metadata, locks, database, or history".to_string(),
));
}
if let Some(parent) = destination
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
fs::create_dir_all(parent)?;
}
atomic_write_file(destination, bytes)
}
fn atomic_write_file(path: &Path, bytes: &[u8]) -> Result<(), WorkspaceError> {
if path_is_symlink(path)? {
return Err(WorkspaceError::Invalid(format!(
"refusing to write through a symbolic link: {}",
path.display()
)));
}
if let Some(parent) = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
fs::create_dir_all(parent)?;
}
let temporary = temporary_path(path);
let write_result = (|| {
let mut file = OpenOptions::new()
.create_new(true)
.write(true)
.open(&temporary)?;
file.write_all(bytes)?;
file.sync_all()?;
drop(file);
match fs::rename(&temporary, path) {
Ok(()) => Ok(()),
Err(error) if path.exists() => {
fs::remove_file(path)?;
fs::rename(&temporary, path).map_err(|_| error)
}
Err(error) => Err(error),
}?;
sync_parent(path)?;
Ok(())
})();
if write_result.is_err() {
let _ = fs::remove_file(&temporary);
}
write_result.map_err(WorkspaceError::Io)
}
fn write_new_file(path: &Path, bytes: &[u8]) -> Result<(), WorkspaceError> {
if path_is_symlink(path)? {
return Err(WorkspaceError::Invalid(format!(
"refusing to create through a symbolic link: {}",
path.display()
)));
}
let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
file.write_all(bytes)?;
file.sync_all()?;
drop(file);
sync_parent(path)?;
Ok(())
}
fn cleanup_failed_init(
root: &Path,
remove_artifacts: bool,
remove_workspace_lock: bool,
remove_root_if_empty: bool,
) -> Result<(), WorkspaceError> {
if remove_artifacts {
for name in [
MANIFEST_FILE,
DATABASE_FILE,
"data.basalt.wal",
"data.basalt.lock",
"data.basalt.tmp",
] {
let path = root.join(name);
let metadata = match fs::symlink_metadata(&path) {
Ok(metadata) => metadata,
Err(error) if error.kind() == io::ErrorKind::NotFound => continue,
Err(error) => return Err(WorkspaceError::Io(error)),
};
if metadata.file_type().is_symlink() || !metadata.is_file() {
return Err(WorkspaceError::Invalid(format!(
"cannot clean failed workspace initialization artifact: {}",
path.display()
)));
}
fs::remove_file(&path)?;
}
}
if remove_workspace_lock {
let path = root.join(WORKSPACE_LOCK_FILE);
match fs::symlink_metadata(&path) {
Ok(metadata) if metadata.is_file() => fs::remove_file(&path)?,
Ok(_) => {
return Err(WorkspaceError::Invalid(format!(
"cannot clean failed workspace lock: {}",
path.display()
)));
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(error) => return Err(WorkspaceError::Io(error)),
}
}
sync_parent(root)?;
if remove_root_if_empty {
match fs::remove_dir(root) {
Ok(()) => {}
Err(error)
if matches!(
error.kind(),
io::ErrorKind::NotFound | io::ErrorKind::DirectoryNotEmpty
) => {}
Err(error) => return Err(WorkspaceError::Io(error)),
}
}
Ok(())
}
fn path_is_symlink(path: &Path) -> Result<bool, WorkspaceError> {
match fs::symlink_metadata(path) {
Ok(metadata) => Ok(metadata.file_type().is_symlink()),
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false),
Err(error) => Err(WorkspaceError::Io(error)),
}
}
#[cfg(unix)]
fn sync_parent(path: &Path) -> io::Result<()> {
if let Some(parent) = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
let directory = File::open(parent)?;
directory.sync_all()?;
}
Ok(())
}
#[cfg(not(unix))]
fn sync_parent(_path: &Path) -> io::Result<()> {
Ok(())
}
fn acquire_workspace_lock(root: &Path) -> Result<Arc<File>, WorkspaceError> {
let path = root.join(WORKSPACE_LOCK_FILE);
if path_is_symlink(&path)? {
return Err(WorkspaceError::Invalid(
"workspace lock cannot be a symbolic link".to_string(),
));
}
let file = OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&path)?;
match fs4::FileExt::try_lock(&file) {
Ok(()) => Ok(Arc::new(file)),
Err(fs4::TryLockError::WouldBlock) => Err(WorkspaceError::Invalid(format!(
"workspace is already open: {}",
root.display()
))),
Err(fs4::TryLockError::Error(error)) => Err(WorkspaceError::Io(error)),
}
}
fn validate_database_paths(root: &Path) -> Result<(), WorkspaceError> {
let database = root.join(DATABASE_FILE);
if path_is_symlink(&database)? {
return Err(WorkspaceError::Invalid(
"workspace database cannot be a symbolic link".to_string(),
));
}
if database.exists() && !database.is_file() {
return Err(WorkspaceError::Invalid(format!(
"workspace database is not a file: {}",
database.display()
)));
}
for suffix in [".wal", ".lock", ".tmp"] {
let mut value = database.as_os_str().to_os_string();
value.push(suffix);
let path = PathBuf::from(value);
if path_is_symlink(&path)? {
return Err(WorkspaceError::Invalid(format!(
"workspace database sidecar cannot be a symbolic link: {}",
path.display()
)));
}
if path.exists() && !path.is_file() {
return Err(WorkspaceError::Invalid(format!(
"workspace database sidecar is not a file: {}",
path.display()
)));
}
}
let mut wal_value = database.as_os_str().to_os_string();
wal_value.push(".wal");
if !database.exists() && !Path::new(&wal_value).exists() {
return Err(WorkspaceError::Invalid(format!(
"workspace database is missing: {}",
database.display()
)));
}
if !database.exists() {
let frame = crate::wal::latest(Path::new(&wal_value))?;
if frame.as_ref().is_none_or(|frame| frame.generation == 0) {
return Err(WorkspaceError::Invalid(format!(
"workspace database is missing and its WAL has no recoverable committed frame: {}",
database.display()
)));
}
}
Ok(())
}
fn manifest_bytes(manifest: &WorkspaceManifest) -> Result<Vec<u8>, WorkspaceError> {
let mut bytes = serde_json::to_vec_pretty(manifest)?;
bytes.push(b'\n');
Ok(bytes)
}
fn temporary_path(destination: &Path) -> PathBuf {
let stamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_nanos())
.unwrap_or_default();
let name = destination
.file_name()
.and_then(OsStr::to_str)
.unwrap_or("export");
let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
destination.with_file_name(format!(
".{name}.basalt-tmp-{}-{stamp}-{counter}",
std::process::id()
))
}
fn is_protected_workspace_path(workspace: &Workspace, destination: &Path) -> bool {
let database = workspace.database_path();
let protected = [
workspace.root.join(MANIFEST_FILE),
workspace.root.join(WORKSPACE_LOCK_FILE),
database.clone(),
sidecar_path(&database, ".wal"),
sidecar_path(&database, ".lock"),
sidecar_path(&database, ".tmp"),
workspace.root.join(HISTORY_DIR),
];
protected
.iter()
.any(|path| path_is_same_or_descendant(destination, path))
}
fn sidecar_path(path: &Path, suffix: &str) -> PathBuf {
let mut value = path.as_os_str().to_os_string();
value.push(suffix);
PathBuf::from(value)
}
fn path_is_same_or_descendant(path: &Path, protected: &Path) -> bool {
let normalized_destination = normalized_path(path);
let normalized_protected = normalized_path(protected);
if normalized_destination == normalized_protected
|| normalized_destination.starts_with(&normalized_protected)
{
return true;
}
match (resolved_path(path), resolved_path(protected)) {
(Some(path), Some(protected)) => path == protected || path.starts_with(&protected),
_ => false,
}
}
#[cfg(test)]
fn same_path(left: &Path, right: &Path) -> bool {
if left == right {
return true;
}
match (resolved_path(left), resolved_path(right)) {
(Some(left), Some(right)) => left == right,
_ => normalized_path(left) == normalized_path(right),
}
}
fn resolved_path(path: &Path) -> Option<PathBuf> {
let mut candidate = path.to_path_buf();
let mut suffix = Vec::new();
loop {
match fs::canonicalize(&candidate) {
Ok(mut resolved) => {
for component in suffix.iter().rev() {
resolved.push(component);
}
return Some(resolved);
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {
let name = candidate.file_name()?.to_os_string();
suffix.push(name);
if !candidate.pop() {
return None;
}
}
Err(_) => return None,
}
}
}
fn normalized_path(path: &Path) -> PathBuf {
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir()
.map(|directory| directory.join(path))
.unwrap_or_else(|_| path.to_path_buf())
};
let mut components = Vec::new();
for component in absolute.components() {
match component {
Component::CurDir => {}
Component::ParentDir => {
if matches!(components.last(), Some(Component::Normal(_))) {
components.pop();
}
}
_ => components.push(component),
}
}
components
.into_iter()
.fold(PathBuf::new(), |mut path, component| {
path.push(component.as_os_str());
path
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn infers_numeric_and_text_columns() {
let rows = vec![
vec![ImportedCell::Integer(1), ImportedCell::Text("a".into())],
vec![ImportedCell::Real(2.5), ImportedCell::Empty],
];
assert_eq!(
infer_types(&rows, 2),
vec![ColumnType::Real, ColumnType::Text]
);
}
#[test]
fn quotes_identifiers_and_literals() {
assert_eq!(quote_identifier("a\"b"), "\"a\"\"b\"");
assert_eq!(sql_literal(&Value::Text("a'b".into())), "'a''b'");
}
#[test]
fn manifest_is_stable() {
let manifest = WorkspaceManifest {
format_version: FORMAT_VERSION,
database: DATABASE_FILE.to_string(),
};
assert_eq!(
String::from_utf8(manifest_bytes(&manifest).unwrap()).unwrap(),
"{\n \"format_version\": 1,\n \"database\": \"data.basalt\"\n}\n"
);
}
#[test]
fn open_or_init_does_not_replace_existing_directories() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-open-or-init-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
fs::create_dir_all(&root).unwrap();
let marker = root.join("keep.txt");
fs::write(&marker, b"keep").unwrap();
let error = Workspace::open_or_init(&root).unwrap_err();
assert!(error.to_string().contains("not a Basalt workspace"));
assert_eq!(fs::read(&marker).unwrap(), b"keep");
fs::remove_dir_all(root).unwrap();
}
#[test]
fn init_rejects_reserved_database_sidecars_before_writing_a_manifest() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-init-sidecar-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
fs::create_dir_all(&root).unwrap();
let sidecar = root.join("data.basalt.lock");
fs::write(&sidecar, b"reserved").unwrap();
let error = Workspace::init(&root).unwrap_err();
assert!(error.to_string().contains("reserved database sidecar"));
assert!(!root.join(MANIFEST_FILE).exists());
assert!(!root.join(WORKSPACE_LOCK_FILE).exists());
assert_eq!(fs::read(&sidecar).unwrap(), b"reserved");
fs::remove_dir_all(root).unwrap();
}
#[test]
fn rejects_a_workspace_with_a_missing_database() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-missing-database-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
drop(workspace);
fs::remove_file(root.join(DATABASE_FILE)).unwrap();
let error = Workspace::open(&root).unwrap_err();
assert!(error.to_string().contains("workspace database is missing"));
assert!(!root.join(DATABASE_FILE).exists());
fs::remove_dir_all(root).unwrap();
}
#[test]
fn rejects_a_missing_database_with_no_recoverable_wal_frame() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-empty-wal-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
drop(workspace);
fs::remove_file(root.join(DATABASE_FILE)).unwrap();
fs::write(root.join("data.basalt.wal"), []).unwrap();
let error = Workspace::open(&root).unwrap_err();
assert!(error.to_string().contains("no recoverable committed frame"));
fs::remove_dir_all(root).unwrap();
}
#[test]
fn resolves_path_aliases() {
let root = std::env::temp_dir().join(format!(
"basalt-same-path-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
fs::create_dir_all(root.join("workspace")).unwrap();
let direct = root.join("workspace/data.basalt");
let alias = root.join("workspace/../workspace/data.basalt");
fs::write(&direct, b"database").unwrap();
assert!(same_path(&alias, &direct));
fs::remove_dir_all(root).unwrap();
}
#[test]
fn json_values_keep_nested_data_as_text() {
assert!(matches!(
json_cell(&serde_json::json!({"nested": true})),
ImportedCell::Text(value) if value == "{\"nested\":true}"
));
}
#[test]
fn mcp_import_limits_cover_rows_columns_and_cells() {
let limits = ImportLimits::mcp();
let rows = enforce_import_limits(MAX_MCP_IMPORT_ROWS + 1, 1, limits).unwrap_err();
assert!(rows.to_string().contains("limited to 10000 rows"));
let columns = enforce_import_limits(1, MAX_MCP_IMPORT_COLUMNS + 1, limits).unwrap_err();
assert!(columns.to_string().contains("limited to 256 columns"));
let cells = enforce_import_limits(10_000, 101, limits).unwrap_err();
assert!(cells.to_string().contains("limited to 1000000 cells"));
}
#[test]
fn mcp_history_bounds_non_record_directory_entries() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-history-directory-limit-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
let changes = root.join(HISTORY_DIR).join(CHANGES_DIR);
fs::create_dir_all(&changes).unwrap();
fs::write(changes.join("first.tmp"), []).unwrap();
fs::write(changes.join("second.tmp"), []).unwrap();
let error = load_changes_with_limits(&workspace, Some(1), Some(1024)).unwrap_err();
assert!(error.to_string().contains("history directory exceeds"));
drop(workspace);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn mcp_mutation_row_limit_is_cumulative() {
let mut total = 0;
enforce_mutation_row_limit(
&mut total,
&StatementResult::Update {
rows_affected: 6_000,
},
Some(MAX_MCP_MUTATION_ROWS),
)
.unwrap();
let error = enforce_mutation_row_limit(
&mut total,
&StatementResult::Delete {
rows_affected: 4_001,
},
Some(MAX_MCP_MUTATION_ROWS),
)
.unwrap_err();
assert!(error.to_string().contains("limited to 10000 affected rows"));
}
#[test]
fn row_delta_counts_ignore_order_and_preserve_duplicates() {
let before = vec![
vec![Value::Integer(1), Value::Text("Ada".into())],
vec![Value::Integer(2), Value::Text("Grace".into())],
vec![Value::Integer(2), Value::Text("Grace".into())],
];
let after = vec![
vec![Value::Integer(2), Value::Text("Grace".into())],
vec![Value::Integer(1), Value::Text("Augusta".into())],
];
assert_eq!(row_delta_counts(&before, &after), (1, 2));
assert_eq!(row_delta_counts(&before, &before), (0, 0));
}
#[test]
fn mcp_rejects_an_over_limit_mutation_without_persisting_or_committing_it() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-mutation-limit-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
let database = workspace.database().unwrap();
database
.execute_sql("CREATE TABLE events (id INTEGER)")
.unwrap();
let values = (1..=MAX_MCP_MUTATION_ROWS + 1)
.map(|id| format!("({id})"))
.collect::<Vec<_>>()
.join(", ");
database
.execute_sql(&format!("INSERT INTO events VALUES {values}"))
.unwrap();
database.checkpoint().unwrap();
drop(database);
let error = mcp_preview(&workspace, "DELETE FROM events", 1_048_576).unwrap_err();
assert!(error.to_string().contains("limited to 10000 affected rows"));
assert!(!root.join(HISTORY_DIR).exists());
let plan = preview_plan(&workspace, "DELETE FROM events").unwrap();
let error = mcp_apply(&workspace, &plan.plan_id).unwrap_err();
assert!(error.to_string().contains("limited to 10000 affected rows"));
let changes = load_changes(&workspace).unwrap();
assert_eq!(changes.len(), 1);
assert_eq!(changes[0].status, ChangeStatus::Failed);
let database = workspace.database().unwrap();
assert_eq!(
database.row_count("events").unwrap(),
MAX_MCP_MUTATION_ROWS + 1
);
drop(database);
drop(workspace);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn rejects_an_oversized_mcp_preview_before_persisting_its_plan() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-preview-limit-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
let error = mcp_preview(&workspace, "CREATE TABLE users (id INTEGER)", 1).unwrap_err();
assert!(error.to_string().contains("response limit is 1 bytes"));
assert!(!root.join(HISTORY_DIR).exists());
drop(workspace);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn mcp_history_rejects_an_oversized_external_ledger_before_loading_it() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-history-limit-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
ensure_history_dirs(&workspace).unwrap();
let change_id = "a".repeat(64);
let change = ChangeRecord {
format_version: FORMAT_VERSION,
sequence: 1,
change_id: change_id.clone(),
kind: ChangeKind::Apply,
plan_id: None,
target_change_id: None,
base_generation: 0,
base_state: format!("sha256:{}", "0".repeat(64)),
expected_state: None,
snapshot_id: change_id.clone(),
sql: None,
status: ChangeStatus::Failed,
committed_generation: None,
after_state: None,
error: Some("test".to_string()),
import: None,
};
write_new_json(&change_path(&workspace, &change_id), &change).unwrap();
let error = load_changes_with_limits(&workspace, Some(0), None).unwrap_err();
assert!(error.to_string().contains("exceeds the 0-entry limit"));
let error = load_changes_with_limits(&workspace, None, Some(1)).unwrap_err();
assert!(
error
.to_string()
.contains("history metadata exceeds the 1-byte limit")
);
drop(workspace);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn rejects_tampered_plan_statement_metadata() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-plan-integrity-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
let plan = preview_plan(&workspace, "CREATE TABLE users (id INTEGER)").unwrap();
let path = plan_path(&workspace, &plan.plan_id);
let mut document = serde_json::to_value(&plan).unwrap();
document["statements"][0]["kind"] = JsonValue::String("select".to_string());
fs::write(&path, serde_json::to_vec_pretty(&document).unwrap()).unwrap();
let error = load_plan(&workspace, &plan.plan_id).unwrap_err();
assert!(
error
.to_string()
.contains("statement metadata does not match")
);
drop(workspace);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn rejects_duplicate_workspace_change_sequences() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-sequence-integrity-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
let report = mcp_import(&workspace, Some("users"), "csv", "id\n1\n").unwrap();
let changes = load_changes(&workspace).unwrap();
let mut duplicate = changes[0].clone();
duplicate.change_id = "b".repeat(64);
duplicate.snapshot_id = duplicate.change_id.clone();
write_new_json(&change_path(&workspace, &duplicate.change_id), &duplicate).unwrap();
let error = load_changes(&workspace).unwrap_err();
assert!(
error
.to_string()
.contains("duplicate change sequence numbers")
);
assert_eq!(report.summary, "table users (1 rows, 1 columns)");
drop(workspace);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn retries_a_failed_mcp_import_when_the_base_state_is_unchanged() {
let root = std::env::temp_dir().join(format!(
"basalt-workspace-retry-test-{}-{}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
let workspace = Workspace::init(&root).unwrap();
let content = "id,name\n1,Ada\n";
let format = DataFormat::Csv;
let table = "users";
let database = workspace.database().unwrap();
database.checkpoint().unwrap();
let base_state = state_fingerprint(&workspace.database_path()).unwrap();
let base_generation = database.generation();
drop(database);
ensure_history_dirs(&workspace).unwrap();
let change_id = import_change_id(base_state.as_str(), format, table, content.as_bytes());
copy_atomic(
&workspace.database_path(),
&snapshot_path(&workspace, &change_id),
)
.unwrap();
let request_key = import_request_key(format, table, content.as_bytes());
let summary =
validate_import(format, table, content.as_bytes(), ImportLimits::mcp()).unwrap();
let failed = ChangeRecord {
format_version: FORMAT_VERSION,
sequence: 1,
change_id: change_id.clone(),
kind: ChangeKind::Apply,
plan_id: None,
target_change_id: None,
base_generation,
base_state: base_state.clone(),
expected_state: None,
snapshot_id: change_id.clone(),
sql: None,
status: ChangeStatus::Failed,
committed_generation: None,
after_state: None,
error: Some("transient test failure".to_string()),
import: Some(ImportMetadata {
request_key,
format: format.name().to_string(),
table: Some(table.to_string()),
bytes: content.len(),
summary,
}),
};
write_new_json(&change_path(&workspace, &change_id), &failed).unwrap();
let report = mcp_import(&workspace, Some(table), format.name(), content).unwrap();
assert_eq!(report.change_id, change_id);
assert_eq!(report.summary, "table users (1 rows, 2 columns)");
let changes = load_changes(&workspace).unwrap();
assert_eq!(changes[0].status, ChangeStatus::Committed);
assert_eq!(changes[0].sequence, 1);
let database = workspace.database().unwrap();
let rows = database.execute_sql("SELECT id, name FROM users").unwrap();
assert!(matches!(
&rows[0],
StatementResult::Select { rows, .. } if rows.len() == 1
));
drop(database);
drop(workspace);
fs::remove_dir_all(root).unwrap();
}
}