refactor: Optimize Nextcloud access verification with centralized filtering

Move access verification from individual search algorithms to final output
stage, eliminating redundant API calls and improving performance.

## Changes

**New:**
- `search/verification.py`: Centralized verification using anyio task groups
  - Deduplicates results by (doc_id, doc_type) before verification
  - Verifies all unique documents in parallel using structured concurrency
  - Filters out inaccessible documents in single pass

**Modified Search Algorithms:**
- `search/semantic.py`: Removed _deduplicate_and_verify() and _verify_document_access()
- `search/keyword.py`: Removed _verify_access() and parallel verification
- `search/fuzzy.py`: Removed _verify_access() and parallel verification
- `search/hybrid.py`: Removed nextcloud_client parameter passing

All algorithms now return unverified results from Qdrant payload.

**Modified Output Stages:**
- `server/semantic.py`: Added verify_search_results() call after search
- `auth/viz_routes.py`: Added verify_search_results() call after search

Both endpoints now verify access once at final stage with deduplication.

## Performance Impact

**Before:**
- Hybrid mode (limit=10): 30 API calls (10 per algorithm × 3 algorithms)
- Single algorithm: 10-20 API calls (with verification buffer)

**After:**
- Hybrid mode (limit=10): 10 API calls (deduplicated verification)
- Single algorithm: 10 API calls (deduplicated verification)

**Performance Gain:** 3x reduction in API calls for hybrid search

## Architecture Benefits

- **Separation of concerns**: Algorithms handle scoring, output stage handles security
- **Deduplication**: Each document verified exactly once
- **Parallel execution**: All verifications run concurrently via anyio task groups
- **Consistency**: Same verification logic across MCP tools and viz endpoints

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2025-11-15 06:21:06 +01:00
co-authored by Claude
parent ed0825e661
commit 42376483ab
7 changed files with 224 additions and 406 deletions
+122
View File
@@ -0,0 +1,122 @@
"""Access verification for search results.
This module provides centralized verification of Nextcloud access permissions
for search results. Verification happens at the final output stage (MCP tool/viz endpoint)
rather than within individual search algorithms, preventing redundant API calls.
Key benefits:
- Deduplication: Each document verified exactly once (even in hybrid mode)
- Parallel execution: All verifications run concurrently via anyio task groups
- Separation of concerns: Algorithms handle scoring, this module handles security
"""
import logging
from dataclasses import replace
from typing import Protocol
import anyio
from nextcloud_mcp_server.search.algorithms import SearchResult
logger = logging.getLogger(__name__)
class NextcloudClientProtocol(Protocol):
"""Protocol for Nextcloud client with app-specific access."""
@property
def notes(self):
"""Notes client for accessing notes API."""
...
async def verify_search_results(
results: list[SearchResult],
nextcloud_client: NextcloudClientProtocol,
) -> list[SearchResult]:
"""
Verify Nextcloud access for search results.
Deduplicates by (doc_id, doc_type), verifies in parallel using anyio task groups,
and filters out inaccessible documents. Maintains original result ordering.
Args:
results: Unverified search results from Qdrant
nextcloud_client: Nextcloud client for access checks
Returns:
Verified and accessible results (same order as input)
Example:
>>> unverified = await search_algo.search(query="test", limit=10)
>>> verified = await verify_search_results(unverified, client)
>>> # verified contains only documents user can access
"""
# Deduplicate by (doc_id, doc_type) while preserving order
# This is critical for hybrid search where same doc may appear in multiple algorithm results
seen = set()
unique_results = []
for result in results:
key = (result.id, result.doc_type)
if key not in seen:
seen.add(key)
unique_results.append(result)
if not unique_results:
return []
logger.debug(
f"Verifying access for {len(unique_results)} unique documents "
f"(from {len(results)} total results)"
)
# Verify all unique documents in parallel using anyio task group
# Use list to maintain order (index-based storage)
verified_results = [None] * len(unique_results)
async def verify_one(index: int, result: SearchResult):
"""
Verify a single document and store result at index.
Args:
index: Position in verified_results list
result: Search result to verify
"""
try:
if result.doc_type == "note":
# Fetch note to verify access and get fresh metadata
note = await nextcloud_client.notes.get_note(result.id)
# Update metadata with fresh data from Nextcloud
updated_metadata = {**(result.metadata or {}), **note}
verified_results[index] = replace(result, metadata=updated_metadata)
# TODO: Add verification for other doc types (calendar, deck, file, etc.)
else:
# For now, assume other types are accessible
# In production, add proper verification for each type
logger.debug(
f"No verification implemented for doc_type={result.doc_type}, "
"assuming accessible"
)
verified_results[index] = result
except Exception as e:
# Document is inaccessible (403, 404, or other error)
# Log at debug level since this is expected for filtered results
logger.debug(f"Document {result.doc_type}/{result.id} not accessible: {e}")
verified_results[index] = None
# Run all verifications in parallel using anyio task group
# This provides structured concurrency with automatic cancellation on errors
async with anyio.create_task_group() as tg:
for idx, result in enumerate(unique_results):
tg.start_soon(verify_one, idx, result)
# Filter out None (inaccessible) and return verified results
accessible = [r for r in verified_results if r is not None]
logger.debug(
f"Verification complete: {len(accessible)} accessible, "
f"{len(unique_results) - len(accessible)} filtered out"
)
return accessible