//! In-memory storage backend for integration tests.

use super::faults::Faults;
use makenotwork::error::{AppError, Result};
use makenotwork::storage::{S3DeleteAuthority, S3Key, StorageBackend};
use std::collections::HashMap;
use std::sync::Mutex;

/// In-memory implementation of `StorageBackend` for tests.
/// Files are stored in a `HashMap<String, Vec<u8>>` behind a `Mutex`.
pub(crate) struct InMemoryStorage {
    objects: Mutex<HashMap<String, Vec<u8>>>,
    /// Open multipart sessions: `upload_id -> s3_key`. Tracked for real (not
    /// stubbed) so the orphan reaper's abort path is observable in tests, an
    /// unaborted session is exactly the leak this models.
    multipart: Mutex<HashMap<String, String>>,
    bucket: String,
    /// Injected transport failures. Empty by default, so a backend with no
    /// policy installed behaves exactly as it did before this existed.
    faults: Faults,
}

#[allow(dead_code)]
impl InMemoryStorage {
    pub(crate) fn new() -> Self {
        InMemoryStorage {
            objects: Mutex::new(HashMap::new()),
            multipart: Mutex::new(HashMap::new()),
            bucket: "test-bucket".to_string(),
            faults: Faults::new(),
        }
    }

    /// The failure policy. Install rules on it to reach the retry and
    /// compensation paths that no test could otherwise enter.
    pub(crate) fn faults(&self) -> &Faults {
        &self.faults
    }

    /// Number of multipart sessions still open. Zero after a clean reap.
    pub(crate) fn open_multipart_count(&self) -> usize {
        self.multipart.lock().unwrap().len()
    }

    /// Open a multipart session directly, standing in for one a client started
    /// and abandoned.
    pub(crate) fn put_open_multipart(&self, upload_id: &str, s3_key: &str) {
        self.multipart
            .lock()
            .unwrap()
            .insert(upload_id.to_string(), s3_key.to_string());
    }

    /// Pre-populate a file so that subsequent `object_exists` / `download_object`
    /// calls see it. Useful for testing confirm_upload flows.
    pub(crate) fn put(&self, key: &str, data: Vec<u8>) {
        self.objects.lock().unwrap().insert(key.to_string(), data);
    }

    /// Retrieve stored bytes for a key. Panics if not found.
    pub(crate) fn get(&self, key: &str) -> Vec<u8> {
        self.objects
            .lock()
            .unwrap()
            .get(key)
            .cloned()
            .expect("key not found in storage")
    }
}

#[async_trait::async_trait]
impl StorageBackend for InMemoryStorage {
    async fn presign_upload(
        &self,
        s3_key: &S3Key,
        _content_type: &str,
        _expiry_secs: Option<u64>,
        _cache_control: Option<&str>,
        _max_bytes: Option<i64>,
    ) -> Result<String> {
        self.faults.check("presign_upload")?;
        Ok(format!("http://test-storage/{s3_key}"))
    }

    async fn presign_download(&self, s3_key: &S3Key, _expiry_secs: Option<u64>) -> Result<String> {
        self.faults.check("presign_download")?;
        if self.objects.lock().unwrap().contains_key(s3_key.as_str()) {
            Ok(format!("http://test-storage/{s3_key}"))
        } else {
            Err(AppError::Storage(format!("Object not found: {s3_key}")))
        }
    }

    async fn object_exists(&self, s3_key: &str) -> Result<bool> {
        self.faults.check("object_exists")?;
        Ok(self.objects.lock().unwrap().contains_key(s3_key))
    }

    async fn object_size(&self, s3_key: &str) -> Result<Option<i64>> {
        self.faults.check("object_size")?;
        Ok(self
            .objects
            .lock()
            .unwrap()
            .get(s3_key)
            .map(|v| v.len() as i64))
    }

    // `download_object_buf` and the two `copy_object_*` wrappers delegate to the
    // methods below them, so a policy on the delegate is what fires and the call
    // count is recorded once. Name the underlying operation in a rule, not the
    // wrapper.
    async fn download_object(&self, s3_key: &str) -> Result<Vec<u8>> {
        self.faults.check("download_object")?;
        self.objects
            .lock()
            .unwrap()
            .get(s3_key)
            .cloned()
            .ok_or_else(|| AppError::Storage(format!("Object not found: {s3_key}")))
    }

    async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes> {
        self.download_object(s3_key).await.map(bytes::Bytes::from)
    }

    async fn download_stream(&self, s3_key: &str) -> Result<s3_storage::ByteStream> {
        self.faults.check("download_stream")?;
        let bytes = self
            .objects
            .lock()
            .unwrap()
            .get(s3_key)
            .cloned()
            .ok_or_else(|| AppError::Storage(format!("Object not found: {s3_key}")))?;
        Ok(s3_storage::ByteStream::from(bytes))
    }

    async fn upload_object(
        &self,
        s3_key: &S3Key,
        _content_type: &str,
        data: Vec<u8>,
        _cache_control: Option<&str>,
    ) -> Result<()> {
        self.faults.check("upload_object")?;
        self.objects
            .lock()
            .unwrap()
            .insert(s3_key.as_str().to_string(), data);
        Ok(())
    }

    async fn upload_multipart(
        &self,
        s3_key: &S3Key,
        _content_type: &str,
        file_path: &std::path::Path,
    ) -> Result<()> {
        self.faults.check("upload_multipart")?;
        let data = tokio::fs::read(file_path)
            .await
            .map_err(|e| AppError::Storage(format!("read multipart source: {e}")))?;
        self.objects
            .lock()
            .unwrap()
            .insert(s3_key.as_str().to_string(), data);
        Ok(())
    }

    async fn delete_object(&self, _auth: &S3DeleteAuthority, s3_key: &S3Key) -> Result<()> {
        self.faults.check("delete_object")?;
        self.objects.lock().unwrap().remove(s3_key.as_str());
        Ok(())
    }

    async fn copy_object(&self, src_key: &S3Key, dst_key: &S3Key) -> Result<()> {
        self.faults.check("copy_object")?;
        let mut objects = self.objects.lock().unwrap();
        let bytes = objects
            .get(src_key.as_str())
            .cloned()
            .ok_or_else(|| AppError::Storage(format!("copy source not found: {src_key}")))?;
        objects.insert(dst_key.as_str().to_string(), bytes);
        Ok(())
    }

    async fn copy_object_from(
        &self,
        _src_bucket: &str,
        src_key: &S3Key,
        dst_key: &S3Key,
    ) -> Result<()> {
        // The in-memory store is one flat map with no bucket isolation, so the
        // cross-bucket promote is just a same-map copy. Tests wire `public_s3` to
        // the same backend as `s3` so the staging source resolves here.
        self.copy_object(src_key, dst_key).await
    }

    async fn copy_object_multipart(
        &self,
        _src_bucket: &str,
        src_key: &S3Key,
        dst_key: &S3Key,
        _content_type: &str,
        _src_size: u64,
        _part_size: Option<usize>,
    ) -> Result<()> {
        // The flat map has no 5 GiB single-copy limit, so the multipart promote
        // is the same copy. Implemented for real (not stubbed) so a promote test
        // gets the same observable result down either branch.
        self.copy_object(src_key, dst_key).await
    }

    // Client-direct multipart sessions
    //
    // Stubbed the same way presigned single-PUT uploads already are: the URLs
    // this backend hands out are fake, so no client bytes can flow back into the
    // map. Tests simulate the finished object with `put()` and then exercise the
    // confirm path, exactly as they do for `presign_upload` + confirm.

    async fn create_multipart_upload(&self, s3_key: &S3Key, _content_type: &str) -> Result<String> {
        self.faults.check("create_multipart_upload")?;
        let upload_id = format!("test-upload-id/{s3_key}");
        self.put_open_multipart(&upload_id, s3_key.as_str());
        Ok(upload_id)
    }

    async fn presign_upload_part(
        &self,
        s3_key: &S3Key,
        upload_id: &str,
        part_number: i32,
        _expiry_secs: Option<u64>,
        _max_bytes: Option<i64>,
        checksum_sha256: Option<&str>,
    ) -> Result<String> {
        self.faults.check("presign_upload_part")?;
        // Mirror the production range check so a bad part number fails in tests
        // the same way it would against S3.
        if !(1..=s3_storage::MULTIPART_MAX_PARTS as i32).contains(&part_number) {
            return Err(AppError::Storage(format!(
                "part number {part_number} out of range 1..={}",
                s3_storage::MULTIPART_MAX_PARTS
            )));
        }
        // Echo the bound checksum into the URL so tests can assert it reached
        // the signer, standing in for the SignedHeaders a real presign carries.
        let checksum = checksum_sha256
            .map(|c| format!("&checksum={c}"))
            .unwrap_or_default();
        Ok(format!(
            "http://test-storage/{s3_key}?uploadId={upload_id}&partNumber={part_number}{checksum}"
        ))
    }

    async fn complete_multipart_upload(
        &self,
        _s3_key: &S3Key,
        upload_id: &str,
        parts: &[(i32, String)],
    ) -> Result<()> {
        self.faults.check("complete_multipart_upload")?;
        if parts.is_empty() {
            return Err(AppError::Storage(
                "cannot complete a multipart upload with no parts".to_string(),
            ));
        }
        // Completing closes the session, so it no longer holds billed parts.
        self.multipart.lock().unwrap().remove(upload_id);
        Ok(())
    }

    async fn abort_multipart_upload(&self, _s3_key: &S3Key, upload_id: &str) -> Result<()> {
        self.faults.check("abort_multipart_upload")?;
        // Idempotent: aborting an unknown session is fine.
        self.multipart.lock().unwrap().remove(upload_id);
        Ok(())
    }

    async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>> {
        self.faults.check("list_multipart_uploads_for_key")?;
        Ok(self
            .multipart
            .lock()
            .unwrap()
            .iter()
            .filter(|(_, key)| key.as_str() == s3_key)
            .map(|(id, _)| id.clone())
            .collect())
    }

    async fn check_connectivity(&self) -> std::result::Result<(), String> {
        // Health checks report a string, not an `AppError`, so the injected
        // error is rendered rather than propagated.
        self.faults
            .check("check_connectivity")
            .map_err(|e| e.to_string())
    }

    fn bucket(&self) -> &str {
        &self.bucket
    }
}