1use cratefield_adapter_sqlite::SqliteDatabase;
22use cratefield_core::{Database, Harness, Row, Statement};
23use serde::{Deserialize, Serialize};
24use serde_json::{Map, Value as Json};
25use std::path::Path;
26
27#[cfg(feature = "postgres")]
30const IMPORT_BATCH_ROWS: usize = 200;
31
32#[derive(Debug, Serialize, Deserialize)]
36struct TableManifest {
37 table: String,
38 columns: Vec<String>,
39 rows: u64,
40 sha256: String,
41}
42
43#[derive(Debug, Serialize, Deserialize)]
44struct Manifest {
45 tables: Vec<TableManifest>,
46 #[serde(default, skip_serializing_if = "Vec::is_empty")]
51 omitted_sidecars: Vec<String>,
52}
53
54#[derive(Debug, Serialize, Deserialize)]
55struct Record {
56 table: String,
57 row: Map<String, Json>,
58}
59
60struct ExportFile {
63 manifest: Manifest,
64 #[cfg_attr(not(feature = "postgres"), allow(dead_code))]
65 records: Vec<(String, Vec<u8>)>,
66}
67
68fn parse_export_file(bytes: &[u8], path: &Path) -> Result<ExportFile, String> {
69 let mut lines = bytes.split(|b| *b == b'\n');
70 let first = lines.next().unwrap_or_default();
71 let manifest: Manifest = serde_json::from_slice(first).map_err(|err| {
72 format!(
73 "{}: the first line must be the manifest: {err}",
74 path.display()
75 )
76 })?;
77 if manifest.tables.is_empty() {
78 return Err(format!("{}: the manifest lists no tables", path.display()));
79 }
80 let mut records = Vec::new();
81 for raw in lines {
82 if raw.is_empty() {
83 continue;
84 }
85 let record: Record = serde_json::from_slice(raw).map_err(|err| {
86 format!(
87 "{}: line {} is not a data record: {err}",
88 path.display(),
89 records.len() + 2
90 )
91 })?;
92 records.push((record.table, raw.to_vec()));
93 }
94 Ok(ExportFile { manifest, records })
95}
96
97fn sha256_hex(data: &[u8]) -> String {
98 use sha2::{Digest, Sha256};
99 let digest = Sha256::digest(data);
100 let mut hex = String::with_capacity(digest.len() * 2);
101 for byte in digest {
102 use std::fmt::Write as _;
103 let _ = write!(hex, "{byte:02x}");
104 }
105 hex
106}
107
108fn validate_identifier(kind: &str, name: &str) -> Result<(), String> {
112 let mut chars = name.chars();
113 let valid = matches!(chars.next(), Some('a'..='z' | 'A'..='Z' | '_'))
114 && chars.all(|c| matches!(c, 'a'..='z' | 'A'..='Z' | '0'..='9' | '_'));
115 if valid {
116 Ok(())
117 } else {
118 Err(format!(
119 "invalid {kind} {name:?}: only [A-Za-z_][A-Za-z0-9_]* is allowed"
120 ))
121 }
122}
123
124fn harness_tables(harness: &Harness) -> Vec<String> {
129 harness
130 .modules()
131 .iter()
132 .flat_map(|module| module.tables().iter().map(|t| (*t).to_owned()))
133 .collect()
134}
135
136pub fn export(
149 harness: &Harness,
150 db_path: &Path,
151 out_path: &Path,
152 plan: bool,
153 without_sidecar_tables: bool,
154 sidecars: Option<&str>,
155) -> Result<(), String> {
156 let omitted_sidecars = match crate::sidecars::parse(sidecars) {
161 Ok(mounts) if !mounts.is_empty() => {
162 if !without_sidecar_tables {
163 return Err(format!(
164 "this venture mounts {} as sidecars. Their tables are in this database but \
165 `fz` cannot see them: they are declared in another repository. Export those \
166 tables from there, then re-run with --without-sidecar-tables to record the \
167 omission in the manifest. See docs/MOUNTING.md",
168 crate::sidecars::listed(&mounts)
169 ));
170 }
171 mounts.iter().map(|m| m.name.clone()).collect()
172 }
173 Ok(_) => Vec::new(),
174 Err(errors) => return Err(errors.join("; ")),
175 };
176 let tables = harness_tables(harness);
177 if tables.is_empty() {
178 return Err("the harness's modules declare no tables; nothing to export".to_owned());
179 }
180 for table in &tables {
181 validate_identifier("table", table)?;
182 }
183 let db = SqliteDatabase::open(&db_path.to_string_lossy())
184 .map_err(|err| format!("cannot open {}: {err}", db_path.display()))?;
185
186 let mut manifest = Manifest {
187 tables: Vec::new(),
188 omitted_sidecars,
189 };
190 let mut body = String::new();
191 for table in &tables {
192 let rows =
193 pollster::block_on(db.query(&Statement::new(format!("SELECT * FROM \"{table}\""))))
194 .map_err(|err| format!("cannot read {table}: {err} (is the database migrated?)"))?;
195 let columns: Vec<String> = rows
196 .first()
197 .map(|first| first.column_names().map(str::to_owned).collect())
198 .unwrap_or_default();
199 for column in &columns {
200 validate_identifier("column", column)?;
201 }
202 if plan {
203 println!(" {table}: {} rows, {} columns", rows.len(), columns.len());
204 manifest.tables.push(TableManifest {
205 table: table.clone(),
206 columns,
207 rows: rows.len() as u64,
208 sha256: String::new(),
209 });
210 continue;
211 }
212 let mut table_bytes = Vec::new();
213 for row in &rows.rows {
214 let record = record_line(table, row)?;
215 table_bytes.extend_from_slice(record.as_bytes());
216 table_bytes.push(b'\n');
217 body.push_str(&record);
218 body.push('\n');
219 }
220 manifest.tables.push(TableManifest {
221 table: table.clone(),
222 columns,
223 rows: rows.len() as u64,
224 sha256: sha256_hex(&table_bytes),
225 });
226 }
227
228 if plan {
229 println!(
230 "plan: export {} ({} tables) -> {}",
231 db_path.display(),
232 manifest.tables.len(),
233 out_path.display()
234 );
235 return Ok(());
236 }
237 let manifest_line = serde_json::to_string(&manifest)
238 .map_err(|err| format!("cannot serialize the manifest: {err}"))?;
239 std::fs::write(out_path, format!("{manifest_line}\n{body}"))
240 .map_err(|err| format!("cannot write {}: {err}", out_path.display()))?;
241 let total: u64 = manifest.tables.iter().map(|t| t.rows).sum();
242 println!(
243 "exported {total} rows across {} tables to {}",
244 manifest.tables.len(),
245 out_path.display()
246 );
247 Ok(())
248}
249
250fn record_line(table: &str, row: &Row) -> Result<String, String> {
251 let mut names: Vec<&str> = row.column_names().collect();
255 names.sort_unstable();
256 let mut object = Map::new();
257 for name in names {
258 let value = row
259 .get::<sea_query::Value>(name)
260 .unwrap_or(sea_query::Value::String(None));
261 let json = sea_value_to_json(&value)
262 .map_err(|problem| format!("table {table}, column {name}: {problem}"))?;
263 object.insert(name.to_owned(), json);
264 }
265 let record = Record {
266 table: table.to_owned(),
267 row: object,
268 };
269 serde_json::to_string(&record).map_err(|err| format!("cannot serialize a row: {err}"))
270}
271
272fn sea_value_to_json(value: &sea_query::Value) -> Result<Json, String> {
273 use sea_query::Value as Sea;
274 Ok(match value {
275 Sea::Bool(None)
276 | Sea::TinyInt(None)
277 | Sea::SmallInt(None)
278 | Sea::Int(None)
279 | Sea::BigInt(None)
280 | Sea::TinyUnsigned(None)
281 | Sea::SmallUnsigned(None)
282 | Sea::Unsigned(None)
283 | Sea::BigUnsigned(None)
284 | Sea::Float(None)
285 | Sea::Double(None)
286 | Sea::String(None)
287 | Sea::Char(None)
288 | Sea::Bytes(None) => Json::Null,
289 Sea::Bool(Some(v)) => Json::Bool(*v),
290 Sea::TinyInt(Some(v)) => i64::from(*v).into(),
291 Sea::SmallInt(Some(v)) => i64::from(*v).into(),
292 Sea::Int(Some(v)) => i64::from(*v).into(),
293 Sea::BigInt(Some(v)) => (*v).into(),
294 Sea::TinyUnsigned(Some(v)) => u64::from(*v).into(),
295 Sea::SmallUnsigned(Some(v)) => u64::from(*v).into(),
296 Sea::Unsigned(Some(v)) => (*v).into(),
297 Sea::BigUnsigned(Some(v)) => (*v).into(),
298 Sea::Float(Some(v)) => Json::from(f64::from(*v)),
299 Sea::Double(Some(v)) => Json::from(*v),
300 Sea::String(Some(v)) => Json::String((**v).clone()),
301 Sea::Char(Some(v)) => Json::String(v.to_string()),
302 Sea::Bytes(Some(_)) => {
305 return Err(
306 "BLOB data is outside the portable subset and cannot be exported".to_owned(),
307 );
308 }
309 })
310}
311
312pub fn import(
326 harness: &Harness,
327 file_path: &Path,
328 url: &str,
329 append: bool,
330 plan: bool,
331) -> Result<(), String> {
332 let bytes = std::fs::read(file_path)
333 .map_err(|err| format!("cannot read {}: {err}", file_path.display()))?;
334 let file = parse_export_file(&bytes, file_path)?;
335 validate_manifest(harness, &file.manifest)?;
336
337 #[cfg(not(feature = "postgres"))]
338 {
339 let _ = (url, append, plan);
340 Err(
341 "this fz binary was built without cratefield-cli's `postgres` feature — \
342 rebuild the fz bin with `--features cratefield-cli/postgres` to import \
343 into Postgres"
344 .to_owned(),
345 )
346 }
347 #[cfg(feature = "postgres")]
348 import_postgres(file_path, &file, url, append, plan)
349}
350
351fn validate_manifest(harness: &Harness, manifest: &Manifest) -> Result<(), String> {
355 let expected = harness_tables(harness);
356 let mut seen = std::collections::BTreeSet::new();
357 for table in &manifest.tables {
358 validate_identifier("table", &table.table)?;
359 if !seen.insert(table.table.clone()) {
360 return Err(format!(
361 "table {} appears twice in the manifest",
362 table.table
363 ));
364 }
365 for column in &table.columns {
366 validate_identifier("column", column)?;
367 }
368 }
369 for table in &manifest.tables {
370 if !expected.contains(&table.table) {
371 return Err(format!(
372 "manifest table {} is not a table of this venture's modules (expected: {})",
373 table.table,
374 expected.join(", ")
375 ));
376 }
377 }
378 Ok(())
379}
380
381#[cfg(feature = "postgres")]
382fn import_postgres(
383 file_path: &Path,
384 file: &ExportFile,
385 url: &str,
386 append: bool,
387 plan: bool,
388) -> Result<(), String> {
389 let runtime = tokio::runtime::Builder::new_current_thread()
390 .enable_all()
391 .build()
392 .map_err(|err| format!("cannot start the async runtime: {err}"))?;
393 runtime.block_on(async {
394 let db = cratefield_adapter_postgres::Postgres::connect(url)
395 .await
396 .map_err(|err| format!("cannot connect (check --url): {err}"))?;
397
398 let schemas = load_target_schemas(&db, &file.manifest).await?;
399 check_column_drift(&file.manifest, &schemas)?;
400 let existing = preflight_counts(&db, &file.manifest, append, plan).await?;
401
402 if plan {
403 println!(
404 "plan: import {} ({} tables, {} rows) -> postgres{}",
405 file_path.display(),
406 file.manifest.tables.len(),
407 file.manifest.tables.iter().map(|t| t.rows).sum::<u64>(),
408 if append { " with --append" } else { "" }
409 );
410 for (entry, had) in file.manifest.tables.iter().zip(&existing) {
411 println!(
412 " {}: incoming {} rows, target has {had} rows{}",
413 entry.table,
414 entry.rows,
415 if *had > 0 && !append {
416 " — REFUSES without --append"
417 } else {
418 ""
419 }
420 );
421 }
422 return Ok(());
423 }
424
425 let slices = table_slices(&file.manifest, file)?;
426 verify_hashes(&file.manifest, &slices)?;
427 import_tables(&db, &file.manifest, &schemas, &slices).await?;
428 verify_counts(&db, &file.manifest, &existing).await?;
429 println!("import verified against the manifest");
430 Ok(())
431 })
432}
433
434#[cfg(feature = "postgres")]
438async fn load_target_schemas(
439 db: &cratefield_adapter_postgres::Postgres,
440 manifest: &Manifest,
441) -> Result<Vec<Vec<(String, String)>>, String> {
442 let mut schemas = Vec::with_capacity(manifest.tables.len());
443 for entry in &manifest.tables {
444 let rows = db
445 .query(&Statement::with_values(
446 "SELECT column_name, data_type FROM information_schema.columns \
447 WHERE table_name = ? ORDER BY ordinal_position",
448 vec![entry.table.clone().into()],
449 ))
450 .await
451 .map_err(|err| format!("cannot read the schema of {}: {err}", entry.table))?;
452 let columns: Vec<(String, String)> = rows
453 .rows
454 .iter()
455 .map(|row| {
456 (
457 row.get::<String>("column_name").unwrap_or_default(),
458 row.get::<String>("data_type").unwrap_or_default(),
459 )
460 })
461 .collect();
462 if columns.is_empty() {
463 return Err(format!(
464 "table {} does not exist in the target — run `fz migrations apply \
465 --dialect postgres` first",
466 entry.table
467 ));
468 }
469 schemas.push(columns);
470 }
471 Ok(schemas)
472}
473
474#[cfg(feature = "postgres")]
477fn check_column_drift(
478 manifest: &Manifest,
479 schemas: &[Vec<(String, String)>],
480) -> Result<(), String> {
481 for (entry, columns) in manifest.tables.iter().zip(schemas) {
482 if entry.columns.is_empty() {
483 continue;
484 }
485 let target: Vec<&str> = columns.iter().map(|(name, _)| name.as_str()).collect();
486 let mut exported = entry.columns.clone();
487 exported.sort_unstable();
488 let mut target_sorted = target.clone();
489 target_sorted.sort_unstable();
490 if exported != target_sorted {
491 return Err(format!(
492 "table {}: the export has columns [{}] but the target has [{}]",
493 entry.table,
494 entry.columns.join(", "),
495 target.join(", ")
496 ));
497 }
498 }
499 Ok(())
500}
501
502#[cfg(feature = "postgres")]
505async fn preflight_counts(
506 db: &cratefield_adapter_postgres::Postgres,
507 manifest: &Manifest,
508 append: bool,
509 plan: bool,
510) -> Result<Vec<u64>, String> {
511 let mut existing = Vec::with_capacity(manifest.tables.len());
512 for entry in &manifest.tables {
513 let count = row_count(db, &entry.table).await?;
514 if count > 0 && !append && !plan {
515 return Err(format!(
516 "table {} already has {count} rows — pass --append to add to it",
517 entry.table
518 ));
519 }
520 existing.push(count);
521 }
522 Ok(existing)
523}
524
525#[cfg(feature = "postgres")]
527type RecordGroup = [(String, Vec<u8>)];
528
529#[cfg(feature = "postgres")]
533fn table_slices<'a>(
534 manifest: &Manifest,
535 file: &'a ExportFile,
536) -> Result<Vec<&'a RecordGroup>, String> {
537 let mut slices = Vec::with_capacity(manifest.tables.len());
538 let mut cursor = 0usize;
539 for entry in &manifest.tables {
540 let expected_rows = usize::try_from(entry.rows).unwrap_or(usize::MAX);
541 let end = (cursor + expected_rows).min(file.records.len());
542 let slice = &file.records[cursor..end];
543 for (table, _) in slice {
544 if table != &entry.table {
545 return Err(format!(
546 "records for {} are out of lock order (found {}) — the file was \
547 edited after export",
548 entry.table, table
549 ));
550 }
551 }
552 if slice.len() != expected_rows {
553 return Err(format!(
554 "table {}: the manifest says {} rows but the file holds {}",
555 entry.table,
556 entry.rows,
557 slice.len()
558 ));
559 }
560 slices.push(slice);
561 cursor += expected_rows;
562 }
563 Ok(slices)
564}
565
566#[cfg(feature = "postgres")]
570fn verify_hashes(manifest: &Manifest, slices: &[&RecordGroup]) -> Result<(), String> {
571 for (entry, slice) in manifest.tables.iter().zip(slices) {
572 let mut hashed = Vec::new();
573 for (_, line) in *slice {
574 hashed.extend_from_slice(line);
575 hashed.push(b'\n');
576 }
577 let digest = sha256_hex(&hashed);
578 if digest != entry.sha256 {
579 return Err(format!(
580 "table {}: sha256 mismatch (manifest {}, file {}) — the file was \
581 edited or truncated after export",
582 entry.table, entry.sha256, digest
583 ));
584 }
585 }
586 Ok(())
587}
588
589#[cfg(feature = "postgres")]
593async fn import_tables(
594 db: &cratefield_adapter_postgres::Postgres,
595 manifest: &Manifest,
596 schemas: &[Vec<(String, String)>],
597 slices: &[&RecordGroup],
598) -> Result<(), String> {
599 for ((entry, columns), slice) in manifest.tables.iter().zip(schemas).zip(slices) {
600 if entry.rows > 0 {
601 insert_table(db, entry, columns, slice).await?;
602 }
603 }
604 Ok(())
605}
606
607#[cfg(feature = "postgres")]
610async fn verify_counts(
611 db: &cratefield_adapter_postgres::Postgres,
612 manifest: &Manifest,
613 existing: &[u64],
614) -> Result<(), String> {
615 for (entry, had) in manifest.tables.iter().zip(existing) {
616 let now = row_count(db, &entry.table).await?;
617 let expected = had + entry.rows;
618 if now != expected {
619 return Err(format!(
620 "table {}: expected {expected} rows after import, found {now}",
621 entry.table
622 ));
623 }
624 println!(
625 " {}: {} rows imported, {now} total",
626 entry.table, entry.rows
627 );
628 }
629 Ok(())
630}
631
632#[cfg(feature = "postgres")]
633async fn row_count(db: &cratefield_adapter_postgres::Postgres, table: &str) -> Result<u64, String> {
634 let rows = db
635 .query(&Statement::new(format!(
636 "SELECT COUNT(*) AS n FROM \"{table}\""
637 )))
638 .await
639 .map_err(|err| format!("cannot count {table}: {err}"))?;
640 let count = rows
641 .first()
642 .and_then(|row| row.get::<i64>("n"))
643 .unwrap_or_default();
644 Ok(u64::try_from(count).unwrap_or_default())
645}
646
647#[cfg(feature = "postgres")]
648async fn insert_table(
649 db: &cratefield_adapter_postgres::Postgres,
650 entry: &TableManifest,
651 columns: &[(String, String)],
652 records: &RecordGroup,
653) -> Result<(), String> {
654 let column_list = entry
655 .columns
656 .iter()
657 .map(|c| format!("\"{c}\""))
658 .collect::<Vec<_>>()
659 .join(", ");
660 let placeholders = format!("({})", vec!["?"; entry.columns.len()].join(", "));
661
662 let mut statements = Vec::new();
663 for chunk in records.chunks(IMPORT_BATCH_ROWS) {
664 let mut sql = format!("INSERT INTO \"{}\" ({column_list}) VALUES ", entry.table);
665 let mut values = Vec::with_capacity(chunk.len() * entry.columns.len());
666 for (index, (_, line)) in chunk.iter().enumerate() {
667 let record: Record = serde_json::from_slice(line)
668 .map_err(|err| format!("table {}: row {}: {err}", entry.table, index))?;
669 if index > 0 {
670 sql.push_str(", ");
671 }
672 sql.push_str(&placeholders);
673 for column in &entry.columns {
674 let value = record.row.get(column).unwrap_or(&Json::Null);
675 let Some(pg_type) = columns
676 .iter()
677 .find(|(name, _)| name == column)
678 .map(|(_, kind)| kind.as_str())
679 else {
680 return Err(format!(
681 "table {}: column {column} vanished from the target schema",
682 entry.table
683 ));
684 };
685 values.push(bind_json_for_type(
686 &entry.table,
687 column,
688 pg_type,
689 value,
690 index,
691 )?);
692 }
693 }
694 statements.push(Statement::with_values(sql, values));
695 }
696 db.batch_atomic(&statements)
697 .await
698 .map_err(|err| format!("table {}: {err}", entry.table))
699}
700
701#[cfg(feature = "postgres")]
705fn bind_json_for_type(
706 table: &str,
707 column: &str,
708 pg_type: &str,
709 value: &Json,
710 row_index: usize,
711) -> Result<sea_query::Value, String> {
712 use sea_query::Value as Sea;
713 let where_ = |problem: String| format!("{table}.{column}, row {row_index}: {problem}");
714 let kind = match value {
715 Json::Null => "null",
716 Json::Bool(_) => "a boolean",
717 Json::Number(_) => "a number",
718 Json::String(_) => "a string",
719 Json::Array(_) => "an array",
720 Json::Object(_) => "an object",
721 };
722 let int_of = || {
723 value
724 .as_i64()
725 .ok_or_else(|| where_(format!("expected an integer, found {kind}")))
726 };
727 Ok(match pg_type {
728 "smallint" => match value {
729 Json::Null => Sea::SmallInt(None),
730 _ => Sea::SmallInt(Some(i16::try_from(int_of()?).map_err(|_| {
731 where_("the value does not fit the SMALLINT column".to_owned())
732 })?)),
733 },
734 "integer" => match value {
735 Json::Null => Sea::Int(None),
736 _ => Sea::Int(Some(i32::try_from(int_of()?).map_err(|_| {
737 where_("the value does not fit the INTEGER column".to_owned())
738 })?)),
739 },
740 "bigint" => match value {
741 Json::Null => Sea::BigInt(None),
742 _ => Sea::BigInt(Some(int_of()?)),
743 },
744 "real" | "double precision" => match value {
745 Json::Null => Sea::Double(None),
746 _ => {
747 Sea::Double(Some(value.as_f64().ok_or_else(|| {
748 where_(format!("expected a number, found {kind}"))
749 })?))
750 }
751 },
752 "boolean" => match value {
753 Json::Null => Sea::Bool(None),
754 Json::Bool(v) => Sea::Bool(Some(*v)),
755 Json::Number(_) => Sea::Bool(Some(int_of()? != 0)),
756 _ => return Err(where_(format!("expected a boolean, found {kind}"))),
757 },
758 "text" | "character varying" | "character" | "varchar" | "bpchar" | "name" => match value {
759 Json::Null => Sea::String(None),
760 Json::String(v) => Sea::String(Some(Box::new(v.clone()))),
761 _ => return Err(where_(format!("expected a string, found {kind}"))),
762 },
763 other => {
764 return Err(where_(format!(
765 "target type {other:?} is outside the portable subset (TEXT, INTEGER, \
766 REAL, BOOLEAN)"
767 )));
768 }
769 })
770}