Complete paid-file caching and deliver LND waiting UI to existing nodes

core/archipelago/src/api/rpc/content.rs

@@ -43,6 +43,90 @@ async fn reclaim_spent_ecash(data_dir: &std::path::Path, token: &str, backend: &
     }
 }
 
+/// Keep first purchases and cached repeats compatible with both existing clients.
+fn paid_content_response(bytes: &[u8], mime: &str, paid_sats: u64) -> serde_json::Value {
+    use base64::Engine;
+    let data = base64::engine::general_purpose::STANDARD.encode(bytes);
+    serde_json::json!({
+        "data": data, "data_base64": data,
+        "size": bytes.len(), "size_bytes": bytes.len(),
+        "mime_type": mime, "paid_sats": paid_sats, "owned": true,
+    })
+}
+
+/// FileBrowser owns its files through a rootless UID mapping. Use its authenticated
+/// API rather than writing host paths with the backend's unrelated UID. Its
+/// override=false upload atomically refuses existing names, including races.
+async fn file_purchase_in_files(
+    client: &reqwest::Client,
+    base_url: &str,
+    token: &str,
+    filename: &str,
+    mime: &str,
+    bytes: &[u8],
+) -> Result<String> {
+    let folder = if mime.starts_with("image/") || mime.starts_with("video/") {
+        "Photos"
+    } else if mime.starts_with("audio/") {
+        "Music"
+    } else {
+        "Documents"
+    };
+    let mut folder_url = reqwest::Url::parse(base_url)?;
+    folder_url
+        .path_segments_mut()
+        .map_err(|_| anyhow::anyhow!("Invalid Files URL"))?
+        .extend(["api", "resources", folder, ""]);
+    let response = client
+        .get(folder_url.clone())
+        .header("X-Auth", token)
+        .send()
+        .await?;
+    if response.status() == reqwest::StatusCode::NOT_FOUND {
+        let response = client
+            .post(folder_url.clone())
+            .header("X-Auth", token)
+            .send()
+            .await?;
+        if response.status() != reqwest::StatusCode::CONFLICT {
+            response.error_for_status()?;
+        }
+    } else {
+        response.error_for_status()?;
+    }
+    let base = std::path::Path::new(filename)
+        .file_name()
+        .and_then(|n| n.to_str())
+        .filter(|n| !n.is_empty())
+        .unwrap_or("download");
+    let (stem, extension) = match base.rsplit_once('.') {
+        Some((stem, ext)) if !stem.is_empty() => (stem, format!(".{ext}")),
+        _ => (base, String::new()),
+    };
+    for attempt in 1..=100 {
+        let name = if attempt == 1 {
+            base.to_string()
+        } else {
+            format!("{stem} ({attempt}){extension}")
+        };
+        let mut url = folder_url.clone();
+        url.path_segments_mut().unwrap().pop_if_empty().push(&name);
+        url.query_pairs_mut().append_pair("override", "false");
+        let response = client
+            .post(url)
+            .header("X-Auth", token)
+            .body(bytes.to_vec())
+            .send()
+            .await?;
+        if response.status() == reqwest::StatusCode::CONFLICT {
+            continue;
+        }
+        response.error_for_status()?;
+        return Ok(format!("{folder}/{name}"));
+    }
+    anyhow::bail!("Too many existing copies; purchased file remains in the purchase cache")
+}
+
 impl RpcHandler {
     /// List content I'm sharing.
     pub(super) async fn handle_content_list_mine(&self) -> Result<serde_json::Value> {
@@ -464,17 +548,10 @@ impl RpcHandler {
                     crate::content_owned::read_owned(&self.config.data_dir, &o.onion, &o.content_id)
                         .await
                 {
-                    use base64::Engine;
-                    return Ok(serde_json::json!({
-                        "owned": true,
-                        "already_owned": true,
-                        "filename": o.filename,
-                        "mime_type": mime,
-                        "size_bytes": bytes.len(),
-                        "paid_sats": 0,
-                        "data_base64":
-                            base64::engine::general_purpose::STANDARD.encode(&bytes),
-                    }));
+                    let mut result = paid_content_response(&bytes, &mime, 0);
+                    result["already_owned"] = serde_json::json!(true);
+                    result["filename"] = serde_json::json!(o.filename);
+                    return Ok(result);
                 }
                 // Cache record exists but bytes are gone — fall through and
                 // repurchase rather than stranding the user.
@@ -649,63 +726,41 @@ impl RpcHandler {
             tracing::warn!("paid download: failed to cache purchased content (non-fatal): {e:#}");
         }
 
-        // Auto-file the purchase into the user's Files area (2026-07-22):
-        // Photos for images/video, Music for audio, Documents otherwise —
-        // same buckets the Cloud view uses. The in-app viewer still plays
-        // from the purchase cache; this makes the file ALSO show up where
-        // files live, on every device, without relying on a browser
-        // download. Best-effort: never fail a paid download over it.
-        {
-            let folder = if mime_type.starts_with("image/") || mime_type.starts_with("video/") {
-                "Photos"
-            } else if mime_type.starts_with("audio/") {
-                "Music"
-            } else {
-                "Documents"
-            };
-            let base = std::path::Path::new(&filename)
-                .file_name()
-                .and_then(|n| n.to_str())
-                .unwrap_or("download")
-                .to_string();
-            let dir = self.config.data_dir.join("filebrowser").join(folder);
-            if let Err(e) = tokio::fs::create_dir_all(&dir).await {
-                tracing::warn!("paid download: cannot create {}: {e}", dir.display());
-            } else {
-                // Don't clobber an existing file of the same name: "x.jpg"
-                // → "x (2).jpg" etc.
-                let mut target = dir.join(&base);
-                let (stem, ext) = match base.rsplit_once('.') {
-                    Some((s, e)) if !s.is_empty() => (s.to_string(), format!(".{e}")),
-                    _ => (base.clone(), String::new()),
-                };
-                let mut n = 2;
-                while target.exists() {
-                    target = dir.join(format!("{stem} ({n}){ext}"));
-                    n += 1;
-                }
-                match tokio::fs::write(&target, &bytes).await {
-                    Ok(()) => tracing::info!("paid download: filed into {}", target.display()),
-                    Err(e) => tracing::warn!(
-                        "paid download: filing into {} failed (non-fatal): {e}",
-                        target.display()
-                    ),
-                }
-            }
+        // The durable purchased-content cache above is primary. A Files copy
+        // remains optional: a stopped FileBrowser must not undo a paid download.
+        let filed = async {
+            let auth = self.handle_filebrowser_token().await?;
+            let token = auth
+                .get("token")
+                .and_then(|v| v.as_str())
+                .context("FileBrowser omitted its authentication token")?;
+            let client = reqwest::Client::builder()
+                .no_proxy()
+                .redirect(reqwest::redirect::Policy::none())
+                .timeout(std::time::Duration::from_secs(30))
+                .build()?;
+            file_purchase_in_files(
+                &client,
+                "http://127.0.0.1:8083",
+                token,
+                &filename,
+                &mime_type,
+                &bytes,
+            )
+            .await
+        }
+        .await;
+        match filed {
+            Ok(path) => tracing::info!("paid download: filed into Files/{path}"),
+            Err(error) => tracing::warn!(
+                "paid download: optional Files copy failed; purchase cache retained: {error}"
+            ),
         }
 
-        use base64::Engine;
-        let encoded = base64::engine::general_purpose::STANDARD.encode(&bytes);
-
         tracing::info!("paid download: received {} bytes from {onion} (paid {price_sats} sats via {used_backend})", bytes.len());
-        Ok(serde_json::json!({
-            "data": encoded,
-            "size": bytes.len(),
-            "paid_sats": price_sats,
-            "ecash_backend": used_backend,
-            "mime_type": mime_type,
-            "owned": true,
-        }))
+        let mut result = paid_content_response(&bytes, &mime_type, price_sats);
+        result["ecash_backend"] = serde_json::json!(used_backend);
+        Ok(result)
     }
 
     /// Buyer side (#46): ask the selling node to mint a Lightning invoice for a
@@ -1378,3 +1433,7 @@ impl RpcHandler {
         }
     }
 }
+
+#[cfg(test)]
+#[path = "content_tests.rs"]
+mod tests;

core/archipelago/src/api/rpc/content_tests.rs

@@ -0,0 +1,164 @@
+use super::*;
+use hyper::{
+    service::{make_service_fn, service_fn},
+    Body, Response, Server,
+};
+use std::{
+    collections::VecDeque,
+    convert::Infallible,
+    sync::{Arc, Mutex},
+};
+
+struct FilesApi {
+    url: String,
+    seen: Arc<Mutex<Vec<(String, String, Vec<u8>)>>>,
+    task: tokio::task::JoinHandle<()>,
+}
+impl Drop for FilesApi {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+fn files_api(statuses: Vec<u16>) -> FilesApi {
+    let statuses = Arc::new(Mutex::new(VecDeque::from(statuses)));
+    let seen = Arc::new(Mutex::new(Vec::new()));
+    let history = seen.clone();
+    let server = Server::bind(&([127, 0, 0, 1], 0).into());
+    let address = server.local_addr();
+    let service = make_service_fn(move |_| {
+        let statuses = statuses.clone();
+        let seen = history.clone();
+        async move {
+            Ok::<_, Infallible>(service_fn(move |request: hyper::Request<Body>| {
+                let statuses = statuses.clone();
+                let seen = seen.clone();
+                async move {
+                    assert_eq!(request.headers().get("X-Auth").unwrap(), "test-session");
+                    let method = request.method().to_string();
+                    let uri = request.uri().to_string();
+                    let body = hyper::body::to_bytes(request.into_body())
+                        .await
+                        .unwrap()
+                        .to_vec();
+                    seen.lock().unwrap().push((method, uri, body));
+                    let status = statuses
+                        .lock()
+                        .unwrap()
+                        .pop_front()
+                        .expect("unexpected extra Files request");
+                    Ok::<_, Infallible>(
+                        Response::builder()
+                            .status(status)
+                            .body(Body::empty())
+                            .unwrap(),
+                    )
+                }
+            }))
+        }
+    });
+    FilesApi {
+        url: format!("http://{address}"),
+        seen,
+        task: tokio::spawn(async move {
+            server.serve(service).await.unwrap();
+        }),
+    }
+}
+
+#[test]
+fn first_and_cached_paid_downloads_have_the_same_client_payload_contract() {
+    use base64::Engine;
+    for paid in [0, 1] {
+        let response = paid_content_response(&[0, 255, 123], "application/octet-stream", paid);
+        assert_eq!(response["data"], response["data_base64"]);
+        assert_eq!(
+            base64::engine::general_purpose::STANDARD
+                .decode(response["data"].as_str().unwrap())
+                .unwrap(),
+            [0, 255, 123]
+        );
+        assert_eq!(response["size"], 3);
+        assert_eq!(response["size_bytes"], 3);
+        assert_eq!(response["paid_sats"], paid);
+        assert_eq!(response["owned"], true);
+    }
+}
+
+#[tokio::test]
+async fn files_copy_uses_authenticated_api_and_preserves_existing_names() {
+    let api = files_api(vec![200, 409, 200]);
+    let client = reqwest::Client::new();
+    let path = file_purchase_in_files(
+        &client,
+        &api.url,
+        "test-session",
+        "../my #file?.txt",
+        "text/plain",
+        b"paid bytes",
+    )
+    .await
+    .unwrap();
+    assert_eq!(path, "Documents/my #file? (2).txt");
+    let seen = api.seen.lock().unwrap();
+    assert_eq!(seen[0].0, "GET");
+    assert_eq!(seen[0].1, "/api/resources/Documents/");
+    assert_eq!(seen.len(), 3);
+    for (_, uri, body) in &seen[1..] {
+        assert!(uri.contains("override=false"));
+        assert!(uri.contains("%23file%3F"));
+        assert!(!uri.contains("../"));
+        assert_eq!(body, b"paid bytes");
+    }
+}
+
+#[tokio::test]
+async fn files_copy_creates_missing_media_folder() {
+    for (mime, folder) in [
+        ("image/png", "Photos"),
+        ("video/mp4", "Photos"),
+        ("audio/ogg", "Music"),
+    ] {
+        let api = files_api(vec![404, 200, 200]);
+        let path = file_purchase_in_files(
+            &reqwest::Client::new(),
+            &api.url,
+            "test-session",
+            "file",
+            mime,
+            b"bytes",
+        )
+        .await
+        .unwrap();
+        assert_eq!(path, format!("{folder}/file"));
+        let seen = api.seen.lock().unwrap();
+        assert_eq!(seen[1].0, "POST");
+        assert!(seen[1].1.ends_with('/'));
+        assert!(seen[1].2.is_empty());
+        assert_eq!(seen[2].2, b"bytes");
+    }
+}
+
+#[tokio::test]
+async fn files_copy_fails_without_overwriting_or_claiming_success_on_errors() {
+    for statuses in [
+        vec![401],
+        vec![503],
+        vec![404, 500],
+        vec![200, 507],
+        vec![200, 403],
+    ] {
+        let expected = statuses.len();
+        let api = files_api(statuses);
+        assert!(file_purchase_in_files(
+            &reqwest::Client::new(),
+            &api.url,
+            "test-session",
+            "file.txt",
+            "text/plain",
+            b"bytes"
+        )
+        .await
+        .is_err());
+        assert_eq!(api.seen.lock().unwrap().len(), expected);
+    }
+}

core/archipelago/src/container/companion.rs

@@ -313,7 +313,7 @@ async fn image_id(image_ref: &str) -> Option<String> {
 /// should reference (`localhost/<base>:latest` for build, registry
 /// URL for pull).
 async fn ensure_image_present(spec: &CompanionSpec) -> Result<String> {
-    let local_image = format!("localhost/{}:latest", spec.image_base);
+    let mut local_image = format!("localhost/{}:latest", spec.image_base);
     let local_image_compat = format!("localhost/{}:local", spec.image_base);
     let registry_image = format!("{}/{}:latest", COMPANION_REGISTRY, spec.image_base);
 
@@ -322,11 +322,13 @@ async fn ensure_image_present(spec: &CompanionSpec) -> Result<String> {
     for dir in spec.build_dir_candidates {
         let dockerfile = PathBuf::from(dir).join("Dockerfile");
         if fs::try_exists(&dockerfile).await.unwrap_or(false) {
-            // `:local` is a deliberate manual override — never auto-rebuild it.
+            // Older installers and self-update create :local themselves. It
+            // must receive source updates too; treating it as a permanent
+            // manual override silently kept the old LND UI after an OTA.
             if image_exists(&local_image_compat).await {
-                return Ok(local_image_compat);
+                local_image = local_image_compat.clone();
             }
-            // Reuse the auto-built `:latest` only when the build context has NOT
+            // Reuse either local tag only when the build context has NOT
             // changed since it was built. Without this staleness check an
             // already-present image is reused forever, so edits to the baked-in
             // context (Dockerfile, nginx.conf, …) never reach the node — this is
@@ -849,20 +851,43 @@ async fn needs_repair(spec: &CompanionSpec) -> Result<bool> {
     if !matches_known_shape {
         return Ok(true);
     }
-    if on_disk.contains(&local_image) && !on_disk.contains(&local_image_compat) {
+    if let Some(image) = managed_local_image(spec, &on_disk) {
         for dir in spec.build_dir_candidates {
             let dockerfile = PathBuf::from(dir).join("Dockerfile");
             if fs::try_exists(&dockerfile).await.unwrap_or(false) {
                 // Conservative on any timeout/error inside: reuse the cache.
-                return Ok(context_is_newer_than_image(dir, &local_image).await);
+                return Ok(context_is_newer_than_image(dir, &image).await);
             }
         }
     }
     Ok(false)
 }
 
+fn managed_local_image(spec: &CompanionSpec, unit: &str) -> Option<String> {
+    ["latest", "local"]
+        .iter()
+        .map(|tag| format!("localhost/{}:{tag}", spec.image_base))
+        .find(|image| build_unit(spec, image).render() == unit)
+}
+
 #[cfg(test)]
 mod tests {
+    #[test]
+    fn legacy_installer_local_tag_is_checked_for_source_updates_like_latest() {
+        for spec in ALL_COMPANIONS.iter().flat_map(|group| group.iter()) {
+            for tag in ["local", "latest"] {
+                let image = format!("localhost/{}:{tag}", spec.image_base);
+                let unit = build_unit(spec, &image).render();
+                assert_eq!(managed_local_image(spec, &unit), Some(image));
+            }
+            let registry = format!("{}/{}:latest", COMPANION_REGISTRY, spec.image_base);
+            assert_eq!(
+                managed_local_image(spec, &build_unit(spec, &registry).render()),
+                None
+            );
+        }
+    }
+
     use super::*;
 
     fn names(specs: &[&'static CompanionSpec]) -> Vec<&'static str> {