hayson-engine 0.27.0

use std::collections::HashMap;
use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use serde_json::Value;

use crate::catalog::{read_catalog_text, Catalog, ReadCatalogError, CATALOG_SCOPE_NAME};
use crate::crypto::{keyed_physical_path, CipherOptions, Sealer};
use crate::schema::{is_physical_path, physical_path_of, FieldType};
use crate::storage::LOG_SUFFIX;

use super::backup::{
    owned_entries, read_catalog_of, refuse, restore_staged, sealer_of, ManifestEntry, Owned, RestorePart,
};
use super::database::{Database, DatabaseError, LockMode};
use super::layout::LayoutOptions;

#[derive(Debug, Clone, Default)]
pub struct ConvertOptions {
    /// The keys the database is sealed with now; `None` for a plain database.
    pub from: Option<CipherOptions>,
    /// The keys it is to be sealed with afterwards; `None` to decrypt it.
    pub to: Option<CipherOptions>,
    pub lock: LockMode,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConvertResult {
    pub files: usize,
    pub tables: usize,
    pub images: usize,
    /// Log lines whose image references had to be renamed.
    pub lines_renamed: u64,
}

fn read_error(error: ReadCatalogError) -> DatabaseError {
    match error {
        ReadCatalogError::Io(error) => DatabaseError::Io(error),
        ReadCatalogError::Cipher(error)
        | ReadCatalogError::Encrypted(error)
        | ReadCatalogError::NotEncrypted(error) => DatabaseError::Cipher(Box::new(error)),
    }
}

fn open_stored(sealer: Option<&Sealer>, scope: &str, bytes: Vec<u8>) -> Result<Vec<u8>, DatabaseError> {
    match sealer {
        Some(sealer) => Ok(sealer.open_blob(scope, &bytes)?),
        None => Ok(bytes),
    }
}

fn seal_stored(sealer: Option<&Sealer>, scope: &str, bytes: &[u8]) -> Vec<u8> {
    match sealer {
        Some(sealer) => sealer.seal_blob(scope, bytes),
        None => bytes.to_vec(),
    }
}

fn write_synced(path: &Path, bytes: &[u8]) -> io::Result<()> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    let mut file = fs::File::create(path)?;
    io::Write::write_all(&mut file, bytes)?;
    file.sync_all()
}

/// Rewrites every `image` field of a record line so its `physicalPath` follows `renamed`.
fn rename_images(line: &str, image_fields: &[String], renamed: &HashMap<String, String>) -> Option<String> {
    let mut value: Value = serde_json::from_str(line).ok()?;
    let object = value.as_object_mut()?;
    let mut changed = false;
    for field in image_fields {
        let Some(Value::Object(image)) = object.get_mut(field) else {
            continue;
        };
        let Some(Value::String(path)) = image.get_mut("physicalPath") else {
            continue;
        };
        if let Some(new) = renamed.get(path.as_str()) {
            *path = new.clone();
            changed = true;
        }
    }
    changed.then(|| serde_json::to_string(&value).expect("a parsed value serializes"))
}

/// Seals a plain database, unseals an encrypted one, or re-seals one under other keys, in place and
/// recoverably (the stage, verify, swap and roll-back of [`super::restore_database`]).
///
/// Every file is read through `from` and written through `to`: the metadata document, every log line
/// (fresh IVs; image references renamed where the file name changes), every image file. Image file
/// names are a plain content hash in a plain database and a keyed one in an encrypted one, so they are
/// recomputed and the records that name them are rewritten. Index snapshots are dropped; the next open
/// rebuilds them. A database with an unfinished unit of work is refused.
pub fn convert_encryption(dir: &Path, options: &ConvertOptions) -> Result<ConvertResult, DatabaseError> {
    if options.from.is_none() && options.to.is_none() {
        return Err(refuse(
            "neither \"from\" nor \"to\" names keys — there is nothing to convert".to_string(),
        ));
    }
    let from = sealer_of(options.from.as_ref())?;
    let to = sealer_of(options.to.as_ref())?;
    let layout = Database::resolve_layout_sealed(dir, &LayoutOptions::default(), from.as_deref())?;
    let catalog = read_catalog_of(&layout, from.as_deref())?;
    // NOTE: read again under the lock by `convert_files`, which refuses if the table set moved.
    let tables_before: Vec<String> = catalog.tables.keys().cloned().collect();
    let part = RestorePart {
        sub: PathBuf::new(),
        encryption: options.to.clone(),
        current_encryption: Some(options.from.clone()),
        metadata_file: layout.metadata_file().to_string(),
        entries: Vec::new(),
    };
    let mut summary = ConvertResult {
        files: 0,
        tables: tables_before.len(),
        images: 0,
        lines_renamed: 0,
    };
    let outcome = {
        let summary = &mut summary;
        restore_staged(dir, vec![part], &options.lock, None, |staged, parts| {
            let (bytes, entries, result) = convert_files(dir, staged, &tables_before, from.clone(), to.clone())?;
            parts[0].entries = entries;
            *summary = result;
            Ok(bytes)
        })?
    };
    let _ = outcome;
    Ok(summary)
}

fn convert_files(
    root: &Path,
    staged: &Path,
    tables_before: &[String],
    from: Option<Arc<Sealer>>,
    to: Option<Arc<Sealer>>,
) -> Result<(u64, Vec<ManifestEntry>, ConvertResult), DatabaseError> {
    let layout = Database::resolve_layout_sealed(root, &LayoutOptions::default(), from.as_deref())?;
    let catalog: Catalog = read_catalog_of(&layout, from.as_deref())?;
    if catalog.tables.keys().cloned().collect::<Vec<_>>() != tables_before {
        return Err(refuse(
            "the database changed while the conversion was being prepared — run it again".to_string(),
        ));
    }
    if layout.journal_path().exists() {
        return Err(refuse(format!(
            "\"{}\" holds an unfinished unit of work — open the database for writing once so it is finished, then convert it",
            layout.journal_path().display()
        )));
    }
    let files = owned_entries(root, &layout, &catalog, true, &[])?;

    let mut entries = Vec::new();
    let mut total = 0u64;
    let mut image_count = 0usize;
    let mut lines_renamed = 0u64;

    // Images first: their new names decide how the log lines are rewritten.
    let mut renamed: HashMap<String, HashMap<String, String>> = HashMap::new();
    let mut written: Vec<(PathBuf, u64)> = Vec::new();
    for (relative, kind) in &files {
        let Owned::Image(table) = kind else { continue };
        let name = relative.file_name().and_then(|n| n.to_str()).unwrap_or_default();
        if !is_physical_path(name) {
            // Not a file this engine wrote; carried over untouched.
            let bytes = fs::read(root.join(relative))?;
            write_synced(&staged.join(relative), &bytes)?;
            written.push((relative.clone(), bytes.len() as u64));
            continue;
        }
        let stored = fs::read(root.join(relative))?;
        let plain = open_stored(from.as_deref(), &format!("image:{table}"), stored)?;
        let new_name = match to.as_deref() {
            Some(sealer) => keyed_physical_path(&sealer.image_name_key(), &plain),
            None => physical_path_of(&plain),
        };
        let target = relative.with_file_name(&new_name);
        let sealed = seal_stored(to.as_deref(), &format!("image:{table}"), &plain);
        write_synced(&staged.join(&target), &sealed)?;
        written.push((target, sealed.len() as u64));
        if new_name != name {
            renamed
                .entry(table.clone())
                .or_default()
                .insert(name.to_string(), new_name);
        }
        image_count += 1;
    }

    for (relative, kind) in &files {
        match kind {
            Owned::Image(_) | Owned::Index(_) => continue,
            Owned::Metadata => {
                let text = read_catalog_text(&root.join(relative), from.as_deref()).map_err(read_error)?;
                let text = text.ok_or_else(|| refuse("the metadata document disappeared".to_string()))?;
                let bytes = seal_stored(to.as_deref(), CATALOG_SCOPE_NAME, text.as_bytes());
                write_synced(&staged.join(relative), &bytes)?;
                written.push((relative.clone(), bytes.len() as u64));
            }
            Owned::Log(table) => {
                let image_fields: Vec<String> = catalog.tables[table]
                    .fields
                    .iter()
                    .filter(|(_, spec)| spec.spec.field_type == FieldType::Image)
                    .map(|(field, _)| field.clone())
                    .collect();
                let names = renamed.get(table);
                let scope = format!("log:{table}");
                let text = fs::read_to_string(root.join(relative))?;
                let mut out = String::with_capacity(text.len());
                for line in text.split_inclusive('\n') {
                    if !line.ends_with('\n') {
                        // A half-written tail is what a writable open cuts off; it is not carried.
                        break;
                    }
                    let stored = &line[..line.len() - 1];
                    if stored.is_empty() {
                        continue;
                    }
                    let mut plain = match from.as_deref() {
                        Some(sealer) => sealer.open_line(&scope, stored)?,
                        None => stored.to_string(),
                    };
                    if let (Some(names), false) = (names, image_fields.is_empty()) {
                        if let Some(rewritten) = rename_images(&plain, &image_fields, names) {
                            plain = rewritten;
                            lines_renamed += 1;
                        }
                    }
                    out.push_str(&match to.as_deref() {
                        Some(sealer) => sealer.seal_line(&scope, &plain),
                        None => plain,
                    });
                    out.push('\n');
                }
                write_synced(&staged.join(relative), out.as_bytes())?;
                written.push((relative.clone(), out.len() as u64));
            }
        }
    }
    let _ = LOG_SUFFIX;

    for (relative, bytes) in written {
        total += bytes;
        entries.push(ManifestEntry { relative, bytes });
    }
    let result = ConvertResult {
        files: entries.len(),
        tables: tables_before.len(),
        images: image_count,
        lines_renamed,
    };
    Ok((total, entries, result))
}