arr/crates/arr-api/src/relocate.rs

//! Moving library files when the layout under them changes: a title changing
//! its `root_id` (issue #228), and a root changing its `path` (issue #236).
//! Both rename §7.4 folders and rewrite the `media_files` and `subtitle_files`
//! rows to match, so the layout keeps describing the disk. Sidecars (§15) live
//! inside the same folder as the video, so the rename already carries them —
//! only their rows need rewriting (issue #237). This module only moves
//! things: the stored verdicts a root change invalidates (§5.1) are
//! re-derived by the calling handler through `reclassify`, after the row
//! commits.
//!
//! Every root shares one filesystem — one ZFS dataset, bind-mounted — so this
//! is a directory rename, never a copy. Hardlinked files keep their inodes
//! and the torrent keeps seeding against them (§7.3).
//!
//! Ordering mirrors `remove_library_files`: the disk is touched before the
//! row changes, so a failed rename leaves the title where its files actually
//! are and the operator can retry.
//!
//! A root path change is the same move repeated over every title under the
//! root, and it is all or nothing. If the seventh of ten folders fails to
//! move, the six already renamed are moved back and the root row is left
//! alone: a root row carries one path, so a half-moved library would have to
//! describe both, and neither the operator nor the next import could tell
//! which titles were where. Refusing leaves one answer — everything is still
//! at the old path — and the retry is the same request again. A retry after
//! an undo that itself failed still converges, because a source that is no
//! longer on disk is skipped while its row is still rewritten.

use std::path::PathBuf;

use crate::movies::{pool, title_target, ApiError};
use crate::state::AppState;

/// Which table owns the moving title's files.
#[derive(Debug, Clone, Copy)]
pub(crate) enum TitleKind {
    Movie,
    Series,
}

/// Which table a rewritten row belongs to — `rewrite_rows` needs this to
/// target the right `UPDATE`, since a `media_files` id and a `subtitle_files`
/// id share no namespace.
#[derive(Debug, Clone, Copy)]
enum FileTable {
    Media,
    Subtitle,
}

/// One rename from the old root into the new one: a §7.4 title folder, or a
/// loose file sitting straight in the root.
#[derive(Debug)]
struct PlannedRename {
    source: PathBuf,
    destination: PathBuf,
}

/// The renames already performed on disk and the row rewrites they imply.
/// The database half is the caller's transaction; [`Self::undo`] is for when
/// that transaction fails after the disk already changed.
#[derive(Debug)]
pub(crate) struct Relocation {
    performed: Vec<PlannedRename>,
    rewrites: Vec<(FileTable, i64, String)>,
    /// Every directory level this request materialised for the new root,
    /// deepest first. `create_dir_all` can make more than one — moving a
    /// root to `/mnt/media-v2/tv/kids` when `/mnt/media-v2` is all that
    /// exists creates both `tv` and `kids` — and [`Self::undo`] removes all
    /// of them, so a failed move leaves nothing behind (#244). Levels that
    /// were already on disk are never in this list and are never touched. A
    /// move that commits never calls `undo` and the directories stay.
    created_dirs: Vec<PathBuf>,
}

/// Rename the title's folders into the new root. Called before the row is
/// written, and only when the root actually changes.
///
/// A title with nothing on disk — no `media_files` rows, or rows whose
/// targets are already gone — changes root with no filesystem work at all.
///
/// # Errors
///
/// [`ApiError::Conflict`] when the destination already holds an entry of the
/// same name — a conflict, never an overwrite. [`ApiError::Filesystem`] when
/// a rename fails; whatever had already been renamed is moved back first, so
/// the row the caller then leaves unchanged still describes the disk.
pub(crate) async fn relocate_title(
    state: &AppState,
    kind: TitleKind,
    title_id: i64,
    old_root_id: i64,
    new_root_id: i64,
) -> Result<Relocation, ApiError> {
    let old_root = root_path(state, old_root_id).await?;
    let new_root = root_path(state, new_root_id).await?;
    let files = title_files(state, kind, title_id).await?;
    relocate_files(&files, &old_root, &new_root).await
}

/// Rename every title folder under a root into the root's new path, for a
/// `PUT /api/roots/{id}` that changes `path` (issue #236).
///
/// A root with no titles under it — or whose titles have nothing on disk —
/// changes path with no filesystem work at all. The new path is created when
/// there is something to move into it, since a root is normally pointed at a
/// directory that does not exist yet.
///
/// # Errors
///
/// The same two as [`relocate_title`], with the whole root's move treated as
/// one unit: one folder that cannot move takes the entire change down and
/// moves back whatever had already moved.
pub(crate) async fn relocate_root(
    state: &AppState,
    root_id: i64,
    old_path: &str,
    new_path: &str,
) -> Result<Relocation, ApiError> {
    let files = root_files(state, root_id).await?;
    relocate_files(&files, old_path, new_path).await
}

/// The one mover both callers share: plan every rename, refuse every
/// destination that already exists, then perform them, undoing what was
/// performed if one fails.
async fn relocate_files(
    files: &[(FileTable, i64, String)],
    old_root: &str,
    new_root: &str,
) -> Result<Relocation, ApiError> {
    let mut renames: Vec<PlannedRename> = Vec::new();
    let mut rewrites: Vec<(FileTable, i64, String)> = Vec::new();
    for (table, file_id, path) in files {
        let Some(source) = title_target(old_root, path) else {
            // Outside its own root: not ours to move, and the row keeps
            // pointing at where the file really is.
            tracing::warn!(%path, %old_root, "media file is outside its root, not moved");
            continue;
        };
        let Some(name) = source.file_name() else {
            continue;
        };
        let destination = std::path::Path::new(new_root).join(name);
        if !renames.iter().any(|rename| rename.source == source) {
            renames.push(PlannedRename {
                source,
                destination,
            });
        }
        let relative = std::path::Path::new(path)
            .strip_prefix(std::path::Path::new(old_root))
            .map_err(|error| ApiError::Filesystem(error.to_string()))?;
        let rewritten = std::path::Path::new(new_root).join(relative);
        let Some(rewritten) = rewritten.to_str() else {
            return Err(ApiError::Filesystem(format!(
                "non-UTF-8 path under {new_root}"
            )));
        };
        rewrites.push((*table, *file_id, rewritten.to_owned()));
    }

    // Every destination is checked before anything is renamed, so a conflict
    // never leaves a half-moved library behind. A rename whose source is
    // already gone is skipped here rather than refused: the perform loop below
    // skips it too, and refusing it would make the documented retry
    // impossible. After an undo that itself failed, the folder sits at the
    // destination while the row still names the source, and every later
    // attempt would 409 on a conflict with the operator's own half-moved
    // library.
    for rename in &renames {
        match tokio::fs::symlink_metadata(&rename.source).await {
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
            _ => {}
        }
        match tokio::fs::symlink_metadata(&rename.destination).await {
            Ok(_) => {
                return Err(ApiError::Conflict(format!(
                    "the destination already has '{}'",
                    rename.destination.display()
                )))
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(ApiError::Filesystem(error.to_string())),
        }
    }

    // Tracked only for the levels this call is the one to create, so a
    // failed move can remove them again without ever touching a directory
    // that already existed on disk. Recorded before `create_dir_all`, since
    // afterwards there is no way to tell which levels it made.
    let mut created_dirs: Vec<PathBuf> = Vec::new();
    // A configured root need not exist on disk. Without creating it here,
    // `rename` reports its missing parent as the same `NotFound` as a missing
    // source and sends the operator looking at the wrong end of the move.
    if !renames.is_empty() {
        created_dirs = missing_levels(std::path::Path::new(new_root)).await;
        if let Err(error) = tokio::fs::create_dir_all(new_root).await {
            return Err(ApiError::Filesystem(format!(
                "could not create '{new_root}': {error}"
            )));
        }
    }

    let mut performed: Vec<PlannedRename> = Vec::new();
    for rename in renames {
        // A recorded file with nothing on disk: the rows still follow the
        // title, the same way a delete treats already-gone as done. Checked
        // on the source, so a missing *destination* parent stays an error.
        match tokio::fs::symlink_metadata(&rename.source).await {
            Ok(_) => {}
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                tracing::warn!(
                    source = %rename.source.display(),
                    "nothing on disk to move for this title"
                );
                continue;
            }
            Err(error) => return Err(failed(&rename, &error, performed, created_dirs).await),
        }
        match tokio::fs::rename(&rename.source, &rename.destination).await {
            Ok(()) => {
                tracing::info!(
                    source = %rename.source.display(),
                    destination = %rename.destination.display(),
                    "moved a title folder"
                );
                performed.push(rename);
            }
            Err(error) => return Err(failed(&rename, &error, performed, created_dirs).await),
        }
    }

    Ok(Relocation {
        performed,
        rewrites,
        created_dirs,
    })
}

/// The levels of `path` that are not on disk, deepest first — exactly what a
/// following `create_dir_all` will materialise. The walk stops at the first
/// ancestor that exists, so nothing already there is ever listed.
async fn missing_levels(path: &std::path::Path) -> Vec<PathBuf> {
    let mut missing = Vec::new();
    for ancestor in path.ancestors() {
        // `ancestors` ends in an empty path for a relative input; there is
        // no level above that to create.
        if ancestor.as_os_str().is_empty() || tokio::fs::symlink_metadata(ancestor).await.is_ok() {
            break;
        }
        missing.push(ancestor.to_path_buf());
    }
    missing
}

/// One rename failed: move back everything that had already moved, remove
/// every directory level this request created, and name the folder that
/// stopped the move so the operator knows which title to look at before
/// retrying.
async fn failed(
    rename: &PlannedRename,
    error: &std::io::Error,
    performed: Vec<PlannedRename>,
    created_dirs: Vec<PathBuf>,
) -> ApiError {
    Relocation {
        performed,
        rewrites: Vec::new(),
        created_dirs,
    }
    .undo()
    .await;
    // Name both ends. `rename` returns NotFound for a missing destination
    // parent as readily as for a missing source, and naming only the source
    // sent an operator looking for a file that was sitting where they left it.
    ApiError::Filesystem(format!(
        "could not move '{}' to '{}': {error}",
        rename.source.display(),
        rename.destination.display()
    ))
}

impl Relocation {
    /// Whether anything on disk actually moved — the trigger for the same
    /// single Jellyfin refresh import performs (§7.5).
    pub(crate) fn moved_files(&self) -> bool {
        !self.performed.is_empty()
    }

    /// Point the `media_files` and `subtitle_files` rows at the new root,
    /// inside the caller's transaction so they land together with the
    /// `root_id` change or not at all.
    pub(crate) async fn rewrite_rows(
        &self,
        transaction: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
    ) -> Result<(), sqlx::Error> {
        for (table, file_id, path) in &self.rewrites {
            match table {
                FileTable::Media => {
                    sqlx::query!(
                        "UPDATE media_files SET path = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
                        path,
                        file_id
                    )
                    .execute(&mut **transaction)
                    .await?;
                }
                FileTable::Subtitle => {
                    sqlx::query!(
                        "UPDATE subtitle_files SET path = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') WHERE id = ?",
                        path,
                        file_id
                    )
                    .execute(&mut **transaction)
                    .await?;
                }
            }
        }
        Ok(())
    }

    /// Best-effort reversal of the renames, for when the disk moved but the
    /// database write failed. A reversal that itself fails is logged: at that
    /// point the operator's retry is the recovery path.
    pub(crate) async fn undo(&self) {
        for rename in self.performed.iter().rev() {
            if let Err(error) = tokio::fs::rename(&rename.destination, &rename.source).await {
                tracing::error!(
                    source = %rename.source.display(),
                    destination = %rename.destination.display(),
                    %error,
                    "could not move the title folder back after a failed root change"
                );
            }
        }
        // `remove_dir` rather than `remove_dir_all`: it only succeeds on an
        // empty directory, so anything unexpected left inside it — this
        // request's own undo failing, say — is a reason to leave it alone.
        // Deepest first, since a parent cannot go while its child is there;
        // the first level that will not go stops the walk, because every
        // level above it now has content and refusing is the right answer.
        for directory in &self.created_dirs {
            if let Err(error) = tokio::fs::remove_dir(directory).await {
                tracing::warn!(
                    path = %directory.display(),
                    %error,
                    "could not remove a directory created for a move that did not complete"
                );
                break;
            }
        }
    }
}

/// §7.5 after a move: the same single best-effort refresh import performs.
/// Failure logs and never fails the write that already committed.
pub(crate) async fn refresh_jellyfin(state: &AppState) {
    if let Some(jellyfin) = state.jellyfin() {
        if let Err(error) = jellyfin.refresh().await {
            tracing::warn!(%error, "jellyfin refresh after a root change failed");
        }
    }
}

async fn root_path(state: &AppState, root_id: i64) -> Result<String, ApiError> {
    Ok(sqlx::query_scalar!(
        r#"SELECT path AS "path!: String" FROM roots WHERE id = ?"#,
        root_id
    )
    .fetch_one(pool(state)?)
    .await?)
}

/// Every file the service recorded under a root: the `media_files` rows of
/// every movie and episode in it, and the `subtitle_files` sidecars (§15) on
/// those files — an embedded track has no `path` and nothing on disk to
/// carry, so it is excluded rather than rewritten to nothing. Ordered so the
/// renames happen in a stable order, which is what makes a failure part-way
/// through reproducible.
async fn root_files(
    state: &AppState,
    root_id: i64,
) -> Result<Vec<(FileTable, i64, String)>, ApiError> {
    let mut files: Vec<(FileTable, i64, String)> = sqlx::query!(
        r#"SELECT mf.id AS "id!: i64", mf.path AS "path!: String"
           FROM media_files mf
           JOIN movies m ON mf.owner_kind = 'movie' AND m.id = mf.owner_id
           WHERE m.root_id = ?
           ORDER BY mf.id"#,
        root_id
    )
    .fetch_all(pool(state)?)
    .await?
    .into_iter()
    .map(|row| (FileTable::Media, row.id, row.path))
    .collect();
    files.extend(
        sqlx::query!(
            r#"SELECT mf.id AS "id!: i64", mf.path AS "path!: String"
               FROM media_files mf
               JOIN episodes e ON mf.owner_kind = 'episode' AND e.id = mf.owner_id
               JOIN seasons se ON se.id = e.season_id
               JOIN series s ON s.id = se.series_id
               WHERE s.root_id = ?
               ORDER BY mf.id"#,
            root_id
        )
        .fetch_all(pool(state)?)
        .await?
        .into_iter()
        .map(|row| (FileTable::Media, row.id, row.path)),
    );
    files.extend(
        sqlx::query!(
            r#"SELECT sf.id AS "id!: i64", sf.path AS "path!: String"
               FROM subtitle_files sf
               JOIN media_files mf ON mf.id = sf.media_file_id
               JOIN movies m ON mf.owner_kind = 'movie' AND m.id = mf.owner_id
               WHERE m.root_id = ? AND sf.path IS NOT NULL
               ORDER BY sf.id"#,
            root_id
        )
        .fetch_all(pool(state)?)
        .await?
        .into_iter()
        .map(|row| (FileTable::Subtitle, row.id, row.path)),
    );
    files.extend(
        sqlx::query!(
            r#"SELECT sf.id AS "id!: i64", sf.path AS "path!: String"
               FROM subtitle_files sf
               JOIN media_files mf ON mf.id = sf.media_file_id
               JOIN episodes e ON mf.owner_kind = 'episode' AND e.id = mf.owner_id
               JOIN seasons se ON se.id = e.season_id
               JOIN series s ON s.id = se.series_id
               WHERE s.root_id = ? AND sf.path IS NOT NULL
               ORDER BY sf.id"#,
            root_id
        )
        .fetch_all(pool(state)?)
        .await?
        .into_iter()
        .map(|row| (FileTable::Subtitle, row.id, row.path)),
    );
    Ok(files)
}

/// Every file the service recorded for the title: a movie's own rows, or the
/// rows of every episode below a series — plus the `subtitle_files` sidecars
/// (§15) on those files. An embedded track has no `path` and nothing on disk
/// to carry, so it is excluded rather than rewritten to nothing.
async fn title_files(
    state: &AppState,
    kind: TitleKind,
    title_id: i64,
) -> Result<Vec<(FileTable, i64, String)>, ApiError> {
    Ok(match kind {
        TitleKind::Movie => {
            let mut files: Vec<(FileTable, i64, String)> = sqlx::query!(
                r#"SELECT id AS "id!: i64", path AS "path!: String"
                   FROM media_files WHERE owner_kind = 'movie' AND owner_id = ?"#,
                title_id
            )
            .fetch_all(pool(state)?)
            .await?
            .into_iter()
            .map(|row| (FileTable::Media, row.id, row.path))
            .collect();
            files.extend(
                sqlx::query!(
                    r#"SELECT sf.id AS "id!: i64", sf.path AS "path!: String"
                       FROM subtitle_files sf
                       JOIN media_files mf ON mf.id = sf.media_file_id
                       WHERE mf.owner_kind = 'movie' AND mf.owner_id = ? AND sf.path IS NOT NULL"#,
                    title_id
                )
                .fetch_all(pool(state)?)
                .await?
                .into_iter()
                .map(|row| (FileTable::Subtitle, row.id, row.path)),
            );
            files
        }
        TitleKind::Series => {
            let mut files: Vec<(FileTable, i64, String)> = sqlx::query!(
                r#"SELECT mf.id AS "id!: i64", mf.path AS "path!: String"
                   FROM media_files mf
                   JOIN episodes e ON mf.owner_kind = 'episode' AND e.id = mf.owner_id
                   JOIN seasons se ON se.id = e.season_id
                   WHERE se.series_id = ?"#,
                title_id
            )
            .fetch_all(pool(state)?)
            .await?
            .into_iter()
            .map(|row| (FileTable::Media, row.id, row.path))
            .collect();
            files.extend(
                sqlx::query!(
                    r#"SELECT sf.id AS "id!: i64", sf.path AS "path!: String"
                       FROM subtitle_files sf
                       JOIN media_files mf ON mf.id = sf.media_file_id
                       JOIN episodes e ON mf.owner_kind = 'episode' AND e.id = mf.owner_id
                       JOIN seasons se ON se.id = e.season_id
                       WHERE se.series_id = ? AND sf.path IS NOT NULL"#,
                    title_id
                )
                .fetch_all(pool(state)?)
                .await?
                .into_iter()
                .map(|row| (FileTable::Subtitle, row.id, row.path)),
            );
            files
        }
    })
}