hayson-engine 0.13.0

use std::collections::HashSet;
use std::fs;
use std::path::{Path, PathBuf};

use serde_json::{Map, Value};

use crate::atomic::write_atomic;
use crate::catalog::{parse_catalog, write_catalog, Catalog, CatalogLayout};
use crate::errors::HaysonSchemaError;
use crate::lock::DirLock;

use super::database::{DatabaseError, HeldLock, LockMode};
use super::layout::{stored_kind_of, Layout, LayoutKind, DEFAULT_METADATA_FILE, SCHEMA_SUFFIX};

/// The marker a format-1 manifest carries under `hayson`, and a format-2 catalog under
/// `catalogVersion` (or, in a catalog written before 5.0.0, under `hayson` as well).
const MANIFEST_MARKER: u64 = 1;
const CATALOG_MARKER: u64 = 2;

#[cfg(unix)]
const PERMISSION_BITS: u32 = 0o777;

/// Which arrangement to move a database into, and under what name to leave its metadata document.
///
/// Every member is optional, and `Default` is what `migrateLayout(root)` does in TS: move into the
/// `nested` arrangement, keep the document's name, take the writer lock on the engine's own terms.
#[derive(Debug, Clone, Default)]
pub struct LayoutMigrationOptions {
    /// Arrangement to move into — [`LayoutKind::Nested`] when absent, as in TS.
    pub to: Option<LayoutKind>,
    /// Name the metadata document has now, [`DEFAULT_METADATA_FILE`] when absent — the metadata
    /// catalog in format 2, the manifest in format 1.
    pub from_metadata_file: Option<String>,
    /// Deprecated alias of `from_metadata_file`, taking the same value.
    pub from_manifest: Option<String>,
    /// Name to give the metadata document as part of the move. Unchanged when absent.
    pub metadata_file: Option<String>,
    /// Deprecated alias of `metadata_file`, taking the same value.
    pub manifest_file: Option<String>,
    /// Report the moves without touching the disk. A dry run takes no writer lock: it changes
    /// nothing, so refusing it because somebody else is writing would only hide the plan.
    pub dry_run: bool,
    /// How the writer lock is taken for the length of the move. [`LockMode::Skip`] is as dangerous
    /// here as it is in `OpenOptions`, and for the same reason: the lock is the only thing keeping
    /// this move and another process's writes off the same files.
    pub lock: LockMode,
}

/// One rename the move makes, as it will be applied.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LayoutMove {
    pub from: PathBuf,
    pub to: PathBuf,
}

/// What [`migrate_layout`] moved, or would move under `dry_run`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LayoutMigrationResult {
    pub root: PathBuf,
    pub from: LayoutKind,
    pub to: LayoutKind,
    /// Every rename the move needs, in the order it is applied. Empty when nothing had to move.
    pub moves: Vec<LayoutMove>,
    /// Files in the root the move does not understand and therefore left where they were.
    pub unclaimed: Vec<PathBuf>,
    /// `false` when the database was already in the target arrangement under the target name.
    pub changed: bool,
    pub dry_run: bool,
}

/// The metadata document as it sits on disk, in whichever of the two formats wrote it. A format-1
/// manifest is kept as the raw document rather than parsed into a type of its own: the move only
/// ever adds or removes its `layout` member, and carrying the rest through untouched is both
/// simpler and the only way a member this build does not know survives the move.
enum StoredMetadata {
    Manifest(Map<String, Value>),
    Catalog(Box<Catalog>),
}

impl StoredMetadata {
    /// The arrangement the document records. A document with no `layout` at all is flat — every
    /// database written before layouts existed is.
    fn layout_kind(&self) -> LayoutKind {
        match self {
            StoredMetadata::Manifest(manifest) => match manifest.get("layout").and_then(Value::as_str) {
                Some("nested") => LayoutKind::Nested,
                _ => LayoutKind::Flat,
            },
            StoredMetadata::Catalog(catalog) => catalog.layout.map_or(LayoutKind::Flat, LayoutKind::from),
        }
    }
}

/// Every path the move has to account for, in one arrangement of one database directory.
///
/// A superset of [`Layout`], which is narrowed to the roles this engine writes: a move also has to
/// carry the two roles only format 1 puts on disk — the per-table `<table>.schema.json` files and
/// the `pending/` staging directory — because a format-1 database is moved like any other, and is
/// left in format 1 by the move.
struct Roles {
    layout: Layout,
    schemas: PathBuf,
    pending: PathBuf,
}

impl Roles {
    fn new(root: &Path, kind: LayoutKind, metadata_file: &str) -> Roles {
        let under = |name: &str| {
            if kind == LayoutKind::Nested {
                root.join(name)
            } else {
                root.to_path_buf()
            }
        };
        Roles {
            layout: Layout::new(root, kind, metadata_file),
            schemas: under("schema"),
            pending: under("pending"),
        }
    }

    fn schema_path(&self, table: &str) -> PathBuf {
        self.schemas.join(format!("{table}{SCHEMA_SUFFIX}"))
    }

    /// Directory a partitioned table keeps its per-value logs in — moved whole, as one rename.
    fn partition_dir(&self, table: &str) -> PathBuf {
        self.layout.tables().join(table)
    }

    /// Directories the arrangement needs before anything is renamed into them, and — read the
    /// other way round — the ones left empty behind a move away from it.
    fn directories(&self) -> Vec<PathBuf> {
        if self.layout.kind() == LayoutKind::Nested {
            vec![
                self.layout.tables().to_path_buf(),
                self.schemas.clone(),
                self.layout.indexes().to_path_buf(),
                self.layout.runtime().to_path_buf(),
            ]
        } else {
            vec![self.layout.root().to_path_buf()]
        }
    }
}

/// The name two options spell the same metadata document under, refusing the pair that names two
/// different files: one of the two would otherwise be silently ignored.
fn metadata_file_of(
    current: (&str, Option<&str>),
    legacy: (&str, Option<&str>),
) -> Result<Option<String>, HaysonSchemaError> {
    match (current, legacy) {
        ((current_name, Some(current)), (legacy_name, Some(legacy))) if current != legacy => {
            Err(HaysonSchemaError(format!(
                "\"{current_name}\" is \"{current}\" and \"{legacy_name}\" is \"{legacy}\" — \"{legacy_name}\" is the deprecated alias of \"{current_name}\" and both name the same file, so one of the two would be silently ignored; pass only one"
            )))
        }
        ((_, Some(current)), _) => Ok(Some(current.to_string())),
        ((_, None), (_, legacy)) => Ok(legacy.map(str::to_owned)),
    }
}

fn read_stored_metadata(path: &Path) -> Result<Option<StoredMetadata>, DatabaseError> {
    let raw = match fs::read_to_string(path) {
        Ok(raw) => raw,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(DatabaseError::Io(error)),
    };
    let document: Value = serde_json::from_str(&raw).map_err(|error| {
        HaysonSchemaError(format!(
            "\"{}\" is not valid JSON — it may have been left truncated by an interrupted write: {error}",
            path.display()
        ))
    })?;
    let Value::Object(document) = document else {
        return Err(HaysonSchemaError(format!("manifest \"{}\" is not a hayson database", path.display())).into());
    };
    let marker = document
        .get("catalogVersion")
        .or_else(|| document.get("hayson"))
        .and_then(Value::as_u64);
    match marker {
        Some(MANIFEST_MARKER) => Ok(Some(StoredMetadata::Manifest(document))),
        Some(CATALOG_MARKER) => Ok(Some(StoredMetadata::Catalog(Box::new(parse_catalog(&Value::Object(
            document,
        ))?)))),
        Some(newer) => Err(HaysonSchemaError(format!(
            "manifest \"{}\" carries marker {newer}, and this build of hayson knows {MANIFEST_MARKER} (schema files) and {CATALOG_MARKER} (metadata catalog) — the database was written by a newer hayson, so upgrade it to open this directory",
            path.display()
        ))
        .into()),
        None => Err(HaysonSchemaError(format!("manifest \"{}\" is not a hayson database", path.display())).into()),
    }
}

/// Every table a format-1 database declares by having a `<table>.schema.json` for it, sorted. An
/// unreadable directory reads as no tables, exactly as the TS `readdir(...).catch(() => [])` does:
/// a flat database keeps its schema files in the root, which is never missing here.
fn schema_file_tables(dir: &Path) -> Vec<String> {
    let Ok(entries) = fs::read_dir(dir) else {
        return Vec::new();
    };
    let mut tables: Vec<String> = entries
        .filter_map(Result::ok)
        .filter_map(|entry| entry.file_name().to_str().map(str::to_owned))
        .filter_map(|name| name.strip_suffix(SCHEMA_SUFFIX).map(str::to_owned))
        .filter(|table| !table.is_empty())
        .collect();
    tables.sort();
    tables
}

#[cfg(unix)]
fn permissions_of(path: &Path) -> std::io::Result<Option<u32>> {
    use std::os::unix::fs::PermissionsExt;
    match fs::metadata(path) {
        Ok(metadata) => Ok(Some(metadata.permissions().mode() & PERMISSION_BITS)),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error),
    }
}

// NOTE: wasi has no permission bits at all; the host sandbox bounds fs access instead.
#[cfg(not(unix))]
fn permissions_of(_path: &Path) -> std::io::Result<Option<u32>> {
    Ok(None)
}

/// Writes a format-1 manifest the way [`write_catalog`] writes a catalog: `.tmp` on exclusive
/// terms, `fsync`, `rename`, inheriting the permissions of the file it replaces. Rendered with the
/// two-space indent and the trailing newline the TS writer used, so a manifest this moves reads
/// back byte for byte as one hayson has always written.
fn write_manifest(path: &Path, manifest: &Map<String, Value>) -> Result<(), DatabaseError> {
    let mode = permissions_of(path)?;
    let contents = format!(
        "{}\n",
        serde_json::to_string_pretty(manifest).map_err(std::io::Error::other)?
    );
    write_atomic(path, contents.as_bytes(), mode)?;
    Ok(())
}

/// Takes the writer lock of the directory for the length of a move, at the path the arrangement on
/// disk puts it — `.hayson.lock` in the root of a flat database, `runtime/.hayson.lock` in a
/// nested one — and names the root, not the lock file, in a refusal.
///
/// This is the whole of the mutual exclusion between a move and the rest of the engine: it is the
/// same [`DirLock`] on the same path that `Database::open` takes, so a move and a writer contend
/// for one lock file under one protocol, and neither can be running while the other is.
fn hold_for_move(root: &Path, layout: &Layout, lock: LockMode) -> Result<HeldLock, DatabaseError> {
    let LockMode::Enabled(options) = lock else {
        return Ok(HeldLock(None));
    };
    // NOTE: as in `open` — a nested database whose runtime/ was cleaned away stays lockable.
    if root.is_dir() {
        fs::create_dir_all(layout.runtime())?;
    }
    Ok(HeldLock(Some(DirLock::acquire_as(
        layout.lock_path(),
        options,
        root.to_path_buf(),
    )?)))
}

/// Moves a database between the `flat` and `nested` on-disk arrangements.
///
/// Idempotent: a database already in the target arrangement, under the target document name,
/// reports `changed: false` and moves nothing. Never automatic — in a repository this is a visible
/// change to tracked files, so it has to be asked for. Only files the database accounts for are
/// moved; anything else in the root is reported in `unclaimed` and left alone.
///
/// The directory is held under the writer lock for the length of the move (see [`hold_for_move`]),
/// so nothing here races a writer that has the directory open. The lock is released before the
/// directories the move emptied are removed, because in a nested database the lock file itself
/// lives in one of them.
///
/// A format-1 database is moved without being migrated: its `<table>.schema.json` files travel with
/// it and its manifest is left in format 1, so what `open()` says about it afterwards is exactly
/// what it said before.
///
/// @param root the database directory
/// @param options the arrangement to move into, the document names on either side of the move, and
/// whether to only report the plan
/// @returns what was moved, or would be moved under `dry_run`
pub fn migrate_layout(root: &Path, options: LayoutMigrationOptions) -> Result<LayoutMigrationResult, DatabaseError> {
    let target = options.to.unwrap_or(LayoutKind::Nested);
    let from_metadata_file = metadata_file_of(
        ("fromMetadataFile", options.from_metadata_file.as_deref()),
        ("fromManifest", options.from_manifest.as_deref()),
    )?
    .unwrap_or_else(|| DEFAULT_METADATA_FILE.to_string());
    let to_metadata_file = metadata_file_of(
        ("metadataFile", options.metadata_file.as_deref()),
        ("manifestFile", options.manifest_file.as_deref()),
    )?
    .unwrap_or_else(|| from_metadata_file.clone());

    // NOTE: read leniently first — only to learn where the lock lives; the read under it decides.
    let probed = stored_kind_of(&root.join(&from_metadata_file)).unwrap_or_default();
    let probed_layout = Layout::new(root, probed, &from_metadata_file);
    let held = if options.dry_run {
        HeldLock(None)
    } else {
        hold_for_move(root, &probed_layout, options.lock)?
    };

    let Some(stored) = read_stored_metadata(probed_layout.metadata_path())? else {
        return Err(HaysonSchemaError(format!(
            "\"{}\" holds no metadata document \"{from_metadata_file}\" — there is no hayson database here to move",
            root.display()
        ))
        .into());
    };
    let current = stored.layout_kind();
    if !options.dry_run && current != probed {
        return Err(HaysonSchemaError(format!(
            "\"{}\" was in the \"{}\" arrangement when the lock for the move was taken and is in the \"{}\" one now — the lock taken is the one the old arrangement names, so nothing was moved; open or move it again",
            root.display(),
            probed.as_str(),
            current.as_str()
        ))
        .into());
    }

    let from = Roles::new(root, current, &from_metadata_file);
    let to = Roles::new(root, target, &to_metadata_file);

    if current == target && from_metadata_file == to_metadata_file {
        return Ok(LayoutMigrationResult {
            root: root.to_path_buf(),
            from: current,
            to: target,
            moves: Vec::new(),
            unclaimed: Vec::new(),
            changed: false,
            dry_run: options.dry_run,
        });
    }

    if from.layout.journal_path().exists() {
        return Err(HaysonSchemaError(format!(
            "\"{}\" holds an unfinished unit of work — open the database once so it rolls forward, then move it",
            root.display()
        ))
        .into());
    }

    let tables: Vec<String> = match &stored {
        StoredMetadata::Catalog(catalog) => catalog.tables.keys().cloned().collect(),
        StoredMetadata::Manifest(_) => schema_file_tables(&from.schemas),
    };

    let mut planned: Vec<LayoutMove> = Vec::with_capacity(tables.len() * 4 + 1);
    for table in &tables {
        planned.push(LayoutMove {
            from: from.schema_path(table),
            to: to.schema_path(table),
        });
        planned.push(LayoutMove {
            from: from.layout.index_path(table),
            to: to.layout.index_path(table),
        });
        planned.push(LayoutMove {
            from: from.layout.log_path(table),
            to: to.layout.log_path(table),
        });
        planned.push(LayoutMove {
            from: from.partition_dir(table),
            to: to.partition_dir(table),
        });
    }
    // NOTE: last, so a table's own refusal is reported before the metadata document's.
    planned.push(LayoutMove {
        from: from.layout.metadata_path().to_path_buf(),
        to: to.layout.metadata_path().to_path_buf(),
    });

    let mut moves: Vec<LayoutMove> = Vec::new();
    for pair in planned {
        if pair.from == pair.to || !pair.from.exists() {
            continue;
        }
        if pair.to.exists() {
            return Err(HaysonSchemaError(format!(
                "\"{}\" already exists — refusing to move over it",
                pair.to.display()
            ))
            .into());
        }
        moves.push(pair);
    }

    let unclaimed = unclaimed_in(root, &from, &to, &moves);

    if options.dry_run {
        return Ok(LayoutMigrationResult {
            root: root.to_path_buf(),
            from: current,
            to: target,
            moves,
            unclaimed,
            changed: true,
            dry_run: true,
        });
    }

    for directory in to.directories() {
        fs::create_dir_all(&directory)?;
    }
    for step in &moves {
        fs::rename(&step.from, &step.to)?;
    }
    match stored {
        StoredMetadata::Catalog(mut catalog) => {
            // K4 requires the catalog to carry its layout out loud, `flat` included — never dropped.
            catalog.layout = Some(CatalogLayout::from(target));
            write_catalog(to.layout.metadata_path(), &catalog)
                .map_err(|error| DatabaseError::Catalog(error.to_string()))?;
        }
        StoredMetadata::Manifest(mut manifest) => {
            if target == LayoutKind::Nested {
                manifest.insert("layout".to_string(), Value::from(target.as_str()));
            } else {
                manifest.remove("layout");
            }
            write_manifest(to.layout.metadata_path(), &manifest)?;
        }
    }

    // NOTE: before the cleanup — in a nested database the lock file lives in one of these.
    drop(held);
    let kept = to.directories();
    for directory in from.directories() {
        if directory == root || kept.contains(&directory) {
            continue;
        }
        let _ = fs::remove_dir(&directory);
    }

    Ok(LayoutMigrationResult {
        root: root.to_path_buf(),
        from: current,
        to: target,
        moves,
        unclaimed,
        changed: true,
        dry_run: false,
    })
}

/// Everything in the root the move neither carries nor owns — reported rather than moved, because
/// a file this cannot name is as likely to be a hand-written note as a stray log, and guessing
/// which would mean moving somebody's file on a guess.
fn unclaimed_in(root: &Path, from: &Roles, to: &Roles, moves: &[LayoutMove]) -> Vec<PathBuf> {
    let claimed: HashSet<&PathBuf> = moves.iter().map(|step| &step.from).collect();
    let mut structural: HashSet<PathBuf> = from.directories().into_iter().collect();
    structural.extend(to.directories());
    structural.insert(from.pending.clone());
    structural.insert(to.pending.clone());
    structural.insert(from.layout.metadata_path().to_path_buf());
    structural.insert(from.layout.lock_path());

    let Ok(entries) = fs::read_dir(root) else {
        return Vec::new();
    };
    let mut unclaimed: Vec<PathBuf> = entries
        .filter_map(Result::ok)
        .map(|entry| root.join(entry.file_name()))
        .filter(|path| !claimed.contains(path) && !structural.contains(path))
        .collect();
    unclaimed.sort();
    unclaimed
}

#[cfg(test)]
mod tests {
    use std::sync::mpsc;
    use std::thread;
    use std::time::Duration;

    use serde_json::json;
    use tempfile::TempDir;

    use super::*;
    use crate::database::{Database, LayoutOptions, OpenOptions};
    use crate::lock::{DirLockError, LockOptions, LockState};

    fn users_schema() -> Value {
        json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
        })
    }

    fn seeded(kind: LayoutKind) -> TempDir {
        let dir = TempDir::new().unwrap();
        let mut database = Database::open_with(
            dir.path(),
            OpenOptions {
                layout: LayoutOptions {
                    kind: Some(kind),
                    ..Default::default()
                },
                ..Default::default()
            },
        )
        .unwrap();
        database.create_table("users", users_schema()).unwrap();
        database
            .table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        database.close().unwrap();
        dir
    }

    fn fail_fast() -> LayoutMigrationOptions {
        LayoutMigrationOptions {
            lock: LockMode::Enabled(LockOptions {
                timeout_ms: Some(0),
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    #[test]
    fn the_lock_a_move_takes_is_the_one_an_engine_writer_is_refused_by() {
        let dir = seeded(LayoutKind::Flat);
        let layout = Layout::new(dir.path(), LayoutKind::Flat, DEFAULT_METADATA_FILE);

        let held = hold_for_move(dir.path(), &layout, LockMode::default()).unwrap();

        let state = DirLock::inspect(layout.lock_path(), LockOptions::default()).unwrap();
        assert!(
            matches!(state, LockState::Held(_)),
            "a move must leave a live holder in the lock file, got {state:?}"
        );
        let refused = match Database::open_with(
            dir.path(),
            OpenOptions {
                lock: LockMode::Enabled(LockOptions {
                    timeout_ms: Some(0),
                    ..Default::default()
                }),
                ..Default::default()
            },
        ) {
            Ok(_) => panic!("a writer opened a directory the move was holding"),
            Err(refused) => refused,
        };
        match refused {
            DatabaseError::Lock(DirLockError::Locked(error)) => {
                assert_eq!(
                    error.holder.map(|holder| holder.owner),
                    state.holder().map(|holder| holder.owner.clone()),
                    "the writer must be refused by the very holder the move wrote"
                );
            }
            other => panic!("expected the writer to be refused by the move's lock, got {other}"),
        }

        drop(held);
        Database::open(dir.path()).unwrap().close().unwrap();
    }

    #[test]
    fn a_move_is_refused_while_an_engine_writer_holds_the_directory_and_moves_nothing() {
        let dir = seeded(LayoutKind::Flat);
        let database = Database::open(dir.path()).unwrap();

        let refused = migrate_layout(dir.path(), fail_fast()).unwrap_err();

        assert!(
            matches!(refused, DatabaseError::Lock(DirLockError::Locked(_))),
            "expected the open writer to refuse the move, got {refused}"
        );
        assert!(dir.path().join("users.jsonl").is_file(), "the log must not have moved");
        assert!(!dir.path().join("tables").exists(), "nothing may be created either");
        database.close().unwrap();
    }

    #[test]
    fn a_move_waits_for_an_engine_writer_to_leave_and_then_lands() {
        let dir = seeded(LayoutKind::Flat);
        let database = Database::open(dir.path()).unwrap();
        let (opened, held) = mpsc::channel();

        let writer = thread::spawn(move || {
            opened.send(()).unwrap();
            thread::sleep(Duration::from_millis(250));
            database.close().unwrap();
        });
        held.recv().unwrap();

        let started_at = std::time::Instant::now();
        let moved = migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                lock: LockMode::Enabled(LockOptions {
                    timeout_ms: Some(10_000),
                    retry_interval_ms: Some(10),
                    ..Default::default()
                }),
                ..Default::default()
            },
        )
        .unwrap();
        writer.join().unwrap();

        assert!(
            started_at.elapsed() >= Duration::from_millis(200),
            "the move must have waited the writer out, not run beside it"
        );
        assert!(moved.changed);
        assert!(dir.path().join("tables").join("users.jsonl").is_file());
    }

    #[test]
    fn a_dry_run_plans_the_move_without_taking_the_lock_or_touching_the_disk() {
        let dir = seeded(LayoutKind::Flat);
        let database = Database::open(dir.path()).unwrap();

        let planned = migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                dry_run: true,
                ..fail_fast()
            },
        )
        .unwrap();

        assert!(planned.dry_run && planned.changed);
        assert!(planned.moves.iter().any(|step| step.from.ends_with("users.jsonl")));
        assert!(!dir.path().join("tables").exists());
        database.close().unwrap();
    }

    #[test]
    fn a_move_and_a_move_back_leave_the_records_and_the_arrangement_where_they_started() {
        let dir = seeded(LayoutKind::Flat);

        let there = migrate_layout(dir.path(), LayoutMigrationOptions::default()).unwrap();
        assert_eq!((there.from, there.to), (LayoutKind::Flat, LayoutKind::Nested));
        assert!(dir.path().join("tables").join("users.jsonl").is_file());
        assert!(!dir.path().join("users.jsonl").exists());

        let back = migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                to: Some(LayoutKind::Flat),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(back.changed);
        for emptied in ["tables", "index", "runtime"] {
            assert!(!dir.path().join(emptied).exists(), "{emptied} should be gone");
        }

        let database = Database::open(dir.path()).unwrap();
        assert_eq!(database.layout().kind(), LayoutKind::Flat);
        assert_eq!(database.table("users").unwrap().len(), 1);
    }

    #[test]
    fn moving_a_database_that_is_already_there_changes_nothing() {
        let dir = seeded(LayoutKind::Nested);

        let again = migrate_layout(dir.path(), LayoutMigrationOptions::default()).unwrap();

        assert!(!again.changed);
        assert!(again.moves.is_empty());
    }

    #[test]
    fn a_directory_with_no_metadata_document_is_refused_and_left_without_a_lock_file() {
        let dir = TempDir::new().unwrap();

        let refused = migrate_layout(dir.path(), LayoutMigrationOptions::default()).unwrap_err();

        assert!(refused.to_string().contains("no metadata document"), "{refused}");
        assert!(
            !dir.path().join(crate::lock::LOCK_FILE).exists(),
            "a refused move must take its own lock file with it"
        );
    }

    #[test]
    fn a_truncated_metadata_document_is_refused_naming_the_file() {
        let dir = TempDir::new().unwrap();
        fs::write(dir.path().join("hayson.json"), "{\"catalogVersion\": 2, \"tables\": {").unwrap();

        let refused = migrate_layout(dir.path(), LayoutMigrationOptions::default()).unwrap_err();

        assert!(matches!(refused, DatabaseError::Schema(_)), "{refused}");
        assert!(refused.to_string().contains("hayson.json"), "{refused}");
    }

    #[test]
    fn a_taken_target_name_is_refused_before_anything_moves() {
        let dir = seeded(LayoutKind::Flat);
        fs::write(dir.path().join("cerebro.json"), "{\"this\":\"is not ours\"}\n").unwrap();

        let refused = migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                metadata_file: Some("cerebro.json".to_string()),
                ..Default::default()
            },
        )
        .unwrap_err();

        assert!(refused.to_string().contains("refusing to move over it"), "{refused}");
        assert_eq!(
            fs::read_to_string(dir.path().join("cerebro.json")).unwrap(),
            "{\"this\":\"is not ours\"}\n"
        );
        assert!(dir.path().join("users.jsonl").is_file());
        assert!(!dir.path().join("tables").exists());
    }

    #[test]
    fn two_disagreeing_names_for_one_document_are_refused_naming_both_options() {
        let dir = seeded(LayoutKind::Flat);

        let refused = migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                from_metadata_file: Some("cerebro.json".to_string()),
                from_manifest: Some("hayson.json".to_string()),
                ..Default::default()
            },
        )
        .unwrap_err();

        let message = refused.to_string();
        assert!(
            message.contains("fromMetadataFile") && message.contains("fromManifest"),
            "{message}"
        );
    }

    #[test]
    fn a_file_the_move_does_not_understand_is_reported_rather_than_carried() {
        let dir = seeded(LayoutKind::Flat);
        fs::write(dir.path().join("notes.md"), "hand-written\n").unwrap();

        let planned = migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                dry_run: true,
                ..Default::default()
            },
        )
        .unwrap();

        assert_eq!(planned.unclaimed, vec![dir.path().join("notes.md")]);
        assert!(dir.path().join("notes.md").is_file());
    }

    #[test]
    fn an_unfinished_unit_of_work_stops_the_move_before_it_starts() {
        let dir = seeded(LayoutKind::Flat);
        fs::write(dir.path().join(crate::txn::JOURNAL_FILE), "{}\n").unwrap();

        let refused = migrate_layout(dir.path(), LayoutMigrationOptions::default()).unwrap_err();

        assert!(refused.to_string().contains("unfinished unit of work"), "{refused}");
        assert!(!dir.path().join("tables").exists());
    }

    #[test]
    fn a_format_1_database_moves_with_its_schema_files_and_stays_on_format_1() {
        let dir = TempDir::new().unwrap();
        fs::write(
            dir.path().join("users.schema.json"),
            serde_json::to_string_pretty(&users_schema()).unwrap(),
        )
        .unwrap();
        fs::write(dir.path().join("users.jsonl"), "").unwrap();
        let manifest = json!({
            "hayson": 1,
            "formatVersion": 1,
            "createdAt": 1_786_995_100_337u64,
            "tables": ["users"],
        });
        fs::write(
            dir.path().join("hayson.json"),
            format!("{}\n", serde_json::to_string_pretty(&manifest).unwrap()),
        )
        .unwrap();

        let moved = migrate_layout(dir.path(), LayoutMigrationOptions::default()).unwrap();

        assert!(moved.changed);
        assert!(dir.path().join("schema").join("users.schema.json").is_file());
        assert!(dir.path().join("tables").join("users.jsonl").is_file());
        let written: Value =
            serde_json::from_str(&fs::read_to_string(dir.path().join("hayson.json")).unwrap()).unwrap();
        assert_eq!(written["hayson"], json!(1));
        assert_eq!(written["formatVersion"], json!(1));
        assert_eq!(written["layout"], json!("nested"));
        assert_eq!(written["tables"], json!(["users"]));
    }

    #[test]
    fn moving_a_format_1_database_back_drops_the_layout_member_instead_of_writing_flat() {
        let dir = TempDir::new().unwrap();
        fs::create_dir_all(dir.path().join("schema")).unwrap();
        fs::write(
            dir.path().join("schema").join("users.schema.json"),
            serde_json::to_string_pretty(&users_schema()).unwrap(),
        )
        .unwrap();
        let manifest = json!({
            "hayson": 1,
            "formatVersion": 1,
            "createdAt": 1_786_995_100_337u64,
            "tables": ["users"],
            "layout": "nested",
        });
        fs::write(
            dir.path().join("hayson.json"),
            format!("{}\n", serde_json::to_string_pretty(&manifest).unwrap()),
        )
        .unwrap();

        migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                to: Some(LayoutKind::Flat),
                ..Default::default()
            },
        )
        .unwrap();

        let written: Value =
            serde_json::from_str(&fs::read_to_string(dir.path().join("hayson.json")).unwrap()).unwrap();
        assert!(written.get("layout").is_none(), "a flat manifest names no layout");
        assert!(dir.path().join("users.schema.json").is_file());
        assert!(!dir.path().join("schema").exists());
    }

    #[test]
    fn a_catalog_names_its_arrangement_out_loud_after_a_move_back_to_flat() {
        let dir = seeded(LayoutKind::Nested);

        migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                to: Some(LayoutKind::Flat),
                ..Default::default()
            },
        )
        .unwrap();

        let written: Value =
            serde_json::from_str(&fs::read_to_string(dir.path().join("hayson.json")).unwrap()).unwrap();
        assert_eq!(written["layout"], json!("flat"));
    }

    #[test]
    fn renaming_the_document_alone_moves_it_and_nothing_else() {
        let dir = seeded(LayoutKind::Flat);

        let renamed = migrate_layout(
            dir.path(),
            LayoutMigrationOptions {
                to: Some(LayoutKind::Flat),
                metadata_file: Some("cerebro.json".to_string()),
                ..Default::default()
            },
        )
        .unwrap();

        assert_eq!(
            renamed.moves,
            vec![LayoutMove {
                from: dir.path().join("hayson.json"),
                to: dir.path().join("cerebro.json"),
            }]
        );
        assert!(dir.path().join("cerebro.json").is_file());
        assert!(!dir.path().join("hayson.json").exists());
    }

    #[test]
    fn a_partitioned_table_carries_its_whole_directory_across() {
        let dir = TempDir::new().unwrap();
        let mut database = Database::open(dir.path()).unwrap();
        database
            .create_table(
                "notes",
                json!({
                    "schemaVersion": 1,
                    "table": "notes",
                    "primaryKey": "id",
                    "partitionBy": "kind",
                    "fields": {
                        "id": { "type": "string" },
                        "kind": { "type": "string", "required": true },
                    },
                }),
            )
            .unwrap();
        let notes = database.table_mut("notes").unwrap();
        notes.put(&json!({ "id": "a", "kind": "adr" }), false).unwrap();
        notes.put(&json!({ "id": "n", "kind": "note" }), false).unwrap();
        database.close().unwrap();

        migrate_layout(dir.path(), LayoutMigrationOptions::default()).unwrap();

        assert!(dir.path().join("tables").join("notes").join("adr.jsonl").is_file());
        assert!(dir.path().join("tables").join("notes").join("note.jsonl").is_file());
        let database = Database::open(dir.path()).unwrap();
        assert_eq!(database.table("notes").unwrap().len(), 2);
    }
}