documents: generalize revision recorder to batched markdown writes

record_markdown_files records many paths as one transaction; a save is
now a batch of one, and record_prepared_documents maps a sync batch to
a single revision. Unchanged content stays a natural no-op.

surfsense_backend/app/services/document_revision_recorder.py

@@ -1,12 +1,15 @@
-"""Direct-caller adapter: a document save becomes one knowledge-store revision.
+"""Direct-caller adapter: document content changes become knowledge-store revisions.
 
-Editor saves and upload-extracted markdown share this path — the same single
-write path agent turns use — behind ``KNOWLEDGE_STORE_ENABLED``.
+Editor saves, upload-extracted markdown, and connector sync batches share this
+path — the same single write path agent turns use — behind
+``KNOWLEDGE_STORE_ENABLED``.
 """
 
 from __future__ import annotations
 
 import logging
+from collections.abc import Mapping, Sequence
+from typing import TYPE_CHECKING
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -18,25 +21,32 @@ from app.knowledge_store import KnowledgeStore
 from app.knowledge_store.identities import user_identity
 from app.knowledge_store.settings import load_knowledge_store_settings
 
+if TYPE_CHECKING:
+    from app.db import Document
+
 logger = logging.getLogger(__name__)
 
 
-async def record_document_markdown(
+async def record_markdown_files(
     *,
     workspace_id: int | str,
-    store_path: str,
-    markdown: str,
+    files: Mapping[str, str],
+    message: str,
     author_user_id: str | None,
 ) -> str | None:
-    """Record one revision for one saved document; ``None`` when disabled."""
-    if not load_knowledge_store_settings().enabled:
+    """Record ``files`` (store path → markdown) as one revision.
+
+    ``None`` when the store is disabled, the batch is empty, or nothing
+    actually changed (identical content is a no-op by construction).
+    """
+    if not files or not load_knowledge_store_settings().enabled:
         return None
     store = KnowledgeStore.for_workspace(workspace_id)
-    filename = store_path.rsplit("/", 1)[-1]
     async with store.transaction(
-        message=f"docs: save {filename}", author=user_identity(author_user_id)
+        message=message, author=user_identity(author_user_id)
     ) as tx:
-        tx.write(store_path, markdown.encode())
+        for path, markdown in files.items():
+            tx.write(path, markdown.encode())
     return tx.revision
 
 
@@ -50,7 +60,7 @@ async def record_saved_document(
     markdown: str,
     author_user_id: str | None,
 ) -> str | None:
-    """Resolve the document's canonical store path and record the save.
+    """Resolve one document's canonical store path and record the save.
 
     Never raises: while the store coexists with the Postgres write path
     (until the Phase 5 cut), a recording failure must not fail the save
@@ -63,10 +73,11 @@ async def record_saved_document(
         virtual_path = doc_to_virtual_path(
             doc_id=doc_id, title=title, folder_id=folder_id, index=index
         )
-        return await record_document_markdown(
+        filename = virtual_path.rsplit("/", 1)[-1]
+        return await record_markdown_files(
             workspace_id=workspace_id,
-            store_path=virtual_path.lstrip("/"),
-            markdown=markdown,
+            files={virtual_path.lstrip("/"): markdown},
+            message=f"docs: save {filename}",
             author_user_id=author_user_id,
         )
     except Exception:
@@ -77,3 +88,44 @@ async def record_saved_document(
             exc_info=True,
         )
         return None
+
+
+async def record_prepared_documents(
+    session: AsyncSession, documents: Sequence[Document]
+) -> str | None:
+    """Record a sync batch's accepted markdown as one revision.
+
+    Called after ``prepare_for_indexing`` commits — the moment content becomes
+    durable — so chunking/embedding failures can never block the record.
+    Never raises, for the same coexistence reason as ``record_saved_document``.
+    """
+    if not documents or not load_knowledge_store_settings().enabled:
+        return None
+    try:
+        workspace_id = documents[0].workspace_id
+        index = await build_path_index(session, workspace_id)
+        files: dict[str, str] = {}
+        for doc in documents:
+            if not doc.source_markdown:
+                continue
+            virtual_path = doc_to_virtual_path(
+                doc_id=doc.id, title=doc.title, folder_id=doc.folder_id, index=index
+            )
+            files[virtual_path.lstrip("/")] = doc.source_markdown
+        return await record_markdown_files(
+            workspace_id=workspace_id,
+            files=files,
+            message=f"sync: index {len(files)} document(s)",
+            author_user_id=(
+                str(documents[0].created_by_id)
+                if documents[0].created_by_id is not None
+                else None
+            ),
+        )
+    except Exception:
+        logger.warning(
+            "Knowledge store recording failed for a sync batch in workspace %s",
+            documents[0].workspace_id,
+            exc_info=True,
+        )
+        return None

surfsense_backend/tests/integration/knowledge_store/test_document_recorder.py

@@ -6,7 +6,7 @@ import pytest
 
 from app.config import config as app_config
 from app.knowledge_store import KnowledgeStore
-from app.services.document_revision_recorder import record_document_markdown
+from app.services.document_revision_recorder import record_markdown_files
 
 pytestmark = pytest.mark.integration
 
@@ -19,10 +19,10 @@ def knowledge_root(tmp_path, monkeypatch):
 
 
 async def test_one_save_records_one_revision(knowledge_root, workspace_id):
-    revision = await record_document_markdown(
+    revision = await record_markdown_files(
         workspace_id=workspace_id,
-        store_path="documents/notes/meeting.md",
-        markdown="# Meeting",
+        files={"documents/notes/meeting.md": "# Meeting"},
+        message="docs: save meeting.md",
         author_user_id="1",
     )
 
@@ -34,14 +34,54 @@ async def test_one_save_records_one_revision(knowledge_root, workspace_id):
     assert "meeting.md" in rev.message
 
 
+async def test_a_sync_batch_records_one_revision(knowledge_root, workspace_id):
+    revision = await record_markdown_files(
+        workspace_id=workspace_id,
+        files={
+            "documents/notion/roadmap.md": "# Roadmap",
+            "documents/notion/okrs.md": "# OKRs",
+        },
+        message="sync: index 2 document(s)",
+        author_user_id="1",
+    )
+
+    store = KnowledgeStore.for_workspace(workspace_id)
+    revisions = await store.list_revisions()
+    assert [r.id for r in revisions] == [revision]
+    assert await store.read_as_of(revision, "documents/notion/okrs.md") == b"# OKRs"
+
+
+async def test_unchanged_content_records_nothing(knowledge_root, workspace_id):
+    files = {"documents/notion/roadmap.md": "# Roadmap"}
+    first = await record_markdown_files(
+        workspace_id=workspace_id, files=files, message="sync", author_user_id="1"
+    )
+    second = await record_markdown_files(
+        workspace_id=workspace_id, files=files, message="sync", author_user_id="1"
+    )
+
+    assert first is not None
+    assert second is None
+    assert len(await KnowledgeStore.for_workspace(workspace_id).list_revisions()) == 1
+
+
+async def test_empty_batch_records_nothing(knowledge_root, workspace_id):
+    revision = await record_markdown_files(
+        workspace_id=workspace_id, files={}, message="sync", author_user_id="1"
+    )
+
+    assert revision is None
+    assert not (knowledge_root / str(workspace_id)).exists()
+
+
 async def test_disabled_store_records_nothing(monkeypatch, tmp_path, workspace_id):
     monkeypatch.setattr(app_config, "KNOWLEDGE_STORE_ENABLED", False)
     monkeypatch.setattr(app_config, "KNOWLEDGE_STORE_ROOT", str(tmp_path))
 
-    revision = await record_document_markdown(
+    revision = await record_markdown_files(
         workspace_id=workspace_id,
-        store_path="documents/notes/meeting.md",
-        markdown="# Meeting",
+        files={"documents/notes/meeting.md": "# Meeting"},
+        message="docs: save meeting.md",
         author_user_id="1",
     )