feat(nc-webhook): receive Nextcloud file events instead of polling

The watchfiles-based watcher works but duplicates Nextcloud's own
notion of "this file changed." NC has a webhook_listeners app that
can POST file events to an external URL. This adds the mule side of
that handshake.

- POST /api/v1/internal/nc-webhook authenticates a Bearer token
  (NEXTCLOUD_WEBHOOK_SECRET, hmac.compare_digest) and dispatches the
  same scan_folder / handle_file_deletion machinery the watcher used.
- Handles NodeCreated, NodeWritten, NodeDeleted, NodeRenamed.
  Renamed is mapped to delete-old + scan-new-parent. Maps NC's
  /admin/files/... path to the bind-mounted /nextcloud-users/admin/files/...
- backend/scripts/register_nc_webhooks.py is the idempotent
  registrar: lists existing webhooks, deletes any pointing at the
  target URL, then POSTs four fresh ones via OCS.
- Sets the env passthrough on backend + all workers in compose so
  the same secret is available wherever the registrar might run.

watch_folders stays in place for now — webhooks become primary, the
watcher is a belt-and-suspenders fallback. Drop the watcher in a
follow-up once webhooks are proven reliable on this NC instance.

backend/app/main.py

@@ -12,7 +12,7 @@ import os
 
 from app.config import settings
 from app.database import init_db
-from app.routers import photos, folders, heaps, tags, discard, library, search, auth, admin, sharing, upload, download, features, nextcloud
+from app.routers import photos, folders, heaps, tags, discard, library, search, auth, admin, sharing, upload, download, features, nextcloud, nc_webhook
 from app.services.scanner import start_initial_scan, bootstrap_default_source_root
 from app.services.cleanup import cleanup_data_integrity
 
@@ -113,6 +113,7 @@ app.include_router(upload.router, prefix="/api/v1/upload", tags=["upload"])
 app.include_router(download.router, prefix="/api/v1/download", tags=["download"])
 app.include_router(features.router, prefix="/api/v1/features", tags=["features"])
 app.include_router(nextcloud.router, prefix="/api/v1/nextcloud", tags=["nextcloud"])
+app.include_router(nc_webhook.router, prefix="/api/v1/internal", tags=["nc-webhook"])
 
 @app.get("/")
 async def root():

backend/app/routers/nc_webhook.py

@@ -0,0 +1,179 @@
+"""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
+
+    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"}
+        if Path(abs_path).suffix.lower() not in supported:
+            return {"status": "ignored", "reason": "unsupported extension"}
+        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 old_abs and Path(old_abs).suffix.lower() in supported:
+            await handle_file_deletion(old_abs)
+        if new_abs and 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"}

backend/scripts/register_nc_webhooks.py

@@ -0,0 +1,173 @@
+"""Register (or re-register) the four file-event webhooks against the
+Nextcloud webhook_listeners app, pointing them at mule's internal
+receiver.
+
+Idempotent: deletes any existing webhooks whose URI matches the target
+mule URL before posting fresh ones. Run after a config change (target
+URL, secret) or after the NC stack is rebuilt fresh.
+
+    docker exec mulita-backend python -m scripts.register_nc_webhooks
+
+Reads:
+- NEXTCLOUD_BASE_URL          — already set for the DAV client
+- NEXTCLOUD_WEBHOOK_TARGET    — http URL of mule's webhook endpoint
+                                 (default: http://192.168.8.136:8001/api/v1/internal/nc-webhook)
+- NEXTCLOUD_WEBHOOK_SECRET    — shared bearer secret; must match the
+                                 backend env var of the same name
+
+Registers as the first mule user with `is_admin=true` (or, failing
+that, the first user with NC credentials configured). The OCS endpoint
+itself only requires basic auth as that NC user.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import sys
+from typing import Any
+
+import httpx
+from sqlalchemy import select
+
+from app.database import AsyncSessionLocal
+from app.models.user import User
+from app.services.secrets import decrypt
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+)
+logger = logging.getLogger("register_nc_webhooks")
+
+DEFAULT_TARGET = "http://192.168.8.136:8001/api/v1/internal/nc-webhook"
+
+EVENTS = [
+    "OCP\\Files\\Events\\Node\\NodeCreatedEvent",
+    "OCP\\Files\\Events\\Node\\NodeWrittenEvent",
+    "OCP\\Files\\Events\\Node\\NodeDeletedEvent",
+    "OCP\\Files\\Events\\Node\\NodeRenamedEvent",
+]
+
+
+async def _pick_admin() -> User | None:
+    """Pick a mule user we can use to authenticate against NC's OCS
+    API. Prefer admin role, fall back to any user with NC creds set."""
+    async with AsyncSessionLocal() as s:
+        # First try admins.
+        r = await s.execute(
+            select(User).where(
+                User.role == "admin",
+                User.nextcloud_app_password_enc.is_not(None),
+            )
+        )
+        u = r.scalars().first()
+        if u:
+            return u
+        # Fall back to any user with creds.
+        r = await s.execute(
+            select(User).where(User.nextcloud_app_password_enc.is_not(None))
+        )
+        return r.scalars().first()
+
+
+def _ocs(base: str, path: str) -> str:
+    return f"{base.rstrip('/')}/ocs/v2.php/apps/webhook_listeners/api/v1{path}"
+
+
+def _ocs_headers() -> dict[str, str]:
+    # OCS-APIRequest header is mandatory for OCS endpoints.
+    return {
+        "OCS-APIRequest": "true",
+        "Accept": "application/json",
+    }
+
+
+def _list_existing(c: httpx.Client, base: str) -> list[dict[str, Any]]:
+    r = c.get(_ocs(base, "/webhooks"), headers=_ocs_headers())
+    r.raise_for_status()
+    body = r.json()
+    return body.get("ocs", {}).get("data", []) or []
+
+
+def _delete(c: httpx.Client, base: str, webhook_id: str) -> None:
+    r = c.delete(_ocs(base, f"/webhooks/{webhook_id}"), headers=_ocs_headers())
+    if r.status_code not in (200, 204):
+        logger.warning(
+            "delete webhook %s returned %s: %s",
+            webhook_id, r.status_code, r.text[:200],
+        )
+
+
+def _register(
+    c: httpx.Client,
+    base: str,
+    target: str,
+    secret: str,
+    event_class: str,
+) -> dict[str, Any]:
+    body = {
+        "uri": target,
+        "httpMethod": "POST",
+        "event": event_class,
+        "authMethod": "header",
+        "authData": {"Authorization": f"Bearer {secret}"},
+        "headers": {"Content-Type": "application/json"},
+    }
+    r = c.post(_ocs(base, "/webhooks"), headers=_ocs_headers(), json=body)
+    r.raise_for_status()
+    return r.json().get("ocs", {}).get("data", {})
+
+
+async def main() -> int:
+    base = os.environ.get("NEXTCLOUD_BASE_URL", "").rstrip("/")
+    target = os.environ.get("NEXTCLOUD_WEBHOOK_TARGET", DEFAULT_TARGET)
+    secret = os.environ.get("NEXTCLOUD_WEBHOOK_SECRET", "")
+    if not base:
+        logger.error("NEXTCLOUD_BASE_URL is not set")
+        return 2
+    if not secret:
+        logger.error("NEXTCLOUD_WEBHOOK_SECRET is not set")
+        return 2
+
+    user = await _pick_admin()
+    if user is None:
+        logger.error(
+            "no mule user has nextcloud_app_password_enc set; can't auth to OCS"
+        )
+        return 2
+
+    nc_user = user.nextcloud_username or user.username
+    pw = decrypt(user.nextcloud_app_password_enc)
+    if not nc_user or not pw:
+        logger.error("user %s has incomplete NC credentials", user.username)
+        return 2
+
+    auth = httpx.BasicAuth(nc_user, pw)
+    with httpx.Client(timeout=30, auth=auth, follow_redirects=False) as c:
+        existing = _list_existing(c, base)
+        logger.info("found %d existing webhook(s)", len(existing))
+
+        # Delete any pointing at the same target URI — idempotent rerun.
+        for w in existing:
+            if w.get("uri") == target:
+                logger.info(
+                    "removing existing webhook id=%s event=%s",
+                    w.get("id"), w.get("event"),
+                )
+                _delete(c, base, str(w["id"]))
+
+        # Register fresh.
+        for event_class in EVENTS:
+            data = _register(c, base, target, secret, event_class)
+            logger.info(
+                "registered: id=%s event=%s",
+                data.get("id"), event_class,
+            )
+
+    logger.info("done — %d webhooks registered against %s", len(EVENTS), target)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

docker-compose.yml

@@ -97,6 +97,7 @@ services:
       # but mutating endpoints fail with a clear error.
       - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users}
       - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-}
+      - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-}
       - LOG_LEVEL=${LOG_LEVEL:-INFO}
       - TZ=${TZ:-UTC}
     depends_on:
@@ -160,6 +161,7 @@ services:
       - PHOTO_DIRS=/photos
       - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users}
       - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-}
+      - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-}
       - LOG_LEVEL=${LOG_LEVEL:-INFO}
       - TZ=${TZ:-UTC}
       # NullPool — see app/database.py for rationale.
@@ -215,6 +217,7 @@ services:
       - PHOTO_DIRS=/photos
       - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users}
       - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-}
+      - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-}
       - LOG_LEVEL=${LOG_LEVEL:-INFO}
       - TZ=${TZ:-UTC}
       - MULITA_CELERY_WORKER=1
@@ -258,6 +261,7 @@ services:
       - PHOTO_DIRS=/photos
       - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users}
       - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-}
+      - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-}
       - LOG_LEVEL=${LOG_LEVEL:-INFO}
       - TZ=${TZ:-UTC}
       - MULITA_CELERY_WORKER=1