iris-chat 0.1.62

use super::*;
use crate::state::AttachmentDownloadResult;
use async_trait::async_trait;
use base64::Engine;
use fips_core::FipsEndpoint;
use hashtree_blossom::BlossomClient;
use hashtree_config::Config as HashtreeConfig;
use hashtree_core::{
    nhash_decode, nhash_encode_full, to_hex, BlobRoute, Cid, Hash, HashTree, HashTreeConfig,
    NHashData, Store, StoreBlobRoute, StoreError,
};
use hashtree_fips_transport::{FipsBlobRoute, TcpBlobTransport, TcpBlobTransportConfig};
use hashtree_network::{BlobRouteEntry, BlobRouter, BlobRouterConfig, RoutedStore};
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{OnceLock, RwLock, Weak};
use std::time::Duration;
use tokio::io::AsyncReadExt;
use tokio::sync::RwLock as AsyncRwLock;
use tokio_util::compat::TokioAsyncReadCompatExt;

mod local_store;
mod message_commit;
mod provider_reads;
mod provider_route;
mod readiness;
mod replication;
use local_store::AttachmentChunkStore;
pub(in crate::core) use provider_reads::ProviderReadOptions;
pub(super) use replication::AttachmentService;

fn active_local_store() -> &'static RwLock<Option<Weak<AttachmentChunkStore>>> {
    static STORE: OnceLock<RwLock<Option<Weak<AttachmentChunkStore>>>> = OnceLock::new();
    STORE.get_or_init(|| RwLock::new(None))
}

fn current_local_store() -> Option<Arc<AttachmentChunkStore>> {
    active_local_store()
        .read()
        .unwrap_or_else(|p| p.into_inner())
        .as_ref()
        .and_then(Weak::upgrade)
}

pub(super) struct AttachmentBlobRuntime {
    store: Arc<RoutedStore<AttachmentChunkStore>>,
    _transport: Arc<TcpBlobTransport<AttachmentChunkStore>>,
    reads: Arc<provider_reads::ProviderReads>,
    provider: Arc<provider_route::ProviderRoute>,
    siblings: Arc<FipsBlobRoute<AttachmentChunkStore>>,
}

impl AttachmentBlobRuntime {
    #[cfg(test)]
    pub(in crate::core) fn provider_read_endpoint_for_test(&self) -> Arc<FipsEndpoint> {
        self.reads.endpoint()
    }

    pub(in crate::core) fn update_store(&self) -> Arc<dyn Store> {
        self.store.clone()
    }

    pub(super) async fn set_peers(
        &self,
        peers: Vec<fips_core::PeerIdentity>,
        siblings: Vec<fips_core::PeerIdentity>,
    ) -> anyhow::Result<()> {
        // Apply the admission barrier before publishing the new sibling/cache
        // policy. An active unrelated provider remains usable after linking.
        let result = self.reads.set_siblings(siblings.clone()).await;
        if let Err(error) = &result {
            crate::perflog!("attachment.provider.admission.error={error}");
        }
        self.provider.set_explicit_peers(
            peers
                .into_iter()
                .filter(|peer| !siblings.contains(peer))
                .collect(),
        );
        self.siblings.set_explicit_peers(siblings);
        result
    }

    pub(super) async fn update_read_peers(&self, peers: Vec<fips_core::config::PeerConfig>) {
        self.reads.update_peers(peers);
    }

    pub(super) async fn shutdown_reads(&self) {
        self.reads.shutdown().await;
    }
}

pub(super) async fn bind_same_host_attachment_store(
    endpoint: Arc<FipsEndpoint>,
    read_options: ProviderReadOptions,
    standalone: Arc<dyn BlobRoute>,
    peers: Vec<fips_core::PeerIdentity>,
    siblings: Vec<fips_core::PeerIdentity>,
    primary: Arc<AttachmentChunkStore>,
) -> anyhow::Result<Arc<AttachmentBlobRuntime>> {
    // A public provider can otherwise probe a known upload hash at this device
    // and identify its recipient, even when the original download was anonymous.
    let allowed_siblings = Arc::new(RwLock::new(siblings.clone()));
    let admission = allowed_siblings.clone();
    let transport = Arc::new(
        TcpBlobTransport::bind_advertised_route_with_config_and_policy(
            endpoint.clone(),
            primary.clone(),
            Arc::new(StoreBlobRoute::new(primary.clone())),
            TcpBlobTransportConfig::default(),
            25,
            Arc::new(move |peer| {
                admission
                    .read()
                    .unwrap_or_else(|error| error.into_inner())
                    .contains(&peer)
            }),
        )
        .await?,
    );
    let reads = Arc::new(
        provider_reads::ProviderReads::bind(
            read_options,
            primary.clone(),
            endpoint.npub().to_string(),
            allowed_siblings.clone(),
        )
        .await?,
    );
    let provider = Arc::new(provider_route::ProviderRoute::new(
        endpoint,
        reads.clone(),
        peers,
    )?);
    // Only verified same-account devices may use the identified read path.
    // Ordinary contacts/providers never enter this route, including retries.
    let siblings = Arc::new(FipsBlobRoute::explicit(transport.clone(), siblings, 4)?);
    let router = Arc::new(BlobRouter::new(
        vec![
            BlobRouteEntry::new(
                "attachment-cache",
                Arc::new(StoreBlobRoute::new(primary.clone())),
            ),
            BlobRouteEntry::new(
                "linked-devices",
                super::support::offline_lan::route(siblings.clone(), true),
            ),
            BlobRouteEntry::new(
                "same-host-fips",
                super::support::offline_lan::route(provider.clone(), true),
            ),
            BlobRouteEntry::new(
                "blossom",
                super::support::offline_lan::route(standalone, false),
            ),
        ],
        Some(primary.clone()),
        BlobRouterConfig::default(),
    )?);
    let runtime = Arc::new(AttachmentBlobRuntime {
        store: Arc::new(RoutedStore::new(primary, router)),
        _transport: transport,
        reads,
        provider,
        siblings,
    });
    runtime.store.primary().transport_ready(&runtime.store)?;
    Ok(runtime)
}

pub(super) async fn start_same_host_attachment_reuse(
    endpoint: Arc<FipsEndpoint>,
    read_options: ProviderReadOptions,
    peers: Vec<fips_core::PeerIdentity>,
    siblings: Vec<fips_core::PeerIdentity>,
    primary: Arc<AttachmentChunkStore>,
) -> anyhow::Result<Arc<AttachmentBlobRuntime>> {
    let standalone: Arc<dyn BlobRoute> = Arc::new(StoreBlobRoute::new(
        blossom_read_store_with_local(Some(primary.clone()))?,
    ));
    bind_same_host_attachment_store(endpoint, read_options, standalone, peers, siblings, primary)
        .await
}

impl AppCore {
    pub(super) fn send_attachment(
        &mut self,
        chat_id: &str,
        file_path: &str,
        filename: &str,
        caption: &str,
    ) {
        self.send_attachments(
            chat_id,
            &[OutgoingAttachment {
                file_path: file_path.to_string(),
                filename: filename.to_string(),
            }],
            caption,
        );
    }

    pub(super) fn send_attachments(
        &mut self,
        chat_id: &str,
        attachments: &[OutgoingAttachment],
        caption: &str,
    ) {
        let chat_id = chat_id.trim();
        if chat_id.is_empty() || attachments.is_empty() {
            self.state.toast = Some("Attachment could not be sent.".to_string());
            self.emit_state();
            return;
        }
        if self.state.busy.uploading_attachment {
            self.state.toast = Some("Attachment upload already in progress.".to_string());
            self.emit_state();
            return;
        }
        if self.logged_in.is_none() {
            self.state.toast = Some("Create or restore a profile first.".to_string());
            self.emit_state();
            return;
        }
        if !self.can_use_chats() {
            self.state.toast = Some(chat_unavailable_message(self.logged_in.as_ref()).to_string());
            self.emit_state();
            return;
        }
        let Some(normalized_chat_id) = self.normalize_chat_id(chat_id) else {
            self.state.toast = Some("Invalid chat id.".to_string());
            self.emit_state();
            return;
        };
        if self.reject_removed_group_send(&normalized_chat_id) {
            return;
        }
        let prepared = match prepare_outgoing_attachments(attachments) {
            Ok(prepared) => prepared,
            Err(message) => {
                self.state.toast = Some(message.to_string());
                self.emit_state();
                return;
            }
        };

        let store = match self.ensure_attachment_service() {
            Ok(store) => store,
            Err(error) => {
                self.push_debug_log("attachment.storage.open.error", error.to_string());
                self.state.toast = Some("Attachment could not be saved.".to_string());
                self.emit_state();
                return;
            }
        };
        let store_id = store.id;
        let caption = caption.trim().to_string();
        let sender = self.core_sender.clone();
        let upload_chat_id = normalized_chat_id.clone();
        let upload_attachments = prepared.clone();

        self.push_debug_log(
            "attachment.upload.start",
            format!(
                "chat_id={} count={} files={}",
                normalized_chat_id,
                prepared.len(),
                prepared
                    .iter()
                    .map(|attachment| attachment.filename.as_str())
                    .collect::<Vec<_>>()
                    .join(",")
            ),
        );
        self.active_chat_id = Some(normalized_chat_id.clone());
        self.screen_stack = vec![Screen::Chat {
            chat_id: normalized_chat_id,
        }];
        let total_bytes: u64 = prepared
            .iter()
            .map(|attachment| attachment.size_bytes)
            .sum();
        self.state.busy.uploading_attachment = true;
        self.state.busy.upload_progress = Some(crate::state::UploadProgress {
            bytes_uploaded: 0,
            total_bytes,
        });
        self.rebuild_state();
        self.emit_state();

        self.runtime.spawn(async move {
            let bytes_uploaded = Arc::new(AtomicU64::new(0));
            let progress_sender = sender.clone();
            let progress_counter = bytes_uploaded.clone();
            let progress_total = total_bytes;
            let progress_handle = tokio::spawn(async move {
                let mut last_reported: u64 = u64::MAX;
                loop {
                    tokio::time::sleep(Duration::from_millis(120)).await;
                    let current = progress_counter.load(Ordering::Relaxed);
                    if current == last_reported {
                        continue;
                    }
                    last_reported = current;
                    let _ = progress_sender.send(CoreMsg::Internal(Box::new(
                        InternalEvent::AttachmentUploadProgress {
                            store_id,
                            bytes_uploaded: current,
                            total_bytes: progress_total,
                        },
                    )));
                }
            });
            let result =
                upload_files_locally(store, &upload_attachments, Some(bytes_uploaded.clone()))
                    .await
                    .map(|uploaded| format_attachment_links_message(&caption, &uploaded))
                    .map_err(|error| error.to_string());
            progress_handle.abort();
            let _ = sender.send(CoreMsg::Internal(Box::new(
                InternalEvent::AttachmentUploadFinished {
                    store_id,
                    chat_id: upload_chat_id,
                    result,
                },
            )));
        });
    }

    pub(super) fn handle_attachment_upload_progress(
        &mut self,
        bytes_uploaded: u64,
        total_bytes: u64,
    ) {
        if !self.state.busy.uploading_attachment {
            return;
        }
        let clamped = if total_bytes == 0 {
            bytes_uploaded
        } else {
            bytes_uploaded.min(total_bytes)
        };
        self.state.busy.upload_progress = Some(crate::state::UploadProgress {
            bytes_uploaded: clamped,
            total_bytes,
        });
        self.emit_state();
    }

    pub(super) fn handle_attachment_upload_finished(
        &mut self,
        chat_id: String,
        result: Result<String, String>,
    ) {
        self.state.busy.uploading_attachment = false;
        self.state.busy.upload_progress = None;
        match result {
            Ok(message_text) => {
                self.push_debug_log(
                    "attachment.upload.finish",
                    format!("chat_id={} success=true", chat_id),
                );
                self.send_prepared_attachment_message(&chat_id, &message_text);
            }
            Err(error) => {
                self.push_debug_log(
                    "attachment.upload.finish",
                    format!("chat_id={} success=false error={}", chat_id, error),
                );
                self.state.toast = Some("Attachment upload failed.".to_string());
                self.emit_state();
            }
        }
    }
}

#[uniffi::export]
pub fn download_hashtree_attachment(nhash: String) -> AttachmentDownloadResult {
    download_hashtree_attachment_with_limit(nhash, MAX_ATTACHMENT_BYTES)
}

const MAX_ATTACHMENT_BYTES: u64 = 64 * 1024 * 1024;

/// Automatic previews may request a lower ceiling; callers cannot raise the app limit.
#[uniffi::export]
pub fn download_hashtree_attachment_with_limit(
    nhash: String,
    max_bytes: u64,
) -> AttachmentDownloadResult {
    match download_hashtree_attachment_blocking(&nhash, max_bytes.min(MAX_ATTACHMENT_BYTES)) {
        Ok(data_base64) => AttachmentDownloadResult {
            data_base64: Some(data_base64),
            error: None,
        },
        Err(error) => AttachmentDownloadResult {
            data_base64: None,
            error: Some(error.to_string()),
        },
    }
}

#[derive(Clone, Debug)]
struct PreparedOutgoingAttachment {
    file_path: PathBuf,
    filename: String,
    size_bytes: u64,
}

fn prepare_outgoing_attachments(
    attachments: &[OutgoingAttachment],
) -> Result<Vec<PreparedOutgoingAttachment>, &'static str> {
    if attachments.len() > local_store::MAX_UPLOADS {
        return Err("Select up to 32 attachments at once.");
    }
    let mut prepared = Vec::with_capacity(attachments.len());
    for attachment in attachments {
        let file_path = PathBuf::from(attachment.file_path.trim());
        if file_path.as_os_str().is_empty() {
            return Err("Attachment could not be sent.");
        }
        let metadata = file_path
            .metadata()
            .map_err(|_| "Attachment file was not found.")?;
        if !metadata.is_file() {
            return Err("Attachment file was not found.");
        }
        if metadata.len() > MAX_ATTACHMENT_BYTES {
            return Err("Attachment is too large (64 MB maximum).");
        }
        prepared.push(PreparedOutgoingAttachment {
            filename: display_filename(&attachment.filename, &file_path),
            file_path,
            size_bytes: metadata.len(),
        });
    }
    Ok(prepared)
}

pub(super) fn display_filename(filename: &str, file_path: &Path) -> String {
    let from_input = filename.trim();
    let candidate = if from_input.is_empty() {
        file_path.file_name().and_then(|value| value.to_str())
    } else {
        Path::new(from_input)
            .file_name()
            .and_then(|value| value.to_str())
    };
    candidate
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or("attachment")
        .to_string()
}

async fn upload_files_locally(
    store: Arc<AttachmentChunkStore>,
    attachments: &[PreparedOutgoingAttachment],
    progress: Option<Arc<AtomicU64>>,
) -> anyhow::Result<Vec<(String, String)>> {
    let servers = replication::bounded_servers(&blossom_servers_from_config().1);
    let mut staged = Vec::with_capacity(attachments.len());
    let mut uploaded = Vec::with_capacity(attachments.len());
    for attachment in attachments {
        let session = store.begin(progress.clone())?;
        let nhash = upload_file_to_store(&attachment.file_path, session.clone()).await?;
        staged.push((session, nhash.clone()));
        uploaded.push((nhash, attachment.filename.clone()));
    }
    if let Err(error) = store.finish_batch(&staged, &servers) {
        for (_, nhash) in &staged {
            if let Ok(data) = nhash_decode(nhash) {
                // Never remove an existing committed message reference.
                let _ = store.resolve_staged(&data.hash, false);
            }
        }
        return Err(error);
    }
    Ok(uploaded)
}

pub(super) async fn upload_file_to_hashtree(
    secret_hex: &str,
    path: &Path,
    progress: Option<Arc<AtomicU64>>,
) -> anyhow::Result<String> {
    let secret_key = nostr::SecretKey::from_hex(secret_hex)
        .map_err(|error| anyhow::anyhow!("invalid upload key: {error}"))?;
    let keys = nostr::Keys::new(secret_key);
    let (read_servers, write_servers) = blossom_servers_from_config();
    if write_servers.is_empty() {
        anyhow::bail!("no hashtree write servers configured");
    }

    let store = Arc::new(
        UploadingBlossomStore::new(keys, read_servers, write_servers, progress)?
            .with_local(current_local_store()),
    );
    upload_file_to_store(path, store).await
}

pub(super) async fn upload_invite_record(
    secret_hex: &str,
    record: &crate::invite_record::InviteRecord,
) -> anyhow::Result<String> {
    let (reference, ciphertext) = crate::invite_record::encode_record(record)?;
    let keys = Keys::new(nostr::SecretKey::from_hex(secret_hex)?);
    let (read, write) = blossom_servers_from_config();
    anyhow::ensure!(!write.is_empty(), "No file servers configured.");
    let store = UploadingBlossomStore::new(keys, read, write, None)?;
    // Return a shareable link only after a server acknowledged the ciphertext.
    tokio::time::timeout(
        Duration::from_secs(30),
        store.put(hashtree_core::sha256(&ciphertext), ciphertext),
    )
    .await??;
    Ok(format!("https://chat.iris.to/#/{reference}"))
}

async fn upload_file_to_store<S: Store + 'static>(
    path: &Path,
    store: Arc<S>,
) -> anyhow::Result<String> {
    let tree = HashTree::new(HashTreeConfig {
        encrypted: true,
        ..HashTreeConfig::new(store)
    });
    let file = tokio::fs::File::open(path).await?;
    let metadata = file.metadata().await?;
    if !metadata.is_file() || metadata.len() > MAX_ATTACHMENT_BYTES {
        anyhow::bail!("attachment is too large or is not a regular file");
    }
    // Recheck the opened file and cap the stream in case it grows during upload.
    // Never send a link that the attachment download limit makes unusable.
    let (cid, size) = tree
        .put_stream(file.take(MAX_ATTACHMENT_BYTES + 1).compat())
        .await
        .map_err(|error| anyhow::anyhow!("hashtree upload failed: {error}"))?;
    if size > MAX_ATTACHMENT_BYTES {
        anyhow::bail!("attachment is too large");
    }

    nhash_encode_full(&NHashData {
        hash: cid.hash,
        decrypt_key: cid.key,
    })
    .map_err(|error| anyhow::anyhow!("nhash encode failed: {error}"))
}

pub(super) async fn upload_profile_picture_to_hashtree(
    secret_hex: &str,
    path: &Path,
) -> anyhow::Result<String> {
    let data = tokio::fs::read(path).await?;
    if data.is_empty() {
        anyhow::bail!("profile picture is empty");
    }
    if data.len() > 10 * 1024 * 1024 {
        anyhow::bail!("profile picture is too large");
    }
    if !looks_like_image(path, &data) {
        anyhow::bail!("profile picture must be an image");
    }
    let nhash = upload_file_to_hashtree(secret_hex, path, None).await?;
    Ok(format!("htree://{nhash}"))
}

pub(super) fn looks_like_image(path: &Path, data: &[u8]) -> bool {
    let extension = path
        .extension()
        .and_then(|value| value.to_str())
        .unwrap_or_default()
        .to_ascii_lowercase();
    matches!(
        extension.as_str(),
        "avif" | "bmp" | "gif" | "heic" | "heif" | "jpg" | "jpeg" | "png" | "webp"
    ) || data.starts_with(b"\x89PNG")
        || data.starts_with(b"\xff\xd8\xff")
        || data.starts_with(b"GIF87a")
        || data.starts_with(b"GIF89a")
        || data.starts_with(b"RIFF")
}

fn download_hashtree_attachment_blocking(nhash: &str, max_bytes: u64) -> anyhow::Result<String> {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()?;
    runtime.block_on(download_hashtree_attachment_base64_with_limit(
        nhash, max_bytes,
    ))
}

#[cfg(test)]
pub(super) async fn download_hashtree_attachment_base64(nhash: &str) -> anyhow::Result<String> {
    download_hashtree_attachment_base64_with_limit(nhash, MAX_ATTACHMENT_BYTES).await
}

async fn download_hashtree_attachment_base64_with_limit(
    nhash: &str,
    max_bytes: u64,
) -> anyhow::Result<String> {
    let nhash = nhash.trim();
    if nhash.is_empty() {
        anyhow::bail!("missing attachment hash");
    }
    let data = nhash_decode(nhash).map_err(|error| anyhow::anyhow!("invalid nhash: {error}"))?;
    let cid = Cid {
        hash: data.hash,
        key: data.decrypt_key,
    };
    // Capture one account before starting HTTP; a later account switch must
    // neither replace its cache nor supply a different account's peer route.
    let local = current_local_store();
    let fallback = blossom_read_store_with_local(local.clone())?;
    let bytes = readiness::read(&cid, local, fallback, max_bytes, Duration::from_secs(8)).await?;
    Ok(base64::engine::general_purpose::STANDARD.encode(bytes))
}

async fn read_hashtree_attachment<S: Store + 'static>(
    cid: &Cid,
    store: Arc<S>,
    max_bytes: u64,
) -> anyhow::Result<Vec<u8>> {
    let tree = HashTree::new(HashTreeConfig::new(store));
    let bytes = tree
        .get(cid, Some(max_bytes.min(MAX_ATTACHMENT_BYTES)))
        .await
        .map_err(|error| anyhow::anyhow!("hashtree download failed: {error}"))?
        .ok_or_else(|| anyhow::anyhow!("attachment was not found"))?;
    Ok(bytes)
}

fn blossom_read_store_with_local(
    local: Option<Arc<AttachmentChunkStore>>,
) -> anyhow::Result<Arc<UploadingBlossomStore>> {
    let (read_servers, write_servers) = blossom_servers_from_config();
    Ok(Arc::new(
        UploadingBlossomStore::new(
            nostr::Keys::generate(),
            merge_read_servers(read_servers, &write_servers),
            Vec::new(),
            None,
        )?
        .with_local(local),
    ))
}

pub(in crate::core) fn blossom_servers_from_config() -> (Vec<String>, Vec<String>) {
    let cfg = HashtreeConfig::load_or_default();
    let mut read = cfg.blossom.all_read_servers();
    let configured_write = cfg.blossom.all_write_servers();
    let mut write: Vec<String> = configured_write
        .iter()
        .filter(|server| !is_local_server_url(server))
        .cloned()
        .collect();
    if write.is_empty() {
        write = configured_write;
    }

    if let Some(local_url) =
        hashtree_config::detect_local_daemon_url(Some(&cfg.server.bind_address))
    {
        if !read.iter().any(|server| server == &local_url) {
            read.insert(0, local_url);
        }
    }

    read = merge_read_servers(read, &write);
    (read, write)
}

fn merge_read_servers(mut read: Vec<String>, write: &[String]) -> Vec<String> {
    let mut seen: HashSet<String> = read.iter().cloned().collect();
    for server in write {
        if seen.insert(server.clone()) {
            read.push(server.clone());
        }
    }
    read
}

fn is_local_server_url(value: &str) -> bool {
    let Ok(parsed) = url::Url::parse(value) else {
        return false;
    };
    let Some(host) = parsed.host_str() else {
        return false;
    };
    matches!(
        host.trim_matches(['[', ']']).to_ascii_lowercase().as_str(),
        "localhost" | "127.0.0.1" | "::1"
    )
}

async fn download_attachment_blob(
    client: &reqwest::Client,
    server: &str,
    hash: &str,
    max_bytes: u64,
) -> anyhow::Result<Option<Vec<u8>>> {
    let url = format!("{}/{}.bin", server.trim_end_matches('/'), hash);
    let mut response = client.get(url).send().await?;
    if !response.status().is_success() {
        return Ok(None);
    }
    if response
        .content_length()
        .is_some_and(|size| size > max_bytes)
    {
        anyhow::bail!("attachment blob is too large");
    }
    let mut bytes = Vec::new();
    while let Some(chunk) = response.chunk().await? {
        if (bytes.len() as u64).saturating_add(chunk.len() as u64) > max_bytes {
            anyhow::bail!("attachment blob is too large");
        }
        bytes.extend_from_slice(&chunk);
    }
    if hashtree_blossom::compute_sha256(&bytes) != hash {
        anyhow::bail!("attachment hash mismatch");
    }
    Ok(Some(bytes))
}

struct UploadingBlossomStore {
    client: BlossomClient,
    download_client: reqwest::Client,
    uploaded: AsyncRwLock<HashSet<String>>,
    progress: Option<Arc<AtomicU64>>,
    local: Option<Arc<AttachmentChunkStore>>,
}

impl UploadingBlossomStore {
    fn new(
        keys: nostr::Keys,
        read_servers: Vec<String>,
        write_servers: Vec<String>,
        progress: Option<Arc<AtomicU64>>,
    ) -> anyhow::Result<Self> {
        let client = BlossomClient::new_empty(keys)
            .with_read_servers(read_servers)
            .with_write_servers(write_servers);
        Ok(Self {
            client,
            download_client: reqwest::Client::builder()
                .timeout(Duration::from_secs(30))
                .build()?,
            uploaded: AsyncRwLock::new(HashSet::new()),
            progress,
            local: None,
        })
    }

    fn with_local(mut self, local: Option<Arc<AttachmentChunkStore>>) -> Self {
        self.local = local;
        self
    }
}

#[async_trait]
impl Store for UploadingBlossomStore {
    async fn put(&self, hash: Hash, data: Vec<u8>) -> Result<bool, StoreError> {
        let hash_hex = to_hex(&hash);
        let computed = hashtree_blossom::compute_sha256(&data);
        if computed != hash_hex {
            return Err(StoreError::Other(
                "hash mismatch for blossom upload".to_string(),
            ));
        }

        {
            let uploaded = self.uploaded.read().await;
            if uploaded.contains(&hash_hex) {
                return Ok(false);
            }
        }

        let chunk_size = data.len() as u64;
        let remote_hash = self
            .client
            .upload(&data)
            .await
            .map_err(|error| StoreError::Other(format!("blossom upload failed: {error}")))?;
        if remote_hash != hash_hex {
            return Err(StoreError::Other(format!(
                "remote hash mismatch: expected {hash_hex}, got {remote_hash}"
            )));
        }

        if let Some(local) = &self.local {
            // The strict path has a durable remote copy; caching is best effort.
            let _ = local.put(hash, data).await;
        }
        let mut uploaded = self.uploaded.write().await;
        uploaded.insert(hash_hex);
        if let Some(progress) = &self.progress {
            progress.fetch_add(chunk_size, Ordering::Relaxed);
        }
        Ok(true)
    }

    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
        let key = to_hex(hash);
        if let Some(local) = &self.local {
            if let Ok(Some(data)) = local.get(hash).await {
                return Ok(Some(data));
            }
        }

        // Bound the HTTP body as well as the assembled tree. A single oversized
        // blob or a server that omits Content-Length must not fill memory first.
        for server in self.client.read_servers() {
            if let Ok(Some(data)) = download_attachment_blob(
                &self.download_client,
                server,
                &key,
                MAX_ATTACHMENT_BYTES + 64,
            )
            .await
            {
                if let Some(local) = &self.local {
                    let _ = local.put(*hash, data.clone()).await;
                }
                return Ok(Some(data));
            }
        }
        Ok(None)
    }

    async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
        let key = to_hex(hash);
        if let Some(local) = &self.local {
            if local.has(hash).await.unwrap_or(false) {
                return Ok(true);
            }
        }

        for server in self.client.read_servers() {
            if self.client.exists_on_server(&key, server).await {
                return Ok(true);
            }
        }

        Ok(false)
    }

    async fn delete(&self, hash: &Hash) -> Result<bool, StoreError> {
        let key = to_hex(hash);
        let mut removed = match &self.local {
            Some(local) => local.delete(hash).await.unwrap_or(false),
            None => false,
        };
        let mut uploaded = self.uploaded.write().await;
        removed |= uploaded.remove(&key);
        Ok(removed)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn display_filename_uses_input_basename_then_path_fallback() {
        assert_eq!(
            display_filename("nested/photo.png", Path::new("/tmp/source.bin")),
            "photo.png"
        );
        assert_eq!(
            display_filename("", Path::new("/tmp/source.bin")),
            "source.bin"
        );
        assert_eq!(display_filename("", Path::new("/")), "attachment");
    }

    #[test]
    fn prepares_multiple_outgoing_attachments() {
        let dir = tempfile::tempdir().expect("tempdir");
        let first = dir.path().join("first.bin");
        let second = dir.path().join("second.bin");
        fs::write(&first, b"one").expect("write first");
        fs::write(&second, b"two").expect("write second");

        let prepared = prepare_outgoing_attachments(&[
            OutgoingAttachment {
                file_path: first.to_string_lossy().to_string(),
                filename: "nested/photo.png".to_string(),
            },
            OutgoingAttachment {
                file_path: second.to_string_lossy().to_string(),
                filename: String::new(),
            },
        ])
        .expect("prepared");

        assert_eq!(prepared.len(), 2);
        assert_eq!(prepared[0].filename, "photo.png");
        assert_eq!(prepared[0].size_bytes, 3);
        assert_eq!(prepared[1].filename, "second.bin");
        assert_eq!(prepared[1].size_bytes, 3);
    }

    #[test]
    fn outgoing_attachments_obey_the_download_size_limit() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("large.bin");
        let file = fs::File::create(&path).unwrap();
        let attachment = OutgoingAttachment {
            file_path: path.to_string_lossy().into_owned(),
            filename: String::new(),
        };

        file.set_len(MAX_ATTACHMENT_BYTES).unwrap();
        let prepared = prepare_outgoing_attachments(std::slice::from_ref(&attachment)).unwrap();
        assert_eq!(prepared[0].size_bytes, MAX_ATTACHMENT_BYTES);

        file.set_len(MAX_ATTACHMENT_BYTES + 1).unwrap();
        assert_eq!(
            prepare_outgoing_attachments(&[attachment]).unwrap_err(),
            "Attachment is too large (64 MB maximum)."
        );
    }
}

#[cfg(test)]
#[path = "attachment_upload/upload_tests.rs"]
mod upload_tests;

#[cfg(test)]
#[path = "attachment_upload/same_host_tests.rs"]
mod same_host_tests;

#[cfg(test)]
mod storage_tests;

#[cfg(test)]
mod lifecycle_tests;

#[cfg(test)]
mod read_pressure_tests;