archy/core/archipelago/src/cloud_sources/immich.rs

use super::{
    InstalledTarget, PreviewKind, SourceAlbum, SourceIdentity, SourceItem, SourceKind, SourcePage,
};
use anyhow::{ensure, Context, Result};
use reqwest::{header::HeaderValue, Method};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};

/// The API key stays in a sensitive header and is never part of an item or URL.
pub(crate) struct ImmichAdapter {
    target: InstalledTarget,
    identity: SourceIdentity,
    key: HeaderValue,
    client: reqwest::Client,
}
impl ImmichAdapter {
    /// Setup derives the account from the supplied key, never from user input.
    pub async fn connect(target: InstalledTarget, key: &str) -> Result<Self> {
        tokio::time::timeout(std::time::Duration::from_secs(30), async {
            let identity = SourceIdentity {
                connection_id: uuid::Uuid::new_v4().to_string(),
                kind: SourceKind::Immich,
                installation: target.installation.clone(),
                account_id: uuid::Uuid::nil().to_string(),
            };
            let mut adapter = Self::new(target, identity, key)?;
            let user = adapter
                .json(Method::GET, &["api", "users", "me"], None)
                .await?;
            let account = user
                .get("id")
                .and_then(Value::as_str)
                .context("Immich account identity unavailable")?;
            uuid::Uuid::parse_str(account).context("Invalid Immich account identity")?;
            adapter.identity.account_id = account.into();
            let permissions = adapter.permissions().await?;
            ensure!(
                permissions.iter().any(|v| v == "asset.read" || v == "all"),
                "Immich key needs asset.read permission"
            );
            Ok(adapter)
        })
        .await
        .context("Immich account verification timed out")?
    }
    pub fn identity(&self) -> &SourceIdentity {
        &self.identity
    }
    pub fn new(target: InstalledTarget, identity: SourceIdentity, key: &str) -> Result<Self> {
        ensure!(
            target.kind == SourceKind::Immich && identity.kind == SourceKind::Immich,
            "Cloud source type mismatch"
        );
        ensure!(
            identity.installation == target.installation,
            "Cloud source installation changed; reconnect this account"
        );
        ensure!(
            uuid::Uuid::parse_str(&identity.connection_id).is_ok()
                && uuid::Uuid::parse_str(&identity.account_id).is_ok(),
            "Invalid Cloud source identity"
        );
        ensure!(
            !key.is_empty() && key.len() <= 4096,
            "Invalid Immich API key"
        );
        let mut key = HeaderValue::from_str(key).context("Invalid Immich API key format")?;
        key.set_sensitive(true);
        Ok(Self {
            target,
            identity,
            key,
            client: InstalledTarget::client()?,
        })
    }
    async fn verified_asset(&self, request: &super::handles::StreamRequest) -> Result<Value> {
        uuid::Uuid::parse_str(&request.upstream_id).context("Invalid Immich item identity")?;
        ensure!(
            request.locator.is_none(),
            "Immich does not accept a file path"
        );
        let item = self
            .json(Method::GET, &["api", "assets", &request.upstream_id], None)
            .await?;
        ensure!(
            item.get("id").and_then(Value::as_str) == Some(request.upstream_id.as_str())
                && item.get("isTrashed").and_then(Value::as_bool) != Some(true),
            "Immich item unavailable"
        );
        ensure!(
            asset_revision(&item)? == request.revision,
            "Immich item changed; refresh this source"
        );
        Ok(item)
    }
    pub(super) async fn delete(&self, request: &super::handles::StreamRequest) -> Result<()> {
        let permissions = self.permissions().await?;
        ensure!(
            permissions
                .iter()
                .any(|p| p == "asset.delete" || p == "all"),
            "Immich key cannot delete assets"
        );
        let item = self.verified_asset(request).await?;
        ensure!(
            item.get("ownerId").and_then(Value::as_str) == Some(self.identity.account_id.as_str()),
            "Only your own assets can be removed"
        );
        self.json(
            Method::DELETE,
            &["api", "assets"],
            Some(json!({"ids": [request.upstream_id], "force": false})),
        )
        .await?;
        Ok(())
    }
    pub(super) async fn upload(
        &self,
        folder: &[String],
        name: &str,
        file: tokio::fs::File,
        length: u64,
    ) -> Result<()> {
        ensure!(
            folder.is_empty(),
            "Upload Immich assets into its main folder"
        );
        let permissions = self.permissions().await?;
        ensure!(
            permissions
                .iter()
                .any(|p| p == "asset.upload" || p == "all"),
            "Immich key cannot upload assets"
        );
        let now = chrono::Utc::now().to_rfc3339();
        let part = reqwest::multipart::Part::stream_with_length(
            reqwest::Body::wrap_stream(tokio_util::io::ReaderStream::new(file)),
            length,
        )
        .file_name(name.to_owned());
        let form = reqwest::multipart::Form::new()
            .text("deviceAssetId", uuid::Uuid::new_v4().to_string())
            .text("deviceId", "archipelago-cloud")
            .text("fileCreatedAt", now.clone())
            .text("fileModifiedAt", now)
            .part("assetData", part);
        let response = InstalledTarget::stream_client()?
            .post(self.target.url(&["api", "assets"])?)
            .header("x-api-key", self.key.clone())
            .multipart(form)
            .send()
            .await?;
        ensure!(
            response.status().is_success(),
            "Immich upload refused (HTTP {})",
            response.status().as_u16()
        );
        Ok(())
    }
    pub(super) async fn open_stream(
        &self,
        request: &super::handles::StreamRequest,
        headers: &hyper::HeaderMap,
    ) -> Result<(super::stream::Representation, hyper::Response<hyper::Body>)> {
        tokio::time::timeout(std::time::Duration::from_secs(30), async {
            let permissions = self.permissions().await?;
            let required = if request.download {
                "asset.download"
            } else {
                "asset.view"
            };
            ensure!(
                permissions.iter().any(|v| v == required || v == "all"),
                "Immich key lacks permission for these bytes"
            );
            let item = self.verified_asset(request).await?;
            let filename = item
                .get("originalFileName")
                .and_then(Value::as_str)
                .context("Immich filename unavailable")?;
            let preview_kind = asset_preview_kind(&item);
            let mut path = vec!["api", "assets", request.upstream_id.as_str()];
            if request.download {
                path.push("original");
            } else {
                match preview_kind.context("Immich asset has no supported preview")? {
                    PreviewKind::Video => path.extend(["video", "playback"]),
                    PreviewKind::Image => path.push("thumbnail"),
                }
            }
            let mut url = self.target.url(&path)?;
            if !request.download && preview_kind == Some(PreviewKind::Image) {
                url.query_pairs_mut().append_pair("size", "preview");
            }
            if request.download || preview_kind == Some(PreviewKind::Image) {
                url.query_pairs_mut().append_pair("edited", "false");
            }
            let client = InstalledTarget::stream_client()?;
            let head = client
                .head(url.clone())
                .header("x-api-key", self.key.clone())
                .header("Accept-Encoding", "identity")
                .send()
                .await?;
            let representation = super::stream::Representation::from_head(&head)?;
            let range = super::stream::selected_range(headers, representation.length)?;
            let upstream = representation
                .request(
                    client
                        .get(url)
                        .header("x-api-key", self.key.clone())
                        .header("Accept-Encoding", "identity"),
                    range,
                )?
                .send()
                .await?;
            let mime = upstream
                .headers()
                .get("content-type")
                .and_then(|v| v.to_str().ok())
                .unwrap_or("application/octet-stream")
                .to_owned();
            if !request.download {
                ensure!(
                    match preview_kind {
                        Some(PreviewKind::Video) => matches!(
                            mime.split(';').next().unwrap_or("").trim(),
                            "video/mp4" | "video/webm" | "video/ogg"
                        ),
                        Some(PreviewKind::Image) => matches!(
                            mime.as_str(),
                            "image/jpeg" | "image/png" | "image/webp" | "image/avif"
                        ),
                        None => false,
                    },
                    "Immich preview is not a supported media representation"
                );
            }
            self.verified_asset(request).await?;
            let response = super::stream::response(
                upstream,
                representation.clone(),
                range,
                &mime,
                filename,
                request.download,
            )?;
            Ok((representation, response))
        })
        .await
        .context("Immich stream preparation timed out")?
    }
    async fn json(&self, method: Method, path: &[&str], body: Option<Value>) -> Result<Value> {
        let mut request = self
            .client
            .request(method, self.target.url(path)?)
            .header("x-api-key", self.key.clone());
        if let Some(body) = body {
            request = request.json(&body)
        }
        let mut response = request
            .send()
            .await
            .context("Cloud source request failed")?;
        ensure!(
            response.status().is_success(),
            "Immich refused request (HTTP {})",
            response.status().as_u16()
        );
        let mut bytes = Vec::new();
        while let Some(chunk) = response.chunk().await? {
            ensure!(
                bytes.len() + chunk.len() <= 4 * 1024 * 1024,
                "Immich response exceeds source listing limit"
            );
            bytes.extend_from_slice(&chunk);
        }
        if bytes.is_empty() {
            return Ok(Value::Null);
        }
        serde_json::from_slice(&bytes).context("Invalid Immich response")
    }
    async fn permissions(&self) -> Result<Vec<String>> {
        let version = self
            .json(Method::GET, &["api", "server", "version"], None)
            .await?;
        ensure!(
            version.get("major").and_then(Value::as_u64) == Some(2)
                && version.get("minor").and_then(Value::as_u64) == Some(7)
                && version.get("patch").and_then(Value::as_u64) == Some(4),
            "This Cloud adapter requires the verified Immich 2.7.4 API"
        );
        let user = self
            .json(Method::GET, &["api", "users", "me"], None)
            .await?;
        ensure!(
            user.get("id").and_then(Value::as_str) == Some(self.identity.account_id.as_str()),
            "Cloud source account changed; reconnect this account"
        );
        let info = self
            .json(Method::GET, &["api", "api-keys", "me"], None)
            .await?;
        let permissions = info
            .get("permissions")
            .and_then(Value::as_array)
            .context("Immich key permissions unavailable")?;
        permissions
            .iter()
            .map(|v| {
                v.as_str()
                    .map(str::to_owned)
                    .context("Invalid Immich key permission")
            })
            .collect()
    }
    #[cfg(test)]
    pub async fn list(&self, page: u32) -> Result<SourcePage> {
        self.list_album(page, None).await
    }
    pub async fn list_album(&self, page: u32, album_id: Option<&str>) -> Result<SourcePage> {
        tokio::time::timeout(
            std::time::Duration::from_secs(30),
            self.list_inner(page, album_id),
        )
        .await
        .context("Immich listing timed out")?
    }
    async fn list_inner(&self, page: u32, album_id: Option<&str>) -> Result<SourcePage> {
        ensure!((1..=100_000).contains(&page), "Invalid Cloud source page");
        let album_id = album_id
            .map(uuid::Uuid::parse_str)
            .transpose()
            .context("Invalid Immich album identity")?
            .map(|id| id.to_string());
        let permissions = self.permissions().await?;
        let has = |permission: &str| permissions.iter().any(|v| v == permission || v == "all");
        ensure!(has("asset.read"), "Immich key needs asset.read permission");
        // Do not make album.read a new prerequisite for existing asset-only keys.
        // When available, the upstream account-authorized inventory is the only
        // accepted source of selectable album IDs, refreshed on every page.
        let albums = if has("album.read") {
            Some(parse_albums(
                self.json(Method::GET, &["api", "albums"], None).await?,
            )?)
        } else {
            None
        };
        if let Some(id) = &album_id {
            ensure!(
                albums
                    .as_ref()
                    .is_some_and(|items| items.iter().any(|item| &item.id == id)),
                "Immich album is unavailable or the key lacks album.read permission"
            );
        }
        let mut query = json!({"page":page,"size":100,"withDeleted":false,"withExif":true});
        if let Some(id) = &album_id {
            query["albumIds"] = json!([id]);
        }
        let response = self
            .json(Method::POST, &["api", "search", "metadata"], Some(query))
            .await?;
        let assets = response
            .get("assets")
            .context("Immich listing missing assets")?;
        let items = assets
            .get("items")
            .and_then(Value::as_array)
            .context("Immich listing missing items")?;
        ensure!(items.len() <= 100, "Immich page exceeds requested limit");
        let mut output = Vec::with_capacity(items.len());
        for item in items {
            if item.get("isTrashed").and_then(Value::as_bool) == Some(true) {
                continue;
            }
            let id = item
                .get("id")
                .and_then(Value::as_str)
                .context("Immich asset missing identity")?;
            ensure!(
                uuid::Uuid::parse_str(id).is_ok(),
                "Invalid Immich asset identity"
            );
            let name = item
                .get("originalFileName")
                .and_then(Value::as_str)
                .context("Immich asset missing name")?;
            let revision = asset_revision(item)?;
            output.push(SourceItem {
                source: self.identity.clone(),
                upstream_id: id.into(),
                locator: None,
                name: name.into(),
                preview_kind: asset_preview_kind(item),
                mime_type: item
                    .get("originalMimeType")
                    .and_then(Value::as_str)
                    .map(str::to_owned),
                size: item
                    .pointer("/exifInfo/fileSizeInByte")
                    .and_then(Value::as_u64),
                revision,
                directory: false,
                preview: has("asset.view") && asset_preview_kind(item).is_some(),
                download: has("asset.download"),
                deletable: has("asset.delete")
                    && item.get("ownerId").and_then(Value::as_str)
                        == Some(self.identity.account_id.as_str()),
            });
        }
        let next_page = match assets.get("nextPage") {
            Some(Value::Null) => None,
            Some(Value::String(value)) => Some(
                value
                    .parse::<u32>()
                    .context("Invalid Immich continuation")?,
            ),
            _ => anyhow::bail!("Immich continuation unavailable"),
        };
        ensure!(
            next_page.is_none_or(|next| next == page + 1 && next <= 100_000),
            "Immich continuation did not advance safely"
        );
        Ok(SourcePage {
            upload: has("asset.upload"),
            listing_revision: None,
            albums,
            items: output,
            next_page,
        })
    }
}

fn parse_albums(value: Value) -> Result<Vec<SourceAlbum>> {
    let items = value
        .as_array()
        .context("Immich album inventory unavailable")?;
    ensure!(
        items.len() <= 1000,
        "Immich album inventory exceeds supported limit"
    );
    let mut ids = std::collections::HashSet::new();
    let mut albums = Vec::with_capacity(items.len());
    for item in items {
        let id = uuid::Uuid::parse_str(
            item.get("id")
                .and_then(Value::as_str)
                .context("Immich album identity unavailable")?,
        )?
        .to_string();
        ensure!(ids.insert(id.clone()), "Duplicate Immich album identity");
        let name = item
            .get("albumName")
            .and_then(Value::as_str)
            .context("Immich album name unavailable")?;
        ensure!(
            name.len() <= 4096 && !name.chars().any(char::is_control),
            "Invalid Immich album name"
        );
        albums.push(SourceAlbum {
            id,
            name: name.to_owned(),
        });
    }
    Ok(albums)
}

fn asset_preview_kind(item: &Value) -> Option<PreviewKind> {
    match item.get("type").and_then(Value::as_str) {
        Some("IMAGE") => Some(PreviewKind::Image),
        Some("VIDEO") => Some(PreviewKind::Video),
        _ => None,
    }
}

fn asset_revision(item: &Value) -> Result<String> {
    let checksum = item
        .get("checksum")
        .and_then(Value::as_str)
        .context("Immich asset revision unavailable")?;
    let updated = item
        .get("updatedAt")
        .and_then(Value::as_str)
        .context("Immich asset revision unavailable")?;
    Ok(hex::encode(Sha256::digest(serde_json::to_vec(&(
        checksum,
        updated,
        item.get("type").and_then(Value::as_str),
    ))?)))
}

#[cfg(test)]
mod tests {
    use super::*;
    use hyper::{
        service::{make_service_fn, service_fn},
        Body, Request, Response, Server,
    };
    use std::{
        convert::Infallible,
        sync::{
            atomic::{AtomicUsize, Ordering},
            Arc,
        },
    };
    const ACCOUNT_A: &str = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa";
    const ACCOUNT_B: &str = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb";
    const ASSET: &str = "cccccccc-cccc-4ccc-8ccc-cccccccccccc";
    const ALBUM_A: &str = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee";
    const ALBUM_B: &str = "ffffffff-ffff-4fff-8fff-ffffffffffff";

    async fn fixture() -> (
        InstalledTarget,
        Arc<AtomicUsize>,
        tokio::task::JoinHandle<()>,
    ) {
        fixture_media(false, Arc::new(AtomicUsize::new(0))).await
    }
    // Modes: 0 normal, 1 revoked view permission, 2 changed metadata, 3 wrong media MIME.
    async fn fixture_media(
        video: bool,
        mode: Arc<AtomicUsize>,
    ) -> (
        InstalledTarget,
        Arc<AtomicUsize>,
        tokio::task::JoinHandle<()>,
    ) {
        assert_eq!(std::env::var("ARCHY_TEST_ISOLATED").as_deref(), Ok("1"));
        let searches = Arc::new(AtomicUsize::new(0));
        let observed = searches.clone();
        let socket = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        socket.set_nonblocking(true).unwrap();
        let port = socket.local_addr().unwrap().port();
        let service = make_service_fn(move |_| {
            let searches = observed.clone();
            let mode = mode.clone();
            async move {
                Ok::<_, Infallible>(service_fn(move |request: Request<Body>| {
                    let searches = searches.clone();
                    let mode = mode.clone();
                    async move {
                        let account = match request
                            .headers()
                            .get("x-api-key")
                            .and_then(|h| h.to_str().ok())
                        {
                            Some("synthetic-a") => ACCOUNT_A,
                            Some("synthetic-b") => ACCOUNT_B,
                            _ => {
                                return Ok::<_, Infallible>(
                                    Response::builder().status(401).body(Body::empty()).unwrap(),
                                )
                            }
                        };
                        let media_path = if video {
                            format!("/api/assets/{ASSET}/video/playback")
                        } else {
                            format!("/api/assets/{ASSET}/thumbnail")
                        };
                        if request.uri().path() == media_path {
                            if video {
                                assert!(request.uri().query().is_none());
                            }
                            let bytes = b"abcdefgh";
                            let selected = request.headers().get("range").map(|_| (2usize, 5usize));
                            if request.method() != hyper::Method::HEAD {
                                assert_eq!(
                                    request.headers()["if-unmodified-since"],
                                    "Thu, 08 Oct 2026 00:00:00 GMT"
                                );
                            }
                            let payload = if request.method() == hyper::Method::HEAD {
                                &bytes[..0]
                            } else if let Some((start, end)) = selected {
                                &bytes[start..=end]
                            } else {
                                &bytes[..]
                            };
                            let mut response = Response::builder()
                                .status(if selected.is_some() { 206 } else { 200 })
                                .header("etag", "W/\"synthetic-one\"")
                                .header("last-modified", "Thu, 08 Oct 2026 00:00:00 GMT")
                                .header(
                                    "content-type",
                                    if mode.load(Ordering::SeqCst) == 3 {
                                        "text/html"
                                    } else if video {
                                        "video/mp4"
                                    } else {
                                        "image/jpeg"
                                    },
                                )
                                .header(
                                    "content-length",
                                    if selected.is_some() { 4u64 } else { 8u64 },
                                );
                            if selected.is_some() {
                                response = response.header("content-range", "bytes 2-5/8");
                            }
                            return Ok(response.body(Body::from(payload.to_vec())).unwrap());
                        }
                        let value = match request.uri().path() {
                            "/api/server/version" => json!({"major":2,"minor":7,"patch":4}),
                            "/api/users/me" => json!({"id":account}),
                            path if path == format!("/api/assets/{ASSET}") => {
                                json!({"id":ASSET,"originalFileName":"same-name.jpg","type":if video {"VIDEO"} else {"IMAGE"},"checksum":if mode.load(Ordering::SeqCst) == 2 {"changed"} else {"synthetic-revision"},"updatedAt":"2026-10-08T00:00:00Z","isTrashed":false})
                            }
                            "/api/api-keys/me" => {
                                json!({"permissions":match mode.load(Ordering::SeqCst) {
                                    1 => vec!["user.read","asset.read"],
                                    4 | 5 => vec!["user.read","asset.read","asset.view","album.read"],
                                    _ => vec!["user.read","asset.read","asset.view"],
                                }})
                            }
                            "/api/albums" => {
                                if mode.load(Ordering::SeqCst) == 5 {
                                    json!([])
                                } else {
                                    json!([{"id":if account == ACCOUNT_A { ALBUM_A } else { ALBUM_B },"albumName":"Private album"}])
                                }
                            }
                            "/api/search/metadata" => {
                                searches.fetch_add(1, Ordering::SeqCst);
                                let body: Value = serde_json::from_slice(
                                    &hyper::body::to_bytes(request.into_body()).await.unwrap(),
                                )
                                .unwrap();
                                assert_eq!(body["size"], 100);
                                assert_eq!(body["withDeleted"], false);
                                if let Some(albums) = body.get("albumIds") {
                                    assert_eq!(
                                        *albums,
                                        json!([if account == ACCOUNT_A {
                                            ALBUM_A
                                        } else {
                                            ALBUM_B
                                        }])
                                    );
                                }
                                json!({"assets":{"items":[{"id":ASSET,"ownerId":account,"originalFileName":"same-name.jpg","originalPath":"/private/source/path-never-exposed","originalMimeType":if video {"video/quicktime"} else {"image/jpeg"},"type":if video {"VIDEO"} else {"IMAGE"},"checksum":"synthetic-revision","updatedAt":"2026-10-08T00:00:00Z","exifInfo":{"fileSizeInByte":128},"isTrashed":false}],"nextPage":null}})
                            }
                            _ => panic!("Unexpected synthetic source route"),
                        };
                        Ok(Response::new(Body::from(
                            serde_json::to_vec(&value).unwrap(),
                        )))
                    }
                }))
            }
        });
        let task = tokio::spawn(async move {
            Server::from_tcp(socket)
                .unwrap()
                .serve(service)
                .await
                .unwrap()
        });
        (
            InstalledTarget::fixture(SourceKind::Immich, port),
            searches,
            task,
        )
    }
    fn identity(account: &str) -> SourceIdentity {
        SourceIdentity {
            connection_id: "dddddddd-dddd-4ddd-8ddd-dddddddddddd".into(),
            kind: SourceKind::Immich,
            installation: "synthetic-installation".into(),
            account_id: account.into(),
        }
    }
    #[test]
    fn album_inventory_rejects_invalid_duplicate_or_oversized_metadata() {
        assert!(parse_albums(json!([{"id":ALBUM_A,"albumName":"A"}])).is_ok());
        assert!(parse_albums(json!([{"id":"../private","albumName":"A"}])).is_err());
        assert!(parse_albums(
            json!([{"id":ALBUM_A,"albumName":"A"},{"id":ALBUM_A.to_uppercase(),"albumName":"B"}])
        )
        .is_err());
        assert!(parse_albums(json!([{"id":ALBUM_A,"albumName":"bad\nname"}])).is_err());
        assert!(parse_albums(json!([{"id":ALBUM_A,"albumName":"x".repeat(4097)}])).is_err());
        assert!(parse_albums(json!(vec![json!({"id":ALBUM_A,"albumName":"A"}); 1001])).is_err());
    }

    #[tokio::test]
    async fn album_filter_is_account_permission_bound_and_rechecked_before_search() {
        let mode = Arc::new(AtomicUsize::new(4));
        let (target, count, task) = fixture_media(false, mode.clone()).await;
        let a = ImmichAdapter::new(target.clone(), identity(ACCOUNT_A), "synthetic-a").unwrap();
        let b = ImmichAdapter::new(target, identity(ACCOUNT_B), "synthetic-b").unwrap();
        let page = a.list_album(1, Some(ALBUM_A)).await.unwrap();
        assert_eq!(page.albums.unwrap()[0].id, ALBUM_A);
        assert_eq!(page.items.len(), 1);
        assert_eq!(count.load(Ordering::SeqCst), 1);
        assert!(b.list_album(1, Some(ALBUM_A)).await.is_err());
        assert!(a.list_album(1, Some("../private")).await.is_err());
        mode.store(5, Ordering::SeqCst);
        assert!(a.list_album(2, Some(ALBUM_A)).await.is_err());
        mode.store(0, Ordering::SeqCst);
        assert!(a.list_album(1, Some(ALBUM_A)).await.is_err());
        assert_eq!(count.load(Ordering::SeqCst), 1);
        assert!(
            a.list(1).await.unwrap().albums.is_none(),
            "Existing asset-only key still works"
        );
        task.abort();
    }

    #[tokio::test]
    async fn separate_accounts_keep_distinct_item_scope_and_do_not_invent_download_permission() {
        let (target, count, task) = fixture().await;
        let a = ImmichAdapter::new(target.clone(), identity(ACCOUNT_A), "synthetic-a")
            .unwrap()
            .list(1)
            .await
            .unwrap();
        let b = ImmichAdapter::new(target, identity(ACCOUNT_B), "synthetic-b")
            .unwrap()
            .list(1)
            .await
            .unwrap();
        task.abort();
        assert_eq!(count.load(Ordering::SeqCst), 2);
        assert_eq!(a.items[0].name, b.items[0].name);
        assert_ne!(a.items[0].source, b.items[0].source);
        assert!(a.items[0].preview);
        assert!(!a.items[0].download);
        let rendered = serde_json::to_string(&a).unwrap();
        assert!(!rendered.contains("private/source"));
        assert!(!rendered.contains("synthetic-a"));
    }
    #[tokio::test]
    async fn authorized_preview_stream_rechecks_metadata_and_forwards_exact_range_without_download_scope(
    ) {
        let (target, _, task) = fixture().await;
        let adapter = ImmichAdapter::new(target, identity(ACCOUNT_A), "synthetic-a").unwrap();
        let revision = asset_revision(
            &json!({"type":"IMAGE","checksum":"synthetic-revision","updatedAt":"2026-10-08T00:00:00Z"}),
        )
        .unwrap();
        let mut request = super::super::handles::StreamRequest {
            connection_id: identity(ACCOUNT_A).connection_id,
            upstream_id: ASSET.into(),
            revision,
            locator: None,
            download: false,
        };
        let mut headers = hyper::HeaderMap::new();
        headers.insert("range", "bytes=2-5".parse().unwrap());
        let (_, response) = adapter.open_stream(&request, &headers).await.unwrap();
        assert_eq!(response.status(), 206);
        assert_eq!(response.headers()["content-range"], "bytes 2-5/8");
        assert_eq!(
            &hyper::body::to_bytes(response.into_body()).await.unwrap()[..],
            b"cdef"
        );
        request.download = true;
        assert!(adapter.open_stream(&request, &headers).await.is_err());
        task.abort();
    }
    #[tokio::test]
    async fn read_only_key_cannot_upload_or_delete() {
        let (target, _, task) = fixture().await;
        let adapter = ImmichAdapter::connect(target, "synthetic-a").await.unwrap();
        let page = adapter.list(1).await.unwrap();
        assert!(!page.upload);
        assert!(!page.items[0].deletable);
        let item = &page.items[0];
        let request = super::super::handles::StreamRequest {
            connection_id: item.source.connection_id.clone(),
            upstream_id: item.upstream_id.clone(),
            revision: item.revision.clone(),
            locator: None,
            download: false,
        };
        assert!(adapter
            .delete(&request)
            .await
            .unwrap_err()
            .to_string()
            .contains("cannot delete"));
        let file = tokio::fs::File::from_std(tempfile::tempfile().unwrap());
        assert!(adapter
            .upload(&[], "test.png", file, 0)
            .await
            .unwrap_err()
            .to_string()
            .contains("cannot upload"));
        task.abort();
    }
    #[tokio::test]
    async fn setup_derives_verified_account_without_user_supplied_uuid() {
        let (target, count, task) = fixture().await;
        let adapter = ImmichAdapter::connect(target, "synthetic-b").await.unwrap();
        assert_eq!(adapter.identity().account_id, ACCOUNT_B);
        assert!(uuid::Uuid::parse_str(&adapter.identity().connection_id).is_ok());
        assert_eq!(count.load(Ordering::SeqCst), 0);
        task.abort();
    }
    #[tokio::test]
    async fn wrong_account_and_changed_installation_fail_before_listing() {
        let (target, count, task) = fixture().await;
        let wrong = ImmichAdapter::new(target.clone(), identity(ACCOUNT_A), "synthetic-b").unwrap();
        assert!(wrong.list(1).await.is_err());
        let mut replaced = identity(ACCOUNT_A);
        replaced.installation = "replacement".into();
        assert!(ImmichAdapter::new(target, replaced, "synthetic-a").is_err());
        task.abort();
        assert_eq!(count.load(Ordering::SeqCst), 0);
    }
    #[tokio::test]
    async fn video_preview_uses_playback_ranges_and_rechecks_permission_revision_and_mime() {
        let mode = Arc::new(AtomicUsize::new(0));
        let (target, _, task) = fixture_media(true, mode.clone()).await;
        let adapter =
            ImmichAdapter::new(target.clone(), identity(ACCOUNT_A), "synthetic-a").unwrap();
        let page = adapter.list(1).await.unwrap();
        assert_eq!(page.items[0].preview_kind, Some(PreviewKind::Video));
        assert_eq!(page.items[0].mime_type.as_deref(), Some("video/quicktime"));
        assert!(page.items[0].preview);
        assert!(!page.items[0].download);
        let mut request = super::super::handles::StreamRequest {
            connection_id: identity(ACCOUNT_A).connection_id,
            upstream_id: ASSET.into(),
            revision: page.items[0].revision.clone(),
            locator: None,
            download: false,
        };
        let mut headers = hyper::HeaderMap::new();
        headers.insert("range", "bytes=2-5".parse().unwrap());
        let (_, response) = adapter.open_stream(&request, &headers).await.unwrap();
        assert_eq!(response.status(), 206);
        assert_eq!(response.headers()["content-type"], "video/mp4");
        assert_eq!(response.headers()["content-range"], "bytes 2-5/8");
        assert_eq!(
            &hyper::body::to_bytes(response.into_body()).await.unwrap()[..],
            b"cdef"
        );
        request.download = true;
        assert!(adapter.open_stream(&request, &headers).await.is_err());
        request.download = false;
        for failure in [1, 2, 3] {
            mode.store(failure, Ordering::SeqCst);
            assert!(adapter.open_stream(&request, &headers).await.is_err());
        }
        mode.store(0, Ordering::SeqCst);
        let wrong = ImmichAdapter::new(target, identity(ACCOUNT_B), "synthetic-a").unwrap();
        assert!(wrong.open_stream(&request, &headers).await.is_err());
        task.abort();
    }

    #[test]
    fn preview_kind_uses_verified_asset_type_and_revision_pins_that_type() {
        let image = json!({"type":"IMAGE","checksum":"same","updatedAt":"same"});
        let video = json!({"type":"VIDEO","checksum":"same","updatedAt":"same"});
        assert_ne!(
            asset_revision(&image).unwrap(),
            asset_revision(&video).unwrap()
        );
        assert_eq!(
            asset_preview_kind(&json!({"type":"UNKNOWN","originalMimeType":"video/mp4"})),
            None
        );
    }
}