//! Remote file operations for the File Manager module (port of `files/remoteFileService.js`).
//!
//! Plain operations use an SFTP channel; operations with `sudo: true` go through the fixed
//! privileged helper. The text editor keeps a baseline per opened document (conflict
//! detection) and saves through prepare → upload temp copy → finalize → cleanup with
//! recovery copies, exactly like 0.1.0.

use super::editor::{remote_editor, Authorization};
use super::pipeline::{pump, Consumer, LocalFile, Producer, PumpStop};
use super::posix;
use crate::core::error::{CoreError, CoreResult};
use crate::core::logging::Logger;
use crate::core::ssh::privileged::PrivilegedExecutor;
use crate::core::ssh::session::SessionManager;
use crate::core::ssh::sftp::{FileHandle, Sftp, SftpError, Stat};
use parking_lot::Mutex;
use rand::RngCore;
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

pub const MAX_TEXT_BYTES: usize = 25 * 1024 * 1024;
pub const MAX_TRANSFER_BYTES: u64 = 16 * 1024 * 1024 * 1024;
pub const MAX_BATCH_ENTRIES: u64 = 10_000;
pub const MAX_SELECTED_ENTRIES: usize = 100;
const DEFINITE_EDITOR_FAILURES: [&str; 31] = [
    "FILE_CONFLICT", "FILE_PERMISSION_DENIED", "FILE_NOT_FOUND", "FILE_UNSAFE_PATH", "FILE_UNSAFE_TYPE_OR_LINK", "FILE_UNSAFE_WORKSPACE",
    "FILE_UNSAFE_PRIVILEGED_DIRECTORY", "FILE_UPLOAD_INCOMPLETE", "FILE_TOO_LARGE", "FILE_METADATA_UNSUPPORTED", "FILE_REMOTE_IO_FAILED",
    "FILE_SUDO_FAILED", "FILE_DEPENDENCY_UNAVAILABLE", "FILE_SUDO_PASSWORD_INVALID", "FILE_SUDO_AUTHORIZATION_REQUIRED", "FILE_LINUX_REQUIRED",
    "FILE_HARDLINK_UNSUPPORTED", "FILE_METADATA_PERMISSION", "FILE_NO_SPACE", "FILE_READ_ONLY_FILESYSTEM", "FILE_SYMLINK_TARGET_MISSING",
    "FILE_SYMLINK_LOOP", "FILE_BUSY", "SUDO_PASSWORD_REQUIRED", "SUDO_PASSWORD_INCORRECT", "SUDO_NOT_PERMITTED", "SUDO_TTY_REQUIRED",
    "SUDO_AUTH_FAILED", "SUDO_UNAVAILABLE", "SUDO_FAILED", "SUDO_PASSWORD_INVALID",
];
/// Removing these directly (even with sudo) is refused; their content can still be managed.
pub const PROTECTED_PATHS: [&str; 23] = [
    "/", "/bin", "/boot", "/dev", "/etc", "/home", "/lib", "/lib32", "/lib64", "/libx32", "/media", "/mnt", "/opt", "/proc", "/root", "/run",
    "/sbin", "/snap", "/srv", "/sys", "/tmp", "/usr", "/var",
];

pub fn file_error(code: &str, message: &str) -> CoreError {
    CoreError::new(code, format!("{code}: {message}"))
}

fn random_hex(bytes: usize) -> String {
    let mut buffer = vec![0u8; bytes];
    rand::rngs::OsRng.fill_bytes(&mut buffer);
    hex::encode(buffer)
}

pub fn normalize_remote_path(value: &Value) -> CoreResult<String> {
    let raw = value.as_str().map(str::to_string).unwrap_or_else(|| if value.is_null() { String::new() } else { value.to_string() });
    let raw = raw.trim();
    if raw.is_empty() || raw.chars().count() > 4096 || !raw.starts_with('/') || raw.contains(['\0', '\r', '\n']) {
        return Err(CoreError::plain("Remote path must be an absolute path."));
    }
    Ok(posix::normalize(raw))
}

pub fn safe_entry_name(value: &str) -> CoreResult<String> {
    if value.is_empty() || value == "." || value == ".." || value.contains(['/', '\\', '\0']) {
        return Err(CoreError::plain("Remote entry name cannot be represented safely on the local filesystem."));
    }
    Ok(value.to_string())
}

/// `remotePath || "/"`.
pub fn or_root(value: &Value) -> Value {
    match value {
        Value::Null => json!("/"),
        Value::String(s) if s.is_empty() => json!("/"),
        Value::Bool(false) => json!("/"),
        other => other.clone(),
    }
}

fn iso_from_secs(seconds: u64) -> Value {
    if seconds == 0 {
        return Value::Null;
    }
    chrono::DateTime::<chrono::Utc>::from_timestamp(seconds as i64, 0).map(|t| json!(t.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))).unwrap_or(Value::Null)
}

fn sort_entries(entries: &mut [Value]) {
    let order = |t: &Value| match t.as_str() {
        Some("directory") => 0,
        Some("file") => 1,
        _ => 2,
    };
    entries.sort_by(|a, b| {
        order(&a["type"]).cmp(&order(&b["type"])).then_with(|| {
            let (x, y) = (a["name"].as_str().unwrap_or(""), b["name"].as_str().unwrap_or(""));
            x.to_lowercase().cmp(&y.to_lowercase()).then_with(|| x.cmp(y))
        })
    });
}

#[derive(Debug, Clone)]
struct Edit {
    target: String,
    identity: String,
    owner: String,
    baseline: Value,
    resolved_path: Option<String>,
    busy: bool,
    uncertain: bool,
    dispose: bool,
}

pub struct RemoteFileService {
    sessions: Arc<SessionManager>,
    privileged: Arc<PrivilegedExecutor>,
    logger: Arc<Logger>,
    edits: Mutex<HashMap<String, Edit>>,
    legacy_edits: Mutex<HashMap<String, Option<String>>>,
}

#[derive(Default)]
struct BatchState {
    entries: u64,
    bytes: u64,
}

impl BatchState {
    fn count(&mut self, bytes: u64) -> CoreResult<()> {
        self.entries += 1;
        self.bytes += bytes;
        if self.entries > MAX_BATCH_ENTRIES {
            return Err(CoreError::plain(format!("Transfer exceeds the {MAX_BATCH_ENTRIES} entry limit.")));
        }
        if self.bytes > MAX_TRANSFER_BYTES {
            return Err(CoreError::plain("Transfer exceeds the aggregate size limit."));
        }
        Ok(())
    }
}

fn sftp_plain(error: SftpError) -> CoreError {
    error.file_error("")
}

fn io_plain(error: std::io::Error) -> CoreError {
    CoreError::plain(error.to_string())
}

impl RemoteFileService {
    pub fn new(sessions: Arc<SessionManager>, privileged: Arc<PrivilegedExecutor>, logger: Arc<Logger>) -> Self {
        Self { sessions, privileged, logger, edits: Mutex::new(HashMap::new()), legacy_edits: Mutex::new(HashMap::new()) }
    }

    pub async fn sftp(&self, session_id: &str) -> CoreResult<Arc<Sftp>> {
        Sftp::open(&self.sessions, session_id).await
    }

    /// Shared fixed helper (remoteEditor.py) for editor, privileged file operations and transfers.
    pub async fn helper(&self, session_id: &str, request: Value, auth: &Authorization) -> CoreResult<Value> {
        remote_editor(&self.sessions, &self.privileged, session_id, &request, auth).await
    }

    // --- listing --------------------------------------------------------------------------------

    pub async fn list(&self, session_id: &str, path: &Value, options: &Value) -> CoreResult<Value> {
        let target = normalize_remote_path(&or_root(path))?;
        if options["sudo"] == true {
            return self.privileged_list(session_id, &target, options).await;
        }
        let sftp = self.sftp(session_id).await?;
        let entries = sftp.read_dir(&target).await.map_err(|e| e.file_error(&format!("Directory {target} cannot be listed.")))?;
        let mut result: Vec<Value> = entries
            .into_iter()
            .filter(|e| !e.name.is_empty() && e.name != "." && e.name != "..")
            .map(|e| {
                json!({
                    "name": e.name,
                    "path": posix::join(&target, &e.name),
                    "type": if e.stat.is_dir() { "directory" } else if e.stat.is_symlink() { "symlink" } else { "file" },
                    "size": e.stat.size,
                    "modifiedAt": iso_from_secs(e.stat.mtime as u64),
                    "mode": e.stat.mode
                })
            })
            .collect();
        // A link to a directory must be navigable, a link to a file editable (the target is edited).
        let links: Vec<usize> = result.iter().enumerate().filter(|(_, e)| e["type"] == "symlink").map(|(i, _)| i).take(500).collect();
        let stats = futures::future::join_all(links.iter().map(|i| {
            let path = result[*i]["path"].as_str().unwrap_or("").to_string();
            let sftp = sftp.clone();
            async move { sftp.stat(&path).await.ok() }
        }))
        .await;
        for (i, stat) in links.into_iter().zip(stats) {
            let entry = &mut result[i];
            match stat {
                None => entry["targetType"] = json!("missing"),
                Some(s) if s.is_dir() => entry["targetType"] = json!("directory"),
                Some(s) => {
                    entry["targetType"] = json!("file");
                    entry["size"] = json!(s.size);
                }
            }
        }
        sort_entries(&mut result);
        Ok(Value::Array(result))
    }

    async fn privileged_list(&self, session_id: &str, target: &str, options: &Value) -> CoreResult<Value> {
        let value = self.helper(session_id, json!({ "operation": "listdir", "path": target }), &Authorization::from_options(options).sudo()).await?;
        let mut entries: Vec<Value> = value["entries"]
            .as_array()
            .cloned()
            .unwrap_or_default()
            .into_iter()
            .map(|entry| {
                let name = entry["name"].as_str().map(str::to_string).unwrap_or_else(|| entry["name"].to_string());
                let kind = match entry["type"].as_str() {
                    Some("directory") => "directory",
                    Some("symlink") => "symlink",
                    _ => "file",
                };
                let mut out = json!({
                    "name": name,
                    "path": posix::join(target, &name),
                    "type": kind,
                    "size": entry["size"].as_f64().unwrap_or(0.0).max(0.0) as u64,
                    "modifiedAt": iso_from_secs(entry["mtime"].as_f64().unwrap_or(0.0).max(0.0) as u64),
                    "mode": entry["mode"].as_u64().unwrap_or(0),
                    "privileged": true
                });
                if !entry["targetType"].is_null() {
                    out["targetType"] = entry["targetType"].clone();
                }
                out
            })
            .collect();
        sort_entries(&mut entries);
        Ok(Value::Array(entries))
    }

    pub async fn privileges(&self, session_id: &str) -> CoreResult<Value> {
        let mut privilege = self.privileged.probe(session_id, false).await?;
        privilege["sudoCached"] = json!(self.privileged.has_cached_password(session_id));
        Ok(privilege)
    }

    // --- text editor ----------------------------------------------------------------------------

    fn editor_path(value: &Value) -> CoreResult<String> {
        let raw = value.as_str().unwrap_or("");
        if raw.is_empty() || raw.chars().count() > 4096 || !raw.starts_with('/') || raw.contains(['\0', '\r', '\n']) {
            return Err(CoreError::plain("Remote path must be an absolute path."));
        }
        if raw.split('/').any(|p| p == "." || p == "..") {
            return Err(CoreError::plain("Editor paths cannot contain traversal components."));
        }
        Ok(posix::normalize(raw))
    }

    fn identity(&self, session_id: &str) -> CoreResult<String> {
        let session = self.sessions.list().into_iter().find(|s| s["sessionId"] == session_id && s["state"] == "connected").ok_or_else(|| CoreError::plain("The editor session is not connected."))?;
        Ok(serde_json::to_string(&json!([session_id, session["profileId"], session["host"], session["port"], session["username"]])).unwrap_or_default())
    }

    fn legacy_key(owner: &str, identity: &str, target: &str) -> String {
        serde_json::to_string(&json!([owner, identity, target])).unwrap_or_default()
    }

    pub async fn read_text(&self, session_id: &str, path: &Value, options: &Value, owner: &str) -> CoreResult<Value> {
        let target = Self::editor_path(path)?;
        let identity = self.identity(session_id)?;
        let value = self.helper(session_id, json!({ "operation": "read", "path": target }), &Authorization::from_options(options)).await?;
        let data = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, value["data"].as_str().unwrap_or("")).map_err(|_| CoreError::plain("Remote editor content verification failed."))?;
        if data.len() > MAX_TEXT_BYTES {
            return Err(CoreError::plain(format!("Text editor limit is {MAX_TEXT_BYTES} bytes.")));
        }
        let text = String::from_utf8(data.clone()).map_err(|_| CoreError::plain("The selected file is not valid UTF-8 text."))?;
        if hex::encode(Sha256::digest(&data)) != value["baseline"]["hash"].as_str().unwrap_or("") {
            return Err(CoreError::plain("Remote editor content verification failed."));
        }
        let edit_id = random_hex(16);
        {
            let mut edits = self.edits.lock();
            if edits.len() >= 1000 {
                return Err(CoreError::plain("Close unused editor documents before opening more files."));
            }
            edits.insert(
                edit_id.clone(),
                Edit {
                    target: target.clone(),
                    identity: identity.clone(),
                    owner: owner.to_string(),
                    baseline: value["baseline"].clone(),
                    resolved_path: value["resolvedPath"].as_str().map(str::to_string),
                    busy: false,
                    uncertain: false,
                    dispose: false,
                },
            );
        }
        // A legacy call has no document identifier. Once two documents are open,
        // choosing either baseline would silently defeat conflict detection.
        {
            let key = Self::legacy_key(owner, &identity, &target);
            let mut legacy = self.legacy_edits.lock();
            let next = if legacy.contains_key(&key) { None } else { Some(edit_id.clone()) };
            legacy.insert(key, next);
        }
        let mtime_ns: i128 = match &value["baseline"]["mtimeNs"] {
            Value::String(s) => s.parse().unwrap_or(0),
            Value::Number(n) => n.as_f64().unwrap_or(0.0) as i128,
            _ => 0,
        };
        let modified = chrono::DateTime::<chrono::Utc>::from_timestamp_millis((mtime_ns / 1_000_000) as i64).map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Millis, true));
        Ok(json!({
            "path": target,
            "text": text,
            "size": data.len(),
            "editId": edit_id,
            "baselineHash": value["baseline"]["hash"],
            "metadata": value["baseline"],
            "resolvedPath": value["resolvedPath"].as_str().unwrap_or(&target),
            "link": value["link"] == true,
            "directSave": value["directSave"] != false,
            "privileged": options["sudo"] == true,
            "modifiedAt": modified
        }))
    }

    pub async fn write_text(&self, session_id: &str, path: &Value, text: &Value, options: &Value, owner: &str) -> CoreResult<Value> {
        let target = Self::editor_path(path)?;
        let data = match text {
            Value::String(s) => s.as_bytes().to_vec(),
            Value::Null => Vec::new(),
            other => other.to_string().into_bytes(),
        };
        if data.len() > MAX_TEXT_BYTES {
            return Err(CoreError::plain(format!("Text editor limit is {MAX_TEXT_BYTES} bytes.")));
        }
        let identity = self.identity(session_id)?;
        let requested = options["editId"].as_str().filter(|s| !s.is_empty()).map(str::to_string);
        let legacy = self.legacy_edits.lock().get(&Self::legacy_key(owner, &identity, &target)).cloned();
        if requested.is_none() && matches!(legacy, Some(None)) {
            return Err(CoreError::new("FILE_EDIT_ID_REQUIRED", "FILE_EDIT_ID_REQUIRED: Multiple documents are open for this path; pass the editId returned by readText."));
        }
        let edit_id = requested.or(legacy.flatten()).unwrap_or_default();
        let edit = {
            let mut edits = self.edits.lock();
            let edit = edits.get_mut(&edit_id).filter(|e| e.target == target && e.identity == identity && e.owner == owner).ok_or_else(|| CoreError::plain("Reopen this file to capture an editor baseline before saving."))?;
            if edit.busy || edit.uncertain {
                return Err(CoreError::new("FILE_RESULT_UNKNOWN", "FILE_RESULT_UNKNOWN: Inspect recovery before starting another save."));
            }
            edit.busy = true;
            edit.clone()
        };
        let auth = Authorization::from_options(options);
        let hash = hex::encode(Sha256::digest(&data));
        let id = random_hex(16);
        let mut recovery: Option<Value> = None;
        let mut submitted = false;
        let outcome = async {
            let mut request = json!({ "operation": "prepare", "id": id, "path": target, "baseline": edit.baseline, "hash": hash });
            if let Some(resolved) = &edit.resolved_path {
                request["resolvedPath"] = json!(resolved);
            }
            let prepared = self.helper(session_id, request, &Authorization::default()).await?;
            recovery = Some(prepared.clone());
            self.upload_editor_copy(session_id, prepared["temporary"].as_str().unwrap_or(""), &data).await?;
            submitted = true;
            let mut finalize = json!({ "operation": "finalize", "id": id, "uid": prepared["uid"], "path": target, "baseline": edit.baseline, "hash": hash });
            if let Some(resolved) = &edit.resolved_path {
                finalize["resolvedPath"] = json!(resolved);
            }
            if options["backup"] == true {
                finalize["backup"] = json!(true);
            }
            let saved = self.helper(session_id, finalize, &auth).await?;
            if saved["status"] != "saved" || saved["hash"] != hash.as_str() || saved["baseline"]["hash"] != hash.as_str() {
                return Err(CoreError::new("FILE_RESULT_UNKNOWN", "Remote save could not be confirmed."));
            }
            if let Some(e) = self.edits.lock().get_mut(&edit_id) {
                e.baseline = saved["baseline"].clone();
            }
            let recovery_available = self.helper(session_id, json!({ "operation": "cleanup", "id": id, "uid": prepared["uid"] }), &auth).await.is_err();
            self.logger.info("Remote editor save confirmed.", json!({ "sessionId": session_id, "path": target, "resolvedPath": saved["resolvedPath"], "privileged": auth.sudo, "bytes": data.len() }));
            Ok(json!({
                "path": target,
                "size": data.len(),
                "status": "saved",
                "editId": edit_id,
                "baselineHash": hash,
                "recoveryId": id,
                "recoveryAvailable": recovery_available,
                "resolvedPath": saved["resolvedPath"].as_str().unwrap_or(&target),
                "link": saved["link"] == true,
                "backupPath": saved.get("backupPath").cloned().filter(|v| !v.is_null()).unwrap_or(Value::Null)
            }))
        }
        .await;
        let dispose = {
            let mut edits = self.edits.lock();
            match edits.get_mut(&edit_id) {
                Some(e) => {
                    e.busy = false;
                    if let Err(error) = &outcome {
                        e.uncertain = submitted && !DEFINITE_EDITOR_FAILURES.contains(&error.code.as_str());
                    }
                    e.dispose
                }
                None => false,
            }
        };
        if dispose {
            self.close_text(&edit_id, owner);
        }
        outcome.map_err(|error| {
            let uncertain = submitted && !DEFINITE_EDITOR_FAILURES.contains(&error.code.as_str());
            let code = if uncertain { "FILE_RESULT_UNKNOWN".to_string() } else if error.code.is_empty() { "FILE_UPLOAD_FAILED".to_string() } else { error.code.clone() };
            let recovery_id = recovery.as_ref().and_then(|r| r["recoveryId"].as_str().map(str::to_string));
            let message = format!("{code}: Save was not confirmed. Keep editor content and inspect recovery{}.", recovery_id.as_ref().map(|r| format!(" ({r})")).unwrap_or_default());
            self.logger.warn("Remote editor save not confirmed.", json!({ "sessionId": session_id, "path": target, "code": code, "status": if uncertain { "unknown" } else { "failed" }, "privileged": auth.sudo }));
            CoreError::new(&code, message)
                .with("recoveryId", recovery_id.map(Value::from).unwrap_or(Value::Null))
                .with("recoveryPath", recovery.as_ref().map(|r| r["temporary"].clone()).unwrap_or(Value::Null))
                .with("status", if uncertain { "unknown" } else { "failed" })
        })
    }

    pub async fn save_text(&self, session_id: &str, path: &Value, text: &Value, options: &Value, owner: &str) -> Value {
        match self.write_text(session_id, path, text, options, owner).await {
            Ok(value) => json!({ "ok": true, "value": value }),
            Err(error) => json!({ "ok": false, "error": {
                "code": if error.code.is_empty() { "FILE_SAVE_FAILED" } else { error.code.as_str() },
                "message": error.message,
                "status": error.details.get("status").cloned().unwrap_or(json!("failed")),
                "recoveryId": error.details.get("recoveryId").cloned().unwrap_or(Value::Null),
                "recoveryPath": error.details.get("recoveryPath").cloned().unwrap_or(Value::Null)
            } }),
        }
    }

    async fn upload_editor_copy(&self, session_id: &str, temporary: &str, data: &[u8]) -> CoreResult<()> {
        let work = async {
            let sftp = self.sftp(session_id).await?;
            sftp.write_new_file(temporary, data, 0o600).await.map_err(sftp_plain)
        };
        tokio::time::timeout(Duration::from_secs(60), work).await.map_err(|_| CoreError::new("FILE_UPLOAD_TIMEOUT", "Editor upload timed out."))?
    }

    async fn recovery(&self, session_id: &str, operation: &str, recovery_id: Option<&Value>, options: &Value) -> CoreResult<Value> {
        let mut request = json!({ "operation": operation });
        if let Some(id) = recovery_id.filter(|v| !v.is_null()) {
            request["id"] = id.clone();
        }
        if options["sudo"] == true {
            let identity = self.helper(session_id, json!({ "operation": "identity" }), &Authorization::default()).await?;
            request["uid"] = identity["uid"].clone();
        }
        self.helper(session_id, request, &Authorization::from_options(options)).await
    }

    pub async fn list_recovery(&self, session_id: &str, options: &Value) -> CoreResult<Value> {
        self.recovery(session_id, "list", None, options).await
    }

    pub async fn read_recovery(&self, session_id: &str, recovery_id: &Value, options: &Value) -> CoreResult<Value> {
        let result = self.recovery(session_id, "inspect", Some(recovery_id), options).await?;
        let unverified = || CoreError::plain("FILE_RECOVERY_UNVERIFIED: Recovery content could not be verified.");
        let data = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, result["data"].as_str().unwrap_or("")).map_err(|_| unverified())?;
        if result["complete"] != true || data.len() > MAX_TEXT_BYTES || hex::encode(Sha256::digest(&data)) != result["hash"].as_str().unwrap_or("") {
            return Err(unverified());
        }
        let text = String::from_utf8(data).map_err(|_| CoreError::plain("Recovery file is not valid UTF-8 text."))?;
        let mut summary = result;
        if let Some(map) = summary.as_object_mut() {
            map.remove("data");
            map.insert("text".into(), json!(text));
        }
        Ok(summary)
    }

    pub async fn cleanup_recovery(&self, session_id: &str, recovery_id: &Value, options: &Value) -> CoreResult<Value> {
        self.recovery(session_id, "cleanup", Some(recovery_id), options).await
    }

    pub fn close_text(&self, edit_id: &str, owner: &str) -> Value {
        let mut edits = self.edits.lock();
        let Some(edit) = edits.get(edit_id).filter(|e| e.owner == owner && !e.busy).cloned() else { return json!({ "closed": false }) };
        edits.remove(edit_id);
        let key = Self::legacy_key(owner, &edit.identity, &edit.target);
        let remaining: Vec<String> = edits.iter().filter(|(_, o)| o.owner == owner && o.identity == edit.identity && o.target == edit.target).map(|(id, _)| id.clone()).collect();
        let mut legacy = self.legacy_edits.lock();
        if remaining.is_empty() {
            legacy.remove(&key);
        } else {
            legacy.insert(key, if remaining.len() == 1 { Some(remaining[0].clone()) } else { None });
        }
        json!({ "closed": true })
    }

    pub fn cleanup_module(&self, owner: &str) {
        let ids: Vec<(String, bool)> = self.edits.lock().iter().filter(|(_, e)| e.owner == owner).map(|(id, e)| (id.clone(), e.busy)).collect();
        for (id, busy) in ids {
            if busy {
                if let Some(e) = self.edits.lock().get_mut(&id) {
                    e.dispose = true;
                }
            } else {
                self.close_text(&id, owner);
            }
        }
    }

    // --- mutations ------------------------------------------------------------------------------

    async fn privileged_mutation(&self, session_id: &str, request: Value, options: &Value, log: &str) -> CoreResult<Value> {
        let value = self.helper(session_id, request.clone(), &Authorization::from_options(options).sudo()).await?;
        self.logger.warn(log, json!({ "sessionId": session_id, "operation": request["operation"], "path": request.get("path").or(request.get("from")), "to": request["to"] }));
        Ok(value)
    }

    /// OpenSSH reports an existing target as a generic failure; tell the user what actually happened.
    async fn mutation_error(&self, sftp: &Sftp, error: SftpError, target: &str, message: &str) -> CoreError {
        if matches!(error.status(), Some(4) | Some(11)) && sftp.lstat(target).await.is_ok() {
            return file_error("FILE_EXISTS", &format!("{target} already exists."));
        }
        error.file_error(message)
    }

    pub async fn create_file(&self, session_id: &str, path: &Value, options: &Value) -> CoreResult<Value> {
        let target = normalize_remote_path(path)?;
        if target == "/" {
            return Err(CoreError::plain("The remote root cannot be a file."));
        }
        if options["sudo"] == true {
            self.privileged_mutation(session_id, json!({ "operation": "create", "path": target }), options, "Remote file created with privileges.").await?;
            return Ok(json!({ "path": target, "size": 0 }));
        }
        let sftp = self.sftp(session_id).await?;
        if let Err(error) = sftp.write_new_file(&target, &[], 0o644).await {
            return Err(self.mutation_error(&sftp, error, &target, &format!("File {target} could not be created.")).await);
        }
        self.logger.warn("Remote file created.", json!({ "sessionId": session_id, "path": target }));
        Ok(json!({ "path": target, "size": 0 }))
    }

    pub async fn mkdir(&self, session_id: &str, path: &Value, options: &Value) -> CoreResult<Value> {
        let target = normalize_remote_path(path)?;
        if target == "/" {
            return Err(CoreError::plain("The remote root already exists."));
        }
        if options["sudo"] == true {
            self.privileged_mutation(session_id, json!({ "operation": "mkdir", "path": target }), options, "Remote directory created with privileges.").await?;
            return Ok(json!({ "path": target }));
        }
        let sftp = self.sftp(session_id).await?;
        if let Err(error) = sftp.mkdir(&target, 0o755).await {
            return Err(self.mutation_error(&sftp, error, &target, &format!("Directory {target} could not be created.")).await);
        }
        self.logger.warn("Remote directory created.", json!({ "sessionId": session_id, "path": target }));
        Ok(json!({ "path": target }))
    }

    pub async fn rename(&self, session_id: &str, from: &Value, to: &Value, options: &Value) -> CoreResult<Value> {
        let (from, to) = (normalize_remote_path(from)?, normalize_remote_path(to)?);
        if from == "/" || to == "/" {
            return Err(CoreError::plain("The remote root cannot be renamed."));
        }
        if PROTECTED_PATHS.contains(&from.as_str()) {
            return Err(file_error("FILE_PROTECTED_PATH", &format!("{from} is a protected system directory.")));
        }
        if options["sudo"] == true {
            self.privileged_mutation(session_id, json!({ "operation": "rename", "from": from, "to": to }), options, "Remote entry renamed with privileges.").await?;
            return Ok(json!({ "from": from, "to": to }));
        }
        let sftp = self.sftp(session_id).await?;
        if let Err(error) = sftp.rename(&from, &to).await {
            return Err(self.mutation_error(&sftp, error, &to, &format!("{from} could not be renamed to {to}.")).await);
        }
        self.logger.warn("Remote entry renamed.", json!({ "sessionId": session_id, "from": from, "to": to }));
        Ok(json!({ "from": from, "to": to }))
    }

    pub async fn remove(&self, session_id: &str, path: &Value, options: &Value) -> CoreResult<Value> {
        let target = normalize_remote_path(path)?;
        if target == "/" {
            return Err(CoreError::plain("The remote root cannot be removed."));
        }
        if PROTECTED_PATHS.contains(&target.as_str()) {
            return Err(file_error("FILE_PROTECTED_PATH", &format!("{target} is a protected system directory.")));
        }
        if options["sudo"] == true {
            let value = self.privileged_mutation(session_id, json!({ "operation": "remove", "path": target }), options, "Remote entry removed with privileges.").await?;
            return Ok(json!({ "path": target, "type": if value["type"] == "directory" { "directory" } else { "file" } }));
        }
        let sftp = self.sftp(session_id).await?;
        let failed = |e: SftpError| e.file_error(&format!("{target} could not be removed."));
        let stat = sftp.lstat(&target).await.map_err(failed)?;
        if stat.is_dir() { sftp.rmdir(&target).await } else { sftp.unlink(&target).await }.map_err(failed)?;
        let kind = if stat.is_dir() { "directory" } else { "file" };
        self.logger.warn("Remote entry removed.", json!({ "sessionId": session_id, "path": target, "type": kind }));
        Ok(json!({ "path": target, "type": kind }))
    }

    pub fn normalize_selection(paths: &Value) -> CoreResult<Vec<String>> {
        let values: Vec<Value> = match paths {
            Value::Array(list) => list.clone(),
            other => vec![other.clone()],
        };
        let mut targets: Vec<String> = Vec::new();
        for value in values.into_iter().filter(|v| !v.is_null()) {
            let target = normalize_remote_path(&value)?;
            if !targets.contains(&target) {
                targets.push(target);
            }
        }
        if targets.is_empty() || targets.len() > MAX_SELECTED_ENTRIES {
            return Err(CoreError::plain(format!("Select between 1 and {MAX_SELECTED_ENTRIES} remote entries.")));
        }
        Ok(targets)
    }

    pub async fn remove_many(&self, session_id: &str, paths: &Value, recursive: bool, options: &Value) -> CoreResult<Value> {
        let targets = Self::normalize_selection(paths)?;
        if targets.iter().any(|t| t == "/") {
            return Err(CoreError::plain("The remote root cannot be removed."));
        }
        if let Some(protected) = targets.iter().find(|t| PROTECTED_PATHS.contains(&t.as_str())) {
            return Err(file_error("FILE_PROTECTED_PATH", &format!("{protected} is a protected system directory.")));
        }
        let mut ordered = targets.clone();
        ordered.sort_by_key(|t| std::cmp::Reverse(t.len()));
        if options["sudo"] == true {
            let mut entries = 0u64;
            for target in &ordered {
                let value = self.helper(session_id, json!({ "operation": "remove", "path": target, "recursive": recursive }), &Authorization::from_options(options).sudo()).await?;
                entries += value["entries"].as_u64().filter(|n| *n > 0).unwrap_or(1);
            }
            self.logger.warn("Remote entries removed with privileges.", json!({ "sessionId": session_id, "selected": targets.len(), "entries": entries, "recursive": recursive }));
            return Ok(json!({ "selected": targets.len(), "entries": entries }));
        }
        let sftp = self.sftp(session_id).await?;
        let mut state = BatchState::default();
        for target in &ordered {
            remove_remote_entry(&sftp, target, recursive, &mut state).await?;
        }
        self.logger.warn("Remote entries removed.", json!({ "sessionId": session_id, "selected": targets.len(), "entries": state.entries, "recursive": recursive }));
        Ok(json!({ "selected": targets.len(), "entries": state.entries }))
    }

    // --- simple (non-queued) transfers ------------------------------------------------------------

    pub async fn upload(&self, session_id: &str, directory: &Value, picked: Vec<PathBuf>) -> CoreResult<Value> {
        let directory = normalize_remote_path(&or_root(directory))?;
        let mut local_paths: Vec<PathBuf> = Vec::new();
        for p in picked {
            if !local_paths.contains(&p) {
                local_paths.push(p);
            }
        }
        if local_paths.is_empty() {
            return Ok(json!({ "canceled": true, "items": [] }));
        }
        if local_paths.len() > MAX_SELECTED_ENTRIES {
            return Err(CoreError::plain(format!("At most {MAX_SELECTED_ENTRIES} entries can be uploaded at once.")));
        }
        let sftp = self.sftp(session_id).await?;
        let mut state = BatchState::default();
        let mut items = Vec::new();
        for local in &local_paths {
            let name = local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default();
            let target = posix::join(&directory, &name);
            let (kind, size) = Box::pin(upload_local_entry(&sftp, local, &target, &mut state)).await?;
            items.push(json!({ "path": target, "type": kind, "size": size }));
        }
        self.logger.info("Entries uploaded through Core.", json!({ "sessionId": session_id, "directory": directory, "entries": state.entries, "bytes": state.bytes }));
        Ok(json!({ "canceled": false, "items": items, "entries": state.entries, "size": state.bytes }))
    }

    /// Download step 1 (before the save dialog): the file must be a regular file within limits.
    pub async fn download_check(&self, session_id: &str, path: &Value) -> CoreResult<(String, u64)> {
        let target = normalize_remote_path(path)?;
        let sftp = self.sftp(session_id).await?;
        let stat = sftp.stat(&target).await.map_err(sftp_plain)?;
        if stat.is_dir() {
            return Err(CoreError::plain("Directory download is not available in this first File Manager version."));
        }
        if stat.size > MAX_TRANSFER_BYTES {
            return Err(CoreError::plain("Remote file exceeds the transfer size limit."));
        }
        Ok((target, stat.size))
    }

    pub async fn download_to(&self, session_id: &str, target: &str, size: u64, local: &Path) -> CoreResult<Value> {
        let sftp = self.sftp(session_id).await?;
        get_file(&sftp, target, local).await?;
        self.logger.info("File downloaded through Core.", json!({ "sessionId": session_id, "path": target, "bytes": size }));
        Ok(json!({ "canceled": false, "path": target, "size": size }))
    }

    pub async fn download_many(&self, session_id: &str, targets: &[String], directory: &Path) -> CoreResult<Value> {
        let sftp = self.sftp(session_id).await?;
        let mut state = BatchState::default();
        let mut items = Vec::new();
        for target in targets {
            if target == "/" {
                return Err(CoreError::plain("Downloading the entire remote root is not allowed in one batch."));
            }
            let local = directory.join(safe_entry_name(&posix::basename(target))?);
            let (kind, size) = Box::pin(download_remote_entry(&sftp, target, &local, &mut state)).await?;
            items.push(json!({ "path": target, "type": kind, "size": size }));
        }
        self.logger.info("Entries downloaded through Core.", json!({ "sessionId": session_id, "entries": state.entries, "bytes": state.bytes }));
        Ok(json!({ "canceled": false, "items": items, "entries": state.entries, "size": state.bytes }))
    }

    pub fn archive_format(requested: &Value) -> CoreResult<String> {
        let raw = requested.as_str().filter(|s| !s.is_empty()).unwrap_or("zip").to_lowercase();
        let format = if raw == "tgz" { "tar.gz".to_string() } else { raw };
        if !["zip", "tar", "tar.gz"].contains(&format.as_str()) {
            return Err(CoreError::plain("Supported archive formats are zip, tar and tar.gz."));
        }
        Ok(format)
    }

    pub async fn download_archive(&self, session_id: &str, targets: &[String], format: &str, destination: &Path) -> CoreResult<Value> {
        let root = std::env::temp_dir().join(format!("chj-file-archive-{}", random_hex(6)));
        std::fs::create_dir(&root).map_err(io_plain)?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let _ = std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o700));
        }
        let result = async {
            let sftp = self.sftp(session_id).await?;
            let mut state = BatchState::default();
            for target in targets {
                let local = root.join(safe_entry_name(&posix::basename(target))?);
                Box::pin(download_remote_entry(&sftp, target, &local, &mut state)).await?;
            }
            let (source, dest, fmt) = (root.clone(), destination.to_path_buf(), format.to_string());
            let size = tokio::task::spawn_blocking(move || super::archive::create(&source, &dest, &fmt)).await.map_err(|_| CoreError::plain("Archive creation failed."))??;
            self.logger.info("Remote selection archived through Core.", json!({ "sessionId": session_id, "format": format, "entries": state.entries, "sourceBytes": state.bytes, "archiveBytes": size }));
            Ok(json!({ "canceled": false, "path": destination.to_string_lossy(), "format": format, "entries": state.entries, "sourceSize": state.bytes, "size": size }))
        }
        .await;
        let _ = std::fs::remove_dir_all(&root);
        result
    }
}

async fn remove_remote_entry(sftp: &Sftp, path: &str, recursive: bool, state: &mut BatchState) -> CoreResult<()> {
    let failed = |e: SftpError| e.file_error("Remote entries could not be removed.");
    let stat = sftp.lstat(path).await.map_err(failed)?;
    state.count(stat.size)?;
    if stat.is_dir() {
        if recursive {
            for entry in sftp.read_dir(path).await.map_err(failed)? {
                if entry.name == "." || entry.name == ".." {
                    continue;
                }
                let child = posix::join(path, &safe_entry_name(&entry.name)?);
                Box::pin(remove_remote_entry(sftp, &child, true, state)).await?;
            }
        }
        sftp.rmdir(path).await.map_err(failed)
    } else {
        sftp.unlink(path).await.map_err(failed)
    }
}

async fn upload_local_entry(sftp: &Arc<Sftp>, local: &Path, remote: &str, state: &mut BatchState) -> CoreResult<(&'static str, u64)> {
    let meta = std::fs::symlink_metadata(local).map_err(io_plain)?;
    if meta.file_type().is_symlink() {
        return Err(CoreError::plain("Uploading symbolic links is not supported."));
    }
    if meta.is_dir() {
        state.count(0)?;
        if let Err(error) = sftp.mkdir(remote, 0o755).await {
            if !sftp.stat(remote).await.map(|s| s.is_dir()).unwrap_or(false) {
                return Err(sftp_plain(error));
            }
        }
        for entry in std::fs::read_dir(local).map_err(io_plain)? {
            let entry = entry.map_err(io_plain)?;
            let name = safe_entry_name(&entry.file_name().to_string_lossy())?;
            Box::pin(upload_local_entry(sftp, &local.join(&name), &posix::join(remote, &name), state)).await?;
        }
        return Ok(("directory", 0));
    }
    if !meta.is_file() {
        return Err(CoreError::plain("Only regular files and directories can be uploaded."));
    }
    state.count(meta.len())?;
    let temporary = posix::join(&posix::dirname(remote), &format!(".{}.chj-upload-{}.tmp", posix::basename(remote), random_hex(6)));
    let result = async {
        put_file(sftp, local, &temporary).await?;
        sftp.rename(&temporary, remote).await.map_err(sftp_plain)
    }
    .await;
    if result.is_err() {
        let _ = sftp.unlink(&temporary).await;
    }
    result.map(|_| ("file", meta.len()))
}

async fn download_remote_entry(sftp: &Arc<Sftp>, remote: &str, local: &Path, state: &mut BatchState) -> CoreResult<(&'static str, u64)> {
    let stat = sftp.lstat(remote).await.map_err(sftp_plain)?;
    if stat.is_symlink() {
        return Err(CoreError::plain(format!("Symbolic links are not included in recursive downloads: {remote}")));
    }
    if stat.is_dir() {
        state.count(0)?;
        match std::fs::create_dir(local) {
            Ok(()) => {
                #[cfg(unix)]
                {
                    use std::os::unix::fs::PermissionsExt;
                    let _ = std::fs::set_permissions(local, std::fs::Permissions::from_mode(0o700));
                }
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {}
            Err(e) => return Err(io_plain(e)),
        }
        for entry in sftp.read_dir(remote).await.map_err(sftp_plain)? {
            if entry.name == "." || entry.name == ".." {
                continue;
            }
            let name = safe_entry_name(&entry.name)?;
            Box::pin(download_remote_entry(sftp, &posix::join(remote, &name), &local.join(&name), state)).await?;
        }
        return Ok(("directory", 0));
    }
    state.count(stat.size)?;
    if let Some(parent) = local.parent() {
        std::fs::create_dir_all(parent).map_err(io_plain)?;
    }
    let file_name = local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default();
    let temporary = local.with_file_name(format!(".{file_name}.chj-download-{}.tmp", random_hex(6)));
    let result = async {
        get_file(sftp, remote, &temporary).await?;
        std::fs::rename(&temporary, local).map_err(io_plain)
    }
    .await;
    if result.is_err() {
        let _ = std::fs::remove_file(&temporary);
    }
    result.map(|_| ("file", stat.size))
}

fn pump_error(stop: PumpStop) -> CoreError {
    match stop {
        PumpStop::Failed(error) => error,
        PumpStop::Stopped(kind) => CoreError::plain(format!("Transfer {kind}")),
    }
}

/// ssh2 `fastPut` equivalent (pipelined).
pub async fn put_file(sftp: &Arc<Sftp>, local: &Path, remote: &str) -> CoreResult<()> {
    let file = LocalFile::open(local, "r").map_err(io_plain)?;
    let size = file.metadata().map_err(io_plain)?.len();
    let handle = sftp.open_file(remote, "w", None).await.map_err(sftp_plain)?;
    let produce: Producer = {
        let file = file.clone();
        Arc::new(move |pos, len| {
            let file = file.clone();
            Box::pin(async move {
                let data = file.read_at(pos, len).await.map_err(io_plain)?;
                if data.len() != len {
                    return Err(file_error("FILE_SOURCE_CHANGED", "The local file shrank during the transfer."));
                }
                Ok(data)
            })
        })
    };
    let consume: Consumer = {
        let (sftp, handle) = (sftp.clone(), handle.clone());
        Arc::new(move |buf, pos| {
            let (sftp, handle) = (sftp.clone(), handle.clone());
            Box::pin(async move { sftp.write_at(&handle, pos, &buf).await.map_err(sftp_plain) })
        })
    };
    let (_keep, stop) = tokio::sync::watch::channel(String::new());
    let result = pump(0, size, produce, consume, stop, |_| {}).await.map_err(pump_error);
    let closed = sftp.close_file(&handle).await.map_err(sftp_plain);
    result.and(closed)
}

/// ssh2 `fastGet` equivalent (pipelined).
pub async fn get_file(sftp: &Arc<Sftp>, remote: &str, local: &Path) -> CoreResult<()> {
    let handle: FileHandle = sftp.open_file(remote, "r", None).await.map_err(sftp_plain)?;
    let stat: Stat = sftp.fstat(&handle).await.map_err(sftp_plain)?;
    let _ = std::fs::remove_file(local);
    let file = LocalFile::open(local, "wx").map_err(io_plain)?;
    let produce: Producer = {
        let (sftp, handle) = (sftp.clone(), handle.clone());
        Arc::new(move |pos, len| {
            let (sftp, handle) = (sftp.clone(), handle.clone());
            Box::pin(async move {
                let data = sftp.read_full(&handle, pos, len).await.map_err(sftp_plain)?;
                if data.len() != len {
                    return Err(file_error("FILE_SOURCE_CHANGED", "The remote file shrank during the transfer."));
                }
                Ok(data)
            })
        })
    };
    let consume: Consumer = {
        let file = file.clone();
        Arc::new(move |buf, pos| {
            let file = file.clone();
            Box::pin(async move { file.write_at(pos, buf).await.map_err(io_plain) })
        })
    };
    let (_keep, stop) = tokio::sync::watch::channel(String::new());
    let result = pump(0, stat.size, produce, consume, stop, |_| {}).await.map_err(pump_error);
    let _ = sftp.close_file(&handle).await;
    result?;
    file.finish(stat.size).await.map_err(io_plain)
}