From 1322e5aba0d3a30a55938ec77baa1bfe5c8255ba Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 5 Jun 2026 16:20:05 +0200 Subject: [PATCH] feat(vector): index files in real time on vector-index tag changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tagging an existing file/folder emits only OCP\SystemTag\MapperEvent — never a Node*Event — so tagged PDFs were previously only picked up by the hourly scanner. Subscribe to the tag event and reconcile membership so adding/removing the `vector-index` tag (re)indexes in near-real time. - webhook_presets: add OCP\SystemTag\MapperEvent to the files_sync preset (NC 32+, where MapperEvent gained getWebhookSerializable(); harmless on older servers — it just never fires). - webhook_parser: parse MapperEvent (objectType=files) into a path-less file "reconcile" task. The payload carries only a fileid + tagIds (no name/path), so assign and unassign both collapse to a reconcile. - processor._reconcile_tag_event: resolve the fileid against the user's current vector-index PDFs (find_files_by_tag). Present -> index with the resolved path/etag; absent -> flip to delete. Naturally handles "an unrelated tag changed" and a tagged folder's own fileid (no-op; the scanner still expands folders to descendants). - Unit tests for the parser branch and the reconcile. The matching admin-UI preset change ships separately in the astrolabe app repo. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../server/webhook_presets.py | 16 +++- nextcloud_mcp_server/vector/processor.py | 64 +++++++++++++ nextcloud_mcp_server/vector/webhook_parser.py | 47 +++++++++- tests/unit/test_webhook_parser.py | 91 +++++++++++++++++++ tests/unit/test_webhook_presets.py | 12 +++ tests/unit/vector/test_tag_reconcile.py | 87 ++++++++++++++++++ 6 files changed, 313 insertions(+), 4 deletions(-) create mode 100644 tests/unit/vector/test_tag_reconcile.py diff --git a/nextcloud_mcp_server/server/webhook_presets.py b/nextcloud_mcp_server/server/webhook_presets.py index 2b435a86..08a04ae5 100644 --- a/nextcloud_mcp_server/server/webhook_presets.py +++ b/nextcloud_mcp_server/server/webhook_presets.py @@ -30,6 +30,13 @@ FILE_EVENT_WRITTEN = "OCP\\Files\\Events\\Node\\NodeWrittenEvent" # See: https://github.com/nextcloud/server/issues/56371 FILE_EVENT_DELETED = "OCP\\Files\\Events\\Node\\BeforeNodeDeletedEvent" +# System-tag assign/unassign (Nextcloud 32+). A single event class covers both +# directions; the payload's ``eventType`` distinguishes them. Lets adding/removing +# the ``vector-index`` tag trigger near-real-time (re)indexing instead of waiting +# for the hourly scan. Requires NC >= 32 — MapperEvent gained +# getWebhookSerializable() in 32.0.0; on older servers the event isn't delivered. +SYSTEMTAG_EVENT_MAPPER = "OCP\\SystemTag\\MapperEvent" + # Calendar webhook events CALENDAR_EVENT_CREATED = "OCP\\Calendar\\Events\\CalendarObjectCreatedEvent" CALENDAR_EVENT_UPDATED = "OCP\\Calendar\\Events\\CalendarObjectUpdatedEvent" @@ -128,7 +135,7 @@ WEBHOOK_PRESETS: Dict[str, WebhookPreset] = { }, "files_sync": { "name": "All Files Sync", - "description": "Real-time synchronization for all file operations (create, update, delete)", + "description": "Real-time synchronization for all file operations (create, update, delete) and tag changes (Nextcloud 32+)", "app": "files", "events": [ { @@ -143,6 +150,13 @@ WEBHOOK_PRESETS: Dict[str, WebhookPreset] = { "event": FILE_EVENT_DELETED, "filter": {}, }, + # Tag assign/unassign. Drives vector-index (re)indexing when a file + # or folder is tagged/untagged. Delivered only on NC >= 32; harmless + # to register on older servers (the event simply never fires). + { + "event": SYSTEMTAG_EVENT_MAPPER, + "filter": {}, + }, ], }, "deck_sync": { diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index a0669a71..490f129d 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -165,6 +165,58 @@ async def processor_task( logger.info("Processor %s stopped", worker_id) +async def _reconcile_tag_event( + doc_task: DocumentTask, nc_client: NextcloudClient +) -> None: + """Resolve a tag-webhook file task into a concrete index or delete. + + A SystemTag ``MapperEvent`` only tells us a fileid's tags changed — not the + path, nor whether our ``vector-index`` tag is (still) on it. Look up the + user's current ``vector-index`` PDFs (the same call the scanner uses, which + also expands tagged folders into their PDF descendants) and reconcile the + task in place: + + - fileid present -> index it; fill path/etag/mtime from the tag listing. + - fileid absent -> it isn't a tagged PDF (anymore); flip ``operation`` to + ``delete`` so any existing points are released for this user. + + A tagged *folder*'s own fileid won't appear in the file-level listing, so it + resolves to a harmless no-op delete here; the hourly scanner still expands + tagged folders into their descendants. + """ + tag_name = get_settings().vector_sync_pdf_tag + tagged = await nc_client.find_files_by_tag( + tag_name, mime_type_filter="application/pdf" + ) + match = next( + (f for f in tagged if str(f.get("id")) == str(doc_task.doc_id)), + None, + ) + + if match is None: + doc_task.operation = "delete" + logger.info( + "Tag reconcile: file %s is not a %r PDF; releasing for %s", + doc_task.doc_id, + tag_name, + doc_task.user_id, + ) + return + + doc_task.file_path = match["path"] + if not doc_task.etag: + doc_task.etag = match.get("etag") + last_modified = match.get("last_modified_timestamp") + if last_modified: + doc_task.modified_at = int(last_modified) + logger.info( + "Tag reconcile: indexing %s (file %s) for %s", + doc_task.file_path, + doc_task.doc_id, + doc_task.user_id, + ) + + async def process_document( doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3 ): @@ -204,6 +256,18 @@ async def process_document( try: qdrant_client = await get_qdrant_client() + # Tag-webhook reconcile: a SystemTag MapperEvent enqueues a file task + # carrying only a fileid (file_path is None — see + # webhook_parser._parse_tag_event). Resolve the file's current + # vector-index membership into a concrete index (path/etag filled) or + # a delete before dispatching below. + if ( + doc_task.doc_type == "file" + and doc_task.operation == "index" + and doc_task.file_path is None + ): + await _reconcile_tag_event(doc_task, nc_client) + # Handle deletion if doc_task.operation == "delete": # Release this user rather than blind-delete: a file shared across diff --git a/nextcloud_mcp_server/vector/webhook_parser.py b/nextcloud_mcp_server/vector/webhook_parser.py index 40f079a8..b0f67f94 100644 --- a/nextcloud_mcp_server/vector/webhook_parser.py +++ b/nextcloud_mcp_server/vector/webhook_parser.py @@ -4,9 +4,10 @@ Maps Nextcloud webhook events to vector-sync DocumentTasks. The handler at ``/webhooks/nextcloud`` calls :func:`extract_document_task` and forwards any non-None result to the same processor send-stream the scanner uses. -Currently scoped to file (note) events and Deck card events. Calendar / -Tables events fall through to ``None`` for now; those parsers can be added -in follow-up changes. +Currently scoped to file (note) events, Deck card events, and SystemTag +assign/unassign events (which drive ``vector-index`` (re)indexing of files). +Calendar / Tables events fall through to ``None`` for now; those parsers can +be added in follow-up changes. See ADR-010 for the design and ``webhook-testing-findings.md`` for real captured payloads. @@ -28,6 +29,11 @@ _DECK_EVENT_CARD_UPDATED = "OCA\\Deck\\Event\\CardUpdatedEvent" _DECK_EVENT_CARD_DELETED = "OCA\\Deck\\Event\\CardDeletedEvent" _DECK_EVENT_BOARD_UPDATED = "OCA\\Deck\\Event\\BoardUpdatedEvent" +# System-tag assign/unassign. A single event class covers both directions; the +# payload's ``eventType`` distinguishes them. Webhook-deliverable on NC 32+ +# (MapperEvent gained ``getWebhookSerializable()`` in 32.0.0). +_SYSTEMTAG_EVENT_MAPPER = "OCP\\SystemTag\\MapperEvent" + _DECK_CARD_EVENTS = frozenset( { _DECK_EVENT_CARD_CREATED, @@ -67,6 +73,9 @@ def extract_document_task(payload: dict) -> DocumentTask | None: if event_class in _DECK_CARD_EVENTS or event_class == _DECK_EVENT_BOARD_UPDATED: return _parse_deck_event(event_class, event, user_id, time) + if event_class == _SYSTEMTAG_EVENT_MAPPER: + return _parse_tag_event(event, user_id, time) + logger.debug("Ignoring webhook for unsupported event: %s", event_class) return None @@ -104,6 +113,38 @@ def _parse_file_event( ) +def _parse_tag_event(event: dict, user_id: str, time: int) -> DocumentTask | None: + """Convert a SystemTag ``MapperEvent`` (assign/unassign) into a reconcile task. + + The NC 32+ payload (``getWebhookSerializable``) carries only ``objectType``, + ``objectId`` (a fileid) and ``tagIds`` — not the tag *name*, the file path, + or whether our ``vector-index`` tag specifically changed. So we can't decide + index-vs-delete here. Emit a file task with ``file_path=None``; the processor + resolves the file's *current* ``vector-index`` membership and indexes or + deletes accordingly (see ``processor._reconcile_tag_event``). Both assign and + unassign collapse to the same reconcile, which also makes "an unrelated tag + changed" a cheap no-op. + """ + object_type = event.get("objectType") + object_id = event.get("objectId") + + # Only file tags drive vector sync; NC tags other object types too + # (comments, etc.). ``objectId`` of 0/"" is malformed — skip. + if object_type != "files" or not object_id: + return None + + return DocumentTask( + user_id=user_id, + doc_id=str(object_id), + doc_type="file", + # Reconciled in the processor; may be flipped to "delete" if the file is + # no longer a tagged PDF. file_path=None is the "needs reconcile" signal. + operation="index", + modified_at=time, + file_path=None, + ) + + def _parse_deck_event( event_class: str, event: dict, user_id: str, time: int ) -> DocumentTask | None: diff --git a/tests/unit/test_webhook_parser.py b/tests/unit/test_webhook_parser.py index ddcc86a6..a42af4da 100644 --- a/tests/unit/test_webhook_parser.py +++ b/tests/unit/test_webhook_parser.py @@ -216,3 +216,94 @@ def test_missing_time_field_defaults_to_zero(): task = extract_document_task(payload) assert task is not None assert task.modified_at == 0 + + +# --- SystemTag MapperEvent (assign/unassign) ------------------------------- +# +# NC 32+ serializes the event as {eventType, objectType, objectId, tagIds}. +# Both directions collapse to a path-less file "reconcile" task; the processor +# resolves current vector-index membership to decide index-vs-delete. + +_TAG_MAPPER = "OCP\\SystemTag\\MapperEvent" + + +@pytest.mark.unit +def test_tag_assign_event_returns_reconcile_task(): + payload = { + "user": {"uid": "alice"}, + "time": 1762850245, + "event": { + "class": _TAG_MAPPER, + "eventType": "OCP\\SystemTag\\ISystemTagObjectMapper::assignTags", + "objectType": "files", + "objectId": "478087", + "tagIds": [7], + }, + } + + task = extract_document_task(payload) + assert task is not None + assert task.user_id == "alice" + assert task.doc_type == "file" + assert task.doc_id == "478087" + assert task.operation == "index" + # No path in the payload -> the processor reconciles membership. + assert task.file_path is None + assert task.modified_at == 1762850245 + + +@pytest.mark.unit +def test_tag_unassign_event_also_returns_reconcile_task(): + """Unassign collapses to the same reconcile (processor flips to delete).""" + payload = { + "user": {"uid": "alice"}, + "time": 1762850245, + "event": { + "class": _TAG_MAPPER, + "eventType": "OCP\\SystemTag\\ISystemTagObjectMapper::unassignTags", + "objectType": "files", + "objectId": "478087", + "tagIds": [7], + }, + } + + task = extract_document_task(payload) + assert task is not None + assert task.doc_type == "file" + assert task.operation == "index" + assert task.file_path is None + + +@pytest.mark.unit +def test_tag_event_non_files_object_type_returns_none(): + """Tags on non-file objects (e.g. comments) don't drive vector sync.""" + payload = { + "user": {"uid": "alice"}, + "time": 1762850245, + "event": { + "class": _TAG_MAPPER, + "eventType": "OCP\\SystemTag\\ISystemTagObjectMapper::assignTags", + "objectType": "comments", + "objectId": "12", + "tagIds": [7], + }, + } + + assert extract_document_task(payload) is None + + +@pytest.mark.unit +def test_tag_event_missing_object_id_returns_none(): + payload = { + "user": {"uid": "alice"}, + "time": 1762850245, + "event": { + "class": _TAG_MAPPER, + "eventType": "OCP\\SystemTag\\ISystemTagObjectMapper::assignTags", + "objectType": "files", + "objectId": "", + "tagIds": [7], + }, + } + + assert extract_document_task(payload) is None diff --git a/tests/unit/test_webhook_presets.py b/tests/unit/test_webhook_presets.py index 248dd9c9..991250d3 100644 --- a/tests/unit/test_webhook_presets.py +++ b/tests/unit/test_webhook_presets.py @@ -89,6 +89,18 @@ def test_get_deck_preset(): assert "OCA\\Deck\\Event\\BoardUpdatedEvent" in event_classes +@pytest.mark.unit +def test_files_sync_includes_systemtag_event(): + """files_sync registers the SystemTag MapperEvent so tag changes (NC 32+) + drive vector-index (re)indexing, alongside the node create/write/delete + events.""" + preset = get_preset("files_sync") + assert preset is not None + event_classes = [e["event"] for e in preset["events"]] + assert "OCP\\SystemTag\\MapperEvent" in event_classes + assert "OCP\\Files\\Events\\Node\\NodeCreatedEvent" in event_classes + + @pytest.mark.unit def test_filter_presets_subset_installed(): """Test filtering when only some apps are installed.""" diff --git a/tests/unit/vector/test_tag_reconcile.py b/tests/unit/vector/test_tag_reconcile.py new file mode 100644 index 00000000..342bb085 --- /dev/null +++ b/tests/unit/vector/test_tag_reconcile.py @@ -0,0 +1,87 @@ +"""Unit tests for processor._reconcile_tag_event. + +A SystemTag MapperEvent enqueues a file task with only a fileid (file_path is +None). The reconcile resolves the file's *current* ``vector-index`` membership +and mutates the task into a concrete index (path/etag filled) or a delete. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nextcloud_mcp_server.vector.processor import _reconcile_tag_event +from nextcloud_mcp_server.vector.scanner import DocumentTask + + +def _tag_task(doc_id: str = "478087") -> DocumentTask: + return DocumentTask( + user_id="alice", + doc_id=doc_id, + doc_type="file", + operation="index", + modified_at=0, + file_path=None, + ) + + +@pytest.mark.unit +async def test_reconcile_tagged_file_becomes_index(): + """A fileid still carrying the tag is resolved to a concrete index task.""" + nc_client = MagicMock() + nc_client.find_files_by_tag = AsyncMock( + return_value=[ + { + "id": "478087", + "path": "/alice/files/Docs/report.pdf", + "etag": "abc123", + "last_modified_timestamp": 1762850245, + } + ] + ) + + task = _tag_task("478087") + await _reconcile_tag_event(task, nc_client) + + assert task.operation == "index" + assert task.file_path == "/alice/files/Docs/report.pdf" + assert task.etag == "abc123" + assert task.modified_at == 1762850245 + nc_client.find_files_by_tag.assert_awaited_once_with( + "vector-index", mime_type_filter="application/pdf" + ) + + +@pytest.mark.unit +async def test_reconcile_untagged_file_becomes_delete(): + """A fileid absent from the tagged set flips the task to a delete.""" + nc_client = MagicMock() + nc_client.find_files_by_tag = AsyncMock( + return_value=[ + {"id": "999", "path": "/alice/files/other.pdf", "etag": "z"}, + ] + ) + + task = _tag_task("478087") + await _reconcile_tag_event(task, nc_client) + + assert task.operation == "delete" + # Path stays None — the delete path addresses points by doc_id only. + assert task.file_path is None + + +@pytest.mark.unit +async def test_reconcile_preserves_existing_etag(): + """An etag already on the task is not overwritten by the tag listing.""" + nc_client = MagicMock() + nc_client.find_files_by_tag = AsyncMock( + return_value=[ + {"id": "478087", "path": "/alice/files/r.pdf", "etag": "from-listing"} + ] + ) + + task = _tag_task("478087") + task.etag = "preset" + await _reconcile_tag_event(task, nc_client) + + assert task.etag == "preset" + assert task.file_path == "/alice/files/r.pdf"