diff --git a/nextcloud_mcp_server/search/access_filter.py b/nextcloud_mcp_server/search/access_filter.py index 5fcccf79..5903a557 100644 --- a/nextcloud_mcp_server/search/access_filter.py +++ b/nextcloud_mcp_server/search/access_filter.py @@ -162,10 +162,14 @@ def build_ownership_filter( """Build the Qdrant ``Filter`` constraining a search to readable points. Matches points whose ``owner_id`` is in ``accessible_owners`` (excluding - self) OR whose ``user_id`` equals ``user_id``. The ``user_id`` branch covers - *all* of the caller's own content — both new points (where - ``owner_id == user_id``) and legacy points indexed before ``owner_id`` - existed — so self is intentionally NOT repeated in the ``owner_id`` branch. + self) OR whose ``user_id`` equals ``user_id`` OR whose ``acl_principals`` + contains ``user:``. The ``user_id`` branch covers *all* of the + caller's own content — both new points (where ``owner_id == user_id``) and + legacy points indexed before ``owner_id`` existed — so self is intentionally + NOT repeated in the ``owner_id`` branch. The ``acl_principals`` branch covers + files that were indexed once and deduplicated across users (user-agnostic + point IDs): such a point's ``user_id``/``owner_id`` are the first indexer's, + so only the observed-access principal set surfaces it to other readers. Args: user_id: Querying user (matched by the ``user_id`` branch, which is the @@ -188,6 +192,13 @@ def build_ownership_filter( other_owners = [owner for owner in owners if owner != user_id] conditions: list[Condition] = [ FieldCondition(key="user_id", match=MatchValue(value=user_id)), + # Observed-access branch: a file deduplicated across users carries one + # user-agnostic point set whose ``user_id``/``owner_id`` are the first + # indexer's. ``acl_principals`` lists every user whose scanner has seen + # (hence can read) the file, so this branch surfaces a shared/group-folder + # file to every reader even when they were not the indexer. Verify-on-read + # (_verify_files) is the precise ACL gate on the returned candidates. + FieldCondition(key="acl_principals", match=MatchAny(any=[f"user:{user_id}"])), ] if other_owners: conditions.insert( diff --git a/nextcloud_mcp_server/vector/eviction.py b/nextcloud_mcp_server/vector/eviction.py index 8c7555a7..d4aaaed1 100644 --- a/nextcloud_mcp_server/vector/eviction.py +++ b/nextcloud_mcp_server/vector/eviction.py @@ -9,10 +9,7 @@ simply re-verify and re-attempt. import logging -from qdrant_client.models import FieldCondition, Filter, MatchValue - -from nextcloud_mcp_server.config import get_settings -from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client +from nextcloud_mcp_server.vector.sharing_state import release_document_for_user logger = logging.getLogger(__name__) @@ -22,39 +19,29 @@ async def delete_document_points( doc_type: str, user_id: str, ) -> None: - """Remove all Qdrant points for a single document. + """Revoke one user's access to a document's points (verify-on-read eviction). - Deletes both real chunk points and any leftover placeholder points for the - given (user_id, doc_id, doc_type) tuple. Safe to call when the document is + A document can be indexed once and shared across users (user-agnostic point + IDs), so eviction must *release* this user rather than blindly delete: it + drops ``user:`` from the point's ``acl_principals`` and removes the + points only when no reader remains. Legacy points without a principal set + fall back to the original per-user delete. Safe to call when the document is not present — Qdrant returns successfully with zero points affected. Args: doc_id: Document ID (str — keyword-indexed in Qdrant payload) doc_type: Document type (note, file, deck_card, news_item) - user_id: Owner of the points being evicted + user_id: User whose access is being revoked Raises: Exception: If the underlying Qdrant client raises. Callers in the search hot path should catch and log; eviction failures must not block search responses. """ - qdrant_client = await get_qdrant_client() - settings = get_settings() - - await qdrant_client.delete( - collection_name=settings.get_collection_name(), - points_selector=Filter( - must=[ - FieldCondition(key="user_id", match=MatchValue(value=user_id)), - FieldCondition(key="doc_id", match=MatchValue(value=doc_id)), - FieldCondition(key="doc_type", match=MatchValue(value=doc_type)), - ] - ), - ) + await release_document_for_user(doc_id, doc_type, user_id) logger.info( - "Evicted Qdrant points for %s_%s (user=%s); " - "document was inaccessible at verification time", + "Released %s_%s for user=%s; document was inaccessible at verification time", doc_type, doc_id, user_id, diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index d6d7c475..a25d919f 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -11,7 +11,7 @@ from typing import Any, cast import anyio from anyio.abc import TaskStatus from anyio.streams.memory import MemoryObjectReceiveStream -from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct +from qdrant_client.models import PointStruct from nextcloud_mcp_server.acl_hash import compute_acl_hash from nextcloud_mcp_server.client import NextcloudClient @@ -33,6 +33,11 @@ from nextcloud_mcp_server.vector.html_processor import html_to_markdown from nextcloud_mcp_server.vector.placeholder import delete_placeholder_point from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client from nextcloud_mcp_server.vector.scanner import DocumentTask +from nextcloud_mcp_server.vector.sharing_state import ( + claim_existing_index, + existing_principals, + release_document_for_user, +) logger = logging.getLogger(__name__) @@ -192,28 +197,15 @@ async def process_document( ): try: qdrant_client = await get_qdrant_client() - settings = get_settings() # Handle deletion if doc_task.operation == "delete": - await qdrant_client.delete( - collection_name=settings.get_collection_name(), - points_selector=Filter( - must=[ - FieldCondition( - key="user_id", - match=MatchValue(value=doc_task.user_id), - ), - FieldCondition( - key="doc_id", - match=MatchValue(value=doc_task.doc_id), - ), - FieldCondition( - key="doc_type", - match=MatchValue(value=doc_task.doc_type), - ), - ] - ), + # Release this user rather than blind-delete: a file shared across + # users has one user-agnostic point set referenced by multiple + # principals, so the points are removed only once the last reader + # is gone (see vector/sharing_state.release_document_for_user). + await release_document_for_user( + doc_task.doc_id, doc_task.doc_type, doc_task.user_id ) logger.info( "Deleted %s_%s for %s", @@ -479,6 +471,28 @@ async def _index_document( ) file_path = doc_task.file_path + # Cross-worker dedup race-guard: two users' tasks for the same shared + # file can be enqueued before either finishes. If another worker has + # already indexed this exact content (fileid + etag + embedding model) + # in the tenant, claim it for this user (observed-access ACL) and skip + # the expensive fetch/parse/embed entirely. + if doc_task.etag and await claim_existing_index( + doc_task.doc_id, "file", doc_task.etag, doc_task.user_id + ): + await delete_placeholder_point( + doc_id=doc_task.doc_id, + doc_type="file", + user_id=doc_task.user_id, + ) + logger.info( + "Dedup hit for file %s (etag=%s); claimed for user %s " + "without reprocessing", + doc_task.doc_id, + doc_task.etag, + doc_task.user_id, + ) + return + # Read file content via WebDAV content_bytes, content_type = await nc_client.webdav.read_file(file_path) else: @@ -510,7 +524,10 @@ async def _index_document( content = result.text file_metadata = result.metadata title = file_metadata.get("title") or file_path.split("/")[-1] - etag = "" # WebDAV read_file doesn't return etag + # etag comes from the scanner's tag REPORT (threaded via the + # DocumentTask); read_file itself returns no etag. It is the + # tenant-wide content-dedup key, so it must be persisted. + etag = doc_task.etag or "" # Diagnostic: Log page boundary information if available if "page_boundaries" in file_metadata: @@ -763,6 +780,28 @@ async def _index_document( _embedding_identity = settings.get_embedding_model_name() _acl_hash = compute_acl_hash([("user", doc_task.user_id)]) + # Observed-access ACL principals (computed once per document, not per chunk). + # Seed with the indexer (and owner, if distinct). For files — the only type + # with cross-user dedup and globally-unique IDs (Nextcloud fileid) — union in + # any principals already recorded so re-indexing after a content change + # preserves visibility for readers who had previously claimed the file. For + # note/news_item/deck_card, IDs are per-user (not globally unique) and point + # IDs are user-agnostic, so merging another user's principals on an ID + # collision would wrongly cross-surface their content; those types are + # seeded with the indexer only. + _prior_principals = ( + await existing_principals(doc_task.doc_id, doc_task.doc_type) + if doc_task.doc_type == "file" + else [] + ) + _acl_principals = sorted( + set(_prior_principals) + | { + f"user:{doc_task.user_id}", + f"user:{doc_task.owner_id or doc_task.user_id}", + } + ) + # Surface deck card data quality issues at indexing time rather than # only at verification time (where _verify_deck_cards falls through to # legacy-data pass-through when board_id/stack_id are missing). This is @@ -807,6 +846,13 @@ async def _index_document( # crawl shared-with-them content can set owner_id to the # true owner without losing the "who indexed this" trail. "owner_id": doc_task.owner_id or doc_task.user_id, + # Observed-access ACL set: every user whose scanner has seen + # (hence can read) this document. Seeded with the indexer (and + # owner, if distinct); grown lazily as other readers' scanners + # hit the tenant-wide dedup path. Search ORs a + # MatchAny(acl_principals, ["user:"]) branch so a + # deduplicated shared file stays findable by every reader. + "acl_principals": _acl_principals, "doc_id": doc_task.doc_id, "doc_type": doc_task.doc_type, "is_placeholder": False, # Real indexed document (not placeholder) diff --git a/nextcloud_mcp_server/vector/qdrant_client.py b/nextcloud_mcp_server/vector/qdrant_client.py index 7417daa9..60fa994f 100644 --- a/nextcloud_mcp_server/vector/qdrant_client.py +++ b/nextcloud_mcp_server/vector/qdrant_client.py @@ -70,6 +70,19 @@ _PAYLOAD_INDEX_FIELDS: dict[str, PayloadSchemaType] = { # migration like modified_at. Local/embedded qdrant-client matches by # substring without an index, so dev stacks work without it too. "file_path": PayloadSchemaType.TEXT, + # etag is the tenant-wide content-dedup key: the scanner/processor scroll for + # a non-placeholder point matching (doc_id, doc_type, etag) to decide whether + # a file's content is already indexed and reprocessing can be skipped (see + # vector/sharing_state.find_indexed_content). KEYWORD for exact match; the + # value is the Nextcloud etag already written to every point. Idempotent + # startup migration like the fields above. + "etag": PayloadSchemaType.KEYWORD, + # acl_principals is the observed-access ACL set ("user:" entries). Search + # ORs MatchAny(key="acl_principals", any=["user:"]) so a deduplicated + # shared point reaches every reader's candidate set (see + # search/access_filter.build_ownership_filter); KEYWORD indexes match array + # membership element-wise. Idempotent startup migration. + "acl_principals": PayloadSchemaType.KEYWORD, } # Sentinel point that records "this collection has been backfilled to str diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index c690eb4c..29311346 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -31,6 +31,7 @@ from nextcloud_mcp_server.vector.placeholder import ( ) from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client from nextcloud_mcp_server.vector.queue.ports import TaskProducer +from nextcloud_mcp_server.vector.sharing_state import claim_existing_index logger = logging.getLogger(__name__) @@ -360,7 +361,16 @@ async def scan_user_documents( return # Scan tagged PDF files (after notes) - # Get indexed file IDs from Qdrant (for deletion tracking) + # Get indexed file IDs from Qdrant (for deletion tracking). + # NOTE: this is filtered by user_id, so a "pure claimer" — a user who + # gained access to a shared file via the tenant-wide dedup path + # (claim_existing_index) without ever indexing it themselves — is NOT in + # this set (the points carry the first indexer's user_id, only the + # claimer's user: in acl_principals). Such a user is therefore never + # enqueued for deletion by the grace-period sweep below; their stale + # acl_principals entry is reclaimed lazily by verify-on-read eviction + # (release_document_for_user) when a search surfaces a now-inaccessible + # result. A future scanner-side cleanup could scroll acl_principals too. indexed_file_ids = set() if not initial_sync: assert qdrant_client is not None # narrow for the type checker @@ -448,6 +458,24 @@ async def scan_user_documents( except (ValueError, KeyError): pass + # Tenant-wide content dedup (Layer 1 / observed-access ACL): if + # this exact file content (fileid + etag) is already indexed under + # the current embedding model by ANY user in the tenant, skip + # re-parsing/re-embedding and just record that this user can read + # it. Eliminates the per-user reprocessing ping-pong that arises + # because chunk point IDs are user-agnostic (note 386945 #5). + etag = str(file_info.get("etag") or "") + if etag and await claim_existing_index(file_id, "file", etag, user_id): + _potentially_deleted.pop((user_id, file_id), None) + logger.debug( + "Dedup: file %s (ID: %s) already indexed in tenant; " + "granted access to %s without reprocessing", + file_path, + file_id, + user_id, + ) + continue + if initial_sync: # Send everything on first sync - write placeholder first await write_placeholder_point( @@ -455,6 +483,7 @@ async def scan_user_documents( doc_type="file", user_id=user_id, modified_at=modified_at, + etag=etag, file_path=file_path, ) await send_stream.send( @@ -465,6 +494,7 @@ async def scan_user_documents( operation="index", modified_at=modified_at, file_path=file_path, + etag=etag, ) ) file_queued += 1 @@ -525,6 +555,7 @@ async def scan_user_documents( doc_type="file", user_id=user_id, modified_at=modified_at, + etag=etag, file_path=file_path, ) await send_stream.send( @@ -535,6 +566,7 @@ async def scan_user_documents( operation="index", modified_at=modified_at, file_path=file_path, + etag=etag, ) ) file_queued += 1 diff --git a/nextcloud_mcp_server/vector/sharing_state.py b/nextcloud_mcp_server/vector/sharing_state.py new file mode 100644 index 00000000..63e7b433 --- /dev/null +++ b/nextcloud_mcp_server/vector/sharing_state.py @@ -0,0 +1,286 @@ +"""Tenant-wide content dedup + observed-access ACL state for the vector index. + +A file shared across users (directly, or via a group folder shared to a group) +has one Nextcloud ``fileid`` and one ``etag`` for everyone, and chunk point IDs +are user-agnostic (``uuid5(tenant_id, doc_id, chunk_index)`` — see +``vector/payload_keys.py``). So two users indexing the same file produce the +*same* points. The per-user freshness gate (filtered by ``user_id``) nonetheless +made them re-parse + re-embed the identical content on every scan (note 386945, +finding #5). This module lets the pipeline detect "already indexed by someone in +this tenant" and skip the expensive work. + +Visibility is handled by an *observed-access* model rather than push-enumeration +of share/group-folder grants (which the server cannot read without admin creds — +group membership and the GroupFolders API are admin-only, and WebDAV PROPFIND +carries no ACL). The per-user scanner crawl is itself the access oracle: a tagged +file appears in a user's ``find_files_by_tag`` REPORT **iff** that user can read +it. So each point carries ``acl_principals`` — the set of ``user:`` whose +scanner has observed (hence can access) the file. The search filter ORs a +``MatchAny(acl_principals, ["user:"])`` branch, and ``_verify_files`` (the +verify-on-read gate) re-checks each result against the user's tagged REPORT, so +an over-broad principal match can never leak content. + +All point IDs are user-agnostic, so deletion must *release one user* (drop their +principal) and only remove the points when the principal set empties — otherwise +one user untagging a shared file would evict it for everyone still reading it. +""" + +from __future__ import annotations + +import logging + +from qdrant_client.models import FieldCondition, Filter, MatchValue + +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.vector import payload_keys +from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter +from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client + +logger = logging.getLogger(__name__) + +ACL_PRINCIPALS_KEY = "acl_principals" + + +def user_principal(user_id: str) -> str: + """The ``acl_principals`` entry representing a single user's read access.""" + return f"user:{user_id}" + + +def _document_filter(doc_id: str, doc_type: str, *, real_only: bool) -> Filter: + """Match every chunk of one document; optionally exclude placeholder points.""" + must: list = [ + FieldCondition(key="doc_id", match=MatchValue(value=doc_id)), + FieldCondition(key="doc_type", match=MatchValue(value=doc_type)), + ] + if real_only: + must.append(get_placeholder_filter()) + return Filter(must=must) + + +async def find_indexed_content( + doc_id: str, + doc_type: str, + etag: str, + embedding_identity: str, +) -> dict | None: + """Return a real point's payload if this exact content is already indexed. + + Looks tenant-wide (no ``user_id`` filter) for a non-placeholder point with + the given ``doc_id``/``doc_type``/``etag``. The match is gated on + ``embedding_identity`` in Python (not the Qdrant filter, to avoid requiring an + index on that field): since point IDs are model-agnostic, a model switch + overwrites the same points, so all live points for a doc share one identity — + a mismatch means the existing vectors were produced by a different model and + must be re-embedded, so we report "not indexed". + + Returns the payload dict (including ``acl_principals``) on a hit, else None. + """ + if not etag: + return None + qdrant_client = await get_qdrant_client() + settings = get_settings() + points, _ = await qdrant_client.scroll( + collection_name=settings.get_collection_name(), + scroll_filter=Filter( + must=[ + FieldCondition(key="doc_id", match=MatchValue(value=doc_id)), + FieldCondition(key="doc_type", match=MatchValue(value=doc_type)), + FieldCondition(key="etag", match=MatchValue(value=etag)), + get_placeholder_filter(), + ] + ), + limit=1, + with_payload=True, + with_vectors=False, + ) + if not points: + return None + payload = dict(points[0].payload or {}) + if payload.get(payload_keys.EMBEDDING_IDENTITY) != embedding_identity: + # Existing vectors were produced by a different embedding model — a + # re-embed is required, so this content is not reusable as-is. + return None + return payload + + +async def existing_principals(doc_id: str, doc_type: str) -> list[str]: + """Return the ``acl_principals`` already recorded for a document (or []). + + Used when re-indexing after a content change (etag differs, so the dedup + race-guard misses and the points are overwritten): seeding the new points + with the prior principal set preserves visibility for readers who had + already claimed the file, instead of resetting it to just the indexer. + """ + qdrant_client = await get_qdrant_client() + settings = get_settings() + points, _ = await qdrant_client.scroll( + collection_name=settings.get_collection_name(), + scroll_filter=_document_filter(doc_id, doc_type, real_only=True), + limit=1, + with_payload=True, + with_vectors=False, + ) + if not points: + return [] + return list(dict(points[0].payload or {}).get(ACL_PRINCIPALS_KEY) or []) + + +async def add_principal( + doc_id: str, + doc_type: str, + user_id: str, + current_principals: list[str] | None, +) -> bool: + """Record that ``user_id`` can read this document (observed-access ACL). + + No-op (returns False) when the user's principal is already present — so the + steady state of a repeat scan writes nothing. Otherwise unions the principal + onto every real chunk of the document via a single ``set_payload`` and + returns True. Concurrent adds race to a last-writer-wins union; a dropped + add is re-applied on the losing user's next scan, and verify-on-read gates + correctness in the meantime. + """ + principal = user_principal(user_id) + existing = current_principals or [] + if principal in existing: + return False + new_principals = sorted(set(existing) | {principal}) + qdrant_client = await get_qdrant_client() + settings = get_settings() + await qdrant_client.set_payload( + collection_name=settings.get_collection_name(), + payload={ACL_PRINCIPALS_KEY: new_principals}, + points=_document_filter(doc_id, doc_type, real_only=True), + wait=True, + ) + logger.debug( + "Granted read principal %s on %s_%s (now %d principal(s))", + principal, + doc_type, + doc_id, + len(new_principals), + ) + return True + + +async def claim_existing_index( + doc_id: str, + doc_type: str, + etag: str, + user_id: str, +) -> bool: + """Tenant-wide dedup claim: skip reprocessing if content is already indexed. + + Returns True when a non-placeholder point for this exact content (fileid + + etag + current embedding model) already exists for some user in the tenant — + in which case ``user_id`` is added to ``acl_principals`` (so the file remains + searchable for them) and the caller should skip fetch/parse/embed. Returns + False when nothing reusable exists and the document must be processed. + + Fail-safe: a Qdrant error during the lookup degrades to False (process the + document normally) rather than aborting the scan — the dedup is an + optimisation, never a correctness gate. A failure to record the principal + after a confirmed hit is non-fatal (logged, not raised): verify-on-read still + gates access and the user's next scan re-claims it. + """ + embedding_identity = get_settings().get_embedding_model_name() + try: + existing = await find_indexed_content( + doc_id, doc_type, etag, embedding_identity + ) + except Exception as exc: # noqa: BLE001 — degrade to "process normally" + logger.warning( + "Dedup lookup failed for %s_%s (%s); processing without dedup", + doc_type, + doc_id, + exc, + ) + return False + if existing is None: + return False + try: + await add_principal(doc_id, doc_type, user_id, existing.get(ACL_PRINCIPALS_KEY)) + except Exception as exc: # noqa: BLE001 — non-fatal; recovered on next scan + logger.warning( + "Failed to grant read principal user:%s on %s_%s (%s); " + "verify-on-read and the next scan will reconcile", + user_id, + doc_type, + doc_id, + exc, + ) + return True + + +async def release_document_for_user( + doc_id: str, + doc_type: str, + user_id: str, +) -> None: + """Drop ``user_id``'s access to a document; delete points only when orphaned. + + Replaces a blind per-document delete. Because point IDs are user-agnostic, a + shared document has one point set referenced by multiple principals; removing + one user must not evict it for the others. Removes the user's principal and + deletes the points only once no principal remains. + + Legacy points written before ``acl_principals`` existed have no principal + set; for those we preserve the original behaviour (delete by + ``user_id``/``doc_id``/``doc_type``) so a single-owner delete still works. + """ + qdrant_client = await get_qdrant_client() + settings = get_settings() + collection = settings.get_collection_name() + + points, _ = await qdrant_client.scroll( + collection_name=collection, + scroll_filter=_document_filter(doc_id, doc_type, real_only=True), + limit=1, + with_payload=True, + with_vectors=False, + ) + principals = ( + (dict(points[0].payload or {}).get(ACL_PRINCIPALS_KEY)) if points else None + ) + + if not principals: + # No real points, or legacy points without a principal set: fall back to + # the original per-user delete (also clears this user's placeholder). + await qdrant_client.delete( + collection_name=collection, + points_selector=Filter( + must=[ + FieldCondition(key="user_id", match=MatchValue(value=user_id)), + FieldCondition(key="doc_id", match=MatchValue(value=doc_id)), + FieldCondition(key="doc_type", match=MatchValue(value=doc_type)), + ] + ), + ) + return + + remaining = sorted(p for p in principals if p != user_principal(user_id)) + if not remaining: + # Last reader released — remove every point (real + placeholder). + await qdrant_client.delete( + collection_name=collection, + points_selector=_document_filter(doc_id, doc_type, real_only=False), + ) + logger.info( + "Released last principal for %s_%s — document removed from index", + doc_type, + doc_id, + ) + else: + await qdrant_client.set_payload( + collection_name=collection, + payload={ACL_PRINCIPALS_KEY: remaining}, + points=_document_filter(doc_id, doc_type, real_only=True), + wait=True, + ) + logger.info( + "Released principal %s for %s_%s — %d reader(s) remain", + user_principal(user_id), + doc_type, + doc_id, + len(remaining), + ) diff --git a/tests/unit/search/test_access_filter.py b/tests/unit/search/test_access_filter.py index 830501c3..58ad663a 100644 --- a/tests/unit/search/test_access_filter.py +++ b/tests/unit/search/test_access_filter.py @@ -2,6 +2,7 @@ from __future__ import annotations +from typing import Any from unittest.mock import AsyncMock import pytest @@ -157,26 +158,32 @@ class TestOwnersCacheBehavior: class TestBuildOwnershipFilter: + @staticmethod + def _by_key(flt: Filter) -> dict[str, Any]: + assert flt.should is not None + return {cond.key: cond for cond in flt.should} + def test_defaults_to_self_only_when_owners_omitted(self) -> None: flt = build_ownership_filter("alice") - # Self-only: just the user_id branch. Self is NOT duplicated into an - # owner_id branch (the user_id branch already covers self-owned content). - assert flt.should is not None - assert len(flt.should) == 1 - (user_branch,) = flt.should - assert user_branch.key == "user_id" - assert user_branch.match.value == "alice" + # Self-only: the user_id branch plus the observed-access acl_principals + # branch (so a deduplicated shared file the user has claimed is still + # findable). No owner_id branch — self is covered by user_id. + branches = self._by_key(flt) + assert set(branches) == {"user_id", "acl_principals"} + assert branches["user_id"].match.value == "alice" + assert branches["acl_principals"].match.any == ["user:alice"] def test_expands_owner_branch_with_accessible_owners(self) -> None: flt = build_ownership_filter("alice", ["alice", "bob", "carol"]) - owner_branch, user_branch = flt.should + branches = self._by_key(flt) + assert set(branches) == {"owner_id", "user_id", "acl_principals"} # Owner branch holds only the OTHER owners — self ("alice") is excluded # because the user_id branch already matches self-owned content. - assert set(owner_branch.match.any) == {"bob", "carol"} - assert user_branch.key == "user_id" - assert user_branch.match.value == "alice" + assert set(branches["owner_id"].match.any) == {"bob", "carol"} + assert branches["user_id"].match.value == "alice" + assert branches["acl_principals"].match.any == ["user:alice"] def test_explicit_empty_list_omits_owner_branch_keeps_legacy(self) -> None: # Edge case: caller passed an explicit empty list. The owner_id branch @@ -185,11 +192,9 @@ class TestBuildOwnershipFilter: # user still finds their own content from before the migration. flt = build_ownership_filter("alice", []) - assert flt.should is not None - assert len(flt.should) == 1 - (user_branch,) = flt.should - assert user_branch.key == "user_id" - assert user_branch.match.value == "alice" + branches = self._by_key(flt) + assert set(branches) == {"user_id", "acl_principals"} + assert branches["user_id"].match.value == "alice" class TestBuildBaseFilterConditions: diff --git a/tests/unit/test_document_parse_metrics.py b/tests/unit/test_document_parse_metrics.py index be0cbf61..e00f9745 100644 --- a/tests/unit/test_document_parse_metrics.py +++ b/tests/unit/test_document_parse_metrics.py @@ -312,8 +312,12 @@ class TestProcessDocumentMetricCounting: ) qmock = MagicMock() - qmock.delete = AsyncMock() - with patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)): + # Deletion now delegates to release_document_for_user (release-one-user + # semantics); stub it so the test exercises only the metric accounting. + with ( + patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)), + patch.object(proc, "release_document_for_user", new=AsyncMock()), + ): await proc.process_document(task, MagicMock()) assert metric_sample( @@ -339,8 +343,16 @@ class TestProcessDocumentMetricCounting: ) qmock = MagicMock() - qmock.delete = AsyncMock(side_effect=RuntimeError("boom")) - with patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)): + # A failed release still counts as a processed (error) delete and must + # not touch the indexed counter. + with ( + patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)), + patch.object( + proc, + "release_document_for_user", + new=AsyncMock(side_effect=RuntimeError("boom")), + ), + ): with pytest.raises(RuntimeError): await proc.process_document(task, MagicMock()) diff --git a/tests/unit/vector/test_sharing_state.py b/tests/unit/vector/test_sharing_state.py new file mode 100644 index 00000000..13a67a99 --- /dev/null +++ b/tests/unit/vector/test_sharing_state.py @@ -0,0 +1,229 @@ +"""Unit tests for tenant-wide content dedup + observed-access ACL state. + +Covers vector/sharing_state.py: the tenant-wide content lookup that lets a +shared/group-folder file be parsed+embedded once per tenant instead of once per +user, and the ``acl_principals`` maintenance (grant/release) that keeps a +deduplicated point findable by every reader without re-indexing. + +All functions reach Qdrant via ``get_qdrant_client`` and resolve the collection +via ``get_settings``; both are monkeypatched here so the logic is exercised +without a live Qdrant. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from nextcloud_mcp_server.vector import payload_keys +from nextcloud_mcp_server.vector import sharing_state as ss + +pytestmark = pytest.mark.unit + +_COLLECTION = "test_collection" +_MODEL = "model-x" + + +class _Settings: + def get_collection_name(self) -> str: + return _COLLECTION + + def get_embedding_model_name(self) -> str: + return _MODEL + + +def _point(payload: dict) -> SimpleNamespace: + """Stand-in for a qdrant_client Record (only id/payload are read).""" + return SimpleNamespace(id="pt", payload=payload) + + +@pytest.fixture +def client(monkeypatch) -> AsyncMock: + """An AsyncMock Qdrant client wired into sharing_state, with a stub Settings. + + ``scroll`` defaults to "no points"; individual tests override + ``client.scroll.return_value``/``side_effect``. + """ + qc = AsyncMock() + qc.scroll.return_value = ([], None) + monkeypatch.setattr(ss, "get_qdrant_client", AsyncMock(return_value=qc)) + monkeypatch.setattr(ss, "get_settings", lambda: _Settings()) + return qc + + +def _must_keys(flt) -> list[str | None]: + """Collect the FieldCondition keys in a Filter's ``must`` clause.""" + return [getattr(c, "key", None) for c in (flt.must or [])] + + +class TestFindIndexedContent: + async def test_returns_payload_on_etag_and_model_match(self, client) -> None: + payload = { + "doc_id": "42", + "etag": "abc", + payload_keys.EMBEDDING_IDENTITY: _MODEL, + ss.ACL_PRINCIPALS_KEY: ["user:alice"], + } + client.scroll.return_value = ([_point(payload)], None) + + result = await ss.find_indexed_content("42", "file", "abc", _MODEL) + assert result == payload + + async def test_none_when_no_points(self, client) -> None: + client.scroll.return_value = ([], None) + assert await ss.find_indexed_content("42", "file", "abc", _MODEL) is None + + async def test_none_on_embedding_model_mismatch(self, client) -> None: + # A model switch overwrites the same point IDs; existing vectors made by + # a different model must be re-embedded, so this reports "not indexed". + client.scroll.return_value = ( + [_point({payload_keys.EMBEDDING_IDENTITY: "other-model"})], + None, + ) + assert await ss.find_indexed_content("42", "file", "abc", _MODEL) is None + + async def test_empty_etag_short_circuits_without_query(self, client) -> None: + assert await ss.find_indexed_content("42", "file", "", _MODEL) is None + client.scroll.assert_not_called() + + +class TestAddPrincipal: + async def test_noop_when_principal_already_present(self, client) -> None: + added = await ss.add_principal("42", "file", "alice", ["user:alice"]) + assert added is False + client.set_payload.assert_not_called() + + async def test_unions_principal_when_absent(self, client) -> None: + added = await ss.add_principal("42", "file", "bob", ["user:alice"]) + assert added is True + client.set_payload.assert_awaited_once() + kwargs = client.set_payload.await_args.kwargs + assert kwargs["payload"][ss.ACL_PRINCIPALS_KEY] == ["user:alice", "user:bob"] + # Updates only real (non-placeholder) chunks of this document. + assert _must_keys(kwargs["points"]) == ["doc_id", "doc_type", "is_placeholder"] + + async def test_handles_none_current_principals(self, client) -> None: + added = await ss.add_principal("42", "file", "alice", None) + assert added is True + kwargs = client.set_payload.await_args.kwargs + assert kwargs["payload"][ss.ACL_PRINCIPALS_KEY] == ["user:alice"] + + +class TestClaimExistingIndex: + async def test_true_and_grants_principal_on_hit(self, client) -> None: + client.scroll.return_value = ( + [ + _point( + { + payload_keys.EMBEDDING_IDENTITY: _MODEL, + ss.ACL_PRINCIPALS_KEY: ["user:alice"], + } + ) + ], + None, + ) + claimed = await ss.claim_existing_index("42", "file", "abc", "bob") + assert claimed is True + # bob was added to the existing point's principals. + client.set_payload.assert_awaited_once() + assert client.set_payload.await_args.kwargs["payload"][ + ss.ACL_PRINCIPALS_KEY + ] == ["user:alice", "user:bob"] + + async def test_false_when_not_indexed(self, client) -> None: + client.scroll.return_value = ([], None) + assert await ss.claim_existing_index("42", "file", "abc", "bob") is False + client.set_payload.assert_not_called() + + async def test_hit_for_already_listed_user_writes_nothing(self, client) -> None: + client.scroll.return_value = ( + [ + _point( + { + payload_keys.EMBEDDING_IDENTITY: _MODEL, + ss.ACL_PRINCIPALS_KEY: ["user:alice"], + } + ) + ], + None, + ) + # alice already present -> claim still True (skip reprocess) but no write. + assert await ss.claim_existing_index("42", "file", "abc", "alice") is True + client.set_payload.assert_not_called() + + async def test_lookup_error_degrades_to_process_normally(self, client) -> None: + # A Qdrant hiccup during dedup must not abort the scan — fall back to + # processing the document (return False), not raise. + client.scroll.side_effect = RuntimeError("qdrant down") + assert await ss.claim_existing_index("42", "file", "abc", "bob") is False + + async def test_principal_grant_failure_after_hit_is_non_fatal(self, client) -> None: + # The content IS indexed (skip reprocess), so a failure to record the + # principal still returns True; verify-on-read + next scan reconcile. + client.scroll.return_value = ( + [_point({payload_keys.EMBEDDING_IDENTITY: _MODEL})], + None, + ) + client.set_payload.side_effect = RuntimeError("set_payload failed") + assert await ss.claim_existing_index("42", "file", "abc", "bob") is True + + +class TestExistingPrincipals: + async def test_returns_recorded_principals(self, client) -> None: + client.scroll.return_value = ( + [_point({ss.ACL_PRINCIPALS_KEY: ["user:alice", "user:bob"]})], + None, + ) + assert await ss.existing_principals("42", "file") == ["user:alice", "user:bob"] + + async def test_empty_when_no_points(self, client) -> None: + client.scroll.return_value = ([], None) + assert await ss.existing_principals("42", "file") == [] + + +class TestReleaseDocumentForUser: + async def test_keeps_points_and_trims_principals_when_readers_remain( + self, client + ) -> None: + client.scroll.return_value = ( + [_point({ss.ACL_PRINCIPALS_KEY: ["user:alice", "user:bob"]})], + None, + ) + await ss.release_document_for_user("42", "file", "alice") + + client.delete.assert_not_called() + kwargs = client.set_payload.await_args.kwargs + assert kwargs["payload"][ss.ACL_PRINCIPALS_KEY] == ["user:bob"] + + async def test_deletes_all_points_when_last_reader_released(self, client) -> None: + client.scroll.return_value = ( + [_point({ss.ACL_PRINCIPALS_KEY: ["user:alice"]})], + None, + ) + await ss.release_document_for_user("42", "file", "alice") + + client.set_payload.assert_not_called() + client.delete.assert_awaited_once() + selector = client.delete.await_args.kwargs["points_selector"] + # Whole document removed: doc_id + doc_type, no user_id, no placeholder gate. + assert _must_keys(selector) == ["doc_id", "doc_type"] + + async def test_legacy_points_without_principals_delete_by_user( + self, client + ) -> None: + # Pre-acl_principals points: preserve the original per-user delete. + client.scroll.return_value = ([_point({"doc_id": "42"})], None) + await ss.release_document_for_user("42", "file", "alice") + + client.set_payload.assert_not_called() + selector = client.delete.await_args.kwargs["points_selector"] + assert _must_keys(selector) == ["user_id", "doc_id", "doc_type"] + + async def test_no_points_falls_back_to_per_user_delete(self, client) -> None: + client.scroll.return_value = ([], None) + await ss.release_document_for_user("42", "file", "alice") + + selector = client.delete.await_args.kwargs["points_selector"] + assert _must_keys(selector) == ["user_id", "doc_id", "doc_type"]