//! The restore pipeline: a snapshot's manifest is seeded into a durable
//! queue, writer threads recreate the files, and a finalization pass
//! applies directories, symlinks, and directory metadata.
//!
//! ```text
//!   seed: manifest entries ──┬── files ────────> restore queue ──> writers
//!                            └── dirs/symlinks ─> meta queue ─┐
//!                                                             v
//!                     finalization: symlinks, empty dirs, then directory
//!                     modes and mtimes deepest-first (after the contents)
//! ```
//!
//! The same durability rules as the backup pipeline apply, with one
//! restore-specific addition: a restored file's bytes are synced before
//! its queue acknowledgement can become durable — data first, then the
//! claim that it was written — and the parent directory entries are
//! synced at each queue sync for the same reason. Directories and
//! symlinks are not tracked per-item at all; the whole finalization pass
//! redoes itself if interrupted, which makes their durability moot.
//!
//! Restoring is idempotent from a fresh or partially-restored target: a
//! redelivered file entry is simply rewritten. Restoring over unrelated
//! existing content is not supported.

use std::{
    collections::HashSet,
    path::{Path, PathBuf},
    sync::{Arc, Mutex, atomic::AtomicBool},
};

use crossbeam_channel::Sender;
use persistent_queue::{Lease, PersistentQueue};
use repository::{Entry, EntryKind, Repository, SnapshotId};

use super::{
    Control, Error, Event, FINALIZE_REPORT_EVERY, Marker, Outcome, PROGRESS_FILE, ProgressCounters,
    Reporter, SkipReason, Stage, StageBacklog, control_loop, load_progress, path_is_safe,
    peek_paths, report_staffing, send_event, store_state_file, worker_loop,
};

const RESTORE_QUEUE: &str = "restore.queue";
const META_QUEUE: &str = "meta.queue";

pub struct RestoreOptions {
    /// The snapshot to restore.
    pub snapshot: SnapshotId,
    /// Cherry-pick patterns (see [`crate::patterns`]): only entries
    /// matching one of these — or living under a matching directory —
    /// are restored, along with the directory entries above them so
    /// their metadata comes out right. Empty means the whole snapshot.
    pub select: Vec<String>,
    /// The directory to restore into; created if absent, expected to be
    /// fresh (existing files at manifest paths are overwritten).
    pub target: PathBuf,
    /// Directory for the pipeline's durable working state; exclusively
    /// owned by the pipeline, like the backup state directory.
    pub state_dir: PathBuf,
    pub writer_threads: usize,
}

impl RestoreOptions {
    pub fn new(
        snapshot: SnapshotId,
        target: impl Into<PathBuf>,
        state_dir: impl Into<PathBuf>,
    ) -> RestoreOptions {
        RestoreOptions {
            snapshot,
            select: Vec::new(),
            target: target.into(),
            state_dir: state_dir.into(),
            writer_threads: std::thread::available_parallelism().map_or(4, usize::from),
        }
    }
}

struct RestoreContext {
    repository: Arc<Repository>,
    target: PathBuf,
    restore: PersistentQueue<Entry>,
    meta: PersistentQueue<Entry>,
    control: Control,
    events: Sender<Event>,
    /// Directories that gained files since the last sync; their entries
    /// are fsynced before the queue acks are, so no durable ack can
    /// claim a file that could still vanish from its directory.
    touched_dirs: Mutex<std::collections::HashSet<PathBuf>>,
    /// Running totals, persisted at every sync so a resumed run's
    /// progress reporting continues from previous sessions.
    progress: ProgressCounters,
}

impl RestoreContext {
    fn sync_state(&self) -> Result<(), Error> {
        let touched: Vec<PathBuf> = {
            let mut guard = self.touched_dirs.lock().unwrap();
            guard.drain().collect()
        };

        for dir in touched {
            std::fs::File::open(&dir)?.sync_all()?;
        }

        self.meta.sync()?;
        self.restore.sync()?;

        Ok(())
    }

    /// Reports an event, folding it into the persistent progress totals.
    fn emit(&self, event: Event) {
        self.progress.record(&event);
        send_event(&self.events, event);
    }

    /// Reports the run's totals and what is queued at each stage — the
    /// state a display needs that no single work item announces.
    fn report_status(&self) {
        send_event(&self.events, Event::Totals(self.progress.snapshot()));
        send_event(
            &self.events,
            Event::Backlog(vec![
                StageBacklog {
                    stage: Stage::Write,
                    waiting: self.restore.pending() as u64,
                    upcoming: Some(peek_paths(&self.restore, |entry| entry.path)),
                },
                StageBacklog {
                    stage: Stage::Apply,
                    waiting: self.meta.pending() as u64,
                    upcoming: Some(peek_paths(&self.meta, |entry| entry.path)),
                },
            ]),
        );
    }

    fn skip(&self, path: &Path, reason: SkipReason) {
        self.emit(Event::Skipped {
            path: path.to_owned(),
            reason,
        });
    }
}

/// Decides which manifest entries a cherry-picked restore includes.
struct Selection(Option<globset::GlobSet>);

impl Selection {
    fn new(patterns: &[String]) -> Result<Selection, Error> {
        if patterns.is_empty() {
            return Ok(Selection(None));
        }

        Ok(Selection(Some(crate::patterns::compile_set(patterns)?)))
    }

    /// True if the entry itself matches, or an ancestor does — which is
    /// what makes selecting a directory bring its whole subtree.
    fn includes(&self, path: &Path) -> bool {
        let Some(set) = &self.0 else {
            return true;
        };

        crate::patterns::matches_with_ancestors(set, path)
    }
}

/// Marks every proper ancestor of `path` as a directory the restore
/// needs, so selected entries get their parent directories' metadata.
fn require_ancestors(path: &Path, needed: &mut HashSet<PathBuf>) {
    for ancestor in path.ancestors().skip(1) {
        if !ancestor.as_os_str().is_empty() {
            needed.insert(ancestor.to_owned());
        }
    }
}

/// Runs a restore to completion or suspension, mirroring
/// [`backup::run`](super::backup::run)'s resume and suspend behavior.
pub fn run(
    repository: Arc<Repository>,
    options: RestoreOptions,
    events: Sender<Event>,
    suspend: Arc<AtomicBool>,
) -> Result<Outcome, Error> {
    let state_dir = options.state_dir.clone();

    // Selection identity is order-insensitive
    let mut selection_id = options.select.clone();
    selection_id.sort();

    std::fs::create_dir_all(&options.target)?;
    std::fs::create_dir_all(&state_dir)?;

    let target = options.target.canonicalize()?;

    let fresh = match Marker::load(&state_dir)? {
        Some(Marker::Completed { snapshot, .. }) => {
            std::fs::remove_dir_all(&state_dir)?;

            return Ok(Outcome::Completed(snapshot));
        }
        Some(Marker::RestoreRunning {
            target: existing,
            snapshot,
            selection,
        }) => {
            if existing != target || snapshot != options.snapshot || selection != selection_id {
                return Err(Error::StateMismatch {
                    existing,
                    requested: target,
                });
            }

            false
        }
        Some(Marker::BackupRunning { root }) => {
            return Err(Error::StateMismatch {
                existing: root,
                requested: target,
            });
        }
        None => {
            std::fs::remove_dir_all(&state_dir)?;
            std::fs::create_dir_all(&state_dir)?;

            true
        }
    };

    let restore: PersistentQueue<Entry> = PersistentQueue::open(state_dir.join(RESTORE_QUEUE))?;
    let meta: PersistentQueue<Entry> = PersistentQueue::open(state_dir.join(META_QUEUE))?;

    if fresh {
        // Seed both queues from the manifest, make them durable, then
        // commit the initialization by writing the marker. Directory
        // entries are buffered and decided last: a directory belongs in
        // the restore if it is selected itself or if anything selected
        // lives beneath it.
        let selection = Selection::new(&selection_id)?;
        let snapshot = repository.load_snapshot(options.snapshot)?;

        let mut directories = Vec::new();
        let mut needed: HashSet<PathBuf> = HashSet::new();

        for entry in repository.manifest_entries(snapshot.manifest) {
            let entry = entry?;

            match entry.kind {
                EntryKind::Directory => directories.push(entry),
                EntryKind::File { .. } => {
                    if selection.includes(&entry.path) {
                        require_ancestors(&entry.path, &mut needed);
                        restore.put(&entry)?;
                    }
                }
                EntryKind::Symlink { .. } => {
                    if selection.includes(&entry.path) {
                        require_ancestors(&entry.path, &mut needed);
                        meta.put(&entry)?;
                    }
                }
            }
        }

        for directory in &directories {
            if selection.includes(&directory.path) {
                require_ancestors(&directory.path, &mut needed);
            }
        }

        for directory in directories {
            if needed.contains(&directory.path) || selection.includes(&directory.path) {
                meta.put(&directory)?;
            }
        }

        meta.sync()?;
        restore.sync()?;
        Marker::RestoreRunning {
            target: target.clone(),
            snapshot: options.snapshot,
            selection: selection_id,
        }
        .store(&state_dir)?;
    }

    let prior = load_progress(&state_dir);

    if !fresh {
        send_event(&events, Event::Resumed(prior));
    }

    let context = Arc::new(RestoreContext {
        repository,
        target,
        restore,
        meta,
        control: Control::new(),
        events,
        touched_dirs: Mutex::new(std::collections::HashSet::new()),
        progress: ProgressCounters::new(prior),
    });

    let mut workers = Vec::new();
    let writer_threads = options.writer_threads.max(1);

    report_staffing(&context.events, Stage::Write, writer_threads);

    for index in 0..writer_threads {
        let context = context.clone();

        workers.push(
            std::thread::Builder::new()
                .name(format!("writer-{index}"))
                .spawn(move || {
                    worker_loop(
                        &context.restore,
                        &context.control,
                        Reporter::new(&context.events, Stage::Write, index),
                        |lease, reporter| write_one(&context, lease, reporter),
                    )
                })?,
        );
    }

    let sync_all = || {
        store_state_file(&state_dir, PROGRESS_FILE, &context.progress.snapshot())?;
        context.sync_state()
    };

    control_loop(
        &context.control,
        &suspend,
        || context.restore.is_empty(),
        || context.report_status(),
        sync_all,
        // A restore publishes nothing as it goes: what it produces is
        // the tree itself, and the queues are what make it resumable
        || Ok(()),
    );

    for worker in workers {
        if worker.join().is_err() {
            context.control.fail(Error::WorkerPanicked);
        }
    }

    sync_all()?;

    // The writers have stopped, so this sample is the exact final word
    // on what they accomplished
    context.report_status();

    if let Some(error) = context.control.take_failure() {
        return Err(error);
    }

    if !context.control.finished() {
        // A restore publishes nothing: what it has written is simply
        // there, in the target directory
        return Ok(Outcome::Suspended { restorable: None });
    }

    finalize(&context)?;

    Marker::Completed {
        snapshot: options.snapshot,
        // A restore has no tree of its own to fall short of
        finished_early: None,
    }
    .store(&state_dir)?;
    context.meta.sync()?;

    drop(context);
    std::fs::remove_dir_all(&state_dir)?;

    Ok(Outcome::Completed(options.snapshot))
}

/// Writer stage: recreate one file from its chunks.
///
/// Failures reading the *repository* are fatal — a restore that cannot
/// produce recorded data must not quietly succeed. Only entries that
/// could never have been produced by a backup are skipped.
fn write_one(
    context: &RestoreContext,
    lease: Lease<'_, Entry>,
    reporter: &Reporter<'_>,
) -> Result<(), Error> {
    let entry: &Entry = &lease;

    reporter.began(&entry.path);

    let EntryKind::File { chunks, len } = &entry.kind else {
        // Seeding routes everything else to the meta queue
        lease.ack();
        return Ok(());
    };

    if !path_is_safe(&entry.path) {
        context.skip(&entry.path, SkipReason::UnsafePath);
        lease.ack();
        return Ok(());
    }

    let absolute = context.target.join(&entry.path);
    let parent = absolute
        .parent()
        .expect("a joined relative path always has a parent")
        .to_owned();

    std::fs::create_dir_all(&parent)?;

    let mut file = std::fs::File::create(&absolute)?;
    let mut written = 0u64;

    {
        use std::io::Write as _;

        let mut writer = std::io::BufWriter::new(&mut file);

        for id in chunks {
            let chunk = context.repository.load_chunk(*id)?;

            // What came off the backend, give or take each chunk's
            // small envelope of nonce and authentication tag
            send_event(
                &context.events,
                Event::Transferred {
                    bytes: chunk.len() as u64,
                },
            );

            writer.write_all(&chunk)?;
            written += chunk.len() as u64;
            reporter.progressed(written, *len);
        }

        writer.flush()?;
    }

    debug_assert_eq!(written, *len);

    // The file's bytes must be on stable storage before the ack that
    // claims it was restored can be synced
    file.sync_all()?;

    apply_metadata(&file, entry)?;

    context.touched_dirs.lock().unwrap().insert(parent);

    context.emit(Event::Restored {
        path: entry.path.clone(),
        bytes: written,
    });

    lease.ack();

    Ok(())
}

/// Applies symlinks, creates any still-missing (empty) directories, and
/// then sets directory modes and mtimes deepest-first — after all
/// content, so that read-only directory modes and accurate mtimes stick.
///
/// This pass is redone in full if interrupted: its acknowledgements only
/// become durable after the completion marker is written.
fn finalize(context: &RestoreContext) -> Result<(), Error> {
    context.emit(Event::Finalizing);

    let mut directories: Vec<Entry> = Vec::new();

    // Finalization is the one worker this stage ever has, so it reports
    // like any other: the entry it has in hand names the apply row that
    // stood empty while the writers were still going.
    let reporter = Reporter::new(&context.events, Stage::Apply, 0);
    let mut applied = 0u64;

    loop {
        let lease = match context.meta.get()? {
            Some(lease) => lease,
            None => break,
        };

        let entry = lease.ack();

        // A snapshot's directories and symlinks can be numerous; name
        // the entry in hand and say how much of the holding queue is
        // left, starting with the very first one so the row is never
        // silent while finalization runs
        if applied.is_multiple_of(FINALIZE_REPORT_EVERY) {
            reporter.began(&entry.path);
            context.report_status();
        }

        applied += 1;

        if !path_is_safe(&entry.path) {
            context.skip(&entry.path, SkipReason::UnsafePath);
            continue;
        }

        match &entry.kind {
            EntryKind::Directory => {
                std::fs::create_dir_all(context.target.join(&entry.path))?;
                directories.push(entry);
            }
            EntryKind::Symlink { target } => {
                apply_symlink(context, &entry, target)?;
            }
            EntryKind::File { .. } => {
                // Cannot happen: seeding routes files to the restore
                // queue. Tolerate it anyway by leaving the file absent.
                context.skip(&entry.path, SkipReason::Vanished);
            }
        }
    }

    // Children before parents, so parent mtimes and read-only modes are
    // applied after the last write inside them
    directories.sort_by_key(|entry| std::cmp::Reverse(entry.path.components().count()));

    for (position, entry) in directories.iter().enumerate() {
        // Deep trees hold a great many directories; keep naming the one
        // being finished rather than going quiet for the whole pass
        if position as u64 % FINALIZE_REPORT_EVERY == 0 {
            reporter.began(&entry.path);
            context.report_status();
        }

        let dir = std::fs::File::open(context.target.join(&entry.path))?;
        apply_metadata(&dir, entry)?;
    }

    reporter.idle();

    Ok(())
}

/// Recreates one symlink, replacing whatever a previous interrupted
/// finalization may have left at its path.
fn apply_symlink(context: &RestoreContext, entry: &Entry, target: &Path) -> Result<(), Error> {
    let absolute = context.target.join(&entry.path);

    if let Some(parent) = absolute.parent() {
        std::fs::create_dir_all(parent)?;
    }

    match std::fs::symlink_metadata(&absolute) {
        Ok(_) => std::fs::remove_file(&absolute)?,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
        Err(err) => return Err(err.into()),
    }

    #[cfg(unix)]
    std::os::unix::fs::symlink(target, &absolute)?;

    #[cfg(not(unix))]
    {
        let _ = target;
        context.skip(&entry.path, SkipReason::SpecialFile);
    }

    Ok(())
}

/// Sets the recorded permissions and modification time on an open file
/// or directory. Absent metadata (from another platform, or a pre-epoch
/// mtime) is simply not applied.
fn apply_metadata(file: &std::fs::File, entry: &Entry) -> Result<(), Error> {
    #[cfg(unix)]
    if let Some(mode) = entry.mode {
        use std::os::unix::fs::PermissionsExt as _;

        file.set_permissions(std::fs::Permissions::from_mode(mode))?;
    }

    if let Some(mtime) = entry.mtime
        && mtime >= 0
    {
        let modified = std::time::UNIX_EPOCH + std::time::Duration::from_nanos(mtime as u64);
        file.set_modified(modified)?;
    }

    Ok(())
}