"""Async Immich API client."""

from __future__ import annotations

import asyncio
import hashlib
import logging
import re
import time
from typing import Any

import aiohttp

from ...notifications.ssrf import UnsafeURLError, validate_outbound_url
from .models import ImmichAlbumData, ImmichAlbumMeta, SharedLinkInfo

_LOGGER = logging.getLogger(__name__)

# Cap user-controlled Immich search parameters so a low-privileged command
# listener (e.g. an Immich ``/search`` command) cannot DoS the upstream.
MAX_SEARCH_QUERY_LEN = 256
MAX_SEARCH_PERSON_IDS = 50

# Module-level TTL caches for album bodies and shared-link listings. The
# Immich ``GET /api/albums/{id}`` response can be tens or hundreds of MB on a
# large album, and bot commands like /random, /latest, /memory all refetch
# the same album in quick succession. A short TTL makes repeat runs nearly
# instant and deduplicates concurrent fetches so a burst of commands issues
# one HTTP call instead of N.
#
# Caches are module-scoped (not instance-scoped) because ``ImmichClient`` is
# constructed fresh per request in several places (api/providers.py,
# services/action_runner.py, command handlers), so an instance cache would
# never survive to serve a second caller. This mirrors ``_users_cache`` in
# ``provider.py``.
_ALBUM_CACHE_TTL_SECONDS = 60
_SHARED_LINKS_CACHE_TTL_SECONDS = 60
# Guard rail against runaway memory — a 200k-asset album response can be
# ~150 MB, so even modest caps bound the worst case.
_ALBUM_CACHE_MAX_ENTRIES = 32
_album_cache_lock = asyncio.Lock()
# key = (server_digest, album_id); value = (monotonic_ts, raw_api_dict)
# Store the raw dict rather than the parsed ``ImmichAlbumData`` so callers
# that pass a ``users_cache`` still get owner-name enrichment on cache hits.
_album_cache: dict[tuple[str, str], tuple[float, dict[str, Any]]] = {}
_shared_links_cache_lock = asyncio.Lock()
# key = server_digest; value = (monotonic_ts, {album_id: [SharedLinkInfo, ...]})
# The underlying ``/api/shared-links`` endpoint has no per-album filter, so
# every call was already paying for the full server-wide list. Caching the
# bucketed result once per server turns N per-album calls into one fetch.
_shared_links_cache: dict[str, tuple[float, dict[str, list[SharedLinkInfo]]]] = {}


def _server_digest(url: str, api_key: str) -> str:
    """Hashed key that avoids putting raw api_key into cache dict keys."""
    return hashlib.sha256(f"{url}|{api_key}".encode("utf-8")).hexdigest()[:32]


def invalidate_album_cache() -> None:
    """Drop every cached album body. Call after mutations that invalidate
    the cached view (e.g. integration tests, manual /refresh commands)."""
    _album_cache.clear()


def invalidate_shared_links_cache() -> None:
    """Drop every cached shared-link listing."""
    _shared_links_cache.clear()

# User-facing error bodies — Immich responses may leak internal paths,
# hostnames, or headers injected by intermediary proxies. These helpers keep
# only a short, scrubbed summary; full bodies are logged server-side only.
_REDACTED_BODY_MAX = 120
_SECRET_PATTERN = re.compile(
    r"(?i)(bearer\s+\S+|x-api-key[:=]\s*\S+|authorization[:=]\s*\S+|cookie[:=]\s*\S+|"
    r"password[:=]?\s*\S+|token[:=]?\s*[A-Za-z0-9._\-]+)"
)


def _redact_body(text: str) -> str:
    """Return a short, credential-scrubbed snippet safe to surface to UI callers.

    Immich error responses are admin-configurable (via reverse proxies, custom
    error pages) and may echo request headers or environment leak. Stripping
    anything that looks like a credential + capping length keeps us from
    persisting secrets into ``ActionExecution.error`` / ``EventLog.details``
    (both of which are returned through the dashboard API).
    """
    if not text:
        return ""
    cleaned = _SECRET_PATTERN.sub("[redacted]", text)
    if len(cleaned) > _REDACTED_BODY_MAX:
        return cleaned[:_REDACTED_BODY_MAX] + "..."
    return cleaned


class ImmichClient:
    """Async client for the Immich API."""

    def __init__(
        self,
        session: aiohttp.ClientSession,
        url: str,
        api_key: str,
    ) -> None:
        self._session = session
        self._url = url.rstrip("/")
        self._api_key = api_key
        self._external_domain: str | None = None
        # SSRF guard — admin-set Immich URLs are loaded from provider config
        # which can be mutated via PATCH /api/providers or imported via
        # prepare-restore, so we revalidate at construction time rather than
        # trusting DB state. Homelab deployments pointing at RFC1918 targets
        # must set ``NOTIFY_BRIDGE_ALLOW_PRIVATE_URLS=1`` in the runtime env.
        if self._url:
            try:
                validate_outbound_url(self._url)
            except UnsafeURLError as err:
                raise UnsafeURLError(
                    f"Refusing to build ImmichClient for unsafe URL {self._url!r}: {err}. "
                    "If this is a LAN/homelab Immich, set NOTIFY_BRIDGE_ALLOW_PRIVATE_URLS=1."
                ) from err

    @property
    def url(self) -> str:
        return self._url

    @property
    def external_domain(self) -> str | None:
        return self._external_domain

    @external_domain.setter
    def external_domain(self, value: str | None) -> None:
        # Mirror the constructor's SSRF guard.  Set
        # ``NOTIFY_BRIDGE_ALLOW_PRIVATE_URLS=1`` for LAN/homelab targets.
        if value:
            try:
                validate_outbound_url(value)
            except UnsafeURLError as err:
                _LOGGER.warning("Ignoring unsafe external_domain %r: %s", value, err)
                return
        self._external_domain = value

    @property
    def external_url(self) -> str:
        if self._external_domain:
            return self._external_domain.rstrip("/")
        return self._url

    @property
    def api_key(self) -> str:
        return self._api_key

    def get_internal_download_url(self, url: str) -> str:
        if self._external_domain:
            external = self._external_domain.rstrip("/")
            if url.startswith(external):
                return url.replace(external, self._url, 1)
        return url

    @property
    def _headers(self) -> dict[str, str]:
        return {"x-api-key": self._api_key}

    @property
    def _json_headers(self) -> dict[str, str]:
        return {
            "x-api-key": self._api_key,
            "Content-Type": "application/json",
        }

    async def ping(self) -> bool:
        try:
            async with self._session.get(
                f"{self._url}/api/server/ping",
                headers=self._headers,
            ) as response:
                return response.status == 200
        except aiohttp.ClientError:
            return False

    async def get_server_config(self) -> str | None:
        try:
            async with self._session.get(
                f"{self._url}/api/server/config",
                headers=self._headers,
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    external_domain = data.get("externalDomain", "") or ""
                    self._external_domain = external_domain
                    return external_domain or None
                _LOGGER.warning(
                    "Failed to fetch server config: HTTP %s", response.status
                )
        except aiohttp.ClientError as err:
            _LOGGER.warning("Failed to fetch server config: %s", err)
        return None

    async def get_users(self) -> dict[str, str]:
        try:
            async with self._session.get(
                f"{self._url}/api/users",
                headers=self._headers,
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    return {
                        u["id"]: u.get("name", u.get("email", "Unknown"))
                        for u in data
                        if u.get("id")
                    }
        except aiohttp.ClientError as err:
            _LOGGER.warning("Failed to fetch users: %s", err)
        return {}

    async def get_people(self) -> dict[str, str]:
        try:
            async with self._session.get(
                f"{self._url}/api/people",
                headers=self._headers,
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    people_list = data.get("people", data) if isinstance(data, dict) else data
                    return {
                        p["id"]: p.get("name", "")
                        for p in people_list
                        if p.get("name")
                    }
        except aiohttp.ClientError as err:
            _LOGGER.warning("Failed to fetch people: %s", err)
        return {}

    async def get_shared_links(self, album_id: str) -> list[SharedLinkInfo]:
        bucketed = await self._get_shared_links_bucketed()
        return list(bucketed.get(album_id, []))

    async def _get_shared_links_bucketed(self) -> dict[str, list[SharedLinkInfo]]:
        """Return ``{album_id: [SharedLinkInfo, ...]}`` for the server, hitting
        the module-level TTL cache first. Underlying Immich endpoint has no
        per-album filter, so one server-wide fetch serves every caller until
        the TTL elapses.
        """
        digest = _server_digest(self._url, self._api_key)
        now = time.monotonic()
        entry = _shared_links_cache.get(digest)
        if entry is not None and (now - entry[0]) < _SHARED_LINKS_CACHE_TTL_SECONDS:
            return entry[1]

        async with _shared_links_cache_lock:
            # Re-check under the lock — another coroutine may have refreshed
            # while we waited.
            entry = _shared_links_cache.get(digest)
            if entry is not None and (time.monotonic() - entry[0]) < _SHARED_LINKS_CACHE_TTL_SECONDS:
                return entry[1]
            fresh = await self.get_all_shared_links_by_album()
            _shared_links_cache[digest] = (time.monotonic(), fresh)
            return fresh

    async def get_all_shared_links_by_album(self) -> dict[str, list[SharedLinkInfo]]:
        """Fetch every shared link on the server, bucketed by album id.

        Immich's ``/api/shared-links`` endpoint is server-wide — there's no
        per-album filter server-side — so every call that wanted the links
        for a single album was already paying the cost of the full listing
        and then discarding most of the response. Callers that need links
        for multiple albums in one tick should use this method and index
        into the returned dict instead of hitting ``get_shared_links`` in
        a loop.

        Returns an empty dict on any error (matches the silent-failure
        contract of ``get_shared_links`` so callers don't need to branch
        on transient outages).
        """
        result: dict[str, list[SharedLinkInfo]] = {}
        try:
            async with self._session.get(
                f"{self._url}/api/shared-links",
                headers=self._headers,
            ) as response:
                if response.status != 200:
                    _LOGGER.warning(
                        "get_all_shared_links non-200: HTTP %s", response.status
                    )
                    return result
                data = await response.json()
                for link in data:
                    album = link.get("album")
                    key = link.get("key")
                    if not (album and key):
                        continue
                    aid = album.get("id")
                    if not aid:
                        continue
                    result.setdefault(aid, []).append(
                        SharedLinkInfo.from_api_response(link)
                    )
        except aiohttp.ClientError as err:
            _LOGGER.warning("Failed to fetch all shared links: %s", err)
        return result

    async def get_album(
        self,
        album_id: str,
        users_cache: dict[str, str] | None = None,
        *,
        use_cache: bool = True,
    ) -> ImmichAlbumData | None:
        """Fetch an album by id, optionally serving from the module-level
        TTL cache. Pass ``use_cache=False`` from paths that must observe the
        current server state (e.g. the notification poll loop's full-fetch
        path, where a stale cached entry would delay asset-removal events).
        Non-cached fetches still populate the cache for subsequent readers.
        """
        cache_key = (_server_digest(self._url, self._api_key), album_id)
        if use_cache:
            entry = _album_cache.get(cache_key)
            if entry is not None and (time.monotonic() - entry[0]) < _ALBUM_CACHE_TTL_SECONDS:
                # Rehydrate per-call so ``users_cache`` enrichment is applied
                # with the caller's dict, not whichever one was live when the
                # cache was populated.
                return ImmichAlbumData.from_api_response(entry[1], users_cache)

        # Deliberately fetch without holding a lock so concurrent calls for
        # *different* album_ids (the common case from asyncio.gather in
        # fetch_albums_with_links) stay parallel. The worst case is a small
        # duplicate-fetch stampede when two requests miss the same album at
        # the same instant — acceptable for our scale.
        try:
            async with self._session.get(
                f"{self._url}/api/albums/{album_id}",
                headers=self._headers,
            ) as response:
                if response.status == 404:
                    return None
                if response.status != 200:
                    raise ImmichApiError(
                        f"Error fetching album {album_id}: HTTP {response.status}"
                    )
                data = await response.json()
        except aiohttp.ClientError as err:
            raise ImmichApiError(f"Error communicating with Immich: {err}") from err

        async with _album_cache_lock:
            # Evict the oldest entry if we're at the cap — simple FIFO is fine
            # for our access pattern (commands touch a small working set).
            if len(_album_cache) >= _ALBUM_CACHE_MAX_ENTRIES and cache_key not in _album_cache:
                oldest = min(_album_cache.items(), key=lambda kv: kv[1][0])[0]
                _album_cache.pop(oldest, None)
            _album_cache[cache_key] = (time.monotonic(), data)
        return ImmichAlbumData.from_api_response(data, users_cache)

    async def get_album_meta(self, album_id: str) -> ImmichAlbumMeta | None:
        """Fetch album metadata without the assets array.

        Uses Immich's ``?withoutAssets=true`` query param, which skips the
        (potentially huge) ``assets`` field. A 200k-asset album response
        drops from ~150 MB to a few hundred bytes, so this is cheap enough
        to run on every poll as a change-detection probe.
        """
        try:
            async with self._session.get(
                f"{self._url}/api/albums/{album_id}",
                params={"withoutAssets": "true"},
                headers=self._headers,
            ) as response:
                if response.status == 404:
                    return None
                if response.status != 200:
                    raise ImmichApiError(
                        f"Error fetching album meta {album_id}: HTTP {response.status}"
                    )
                data = await response.json()
                return ImmichAlbumMeta.from_api_response(data)
        except aiohttp.ClientError as err:
            raise ImmichApiError(f"Error communicating with Immich: {err}") from err

    async def search_album_assets_updated_after(
        self,
        album_id: str,
        updated_after: str,
        *,
        page_size: int = 1000,
        max_pages: int = 50,
    ) -> list[dict[str, Any]]:
        """Fetch assets in ``album_id`` whose ``updatedAt`` is after ``updated_after``.

        Uses ``POST /api/search/metadata`` with ``albumIds=[album_id]`` and
        ``updatedAfter=<iso>``. Paginates through up to ``max_pages`` pages —
        the cap exists so a clock-skew or upstream bug cannot produce an
        infinite loop that exhausts memory on a 200k-asset album. In practice
        an active album sees a few hundred updated assets per tick and
        terminates after one page.

        Returns raw Immich asset dicts (same shape as ``album.assets[*]``
        from ``get_album``), so callers can feed them into
        ``ImmichAssetInfo.from_api_response`` directly.
        """
        if not updated_after:
            return []

        page_size = max(1, min(page_size, 1000))
        results: list[dict[str, Any]] = []
        for page in range(1, max_pages + 1):
            payload: dict[str, Any] = {
                "albumIds": [album_id],
                "updatedAfter": updated_after,
                "page": page,
                "size": page_size,
                # ``withExif`` keeps location/description parity with
                # ``get_album`` so downstream ``ImmichAssetInfo.from_api_response``
                # populates city/country/rating on the delta path too.
                "withExif": True,
                "withPeople": True,
            }
            try:
                async with self._session.post(
                    f"{self._url}/api/search/metadata",
                    headers=self._json_headers,
                    json=payload,
                ) as response:
                    if response.status != 200:
                        body_snip = await response.text()
                        _LOGGER.warning(
                            "Immich delta search non-200: HTTP %s body=%s",
                            response.status, _redact_body(body_snip),
                        )
                        break
                    data = await response.json()
                    assets_block = data.get("assets")
                    if isinstance(assets_block, dict):
                        items = assets_block.get("items", []) or []
                        next_page = assets_block.get("nextPage")
                    elif isinstance(assets_block, list):
                        items = assets_block
                        next_page = None
                    else:
                        _LOGGER.warning(
                            "Immich delta search returned unexpected shape: keys=%s",
                            list(data.keys())[:5],
                        )
                        break

                    results.extend(items)

                    # Stop early on the last page.  Immich returns nextPage as
                    # the next page number (string or int) or None/empty when
                    # exhausted.  Fall back to page-fullness heuristic if the
                    # server omits the pagination hint.
                    if next_page is None or next_page == "" or next_page == 0:
                        break
                    if len(items) < page_size:
                        break
            except aiohttp.ClientError as err:
                _LOGGER.warning("Immich delta search transport error: %s", err)
                break
            except Exception as err:  # noqa: BLE001 — resilience over correctness
                _LOGGER.warning("Immich delta search parse error: %s", err)
                break
        else:
            _LOGGER.warning(
                "Immich delta search for album %s hit max_pages=%d cap",
                album_id, max_pages,
            )
        return results

    async def get_albums(self) -> list[dict[str, Any]]:
        try:
            async with self._session.get(
                f"{self._url}/api/albums",
                headers=self._headers,
            ) as response:
                if response.status == 200:
                    return await response.json()
                _LOGGER.warning("Failed to fetch albums: HTTP %s", response.status)
        except aiohttp.ClientError as err:
            _LOGGER.warning("Failed to fetch albums: %s", err)
        return []

    async def create_shared_link(
        self, album_id: str, password: str | None = None
    ) -> bool:
        payload: dict[str, Any] = {
            "albumId": album_id,
            "type": "ALBUM",
            "allowDownload": True,
            "allowUpload": False,
            "showMetadata": True,
        }
        if password:
            payload["password"] = password

        try:
            async with self._session.post(
                f"{self._url}/api/shared-links",
                headers=self._json_headers,
                json=payload,
            ) as response:
                if response.status == 201:
                    return True
                return False
        except aiohttp.ClientError:
            return False

    async def delete_shared_link(self, link_id: str) -> bool:
        try:
            async with self._session.delete(
                f"{self._url}/api/shared-links/{link_id}",
                headers=self._headers,
            ) as response:
                return response.status == 200
        except aiohttp.ClientError:
            return False

    async def set_shared_link_password(
        self, link_id: str, password: str | None
    ) -> bool:
        payload = {"password": password if password else None}
        try:
            async with self._session.patch(
                f"{self._url}/api/shared-links/{link_id}",
                headers=self._json_headers,
                json=payload,
            ) as response:
                return response.status == 200
        except aiohttp.ClientError:
            return False

    async def search_smart(
        self,
        query: str,
        album_ids: list[str] | None = None,
        limit: int = 10,
        page: int = 1,
    ) -> list[dict[str, Any]]:
        # Cap user-controlled inputs — a low-privileged Telegram listener can
        # craft arbitrarily long queries to DoS the upstream Immich.
        query = (query or "")[:MAX_SEARCH_QUERY_LEN]
        payload: dict[str, Any] = {"query": query, "page": max(1, page), "size": min(max(1, limit), 100)}
        if album_ids:
            payload["albumIds"] = album_ids[:MAX_SEARCH_PERSON_IDS]
        return await self._search_items(
            f"{self._url}/api/search/smart", payload, limit, "smart",
        )

    async def search_metadata(
        self,
        query: str,
        album_ids: list[str] | None = None,
        limit: int = 10,
        page: int = 1,
    ) -> list[dict[str, Any]]:
        query = (query or "")[:MAX_SEARCH_QUERY_LEN]
        payload: dict[str, Any] = {"originalFileName": query, "page": max(1, page), "size": min(max(1, limit), 100)}
        if album_ids:
            payload["albumIds"] = album_ids[:MAX_SEARCH_PERSON_IDS]
        return await self._search_items(
            f"{self._url}/api/search/metadata", payload, limit, "metadata",
        )

    async def _search_items(
        self,
        url: str,
        payload: dict[str, Any],
        limit: int,
        kind: str,
    ) -> list[dict[str, Any]]:
        """Shared POST-and-extract-items helper with error logging.

        Returns an empty list on any error; previously these paths swallowed
        non-200s silently, making "/search always returns no results" on
        misbehaving Immich deployments impossible to diagnose without a
        network trace. Logging keeps the empty-list contract but tells the
        operator *why* it's empty.
        """
        try:
            async with self._session.post(
                url,
                headers=self._json_headers,
                json=payload,
            ) as response:
                if response.status != 200:
                    body_snip = await response.text()
                    _LOGGER.warning(
                        "Immich %s search non-200: HTTP %s body=%s",
                        kind, response.status, _redact_body(body_snip),
                    )
                    return []
                data = await response.json()
                # Modern Immich: {"assets": {"items": [...], ...}}
                assets_block = data.get("assets")
                if isinstance(assets_block, dict):
                    items = assets_block.get("items", []) or []
                elif isinstance(assets_block, list):
                    # Older/alternate shape — flat list of assets.
                    items = assets_block
                else:
                    _LOGGER.warning(
                        "Immich %s search returned unexpected shape: keys=%s",
                        kind, list(data.keys())[:5],
                    )
                    items = []
                return items[:limit]
        except aiohttp.ClientError as err:
            _LOGGER.warning("Immich %s search transport error: %s", kind, err)
        except Exception as err:  # noqa: BLE001 — don't crash caller on unexpected JSON
            _LOGGER.warning("Immich %s search parse error: %s", kind, err)
        return []

    async def search_by_person(
        self, person_id: str, limit: int = 10
    ) -> list[dict[str, Any]]:
        """Fetch up to ``limit`` assets tagged with ``person_id``.

        Uses ``POST /api/search/metadata`` with ``personIds`` — the public
        ``GET /api/people/{id}/assets`` endpoint was removed from Immich
        around v1.106 and now silently 404s, which is why this method used
        to return an empty list on current servers.
        """
        payload: dict[str, Any] = {
            "personIds": [person_id][:MAX_SEARCH_PERSON_IDS],
            "page": 1,
            "size": max(1, min(limit, 100)),
        }
        try:
            async with self._session.post(
                f"{self._url}/api/search/metadata",
                headers=self._json_headers,
                json=payload,
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    items = data.get("assets", {}).get("items", [])
                    return items[:limit]
        except aiohttp.ClientError:
            pass
        return []

    async def get_memories(
        self,
        date: str | None = None,
    ) -> list[dict[str, Any]]:
        """Fetch native Immich memories (On This Day).

        Args:
            date: ISO date string (e.g. "2026-03-20") to fetch memories for.
                  If None, Immich returns memories for today.

        Returns a list of memory objects, each containing an ``assets`` list
        with full ``AssetResponseDto`` items.
        """
        params: dict[str, str] = {}
        if date:
            params["for"] = date
        try:
            async with self._session.get(
                f"{self._url}/api/memories",
                headers=self._headers,
                params=params,
            ) as response:
                if response.status == 200:
                    return await response.json()
                _LOGGER.warning(
                    "Failed to fetch memories: HTTP %s", response.status
                )
        except aiohttp.ClientError as err:
            _LOGGER.warning("Failed to fetch memories: %s", err)
        return []

    # ------------------------------------------------------------------
    # Write methods (used by action executors)
    # ------------------------------------------------------------------

    async def add_assets_to_album(
        self, album_id: str, asset_ids: list[str]
    ) -> dict[str, Any]:
        """Add assets to an album. Returns API response with success/error arrays.

        Immich returns 200 with a per-asset array even when some IDs fail
        individually (already in album, not found, etc). Partial failures
        are data, not errors — surface them as the normal return value.
        Non-2xx responses include Immich's error body in the raised message
        so callers and logs see the real reason (bad UUIDs, stale album,
        permission, etc.) instead of just the HTTP status code.
        """
        payload = {"ids": asset_ids}
        try:
            async with self._session.put(
                f"{self._url}/api/albums/{album_id}/assets",
                headers=self._json_headers,
                json=payload,
            ) as response:
                body_text = await response.text()
                if response.status in (200, 201):
                    try:
                        parsed = await response.json(content_type=None)
                    except Exception:  # noqa: BLE001 — malformed body, still 200
                        return {"raw": body_text}
                    # Per-asset array is the typical shape; wrap for consistency.
                    if isinstance(parsed, list):
                        return {"results": parsed}
                    if isinstance(parsed, dict):
                        return parsed
                    return {"raw": body_text}
                # Log full body server-side (for operators), surface only a
                # redacted snippet to the caller — this string ends up in
                # ActionExecution.error / EventLog.details which are returned
                # through the dashboard API.
                _LOGGER.warning(
                    "add_assets_to_album failed: HTTP %s body=%s",
                    response.status, body_text[:512],
                )
                raise ImmichApiError(
                    f"Failed to add assets to album {album_id}: "
                    f"HTTP {response.status} {_redact_body(body_text)}"
                )
        except aiohttp.ClientError as err:
            raise ImmichApiError(f"Error adding assets to album: {err}") from err

    async def set_album_thumbnail(
        self, album_id: str, asset_id: str
    ) -> None:
        """Set an album's cover/thumbnail to the given asset.

        Uses ``PATCH /api/albums/{id}`` with ``albumThumbnailAssetId``.
        Raises ``ImmichApiError`` on non-2xx so callers can treat it as
        best-effort and log.
        """
        payload = {"albumThumbnailAssetId": asset_id}
        try:
            async with self._session.patch(
                f"{self._url}/api/albums/{album_id}",
                headers=self._json_headers,
                json=payload,
            ) as response:
                if response.status in (200, 201, 204):
                    return
                body_text = await response.text()
                _LOGGER.warning(
                    "set_album_thumbnail failed: HTTP %s body=%s",
                    response.status, body_text[:512],
                )
                raise ImmichApiError(
                    f"Failed to set album thumbnail for {album_id}: "
                    f"HTTP {response.status} {_redact_body(body_text)}"
                )
        except aiohttp.ClientError as err:
            raise ImmichApiError(f"Error setting album thumbnail: {err}") from err

    async def remove_assets_from_album(
        self, album_id: str, asset_ids: list[str]
    ) -> dict[str, Any]:
        """Remove assets from an album."""
        payload = {"ids": asset_ids}
        try:
            async with self._session.delete(
                f"{self._url}/api/albums/{album_id}/assets",
                headers=self._json_headers,
                json=payload,
            ) as response:
                if response.status == 200:
                    return await response.json()
                raise ImmichApiError(
                    f"Failed to remove assets from album {album_id}: HTTP {response.status}"
                )
        except aiohttp.ClientError as err:
            raise ImmichApiError(f"Error removing assets from album: {err}") from err

    async def create_album(
        self, name: str, asset_ids: list[str] | None = None
    ) -> dict[str, Any]:
        """Create a new album, optionally with initial assets."""
        payload: dict[str, Any] = {"albumName": name}
        if asset_ids:
            payload["assetIds"] = asset_ids
        try:
            async with self._session.post(
                f"{self._url}/api/albums",
                headers=self._json_headers,
                json=payload,
            ) as response:
                if response.status == 201:
                    return await response.json()
                raise ImmichApiError(
                    f"Failed to create album '{name}': HTTP {response.status}"
                )
        except aiohttp.ClientError as err:
            raise ImmichApiError(f"Error creating album: {err}") from err

    async def get_person_assets_all(self, person_id: str) -> list[dict[str, Any]]:
        """Fetch ALL assets tagged with a person (paginated, no soft cap).

        Uses ``POST /api/search/metadata`` with ``personIds``. The legacy
        ``GET /api/people/{id}/assets`` endpoint was removed from Immich
        around v1.106 and returns 404 on current servers — switching to
        the search endpoint is the only way to get person-filtered assets
        from modern Immich.
        """
        all_items: list[dict[str, Any]] = []
        page = 1
        page_size = 100
        max_pages = 1000  # hard cap to avoid runaway loops if server misbehaves
        while page <= max_pages:
            payload: dict[str, Any] = {
                "personIds": [person_id],
                "page": page,
                "size": page_size,
            }
            try:
                async with self._session.post(
                    f"{self._url}/api/search/metadata",
                    headers=self._json_headers,
                    json=payload,
                ) as response:
                    if response.status == 200:
                        data = await response.json()
                        items = data.get("assets", {}).get("items", [])
                        if not items:
                            break
                        all_items.extend(items)
                        if len(items) < page_size:
                            break
                        page += 1
                        continue
                    if response.status == 404:
                        # Person doesn't exist — return empty rather than raising
                        return all_items
                    raise ImmichApiError(
                        f"Failed to fetch person {person_id} assets: HTTP {response.status}"
                    )
            except aiohttp.ClientError as err:
                raise ImmichApiError(f"Error fetching person assets: {err}") from err
        return all_items

    async def search_smart_all(
        self, query: str, limit: int = 1000
    ) -> list[dict[str, Any]]:
        """Smart search with pagination up to limit."""
        all_items: list[dict[str, Any]] = []
        page = 1
        page_size = min(limit, 100)
        while len(all_items) < limit:
            payload = {"query": query, "page": page, "size": page_size}
            try:
                async with self._session.post(
                    f"{self._url}/api/search/smart",
                    headers=self._json_headers,
                    json=payload,
                ) as response:
                    if response.status != 200:
                        break
                    data = await response.json()
                    items = data.get("assets", {}).get("items", [])
                    if not items:
                        break
                    all_items.extend(items)
                    if len(items) < page_size:
                        break
                    page += 1
            except aiohttp.ClientError:
                break
        return all_items[:limit]

    # ------------------------------------------------------------------
    # Read methods (continued)
    # ------------------------------------------------------------------

    async def get_asset_thumbnail(self, asset_id: str, size: str = "preview") -> bytes | None:
        try:
            async with self._session.get(
                f"{self._url}/api/assets/{asset_id}/thumbnail",
                headers=self._headers,
                params={"size": size},
            ) as response:
                if response.status == 200:
                    return await response.read()
        except aiohttp.ClientError:
            pass
        return None


class ImmichApiError(Exception):
    """Raised when an Immich API call fails."""
