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

"""Internal webhook receiver for Nextcloud file events.

Replaces the watchfiles-based `watch_folders` Celery task: instead of
mule polling the bind-mount with inotify, Nextcloud's `webhook_listeners`
app POSTs here on every NodeCreated / NodeWritten / NodeDeleted /
NodeRenamed event, and we dispatch the same scan_folder /
handle_file_deletion machinery that the watcher used.

Auth: `Authorization: Bearer <NEXTCLOUD_WEBHOOK_SECRET>` header.
constant_time compare. 401 on mismatch, 401 also when the secret isn't
configured (fail closed).

The route is intentionally outside `/api/v1/photos/...` so it doesn't
get caught by the per-user auth middleware — webhook requests come
from Nextcloud as a service principal, not as a logged-in user. They
get NO mule app session.
"""
from __future__ import annotations

import hmac
import logging
import os
from pathlib import Path
from typing import Any

from fastapi import APIRouter, Header, HTTPException, Request, status
from sqlalchemy import select

from app.database import AsyncSessionLocal
from app.models.folders import SourceRoot
from app.services.nextcloud_dav import NEXTCLOUD_USERS_ROOT

logger = logging.getLogger(__name__)

router = APIRouter()


# Same list the on-disk watcher used (app.tasks.scan.SUPPORTED_EXTENSIONS).
# Imported lazily inside the handler so this module can load without
# pulling in the tasks package at startup.
def _supported_extensions() -> set[str]:
    from app.tasks.scan import SUPPORTED_EXTENSIONS
    return SUPPORTED_EXTENSIONS


def _expected_secret() -> str | None:
    return os.environ.get("NEXTCLOUD_WEBHOOK_SECRET") or None


def _nc_path_to_abs(nc_path: str) -> str | None:
    """Map a Nextcloud-internal path (`/admin/files/Photos/foo.jpg`) to
    the absolute bind-mount path mule's workers operate on
    (`/nextcloud-users/admin/files/Photos/foo.jpg`).

    Returns None for paths that don't sit under `<user>/files/...`
    (NC also emits events for trashbin, versions, etc — we ignore
    those).
    """
    if not nc_path or not nc_path.startswith("/"):
        return None
    parts = nc_path.lstrip("/").split("/", 2)
    if len(parts) < 3 or parts[1] != "files":
        return None
    return os.path.join(NEXTCLOUD_USERS_ROOT, parts[0], "files", parts[2])


def _classify(event_class: str) -> str | None:
    """Bucket the full event class string into the four buckets we act
    on. Returns None for events we don't care about (Before*, copy,
    touched, etc)."""
    short = event_class.rsplit("\\", 1)[-1]
    return {
        "NodeCreatedEvent": "created",
        "NodeWrittenEvent": "written",
        "NodeDeletedEvent": "deleted",
        "NodeRenamedEvent": "renamed",
    }.get(short)


async def _source_root_for(parent_dir: str) -> str | None:
    """Find the SourceRoot id whose path contains `parent_dir`."""
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            select(SourceRoot).where(SourceRoot.is_active == True)  # noqa: E712
        )
        roots = result.scalars().all()
    normalized = os.path.normpath(parent_dir)
    for sr in roots:
        root_path = os.path.normpath(sr.path)
        if normalized == root_path or normalized.startswith(root_path + os.sep):
            return sr.id
    return None


@router.post("/nc-webhook")
async def nc_webhook(
    request: Request,
    authorization: str | None = Header(default=None),
):
    """Receive a Nextcloud file event and dispatch the matching
    scan_folder / handle_file_deletion task. Returns 204 on success
    (Nextcloud doesn't care about the body)."""
    expected = _expected_secret()
    if not expected:
        # Fail closed: a misconfigured server should reject webhooks
        # rather than accept arbitrary POSTs.
        logger.error("nc-webhook hit but NEXTCLOUD_WEBHOOK_SECRET is not set")
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    presented = authorization[len("Bearer "):]
    if not hmac.compare_digest(presented, expected):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

    try:
        payload: dict[str, Any] = await request.json()
    except Exception:
        raise HTTPException(status_code=400, detail="malformed json")

    event = payload.get("event") or {}
    event_class = event.get("class") or ""
    bucket = _classify(event_class)
    if bucket is None:
        return {"status": "ignored", "reason": "unwanted event"}

    # Import lazily so this router can load before the celery app is
    # ready — important when the backend boots before broker is up.
    from app.tasks.scan import (
        scan_folder,
        handle_file_deletion,
        handle_directory_deletion,
        handle_directory_rename,
    )

    supported = _supported_extensions()

    if bucket in ("created", "written"):
        node = event.get("node") or {}
        nc_path = node.get("path")
        abs_path = _nc_path_to_abs(nc_path) if nc_path else None
        if not abs_path:
            return {"status": "ignored", "reason": "non-user-files path"}
        if Path(abs_path).suffix.lower() not in supported:
            return {"status": "ignored", "reason": "unsupported extension"}
        parent_dir = str(Path(abs_path).parent)
        source_root_id = await _source_root_for(parent_dir)
        if source_root_id is None:
            # Outside any registered source root — we don't index this
            # part of Nextcloud at all.
            return {"status": "ignored", "reason": "outside source root"}
        scan_folder.delay(parent_dir, source_root_id)
        logger.info("nc-webhook %s: queued scan_folder for %s", bucket, parent_dir)
        return {"status": "queued", "action": "scan_folder", "path": parent_dir}

    if bucket == "deleted":
        node = event.get("node") or {}
        nc_path = node.get("path")
        abs_path = _nc_path_to_abs(nc_path) if nc_path else None
        if not abs_path:
            return {"status": "ignored", "reason": "non-user-files path"}
        # Folder deletes: NC fires exactly one NodeDeletedEvent for the
        # folder, not one per child file. Detect the directory case by
        # the absence of a supported image extension and recursively
        # discard every Photo under that prefix.
        if Path(abs_path).suffix.lower() not in supported:
            n = await handle_directory_deletion(abs_path)
            logger.info(
                "nc-webhook deleted (dir): %s -> %s photos discarded",
                abs_path, n,
            )
            return {
                "status": "applied",
                "action": "discard_subtree",
                "path": abs_path,
                "discarded": n,
            }
        await handle_file_deletion(abs_path)
        logger.info("nc-webhook deleted: marked %s as discarded", abs_path)
        return {"status": "applied", "action": "discard", "path": abs_path}

    if bucket == "renamed":
        source = event.get("source") or {}
        target = event.get("target") or {}
        old_abs = _nc_path_to_abs(source.get("path") or "")
        new_abs = _nc_path_to_abs(target.get("path") or "")
        if not old_abs or not new_abs:
            return {"status": "ignored", "reason": "non-user-files path"}

        old_is_dir = Path(old_abs).suffix.lower() not in supported
        new_is_dir = Path(new_abs).suffix.lower() not in supported

        # Directory rename: NC fires one event for the directory; the
        # children's paths change implicitly. Prefix-rewrite in mule.
        # Same handler covers the feedback case where the PATCH
        # /folders/{id}/rename endpoint already updated the DB — the
        # SQL UPDATE matches zero rows the second time around.
        if old_is_dir and new_is_dir:
            result = await handle_directory_rename(old_abs, new_abs)
            logger.info(
                "nc-webhook renamed (dir): %s -> %s : %s",
                old_abs, new_abs, result,
            )
            return {
                "status": "applied",
                "action": "rename_subtree",
                "from": old_abs,
                "to": new_abs,
                **result,
            }

        # File rename (existing logic).
        if Path(old_abs).suffix.lower() in supported:
            await handle_file_deletion(old_abs)
        if Path(new_abs).suffix.lower() in supported:
            parent_dir = str(Path(new_abs).parent)
            source_root_id = await _source_root_for(parent_dir)
            if source_root_id is not None:
                scan_folder.delay(parent_dir, source_root_id)
        logger.info("nc-webhook renamed: %s -> %s", old_abs, new_abs)
        return {"status": "applied", "action": "rename", "from": old_abs, "to": new_abs}

    # Shouldn't reach here — classify() already filtered.
    return {"status": "ignored"}