mule-image/backend/app/routers/photos.py

"""
Photos API router
"""
from typing import List, Optional, Dict, Any
from datetime import datetime, timezone
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel
from sqlalchemy import select, and_, or_, func, tuple_
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
import asyncio
import base64
import json
import os
import logging

logger = logging.getLogger(__name__)

from app.database import get_db
from app.models import Photo, Folder, Tag
from app.models.folders import SourceRoot
from app.models.user import User
from app.models.heaps import heap_photos
from app.models.tags import photo_tags
from app.schemas.photos import PhotoResponse, PhotoUpdate, PhotoListResponse, BulkAction
from app.services.exif_writer import ExifWriteError, write_taken_at
from app.services.date_guess import has_date_warning as compute_date_warning
from app.services import video as video_service
from app.dependencies import (
    get_current_user, get_current_user_media, get_user_photo,
    get_user_or_shared_heap, get_user_or_shared_folder,
    can_access_photo_via_share,
)
from app.services.nextcloud_dav import (
    NextcloudCredentialsMissing,
    get_preview_async,
    is_nextcloud_path,
    move_for_user,
)
from app.config import settings


def _nc_unsupported(detail: str) -> HTTPException:
    """Returns a 501 with a UI-friendly hint to use Nextcloud's web UI
    for an op we haven't routed through WebDAV yet. Centralised so the
    message stays consistent."""
    return HTTPException(
        status_code=501,
        detail=(
            f"{detail} For Nextcloud-managed libraries, do this from "
            "Nextcloud's web UI; mule-image will pick up the change."
        ),
    )

router = APIRouter()

@router.get("")
async def list_photos(
    q: Optional[str] = None,
    # Accept either a bare date ("2026-04-10") or a full ISO datetime
    # ("2026-04-10T23:59:59"). pydantic v2's datetime parser rejects
    # the bare form with 422; we coerce manually below so older
    # clients that send a date-only string keep working.
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
    folder_id: Optional[str] = None,
    tag_ids: Optional[str] = None,
    media_type: Optional[str] = None,
    rating_min: Optional[int] = Query(None, ge=0, le=5),
    rating_max: Optional[int] = Query(None, ge=0, le=5),
    color_label: Optional[str] = None,
    is_discarded: Optional[bool] = False,
    is_duplicate: Optional[bool] = None,
    needs_review: Optional[bool] = None,
    has_date_warning: Optional[bool] = None,
    heap_id: Optional[str] = None,
    sort: str = "taken_at",
    order: str = "desc",
    page: int = Query(1, ge=1),
    per_page: int = Query(100, ge=1, le=500),
    cursor: Optional[str] = Query(None, description="Opaque cursor for keyset pagination"),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """List photos with filters and pagination"""

    # Determine if this request is for shared content. When viewing a
    # shared heap or folder the user_id filter is replaced by the
    # heap/folder join that already encodes authorization.
    is_shared_context = False

    if heap_id:
        try:
            _heap, _perm = await get_user_or_shared_heap(heap_id, current_user, db)
            if _perm != "owner":
                is_shared_context = True
        except Exception:
            raise HTTPException(status_code=404, detail="Heap not found")

    if folder_id and not is_shared_context:
        try:
            _folder, _perm = await get_user_or_shared_folder(folder_id, current_user, db)
            if _perm != "owner":
                is_shared_context = True
        except Exception:
            raise HTTPException(status_code=404, detail="Folder not found")

    # Build query — eager-load tags so the response can include them
    # without an N+1 round-trip per photo. Scoped to the current user
    # unless we're in a shared context (scoped by heap/folder instead).
    query = select(Photo).options(selectinload(Photo.tags))
    if not is_shared_context:
        query = query.where(Photo.user_id == current_user.id)

    # Apply filters
    filters = []
    
    # Text search. Walks every metadata field a user might reasonably
    # remember a photo by: basename, title, notes, raw EXIF blob, and
    # tag names (via a subquery so photos with any matching tag come
    # back even when the tag filter isn't set). Case-insensitive ILIKE
    # across all fields — the frontend re-runs the same match logic to
    # render a "matched on …" chip on each thumbnail.
    if q:
        search_pattern = f"%{q}%"
        tag_subq = (
            select(photo_tags.c.photo_id)
            .select_from(photo_tags.join(Tag, photo_tags.c.tag_id == Tag.id))
            .where(Tag.name.ilike(search_pattern))
        )
        filters.append(
            or_(
                Photo.filename.ilike(search_pattern),
                Photo.user_title.ilike(search_pattern),
                Photo.user_notes.ilike(search_pattern),
                Photo.exif_json.ilike(search_pattern),
                Photo.id.in_(tag_subq),
            )
        )
    
    # Date range. Both inputs are strings to keep pydantic from rejecting
    # bare-date forms ("2026-04-10") with 422; coerce here. fromisoformat
    # accepts both bare dates and full ISO datetimes — when given a date,
    # it returns midnight, which is what we want for the lower bound.
    def _parse_bound(s: Optional[str]) -> Optional[datetime]:
        if not s:
            return None
        try:
            dt = datetime.fromisoformat(s)
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail=f"Invalid date: {s!r} (expected YYYY-MM-DD or ISO 8601)",
            )
        # If the caller sent a tz-aware string, normalize to naive UTC —
        # the photos.taken_at column is `timestamp without time zone`.
        if dt.tzinfo is not None:
            from datetime import timezone
            dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
        return dt

    parsed_date_from = _parse_bound(date_from)
    parsed_date_to = _parse_bound(date_to)
    if parsed_date_from:
        filters.append(Photo.taken_at >= parsed_date_from)
    if parsed_date_to:
        filters.append(Photo.taken_at <= parsed_date_to)
    
    # Folder filter. The sidebar can pass either a SourceRoot id or a
    # Folder id; both should include descendants so clicking a parent
    # folder shows everything under it (Lightroom semantics).
    if folder_id:
        sr_check = await db.execute(
            select(SourceRoot).where(SourceRoot.id == folder_id)
        )
        sr_row = sr_check.scalar_one_or_none()

        if sr_row is not None:
            # Source root → all folders under it (any depth).
            child_folders = await db.execute(
                select(Folder.id).where(Folder.source_root_id == folder_id)
            )
            child_ids = [row[0] for row in child_folders.all()]
            if child_ids:
                filters.append(Photo.folder_id.in_(child_ids))
            else:
                filters.append(Photo.id == '__no_match__')
        else:
            # Folder id → that folder + every descendant by path prefix.
            target_check = await db.execute(
                select(Folder).where(Folder.id == folder_id)
            )
            target = target_check.scalar_one_or_none()
            if target is None:
                filters.append(Photo.id == '__no_match__')
            else:
                target_path = os.path.normpath(target.path).rstrip(os.sep)
                desc_result = await db.execute(
                    select(Folder.id).where(
                        (Folder.path == target_path)
                        | (Folder.path.like(target_path + os.sep + '%'))
                    )
                )
                desc_ids = [row[0] for row in desc_result.all()]
                filters.append(Photo.folder_id.in_(desc_ids))
    
    # Media type filter
    if media_type:
        types = media_type.split(',')
        filters.append(Photo.media_type.in_(types))
    
    # Rating filter
    if rating_min is not None:
        filters.append(Photo.rating >= rating_min)
    if rating_max is not None:
        filters.append(Photo.rating <= rating_max)
    
    # Color label filter
    if color_label:
        if color_label == 'none':
            filters.append(Photo.color_label.is_(None))
        else:
            filters.append(Photo.color_label == color_label)
    
    # Discard filter — defaults to hiding discarded photos
    filters.append(Photo.is_discarded == is_discarded)

    # Hidden-folder filter. Photos in folders the user has marked
    # "hidden from views" (or any descendant of one) are excluded from
    # every cross-cutting listing — All Photos, Rated, Colors, Tags,
    # People, Map, search, etc. We only apply the filter when the
    # request isn't already scoped to a user-intentional collection:
    #   - folder_id set: the user is explicitly browsing that folder,
    #     which is precisely how hidden folders are "opened" again.
    #   - heap_id set: heaps are hand-curated. If the user added a
    #     photo to a heap and later hid its folder, the heap still
    #     reflects their explicit pick.
    if not folder_id and not heap_id:
        filters.append(Photo.is_hidden.is_(False))

    # Duplicate filter — only applied when explicitly set, so the default
    # view shows everything regardless of duplicate status.
    if is_duplicate is not None:
        filters.append(Photo.is_duplicate == is_duplicate)
    if needs_review is not None:
        filters.append(Photo.needs_review == needs_review)
    if has_date_warning is not None:
        filters.append(Photo.has_date_warning == has_date_warning)

    # Heap membership filter — restrict to photos that belong to the heap.
    if heap_id:
        filters.append(
            Photo.id.in_(
                select(heap_photos.c.photo_id).where(heap_photos.c.heap_id == heap_id)
            )
        )

    # Tag filter — comma-separated tag ids, AND semantics. A photo must
    # have a row in photo_tags for EVERY listed tag. Implemented as a
    # single GROUP BY ... HAVING COUNT(DISTINCT) = N subquery so the cost
    # is independent of the number of tags being filtered.
    if tag_ids:
        tag_id_list = [t.strip() for t in tag_ids.split(',') if t.strip()]
        if tag_id_list:
            matching_photos = (
                select(photo_tags.c.photo_id)
                .where(photo_tags.c.tag_id.in_(tag_id_list))
                .group_by(photo_tags.c.photo_id)
                .having(
                    func.count(func.distinct(photo_tags.c.tag_id)) == len(tag_id_list)
                )
            )
            filters.append(Photo.id.in_(matching_photos))

    # Apply all filters
    if filters:
        query = query.where(and_(*filters))
    
    # Apply sorting. The sort field is whitelisted so a malicious client
    # can't pass an arbitrary column name (e.g. "filepath" leaks paths or
    # any other Photo attribute the model exposes).
    SORT_WHITELIST = {
        "taken_at": Photo.taken_at,
        "added_at": Photo.added_at,
        "filename": Photo.filename,
        "file_size": Photo.file_size,
        "rating": Photo.rating,
    }
    sort_column = SORT_WHITELIST.get(sort, Photo.taken_at)
    desc = order == "desc"

    # Keyset / cursor pagination — O(1) regardless of page depth.
    # The cursor encodes the last-seen (sort_value, id) pair so the DB
    # can seek directly to the next slice via an indexed range scan
    # instead of skipping N rows with OFFSET.
    if cursor:
        try:
            decoded = json.loads(base64.urlsafe_b64decode(cursor))
            cursor_val = decoded["v"]
            cursor_id = decoded["id"]
            # For datetime columns, parse the ISO string back.
            if sort in ("taken_at", "added_at") and cursor_val is not None:
                cursor_val = datetime.fromisoformat(cursor_val)
        except Exception:
            raise HTTPException(status_code=400, detail="Invalid cursor")

        # Keyset condition: for DESC we want rows "less than" the cursor,
        # for ASC rows "greater than". We use (sort_col, id) tuple
        # comparison which handles NULLs and ties correctly.
        if desc:
            if cursor_val is None:
                # NULL sorts last in DESC with NULLS LAST — seek past it by id
                query = query.where(
                    or_(
                        sort_column.is_(None) & (Photo.id < cursor_id),
                    )
                )
            else:
                query = query.where(
                    or_(
                        sort_column < cursor_val,
                        and_(sort_column == cursor_val, Photo.id < cursor_id),
                        sort_column.is_(None),
                    )
                )
        else:
            if cursor_val is None:
                query = query.where(
                    or_(
                        sort_column.is_(None) & (Photo.id > cursor_id),
                    )
                )
            else:
                query = query.where(
                    or_(
                        sort_column > cursor_val,
                        and_(sort_column == cursor_val, Photo.id > cursor_id),
                    )
                )

    if desc:
        query = query.order_by(sort_column.desc().nulls_last(), Photo.id.desc())
    else:
        query = query.order_by(sort_column.asc().nulls_last(), Photo.id.asc())

    # Count total results (only when no cursor — first page needs it;
    # subsequent pages reuse the total from the first response).
    total = None
    if not cursor:
        count_query = select(func.count()).select_from(query.subquery())
        total_result = await db.execute(count_query)
        total = total_result.scalar()

    # Fallback to offset pagination when no cursor is provided and page > 1
    # (backward compat for any callers not yet using cursors).
    if not cursor and page > 1:
        offset = (page - 1) * per_page
        query = query.offset(offset)

    query = query.limit(per_page)

    # Execute query
    result = await db.execute(query)
    photos = result.scalars().all()

    # Build next_cursor from the last row in this batch.
    next_cursor = None
    if photos and len(photos) == per_page:
        last = photos[-1]
        sort_val = getattr(last, sort if sort in SORT_WHITELIST else "taken_at")
        if isinstance(sort_val, datetime):
            sort_val = sort_val.isoformat()
        cursor_payload = json.dumps({"v": sort_val, "id": last.id})
        next_cursor = base64.urlsafe_b64encode(cursor_payload.encode()).decode()

    # Convert to response, attaching tags inline so the frontend can group
    # client-side without a second round-trip.
    # In shared context, resolve owner usernames for photos from other users.
    owner_cache: dict[str, str] = {}  # user_id → username
    if is_shared_context:
        other_user_ids = {p.user_id for p in photos if p.user_id != current_user.id}
        if other_user_ids:
            from app.models.user import User as UserModel
            user_result = await db.execute(
                select(UserModel.id, UserModel.username).where(UserModel.id.in_(other_user_ids))
            )
            owner_cache = {uid: uname for uid, uname in user_result.all()}

    photo_dicts = []
    for photo in photos:
        d = PhotoResponse.from_orm(photo).dict()
        d["tags"] = [
            {"id": t.id, "name": t.name, "color": t.color}
            for t in (photo.tags or [])
        ]
        if is_shared_context and photo.user_id != current_user.id:
            d["owner_username"] = owner_cache.get(photo.user_id)
        photo_dicts.append(d)

    response = {
        "photos": photo_dicts,
        "per_page": per_page,
        "next_cursor": next_cursor,
    }
    # Include total + legacy page fields on first page / non-cursor requests
    if total is not None:
        response["total"] = total
        response["page"] = page
        response["pages"] = (total + per_page - 1) // per_page if total else 0
    return response

@router.get("/map")
async def list_photos_with_gps(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Lightweight listing of every non-discarded photo that has GPS
    coordinates, used by the Map view."""
    result = await db.execute(
        select(
            Photo.id,
            Photo.latitude,
            Photo.longitude,
            Photo.taken_at,
        ).where(
            Photo.user_id == current_user.id,
            Photo.is_discarded.is_(False),
            Photo.is_hidden.is_(False),
            Photo.latitude.is_not(None),
            Photo.longitude.is_not(None),
        )
    )
    return [
        {
            "id": row.id,
            "latitude": row.latitude,
            "longitude": row.longitude,
            "taken_at": row.taken_at.isoformat() if row.taken_at else None,
        }
        for row in result.all()
    ]


@router.get("/memories")
async def get_memories(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """'On this day' — photos taken on this date in previous years.

    Returns groups keyed by year, each with up to 12 photos. Only
    considers non-discarded, non-hidden photos with an EXIF-sourced
    taken_at (no filesystem-guessed dates to avoid false matches).
    """
    from sqlalchemy import extract

    today = datetime.now().date()

    result = await db.execute(
        select(
            Photo.id,
            Photo.filename,
            Photo.taken_at,
            Photo.thumb_small,
            Photo.thumb_medium,
            Photo.media_type,
            Photo.width,
            Photo.height,
            Photo.rating,
            Photo.color_label,
            Photo.is_discarded,
        )
        .where(
            Photo.user_id == current_user.id,
            Photo.is_discarded.is_(False),
            Photo.is_hidden.is_(False),
            Photo.taken_at.is_not(None),
            Photo.taken_at_source == "exif",
            extract("month", Photo.taken_at) == today.month,
            extract("day", Photo.taken_at) == today.day,
            extract("year", Photo.taken_at) < today.year,
        )
        .order_by(Photo.taken_at.desc())
    )
    rows = result.all()

    # Group by year
    years: dict[int, list] = {}
    for row in rows:
        year = row.taken_at.year
        group = years.setdefault(year, [])
        if len(group) >= 12:
            continue
        group.append({
            "id": row.id,
            "filename": row.filename,
            "taken_at": row.taken_at.isoformat(),
            "thumb_small": row.thumb_small,
            "thumb_medium": row.thumb_medium,
            "media_type": row.media_type,
            "width": row.width,
            "height": row.height,
            "rating": row.rating,
            "color_label": row.color_label,
            "is_discarded": row.is_discarded,
        })

    memories = [
        {"year": year, "years_ago": today.year - year, "photos": photos}
        for year, photos in sorted(years.items())
    ]

    return {"date": today.isoformat(), "memories": memories}


@router.get("/{photo_id}")
async def get_photo(
    photo_id: str,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Get single photo with full EXIF and its tags."""
    photo = await get_user_photo(photo_id, current_user, db)

    # Fetch tags via the join table so we don't need to declare a
    # relationship on the Photo model side.
    tag_result = await db.execute(
        select(Tag)
        .join(photo_tags, Tag.id == photo_tags.c.tag_id)
        .where(photo_tags.c.photo_id == photo_id)
        .order_by(Tag.name.asc())
    )
    tags = tag_result.scalars().all()

    base = PhotoResponse.from_orm(photo).dict()
    base["tags"] = [
        {"id": t.id, "name": t.name, "color": t.color} for t in tags
    ]
    return base


@router.post("/{photo_id}/tags", status_code=201)
async def add_photo_tags(
    photo_id: str,
    body: dict,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Add one or more tags to a photo. Body: { tag_ids: [str, ...] }.
    Idempotent: re-adding existing members is a no-op."""
    await get_user_photo(photo_id, current_user, db)

    tag_ids = body.get("tag_ids") or []
    if not isinstance(tag_ids, list) or not tag_ids:
        return {"status": "success", "added": 0}

    existing = await db.execute(
        select(photo_tags.c.tag_id).where(
            photo_tags.c.photo_id == photo_id,
            photo_tags.c.tag_id.in_(tag_ids),
        )
    )
    existing_ids = {row[0] for row in existing.all()}
    new_ids = [tid for tid in tag_ids if tid not in existing_ids]

    if new_ids:
        from sqlalchemy import insert
        await db.execute(
            insert(photo_tags),
            [{"photo_id": photo_id, "tag_id": tid} for tid in new_ids],
        )
        await db.commit()

    return {"status": "success", "added": len(new_ids)}


@router.delete("/{photo_id}/tags/{tag_id}", status_code=204)
async def remove_photo_tag(
    photo_id: str,
    tag_id: str,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Remove a tag from a photo. Removing a non-member is a no-op."""
    await get_user_photo(photo_id, current_user, db)
    from sqlalchemy import delete as sql_delete
    await db.execute(
        sql_delete(photo_tags).where(
            photo_tags.c.photo_id == photo_id,
            photo_tags.c.tag_id == tag_id,
        )
    )
    await db.commit()
    return None

async def _get_photo_with_share_fallback(
    photo_id: str, user: User, db: AsyncSession,
) -> Photo:
    """Fetch a photo the user owns, or one they can access via a share.

    The fast path (owned photo) does a single indexed query. The share
    fallback only runs when the first query returns nothing — this
    happens only for shared photos, not during normal browsing.
    """
    # Fast path — owned photo (single indexed query, no extra joins).
    result = await db.execute(
        select(Photo).where(Photo.id == photo_id, Photo.user_id == user.id)
    )
    photo = result.scalar_one_or_none()
    if photo:
        return photo
    # Slow path — check share access (only for shared photos).
    if await can_access_photo_via_share(photo_id, user, db):
        result = await db.execute(
            select(Photo).where(Photo.id == photo_id)
        )
        photo = result.scalar_one_or_none()
        if photo:
            return photo
    raise HTTPException(status_code=404, detail="Photo not found")


# Pixel box mule's three logical sizes map to. Nextcloud's preview
# endpoint takes (x, y) as a bounding box and `a=true` preserves the
# source aspect ratio, so passing a square box is fine. Keep these in
# sync with the worker's THUMB_SIZES if you ever change them.
_NC_PREVIEW_PX = {"small": 240, "medium": 640, "large": 1280}


@router.get("/{photo_id}/thumb/{size}")
async def get_thumbnail(
    photo_id: str,
    size: str,
    response: Response,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user_media),
):
    """Serve a thumbnail.

    Primary path: proxy Nextcloud's `/index.php/core/preview` for the
    photo's `nextcloud_fileid`, authenticated with the owner's NC app
    password. Nextcloud already maintains previews for the source file;
    duplicating that work in `/data/thumbs/*` was burning disk and CPU.

    Fallback path: legacy / non-NC photos (where `nextcloud_fileid` is
    NULL) and any NC error keep working through the original on-disk
    thumbnail cache + inline-generate fallback. The fallback is
    intentionally identical to the old handler so a revert is one
    file diff.
    """
    if size not in _NC_PREVIEW_PX:
        raise HTTPException(status_code=400, detail="Invalid thumbnail size")

    photo = await _get_photo_with_share_fallback(photo_id, current_user, db)

    # ── Primary: proxy Nextcloud's preview endpoint ────────────────────
    if photo.nextcloud_fileid and photo.user_id:
        owner = (
            await db.execute(select(User).where(User.id == photo.user_id))
        ).scalar_one_or_none()
        if owner is not None:
            try:
                upstream = await get_preview_async(
                    owner,
                    photo.nextcloud_fileid,
                    _NC_PREVIEW_PX[size],
                    _NC_PREVIEW_PX[size],
                )
            except NextcloudCredentialsMissing:
                # Owner hasn't set their NC app password — fall through
                # to disk; that path still works for them.
                upstream = None
            except Exception as e:
                logger.warning(
                    "NC preview proxy failed for photo %s size=%s: %s",
                    photo_id, size, e,
                )
                upstream = None
            if upstream is not None and upstream.is_success:
                headers = {
                    "Cache-Control": "private, max-age=86400",
                    "X-Mule-Thumb-Source": "nextcloud",
                }
                etag = upstream.headers.get("etag")
                if etag:
                    headers["ETag"] = etag
                media_type = upstream.headers.get(
                    "content-type", "image/jpeg"
                )
                return Response(
                    content=upstream.content,
                    media_type=media_type,
                    headers=headers,
                )
            # Non-success or exception: log + fall through.
            if upstream is not None:
                logger.info(
                    "NC preview returned %s for photo=%s fileid=%s — falling back to disk",
                    upstream.status_code, photo_id, photo.nextcloud_fileid,
                )

    # ── Fallback: on-disk thumbnail (unchanged from pre-NC-proxy) ──────
    if photo.user_id:
        thumb_dir = f"/data/thumbs/{photo.user_id}/{photo_id}"
    else:
        thumb_dir = f"/data/thumbs/{photo_id}"
    thumb_path = f"{thumb_dir}/{size}.webp"

    if not os.path.exists(thumb_path):
        # Queue background generation (handles RAW/HEIC/video properly)
        from app.tasks.thumbs import generate_thumbnails
        generate_thumbnails.delay(photo_id)

        # Best-effort inline fallback for standard images so the first
        # request doesn't have to wait for the worker. RAW/HEIC/video
        # and missing source files fall through to a clean 404 below.
        if photo.filepath and os.path.exists(photo.filepath):
            from PIL import Image
            try:
                os.makedirs(thumb_dir, exist_ok=True)
                img = Image.open(photo.filepath)

                # Auto-rotate based on EXIF
                from PIL import ExifTags
                try:
                    for orientation in ExifTags.TAGS.keys():
                        if ExifTags.TAGS[orientation] == 'Orientation':
                            break
                    exif = img._getexif()
                    if exif is not None:
                        orient = exif.get(orientation)
                        if orient == 3:
                            img = img.rotate(180, expand=True)
                        elif orient == 6:
                            img = img.rotate(270, expand=True)
                        elif orient == 8:
                            img = img.rotate(90, expand=True)
                except:
                    pass

                # Generate thumbnail size
                sizes = {'small': 150, 'medium': 400, 'large': 800}
                target_size = sizes.get(size, 400)
                img.thumbnail((target_size, target_size), Image.Resampling.LANCZOS)

                # Save as WebP
                img.save(thumb_path, 'WEBP', quality=85, optimize=True)

            except HTTPException:
                raise
            except Exception as e:
                logger.warning(
                    f"Inline thumbnail fallback failed for {photo_id} ({size}); "
                    f"waiting on worker: {e}"
                )

    # If the inline fallback didn't (or couldn't) produce the file,
    # tell the client to retry instead of crashing in FileResponse/nginx.
    if not os.path.exists(thumb_path):
        raise HTTPException(
            status_code=404,
            detail="Thumbnail not ready",
            headers={"Retry-After": "2"},
        )

    response.headers["X-Mule-Thumb-Source"] = "disk"
    # Check if we're behind Nginx
    if os.environ.get('USE_X_ACCEL_REDIRECT'):
        # Use Nginx X-Accel-Redirect for better performance
        response.headers['X-Accel-Redirect'] = f'/internal_thumbs/{photo_id}/{size}.webp'
        response.headers['Content-Type'] = 'image/webp'
        return Response()
    else:
        # Direct file serving for development
        return FileResponse(thumb_path, media_type='image/webp')

_INLINE_MEDIA_TYPES = {
    '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg',
    '.png': 'image/png', '.webp': 'image/webp', '.gif': 'image/gif',
    '.mp4': 'video/mp4', '.mov': 'video/quicktime',
    '.webm': 'video/webm', '.mkv': 'video/x-matroska',
    '.m4v': 'video/mp4',
}

# How big each chunk we yield is when streaming a Range response. 1 MB
# strikes a balance between syscall count and memory residency.
_RANGE_CHUNK = 1024 * 1024


def _parse_range(header: str, size: int) -> Optional[tuple[int, int]]:
    """Parse a single-range `Range: bytes=START-END` header.

    Returns (start, end) inclusive on success, or None when the header
    is malformed / multi-range (we don't bother with multipart). The
    caller falls back to a 200 response in that case.
    """
    if not header or not header.startswith("bytes="):
        return None
    spec = header[len("bytes="):]
    if "," in spec:  # multi-range; punt
        return None
    if "-" not in spec:
        return None
    start_s, end_s = spec.split("-", 1)
    try:
        if start_s == "":
            # bytes=-N → last N bytes
            n = int(end_s)
            if n <= 0:
                return None
            start = max(0, size - n)
            end = size - 1
        else:
            start = int(start_s)
            end = int(end_s) if end_s else size - 1
    except ValueError:
        return None
    if start < 0 or start >= size or end < start:
        return None
    end = min(end, size - 1)
    return start, end


@router.get("/{photo_id}/original")
async def get_original(
    photo_id: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user_media),
):
    """Serve the original file inline (web-safe formats) with HTTP Range
    support so `<video>` can seek and stream.

    Browsers refuse to play long `<video>` they can't seek — without
    Accept-Ranges + 206 they surface the failure as "format not
    supported" even when the codec itself is fine.
    """
    photo = await _get_photo_with_share_fallback(photo_id, current_user, db)

    if not os.path.exists(photo.filepath):
        raise HTTPException(status_code=404, detail="File not found")

    ext = Path(photo.filepath).suffix.lower()
    media_type = _INLINE_MEDIA_TYPES.get(ext, 'application/octet-stream')
    return _serve_file_with_range(
        photo.filepath,
        request,
        media_type,
        download_filename=(
            photo.filename if media_type == 'application/octet-stream' else None
        ),
    )


def _serve_file_with_range(
    path: str,
    request: Request,
    media_type: str,
    download_filename: Optional[str] = None,
) -> Response:
    """Range-aware file streamer extracted so /original and /playback share
    one implementation. Returns 200 (full body) when no Range header is
    present, 206 with the requested slice otherwise. Browsers refuse to
    seek long videos without Accept-Ranges + 206, so this is mandatory for
    <video> playback rather than a nice-to-have."""
    file_size = os.path.getsize(path)
    range_header = request.headers.get("range")
    parsed = _parse_range(range_header, file_size) if range_header else None

    if parsed is None:
        return FileResponse(
            path,
            filename=download_filename,
            media_type=media_type,
            headers={"Accept-Ranges": "bytes"},
        )

    start, end = parsed
    length = end - start + 1

    def _iter_range():
        with open(path, 'rb') as f:
            f.seek(start)
            remaining = length
            while remaining > 0:
                chunk = f.read(min(_RANGE_CHUNK, remaining))
                if not chunk:
                    break
                remaining -= len(chunk)
                yield chunk

    return StreamingResponse(
        _iter_range(),
        status_code=206,
        media_type=media_type,
        headers={
            "Content-Range": f"bytes {start}-{end}/{file_size}",
            "Content-Length": str(length),
            "Accept-Ranges": "bytes",
        },
    )


@router.get("/{photo_id}/playback")
async def get_playback(
    photo_id: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user_media),
):
    """Serve a video in a format every browser can play.

    iPhone .mov files are HEVC Main 10 (codec_tag hvc1), which Chrome and
    Firefox cannot decode reliably — Safari is the only browser that
    handles 10-bit HEVC out of the box. New video rows are pre-transcoded
    by `pretranscode_video` from scan_folder, so the typical hit here
    serves straight from `/data/video-cache/{id}.mp4`. The sync
    transcode-on-miss path stays as a fallback for backfill races and
    pre-existing rows that haven't run through the celery task yet."""
    photo = await _get_photo_with_share_fallback(photo_id, current_user, db)
    if not os.path.exists(photo.filepath):
        raise HTTPException(status_code=404, detail="File not found")

    src_path = photo.filepath
    cache_path = video_service.cache_path_for(photo.id)

    # Source-newer-than-cache invalidates the cache. Covers in-place file
    # replacement (re-upload via NC) without leaving stale playback bytes.
    if cache_path.exists():
        try:
            if os.path.getmtime(src_path) > os.path.getmtime(cache_path):
                cache_path.unlink()
        except OSError:
            pass

    if cache_path.exists():
        return _serve_file_with_range(str(cache_path), request, 'video/mp4')

    # Passthrough fast path: source is already h264 in a web-safe container.
    ext = Path(src_path).suffix.lower()
    if ext in video_service.PLAYBACK_OK_EXTS:
        codec = await asyncio.to_thread(video_service.ffprobe_video_codec, src_path)
        if codec in video_service.PLAYBACK_OK_VCODECS:
            return _serve_file_with_range(
                src_path,
                request,
                _INLINE_MEDIA_TYPES.get(ext, 'video/mp4'),
            )

    # Cache miss + needs transcode: sync fallback. Long videos are the
    # painful case — they're why pretranscode_video runs at scan time.
    ok = await asyncio.to_thread(
        video_service.transcode_to_h264_mp4, src_path, str(cache_path),
    )
    if not ok:
        raise HTTPException(status_code=500, detail="Video transcoding failed")
    return _serve_file_with_range(str(cache_path), request, 'video/mp4')


# Extensions that the browser can decode natively. Anything else (RAW, HEIC,
# TIFF) needs the /proxy endpoint to convert to WebP for display.
_WEB_SAFE_DISPLAY_EXTS = {'.jpg', '.jpeg', '.png', '.webp', '.gif'}


def _generate_proxy_webp(src_path: str, dst_path: str) -> None:
    """Decode src_path with the appropriate backend and write a full-res WebP
    to dst_path. Used by GET /photos/{id}/proxy for RAW/HEIC/TIFF display.

    Conservative: catches per-format failures and falls back to extracting an
    embedded preview where possible (RAW), so a single broken file never
    crashes the request.
    """
    from PIL import Image
    ext = Path(src_path).suffix.lower()

    img = None

    # RAW formats — decode via rawpy at full size
    raw_exts = {'.cr2', '.cr3', '.nef', '.nrw', '.arw', '.srf',
                '.raf', '.rw2', '.orf', '.srw', '.pef', '.rwl', '.dng'}
    if ext in raw_exts:
        try:
            import rawpy
            with rawpy.imread(src_path) as raw:
                rgb = raw.postprocess(use_camera_wb=True, no_auto_bright=False)
            img = Image.fromarray(rgb, 'RGB')
        except Exception as e:
            logger.warning(f"rawpy decode failed for {src_path}: {e}; trying embedded preview")
            try:
                import rawpy
                with rawpy.imread(src_path) as raw:
                    thumb = raw.extract_thumb()
                    if thumb.format == rawpy.ThumbFormat.JPEG:
                        from io import BytesIO
                        img = Image.open(BytesIO(thumb.data))
            except Exception as e2:
                # Last-resort: iPhone "Apple ProRAW" / Linear DNG isn't a
                # Bayer-pattern RAW — LibRaw rejects it. The file IS a TIFF
                # container with a developed RGB image inside, so PIL opens
                # it directly. Same fallback covers misnamed TIFFs.
                logger.warning(f"RAW preview extraction failed for {src_path}: {e2}; trying PIL TIFF fallback")
                try:
                    img = Image.open(src_path)
                except Exception as e3:
                    logger.error(f"PIL fallback also failed for {src_path}: {e3}")
                    raise HTTPException(status_code=415, detail="Unable to decode RAW file")

    # HEIC/HEIF — pillow-heif registers a PIL plugin
    elif ext in {'.heic', '.heif'}:
        try:
            from pillow_heif import register_heif_opener
            register_heif_opener()
            img = Image.open(src_path)
        except Exception as e:
            logger.error(f"HEIC decode failed for {src_path}: {e}")
            raise HTTPException(status_code=415, detail="Unable to decode HEIC file")

    # TIFF and any other PIL-supported format
    else:
        try:
            img = Image.open(src_path)
        except Exception as e:
            logger.error(f"PIL open failed for {src_path}: {e}")
            raise HTTPException(status_code=415, detail="Unable to decode image")

    # Auto-rotate via EXIF
    try:
        from PIL import ImageOps
        img = ImageOps.exif_transpose(img)
    except Exception:
        pass

    if img.mode not in ('RGB', 'RGBA'):
        img = img.convert('RGB')

    os.makedirs(os.path.dirname(dst_path), exist_ok=True)
    img.save(dst_path, 'WEBP', quality=90, method=4)


@router.get("/{photo_id}/proxy")
async def get_proxy(
    photo_id: str,
    response: Response,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user_media),
):
    """Serve a full-resolution WebP proxy for non-web-safe formats."""
    photo = await _get_photo_with_share_fallback(photo_id, current_user, db)

    if not os.path.exists(photo.filepath):
        raise HTTPException(status_code=404, detail="File not found")

    ext = Path(photo.filepath).suffix.lower()

    # Web-safe formats don't need a proxy — serve the original directly so the
    # browser uses its native decoder. Saves disk and CPU.
    if ext in _WEB_SAFE_DISPLAY_EXTS:
        return FileResponse(
            photo.filepath,
            media_type={
                '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg',
                '.png': 'image/png', '.webp': 'image/webp', '.gif': 'image/gif',
            }[ext],
        )

    proxy_dir = "/data/proxies"
    proxy_path = f"{proxy_dir}/{photo_id}.webp"

    if not os.path.exists(proxy_path):
        try:
            _generate_proxy_webp(photo.filepath, proxy_path)
        except HTTPException:
            raise
        except Exception as e:
            logger.error(f"Proxy generation failed for {photo_id}: {e}")
            raise HTTPException(status_code=500, detail="Proxy generation failed")

    if os.environ.get('USE_X_ACCEL_REDIRECT'):
        response.headers['X-Accel-Redirect'] = f'/internal_proxies/{photo_id}.webp'
        response.headers['Content-Type'] = 'image/webp'
        return Response()

    return FileResponse(proxy_path, media_type='image/webp')

@router.patch("/{photo_id}", response_model=PhotoResponse)
async def update_photo(
    photo_id: str,
    update: PhotoUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Update photo metadata."""
    photo = await get_user_photo(photo_id, current_user, db)

    update_data = update.dict(exclude_unset=True)

    # Filename rename: validate, rename on disk, then update both filename
    # and filepath atomically. Done before any other field changes so a
    # filesystem failure leaves the rest of the row untouched.
    if 'filename' in update_data:
        new_name = (update_data.pop('filename') or '').strip()
        if not new_name:
            raise HTTPException(status_code=400, detail="Filename cannot be empty")
        # Reject path separators and parent traversal — same-directory only.
        if '/' in new_name or '\\' in new_name or new_name in ('.', '..'):
            raise HTTPException(status_code=400, detail="Invalid filename")

        if new_name != photo.filename:
            current_dir = os.path.dirname(photo.filepath)
            new_path = os.path.join(current_dir, new_name)

            if not os.path.exists(photo.filepath):
                raise HTTPException(status_code=404, detail="Source file missing on disk")
            if os.path.exists(new_path):
                raise HTTPException(status_code=409, detail="A file with that name already exists")

            if is_nextcloud_path(photo.filepath):
                # WebDAV MOVE within the same directory == rename. Lands
                # the new name in oc_filecache so Nextcloud's web UI and
                # sync clients see it; the bind mount reflects the
                # rename for our own watcher.
                move_for_user(current_user, photo.filepath, new_path)
            else:
                try:
                    os.rename(photo.filepath, new_path)
                except OSError as e:
                    logger.error(f"Failed to rename {photo.filepath} -> {new_path}: {e}")
                    raise HTTPException(status_code=500, detail=f"Rename failed: {e}")

            photo.filename = new_name
            photo.filepath = new_path

    # taken_at edits write EXIF first, DB second — we'd rather surface a
    # failure than leave the DB ahead of the file on disk. On success the
    # source flips to 'manual' so the UI can render a badge and the next
    # rescan knows not to overwrite it.
    if 'taken_at' in update_data:
        new_dt = update_data.pop('taken_at')
        if new_dt is not None:
            # The frontend's <input type="datetime-local"> usually serializes
            # without a tz, but a manual edit / paste / certain locales can
            # send a tz-aware ISO (e.g. "2026-05-09T00:12+02:00"). The
            # photos.taken_at column is `timestamp without time zone`, so
            # asyncpg can't bind a tz-aware value — it raises
            # "can't subtract offset-naive and offset-aware datetimes".
            # Normalize to naive UTC so both shapes round-trip cleanly.
            if new_dt.tzinfo is not None:
                new_dt = new_dt.astimezone(timezone.utc).replace(tzinfo=None)
            try:
                await write_taken_at(photo.filepath, new_dt)
            except ExifWriteError as exc:
                raise HTTPException(
                    status_code=500,
                    detail=f"Failed to write EXIF: {exc}",
                )
        photo.taken_at = new_dt
        photo.taken_at_source = 'manual'
        # Recompute warning: a manual edit usually clears it (user just
        # told us the right date), but if they set it to something that
        # still disagrees with the folder path we'd rather keep the
        # flag up than pretend the problem's gone.
        photo.has_date_warning = compute_date_warning(photo.filepath, new_dt)

    # Apply remaining updates
    for field, value in update_data.items():
        setattr(photo, field, value)

    await db.commit()
    await db.refresh(photo)
    
    return PhotoResponse.from_orm(photo)

@router.delete("/{photo_id}")
async def discard_photo(
    photo_id: str,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Soft-discard a photo."""
    photo = await get_user_photo(photo_id, current_user, db)

    photo.is_discarded = True
    photo.discarded_at = datetime.utcnow()
    await db.commit()

    return {"status": "success", "message": "Photo discarded"}

class MoveRequest(BaseModel):
    photo_ids: list[str]
    target_id: str  # folder id OR source root id


class CopyRequest(BaseModel):
    photo_ids: list[str]
    target_id: str  # folder id OR source root id


@router.post("/copy")
async def copy_photos(
    body: CopyRequest,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Copy photos into a target folder. Same target resolution as /move
    (folder id or source root id), but uses shutil.copy2 and creates new
    Photo rows for each copied file. Original photos are unaffected.

    Each new row gets is_duplicate=true so the user can spot the
    duplicates later. The new file's name is suffixed with " (copy)" if
    a name collision would otherwise happen, and " (copy 2)", etc., for
    further conflicts.
    """
    import shutil

    # Resolve target_id → (target_dir, target_folder)
    sr_check = await db.execute(
        select(SourceRoot).where(SourceRoot.id == body.target_id)
    )
    source_root = sr_check.scalar_one_or_none()

    if source_root is not None:
        target_dir = source_root.path
        from app.tasks.scan import get_or_create_folder
        target_folder = await get_or_create_folder(db, target_dir, source_root.id)
    else:
        folder_check = await db.execute(
            select(Folder).where(Folder.id == body.target_id)
        )
        target_folder = folder_check.scalar_one_or_none()
        if target_folder is None:
            raise HTTPException(status_code=404, detail="Target folder not found")
        target_dir = target_folder.path

    if not os.path.isdir(target_dir):
        raise HTTPException(
            status_code=400,
            detail=f"Target directory does not exist: {target_dir}",
        )

    if not body.photo_ids:
        return {"status": "success", "copied": 0, "errors": []}

    photos_result = await db.execute(
        select(Photo).where(Photo.id.in_(body.photo_ids), Photo.user_id == current_user.id)
    )
    photos_to_copy = photos_result.scalars().all()

    copied = 0
    errors: list[dict] = []

    def _unique_target_name(directory: str, filename: str) -> Optional[str]:
        """Find a non-colliding filename in `directory` based on `filename`,
        suffixing " (copy)", " (copy 2)", ... if needed. Gives up after 100
        attempts."""
        if not os.path.exists(os.path.join(directory, filename)):
            return filename
        stem, ext = os.path.splitext(filename)
        for i in range(1, 100):
            candidate = f"{stem} (copy{'' if i == 1 else f' {i}'}){ext}"
            if not os.path.exists(os.path.join(directory, candidate)):
                return candidate
        return None

    target_nc = is_nextcloud_path(target_dir)
    if target_nc:
        # Copy into a Nextcloud-managed folder isn't routed through
        # WebDAV PUT yet — would need a download from the source path
        # plus an upload, doubling IO. Defer until someone needs it.
        raise _nc_unsupported(
            "Copy into a Nextcloud-managed folder isn't supported yet."
        )

    for photo in photos_to_copy:
        if not os.path.exists(photo.filepath):
            errors.append({"id": photo.id, "error": "source file missing"})
            continue

        new_name = _unique_target_name(target_dir, photo.filename)
        if new_name is None:
            errors.append({"id": photo.id, "error": "too many name collisions"})
            continue

        new_path = os.path.join(target_dir, new_name)

        if is_nextcloud_path(photo.filepath):
            # Source is NC, dest is local. We could `shutil.copy2`
            # since NC files are readable on the bind mount, but the
            # local /photos tree has different ownership semantics —
            # leave this off until we've thought about it.
            errors.append({
                "id": photo.id,
                "error": "Cross-system copy (Nextcloud ↔ local) not supported yet",
            })
            continue

        try:
            shutil.copy2(photo.filepath, new_path)
        except OSError as e:
            errors.append({"id": photo.id, "error": str(e)})
            continue

        # Create a new Photo row pointing at the copy. Most metadata is
        # copied verbatim; the file_hash stays so the duplicate flag does
        # the right thing across the library.
        new_photo = Photo(
            filepath=new_path,
            filename=new_name,
            folder_id=target_folder.id,
            file_hash=photo.file_hash,
            media_type=photo.media_type,
            original_format=photo.original_format,
            width=photo.width,
            height=photo.height,
            file_size=photo.file_size,
            taken_at=photo.taken_at,
            taken_at_source=photo.taken_at_source,
            user_id=current_user.id,
            user_title=photo.user_title,
            user_notes=photo.user_notes,
            rating=photo.rating,
            color_label=photo.color_label,
            exif_json=photo.exif_json,
            is_duplicate=True,
            processing_status='pending',
        )
        db.add(new_photo)
        copied += 1

    await db.commit()

    return {
        "status": "success",
        "copied": copied,
        "errors": errors,
    }


@router.post("/move")
async def move_photos(
    body: MoveRequest,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Move photos into a target folder. The target can be either a Folder
    id or a SourceRoot id (since the LeftSidebar only exposes source roots
    today). The handler resolves the target to an on-disk directory, calls
    shutil.move for each photo, and updates photo.filepath + folder_id.

    Per-file failures (target name collision, missing source) are collected
    and returned in the response so a single bad photo doesn't abort the
    batch.
    """
    import shutil

    # Resolve target_id → (target_dir, target_folder)
    sr_check = await db.execute(
        select(SourceRoot).where(SourceRoot.id == body.target_id)
    )
    source_root = sr_check.scalar_one_or_none()

    if source_root is not None:
        target_dir = source_root.path
        # We need a Folder row to point photo.folder_id at. Reuse the
        # scanner's get_or_create helper so we don't duplicate the dedupe
        # / normalization logic.
        from app.tasks.scan import get_or_create_folder
        target_folder = await get_or_create_folder(db, target_dir, source_root.id)
    else:
        folder_check = await db.execute(
            select(Folder).where(Folder.id == body.target_id)
        )
        target_folder = folder_check.scalar_one_or_none()
        if target_folder is None:
            raise HTTPException(status_code=404, detail="Target folder not found")
        target_dir = target_folder.path

    if not os.path.isdir(target_dir):
        raise HTTPException(
            status_code=400,
            detail=f"Target directory does not exist: {target_dir}",
        )

    if not body.photo_ids:
        return {"status": "success", "moved": 0, "errors": []}

    # Fetch the photo rows, scoped to user
    photos_result = await db.execute(
        select(Photo).where(Photo.id.in_(body.photo_ids), Photo.user_id == current_user.id)
    )
    photos_to_move = photos_result.scalars().all()

    moved = 0
    errors: list[dict] = []

    for photo in photos_to_move:
        # Skip if already in the target folder.
        if photo.folder_id == target_folder.id:
            continue

        new_path = os.path.join(target_dir, photo.filename)

        if not os.path.exists(photo.filepath):
            errors.append({"id": photo.id, "error": "source file missing"})
            continue
        if os.path.exists(new_path):
            errors.append({"id": photo.id, "error": f"name already exists in target: {photo.filename}"})
            continue

        src_nc = is_nextcloud_path(photo.filepath)
        dst_nc = is_nextcloud_path(new_path)
        if src_nc != dst_nc:
            # Cross-system moves (NC ↔ local /photos) aren't supported
            # in v1 — the user can copy via Nextcloud's web UI or use
            # the desktop sync client to move the file, and mule-image
            # will pick up both sides via inotify.
            errors.append({
                "id": photo.id,
                "error": "Cross-system move (Nextcloud ↔ local) not supported yet",
            })
            continue

        try:
            if src_nc:
                move_for_user(current_user, photo.filepath, new_path)
            else:
                shutil.move(photo.filepath, new_path)
        except HTTPException as e:
            errors.append({"id": photo.id, "error": str(e.detail)})
            continue
        except OSError as e:
            errors.append({"id": photo.id, "error": str(e)})
            continue

        photo.filepath = new_path
        photo.folder_id = target_folder.id
        moved += 1

    await db.commit()

    return {
        "status": "success",
        "moved": moved,
        "errors": errors,
    }


@router.post("/bulk")
async def bulk_action(
    action: BulkAction,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Perform bulk actions on multiple photos"""
    # Get photos, scoped to user
    result = await db.execute(
        select(Photo).where(Photo.id.in_(action.ids), Photo.user_id == current_user.id)
    )
    photos = result.scalars().all()
    
    if not photos:
        raise HTTPException(status_code=404, detail="No photos found")
    
    # Perform action based on type
    if action.action == 'discard':
        for photo in photos:
            photo.is_discarded = True
            photo.discarded_at = datetime.utcnow()
    elif action.action == 'restore':
        for photo in photos:
            photo.is_discarded = False
            photo.discarded_at = None
    elif action.action == 'set_rating':
        for photo in photos:
            photo.rating = action.value
    elif action.action == 'set_color':
        for photo in photos:
            photo.color_label = action.value
    elif action.action == 'set_notes':
        # value is the replacement notes string (empty string clears).
        # Sent verbatim — no whitespace trimming, callers can pre-trim
        # client-side if they want.
        if action.value is not None and not isinstance(action.value, str):
            raise HTTPException(
                status_code=400,
                detail="set_notes requires a string value (or null to clear)",
            )
        new_notes = action.value or None  # empty string -> null
        for photo in photos:
            photo.user_notes = new_notes
    elif action.action in ('set_taken_at', 'set_taken_at_map'):
        # Two shapes share one code path:
        #   set_taken_at      → value is one ISO datetime, applied to every id
        #   set_taken_at_map  → value is {photo_id: iso datetime}, per-photo
        # The per-photo variant is what the "guess from folder" bulk flow
        # uses when every selected photo gets a different date.
        if action.action == 'set_taken_at':
            if not isinstance(action.value, str) or not action.value:
                raise HTTPException(
                    status_code=400,
                    detail="set_taken_at requires an ISO datetime string",
                )
            try:
                uniform_dt = datetime.fromisoformat(action.value)
            except ValueError:
                raise HTTPException(
                    status_code=400,
                    detail="Invalid ISO datetime for set_taken_at",
                )
            date_map = {p.id: uniform_dt for p in photos}
        else:
            if not isinstance(action.value, dict) or not action.value:
                raise HTTPException(
                    status_code=400,
                    detail="set_taken_at_map requires a {id: iso} mapping",
                )
            date_map = {}
            for pid, raw in action.value.items():
                if not isinstance(raw, str):
                    continue
                try:
                    date_map[pid] = datetime.fromisoformat(raw)
                except ValueError:
                    continue

        updated = 0
        errors: list[dict[str, str]] = []
        for photo in photos:
            new_dt = date_map.get(photo.id)
            if new_dt is None:
                continue
            try:
                await write_taken_at(photo.filepath, new_dt)
            except ExifWriteError as exc:
                errors.append({"id": photo.id, "message": str(exc)})
                continue
            photo.taken_at = new_dt
            photo.taken_at_source = 'manual'
            photo.has_date_warning = compute_date_warning(photo.filepath, new_dt)
            updated += 1

        await db.commit()
        return {
            "status": "success",
            "updated": updated,
            "skipped": len(photos) - updated - len(errors),
            "errors": errors,
        }
    elif action.action == 'add_tags':
        # value is a list of tag ids. We bulk-insert (photo_id, tag_id)
        # rows for every (photo, tag) combination that doesn't already
        # exist, so the operation is idempotent.
        tag_ids = action.value or []
        if not isinstance(tag_ids, list) or not tag_ids:
            return {"status": "success", "added": 0, "message": "No tags supplied"}
        photo_ids = [p.id for p in photos]
        existing = await db.execute(
            select(photo_tags.c.photo_id, photo_tags.c.tag_id).where(
                photo_tags.c.photo_id.in_(photo_ids),
                photo_tags.c.tag_id.in_(tag_ids),
            )
        )
        existing_pairs = {(row[0], row[1]) for row in existing.all()}
        new_rows = [
            {"photo_id": pid, "tag_id": tid}
            for pid in photo_ids
            for tid in tag_ids
            if (pid, tid) not in existing_pairs
        ]
        if new_rows:
            from sqlalchemy import insert
            await db.execute(insert(photo_tags), new_rows)
        await db.commit()
        return {
            "status": "success",
            "added": len(new_rows),
            "message": f"Added {len(new_rows)} tag link{'s' if len(new_rows) != 1 else ''}",
        }
    elif action.action == 'remove_tags':
        tag_ids = action.value or []
        if not isinstance(tag_ids, list) or not tag_ids:
            return {"status": "success", "removed": 0, "message": "No tags supplied"}
        photo_ids = [p.id for p in photos]
        from sqlalchemy import delete as sql_delete
        result = await db.execute(
            sql_delete(photo_tags).where(
                photo_tags.c.photo_id.in_(photo_ids),
                photo_tags.c.tag_id.in_(tag_ids),
            )
        )
        await db.commit()
        return {
            "status": "success",
            "removed": result.rowcount or 0,
            "message": f"Removed tag link{'s' if (result.rowcount or 0) != 1 else ''}",
        }
    else:
        raise HTTPException(status_code=400, detail="Invalid action")
    
    await db.commit()
    
    return {
        "status": "success",
        "message": f"{action.action} applied to {len(photos)} photos"
    }