liteboxfs 0.1.0

use std::collections::HashMap;

use rusqlite::OptionalExtension;

use super::store::SqlStore;

use crate::path::NormalizedPath;
use crate::util::system_time_from_nanos;
use crate::{
    FileOrigin, RootId,
    block::FileId as StoreFileId,
    errors::InternalError,
    id::Root,
    metadata::{FileKind, FileMetadata, FileMode, Owner},
    sql::FileDiscriminant,
};

impl<'conn> SqlStore<'conn> {
    pub fn root_exists(&self, id: RootId) -> crate::Result<bool> {
        let mut stmt = self.db.prepare_cached(
            r#"
                SELECT
                    1
                FROM
                    liteboxfs_roots
                WHERE
                    uuid = ?;
            "#,
        )?;

        match stmt.query_one(rusqlite::params![id.to_string()], |_row| Ok(())) {
            Ok(_) => Ok(true),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(false),
            Err(err) => Err(crate::Error::from(err)),
        }
    }

    pub fn get_root_info(&self, id: RootId) -> crate::Result<Root> {
        let mut stmt = self.db.prepare_cached(
            r#"
                SELECT
                    name,
                    created
                FROM
                    liteboxfs_roots
                WHERE
                    uuid = ?;
            "#,
        )?;

        let query_result = stmt.query_one(rusqlite::params![id.to_string()], |row| {
            Ok((row.get::<_, Option<String>>(0)?, row.get::<_, i128>(1)?))
        });

        match query_result {
            Ok((name, nanos)) => Ok(Root {
                id,
                name,
                created: system_time_from_nanos(nanos),
            }),
            Err(rusqlite::Error::QueryReturnedNoRows) => {
                Err(crate::Error::RootNotFound { root: id.into() })
            }
            Err(err) => Err(err.into()),
        }
    }

    pub fn set_root_name(&self, id: RootId, name: Option<&str>) -> crate::Result<()> {
        let mut stmt = self.db.prepare_cached(
            r#"
                UPDATE OR IGNORE
                    liteboxfs_roots
                SET
                    name = ?
                WHERE
                    uuid = ?;
            "#,
        )?;

        let params = rusqlite::params![name, id.to_string()];
        let rows_updated = stmt.execute(params)?;

        if let Some(name) = name
            && rows_updated == 0
        {
            return Err(crate::Error::RootAlreadyExists {
                name: name.to_string(),
            });
        }

        Ok(())
    }

    pub fn get_root_by_name(&self, name: &str) -> crate::Result<Root> {
        let mut stmt = self.db.prepare_cached(
            r#"
                SELECT
                    uuid,
                    created
                FROM
                    liteboxfs_roots
                WHERE
                    name = ?;
            "#,
        )?;

        let query_result = stmt
            .query_one(rusqlite::params![name], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, i128>(1)?))
            })
            .map(|(uuid, nanos)| {
                crate::Result::Ok(Root {
                    id: uuid.parse().map_err(|_| {
                        crate::Error::from(InternalError::RootIdIsNotAUuid { uuid })
                    })?,
                    name: Some(name.to_string()),
                    created: system_time_from_nanos(nanos),
                })
            });

        match query_result {
            Ok(root_result) => Ok(root_result?),
            Err(rusqlite::Error::QueryReturnedNoRows) => Err(crate::Error::RootNotFound {
                root: name.to_string().into(),
            }),
            Err(err) => Err(err.into()),
        }
    }

    /// Create a new root without creating the root directory.
    pub fn insert_root_record(&self, name: Option<&str>) -> crate::Result<Root> {
        let root_id = RootId::new();

        let mut stmt = self.db.prepare_cached(
            r#"
            INSERT OR IGNORE INTO
                liteboxfs_roots (uuid, name)
            VALUES
                (?, ?)
            RETURNING
                uuid,
                created;
            "#,
        )?;

        let params = rusqlite::params![root_id.to_string(), name];

        let root = stmt
            .query_one(params, |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, i128>(1)?))
            })
            .optional()?
            .map(|(uuid, nanos)| {
                crate::Result::Ok(Root {
                    id: uuid.parse()?,
                    name: name.map(|s| s.to_string()),
                    created: system_time_from_nanos(nanos),
                })
            })
            .transpose()?;

        match (name, root) {
            (Some(name), None) => Err(crate::Error::RootAlreadyExists {
                name: name.to_string(),
            }),
            (None, None) => unreachable!(),
            (_, Some(root)) => Ok(root),
        }
    }

    pub fn insert_root(&self, name: Option<&str>) -> crate::Result<Root> {
        let root = self.insert_root_record(name)?;

        let root_dir_metadata =
            FileMetadata::for_new_file(FileDiscriminant::Dir, FileMode::OTHER_W, Owner::ROOT);

        // Create the root directory.
        self.create_file(
            root.id,
            &NormalizedPath::new("/"),
            &FileKind::Dir,
            root_dir_metadata,
        )?;

        Ok(root)
    }

    pub fn get_roots_page(&self, limit: u32, offset: u32) -> crate::Result<Vec<Root>> {
        let mut stmt = self.db.prepare_cached(
            r#"
                SELECT
                    uuid,
                    name,
                    created
                FROM
                    liteboxfs_roots
                ORDER BY
                    id
                LIMIT
                    ?
                OFFSET
                    ?;
            "#,
        )?;

        let params = rusqlite::params![limit, offset];

        stmt.query_map(params, |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, Option<String>>(1)?,
                row.get::<_, i128>(2)?,
            ))
        })?
        .map(|result| {
            let (uuid, name, created) = result?;

            Ok(Root {
                id: uuid
                    .parse()
                    .map_err(|_| crate::Error::from(InternalError::RootIdIsNotAUuid { uuid }))?,
                name,
                created: system_time_from_nanos(created),
            })
        })
        .collect::<Result<Vec<_>, crate::Error>>()
    }

    pub fn copy_file(
        &self,
        from_root: RootId,
        from_path: &NormalizedPath,
        to_root: RootId,
        to_path: &NormalizedPath,
    ) -> crate::Result<()> {
        use crate::block::FileId as StoreFileId;

        let (source_file_row_id, kind, metadata) = self.get_file(from_path.into(), from_root)?;

        let new_file_id: StoreFileId = {
            let mut stmt = self.db.prepare_cached(
                r#"
                INSERT INTO
                    liteboxfs_files (kind, mode, atime, mtime, ctime, btime, uid, gid, major, minor, target)
                VALUES
                    (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                RETURNING
                    id;
                "#,
            )?;

            stmt.query_row(
                rusqlite::params![
                    metadata.discriminant,
                    metadata.mode.bits(),
                    metadata.atime,
                    metadata.mtime,
                    metadata.ctime,
                    metadata.btime,
                    metadata.uid.as_raw(),
                    metadata.gid.as_raw(),
                    match &kind {
                        FileKind::Block { dev } | FileKind::Char { dev } => Some(dev.major()),
                        _ => None,
                    },
                    match &kind {
                        FileKind::Block { dev } | FileKind::Char { dev } => Some(dev.minor()),
                        _ => None,
                    },
                    match &kind {
                        FileKind::Symlink { target } => Some(target.as_os_str().as_encoded_bytes()),
                        _ => None,
                    },
                ],
                |row| row.get(0),
            )?
        };

        if self.insert_path(to_root, to_path, new_file_id)?
            == crate::sql::path::PathInsertResult::AlreadyExists
        {
            return Err(crate::Error::FileAlreadyExists {
                path: FileOrigin::Litebox {
                    root: to_root,
                    locator: to_path.to_path_buf(),
                },
            });
        }

        self.copy_file_blocks(source_file_row_id, new_file_id)?;
        self.copy_merkle_nodes(source_file_row_id, new_file_id)?;
        self.copy_xattrs(source_file_row_id, new_file_id)?;

        Ok(())
    }

    /// Copy liteboxfs_file_blocks for a single file.
    pub(super) fn copy_file_blocks(&self, from: StoreFileId, to: StoreFileId) -> crate::Result<()> {
        let mut stmt = self.db.prepare_cached(
            r#"
            INSERT INTO
                liteboxfs_file_blocks (file, "index", block)
            SELECT
                ?,
                "index",
                block
            FROM
                liteboxfs_file_blocks
            WHERE
                file = ?;
            "#,
        )?;

        stmt.execute(rusqlite::params![to, from])?;

        Ok(())
    }

    /// Copy liteboxfs_merkle_nodes for a single file.
    pub(super) fn copy_merkle_nodes(
        &self,
        from: StoreFileId,
        to: StoreFileId,
    ) -> crate::Result<()> {
        let mut stmt = self.db.prepare_cached(
            r#"
            INSERT INTO
                liteboxfs_merkle_nodes (file, level, "index", hash, dirty)
            SELECT
                ?,
                level,
                "index",
                hash,
                dirty
            FROM
                liteboxfs_merkle_nodes
            WHERE
                file = ?;
            "#,
        )?;

        stmt.execute(rusqlite::params![to, from])?;

        Ok(())
    }

    /// Copy liteboxfs_xattrs for a single file.
    pub(super) fn copy_xattrs(&self, from: StoreFileId, to: StoreFileId) -> crate::Result<()> {
        let mut stmt = self.db.prepare_cached(
            r#"
            INSERT INTO
                liteboxfs_xattrs (file, name, value)
            SELECT
                ?,
                name,
                value
            FROM
                liteboxfs_xattrs
            WHERE
                file = ?;
            "#,
        )?;

        stmt.execute(rusqlite::params![to, from])?;

        Ok(())
    }

    /// Copy all paths from one root to another using COW semantics.
    ///
    /// File metadata rows are shared between the two roots; a private copy is made on first
    /// mutation via [`SqlStore::materialize_file`]. A single INSERT with ID offset arithmetic
    /// wires up parent pointers correctly with no temp tables or follow-up UPDATE.
    pub fn copy_root(&self, from: RootId, to: RootId) -> crate::Result<()> {
        let from_root_row_id: i64 = self.db.query_row(
            "SELECT id FROM liteboxfs_roots WHERE uuid = ?",
            rusqlite::params![from.to_string()],
            |row| row.get(0),
        )?;

        let to_root_row_id: i64 = self.db.query_row(
            "SELECT id FROM liteboxfs_roots WHERE uuid = ?",
            rusqlite::params![to.to_string()],
            |row| row.get(0),
        )?;

        // Copy all paths in one statement. The offset arithmetic keeps parent pointers correct:
        // new_id = old_id - min_source_id + new_start, and the same shift applies to parent
        // references, so no secondary UPDATE is needed. ORDER BY id ASC guarantees parents are
        // inserted before their children, satisfying the FK constraint.
        self.db.execute(
            r#"
            WITH source_info AS (
                SELECT
                    MIN(id) AS min_id,
                    COALESCE((SELECT MAX(id) FROM liteboxfs_paths), 0) + 1 AS new_start
                FROM
                    liteboxfs_paths
                WHERE
                    root = ?1
            )
            INSERT INTO
                liteboxfs_paths (id, file, root, name, parent)
            SELECT
                liteboxfs_paths.id - source_info.min_id + source_info.new_start,
                liteboxfs_paths.file,
                ?2,
                liteboxfs_paths.name,
                CASE
                    WHEN liteboxfs_paths.parent IS NULL THEN NULL
                    ELSE liteboxfs_paths.parent - source_info.min_id + source_info.new_start
                END
            FROM
                liteboxfs_paths, source_info
            WHERE
                liteboxfs_paths.root = ?1
            ORDER BY
                liteboxfs_paths.id ASC;
            "#,
            rusqlite::params![from_root_row_id, to_root_row_id],
        )?;

        // Copy root metadata.
        self.db.execute(
            r#"
            INSERT INTO
                liteboxfs_root_metadata (root, "key", value)
            SELECT
                ?1, "key", value
            FROM
                liteboxfs_root_metadata
            WHERE
                root = ?2;
            "#,
            rusqlite::params![to_root_row_id, from_root_row_id],
        )?;

        Ok(())
    }

    /// Copy a directory tree from one location to another.
    ///
    /// When copying across roots, file metadata rows are shared via COW semantics: a private copy
    /// is made on first mutation via [`SqlStore::materialize_file`]. When copying within the same
    /// root, a deep copy is performed instead, since COW cannot distinguish between hard links
    /// (which must share state) and copy_tree paths (which must be independent).
    pub fn copy_tree(
        &self,
        from_root: RootId,
        from_path: &NormalizedPath,
        to_root: RootId,
        to_path: &NormalizedPath,
    ) -> crate::Result<()> {
        let from_root_row_id: i64 = self.db.query_row(
            "SELECT id FROM liteboxfs_roots WHERE uuid = ?",
            rusqlite::params![from_root.to_string()],
            |row| row.get(0),
        )?;

        let to_root_row_id: i64 = self.db.query_row(
            "SELECT id FROM liteboxfs_roots WHERE uuid = ?",
            rusqlite::params![to_root.to_string()],
            |row| row.get(0),
        )?;

        let source_ancestor_path_id =
            self.resolve_path_id(from_root_row_id, from_root, from_path)?;

        let dest_exists = match self.resolve_path_id(to_root_row_id, to_root, to_path) {
            Ok(_) => true,
            Err(crate::Error::FileNotFound { .. }) => false,
            Err(e) => return Err(e),
        };

        if dest_exists {
            return Err(crate::Error::FileAlreadyExists {
                path: FileOrigin::Litebox {
                    root: to_root,
                    locator: to_path.to_path_buf(),
                },
            });
        }

        let dest_parent = to_path
            .parent()
            .map(NormalizedPath::new)
            .expect("dest path should have a parent");

        let dest_parent_path_id = self
            .resolve_path_id(to_root_row_id, to_root, &dest_parent)
            .map_err(|e| match e {
                crate::Error::FileNotFound { .. } => crate::Error::NoParentDirectory {
                    file: FileOrigin::Litebox {
                        root: to_root,
                        locator: to_path.to_path_buf().into(),
                    },
                },
                _ => e,
            })?;

        let dest_parent_kind: FileDiscriminant = self.db.query_row(
            r#"
            SELECT
                liteboxfs_files.kind
            FROM
                liteboxfs_files
            JOIN
                liteboxfs_paths ON liteboxfs_files.id = liteboxfs_paths.file
            WHERE
                liteboxfs_paths.id = ?;
            "#,
            rusqlite::params![dest_parent_path_id],
            |row| row.get(0),
        )?;

        if dest_parent_kind != FileDiscriminant::Dir {
            return Err(crate::Error::NotADirectory {
                file: FileOrigin::Litebox {
                    root: to_root,
                    locator: dest_parent.to_path_buf().into(),
                },
            });
        }

        let dest_name = to_path
            .file_name()
            .expect("dest path should have a file name");

        if from_root == to_root {
            self.copy_tree_deep(
                source_ancestor_path_id,
                to_root_row_id,
                dest_name.as_encoded_bytes(),
                dest_parent_path_id,
            )
        } else {
            self.copy_tree_cow(
                source_ancestor_path_id,
                to_root_row_id,
                dest_name.as_encoded_bytes(),
                dest_parent_path_id,
            )
        }
    }

    /// Copy a subtree using COW semantics (cross-root).
    ///
    /// A single INSERT with a recursive CTE and ID offset arithmetic inserts all paths with
    /// correct parent pointers in one pass.
    fn copy_tree_cow(
        &self,
        source_path_id: i64,
        to_root_row_id: i64,
        dest_name: &[u8],
        dest_parent_path_id: i64,
    ) -> crate::Result<()> {
        // The offset arithmetic shifts each source path id into a disjoint range starting at
        // `new_start` so that `new_id = old_id - min_id + new_start`, applied identically to
        // parent pointers. `new_start` and `min_id` are computed separately and bound as
        // parameters rather than referenced from the bulk INSERT via a CTE: referencing the
        // recursive subtree CTE more than once in the outer query causes SQLite to
        // re-evaluate the recursion per reference, and combined with an
        // `id IN (SELECT id FROM subtree)` filter over `liteboxfs_paths` it degenerates to
        // O(N^2) subtree walks.
        let new_start: i64 = self.db.query_row(
            r#"
            SELECT
                COALESCE(MAX(id), 0) + 1
            FROM
                liteboxfs_paths;
            "#,
            [],
            |row| row.get(0),
        )?;

        let min_id: i64 = self.db.query_row(
            r#"
            WITH RECURSIVE subtree(id) AS (
                SELECT ?1
                UNION ALL
                SELECT
                    liteboxfs_paths.id
                FROM
                    liteboxfs_paths
                JOIN
                    subtree ON liteboxfs_paths.parent = subtree.id
            )
            SELECT
                MIN(id)
            FROM
                subtree;
            "#,
            rusqlite::params![source_path_id],
            |row| row.get(0),
        )?;

        // The recursive CTE walks the subtree once and carries every column the INSERT
        // needs, so the outer SELECT drives directly from `subtree` and the recursion is
        // referenced exactly once. ORDER BY subtree.id ASC guarantees parents are inserted
        // before their children, satisfying the FK constraint.
        self.db.execute(
            r#"
            WITH RECURSIVE subtree(id, file, name, parent) AS (
                SELECT
                    liteboxfs_paths.id,
                    liteboxfs_paths.file,
                    liteboxfs_paths.name,
                    liteboxfs_paths.parent
                FROM
                    liteboxfs_paths
                WHERE
                    liteboxfs_paths.id = ?1

                UNION ALL

                SELECT
                    liteboxfs_paths.id,
                    liteboxfs_paths.file,
                    liteboxfs_paths.name,
                    liteboxfs_paths.parent
                FROM
                    liteboxfs_paths
                JOIN
                    subtree ON liteboxfs_paths.parent = subtree.id
            )
            INSERT INTO
                liteboxfs_paths (id, file, root, name, parent)
            SELECT
                subtree.id - ?5 + ?6,
                subtree.file,
                ?2,
                CASE WHEN subtree.id = ?1 THEN ?3 ELSE subtree.name END,
                CASE
                    WHEN subtree.id = ?1 THEN ?4
                    ELSE subtree.parent - ?5 + ?6
                END
            FROM
                subtree
            ORDER BY
                subtree.id ASC;
            "#,
            rusqlite::params![
                source_path_id,
                to_root_row_id,
                dest_name,
                dest_parent_path_id,
                min_id,
                new_start,
            ],
        )?;

        Ok(())
    }

    /// Copy a subtree using deep copy semantics (within-root).
    ///
    /// Each file in the subtree gets its own new `liteboxfs_files` row, so the copy is fully
    /// independent of the source from the start. COW cannot be used within a root because
    /// `materialize_file` updates all paths in a root for a given file_id, which is the correct
    /// behavior for hard links but would clobber the original when an independent copy is mutated.
    fn copy_tree_deep(
        &self,
        source_path_id: i64,
        to_root_row_id: i64,
        dest_name: &[u8],
        dest_parent_path_id: i64,
    ) -> crate::Result<()> {
        // Collect the full subtree in topological order (parents before children).
        let nodes: Vec<(i64, StoreFileId, Vec<u8>, Option<i64>)> = {
            let mut stmt = self.db.prepare_cached(
                r#"
                WITH RECURSIVE subtree(id) AS (
                    SELECT ?1
                    UNION ALL
                    SELECT
                        liteboxfs_paths.id
                    FROM
                        liteboxfs_paths
                    JOIN
                        subtree ON liteboxfs_paths.parent = subtree.id
                )
                SELECT
                    liteboxfs_paths.id,
                    liteboxfs_paths.file,
                    liteboxfs_paths.name,
                    liteboxfs_paths.parent
                FROM
                    liteboxfs_paths
                WHERE
                    liteboxfs_paths.id IN (SELECT id FROM subtree)
                ORDER BY
                    liteboxfs_paths.id ASC;
                "#,
            )?;

            stmt.query_map(rusqlite::params![source_path_id], |row| {
                Ok((
                    row.get::<_, i64>(0)?,
                    StoreFileId::from(row.get::<_, i64>(1)?),
                    row.get::<_, Vec<u8>>(2)?,
                    row.get::<_, Option<i64>>(3)?,
                ))
            })?
            .collect::<Result<Vec<_>, _>>()?
        };

        // Deep copy each unique file_id. Multiple paths (hard links) may share a file_id; we
        // preserve that relationship in the copy by mapping them all to the same new file_id.
        let mut file_id_map: HashMap<StoreFileId, StoreFileId> = HashMap::new();

        let mut stmt = self.db.prepare_cached(
            r#"
            INSERT INTO
                liteboxfs_files (kind, mode, atime, mtime, ctime, btime, uid, gid, major, minor, target)
            SELECT
                kind, mode, atime, mtime, ctime, btime, uid, gid, major, minor, target
            FROM
                liteboxfs_files
            WHERE
                id = ?
            RETURNING
                id;
            "#,
        )?;

        for (_, file_id, _, _) in &nodes {
            if file_id_map.contains_key(file_id) {
                continue;
            }

            let new_file_id: StoreFileId = stmt.query_row(rusqlite::params![file_id], |row| {
                Ok(StoreFileId::from(row.get::<_, i64>(0)?))
            })?;

            self.copy_file_blocks(*file_id, new_file_id)?;
            self.copy_merkle_nodes(*file_id, new_file_id)?;
            self.copy_xattrs(*file_id, new_file_id)?;

            file_id_map.insert(*file_id, new_file_id);
        }

        // Insert new paths in topological order, tracking old→new path_id for parent fixup.
        let mut path_id_map: HashMap<i64, i64> = HashMap::new();

        for (old_path_id, old_file_id, name, old_parent) in &nodes {
            let new_file_id = *file_id_map
                .get(old_file_id)
                .expect("file ID was inserted in the previous loop");

            let effective_name: &[u8] = if *old_path_id == source_path_id {
                dest_name
            } else {
                name.as_slice()
            };

            let new_parent: Option<i64> = if *old_path_id == source_path_id {
                Some(dest_parent_path_id)
            } else {
                old_parent.map(|p| {
                    *path_id_map
                        .get(&p)
                        .expect("parent was inserted before child due to ORDER BY id ASC")
                })
            };

            let new_path_id: i64 = {
                let mut stmt = self.db.prepare_cached(
                    r#"
                    INSERT INTO
                        liteboxfs_paths (file, root, name, parent)
                    VALUES
                        (?, ?, ?, ?)
                    RETURNING
                        id;
                    "#,
                )?;

                stmt.query_one(
                    rusqlite::params![new_file_id, to_root_row_id, effective_name, new_parent],
                    |row| row.get(0),
                )?
            };

            path_id_map.insert(*old_path_id, new_path_id);
        }

        Ok(())
    }

    pub fn delete_root(&self, id: RootId) -> crate::Result<()> {
        let mut stmt = self.db.prepare_cached(
            r#"
                DELETE FROM
                    liteboxfs_roots
                WHERE
                    uuid = ?;
            "#,
        )?;

        let rows_deleted = stmt.execute(rusqlite::params![id])?;

        if rows_deleted == 0 {
            return Err(crate::Error::RootNotFound { root: id.into() });
        }

        // The root deletion cascades through liteboxfs_paths, but liteboxfs_files rows are not
        // tied to roots directly. Clean up any files that are now unreferenced.
        self.delete_all_unlinked_files()?;

        Ok(())
    }
}