use std::collections::BTreeMap;
use std::fs;
use std::path::Path;
use rusqlite::{Connection, Row};
use crate::adapter::git::GitCli;
use crate::adapter::sqlite::ProjectDatabase;
use crate::adapter::sqlite_repos::SqliteEventRepository;
use crate::adapter::xdg::XdgPaths;
use crate::application::interchange::read_bundle;
use crate::domain::pack::{self, PackManifest, PackSource};
use crate::error::CarryCtxError;
use crate::repository::event::{EventRepository, NewEvent};
fn hostname() -> String {
std::env::var("HOSTNAME").unwrap_or_else(|_| "unknown".into())
}
fn db_err(context: &str, error: impl std::fmt::Display) -> CarryCtxError {
CarryCtxError::database_error(format!("{context}: {error}"))
}
fn require_dir_format(pack_format: &str) -> Result<(), CarryCtxError> {
if pack_format == "dir" {
Ok(())
} else {
Err(CarryCtxError::unsupported_operation(format!(
"Pack format '{pack_format}' is not supported in v1; only '--pack-format dir' is available. Single-file transport stays external: 'tar -cf - <dir> | ...'."
)))
}
}
fn reject_file_target(out_dir: &Path) -> Result<(), CarryCtxError> {
if out_dir.exists() && !out_dir.is_dir() {
return Err(CarryCtxError::invalid_arguments(format!(
"Export target '{}' exists and is not a directory.",
out_dir.display()
)));
}
Ok(())
}
fn display_path(out_dir: &Path) -> Result<String, CarryCtxError> {
if out_dir.is_absolute() {
return Ok(out_dir.to_string_lossy().into_owned());
}
let cwd = std::env::current_dir().map_err(|e| {
CarryCtxError::io_error(format!("Failed to resolve working directory: {e}"))
})?;
Ok(cwd.join(out_dir).to_string_lossy().into_owned())
}
fn sql_value_to_json(value: rusqlite::types::Value) -> serde_json::Value {
match value {
rusqlite::types::Value::Null => serde_json::Value::Null,
rusqlite::types::Value::Integer(i) => serde_json::json!(i),
rusqlite::types::Value::Real(f) => serde_json::Number::from_f64(f)
.map(serde_json::Value::Number)
.unwrap_or(serde_json::Value::Null),
rusqlite::types::Value::Text(s) => serde_json::Value::String(s),
rusqlite::types::Value::Blob(bytes) => serde_json::Value::String(hex::encode(bytes)),
}
}
fn row_to_json(names: &[String], row: &Row) -> rusqlite::Result<serde_json::Value> {
let mut map = serde_json::Map::with_capacity(names.len());
for (index, name) in names.iter().enumerate() {
let value: rusqlite::types::Value = row.get(index)?;
map.insert(name.clone(), sql_value_to_json(value));
}
Ok(serde_json::Value::Object(map))
}
fn dump_table(conn: &Connection, table: &str) -> Result<Vec<serde_json::Value>, CarryCtxError> {
if table != "projects" && !pack::PACK_TABLE_FILES.contains(&table) {
return Err(CarryCtxError::database_error(format!(
"Refusing to dump unknown table '{table}'."
)));
}
let mut stmt = conn
.prepare(&format!("SELECT * FROM \"{table}\" ORDER BY rowid"))
.map_err(|e| db_err(&format!("Failed to dump table '{table}'"), e))?;
let names: Vec<String> = stmt
.column_names()
.iter()
.map(|name| (*name).to_string())
.collect();
stmt.query_map([], |row| row_to_json(&names, row))
.map_err(|e| db_err(&format!("Failed to dump table '{table}'"), e))?
.collect::<Result<Vec<_>, _>>()
.map_err(|e| db_err(&format!("Failed to read row of table '{table}'"), e))
}
fn count_table(conn: &Connection, table: &str) -> Result<u64, CarryCtxError> {
debug_assert!(pack::PACK_TABLE_FILES.contains(&table));
let count: i64 = conn
.query_row(&format!("SELECT COUNT(*) FROM \"{table}\""), [], |row| {
row.get(0)
})
.map_err(|e| db_err(&format!("Failed to count table '{table}'"), e))?;
Ok(count.max(0) as u64)
}
fn counts_of(tables: &BTreeMap<String, Vec<serde_json::Value>>) -> BTreeMap<String, u64> {
let mut counts = BTreeMap::new();
for table in pack::PACK_TABLE_FILES {
let len = tables
.get(*table)
.map(|rows| rows.len() as u64)
.unwrap_or(0);
counts.insert((*table).to_string(), len);
}
counts
}
#[derive(Debug)]
struct Snapshot {
project_id: String,
project: serde_json::Value,
tables: BTreeMap<String, Vec<serde_json::Value>>,
sequences: BTreeMap<String, u64>,
schema_version: u32,
}
fn collect_snapshot(conn: &Connection) -> Result<Snapshot, CarryCtxError> {
let project_rows = dump_table(conn, "projects")?;
if project_rows.is_empty() {
return Err(CarryCtxError::resource_not_found(
"No CarryCtx project is initialized here; run `carryctx init` first.",
));
}
if project_rows.len() != 1 {
return Err(CarryCtxError::database_error(
"Project table must contain exactly one row.",
));
}
let project = project_rows.into_iter().next().expect("checked");
let project_id = project
.get("id")
.and_then(|v| v.as_str())
.filter(|id| !id.trim().is_empty())
.ok_or_else(|| CarryCtxError::database_error("Project row has no id."))?
.to_string();
let mut tables = BTreeMap::new();
for table in pack::PACK_TABLE_FILES {
tables.insert((*table).to_string(), dump_table(conn, table)?);
}
let mut sequences = BTreeMap::new();
{
let mut stmt = conn
.prepare("SELECT kind, next_value FROM sequences WHERE project_id = ?1")
.map_err(|e| db_err("Failed to read sequences", e))?;
let rows = stmt
.query_map([&project_id], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
})
.map_err(|e| db_err("Failed to read sequences", e))?;
for row in rows {
let (kind, next) = row.map_err(|e| db_err("Failed to read sequence row", e))?;
sequences.insert(kind, next.max(0) as u64);
}
}
let applied: i64 = conn
.query_row(
"SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
[],
|row| row.get(0),
)
.map_err(|e| db_err("Failed to read schema version", e))?;
let schema_version = if applied > 0 {
applied as u32
} else {
project
.get("schema_version")
.and_then(serde_json::Value::as_u64)
.unwrap_or(0) as u32
};
Ok(Snapshot {
project_id,
project,
tables,
sequences,
schema_version,
})
}
fn build_manifest(
snapshot: &Snapshot,
counts: BTreeMap<String, u64>,
export_id: String,
created_at: String,
git_branch: Option<String>,
git_commit: Option<String>,
) -> PackManifest {
let mut manifest = PackManifest::new(
env!("CARGO_PKG_VERSION"),
snapshot.schema_version,
snapshot.project_id.clone(),
export_id,
created_at,
PackSource {
git_branch,
git_commit,
hostname: Some(hostname()),
},
counts,
);
manifest.sequences = snapshot.sequences.clone();
manifest
}
fn write_bundle(
out_dir: &Path,
manifest: &PackManifest,
project: &serde_json::Value,
tables: &BTreeMap<String, Vec<serde_json::Value>>,
) -> Result<(), CarryCtxError> {
fs::create_dir_all(out_dir).map_err(|e| {
CarryCtxError::io_error(format!(
"Failed to create export directory '{}': {e}",
out_dir.display()
))
})?;
let write = |name: &str, content: String| {
fs::write(out_dir.join(name), content).map_err(|e| {
CarryCtxError::io_error(format!("Failed to write pack file '{name}': {e}"))
})
};
write(
pack::PACK_MANIFEST_FILE,
serde_json::to_string_pretty(manifest)
.map_err(|e| CarryCtxError::io_error(format!("Failed to encode manifest: {e}")))?,
)?;
write(
pack::PACK_PROJECT_FILE,
serde_json::to_string_pretty(project)
.map_err(|e| CarryCtxError::io_error(format!("Failed to encode project row: {e}")))?,
)?;
for table in pack::PACK_TABLE_FILES {
let mut text = String::new();
if let Some(rows) = tables.get(*table) {
for row in rows {
text.push_str(&serde_json::to_string(row).map_err(|e| {
CarryCtxError::io_error(format!("Failed to encode '{table}.jsonl' row: {e}"))
})?);
text.push('\n');
}
}
write(&format!("{table}.jsonl"), text)?;
}
Ok(())
}
pub fn plan_export(
project_path: &Path,
pack_format: &str,
out_dir: &Path,
) -> Result<serde_json::Value, CarryCtxError> {
require_dir_format(pack_format)?;
reject_file_target(out_dir)?;
let git = GitCli::new();
let gp = git.discover(project_path)?;
let xdg = XdgPaths::new();
let db_path = xdg.project_db(&gp.git_common_dir);
if !db_path.exists() {
return Err(CarryCtxError::resource_not_found(
"No CarryCtx project database found; run `carryctx init` first.",
));
}
let database = ProjectDatabase::open_readonly(&db_path)?;
let snapshot = collect_snapshot(database.connection())?;
let counts = counts_of(&snapshot.tables);
let manifest = build_manifest(
&snapshot,
counts.clone(),
ulid::Ulid::generate().to_string(),
chrono::Utc::now().to_rfc3339(),
gp.branch.clone(),
gp.head.clone(),
);
pack::check_counts(&manifest, &counts)?;
Ok(serde_json::json!({
"manifest": manifest,
"counts": counts,
"path": display_path(out_dir)?,
"operation": {"applied": false},
}))
}
pub fn run_export(
project_path: &Path,
pack_format: &str,
out_dir: &Path,
actor_agent_id: Option<String>,
session_id: Option<String>,
) -> Result<serde_json::Value, CarryCtxError> {
require_dir_format(pack_format)?;
reject_file_target(out_dir)?;
let git = GitCli::new();
let gp = git.discover(project_path)?;
let xdg = XdgPaths::new();
let db_path = xdg.project_db(&gp.git_common_dir);
if !db_path.exists() {
return Err(CarryCtxError::resource_not_found(
"No CarryCtx project database found; run `carryctx init` first.",
));
}
let mut database = ProjectDatabase::open(&db_path)?;
database.migrate()?;
let path = display_path(out_dir)?;
let uow = database.begin_unit_of_work()?;
let conn = uow.connection();
let result = (|| {
let project_id: String = conn
.query_row("SELECT id FROM projects LIMIT 1", [], |row| row.get(0))
.map_err(|e| {
if e == rusqlite::Error::QueryReturnedNoRows {
CarryCtxError::resource_not_found(
"No CarryCtx project is initialized here; run `carryctx init` first.",
)
} else {
db_err("Failed to resolve project id", e)
}
})?;
let export_id = ulid::Ulid::generate().to_string();
let created_at = chrono::Utc::now().to_rfc3339();
let mut payload_counts = BTreeMap::new();
for table in pack::PACK_TABLE_FILES {
payload_counts.insert((*table).to_string(), count_table(conn, table)?);
}
*payload_counts
.get_mut("events")
.expect("events is a pack table") += 1;
SqliteEventRepository::new(conn).append(&NewEvent {
id: ulid::Ulid::generate().to_string(),
project_id,
event_type: "project.exported".into(),
actor_agent_id,
session_id,
task_id: None,
payload: serde_json::json!({
"exportId": export_id,
"format": pack::PACK_FORMAT,
"formatVersion": pack::PACK_FORMAT_VERSION,
"counts": payload_counts,
"path": path,
}),
occurred_at: created_at.clone(),
})?;
let snapshot = collect_snapshot(conn)?;
let counts = counts_of(&snapshot.tables);
if counts != payload_counts {
return Err(CarryCtxError::validation_error(
"Export count skew between audit payload and dump; retry the export.",
));
}
let manifest = build_manifest(
&snapshot,
counts.clone(),
export_id,
created_at,
gp.branch.clone(),
gp.head.clone(),
);
pack::check_counts(&manifest, &counts)?;
write_bundle(out_dir, &manifest, &snapshot.project, &snapshot.tables)?;
let bundle = read_bundle(out_dir)?;
pack::check_counts(&bundle.manifest, &bundle.actual_counts())?;
if bundle.manifest != manifest {
return Err(CarryCtxError::validation_error(
"Exported manifest does not round-trip; retry the export.",
));
}
Ok::<(PackManifest, BTreeMap<String, u64>), CarryCtxError>((manifest, counts))
})();
let (manifest, counts) = result?;
uow.commit()?;
Ok(serde_json::json!({
"manifest": manifest,
"counts": counts,
"path": path,
}))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn dir_format_accepted_others_refused_as_unsupported() {
assert!(require_dir_format("dir").is_ok());
for bad in ["tar", "DIR", "", "zip"] {
let error = require_dir_format(bad).unwrap_err();
assert_eq!(error.code, "UNSUPPORTED_OPERATION", "format '{bad}'");
assert_eq!(format!("{}", error.exit_code as i32), "10");
}
}
#[test]
fn file_target_refused_missing_path_allowed() {
let root = tempfile::tempdir().unwrap();
let file = root.path().join("file.sqlite");
fs::write(&file, b"x").unwrap();
let error = reject_file_target(&file).unwrap_err();
assert_eq!(error.code, "INVALID_ARGUMENTS");
assert!(reject_file_target(&root.path().join("absent")).is_ok());
assert!(reject_file_target(root.path()).is_ok());
}
#[test]
fn sql_values_convert_losslessly() {
use rusqlite::types::Value as Sql;
assert_eq!(sql_value_to_json(Sql::Null), serde_json::Value::Null);
assert_eq!(sql_value_to_json(Sql::Integer(-7)), serde_json::json!(-7));
assert_eq!(sql_value_to_json(Sql::Real(1.5)), serde_json::json!(1.5));
assert_eq!(
sql_value_to_json(Sql::Real(f64::INFINITY)),
serde_json::Value::Null
);
assert_eq!(
sql_value_to_json(Sql::Text("hi".into())),
serde_json::json!("hi")
);
assert_eq!(
sql_value_to_json(Sql::Blob(vec![0xab, 0xcd])),
serde_json::json!("abcd")
);
}
#[test]
fn dump_refuses_tables_outside_the_pack_whitelist() {
let conn = Connection::open_in_memory().unwrap();
let error = dump_table(&conn, "sqlite_master").unwrap_err();
assert_eq!(error.code, "DATABASE_ERROR");
let error = dump_table(&conn, "operations").unwrap_err();
assert_eq!(error.code, "DATABASE_ERROR");
}
#[test]
fn snapshot_without_project_row_is_not_found() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch("CREATE TABLE projects (id TEXT PRIMARY KEY, name TEXT);")
.unwrap();
let error = collect_snapshot(&conn).unwrap_err();
assert_eq!(error.code, "RESOURCE_NOT_FOUND");
}
}