feat: implement background processing for binary attachments in Obsidian plugin

- Added a new Celery task for indexing non-markdown attachments.
- Enhanced the Obsidian plugin schema to support binary attachments.
- Updated routes to enqueue binary attachments for background processing.
- Improved metadata handling for binary attachments during indexing.
- Added tests for binary attachment processing and validation.

Author: AnishSarkar22

surfsense_backend/app/celery_app.py

@@ -90,6 +90,7 @@ def parse_schedule_interval(interval: str) -> dict:
         "app.tasks.celery_tasks.podcast_tasks",
         "app.tasks.celery_tasks.video_presentation_tasks",
         "app.tasks.celery_tasks.connector_tasks",
+        "app.tasks.celery_tasks.obsidian_tasks",
         "app.tasks.celery_tasks.schedule_checker_task",
         "app.tasks.celery_tasks.document_reindex_tasks",
         "app.tasks.celery_tasks.stale_notification_cleanup_task",
@@ -153,6 +154,7 @@ def parse_schedule_interval(interval: str) -> dict:
         "index_crawled_urls": {"queue": CONNECTORS_QUEUE},
         "index_bookstack_pages": {"queue": CONNECTORS_QUEUE},
         "index_composio_connector": {"queue": CONNECTORS_QUEUE},
+        "index_obsidian_attachment": {"queue": CONNECTORS_QUEUE},
         # Everything else (document processing, podcasts, reindexing,
         # schedule checker, cleanup) stays on the default fast queue.
     },

surfsense_backend/app/routes/obsidian_plugin_routes.py

@@ -49,6 +49,7 @@
     rename_note,
     upsert_note,
 )
+from app.tasks.celery_tasks.obsidian_tasks import index_obsidian_attachment_task
 from app.users import current_active_user
 
 logger = logging.getLogger(__name__)
@@ -204,6 +205,17 @@ async def _resolve_vault_connector(
     )
 
 
+def _queue_obsidian_attachment(
+    *, connector_id: int, note_payload: dict, user_id: str
+) -> None:
+    """Enqueue one non-markdown Obsidian note for background ETL/indexing."""
+    index_obsidian_attachment_task.delay(
+        connector_id=connector_id,
+        payload_data=note_payload,
+        user_id=user_id,
+    )
+
+
 async def _ensure_search_space_access(
     session: AsyncSession,
     *,
@@ -452,6 +464,16 @@ async def obsidian_sync(
 
     for note in payload.notes:
         try:
+            if note.is_binary:
+                _queue_obsidian_attachment(
+                    connector_id=connector.id,
+                    note_payload=note.model_dump(mode="json"),
+                    user_id=str(user.id),
+                )
+                indexed += 1
+                items.append(SyncAckItem(path=note.path, status="queued"))
+                continue
+
             doc = await upsert_note(
                 session, connector=connector, payload=note, user_id=str(user.id)
             )

surfsense_backend/app/schemas/obsidian_plugin.py

@@ -52,6 +52,24 @@ class NotePayload(_PluginBase):
     content_hash: str = Field(
         ..., description="Plugin-computed SHA-256 of the raw content"
     )
+    is_binary: bool = Field(
+        default=False,
+        description=(
+            "True when payload represents a non-markdown attachment. "
+            "If set, the plugin may include binary_base64 for ETL extraction."
+        ),
+    )
+    binary_base64: str | None = Field(
+        default=None,
+        description=(
+            "Base64-encoded raw file bytes for binary attachments. "
+            "Used by the backend ETL pipeline."
+        ),
+    )
+    mime_type: str | None = Field(
+        default=None,
+        description="Optional MIME type hint for binary attachments.",
+    )
     size: int | None = Field(
         default=None,
         ge=0,
@@ -138,7 +156,7 @@ class HealthResponse(_PluginBase):
 
 class SyncAckItem(_PluginBase):
     path: str
-    status: Literal["ok", "error"]
+    status: Literal["ok", "queued", "error"]
     document_id: int | None = None
     error: str | None = None
 

surfsense_backend/app/services/obsidian_plugin_indexer.py

@@ -32,7 +32,11 @@
 
 from __future__ import annotations
 
+import base64
+import contextlib
 import logging
+import os
+import tempfile
 from datetime import UTC, datetime
 from typing import Any
 from urllib.parse import quote
@@ -113,19 +117,26 @@ def _build_metadata(
         "connector_id": connector_id,
         "url": _build_source_url(vault_name, payload.path),
     }
+    if payload.is_binary:
+        meta["is_binary"] = True
+        if payload.mime_type:
+            meta["mime_type"] = payload.mime_type
     if extra:
         meta.update(extra)
     return meta
 
 
-def _build_document_string(payload: NotePayload, vault_name: str) -> str:
+def _build_document_string(
+    payload: NotePayload, vault_name: str, *, content_override: str | None = None
+) -> str:
     """Compose the indexable string the pipeline embeds and chunks.
 
     Mirrors the legacy obsidian indexer's METADATA + CONTENT framing so
     existing search relevance heuristics keep working unchanged.
     """
     tags_line = ", ".join(payload.tags) if payload.tags else "None"
     links_line = ", ".join(payload.resolved_links) if payload.resolved_links else "None"
+    body = payload.content if content_override is None else content_override
     return (
         "<METADATA>\n"
         f"Title: {payload.name}\n"
@@ -135,11 +146,120 @@ def _build_document_string(payload: NotePayload, vault_name: str) -> str:
         f"Links to: {links_line}\n"
         "</METADATA>\n\n"
         "<CONTENT>\n"
-        f"{payload.content}\n"
+        f"{body}\n"
         "</CONTENT>\n"
     )
 
 
+async def _extract_binary_attachment_markdown(
+    payload: NotePayload, *, vision_llm
+) -> tuple[str, dict[str, Any]]:
+    if not payload.binary_base64:
+        return "", {"attachment_extraction_status": "missing_binary_payload"}
+
+    try:
+        raw_bytes = base64.b64decode(payload.binary_base64, validate=True)
+    except Exception:
+        logger.warning("obsidian attachment payload had invalid base64: %s", payload.path)
+        return "", {"attachment_extraction_status": "invalid_binary_payload"}
+
+    suffix = f".{payload.extension.lstrip('.')}" if payload.extension else ""
+    temp_path: str | None = None
+    filename = payload.path.rsplit("/", 1)[-1] or payload.name
+    try:
+        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+            tmp.write(raw_bytes)
+            temp_path = tmp.name
+
+        result = await _run_etl_extract(
+            file_path=temp_path,
+            filename=filename,
+            vision_llm=vision_llm,
+        )
+        metadata: dict[str, Any] = {
+            "attachment_extraction_status": "ok",
+            "attachment_etl_service": result.etl_service,
+            "attachment_content_type": result.content_type,
+        }
+        return result.markdown_content, metadata
+    except Exception as exc:
+        logger.warning(
+            "obsidian attachment ETL failed for %s: %s", payload.path, exc, exc_info=True
+        )
+        return "", {
+            "attachment_extraction_status": "etl_failed",
+            "attachment_extraction_error": str(exc)[:300],
+        }
+    finally:
+        if temp_path and os.path.exists(temp_path):
+            with contextlib.suppress(Exception):
+                os.unlink(temp_path)
+
+
+async def _run_etl_extract(*, file_path: str, filename: str, vision_llm):
+    """Lazy-load ETL dependencies to avoid module-import cycles."""
+    from app.etl_pipeline.etl_document import EtlRequest
+    from app.etl_pipeline.etl_pipeline_service import EtlPipelineService
+
+    return await EtlPipelineService(vision_llm=vision_llm).extract(
+        EtlRequest(file_path=file_path, filename=filename)
+    )
+
+
+def _is_image_attachment(payload: NotePayload) -> bool:
+    ext = payload.extension.lower().lstrip(".")
+    return ext in {"png", "jpg", "jpeg", "gif", "webp", "bmp", "tiff", "svg"}
+
+
+async def _resolve_attachment_vision_llm(
+    session: AsyncSession,
+    *,
+    connector: SearchSourceConnector,
+    search_space_id: int,
+    payload: NotePayload,
+):
+    """Match connector indexers: only fetch vision LLM for image attachments
+    when the connector has vision indexing enabled."""
+    if not payload.is_binary:
+        return None
+    if not _is_image_attachment(payload):
+        return None
+    if not getattr(connector, "enable_vision_llm", False):
+        return None
+
+    from app.services.llm_service import get_vision_llm
+
+    return await get_vision_llm(session, search_space_id)
+
+
+async def _resolve_summary_llm(
+    session: AsyncSession, *, user_id: str, search_space_id: int, should_summarize: bool
+):
+    """Fetch summary LLM only when indexing summary is enabled."""
+    if not should_summarize:
+        return None
+
+    from app.services.llm_service import get_user_long_context_llm
+
+    return await get_user_long_context_llm(session, user_id, search_space_id)
+
+
+def _require_extracted_attachment_content(
+    *, content: str, etl_meta: dict[str, Any], path: str
+) -> str:
+    extracted = content.strip()
+    if extracted:
+        return extracted
+
+    status = etl_meta.get("attachment_extraction_status", "unknown")
+    reason = etl_meta.get("attachment_extraction_error")
+    if reason:
+        raise RuntimeError(
+            f"Attachment extraction failed for {path} ({status}): {reason}"
+        )
+    raise RuntimeError(f"Attachment extraction failed for {path} ({status})")
+
+
 async def _find_existing_document(
     session: AsyncSession,
     *,
@@ -207,11 +327,42 @@ async def upsert_note(
                 exc_info=True,
             )
 
-    document_string = _build_document_string(payload, vault_name)
+    content_for_index = payload.content
+    extra_meta: dict[str, Any] = {}
+    vision_llm = None
+    if payload.is_binary:
+        vision_llm = await _resolve_attachment_vision_llm(
+            session,
+            connector=connector,
+            search_space_id=search_space_id,
+            payload=payload,
+        )
+        content_for_index, etl_meta = await _extract_binary_attachment_markdown(
+            payload, vision_llm=vision_llm
+        )
+        extra_meta.update(etl_meta)
+        # Strict KB behavior: do not index metadata-only attachments.
+        content_for_index = _require_extracted_attachment_content(
+            content=content_for_index,
+            etl_meta=etl_meta,
+            path=payload.path,
+        )
+
+    llm = await _resolve_summary_llm(
+        session,
+        user_id=str(user_id),
+        search_space_id=search_space_id,
+        should_summarize=connector.enable_summary,
+    )
+
+    document_string = _build_document_string(
+        payload, vault_name, content_override=content_for_index
+    )
     metadata = _build_metadata(
         payload,
         vault_name=vault_name,
         connector_id=connector.id,
+        extra=extra_meta,
     )
 
     connector_doc = ConnectorDocument(
@@ -223,7 +374,7 @@ async def upsert_note(
         connector_id=connector.id,
         created_by_id=str(user_id),
         should_summarize=connector.enable_summary,
-        fallback_summary=f"Obsidian Note: {payload.name}\n\n{payload.content}",
+        fallback_summary=f"Obsidian Note: {payload.name}\n\n{content_for_index}",
         metadata=metadata,
     )
 
@@ -236,9 +387,6 @@ async def upsert_note(
 
     document = prepared[0]
 
-    from app.services.llm_service import get_user_long_context_llm
-
-    llm = await get_user_long_context_llm(session, str(user_id), search_space_id)
     return await pipeline.index(document, connector_doc, llm)
 
 

surfsense_backend/app/tasks/celery_tasks/obsidian_tasks.py

@@ -0,0 +1,59 @@
+"""Celery tasks for Obsidian plugin background processing."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from app.celery_app import celery_app
+from app.db import SearchSourceConnector
+from app.schemas.obsidian_plugin import NotePayload
+from app.services.obsidian_plugin_indexer import upsert_note
+from app.tasks.celery_tasks import get_celery_session_maker
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(name="index_obsidian_attachment", bind=True)
+def index_obsidian_attachment_task(
+    self,
+    connector_id: int,
+    payload_data: dict,
+    user_id: str,
+) -> None:
+    """Process one Obsidian non-markdown attachment asynchronously."""
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        loop.run_until_complete(
+            _index_obsidian_attachment(
+                connector_id=connector_id,
+                payload_data=payload_data,
+                user_id=user_id,
+            )
+        )
+    finally:
+        loop.close()
+
+
+async def _index_obsidian_attachment(
+    *,
+    connector_id: int,
+    payload_data: dict,
+    user_id: str,
+) -> None:
+    async with get_celery_session_maker()() as session:
+        connector = await session.get(SearchSourceConnector, connector_id)
+        if connector is None:
+            logger.warning(
+                "obsidian attachment task skipped: connector %s not found", connector_id
+            )
+            return
+
+        payload = NotePayload.model_validate(payload_data)
+        await upsert_note(
+            session,
+            connector=connector,
+            payload=payload,
+            user_id=user_id,
+        )