feat(vector): index files in real time on vector-index tag changes

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) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-05 16:20:05 +02:00
co-authored by Claude Opus 4.8
parent 78f3f284a6
commit 1322e5aba0
6 changed files with 313 additions and 4 deletions
+15 -1
View File
@@ -30,6 +30,13 @@ FILE_EVENT_WRITTEN = "OCP\\Files\\Events\\Node\\NodeWrittenEvent"
# See: https://github.com/nextcloud/server/issues/56371 # See: https://github.com/nextcloud/server/issues/56371
FILE_EVENT_DELETED = "OCP\\Files\\Events\\Node\\BeforeNodeDeletedEvent" 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 webhook events
CALENDAR_EVENT_CREATED = "OCP\\Calendar\\Events\\CalendarObjectCreatedEvent" CALENDAR_EVENT_CREATED = "OCP\\Calendar\\Events\\CalendarObjectCreatedEvent"
CALENDAR_EVENT_UPDATED = "OCP\\Calendar\\Events\\CalendarObjectUpdatedEvent" CALENDAR_EVENT_UPDATED = "OCP\\Calendar\\Events\\CalendarObjectUpdatedEvent"
@@ -128,7 +135,7 @@ WEBHOOK_PRESETS: Dict[str, WebhookPreset] = {
}, },
"files_sync": { "files_sync": {
"name": "All 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", "app": "files",
"events": [ "events": [
{ {
@@ -143,6 +150,13 @@ WEBHOOK_PRESETS: Dict[str, WebhookPreset] = {
"event": FILE_EVENT_DELETED, "event": FILE_EVENT_DELETED,
"filter": {}, "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": { "deck_sync": {
+64
View File
@@ -165,6 +165,58 @@ async def processor_task(
logger.info("Processor %s stopped", worker_id) 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( async def process_document(
doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3 doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3
): ):
@@ -204,6 +256,18 @@ async def process_document(
try: try:
qdrant_client = await get_qdrant_client() 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 # Handle deletion
if doc_task.operation == "delete": if doc_task.operation == "delete":
# Release this user rather than blind-delete: a file shared across # Release this user rather than blind-delete: a file shared across
+44 -3
View File
@@ -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 ``/webhooks/nextcloud`` calls :func:`extract_document_task` and forwards any
non-None result to the same processor send-stream the scanner uses. non-None result to the same processor send-stream the scanner uses.
Currently scoped to file (note) events and Deck card events. Calendar / Currently scoped to file (note) events, Deck card events, and SystemTag
Tables events fall through to ``None`` for now; those parsers can be added assign/unassign events (which drive ``vector-index`` (re)indexing of files).
in follow-up changes. 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 See ADR-010 for the design and ``webhook-testing-findings.md`` for real
captured payloads. captured payloads.
@@ -28,6 +29,11 @@ _DECK_EVENT_CARD_UPDATED = "OCA\\Deck\\Event\\CardUpdatedEvent"
_DECK_EVENT_CARD_DELETED = "OCA\\Deck\\Event\\CardDeletedEvent" _DECK_EVENT_CARD_DELETED = "OCA\\Deck\\Event\\CardDeletedEvent"
_DECK_EVENT_BOARD_UPDATED = "OCA\\Deck\\Event\\BoardUpdatedEvent" _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_CARD_EVENTS = frozenset(
{ {
_DECK_EVENT_CARD_CREATED, _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: if event_class in _DECK_CARD_EVENTS or event_class == _DECK_EVENT_BOARD_UPDATED:
return _parse_deck_event(event_class, event, user_id, time) 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) logger.debug("Ignoring webhook for unsupported event: %s", event_class)
return None 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( def _parse_deck_event(
event_class: str, event: dict, user_id: str, time: int event_class: str, event: dict, user_id: str, time: int
) -> DocumentTask | None: ) -> DocumentTask | None:
+91
View File
@@ -216,3 +216,94 @@ def test_missing_time_field_defaults_to_zero():
task = extract_document_task(payload) task = extract_document_task(payload)
assert task is not None assert task is not None
assert task.modified_at == 0 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
+12
View File
@@ -89,6 +89,18 @@ def test_get_deck_preset():
assert "OCA\\Deck\\Event\\BoardUpdatedEvent" in event_classes 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 @pytest.mark.unit
def test_filter_presets_subset_installed(): def test_filter_presets_subset_installed():
"""Test filtering when only some apps are installed.""" """Test filtering when only some apps are installed."""
+87
View File
@@ -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"