Skip to main content

cratefield_cli/
data.rs

1//! `fz data export` / `fz data import` (issue #21): moving a venture's
2//! D1 data to Postgres — the migration-path step between "stand up
3//! Postgres, run the migrations" and "switch the runtime" (architecture
4//! section 10).
5//!
6//! **Export** reads a venture SQLite database — D1 is SQLite, and the
7//! Cloudflare-side step is `wrangler d1 export` loaded into a local
8//! SQLite file; `docs/DATA-MOVE.md` is the runbook. It writes one JSON
9//! Lines file: a manifest line first (per-table row counts and sha256,
10//! tables in lock order), then one `{"table","row"}` record per table
11//! row, tables in the same lock order.
12//!
13//! **Import** (requires the crate's `postgres` feature, like `fz
14//! migrations apply`) loads that file into a Postgres database in lock
15//! order: it verifies every table's sha256 against the manifest,
16//! refuses a non-empty table without `--append`, inserts batched
17//! multi-row statements inside one transaction per table, and verifies
18//! row counts afterwards. `--plan` prints what either command would do
19//! and writes nothing.
20
21use 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/// Rows per multi-row INSERT during import: 200 rows × ≤ ~16 columns
28/// stays far below Postgres' 65 535 bind-parameter ceiling.
29#[cfg(feature = "postgres")]
30const IMPORT_BATCH_ROWS: usize = 200;
31
32// ---------------------------------------------------------------------------
33// Manifest and record model
34
35#[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    /// Sidecar modules whose tables this export deliberately omits
47    /// (issue #66). Present only when `--without-sidecar-tables` was
48    /// given: the artifact is the audit record of what moved, so it has
49    /// to record what did not.
50    #[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
60/// The parsed export file. Record lines are kept as raw bytes: the
61/// sha256 must cover exactly what export wrote, not a re-serialization.
62struct 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
108/// A manifest comes from a file anyone can edit, and its table and
109/// column names end up inside generated SQL: accept only the portable
110/// identifier shape, then quote them.
111fn 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
124/// The venture's tables in lock order: modules in harness (config)
125/// order, each module's tables in declared order — the order `fz
126/// migrations collect` pins and the manifest carries. The migrations
127/// bookkeeping table is infrastructure, never venture data.
128fn 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
136// ---------------------------------------------------------------------------
137// Export
138
139/// Exports the venture's tables from the SQLite database at `db_path`
140/// to `out_path` as manifest + JSONL records (see the module docs).
141/// `plan` prints the per-table summary without writing.
142///
143/// # Errors
144///
145/// A human-readable message when the database cannot be opened, a table
146/// is missing, a value is outside the portable subset, or the file
147/// cannot be written.
148pub 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    // A sidecar's tables live in the same database and are invisible
157    // here (issue #66). Exporting them silently would hand someone an
158    // artifact that looks complete and is not, which is the one thing a
159    // data move must never do.
160    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    // Keys sorted explicitly: the file format is "byte-identical across
252    // exports", and that must not depend on whether serde_json was built
253    // with `preserve_order` (it is, for the UI surface, since ADR 0010).
254    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        // BLOBs are outside the portable subset (ADR 0004); refuse them
303        // rather than invent an encoding import could not round-trip.
304        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
312// ---------------------------------------------------------------------------
313// Import (Postgres; requires the `postgres` feature)
314
315/// Imports an export file into the Postgres database at `url` (see the
316/// module docs). `plan` prints the per-table target state without
317/// writing.
318///
319/// # Errors
320///
321/// A human-readable message when the file is malformed, a sha256 or row
322/// count disagrees with the manifest, a target table is missing or
323/// non-empty without `--append`, or any insert fails. The connection
324/// string is never echoed.
325pub 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
351/// The manifest must be exactly this venture's tables, in lock order,
352/// each a valid identifier — the wrong venture's export file fails
353/// before anything touches the network.
354fn 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/// Per table, the target's `(column, data type)` list in ordinal order —
435/// the type drives every bind, including typed NULLs (a text-typed NULL
436/// will not insert into an integer column).
437#[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/// The manifest's columns must be exactly the target's columns before
475/// anything is written — an export from a drifted schema fails loudly.
476#[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/// Row counts before any write: a non-empty table without `--append` is
503/// refused up front, so a partial import never starts.
504#[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/// One table's `(table, raw line)` record group.
526#[cfg(feature = "postgres")]
527type RecordGroup = [(String, Vec<u8>)];
528
529/// Slices the file into per-table record groups in manifest (lock)
530/// order. Records must arrive grouped per table — anything else is a
531/// hand-edited file.
532#[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/// Verifies EVERY table's sha256 against the manifest before anything
567/// is written — a tampered file leaves the target untouched (earlier
568/// tables in lock order are not imported around a later failure).
569#[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/// Inserts each table's records in batched multi-row statements inside
590/// one transaction per table (atomic per table; `--append` re-runs
591/// resume at the tables that are still empty).
592#[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/// After the writes: every table must hold exactly what the manifest
608/// promised on top of what it had.
609#[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/// Maps one JSON value onto the sea-query bind matching the target
702/// column's Postgres type, so SQL NULL binds as a NULL of the right
703/// type and integers do not arrive as text.
704#[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}