megalib 0.11.1

//! Upload operations.

use futures::io::Cursor;
use futures::stream::{self, StreamExt};
use rand::RngCore;
use serde_json::json;

use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _};

use super::utils::{get_chunk_size, upload_checksum};
use crate::api::ApiErrorCode;
use crate::base64::base64url_encode;
use crate::crypto::aes::{
    aes128_cbc_encrypt, aes128_ctr_encrypt, chunk_mac_calculate, meta_mac_calculate,
};
use crate::crypto::keys::pack_node_key;
use crate::error::{MegaError, Result};
use crate::fs::node::Node;
use crate::fs::upload_state::UploadState;
use crate::fs::upload_state::calculate_file_hash;
use crate::http::RequestKind;
use crate::session::Session;
use tracing::{debug, error};

impl Session {
    fn load_resumable_upload_state(
        &self,
        source_path: &std::path::Path,
        source_fingerprint: &str,
    ) -> Result<Option<UploadState>> {
        match self.load_persisted_upload_state_for_path(source_path, source_fingerprint) {
            Ok(state) => Ok(state),
            Err(err) => {
                debug!(
                    ?err,
                    path = %source_path.display(),
                    "failed to load resumable upload checkpoint from persistence runtime"
                );
                Ok(None)
            }
        }
    }

    fn save_resumable_upload_state(
        &self,
        source_path: &std::path::Path,
        state: &UploadState,
    ) -> Result<()> {
        self.save_persisted_upload_state_for_path(source_path, state)
    }

    fn try_save_resumable_upload_state(&self, source_path: &std::path::Path, state: &UploadState) {
        if let Err(err) = self.save_resumable_upload_state(source_path, state) {
            debug!(
                ?err,
                path = %source_path.display(),
                "failed to persist resumable upload checkpoint"
            );
        }
    }

    fn clear_resumable_upload_state(
        &self,
        source_path: &std::path::Path,
        source_fingerprint: &str,
    ) -> Result<()> {
        self.clear_persisted_upload_state_for_path(source_path, source_fingerprint)
    }

    fn should_reset_resumable_upload_on_error(err: &MegaError) -> bool {
        matches!(
            err,
            MegaError::ApiError { code, .. }
                if *code == ApiErrorCode::Again as i32 || *code == ApiErrorCode::Failed as i32
        )
    }

    async fn request_upload_url_for_file(
        &mut self,
        file_size: u64,
        parent: &Node,
    ) -> Result<String> {
        debug!(
            upload_preflight = "sdk-no-key-bootstrap",
            parent_handle = %parent.handle,
            parent_path = parent.path(),
            "upload hot path preflight policy"
        );

        let response = self
            .api_mut()
            .request(json!({
                "a": "u",
                "s": file_size,
                "ssl": 0
            }))
            .await?;

        response
            .get("p")
            .and_then(|v| v.as_str())
            .map(str::to_string)
            .ok_or_else(|| MegaError::Custom("Failed to get upload URL".to_string()))
    }

    /// Upload a node attribute (thumbnail, preview) to MEGA's attribute storage.
    ///
    /// This is used internally when `enable_previews(true)` is set, but can also
    /// be called directly for custom attributes.
    ///
    /// # Arguments
    /// * `data` - Raw attribute data (e.g., JPEG thumbnail bytes)
    /// * `attr_type` - Attribute type: "0" = thumbnail (128x128), "1" = preview  
    /// * `node_key` - The 16-byte file key used to encrypt the attribute
    ///
    /// # Returns
    /// Handle string like "0*ABC123" that can be added to file's `fa` field.
    #[cfg(feature = "preview")]
    pub async fn upload_node_attribute(
        &mut self,
        data: &[u8],
        attr_type: &str,
        node_key: &[u8; 16],
    ) -> Result<String> {
        // Pad data to multiple of 16 bytes
        let pad_len = if data.len().is_multiple_of(16) {
            0
        } else {
            16 - (data.len() % 16)
        };
        let mut padded = data.to_vec();
        padded.extend(std::iter::repeat_n(0u8, pad_len));

        // Get upload URL via a:ufa API
        let response = self
            .api_mut()
            .request(json!({
                "a": "ufa",
                "s": padded.len(),
                "ssl": 0
            }))
            .await?;

        let upload_url = response
            .get("p")
            .and_then(|v| v.as_str())
            .ok_or_else(|| MegaError::Custom("Failed to get attribute upload URL".to_string()))?;

        // Encrypt data with AES-CBC using node key
        let encrypted = aes128_cbc_encrypt(&padded, node_key);

        // Upload encrypted data
        let http = self.api_mut().http_client();
        let upload_response = http
            .post_binary(upload_url, encrypted, RequestKind::TransferUpload)
            .await?;

        if !upload_response.status().is_success() {
            return Err(MegaError::Custom(format!(
                "Attribute upload failed: {}",
                upload_response.status()
            )));
        }

        // Get handle from response (should be 8 bytes raw)
        let handle_bytes = upload_response
            .bytes()
            .await
            .map_err(MegaError::RequestError)?;

        if handle_bytes.len() != 8 {
            return Err(MegaError::Custom(format!(
                "Invalid attribute handle length: {}",
                handle_bytes.len()
            )));
        }

        // Base64 encode the handle
        let handle_b64 = base64url_encode(&handle_bytes);

        // Return in format "type*handle"
        Ok(format!("{}*{}", attr_type, handle_b64))
    }

    /// Upload a file to a directory.
    ///
    /// # Arguments
    /// * `local_path` - Path to the local file to upload
    /// * `remote_parent_path` - Path to the remote parent directory
    pub async fn upload_to_node<P: AsRef<std::path::Path>>(
        &mut self,
        local_path: P,
        parent: &Node,
    ) -> Result<Node> {
        let path = local_path.as_ref();
        let file_name = path
            .file_name()
            .ok_or_else(|| MegaError::Custom("Invalid file path".to_string()))?
            .to_string_lossy()
            .to_string();

        let metadata = tokio::fs::metadata(path)
            .await
            .map_err(|e| MegaError::Custom(format!("Failed to get metadata: {}", e)))?;
        let file_size = metadata.len();

        let parent_handle = parent.handle.clone();

        // 1. Get upload URL
        // [{a:u, s:<SIZE>, ssl:0}]
        debug!(
            upload_preflight = "sdk-no-key-bootstrap",
            parent_handle,
            parent_path = parent.path(),
            "upload hot path preflight policy"
        );

        let response = self
            .api_mut()
            .request(json!({
                "a": "u",
                "s": file_size,
                "ssl": 0
            }))
            .await?;

        let upload_url = response
            .get("p")
            .and_then(|v| v.as_str())
            .ok_or_else(|| MegaError::Custom("Failed to get upload URL".to_string()))?
            .to_string();

        // Generate encryption keys
        let (file_key, nonce) = {
            let mut rng = rand::thread_rng();
            let mut file_key = [0u8; 16];
            let mut nonce = [0u8; 8];
            rng.fill_bytes(&mut file_key);
            rng.fill_bytes(&mut nonce);
            (file_key, nonce)
        };

        // Create state (without file hash - not saving to disk)
        let state = crate::fs::upload_state::UploadState::new(
            upload_url,
            file_key,
            nonce,
            file_size,
            file_name,
            parent_handle,
            String::new(), // Empty hash - not resumable
        );

        // Delegate to internal upload with no state persistence
        self.upload_internal(path, state, None).await
    }

    pub async fn upload<P: AsRef<std::path::Path>>(
        &mut self,
        local_path: P,
        remote_parent_path: &str,
    ) -> Result<Node> {
        let parent_node = self.stat_by_path(remote_parent_path).ok_or_else(|| {
            MegaError::Custom(format!(
                "Parent directory not found: {}",
                remote_parent_path
            ))
        })?;
        let parent = parent_node.clone();
        self.upload_to_node(local_path, &parent).await
    }

    /// Upload a file with resume support.
    ///
    /// This method persists runtime-managed upload checkpoints after each
    /// committed chunk, allowing uploads to resume after interruption.
    /// Legacy `.megalib_upload` sidecar files are ignored.
    ///
    /// # Arguments
    /// * `local_path` - Path to the local file to upload
    /// * `remote_parent_path` - Path to the remote parent directory
    ///
    /// # Example
    /// ```no_run
    /// # use megalib::SessionHandle;
    /// # async fn example() -> megalib::error::Result<()> {
    /// let mut session = SessionHandle::login("user@example.com", "password").await?;
    /// session.refresh().await?;
    ///
    /// // This upload can be safely interrupted and resumed
    /// let node = session.upload_resumable("largefile.zip", "/Root").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn upload_resumable_to_node<P: AsRef<std::path::Path>>(
        &mut self,
        local_path: P,
        parent: &Node,
    ) -> Result<Node> {
        let path = local_path.as_ref();
        let file_name = path
            .file_name()
            .ok_or_else(|| MegaError::Custom("Invalid file path".to_string()))?
            .to_string_lossy()
            .to_string();

        let metadata = tokio::fs::metadata(path)
            .await
            .map_err(|e| MegaError::Custom(format!("Failed to get metadata: {}", e)))?;
        let file_size = metadata.len();
        let file_hash = calculate_file_hash(path)?;
        let parent_handle = parent.handle.clone();

        if let Some(existing_state) = self.load_resumable_upload_state(path, &file_hash)? {
            if existing_state.is_likely_valid() {
                let checkpoint_for_reset = existing_state.clone();
                match self.upload_internal(path, existing_state, Some(path)).await {
                    Ok(node) => return Ok(node),
                    Err(err) if Self::should_reset_resumable_upload_on_error(&err) => {
                        let upload_url =
                            self.request_upload_url_for_file(file_size, parent).await?;
                        let mut reset_state = checkpoint_for_reset;
                        reset_state.reset_for_new_upload_url(upload_url);
                        self.save_resumable_upload_state(path, &reset_state)?;
                        return self.upload_internal(path, reset_state, Some(path)).await;
                    }
                    Err(err) => return Err(err),
                }
            }

            let upload_url = self.request_upload_url_for_file(file_size, parent).await?;
            let mut reset_state = existing_state;
            reset_state.reset_for_new_upload_url(upload_url);
            self.save_resumable_upload_state(path, &reset_state)?;
            return self.upload_internal(path, reset_state, Some(path)).await;
        }

        let upload_url = self.request_upload_url_for_file(file_size, parent).await?;

        let (file_key, nonce) = {
            let mut rng = rand::thread_rng();
            let mut file_key = [0u8; 16];
            let mut nonce = [0u8; 8];
            use rand::RngCore;
            rng.fill_bytes(&mut file_key);
            rng.fill_bytes(&mut nonce);
            (file_key, nonce)
        };

        let state = UploadState::new(
            upload_url,
            file_key,
            nonce,
            file_size,
            file_name,
            parent_handle,
            file_hash,
        );

        self.save_resumable_upload_state(path, &state)?;
        self.upload_internal(path, state, Some(path)).await
    }

    pub async fn upload_resumable<P: AsRef<std::path::Path>>(
        &mut self,
        local_path: P,
        remote_parent_path: &str,
    ) -> Result<Node> {
        let parent_node = self.stat_by_path(remote_parent_path).ok_or_else(|| {
            MegaError::Custom(format!(
                "Parent directory not found: {}",
                remote_parent_path
            ))
        })?;
        let parent = parent_node.clone();
        self.upload_resumable_to_node(local_path, &parent).await
    }

    /// Upload data from a byte slice to a directory.
    ///
    /// This method is useful for uploading in-memory data without writing to disk first.
    /// Note: This method does NOT support resume (the data must be re-provided if interrupted).
    /// For large uploads that need resume support, use `upload_resumable` with a file path.
    ///
    /// # Arguments
    /// * `data` - The bytes to upload
    /// * `file_name` - Name for the uploaded file on MEGA
    /// * `remote_parent_path` - Path to the remote parent directory
    ///
    /// # Example
    /// ```no_run
    /// # use megalib::SessionHandle;
    /// # async fn example() -> megalib::error::Result<()> {
    /// let mut session = SessionHandle::login("user@example.com", "password").await?;
    /// session.refresh().await?;
    ///
    /// let data = b"Hello, MEGA!";
    /// let node = session.upload_from_bytes(data, "hello.txt", "/Root").await?;
    /// println!("Uploaded: {} ({} bytes)", node.name, node.size);
    /// # Ok(())
    /// # }
    /// ```
    pub async fn upload_bytes_to_node(
        &mut self,
        data: &[u8],
        file_name: &str,
        parent: &Node,
    ) -> Result<Node> {
        self.upload_reader_to_node(
            Cursor::new(data.to_vec()),
            file_name,
            data.len() as u64,
            parent,
        )
        .await
    }

    pub async fn upload_from_bytes(
        &mut self,
        data: &[u8],
        file_name: &str,
        remote_parent_path: &str,
    ) -> Result<Node> {
        let parent_node = self.stat_by_path(remote_parent_path).ok_or_else(|| {
            MegaError::Custom(format!(
                "Parent directory not found: {}",
                remote_parent_path
            ))
        })?;
        let parent = parent_node.clone();
        self.upload_bytes_to_node(data, file_name, &parent).await
    }

    /// Upload data from an async reader to a directory.
    ///
    /// This method accepts any type implementing `AsyncRead + AsyncSeek + Unpin + Send`,
    /// enabling uploads from various sources like in-memory buffers, network streams,
    /// or custom data sources.
    ///
    /// Note: This method does NOT support resume. The reader is consumed during upload,
    /// and if the upload is interrupted, you must provide a fresh reader to retry.
    /// For large file uploads that need resume support, use `upload_resumable` instead.
    ///
    /// Note: Preview/thumbnail generation is not supported for stream uploads since
    /// the data source may not be seekable after reading.
    ///
    /// # Arguments
    /// * `reader` - Async reader providing the data
    /// * `file_name` - Name for the uploaded file on MEGA
    /// * `file_size` - Total size of the data in bytes
    /// * `remote_parent_path` - Path to the remote parent directory
    ///
    /// # Example
    /// ```no_run
    /// # use megalib::SessionHandle;
    /// use futures::io::Cursor;
    /// # async fn example() -> megalib::error::Result<()> {
    /// let mut session = SessionHandle::login("user@example.com", "password").await?;
    /// session.refresh().await?;
    ///
    /// let data = vec![0u8; 1024]; // 1KB of zeros
    /// let cursor = Cursor::new(data);
    /// let node = session.upload_from_reader(cursor, "zeros.bin", 1024, "/Root").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn upload_reader_to_node<R>(
        &mut self,
        reader: R,
        file_name: &str,
        file_size: u64,
        parent: &Node,
    ) -> Result<Node>
    where
        R: futures::io::AsyncRead + futures::io::AsyncSeek + Unpin + Send,
    {
        let parent_handle = parent.handle.clone();

        debug!(
            upload_preflight = "sdk-no-key-bootstrap",
            parent_handle,
            parent_path = parent.path(),
            "upload hot path preflight policy"
        );

        // Get upload URL
        let response = self
            .api_mut()
            .request(json!({
                "a": "u",
                "s": file_size,
                "ssl": 0
            }))
            .await?;

        let upload_url = response
            .get("p")
            .and_then(|v| v.as_str())
            .ok_or_else(|| MegaError::Custom("Failed to get upload URL".to_string()))?
            .to_string();

        // Generate encryption keys
        let (file_key, nonce) = {
            let mut rng = rand::thread_rng();
            let mut file_key = [0u8; 16];
            let mut nonce = [0u8; 8];
            rng.fill_bytes(&mut file_key);
            rng.fill_bytes(&mut nonce);
            (file_key, nonce)
        };

        // Create state (no file hash needed - not resumable)
        let state = crate::fs::upload_state::UploadState::new(
            upload_url,
            file_key,
            nonce,
            file_size,
            file_name.to_string(),
            parent_handle,
            String::new(),
        );

        self.upload_internal_stream(reader, state).await
    }

    pub async fn upload_from_reader<R>(
        &mut self,
        reader: R,
        file_name: &str,
        file_size: u64,
        remote_parent_path: &str,
    ) -> Result<Node>
    where
        R: futures::io::AsyncRead + futures::io::AsyncSeek + Unpin + Send,
    {
        let parent_node = self.stat_by_path(remote_parent_path).ok_or_else(|| {
            MegaError::Custom(format!(
                "Parent directory not found: {}",
                remote_parent_path
            ))
        })?;
        let parent = parent_node.clone();
        self.upload_reader_to_node(reader, file_name, file_size, &parent)
            .await
    }

    /// Internal method to upload with optional state tracking.
    async fn upload_internal(
        &mut self,
        path: &std::path::Path,
        mut state: crate::fs::upload_state::UploadState,
        checkpoint_source_path: Option<&std::path::Path>,
    ) -> Result<Node> {
        let file_name = state.file_name.clone();
        let file_key = state.file_key;
        let nonce = state.nonce;
        let file_size = state.file_size;
        let parent_handle = state.parent_handle.clone();
        let upload_url = state.upload_url.clone();

        let mut file = tokio::fs::File::open(path)
            .await
            .map_err(|e| MegaError::Custom(format!("Failed to open file: {}", e)))?;

        // Calculate starting chunk index based on existing MACs
        let chunk_index = state.chunk_macs.len();
        let mut offset = state.offset;
        let mut chunk_macs = state.chunk_macs.clone();

        // If resume state indicates complete upload (offset == file_size),
        // we might have missed the handle if the process crashed before finalization.
        // Rewind by one chunk to force re-upload and get the handle again.
        if offset == file_size && file_size > 0 && !chunk_macs.is_empty() {
            chunk_macs.pop();

            // Recalculate offset from remaining chunks
            let mut new_offset = 0;
            for i in 0..chunk_macs.len() {
                new_offset += get_chunk_size(i, new_offset, file_size);
            }
            offset = new_offset;
        }

        // Seek to resume position
        if offset > 0 {
            file.seek(std::io::SeekFrom::Start(offset))
                .await
                .map_err(|e| MegaError::Custom(format!("Failed to seek: {}", e)))?;
        }

        let mut upload_handle = String::new();

        // Create chunk list for parallel processing
        let mut chunks = Vec::new();
        let mut iter_offset = offset;
        let mut iter_index = chunk_index;

        while iter_offset < file_size {
            let chunk_size = get_chunk_size(iter_index, iter_offset, file_size);
            chunks.push((iter_index, iter_offset, chunk_size));
            iter_offset += chunk_size;
            iter_index += 1;
        }

        let path_buf = path.to_path_buf();
        let workers = self.workers();
        let transfer_http = self.api_mut().http_client();

        let mut stream = stream::iter(chunks)
            .map(|(_index, chunk_offset, chunk_size)| {
                let path = path_buf.clone();
                let upload_url = upload_url.clone();
                let file_name_clone = file_name.clone();
                let transfer_http = transfer_http.clone();

                async move {
                    // Open file for this chunk
                    let mut file = tokio::fs::File::open(&path)
                        .await
                        .map_err(|e| MegaError::Custom(format!("Failed to open file: {}", e)))?;

                    file.seek(std::io::SeekFrom::Start(chunk_offset))
                        .await
                        .map_err(|e| MegaError::Custom(format!("Failed to seek: {}", e)))?;

                    let mut buffer = vec![0u8; chunk_size as usize];
                    file.read_exact(&mut buffer)
                        .await
                        .map_err(|e| MegaError::Custom(format!("Read error: {}", e)))?;

                    // Calculate chunk MAC
                    let mut mac_iv = [0u8; 16];
                    mac_iv[..8].copy_from_slice(&nonce);
                    mac_iv[8..].copy_from_slice(&nonce);

                    let chunk_mac = chunk_mac_calculate(&buffer, &file_key, &mac_iv);

                    // Encrypt chunk
                    let encrypted_chunk =
                        aes128_ctr_encrypt(&buffer, &file_key, &nonce, chunk_offset);

                    // Upload chunk
                    let checksum = upload_checksum(&encrypted_chunk);
                    let chunk_url = format!("{}/{}?d={}", upload_url, chunk_offset, checksum);

                    let response = transfer_http
                        .post_binary(&chunk_url, encrypted_chunk, RequestKind::TransferUpload)
                        .await?;

                    if !response.status().is_success() {
                        return Err(MegaError::Custom(format!(
                            "Chunk upload failed: {}",
                            response.status()
                        )));
                    }

                    let response_text = response.text().await.map_err(MegaError::RequestError)?;
                    let handle = if response_text.is_empty() {
                        return Err(MegaError::Custom("Empty upload response".to_string()));
                    } else if response_text.starts_with('-') {
                        let code = response_text.parse::<i64>().unwrap_or(-999);
                        return Err(MegaError::ApiError {
                            code: code as i32,
                            message: crate::api::ApiErrorCode::from(code)
                                .description()
                                .to_string(),
                        });
                    } else {
                        Some(response_text)
                    };

                    Ok((
                        chunk_mac,
                        chunk_offset + chunk_size,
                        handle,
                        file_name_clone,
                    ))
                }
            })
            .buffered(workers); // Run up to 'workers' chunks in parallel, return in order

        while let Some(result) = stream.next().await {
            match result {
                Ok((chunk_mac, new_offset, handle, fname)) => {
                    if let Some(h) = handle {
                        upload_handle = h;
                    }

                    // Update state - because we used .buffered(), results come in original order
                    chunk_macs.push(chunk_mac);
                    offset = new_offset;

                    // Save state
                    if checkpoint_source_path.is_some() {
                        state.chunk_macs = chunk_macs.clone();
                        state.offset = offset;
                        self.save_resumable_upload_state(path, &state)?;
                    }

                    // Report progress
                    let progress =
                        crate::progress::TransferProgress::new(offset, file_size, &fname);
                    if !self.report_progress(&progress) {
                        // Save state before cancelling
                        if checkpoint_source_path.is_some() {
                            state.chunk_macs = chunk_macs.clone();
                            state.offset = offset;
                            self.try_save_resumable_upload_state(path, &state);
                        }
                        return Err(MegaError::Custom("Upload cancelled by user".to_string()));
                    }
                }
                Err(e) => {
                    // Save state before returning error
                    if checkpoint_source_path.is_some() {
                        state.chunk_macs = chunk_macs.clone();
                        state.offset = offset;
                        self.try_save_resumable_upload_state(path, &state);
                    }
                    return Err(e);
                }
            }
        }

        if upload_handle.is_empty() {
            return Err(MegaError::Custom(
                "Did not receive upload handle".to_string(),
            ));
        }

        // Generate preview if enabled and the preview feature is compiled in.
        let file_attr = if self.previews_enabled() {
            #[cfg(feature = "preview")]
            {
                if let Some(thumbnail_result) = crate::preview::generate_thumbnail(path) {
                    match thumbnail_result {
                        Ok(thumbnail_data) => (self
                            .upload_node_attribute(&thumbnail_data, "0", &file_key)
                            .await)
                            .ok(),
                        Err(_) => None,
                    }
                } else {
                    None
                }
            }
            #[cfg(not(feature = "preview"))]
            {
                None
            }
        } else {
            None
        };

        // Delegate to shared finalization logic
        let node = self
            .finalize_upload(
                &upload_handle,
                &chunk_macs,
                &file_key,
                &nonce,
                &file_name,
                &parent_handle,
                file_attr,
            )
            .await?;

        // Delete state file on success
        if checkpoint_source_path.is_some()
            && let Err(err) = self.clear_resumable_upload_state(path, &state.file_hash)
        {
            error!(
                ?err,
                path = %path.display(),
                parent_handle = %parent_handle,
                "upload finalized remotely but failed to clear resumable upload checkpoint"
            );
            return Err(err);
        }

        Ok(node)
    }

    /// Internal method to upload from an async reader (sequential, no resume support).
    async fn upload_internal_stream<R>(&mut self, mut reader: R, state: UploadState) -> Result<Node>
    where
        R: futures::io::AsyncRead + futures::io::AsyncSeek + Unpin + Send,
    {
        use futures::io::AsyncReadExt;

        let file_name = state.file_name.clone();
        let file_key = state.file_key;
        let nonce = state.nonce;
        let file_size = state.file_size;
        let parent_handle = state.parent_handle.clone();
        let upload_url = state.upload_url.clone();

        let mut chunk_macs: Vec<[u8; 16]> = Vec::new();
        let mut offset: u64 = 0;
        let mut chunk_index: usize = 0;
        let mut upload_handle = String::new();
        let transfer_http = self.api_mut().http_client();

        // Sequential upload - read chunks one at a time from the stream
        while offset < file_size {
            let chunk_size = get_chunk_size(chunk_index, offset, file_size);
            let mut buffer = vec![0u8; chunk_size as usize];

            reader
                .read_exact(&mut buffer)
                .await
                .map_err(|e| MegaError::Custom(format!("Read error: {}", e)))?;

            // Calculate chunk MAC
            let mut mac_iv = [0u8; 16];
            mac_iv[..8].copy_from_slice(&nonce);
            mac_iv[8..].copy_from_slice(&nonce);
            let chunk_mac = chunk_mac_calculate(&buffer, &file_key, &mac_iv);

            // Encrypt chunk
            let encrypted_chunk = aes128_ctr_encrypt(&buffer, &file_key, &nonce, offset);

            // Upload chunk
            let checksum = upload_checksum(&encrypted_chunk);
            let chunk_url = format!("{}/{}?d={}", upload_url, offset, checksum);

            let response = transfer_http
                .post_binary(&chunk_url, encrypted_chunk, RequestKind::TransferUpload)
                .await?;

            if !response.status().is_success() {
                return Err(MegaError::Custom(format!(
                    "Chunk upload failed: {}",
                    response.status()
                )));
            }

            let response_text = response.text().await.map_err(MegaError::RequestError)?;
            if response_text.is_empty() {
                return Err(MegaError::Custom("Empty upload response".to_string()));
            }
            if response_text.starts_with('-') {
                let code = response_text.parse::<i64>().unwrap_or(-999);
                return Err(MegaError::ApiError {
                    code: code as i32,
                    message: crate::api::ApiErrorCode::from(code)
                        .description()
                        .to_string(),
                });
            }
            upload_handle = response_text;

            chunk_macs.push(chunk_mac);
            offset += chunk_size;
            chunk_index += 1;

            // Report progress
            let progress = crate::progress::TransferProgress::new(offset, file_size, &file_name);
            if !self.report_progress(&progress) {
                return Err(MegaError::Custom("Upload cancelled by user".to_string()));
            }
        }

        if upload_handle.is_empty() {
            return Err(MegaError::Custom(
                "Did not receive upload handle".to_string(),
            ));
        }

        // Delegate to shared finalization logic (no preview for stream uploads)
        self.finalize_upload(
            &upload_handle,
            &chunk_macs,
            &file_key,
            &nonce,
            &file_name,
            &parent_handle,
            None,
        )
        .await
    }

    /// Shared upload finalization: creates the file node on MEGA after all chunks are uploaded.
    #[allow(clippy::too_many_arguments)]
    async fn finalize_upload(
        &mut self,
        upload_handle: &str,
        chunk_macs: &[[u8; 16]],
        file_key: &[u8; 16],
        nonce: &[u8; 8],
        file_name: &str,
        parent_handle: &str,
        file_attr: Option<String>,
    ) -> Result<Node> {
        let meta_mac = meta_mac_calculate(chunk_macs, file_key);

        // Encrypt attributes
        let attrs = json!({ "n": file_name }).to_string();
        let attrs_bytes = format!("MEGA{}", attrs).into_bytes();
        let pad_len = 16 - (attrs_bytes.len() % 16);
        let mut padded_attrs = attrs_bytes;
        padded_attrs.extend(std::iter::repeat_n(0, pad_len));
        let encrypted_attrs = aes128_cbc_encrypt(&padded_attrs, file_key);
        let attrs_b64 = base64url_encode(&encrypted_attrs);

        // Pack and encrypt node key
        let node_key = pack_node_key(file_key, nonce, &meta_mac);
        let encrypted_node_key =
            crate::crypto::aes::aes128_ecb_encrypt(&node_key, &self.master_key);
        let key_b64 = base64url_encode(&encrypted_node_key);

        // Create file node
        let mut node_data = json!({
            "h": upload_handle,
            "t": 0,
            "a": attrs_b64,
            "k": key_b64
        });

        if let Some(fa) = &file_attr {
            node_data["fa"] = json!(fa);
        }

        // If uploading into a shared/exported folder, include CR mapping so sharees can decrypt.
        let mut request = json!({
            "a": "p",
            "t": parent_handle,
            "n": [node_data],
            "v": 4,   // request file IDs/handles (matches SDK)
            "sm": 1,  // secure mode flag like SDK/webclient
            "i": self.session_id()
        });

        if let Some((share_handle, share_key)) = self.find_share_for_handle(parent_handle) {
            let targets: Vec<(String, Vec<u8>)> =
                vec![(upload_handle.to_string(), node_key.to_vec())];
            if let Some(cr_value) = self.build_cr_for_nodes(&share_handle, &share_key, &targets) {
                request["cr"] = cr_value;
            }
            // Also persist the parent share key into ^!keys if available (best effort).
            if self.key_manager.is_ready() {
                self.key_manager
                    .add_share_key_from_str(&share_handle, &share_key);
                let _ = self.persist_keys_attribute().await;
            }
        }

        let response = self.api_mut().request(request).await?;
        let seqtag = self.track_seqtag_from_response(&response);

        // Parse result; accept "f" array when available.
        if let Some(nodes_array) = response.get("f").and_then(|v| v.as_array())
            && let Some(node_obj) = nodes_array.first()
        {
            let node = self
                .parse_node(node_obj)
                .ok_or_else(|| MegaError::Custom("Failed to parse new node".to_string()))?;

            // Find parent path
            let parent_path_str = {
                let parent_path = self
                    .nodes
                    .iter()
                    .find(|n| n.handle == parent_handle)
                    .and_then(|n| n.path.as_ref())
                    .map(|p| p.as_str())
                    .unwrap_or("");

                if !parent_path.is_empty() {
                    format!("{}/{}", parent_path.trim_end_matches('/'), node.name)
                } else {
                    format!("/{}", node.name)
                }
            };

            self.nodes.push(node.clone());
            if let Some(last_node) = self.nodes.last_mut() {
                last_node.path = Some(parent_path_str);
            }

            let _ = seqtag;
            return Ok(node);
        }

        // Accept seqtag array response with error list.
        if let Some(arr) = response.as_array()
            && let Some(errors) = arr.get(1)
            && let Some(code) = first_error_code(errors)
        {
            return Err(MegaError::ApiError {
                code: code as i32,
                message: ApiErrorCode::from(code).description().to_string(),
            });
        }

        let _ = seqtag;

        let parent_path = self
            .nodes
            .iter()
            .find(|n| n.handle == parent_handle)
            .and_then(|n| n.path.clone())
            .unwrap_or_else(|| "/".to_string());
        let target_path = format!("{}/{}", parent_path.trim_end_matches('/'), file_name);

        if let Some(existing) = self.stat_by_path(&target_path) {
            return Ok(existing.clone());
        }

        Err(MegaError::Custom(
            "Failed to observe uploaded node via action packets".to_string(),
        ))
    }
}

fn first_error_code(errors: &serde_json::Value) -> Option<i64> {
    if let Some(code) = errors.as_i64() {
        return if code != 0 { Some(code) } else { None };
    }

    if let Some(arr) = errors.as_array() {
        for v in arr {
            if let Some(code) = v.as_i64()
                && code != 0
            {
                return Some(code);
            }
        }
        return None;
    }

    if let Some(obj) = errors.as_object() {
        for v in obj.values() {
            if let Some(code) = v.as_i64()
                && code != 0
            {
                return Some(code);
            }
        }
    }

    None
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::path::{Path, PathBuf};
    use std::sync::Arc;
    use std::time::{SystemTime, UNIX_EPOCH};

    use super::*;
    use crate::session::runtime::persistence::{
        PersistedEngineState, PersistenceBackend, PersistenceRuntime, PersistenceScope,
        TransferCheckpointKey, TransferCheckpointRecord,
    };

    #[derive(Debug, Default)]
    struct LoadFailingUploadStateBackend;

    impl PersistenceBackend for LoadFailingUploadStateBackend {
        fn load_engine_state(
            &self,
            _scope: &PersistenceScope,
        ) -> Result<Option<PersistedEngineState>> {
            Ok(None)
        }

        fn save_engine_state(
            &self,
            _scope: &PersistenceScope,
            _state: &PersistedEngineState,
        ) -> Result<()> {
            Ok(())
        }

        fn clear_engine_state(&self, _scope: &PersistenceScope) -> Result<()> {
            Ok(())
        }

        fn load_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _key: &TransferCheckpointKey,
        ) -> Result<Option<TransferCheckpointRecord>> {
            Err(MegaError::Custom("corrupt transfer checkpoint".to_string()))
        }

        fn save_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _record: &TransferCheckpointRecord,
        ) -> Result<()> {
            Ok(())
        }

        fn clear_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _key: &TransferCheckpointKey,
        ) -> Result<()> {
            Ok(())
        }
    }

    #[derive(Debug, Default)]
    struct SaveFailingUploadStateBackend;

    impl PersistenceBackend for SaveFailingUploadStateBackend {
        fn load_engine_state(
            &self,
            _scope: &PersistenceScope,
        ) -> Result<Option<PersistedEngineState>> {
            Ok(None)
        }

        fn save_engine_state(
            &self,
            _scope: &PersistenceScope,
            _state: &PersistedEngineState,
        ) -> Result<()> {
            Ok(())
        }

        fn clear_engine_state(&self, _scope: &PersistenceScope) -> Result<()> {
            Ok(())
        }

        fn load_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _key: &TransferCheckpointKey,
        ) -> Result<Option<TransferCheckpointRecord>> {
            Ok(None)
        }

        fn save_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _record: &TransferCheckpointRecord,
        ) -> Result<()> {
            Err(MegaError::Custom(
                "failed to save transfer checkpoint".to_string(),
            ))
        }

        fn clear_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _key: &TransferCheckpointKey,
        ) -> Result<()> {
            Ok(())
        }
    }

    #[derive(Debug, Default)]
    struct ClearFailingUploadStateBackend;

    impl PersistenceBackend for ClearFailingUploadStateBackend {
        fn load_engine_state(
            &self,
            _scope: &PersistenceScope,
        ) -> Result<Option<PersistedEngineState>> {
            Ok(None)
        }

        fn save_engine_state(
            &self,
            _scope: &PersistenceScope,
            _state: &PersistedEngineState,
        ) -> Result<()> {
            Ok(())
        }

        fn clear_engine_state(&self, _scope: &PersistenceScope) -> Result<()> {
            Ok(())
        }

        fn load_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _key: &TransferCheckpointKey,
        ) -> Result<Option<TransferCheckpointRecord>> {
            Ok(None)
        }

        fn save_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _record: &TransferCheckpointRecord,
        ) -> Result<()> {
            Ok(())
        }

        fn clear_transfer_checkpoint(
            &self,
            _scope: &PersistenceScope,
            _key: &TransferCheckpointKey,
        ) -> Result<()> {
            Err(MegaError::Custom(
                "failed to clear transfer checkpoint".to_string(),
            ))
        }
    }

    fn sample_state(hash: &str) -> UploadState {
        UploadState::new(
            "https://upload.example.test".to_string(),
            [0x11; 16],
            [0x22; 8],
            4096,
            "story-3-upload.bin".to_string(),
            "parent-handle".to_string(),
            hash.to_string(),
        )
    }

    fn unique_test_path(name: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system clock should be after unix epoch")
            .as_nanos();
        std::env::temp_dir()
            .join("megalib-story-3-tests")
            .join(format!("{name}-{unique}"))
    }

    fn ensure_parent_dir(path: &Path) {
        let parent = path
            .parent()
            .expect("test path should always have a parent directory");
        fs::create_dir_all(parent).expect("test directory creation should succeed");
    }

    #[test]
    fn resumable_upload_load_ignores_sidecar_files() {
        let source_path = unique_test_path("resumable-sidecar");
        let state_path = UploadState::state_file_path(&source_path);
        ensure_parent_dir(&source_path);

        let runtime_state = sample_state("runtime-hash");
        let sidecar_state = sample_state("sidecar-hash");

        let persistence = PersistenceRuntime::new(Arc::new(
            crate::session::runtime::persistence::MemoryPersistenceBackend::default(),
        ));
        let session = Session::test_dummy().with_persistence_for_tests(persistence);

        session
            .save_persisted_upload_state_for_path(&source_path, &runtime_state)
            .expect("runtime-mirrored upload state should persist");
        sidecar_state
            .save(&state_path)
            .expect("sidecar upload state should save");

        let loaded = session
            .load_resumable_upload_state(&source_path, &runtime_state.file_hash)
            .expect("resumable upload state load should succeed")
            .expect("runtime checkpoint should be available");

        assert_eq!(loaded.file_hash, runtime_state.file_hash);

        let empty_store_loaded = Session::test_dummy()
            .load_resumable_upload_state(&source_path, &runtime_state.file_hash)
            .expect("sidecar-only load should succeed");

        assert!(empty_store_loaded.is_none());
    }

    #[test]
    fn resumable_upload_load_treats_runtime_errors_as_invalid_mirror() {
        let source_path = unique_test_path("resumable-runtime-error");

        let persistence = PersistenceRuntime::new(Arc::new(LoadFailingUploadStateBackend));
        let session = Session::test_dummy().with_persistence_for_tests(persistence);

        let loaded = session
            .load_resumable_upload_state(&source_path, "runtime-hash")
            .expect("runtime mirror errors should be swallowed");

        assert!(loaded.is_none());
    }

    #[test]
    fn resumable_upload_save_propagates_runtime_errors() {
        let source_path = unique_test_path("resumable-save-error");
        let state = sample_state("runtime-hash");
        let persistence = PersistenceRuntime::new(Arc::new(SaveFailingUploadStateBackend));
        let session = Session::test_dummy().with_persistence_for_tests(persistence);

        let err = session
            .save_resumable_upload_state(&source_path, &state)
            .expect_err("save errors should propagate when runtime persistence is authoritative");

        assert!(
            err.to_string()
                .contains("failed to save transfer checkpoint"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn resumable_upload_try_save_keeps_failure_best_effort() {
        let source_path = unique_test_path("resumable-try-save-error");
        let state = sample_state("runtime-hash");
        let persistence = PersistenceRuntime::new(Arc::new(SaveFailingUploadStateBackend));
        let session = Session::test_dummy().with_persistence_for_tests(persistence);

        session.try_save_resumable_upload_state(&source_path, &state);
    }

    #[test]
    fn resumable_upload_clear_propagates_runtime_errors() {
        let source_path = unique_test_path("resumable-clear-error");
        let persistence = PersistenceRuntime::new(Arc::new(ClearFailingUploadStateBackend));
        let session = Session::test_dummy().with_persistence_for_tests(persistence);

        let err = session
            .clear_resumable_upload_state(&source_path, "runtime-hash")
            .expect_err("clear errors should propagate after upload finalization");

        assert!(
            err.to_string()
                .contains("failed to clear transfer checkpoint"),
            "unexpected error: {err}"
        );
    }
}