Merge branch 'dev' into sur-90-feat-comments-in-chats

surfsense_backend/alembic/env.py

@@ -25,6 +25,13 @@ database_url = os.getenv("DATABASE_URL")
 if database_url:
     config.set_main_option("sqlalchemy.url", database_url)
 
+# Electric SQL user credentials - centralized configuration for migrations
+# These are used by migrations that set up Electric SQL replication
+config.set_main_option("electric_db_user", os.getenv("ELECTRIC_DB_USER", "electric"))
+config.set_main_option(
+    "electric_db_password", os.getenv("ELECTRIC_DB_PASSWORD", "electric_password")
+)
+
 # Interpret the config file for Python logging.
 # This line sets up loggers basically.
 if config.config_file_name is not None:

surfsense_backend/alembic/versions/66_add_notifications_table_and_electric_replication.py

@@ -0,0 +1,172 @@
+"""Add notifications table and Electric SQL replication
+
+Revision ID: 66
+Revises: 65
+
+Creates notifications table and sets up Electric SQL replication
+(user, publication, REPLICA IDENTITY FULL) for notifications,
+search_source_connectors, and documents tables.
+
+NOTE: Electric SQL user creation is idempotent (uses IF NOT EXISTS).
+- Docker deployments: user is pre-created by scripts/docker/init-electric-user.sh
+- Local PostgreSQL: user is created here during migration
+Both approaches are safe to run together without conflicts as this migraiton is idempotent
+"""
+
+from collections.abc import Sequence
+
+from alembic import context, op
+
+# Get Electric SQL user credentials from env.py configuration
+_config = context.config
+ELECTRIC_DB_USER = _config.get_main_option("electric_db_user", "electric")
+ELECTRIC_DB_PASSWORD = _config.get_main_option(
+    "electric_db_password", "electric_password"
+)
+
+# revision identifiers, used by Alembic.
+revision: str = "66"
+down_revision: str | None = "65"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema - add notifications table and Electric SQL replication."""
+    # Create notifications table
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS notifications (
+            id SERIAL PRIMARY KEY,
+            user_id UUID NOT NULL REFERENCES "user"(id) ON DELETE CASCADE,
+            search_space_id INTEGER REFERENCES searchspaces(id) ON DELETE CASCADE,
+            type VARCHAR(50) NOT NULL,
+            title VARCHAR(200) NOT NULL,
+            message TEXT NOT NULL,
+            read BOOLEAN NOT NULL DEFAULT FALSE,
+            metadata JSONB DEFAULT '{}',
+            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            updated_at TIMESTAMPTZ
+        );
+        """
+    )
+
+    # Create indexes (using IF NOT EXISTS for idempotency)
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_notifications_user_id ON notifications (user_id);"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_notifications_read ON notifications (read);"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_notifications_created_at ON notifications (created_at);"
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS ix_notifications_user_read ON notifications (user_id, read);"
+    )
+
+    # =====================================================
+    # Electric SQL Setup - User and Publication
+    # =====================================================
+
+    # Create Electric SQL replication user if not exists
+    op.execute(
+        f"""
+        DO $$
+        BEGIN
+            IF NOT EXISTS (SELECT FROM pg_user WHERE usename = '{ELECTRIC_DB_USER}') THEN
+                CREATE USER {ELECTRIC_DB_USER} WITH REPLICATION PASSWORD '{ELECTRIC_DB_PASSWORD}';
+            END IF;
+        END
+        $$;
+        """
+    )
+
+    # Grant necessary permissions to electric user
+    op.execute(
+        f"""
+        DO $$
+        DECLARE
+            db_name TEXT := current_database();
+        BEGIN
+            EXECUTE format('GRANT CONNECT ON DATABASE %I TO {ELECTRIC_DB_USER}', db_name);
+        END
+        $$;
+        """
+    )
+    op.execute(f"GRANT USAGE ON SCHEMA public TO {ELECTRIC_DB_USER};")
+    op.execute(f"GRANT SELECT ON ALL TABLES IN SCHEMA public TO {ELECTRIC_DB_USER};")
+    op.execute(f"GRANT SELECT ON ALL SEQUENCES IN SCHEMA public TO {ELECTRIC_DB_USER};")
+    op.execute(
+        f"ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO {ELECTRIC_DB_USER};"
+    )
+    op.execute(
+        f"ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON SEQUENCES TO {ELECTRIC_DB_USER};"
+    )
+
+    # Create the publication if not exists
+    op.execute(
+        """
+        DO $$
+        BEGIN
+            IF NOT EXISTS (SELECT FROM pg_publication WHERE pubname = 'electric_publication_default') THEN
+                CREATE PUBLICATION electric_publication_default;
+            END IF;
+        END
+        $$;
+        """
+    )
+
+    # =====================================================
+    # Electric SQL Setup - Table Configuration
+    # =====================================================
+
+    # Set REPLICA IDENTITY FULL (required by Electric SQL for replication)
+    op.execute("ALTER TABLE notifications REPLICA IDENTITY FULL;")
+    op.execute("ALTER TABLE search_source_connectors REPLICA IDENTITY FULL;")
+    op.execute("ALTER TABLE documents REPLICA IDENTITY FULL;")
+
+    # Add tables to Electric SQL publication for replication
+    op.execute(
+        """
+        DO $$
+        BEGIN
+            -- Add notifications if not already added
+            IF NOT EXISTS (
+                SELECT 1 FROM pg_publication_tables 
+                WHERE pubname = 'electric_publication_default' 
+                AND tablename = 'notifications'
+            ) THEN
+                ALTER PUBLICATION electric_publication_default ADD TABLE notifications;
+            END IF;
+            
+            -- Add search_source_connectors if not already added
+            IF NOT EXISTS (
+                SELECT 1 FROM pg_publication_tables 
+                WHERE pubname = 'electric_publication_default' 
+                AND tablename = 'search_source_connectors'
+            ) THEN
+                ALTER PUBLICATION electric_publication_default ADD TABLE search_source_connectors;
+            END IF;
+            
+            -- Add documents if not already added
+            IF NOT EXISTS (
+                SELECT 1 FROM pg_publication_tables 
+                WHERE pubname = 'electric_publication_default' 
+                AND tablename = 'documents'
+            ) THEN
+                ALTER PUBLICATION electric_publication_default ADD TABLE documents;
+            END IF;
+        END
+        $$;
+        """
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema - remove notifications table."""
+    op.drop_index("ix_notifications_user_read", table_name="notifications")
+    op.drop_index("ix_notifications_created_at", table_name="notifications")
+    op.drop_index("ix_notifications_read", table_name="notifications")
+    op.drop_index("ix_notifications_user_id", table_name="notifications")
+    op.drop_table("notifications")

surfsense_backend/alembic/versions/67_add_pg_trgm_index_for_document_title_search.py

@@ -0,0 +1,76 @@
+"""Add pg_trgm indexes for efficient document title search
+
+Revision ID: 67
+Revises: 66
+
+Adds the pg_trgm extension and GIN trigram indexes on documents.title
+to enable efficient ILIKE searches with leading wildcards (e.g., '%search_term%').
+
+Indexes added:
+1. idx_documents_title_trgm - GIN trigram on title for ILIKE '%term%'
+2. idx_documents_search_space_id - B-tree on search_space_id for filtering
+3. idx_documents_search_space_updated - Composite for recent docs query (covering index)
+4. idx_surfsense_docs_title_trgm - GIN trigram on surfsense docs title
+
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "67"
+down_revision: str | None = "66"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Add pg_trgm extension and optimized indexes for document search."""
+
+    # Create pg_trgm extension if not exists
+    # This extension provides trigram-based text similarity functions and operators
+    op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm;")
+
+    # 1. GIN trigram index on documents.title for ILIKE '%term%' searches
+    op.execute(
+        """
+        CREATE INDEX IF NOT EXISTS idx_documents_title_trgm 
+        ON documents USING gin (title gin_trgm_ops);
+        """
+    )
+
+    # 2. B-tree index on search_space_id for fast filtering
+    # (Every query filters by search_space_id first)
+    op.execute(
+        """
+        CREATE INDEX IF NOT EXISTS idx_documents_search_space_id 
+        ON documents (search_space_id);
+        """
+    )
+
+    # 3. Covering index for "recent documents" query (no search term)
+    # Includes id, title, document_type so PostgreSQL can do index-only scan
+    op.execute(
+        """
+        CREATE INDEX IF NOT EXISTS idx_documents_search_space_updated 
+        ON documents (search_space_id, updated_at DESC NULLS LAST)
+        INCLUDE (id, title, document_type);
+        """
+    )
+
+    # 4. GIN trigram index on surfsense_docs_documents.title
+    op.execute(
+        """
+        CREATE INDEX IF NOT EXISTS idx_surfsense_docs_title_trgm 
+        ON surfsense_docs_documents USING gin (title gin_trgm_ops);
+        """
+    )
+
+
+def downgrade() -> None:
+    """Remove all document search indexes (extension is left in place)."""
+    op.execute("DROP INDEX IF EXISTS idx_surfsense_docs_title_trgm;")
+    op.execute("DROP INDEX IF EXISTS idx_documents_search_space_updated;")
+    op.execute("DROP INDEX IF EXISTS idx_documents_search_space_id;")
+    op.execute("DROP INDEX IF EXISTS idx_documents_title_trgm;")

surfsense_backend/app/agents/new_chat/tools/mcp_client.py

@@ -4,6 +4,7 @@ This module provides a client for communicating with MCP servers via stdio trans
 It handles server lifecycle management, tool discovery, and tool execution.
 """
 
+import asyncio
 import logging
 import os
 from contextlib import asynccontextmanager
@@ -14,6 +15,11 @@ from mcp.client.stdio import StdioServerParameters, stdio_client
 
 logger = logging.getLogger(__name__)
 
+# Retry configuration
+MAX_RETRIES = 3
+RETRY_DELAY = 1.0  # seconds
+RETRY_BACKOFF = 2.0  # exponential backoff multiplier
+
 
 class MCPClient:
     """Client for communicating with an MCP server."""
@@ -35,44 +41,86 @@ class MCPClient:
         self.session: ClientSession | None = None
 
     @asynccontextmanager
-    async def connect(self):
+    async def connect(self, max_retries: int = MAX_RETRIES):
         """Connect to the MCP server and manage its lifecycle.
 
+        Args:
+            max_retries: Maximum number of connection retry attempts
+
         Yields:
             ClientSession: Active MCP session for making requests
 
+        Raises:
+            RuntimeError: If all connection attempts fail
+
         """
-        try:
-            # Merge env vars with current environment
-            server_env = os.environ.copy()
-            server_env.update(self.env)
+        last_error = None
+        delay = RETRY_DELAY
 
-            # Create server parameters with env
-            server_params = StdioServerParameters(
-                command=self.command, args=self.args, env=server_env
-            )
+        for attempt in range(max_retries):
+            try:
+                # Merge env vars with current environment
+                server_env = os.environ.copy()
+                server_env.update(self.env)
 
-            # Spawn server process and create session
-            # Note: Cannot combine these context managers because ClientSession
-            # needs the read/write streams from stdio_client
-            async with stdio_client(server=server_params) as (read, write):  # noqa: SIM117
-                async with ClientSession(read, write) as session:
-                    # Initialize the connection
-                    await session.initialize()
-                    self.session = session
-                    logger.info(
-                        "Connected to MCP server: %s %s",
-                        self.command,
-                        " ".join(self.args),
+                # Create server parameters with env
+                server_params = StdioServerParameters(
+                    command=self.command, args=self.args, env=server_env
+                )
+
+                # Spawn server process and create session
+                # Note: Cannot combine these context managers because ClientSession
+                # needs the read/write streams from stdio_client
+                async with stdio_client(server=server_params) as (read, write):  # noqa: SIM117
+                    async with ClientSession(read, write) as session:
+                        # Initialize the connection
+                        await session.initialize()
+                        self.session = session
+                        
+                        if attempt > 0:
+                            logger.info(
+                                "Connected to MCP server on attempt %d: %s %s",
+                                attempt + 1,
+                                self.command,
+                                " ".join(self.args),
+                            )
+                        else:
+                            logger.info(
+                                "Connected to MCP server: %s %s",
+                                self.command,
+                                " ".join(self.args),
+                            )
+                        yield session
+                        return  # Success, exit retry loop
+
+            except Exception as e:
+                last_error = e
+                if attempt < max_retries - 1:
+                    logger.warning(
+                        "MCP server connection failed (attempt %d/%d): %s. Retrying in %.1fs...",
+                        attempt + 1,
+                        max_retries,
+                        e,
+                        delay,
                     )
-                    yield session
+                    await asyncio.sleep(delay)
+                    delay *= RETRY_BACKOFF  # Exponential backoff
+                else:
+                    logger.error(
+                        "Failed to connect to MCP server after %d attempts: %s",
+                        max_retries,
+                        e,
+                        exc_info=True,
+                    )
+            finally:
+                self.session = None
 
-        except Exception as e:
-            logger.error("Failed to connect to MCP server: %s", e, exc_info=True)
-            raise
-        finally:
-            self.session = None
-            logger.info("Disconnected from MCP server: %s", self.command)
+        # All retries exhausted
+        error_msg = f"Failed to connect to MCP server '{self.command}' after {max_retries} attempts"
+        if last_error:
+            error_msg += f": {last_error}"
+        logger.error(error_msg)
+        raise RuntimeError(error_msg) from last_error
 
     async def list_tools(self) -> list[dict[str, Any]]:
         """List all tools available from the MCP server.

surfsense_backend/app/agents/new_chat/tools/mcp_tool.py

@@ -90,16 +90,22 @@ async def _create_mcp_tool_from_definition(
     input_model = _create_dynamic_input_model_from_schema(tool_name, input_schema)
 
     async def mcp_tool_call(**kwargs) -> str:
-        """Execute the MCP tool call via the client."""
+        """Execute the MCP tool call via the client with retry support."""
         logger.info(f"MCP tool '{tool_name}' called with params: {kwargs}")
 
         try:
-            # Connect to server and call tool
+            # Connect to server and call tool (connect has built-in retry logic)
             async with mcp_client.connect():
                 result = await mcp_client.call_tool(tool_name, kwargs)
                 return str(result)
+        except RuntimeError as e:
+            # Connection failures after all retries
+            error_msg = f"MCP tool '{tool_name}' connection failed after retries: {e!s}"
+            logger.error(error_msg)
+            return f"Error: {error_msg}"
         except Exception as e:
-            error_msg = f"MCP tool '{tool_name}' failed: {e!s}"
+            # Tool execution or other errors
+            error_msg = f"MCP tool '{tool_name}' execution failed: {e!s}"
             logger.exception(error_msg)
             return f"Error: {error_msg}"
 
@@ -146,17 +152,38 @@ async def load_mcp_tools(
         tools: list[StructuredTool] = []
         for connector in result.scalars():
             try:
-                # Extract server config
+                # Early validation: Extract and validate connector config
                 config = connector.config or {}
                 server_config = config.get("server_config", {})
-
-                command = server_config.get("command")
-                args = server_config.get("args", [])
-                env = server_config.get("env", {})
-
-                if not command:
+                
+                # Validate server_config exists and is a dict
+                if not server_config or not isinstance(server_config, dict):
                     logger.warning(
-                        f"MCP connector {connector.id} missing command, skipping"
+                        f"MCP connector {connector.id} (name: '{connector.name}') has invalid or missing server_config, skipping"
+                    )
+                    continue
+
+                # Validate required command field
+                command = server_config.get("command")
+                if not command or not isinstance(command, str):
+                    logger.warning(
+                        f"MCP connector {connector.id} (name: '{connector.name}') missing or invalid command field, skipping"
+                    )
+                    continue
+
+                # Validate args field (must be list if present)
+                args = server_config.get("args", [])
+                if not isinstance(args, list):
+                    logger.warning(
+                        f"MCP connector {connector.id} (name: '{connector.name}') has invalid args field (must be list), skipping"
+                    )
+                    continue
+
+                # Validate env field (must be dict if present)
+                env = server_config.get("env", {})
+                if not isinstance(env, dict):
+                    logger.warning(
+                        f"MCP connector {connector.id} (name: '{connector.name}') has invalid env field (must be dict), skipping"
                     )
                     continue
 
@@ -172,22 +199,21 @@ async def load_mcp_tools(
                         f"'{command}' (connector {connector.id})"
                     )
 
-                    # Create LangChain tools from definitions
-                    for tool_def in tool_definitions:
-                        try:
-                            tool = await _create_mcp_tool_from_definition(
-                                tool_def, mcp_client
-                            )
-                            tools.append(tool)
-                        except Exception as e:
-                            logger.exception(
-                                f"Failed to create tool '{tool_def.get('name')}' "
-                                f"from connector {connector.id}: {e!s}",
-                            )
-
+                # Create LangChain tools from definitions
+                for tool_def in tool_definitions:
+                    try:
+                        tool = await _create_mcp_tool_from_definition(
+                            tool_def, mcp_client
+                        )
+                        tools.append(tool)
+                    except Exception as e:
+                        logger.exception(
+                            f"Failed to create tool '{tool_def.get('name')}' "
+                            f"from connector {connector.id}: {e!s}"
+                        )
             except Exception as e:
                 logger.exception(
-                    f"Failed to load tools from MCP connector {connector.id}: {e!s}",
+                    f"Failed to load tools from MCP connector {connector.id}: {e!s}"
                 )
 
         logger.info(f"Loaded {len(tools)} MCP tools for search space {search_space_id}")

surfsense_backend/app/connectors/google_drive/change_tracker.py

@@ -58,7 +58,7 @@ async def get_changes(
         params = {
             "pageToken": page_token,
             "pageSize": 100,
-            "fields": "nextPageToken, newStartPageToken, changes(fileId, removed, file(id, name, mimeType, modifiedTime, size, webViewLink, parents, trashed))",
+            "fields": "nextPageToken, newStartPageToken, changes(fileId, removed, file(id, name, mimeType, modifiedTime, md5Checksum, size, webViewLink, parents, trashed))",
             "supportsAllDrives": True,
             "includeItemsFromAllDrives": True,
         }

surfsense_backend/app/connectors/google_drive/client.py

@@ -47,7 +47,7 @@ class GoogleDriveClient:
     async def list_files(
         self,
         query: str = "",
-        fields: str = "nextPageToken, files(id, name, mimeType, modifiedTime, size, webViewLink, parents, owners, createdTime, description)",
+        fields: str = "nextPageToken, files(id, name, mimeType, modifiedTime, md5Checksum, size, webViewLink, parents, owners, createdTime, description)",
         page_size: int = 100,
         page_token: str | None = None,
     ) -> tuple[list[dict[str, Any]], str | None, str | None]:

surfsense_backend/app/connectors/google_drive/content_extractor.py

@@ -102,6 +102,8 @@ async def download_and_process_file(
             connector_info["metadata"]["file_size"] = file["size"]
         if "webViewLink" in file:
             connector_info["metadata"]["web_view_link"] = file["webViewLink"]
+        if "md5Checksum" in file:
+            connector_info["metadata"]["md5_checksum"] = file["md5Checksum"]
 
         if is_google_workspace_file(mime_type):
             connector_info["metadata"]["exported_as"] = "pdf"

surfsense_backend/app/connectors/google_drive/folder_manager.py

@@ -157,7 +157,7 @@ async def get_file_by_id(
     try:
         file, error = await client.get_file_metadata(
             file_id,
-            fields="id, name, mimeType, parents, createdTime, modifiedTime, size, webViewLink, iconLink",
+            fields="id, name, mimeType, parents, createdTime, modifiedTime, md5Checksum, size, webViewLink, iconLink",
         )
 
         if error:
@@ -228,7 +228,7 @@ async def list_folder_contents(
         while True:
             items, next_token, error = await client.list_files(
                 query=query,
-                fields="files(id, name, mimeType, parents, createdTime, modifiedTime, size, webViewLink, iconLink)",
+                fields="files(id, name, mimeType, parents, createdTime, modifiedTime, md5Checksum, size, webViewLink, iconLink)",
                 page_size=1000,  # Max allowed by Google Drive API
                 page_token=page_token,
             )

surfsense_backend/app/db.py

@@ -667,6 +667,12 @@ class SearchSpace(BaseModel, TimestampMixin):
         order_by="Log.id",
         cascade="all, delete-orphan",
     )
+    notifications = relationship(
+        "Notification",
+        back_populates="search_space",
+        order_by="Notification.created_at.desc()",
+        cascade="all, delete-orphan",
+    )
     search_source_connectors = relationship(
         "SearchSourceConnector",
         back_populates="search_space",
@@ -805,6 +811,39 @@ class Log(BaseModel, TimestampMixin):
     search_space = relationship("SearchSpace", back_populates="logs")
 
 
+class Notification(BaseModel, TimestampMixin):
+    __tablename__ = "notifications"
+
+    user_id = Column(
+        UUID(as_uuid=True),
+        ForeignKey("user.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    search_space_id = Column(
+        Integer, ForeignKey("searchspaces.id", ondelete="CASCADE"), nullable=True
+    )
+    type = Column(
+        String(50), nullable=False
+    )  # 'connector_indexing', 'document_processing', etc.
+    title = Column(String(200), nullable=False)
+    message = Column(Text, nullable=False)
+    read = Column(
+        Boolean, nullable=False, default=False, server_default=text("false"), index=True
+    )
+    notification_metadata = Column("metadata", JSONB, nullable=True, default={})
+    updated_at = Column(
+        TIMESTAMP(timezone=True),
+        nullable=True,
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
+        index=True,
+    )
+
+    user = relationship("User", back_populates="notifications")
+    search_space = relationship("SearchSpace", back_populates="notifications")
+
+
 class SearchSpaceRole(BaseModel, TimestampMixin):
     """
     Custom roles that can be defined per search space.
@@ -949,6 +988,12 @@ if config.AUTH_TYPE == "GOOGLE":
             "OAuthAccount", lazy="joined"
         )
         search_spaces = relationship("SearchSpace", back_populates="user")
+        notifications = relationship(
+            "Notification",
+            back_populates="user",
+            order_by="Notification.created_at.desc()",
+            cascade="all, delete-orphan",
+        )
 
         # RBAC relationships
         search_space_memberships = relationship(
@@ -986,6 +1031,12 @@ else:
 
     class User(SQLAlchemyBaseUserTableUUID, Base):
         search_spaces = relationship("SearchSpace", back_populates="user")
+        notifications = relationship(
+            "Notification",
+            back_populates="user",
+            order_by="Notification.created_at.desc()",
+            cascade="all, delete-orphan",
+        )
 
         # RBAC relationships
         search_space_memberships = relationship(
@@ -1049,11 +1100,36 @@ async def setup_indexes():
                 "CREATE INDEX IF NOT EXISTS chucks_search_index ON chunks USING gin (to_tsvector('english', content))"
             )
         )
+        # pg_trgm indexes for efficient ILIKE '%term%' searches on titles
+        # Critical for document mention picker (@mentions) to scale
+        await conn.execute(
+            text(
+                "CREATE INDEX IF NOT EXISTS idx_documents_title_trgm ON documents USING gin (title gin_trgm_ops)"
+            )
+        )
+        # B-tree index on search_space_id for fast filtering
+        await conn.execute(
+            text(
+                "CREATE INDEX IF NOT EXISTS idx_documents_search_space_id ON documents (search_space_id)"
+            )
+        )
+        # Covering index for "recent documents" query - enables index-only scan
+        await conn.execute(
+            text(
+                "CREATE INDEX IF NOT EXISTS idx_documents_search_space_updated ON documents (search_space_id, updated_at DESC NULLS LAST) INCLUDE (id, title, document_type)"
+            )
+        )
+        await conn.execute(
+            text(
+                "CREATE INDEX IF NOT EXISTS idx_surfsense_docs_title_trgm ON surfsense_docs_documents USING gin (title gin_trgm_ops)"
+            )
+        )
 
 
 async def create_db_and_tables():
     async with engine.begin() as conn:
         await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
+        await conn.execute(text("CREATE EXTENSION IF NOT EXISTS pg_trgm"))
         await conn.run_sync(Base.metadata.create_all)
     await setup_indexes()
 

surfsense_backend/app/routes/__init__.py

@@ -26,6 +26,7 @@ from .luma_add_connector_route import router as luma_add_connector_router
 from .new_chat_routes import router as new_chat_router
 from .new_llm_config_routes import router as new_llm_config_router
 from .notes_routes import router as notes_router
+from .notifications_routes import router as notifications_router
 from .notion_add_connector_route import router as notion_add_connector_router
 from .podcasts_routes import router as podcasts_router
 from .rbac_routes import router as rbac_router
@@ -63,3 +64,4 @@ router.include_router(new_llm_config_router)  # LLM configs with prompt configur
 router.include_router(logs_router)
 router.include_router(circleback_webhook_router)  # Circleback meeting webhooks
 router.include_router(surfsense_docs_router)  # Surfsense documentation for citations
+router.include_router(notifications_router)  # Notifications with Electric SQL sync

surfsense_backend/app/routes/documents_routes.py

@@ -19,6 +19,8 @@ from app.db import (
 from app.schemas import (
     DocumentRead,
     DocumentsCreate,
+    DocumentTitleRead,
+    DocumentTitleSearchResponse,
     DocumentUpdate,
     DocumentWithChunksRead,
     PaginatedResponse,
@@ -429,6 +431,112 @@ async def search_documents(
         ) from e
 
 
+@router.get("/documents/search/titles", response_model=DocumentTitleSearchResponse)
+async def search_document_titles(
+    search_space_id: int,
+    title: str = "",
+    page: int = 0,
+    page_size: int = 20,
+    session: AsyncSession = Depends(get_async_session),
+    user: User = Depends(current_active_user),
+):
+    """
+    Lightweight document title search optimized for mention picker (@mentions).
+
+    Returns only id, title, and document_type - no content or metadata.
+    Uses pg_trgm fuzzy search with similarity scoring for typo tolerance.
+    Results are ordered by relevance using trigram similarity scores.
+
+    Args:
+        search_space_id: The search space to search in. Required.
+        title: Search query (case-insensitive). If empty or < 2 chars, returns recent documents.
+        page: Zero-based page index. Default: 0.
+        page_size: Number of items per page. Default: 20.
+        session: Database session (injected).
+        user: Current authenticated user (injected).
+
+    Returns:
+        DocumentTitleSearchResponse: Lightweight list with has_more flag (no total count).
+    """
+    from sqlalchemy import desc, func, or_
+
+    try:
+        # Check permission for the search space
+        await check_permission(
+            session,
+            user,
+            search_space_id,
+            Permission.DOCUMENTS_READ.value,
+            "You don't have permission to read documents in this search space",
+        )
+
+        # Base query - only select lightweight fields
+        query = select(
+            Document.id,
+            Document.title,
+            Document.document_type,
+        ).filter(Document.search_space_id == search_space_id)
+
+        # If query is too short, return recent documents ordered by updated_at
+        if len(title.strip()) < 2:
+            query = query.order_by(Document.updated_at.desc().nullslast())
+        else:
+            # Fuzzy search using pg_trgm similarity + ILIKE fallback
+            search_term = title.strip()
+
+            # Similarity threshold for fuzzy matching (0.3 = ~30% trigram overlap)
+            # Lower values = more fuzzy, higher values = stricter matching
+            similarity_threshold = 0.3
+
+            # Match documents that either:
+            # 1. Have high trigram similarity (fuzzy match - handles typos)
+            # 2. Contain the exact substring (ILIKE - handles partial matches)
+            query = query.filter(
+                or_(
+                    func.similarity(Document.title, search_term) > similarity_threshold,
+                    Document.title.ilike(f"%{search_term}%"),
+                )
+            )
+
+            # Order by similarity score (descending) for best relevance ranking
+            # Higher similarity = better match = appears first
+            query = query.order_by(
+                desc(func.similarity(Document.title, search_term)),
+                Document.title,  # Alphabetical tiebreaker
+            )
+
+        # Fetch page_size + 1 to determine has_more without COUNT query
+        offset = page * page_size
+        result = await session.execute(query.offset(offset).limit(page_size + 1))
+        rows = result.all()
+
+        # Check if there are more results
+        has_more = len(rows) > page_size
+        items = rows[:page_size]  # Only return requested page_size
+
+        # Convert to response format
+        api_documents = [
+            DocumentTitleRead(
+                id=row.id,
+                title=row.title,
+                document_type=row.document_type,
+            )
+            for row in items
+        ]
+
+        return DocumentTitleSearchResponse(
+            items=api_documents,
+            has_more=has_more,
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to search document titles: {e!s}"
+        ) from e
+
+
 @router.get("/documents/type-counts")
 async def get_document_type_counts(
     search_space_id: int | None = None,

surfsense_backend/app/routes/notifications_routes.py

@@ -0,0 +1,102 @@
+"""
+Notifications API routes.
+These endpoints allow marking notifications as read.
+Electric SQL automatically syncs the changes to all connected clients.
+"""
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db import Notification, User, get_async_session
+from app.users import current_active_user
+
+router = APIRouter(prefix="/notifications", tags=["notifications"])
+
+
+class MarkReadResponse(BaseModel):
+    """Response for mark as read operations."""
+
+    success: bool
+    message: str
+
+
+class MarkAllReadResponse(BaseModel):
+    """Response for mark all as read operation."""
+
+    success: bool
+    message: str
+    updated_count: int
+
+
+@router.patch("/{notification_id}/read", response_model=MarkReadResponse)
+async def mark_notification_as_read(
+    notification_id: int,
+    user: User = Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+) -> MarkReadResponse:
+    """
+    Mark a single notification as read.
+
+    Electric SQL will automatically sync this change to all connected clients.
+    """
+    # Verify the notification belongs to the user
+    result = await session.execute(
+        select(Notification).where(
+            Notification.id == notification_id,
+            Notification.user_id == user.id,
+        )
+    )
+    notification = result.scalar_one_or_none()
+
+    if not notification:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Notification not found",
+        )
+
+    if notification.read:
+        return MarkReadResponse(
+            success=True,
+            message="Notification already marked as read",
+        )
+
+    # Update the notification
+    notification.read = True
+    await session.commit()
+
+    return MarkReadResponse(
+        success=True,
+        message="Notification marked as read",
+    )
+
+
+@router.patch("/read-all", response_model=MarkAllReadResponse)
+async def mark_all_notifications_as_read(
+    user: User = Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+) -> MarkAllReadResponse:
+    """
+    Mark all notifications as read for the current user.
+
+    Electric SQL will automatically sync these changes to all connected clients.
+    """
+    # Update all unread notifications for the user
+    result = await session.execute(
+        update(Notification)
+        .where(
+            Notification.user_id == user.id,
+            Notification.read == False,  # noqa: E712
+        )
+        .values(read=True)
+    )
+    await session.commit()
+
+    updated_count = result.rowcount
+
+    return MarkAllReadResponse(
+        success=True,
+        message=f"Marked {updated_count} notification(s) as read",
+        updated_count=updated_count,
+    )

surfsense_backend/app/schemas/__init__.py

@@ -4,13 +4,15 @@ from .documents import (
     DocumentBase,
     DocumentRead,
     DocumentsCreate,
+    DocumentTitleRead,
+    DocumentTitleSearchResponse,
     DocumentUpdate,
     DocumentWithChunksRead,
     ExtensionDocumentContent,
     ExtensionDocumentMetadata,
     PaginatedResponse,
 )
-from .google_drive import DriveItem, GoogleDriveIndexRequest
+from .google_drive import DriveItem, GoogleDriveIndexingOptions, GoogleDriveIndexRequest
 from .logs import LogBase, LogCreate, LogFilter, LogRead, LogUpdate
 from .new_chat import (
     ChatMessage,
@@ -85,6 +87,8 @@ __all__ = [
     # Document schemas
     "DocumentBase",
     "DocumentRead",
+    "DocumentTitleRead",
+    "DocumentTitleSearchResponse",
     "DocumentUpdate",
     "DocumentWithChunksRead",
     "DocumentsCreate",
@@ -94,6 +98,7 @@ __all__ = [
     "ExtensionDocumentMetadata",
     "GlobalNewLLMConfigRead",
     "GoogleDriveIndexRequest",
+    "GoogleDriveIndexingOptions",
     # Base schemas
     "IDModel",
     # RBAC schemas

surfsense_backend/app/schemas/documents.py

@@ -67,3 +67,20 @@ class PaginatedResponse[T](BaseModel):
     page: int
     page_size: int
     has_more: bool
+
+
+class DocumentTitleRead(BaseModel):
+    """Lightweight document response for mention picker - only essential fields."""
+
+    id: int
+    title: str
+    document_type: DocumentType
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class DocumentTitleSearchResponse(BaseModel):
+    """Response for document title search - optimized for typeahead."""
+
+    items: list[DocumentTitleRead]
+    has_more: bool

surfsense_backend/app/schemas/google_drive.py

@@ -10,6 +10,25 @@ class DriveItem(BaseModel):
     name: str = Field(..., description="Item display name")
 
 
+class GoogleDriveIndexingOptions(BaseModel):
+    """Indexing options for Google Drive connector."""
+
+    max_files_per_folder: int = Field(
+        default=100,
+        ge=1,
+        le=1000,
+        description="Maximum number of files to index from each folder (1-1000)",
+    )
+    incremental_sync: bool = Field(
+        default=True,
+        description="Only sync changes since last index (faster). Disable for a full re-index.",
+    )
+    include_subfolders: bool = Field(
+        default=True,
+        description="Recursively index files in subfolders of selected folders",
+    )
+
+
 class GoogleDriveIndexRequest(BaseModel):
     """Request body for indexing Google Drive content."""
 
@@ -19,6 +38,10 @@ class GoogleDriveIndexRequest(BaseModel):
     files: list[DriveItem] = Field(
         default_factory=list, description="List of specific files to index"
     )
+    indexing_options: GoogleDriveIndexingOptions = Field(
+        default_factory=GoogleDriveIndexingOptions,
+        description="Indexing configuration options",
+    )
 
     def has_items(self) -> bool:
         """Check if any items are selected."""

surfsense_backend/app/schemas/search_source_connector.py

@@ -95,7 +95,7 @@ class MCPConnectorCreate(BaseModel):
     """Schema for creating an MCP connector."""
 
     name: str
-    server_config: MCPServerConfig
+    server_config: MCPServerConfig  # Single MCP server configuration
 
 
 class MCPConnectorUpdate(BaseModel):
@@ -106,7 +106,7 @@ class MCPConnectorUpdate(BaseModel):
 
 
 class MCPConnectorRead(BaseModel):
-    """Schema for reading an MCP connector with server config."""
+    """Schema for reading an MCP connector with server configs."""
 
     id: int
     name: str
@@ -123,7 +123,8 @@ class MCPConnectorRead(BaseModel):
     def from_connector(cls, connector: SearchSourceConnectorRead) -> "MCPConnectorRead":
         """Convert from base SearchSourceConnectorRead."""
         config = connector.config or {}
-        server_config = MCPServerConfig(**config.get("server_config", {}))
+        server_config_data = config.get("server_config", {})
+        server_config = MCPServerConfig(**server_config_data)
 
         return cls(
             id=connector.id,

surfsense_backend/app/services/notification_service.py

@@ -0,0 +1,664 @@
+"""Service for creating and managing notifications with Electric SQL sync."""
+
+import logging
+from datetime import UTC, datetime
+from typing import Any
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm.attributes import flag_modified
+
+from app.db import Notification
+
+logger = logging.getLogger(__name__)
+
+
+class BaseNotificationHandler:
+    """Base class for notification handlers - provides common functionality."""
+
+    def __init__(self, notification_type: str):
+        """
+        Initialize the notification handler.
+
+        Args:
+            notification_type: Type of notification (e.g., 'connector_indexing', 'document_processing')
+        """
+        self.notification_type = notification_type
+
+    async def find_notification_by_operation(
+        self,
+        session: AsyncSession,
+        user_id: UUID,
+        operation_id: str,
+        search_space_id: int | None = None,
+    ) -> Notification | None:
+        """
+        Find an existing notification by operation ID.
+
+        Args:
+            session: Database session
+            user_id: User ID
+            operation_id: Unique operation identifier
+            search_space_id: Optional search space ID
+
+        Returns:
+            Notification if found, None otherwise
+        """
+        query = select(Notification).where(
+            Notification.user_id == user_id,
+            Notification.type == self.notification_type,
+            Notification.notification_metadata["operation_id"].astext == operation_id,
+        )
+        if search_space_id is not None:
+            query = query.where(Notification.search_space_id == search_space_id)
+
+        result = await session.execute(query)
+        return result.scalar_one_or_none()
+
+    async def find_or_create_notification(
+        self,
+        session: AsyncSession,
+        user_id: UUID,
+        operation_id: str,
+        title: str,
+        message: str,
+        search_space_id: int | None = None,
+        initial_metadata: dict[str, Any] | None = None,
+    ) -> Notification:
+        """
+        Find an existing notification or create a new one.
+
+        Args:
+            session: Database session
+            user_id: User ID
+            operation_id: Unique operation identifier
+            title: Notification title
+            message: Notification message
+            search_space_id: Optional search space ID
+            initial_metadata: Initial metadata dictionary
+
+        Returns:
+            Notification: The found or created notification
+        """
+        # Try to find existing notification
+        notification = await self.find_notification_by_operation(
+            session, user_id, operation_id, search_space_id
+        )
+
+        if notification:
+            # Update existing notification
+            notification.title = title
+            notification.message = message
+            if initial_metadata:
+                notification.notification_metadata = {
+                    **notification.notification_metadata,
+                    **initial_metadata,
+                }
+                # Mark JSONB column as modified so SQLAlchemy detects the change
+                flag_modified(notification, "notification_metadata")
+            await session.commit()
+            await session.refresh(notification)
+            logger.info(
+                f"Updated notification {notification.id} for operation {operation_id}"
+            )
+            return notification
+
+        # Create new notification
+        metadata = initial_metadata or {}
+        metadata["operation_id"] = operation_id
+        metadata["status"] = "in_progress"
+        metadata["started_at"] = datetime.now(UTC).isoformat()
+
+        notification = Notification(
+            user_id=user_id,
+            search_space_id=search_space_id,
+            type=self.notification_type,
+            title=title,
+            message=message,
+            notification_metadata=metadata,
+        )
+        session.add(notification)
+        await session.commit()
+        await session.refresh(notification)
+        logger.info(
+            f"Created notification {notification.id} for operation {operation_id}"
+        )
+        return notification
+
+    async def update_notification(
+        self,
+        session: AsyncSession,
+        notification: Notification,
+        title: str | None = None,
+        message: str | None = None,
+        status: str | None = None,
+        metadata_updates: dict[str, Any] | None = None,
+    ) -> Notification:
+        """
+        Update an existing notification.
+
+        Args:
+            session: Database session
+            notification: Notification to update
+            title: New title (optional)
+            message: New message (optional)
+            status: New status (optional)
+            metadata_updates: Additional metadata to merge (optional)
+
+        Returns:
+            Updated notification
+        """
+        if title is not None:
+            notification.title = title
+        if message is not None:
+            notification.message = message
+
+        if status is not None:
+            notification.notification_metadata["status"] = status
+            if status in ("completed", "failed"):
+                notification.notification_metadata["completed_at"] = datetime.now(
+                    UTC
+                ).isoformat()
+            # Mark JSONB column as modified so SQLAlchemy detects the change
+            flag_modified(notification, "notification_metadata")
+
+        if metadata_updates:
+            notification.notification_metadata = {
+                **notification.notification_metadata,
+                **metadata_updates,
+            }
+            # Mark JSONB column as modified
+            flag_modified(notification, "notification_metadata")
+
+        await session.commit()
+        await session.refresh(notification)
+        logger.info(f"Updated notification {notification.id}")
+        return notification
+
+
+class ConnectorIndexingNotificationHandler(BaseNotificationHandler):
+    """Handler for connector indexing notifications."""
+
+    def __init__(self):
+        super().__init__("connector_indexing")
+
+    def _generate_operation_id(
+        self,
+        connector_id: int,
+        start_date: str | None = None,
+        end_date: str | None = None,
+    ) -> str:
+        """
+        Generate a unique operation ID for a connector indexing operation.
+
+        Args:
+            connector_id: Connector ID
+            start_date: Start date (optional)
+            end_date: End date (optional)
+
+        Returns:
+            Unique operation ID string
+        """
+        timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
+        date_range = ""
+        if start_date or end_date:
+            date_range = f"_{start_date or 'none'}_{end_date or 'none'}"
+        return f"connector_{connector_id}_{timestamp}{date_range}"
+
+    def _generate_google_drive_operation_id(
+        self, connector_id: int, folder_count: int, file_count: int
+    ) -> str:
+        """
+        Generate a unique operation ID for a Google Drive indexing operation.
+
+        Args:
+            connector_id: Connector ID
+            folder_count: Number of folders to index
+            file_count: Number of files to index
+
+        Returns:
+            Unique operation ID string
+        """
+        timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
+        items_info = f"_{folder_count}f_{file_count}files"
+        return f"drive_{connector_id}_{timestamp}{items_info}"
+
+    async def notify_indexing_started(
+        self,
+        session: AsyncSession,
+        user_id: UUID,
+        connector_id: int,
+        connector_name: str,
+        connector_type: str,
+        search_space_id: int,
+        start_date: str | None = None,
+        end_date: str | None = None,
+    ) -> Notification:
+        """
+        Create or update notification when connector indexing starts.
+
+        Args:
+            session: Database session
+            user_id: User ID
+            connector_id: Connector ID
+            connector_name: Connector name
+            connector_type: Connector type
+            search_space_id: Search space ID
+            start_date: Start date for indexing
+            end_date: End date for indexing
+
+        Returns:
+            Notification: The created or updated notification
+        """
+        operation_id = self._generate_operation_id(connector_id, start_date, end_date)
+        title = f"Syncing: {connector_name}"
+        message = "Connecting to your account"
+
+        metadata = {
+            "connector_id": connector_id,
+            "connector_name": connector_name,
+            "connector_type": connector_type,
+            "start_date": start_date,
+            "end_date": end_date,
+            "indexed_count": 0,
+            "sync_stage": "connecting",
+        }
+
+        return await self.find_or_create_notification(
+            session=session,
+            user_id=user_id,
+            operation_id=operation_id,
+            title=title,
+            message=message,
+            search_space_id=search_space_id,
+            initial_metadata=metadata,
+        )
+
+    async def notify_indexing_progress(
+        self,
+        session: AsyncSession,
+        notification: Notification,
+        indexed_count: int,
+        total_count: int | None = None,
+        stage: str | None = None,
+        stage_message: str | None = None,
+    ) -> Notification:
+        """
+        Update notification with indexing progress.
+
+        Args:
+            session: Database session
+            notification: Notification to update
+            indexed_count: Number of items indexed so far
+            total_count: Total number of items (optional)
+            stage: Current sync stage (fetching, processing, storing) (optional)
+            stage_message: Optional custom message for the stage
+
+        Returns:
+            Updated notification
+        """
+        # User-friendly stage messages (clean, no ellipsis - spinner shows activity)
+        stage_messages = {
+            "connecting": "Connecting to your account",
+            "fetching": "Fetching your content",
+            "processing": "Preparing for search",
+            "storing": "Almost done",
+        }
+
+        # Use stage-based message if stage provided, otherwise fallback
+        if stage or stage_message:
+            progress_msg = stage_message or stage_messages.get(stage, "Processing")
+        else:
+            # Fallback for backward compatibility
+            progress_msg = "Fetching your content"
+
+        metadata_updates = {"indexed_count": indexed_count}
+        if total_count is not None:
+            metadata_updates["total_count"] = total_count
+            progress_percent = int((indexed_count / total_count) * 100)
+            metadata_updates["progress_percent"] = progress_percent
+        if stage:
+            metadata_updates["sync_stage"] = stage
+
+        return await self.update_notification(
+            session=session,
+            notification=notification,
+            message=progress_msg,
+            status="in_progress",
+            metadata_updates=metadata_updates,
+        )
+
+    async def notify_indexing_completed(
+        self,
+        session: AsyncSession,
+        notification: Notification,
+        indexed_count: int,
+        error_message: str | None = None,
+    ) -> Notification:
+        """
+        Update notification when connector indexing completes.
+
+        Args:
+            session: Database session
+            notification: Notification to update
+            indexed_count: Total number of items indexed
+            error_message: Error message if indexing failed (optional)
+
+        Returns:
+            Updated notification
+        """
+        connector_name = notification.notification_metadata.get(
+            "connector_name", "Connector"
+        )
+
+        if error_message:
+            title = f"Failed: {connector_name}"
+            message = f"Sync failed: {error_message}"
+            status = "failed"
+        else:
+            title = f"Ready: {connector_name}"
+            if indexed_count == 0:
+                message = "Already up to date! No new items to sync."
+            else:
+                item_text = "item" if indexed_count == 1 else "items"
+                message = f"Now searchable! {indexed_count} {item_text} synced."
+            status = "completed"
+
+        metadata_updates = {
+            "indexed_count": indexed_count,
+            "sync_stage": "completed" if not error_message else "failed",
+            "error_message": error_message,
+        }
+
+        return await self.update_notification(
+            session=session,
+            notification=notification,
+            title=title,
+            message=message,
+            status=status,
+            metadata_updates=metadata_updates,
+        )
+
+    async def notify_google_drive_indexing_started(
+        self,
+        session: AsyncSession,
+        user_id: UUID,
+        connector_id: int,
+        connector_name: str,
+        connector_type: str,
+        search_space_id: int,
+        folder_count: int,
+        file_count: int,
+        folder_names: list[str] | None = None,
+        file_names: list[str] | None = None,
+    ) -> Notification:
+        """
+        Create or update notification when Google Drive indexing starts.
+
+        Args:
+            session: Database session
+            user_id: User ID
+            connector_id: Connector ID
+            connector_name: Connector name
+            connector_type: Connector type
+            search_space_id: Search space ID
+            folder_count: Number of folders to index
+            file_count: Number of files to index
+            folder_names: List of folder names (optional)
+            file_names: List of file names (optional)
+
+        Returns:
+            Notification: The created or updated notification
+        """
+        operation_id = self._generate_google_drive_operation_id(
+            connector_id, folder_count, file_count
+        )
+        title = f"Syncing: {connector_name}"
+        message = "Preparing your files"
+
+        metadata = {
+            "connector_id": connector_id,
+            "connector_name": connector_name,
+            "connector_type": connector_type,
+            "folder_count": folder_count,
+            "file_count": file_count,
+            "indexed_count": 0,
+            "sync_stage": "connecting",
+        }
+
+        if folder_names:
+            metadata["folder_names"] = folder_names
+        if file_names:
+            metadata["file_names"] = file_names
+
+        return await self.find_or_create_notification(
+            session=session,
+            user_id=user_id,
+            operation_id=operation_id,
+            title=title,
+            message=message,
+            search_space_id=search_space_id,
+            initial_metadata=metadata,
+        )
+
+
+class DocumentProcessingNotificationHandler(BaseNotificationHandler):
+    """Handler for document processing notifications."""
+
+    def __init__(self):
+        super().__init__("document_processing")
+
+    def _generate_operation_id(
+        self, document_type: str, filename: str, search_space_id: int
+    ) -> str:
+        """
+        Generate a unique operation ID for a document processing operation.
+
+        Args:
+            document_type: Type of document (FILE, YOUTUBE_VIDEO, CRAWLED_URL, etc.)
+            filename: Name of the file/document
+            search_space_id: Search space ID
+
+        Returns:
+            Unique operation ID string
+        """
+        timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S_%f")
+        # Create a short hash of filename to ensure uniqueness
+        import hashlib
+
+        filename_hash = hashlib.md5(filename.encode()).hexdigest()[:8]
+        return f"doc_{document_type}_{search_space_id}_{timestamp}_{filename_hash}"
+
+    async def notify_processing_started(
+        self,
+        session: AsyncSession,
+        user_id: UUID,
+        document_type: str,
+        document_name: str,
+        search_space_id: int,
+        file_size: int | None = None,
+    ) -> Notification:
+        """
+        Create notification when document processing starts.
+
+        Args:
+            session: Database session
+            user_id: User ID
+            document_type: Type of document (FILE, YOUTUBE_VIDEO, CRAWLED_URL, etc.)
+            document_name: Name/title of the document
+            search_space_id: Search space ID
+            file_size: Size of file in bytes (optional)
+
+        Returns:
+            Notification: The created notification
+        """
+        operation_id = self._generate_operation_id(
+            document_type, document_name, search_space_id
+        )
+        title = f"Processing: {document_name}"
+        message = "Waiting in queue"
+
+        metadata = {
+            "document_type": document_type,
+            "document_name": document_name,
+            "processing_stage": "queued",
+        }
+
+        if file_size is not None:
+            metadata["file_size"] = file_size
+
+        return await self.find_or_create_notification(
+            session=session,
+            user_id=user_id,
+            operation_id=operation_id,
+            title=title,
+            message=message,
+            search_space_id=search_space_id,
+            initial_metadata=metadata,
+        )
+
+    async def notify_processing_progress(
+        self,
+        session: AsyncSession,
+        notification: Notification,
+        stage: str,
+        stage_message: str | None = None,
+        chunks_count: int | None = None,
+    ) -> Notification:
+        """
+        Update notification with processing progress.
+
+        Args:
+            session: Database session
+            notification: Notification to update
+            stage: Current processing stage (parsing, chunking, embedding, storing)
+            stage_message: Optional custom message for the stage
+            chunks_count: Number of chunks created (optional, stored in metadata only)
+
+        Returns:
+            Updated notification
+        """
+        # User-friendly stage messages
+        stage_messages = {
+            "parsing": "Reading your file",
+            "chunking": "Preparing for search",
+            "embedding": "Preparing for search",
+            "storing": "Finalizing",
+        }
+
+        message = stage_message or stage_messages.get(stage, "Processing")
+
+        metadata_updates = {"processing_stage": stage}
+        # Store chunks_count in metadata for debugging, but don't show to user
+        if chunks_count is not None:
+            metadata_updates["chunks_count"] = chunks_count
+
+        return await self.update_notification(
+            session=session,
+            notification=notification,
+            message=message,
+            status="in_progress",
+            metadata_updates=metadata_updates,
+        )
+
+    async def notify_processing_completed(
+        self,
+        session: AsyncSession,
+        notification: Notification,
+        document_id: int | None = None,
+        chunks_count: int | None = None,
+        error_message: str | None = None,
+    ) -> Notification:
+        """
+        Update notification when document processing completes.
+
+        Args:
+            session: Database session
+            notification: Notification to update
+            document_id: ID of the created document (optional)
+            chunks_count: Total number of chunks created (optional)
+            error_message: Error message if processing failed (optional)
+
+        Returns:
+            Updated notification
+        """
+        document_name = notification.notification_metadata.get(
+            "document_name", "Document"
+        )
+
+        if error_message:
+            title = f"Failed: {document_name}"
+            message = f"Processing failed: {error_message}"
+            status = "failed"
+        else:
+            title = f"Ready: {document_name}"
+            message = "Now searchable!"
+            status = "completed"
+
+        metadata_updates = {
+            "processing_stage": "completed" if not error_message else "failed",
+            "error_message": error_message,
+        }
+
+        if document_id is not None:
+            metadata_updates["document_id"] = document_id
+        # Store chunks_count in metadata for debugging, but don't show to user
+        if chunks_count is not None:
+            metadata_updates["chunks_count"] = chunks_count
+
+        return await self.update_notification(
+            session=session,
+            notification=notification,
+            title=title,
+            message=message,
+            status=status,
+            metadata_updates=metadata_updates,
+        )
+
+
+class NotificationService:
+    """Service for creating and managing notifications that sync via Electric SQL."""
+
+    # Handler instances
+    connector_indexing = ConnectorIndexingNotificationHandler()
+    document_processing = DocumentProcessingNotificationHandler()
+
+    @staticmethod
+    async def create_notification(
+        session: AsyncSession,
+        user_id: UUID,
+        notification_type: str,
+        title: str,
+        message: str,
+        search_space_id: int | None = None,
+        notification_metadata: dict[str, Any] | None = None,
+    ) -> Notification:
+        """
+        Create a notification - Electric SQL will automatically sync it to frontend.
+
+        Args:
+            session: Database session
+            user_id: User to notify
+            notification_type: Type of notification (e.g., 'document_processing', 'connector_indexing')
+            title: Notification title
+            message: Notification message
+            search_space_id: Optional search space ID
+            notification_metadata: Optional metadata dictionary
+
+        Returns:
+            Notification: The created notification
+        """
+        notification = Notification(
+            user_id=user_id,
+            search_space_id=search_space_id,
+            type=notification_type,
+            title=title,
+            message=message,
+            notification_metadata=notification_metadata or {},
+        )
+        session.add(notification)
+        await session.commit()
+        await session.refresh(notification)
+        logger.info(f"Created notification {notification.id} for user {user_id}")
+        return notification

surfsense_backend/app/tasks/celery_tasks/connector_tasks.py

@@ -445,31 +445,13 @@ async def _index_google_gmail_messages(
     end_date: str,
 ):
     """Index Google Gmail messages with new session."""
-    from datetime import datetime
-
     from app.routes.search_source_connectors_routes import (
         run_google_gmail_indexing,
     )
 
-    # Parse dates to calculate days_back
-    max_messages = 100
-    days_back = 30  # Default
-
-    if start_date:
-        try:
-            # Parse start_date (format: YYYY-MM-DD)
-            start_dt = datetime.strptime(start_date, "%Y-%m-%d")
-            # Calculate days back from now
-            days_back = (datetime.now() - start_dt).days
-            # Ensure at least 1 day
-            days_back = max(1, days_back)
-        except ValueError:
-            # If parsing fails, use default
-            days_back = 30
-
     async with get_celery_session_maker()() as session:
         await run_google_gmail_indexing(
-            session, connector_id, search_space_id, user_id, max_messages, days_back
+            session, connector_id, search_space_id, user_id, start_date, end_date
         )
 
 
@@ -479,7 +461,7 @@ def index_google_drive_files_task(
     connector_id: int,
     search_space_id: int,
     user_id: str,
-    items_dict: dict,  # Dictionary with 'folders' and 'files' lists
+    items_dict: dict,  # Dictionary with 'folders', 'files', and 'indexing_options'
 ):
     """Celery task to index Google Drive folders and files."""
     import asyncio
@@ -504,7 +486,7 @@ async def _index_google_drive_files(
     connector_id: int,
     search_space_id: int,
     user_id: str,
-    items_dict: dict,  # Dictionary with 'folders' and 'files' lists
+    items_dict: dict,  # Dictionary with 'folders', 'files', and 'indexing_options'
 ):
     """Index Google Drive folders and files with new session."""
     from app.routes.search_source_connectors_routes import (

surfsense_backend/app/tasks/celery_tasks/document_tasks.py

@@ -1,12 +1,14 @@
 """Celery tasks for document processing."""
 
 import logging
+from uuid import UUID
 
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlalchemy.pool import NullPool
 
 from app.celery_app import celery_app
 from app.config import config
+from app.services.notification_service import NotificationService
 from app.services.task_logging_service import TaskLoggingService
 from app.tasks.document_processors import (
     add_extension_received_document,
@@ -84,6 +86,22 @@ async def _process_extension_document(
     async with get_celery_session_maker()() as session:
         task_logger = TaskLoggingService(session, search_space_id)
 
+        # Truncate title for notification display
+        page_title = individual_document.metadata.VisitedWebPageTitle[:50]
+        if len(individual_document.metadata.VisitedWebPageTitle) > 50:
+            page_title += "..."
+
+        # Create notification for document processing
+        notification = (
+            await NotificationService.document_processing.notify_processing_started(
+                session=session,
+                user_id=UUID(user_id),
+                document_type="EXTENSION",
+                document_name=page_title,
+                search_space_id=search_space_id,
+            )
+        )
+
         log_entry = await task_logger.log_task_start(
             task_name="process_extension_document",
             source="document_processor",
@@ -97,6 +115,14 @@ async def _process_extension_document(
         )
 
         try:
+            # Update notification: parsing stage
+            await NotificationService.document_processing.notify_processing_progress(
+                session,
+                notification,
+                stage="parsing",
+                stage_message="Reading page content",
+            )
+
             result = await add_extension_received_document(
                 session, individual_document, search_space_id, user_id
             )
@@ -107,12 +133,31 @@ async def _process_extension_document(
                     f"Successfully processed extension document: {individual_document.metadata.VisitedWebPageTitle}",
                     {"document_id": result.id, "content_hash": result.content_hash},
                 )
+
+                # Update notification on success
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        document_id=result.id,
+                        chunks_count=None,
+                    )
+                )
             else:
                 await task_logger.log_task_success(
                     log_entry,
                     f"Extension document already exists (duplicate): {individual_document.metadata.VisitedWebPageTitle}",
                     {"duplicate_detected": True},
                 )
+
+                # Update notification for duplicate
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        error_message="Page already saved (duplicate)",
+                    )
+                )
         except Exception as e:
             await task_logger.log_task_failure(
                 log_entry,
@@ -120,6 +165,23 @@ async def _process_extension_document(
                 str(e),
                 {"error_type": type(e).__name__},
             )
+
+            # Update notification on failure - wrapped in try-except to ensure it doesn't fail silently
+            try:
+                # Refresh notification to ensure it's not stale after any rollback
+                await session.refresh(notification)
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        error_message=str(e)[:100],
+                    )
+                )
+            except Exception as notif_error:
+                logger.error(
+                    f"Failed to update notification on failure: {notif_error!s}"
+                )
+
             logger.error(f"Error processing extension document: {e!s}")
             raise
 
@@ -150,6 +212,20 @@ async def _process_youtube_video(url: str, search_space_id: int, user_id: str):
     async with get_celery_session_maker()() as session:
         task_logger = TaskLoggingService(session, search_space_id)
 
+        # Extract video title from URL for notification (will be updated later)
+        video_name = url.split("v=")[-1][:11] if "v=" in url else url
+
+        # Create notification for document processing
+        notification = (
+            await NotificationService.document_processing.notify_processing_started(
+                session=session,
+                user_id=UUID(user_id),
+                document_type="YOUTUBE_VIDEO",
+                document_name=f"YouTube: {video_name}",
+                search_space_id=search_space_id,
+            )
+        )
+
         log_entry = await task_logger.log_task_start(
             task_name="process_youtube_video",
             source="document_processor",
@@ -158,6 +234,14 @@ async def _process_youtube_video(url: str, search_space_id: int, user_id: str):
         )
 
         try:
+            # Update notification: parsing (fetching transcript)
+            await NotificationService.document_processing.notify_processing_progress(
+                session,
+                notification,
+                stage="parsing",
+                stage_message="Fetching video transcript",
+            )
+
             result = await add_youtube_video_document(
                 session, url, search_space_id, user_id
             )
@@ -172,12 +256,31 @@ async def _process_youtube_video(url: str, search_space_id: int, user_id: str):
                         "content_hash": result.content_hash,
                     },
                 )
+
+                # Update notification on success
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        document_id=result.id,
+                        chunks_count=None,
+                    )
+                )
             else:
                 await task_logger.log_task_success(
                     log_entry,
                     f"YouTube video document already exists (duplicate): {url}",
                     {"duplicate_detected": True},
                 )
+
+                # Update notification for duplicate
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        error_message="Video already exists (duplicate)",
+                    )
+                )
         except Exception as e:
             await task_logger.log_task_failure(
                 log_entry,
@@ -185,6 +288,23 @@ async def _process_youtube_video(url: str, search_space_id: int, user_id: str):
                 str(e),
                 {"error_type": type(e).__name__},
             )
+
+            # Update notification on failure - wrapped in try-except to ensure it doesn't fail silently
+            try:
+                # Refresh notification to ensure it's not stale after any rollback
+                await session.refresh(notification)
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        error_message=str(e)[:100],
+                    )
+                )
+            except Exception as notif_error:
+                logger.error(
+                    f"Failed to update notification on failure: {notif_error!s}"
+                )
+
             logger.error(f"Error processing YouTube video: {e!s}")
             raise
 
@@ -219,11 +339,31 @@ async def _process_file_upload(
     file_path: str, filename: str, search_space_id: int, user_id: str
 ):
     """Process file upload with new session."""
+    import os
+
     from app.tasks.document_processors.file_processors import process_file_in_background
 
     async with get_celery_session_maker()() as session:
         task_logger = TaskLoggingService(session, search_space_id)
 
+        # Get file size for notification metadata
+        try:
+            file_size = os.path.getsize(file_path)
+        except Exception:
+            file_size = None
+
+        # Create notification for document processing
+        notification = (
+            await NotificationService.document_processing.notify_processing_started(
+                session=session,
+                user_id=UUID(user_id),
+                document_type="FILE",
+                document_name=filename,
+                search_space_id=search_space_id,
+                file_size=file_size,
+            )
+        )
+
         log_entry = await task_logger.log_task_start(
             task_name="process_file_upload",
             source="document_processor",
@@ -237,7 +377,7 @@ async def _process_file_upload(
         )
 
         try:
-            await process_file_in_background(
+            result = await process_file_in_background(
                 file_path,
                 filename,
                 search_space_id,
@@ -245,7 +385,29 @@ async def _process_file_upload(
                 session,
                 task_logger,
                 log_entry,
+                notification=notification,
             )
+
+            # Update notification on success
+            if result:
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        document_id=result.id,
+                        chunks_count=None,
+                    )
+                )
+            else:
+                # Duplicate detected
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        error_message="Document already exists (duplicate)",
+                    )
+                )
+
         except Exception as e:
             # Import here to avoid circular dependencies
             from fastapi import HTTPException
@@ -258,7 +420,23 @@ async def _process_file_upload(
             elif isinstance(e, HTTPException) and "page limit" in str(e.detail).lower():
                 error_message = str(e.detail)
             else:
-                error_message = f"Failed to process file: {filename}"
+                error_message = str(e)[:100]
+
+            # Update notification on failure - wrapped in try-except to ensure it doesn't fail silently
+            try:
+                # Refresh notification to ensure it's not stale after any rollback
+                await session.refresh(notification)
+                await (
+                    NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        error_message=error_message,
+                    )
+                )
+            except Exception as notif_error:
+                logger.error(
+                    f"Failed to update notification on failure: {notif_error!s}"
+                )
 
             await task_logger.log_task_failure(
                 log_entry,
@@ -323,6 +501,22 @@ async def _process_circleback_meeting(
     async with get_celery_session_maker()() as session:
         task_logger = TaskLoggingService(session, search_space_id)
 
+        # Get user_id from metadata if available
+        user_id = metadata.get("user_id")
+
+        # Create notification if user_id is available
+        notification = None
+        if user_id:
+            notification = (
+                await NotificationService.document_processing.notify_processing_started(
+                    session=session,
+                    user_id=UUID(user_id),
+                    document_type="CIRCLEBACK",
+                    document_name=f"Meeting: {meeting_name[:40]}",
+                    search_space_id=search_space_id,
+                )
+            )
+
         log_entry = await task_logger.log_task_start(
             task_name="process_circleback_meeting",
             source="circleback_webhook",
@@ -336,6 +530,17 @@ async def _process_circleback_meeting(
         )
 
         try:
+            # Update notification: parsing stage
+            if notification:
+                await (
+                    NotificationService.document_processing.notify_processing_progress(
+                        session,
+                        notification,
+                        stage="parsing",
+                        stage_message="Reading meeting notes",
+                    )
+                )
+
             result = await add_circleback_meeting_document(
                 session=session,
                 meeting_id=meeting_id,
@@ -355,12 +560,29 @@ async def _process_circleback_meeting(
                         "content_hash": result.content_hash,
                     },
                 )
+
+                # Update notification on success
+                if notification:
+                    await NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        document_id=result.id,
+                        chunks_count=None,
+                    )
             else:
                 await task_logger.log_task_success(
                     log_entry,
                     f"Circleback meeting document already exists (duplicate): {meeting_name}",
                     {"duplicate_detected": True, "meeting_id": meeting_id},
                 )
+
+                # Update notification for duplicate
+                if notification:
+                    await NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        error_message="Meeting already saved (duplicate)",
+                    )
         except Exception as e:
             await task_logger.log_task_failure(
                 log_entry,
@@ -368,5 +590,21 @@ async def _process_circleback_meeting(
                 str(e),
                 {"error_type": type(e).__name__, "meeting_id": meeting_id},
             )
+
+            # Update notification on failure - wrapped in try-except to ensure it doesn't fail silently
+            if notification:
+                try:
+                    # Refresh notification to ensure it's not stale after any rollback
+                    await session.refresh(notification)
+                    await NotificationService.document_processing.notify_processing_completed(
+                        session=session,
+                        notification=notification,
+                        error_message=str(e)[:100],
+                    )
+                except Exception as notif_error:
+                    logger.error(
+                        f"Failed to update notification on failure: {notif_error!s}"
+                    )
+
             logger.error(f"Error processing Circleback meeting: {e!s}")
             raise

surfsense_backend/app/tasks/celery_tasks/schedule_checker_task.py

@@ -72,6 +72,7 @@ async def _check_and_trigger_schedules():
                 index_elasticsearch_documents_task,
                 index_github_repos_task,
                 index_google_calendar_events_task,
+                index_google_drive_files_task,
                 index_google_gmail_messages_task,
                 index_jira_issues_task,
                 index_linear_issues_task,
@@ -96,6 +97,7 @@ async def _check_and_trigger_schedules():
                 SearchSourceConnectorType.LUMA_CONNECTOR: index_luma_events_task,
                 SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR: index_elasticsearch_documents_task,
                 SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: index_crawled_urls_task,
+                SearchSourceConnectorType.GOOGLE_DRIVE_CONNECTOR: index_google_drive_files_task,
             }
 
             # Trigger indexing for each due connector
@@ -106,13 +108,57 @@ async def _check_and_trigger_schedules():
                         f"Triggering periodic indexing for connector {connector.id} "
                         f"({connector.connector_type.value})"
                     )
-                    task.delay(
-                        connector.id,
-                        connector.search_space_id,
-                        str(connector.user_id),
-                        None,  # start_date - uses last_indexed_at
-                        None,  # end_date - uses now
-                    )
+
+                    # Special handling for Google Drive - uses config for folder/file selection
+                    if (
+                        connector.connector_type
+                        == SearchSourceConnectorType.GOOGLE_DRIVE_CONNECTOR
+                    ):
+                        connector_config = connector.config or {}
+                        selected_folders = connector_config.get("selected_folders", [])
+                        selected_files = connector_config.get("selected_files", [])
+                        indexing_options = connector_config.get(
+                            "indexing_options",
+                            {
+                                "max_files_per_folder": 100,
+                                "incremental_sync": True,
+                                "include_subfolders": True,
+                            },
+                        )
+
+                        if selected_folders or selected_files:
+                            task.delay(
+                                connector.id,
+                                connector.search_space_id,
+                                str(connector.user_id),
+                                {
+                                    "folders": selected_folders,
+                                    "files": selected_files,
+                                    "indexing_options": indexing_options,
+                                },
+                            )
+                        else:
+                            # No folders/files selected - skip indexing but still update next_scheduled_at
+                            # to prevent checking every minute
+                            logger.info(
+                                f"Google Drive connector {connector.id} has no folders or files selected, "
+                                "skipping periodic indexing (will check again at next scheduled time)"
+                            )
+                            from datetime import timedelta
+
+                            connector.next_scheduled_at = now + timedelta(
+                                minutes=connector.indexing_frequency_minutes
+                            )
+                            await session.commit()
+                            continue
+                    else:
+                        task.delay(
+                            connector.id,
+                            connector.search_space_id,
+                            str(connector.user_id),
+                            None,  # start_date - uses last_indexed_at
+                            None,  # end_date - uses now
+                        )
 
                     # Update next_scheduled_at for next run
                     from datetime import timedelta

surfsense_backend/app/tasks/chat/stream_new_chat.py

@@ -423,9 +423,9 @@ async def stream_new_chat(
                     title = title[:27] + "..."
                 doc_names.append(title)
             if len(doc_names) == 1:
-                processing_parts.append(f"[📖 {doc_names[0]}]")
+                processing_parts.append(f"[{doc_names[0]}]")
             else:
-                processing_parts.append(f"[📖 {len(doc_names)} docs]")
+                processing_parts.append(f"[{len(doc_names)} docs]")
 
         last_active_step_items = [f"{action_verb}: {' '.join(processing_parts)}"]
 

surfsense_backend/app/tasks/connector_indexers/discord_indexer.py

@@ -549,7 +549,10 @@ async def index_discord_messages(
         logger.info(
             f"Discord indexing completed: {documents_indexed} new messages, {documents_skipped} skipped"
         )
-        return documents_indexed, result_message
+        return (
+            documents_indexed,
+            None,
+        )  # Return None on success (result_message is for logging only)
 
     except SQLAlchemyError as db_error:
         await session.rollback()

surfsense_backend/app/tasks/connector_indexers/google_drive_indexer.py

@@ -37,6 +37,7 @@ async def index_google_drive_files(
     use_delta_sync: bool = True,
     update_last_indexed: bool = True,
     max_files: int = 500,
+    include_subfolders: bool = False,
 ) -> tuple[int, str | None]:
     """
     Index Google Drive files for a specific connector.
@@ -51,6 +52,7 @@ async def index_google_drive_files(
         use_delta_sync: Whether to use change tracking for incremental sync
         update_last_indexed: Whether to update last_indexed_at timestamp
         max_files: Maximum number of files to index
+        include_subfolders: Whether to recursively index files in subfolders
 
     Returns:
         Tuple of (number_of_indexed_files, error_message)
@@ -144,6 +146,7 @@ async def index_google_drive_files(
                 task_logger=task_logger,
                 log_entry=log_entry,
                 max_files=max_files,
+                include_subfolders=include_subfolders,
             )
         else:
             logger.info(f"Using full scan for connector {connector_id}")
@@ -159,6 +162,7 @@ async def index_google_drive_files(
                 task_logger=task_logger,
                 log_entry=log_entry,
                 max_files=max_files,
+                include_subfolders=include_subfolders,
             )
 
         documents_indexed, documents_skipped = result
@@ -168,6 +172,9 @@ async def index_google_drive_files(
             if new_token and not token_error:
                 from sqlalchemy.orm.attributes import flag_modified
 
+                # Refresh connector to reload attributes that may have been expired by earlier commits
+                await session.refresh(connector)
+
                 if "folder_tokens" not in connector.config:
                     connector.config["folder_tokens"] = {}
                 connector.config["folder_tokens"][target_folder_id] = new_token
@@ -375,60 +382,89 @@ async def _index_full_scan(
     task_logger: TaskLoggingService,
     log_entry: any,
     max_files: int,
+    include_subfolders: bool = False,
 ) -> tuple[int, int]:
     """Perform full scan indexing of a folder."""
     await task_logger.log_task_progress(
         log_entry,
-        f"Starting full scan of folder: {folder_name}",
-        {"stage": "full_scan", "folder_id": folder_id},
+        f"Starting full scan of folder: {folder_name} (include_subfolders={include_subfolders})",
+        {
+            "stage": "full_scan",
+            "folder_id": folder_id,
+            "include_subfolders": include_subfolders,
+        },
     )
 
     documents_indexed = 0
     documents_skipped = 0
-    page_token = None
     files_processed = 0
 
-    while files_processed < max_files:
-        files, next_token, error = await get_files_in_folder(
-            drive_client, folder_id, include_subfolders=False, page_token=page_token
-        )
+    # Queue of folders to process: (folder_id, folder_name)
+    folders_to_process = [(folder_id, folder_name)]
 
-        if error:
-            logger.error(f"Error listing files: {error}")
-            break
+    while folders_to_process and files_processed < max_files:
+        current_folder_id, current_folder_name = folders_to_process.pop(0)
+        logger.info(f"Processing folder: {current_folder_name} ({current_folder_id})")
+        page_token = None
 
-        if not files:
-            break
-
-        for file in files:
-            if files_processed >= max_files:
-                break
-
-            files_processed += 1
-
-            indexed, skipped = await _process_single_file(
-                drive_client=drive_client,
-                session=session,
-                file=file,
-                connector_id=connector_id,
-                search_space_id=search_space_id,
-                user_id=user_id,
-                task_logger=task_logger,
-                log_entry=log_entry,
+        while files_processed < max_files:
+            # Get files and folders in current folder
+            # include_subfolders=True here so we get folder items to queue them
+            files, next_token, error = await get_files_in_folder(
+                drive_client,
+                current_folder_id,
+                include_subfolders=True,
+                page_token=page_token,
             )
 
-            documents_indexed += indexed
-            documents_skipped += skipped
+            if error:
+                logger.error(f"Error listing files in {current_folder_name}: {error}")
+                break
 
-            if documents_indexed % 10 == 0 and documents_indexed > 0:
-                await session.commit()
-                logger.info(
-                    f"Committed batch: {documents_indexed} files indexed so far"
+            if not files:
+                break
+
+            for file in files:
+                if files_processed >= max_files:
+                    break
+
+                mime_type = file.get("mimeType", "")
+
+                # If this is a folder and include_subfolders is enabled, queue it for processing
+                if mime_type == "application/vnd.google-apps.folder":
+                    if include_subfolders:
+                        folders_to_process.append(
+                            (file["id"], file.get("name", "Unknown"))
+                        )
+                        logger.debug(f"Queued subfolder: {file.get('name', 'Unknown')}")
+                    continue
+
+                # Process the file
+                files_processed += 1
+
+                indexed, skipped = await _process_single_file(
+                    drive_client=drive_client,
+                    session=session,
+                    file=file,
+                    connector_id=connector_id,
+                    search_space_id=search_space_id,
+                    user_id=user_id,
+                    task_logger=task_logger,
+                    log_entry=log_entry,
                 )
 
-        page_token = next_token
-        if not page_token:
-            break
+                documents_indexed += indexed
+                documents_skipped += skipped
+
+                if documents_indexed % 10 == 0 and documents_indexed > 0:
+                    await session.commit()
+                    logger.info(
+                        f"Committed batch: {documents_indexed} files indexed so far"
+                    )
+
+            page_token = next_token
+            if not page_token:
+                break
 
     logger.info(
         f"Full scan complete: {documents_indexed} indexed, {documents_skipped} skipped"
@@ -448,8 +484,13 @@ async def _index_with_delta_sync(
     task_logger: TaskLoggingService,
     log_entry: any,
     max_files: int,
+    include_subfolders: bool = False,
 ) -> tuple[int, int]:
-    """Perform delta sync indexing using change tracking."""
+    """Perform delta sync indexing using change tracking.
+
+    Note: include_subfolders is accepted for API consistency but delta sync
+    automatically tracks changes across all folders including subfolders.
+    """
     await task_logger.log_task_progress(
         log_entry,
         f"Starting delta sync from token: {start_page_token[:20]}...",
@@ -515,6 +556,131 @@ async def _index_with_delta_sync(
     return documents_indexed, documents_skipped
 
 
+async def _check_rename_only_update(
+    session: AsyncSession,
+    file: dict,
+    search_space_id: int,
+) -> tuple[bool, str | None]:
+    """
+    Check if a file only needs a rename update (no content change).
+
+    Uses md5Checksum comparison (preferred) or modifiedTime (fallback for Google Workspace files)
+    to detect if content has changed. This optimization prevents unnecessary ETL API calls
+    (Docling/LlamaCloud) for rename-only operations.
+
+    Args:
+        session: Database session
+        file: File metadata from Google Drive API
+        search_space_id: ID of the search space
+
+    Returns:
+        Tuple of (is_rename_only, message)
+        - (True, message): Only filename changed, document was updated
+        - (False, None): Content changed or new file, needs full processing
+    """
+    from sqlalchemy import select
+    from sqlalchemy.orm.attributes import flag_modified
+
+    from app.db import Document
+
+    file_id = file.get("id")
+    file_name = file.get("name", "Unknown")
+    incoming_md5 = file.get("md5Checksum")  # None for Google Workspace files
+    incoming_modified_time = file.get("modifiedTime")
+
+    if not file_id:
+        return False, None
+
+    # Try to find existing document by file_id-based hash (primary method)
+    primary_hash = generate_unique_identifier_hash(
+        DocumentType.GOOGLE_DRIVE_FILE, file_id, search_space_id
+    )
+    existing_document = await check_document_by_unique_identifier(session, primary_hash)
+
+    # If not found by primary hash, try searching by metadata (for legacy documents)
+    if not existing_document:
+        result = await session.execute(
+            select(Document).where(
+                Document.search_space_id == search_space_id,
+                Document.document_type == DocumentType.GOOGLE_DRIVE_FILE,
+                Document.document_metadata["google_drive_file_id"].astext == file_id,
+            )
+        )
+        existing_document = result.scalar_one_or_none()
+        if existing_document:
+            logger.debug(f"Found legacy document by metadata for file_id: {file_id}")
+
+    if not existing_document:
+        # New file, needs full processing
+        return False, None
+
+    # Get stored checksums/timestamps from document metadata
+    doc_metadata = existing_document.document_metadata or {}
+    stored_md5 = doc_metadata.get("md5_checksum")
+    stored_modified_time = doc_metadata.get("modified_time")
+
+    # Determine if content changed using md5Checksum (preferred) or modifiedTime (fallback)
+    content_unchanged = False
+
+    if incoming_md5 and stored_md5:
+        # Best case: Compare md5 checksums (only changes when content changes, not on rename)
+        content_unchanged = incoming_md5 == stored_md5
+        logger.debug(f"MD5 comparison for {file_name}: unchanged={content_unchanged}")
+    elif incoming_md5 and not stored_md5:
+        # Have incoming md5 but no stored md5 (legacy doc) - need to reprocess to store it
+        logger.debug(
+            f"No stored md5 for {file_name}, will reprocess to store md5_checksum"
+        )
+        return False, None
+    elif not incoming_md5:
+        # Google Workspace file (no md5Checksum available) - fall back to modifiedTime
+        # Note: modifiedTime is less reliable as it changes on rename too, but it's the best we have
+        if incoming_modified_time and stored_modified_time:
+            content_unchanged = incoming_modified_time == stored_modified_time
+            logger.debug(
+                f"ModifiedTime fallback for Google Workspace file {file_name}: unchanged={content_unchanged}"
+            )
+        else:
+            # No stored modifiedTime (legacy) - reprocess to store it
+            return False, None
+
+    if content_unchanged:
+        # Content hasn't changed - check if filename changed
+        old_name = doc_metadata.get("FILE_NAME") or doc_metadata.get(
+            "google_drive_file_name"
+        )
+
+        if old_name and old_name != file_name:
+            # Rename-only update - update the document without re-processing
+            existing_document.title = file_name
+            if not existing_document.document_metadata:
+                existing_document.document_metadata = {}
+            existing_document.document_metadata["FILE_NAME"] = file_name
+            existing_document.document_metadata["google_drive_file_name"] = file_name
+            # Also update modified_time for Google Workspace files (since it changed on rename)
+            if incoming_modified_time:
+                existing_document.document_metadata["modified_time"] = (
+                    incoming_modified_time
+                )
+            flag_modified(existing_document, "document_metadata")
+            await session.commit()
+
+            logger.info(
+                f"Rename-only update: '{old_name}' → '{file_name}' (skipped ETL)"
+            )
+            return (
+                True,
+                f"File renamed: '{old_name}' → '{file_name}' (no content change)",
+            )
+        else:
+            # Neither content nor name changed
+            logger.debug(f"File unchanged: {file_name}")
+            return True, "File unchanged (same content and name)"
+
+    # Content changed - needs full processing
+    return False, None
+
+
 async def _process_single_file(
     drive_client: GoogleDriveClient,
     session: AsyncSession,
@@ -537,6 +703,27 @@ async def _process_single_file(
     try:
         logger.info(f"Processing file: {file_name} ({mime_type})")
 
+        # Early check: Is this a rename-only update?
+        # This optimization prevents downloading and ETL processing for files
+        # where only the name changed but content is the same.
+        is_rename_only, rename_message = await _check_rename_only_update(
+            session=session,
+            file=file,
+            search_space_id=search_space_id,
+        )
+
+        if is_rename_only:
+            await task_logger.log_task_progress(
+                log_entry,
+                f"Skipped ETL for {file_name}: {rename_message}",
+                {"status": "rename_only", "reason": rename_message},
+            )
+            # Return 1 for renamed files (they are "indexed" in the sense that they're updated)
+            # Return 0 for unchanged files
+            if "renamed" in (rename_message or "").lower():
+                return 1, 0
+            return 0, 1
+
         _, error, _ = await download_and_process_file(
             client=drive_client,
             file=file,
@@ -564,7 +751,15 @@ async def _process_single_file(
 
 
 async def _remove_document(session: AsyncSession, file_id: str, search_space_id: int):
-    """Remove a document that was deleted in Drive."""
+    """Remove a document that was deleted in Drive.
+
+    Handles both new (file_id-based) and legacy (filename-based) hash schemes.
+    """
+    from sqlalchemy import select
+
+    from app.db import Document
+
+    # First try with file_id-based hash (new method)
     unique_identifier_hash = generate_unique_identifier_hash(
         DocumentType.GOOGLE_DRIVE_FILE, file_id, search_space_id
     )
@@ -573,6 +768,19 @@ async def _remove_document(session: AsyncSession, file_id: str, search_space_id:
         session, unique_identifier_hash
     )
 
+    # If not found, search by metadata (for legacy documents with filename-based hash)
+    if not existing_document:
+        result = await session.execute(
+            select(Document).where(
+                Document.search_space_id == search_space_id,
+                Document.document_type == DocumentType.GOOGLE_DRIVE_FILE,
+                Document.document_metadata["google_drive_file_id"].astext == file_id,
+            )
+        )
+        existing_document = result.scalar_one_or_none()
+        if existing_document:
+            logger.info(f"Found legacy document by metadata for file_id: {file_id}")
+
     if existing_document:
         await session.delete(existing_document)
         logger.info(f"Removed deleted file document: {file_id}")

surfsense_backend/app/tasks/connector_indexers/notion_indexer.py

@@ -464,7 +464,10 @@ async def index_notion_pages(
         # Clean up the async client
         await notion_client.close()
 
-        return total_processed, result_message
+        return (
+            total_processed,
+            None,
+        )  # Return None on success (result_message is for logging only)
 
     except SQLAlchemyError as db_error:
         await session.rollback()

surfsense_backend/app/tasks/connector_indexers/slack_indexer.py

@@ -413,7 +413,10 @@ async def index_slack_messages(
         logger.info(
             f"Slack indexing completed: {documents_indexed} new channels, {documents_skipped} skipped"
         )
-        return total_processed, result_message
+        return (
+            total_processed,
+            None,
+        )  # Return None on success (result_message is for logging only)
 
     except SQLAlchemyError as db_error:
         await session.rollback()

surfsense_backend/app/tasks/connector_indexers/teams_indexer.py

@@ -460,7 +460,10 @@ async def index_teams_messages(
             documents_indexed,
             documents_skipped,
         )
-        return total_processed, result_message
+        return (
+            total_processed,
+            None,
+        )  # Return None on success (result_message is for logging only)
 
     except SQLAlchemyError as db_error:
         await session.rollback()

surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py

@@ -371,17 +371,14 @@ async def index_crawled_urls(
         )
         await session.commit()
 
-        # Build result message
-        result_message = None
+        # Log failed URLs if any (for debugging purposes)
         if failed_urls:
             failed_summary = "; ".join(
                 [f"{url}: {error}" for url, error in failed_urls[:5]]
             )
             if len(failed_urls) > 5:
                 failed_summary += f" (and {len(failed_urls) - 5} more)"
-            result_message = (
-                f"Completed with {len(failed_urls)} failures: {failed_summary}"
-            )
+            logger.warning(f"Some URLs failed to index: {failed_summary}")
 
         await task_logger.log_task_success(
             log_entry,
@@ -400,7 +397,10 @@ async def index_crawled_urls(
             f"{documents_updated} updated, {documents_skipped} skipped, "
             f"{len(failed_urls)} failed"
         )
-        return total_processed, result_message
+        return (
+            total_processed,
+            None,
+        )  # Return None on success (result_message is for logging only)
 
     except SQLAlchemyError as db_error:
         await session.rollback()

surfsense_backend/app/tasks/document_processors/file_processors.py

@@ -17,8 +17,9 @@ from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import config as app_config
-from app.db import Document, DocumentType, Log
+from app.db import Document, DocumentType, Log, Notification
 from app.services.llm_service import get_user_long_context_llm
+from app.services.notification_service import NotificationService
 from app.services.task_logging_service import TaskLoggingService
 from app.utils.document_converters import (
     convert_document_to_markdown,
@@ -30,6 +31,7 @@ from app.utils.document_converters import (
 
 from .base import (
     check_document_by_unique_identifier,
+    check_duplicate_document,
     get_current_timestamp,
 )
 from .markdown_processor import add_received_markdown_file_document
@@ -48,6 +50,160 @@ LLAMACLOUD_RETRYABLE_EXCEPTIONS = (
 )
 
 
+def get_google_drive_unique_identifier(
+    connector: dict | None,
+    filename: str,
+    search_space_id: int,
+) -> tuple[str, str | None]:
+    """
+    Get unique identifier hash for a file, with special handling for Google Drive.
+
+    For Google Drive files, uses file_id as the unique identifier (doesn't change on rename).
+    For other files, uses filename.
+
+    Args:
+        connector: Optional connector info dict with type and metadata
+        filename: The filename (used for non-Google Drive files or as fallback)
+        search_space_id: The search space ID
+
+    Returns:
+        Tuple of (primary_hash, legacy_hash or None)
+        - For Google Drive: (file_id_based_hash, filename_based_hash for migration)
+        - For other sources: (filename_based_hash, None)
+    """
+    if connector and connector.get("type") == DocumentType.GOOGLE_DRIVE_FILE:
+        metadata = connector.get("metadata", {})
+        file_id = metadata.get("google_drive_file_id")
+
+        if file_id:
+            # New method: use file_id as unique identifier (doesn't change on rename)
+            primary_hash = generate_unique_identifier_hash(
+                DocumentType.GOOGLE_DRIVE_FILE, file_id, search_space_id
+            )
+            # Legacy method: for backward compatibility with existing documents
+            # that were indexed with filename-based hash
+            legacy_hash = generate_unique_identifier_hash(
+                DocumentType.GOOGLE_DRIVE_FILE, filename, search_space_id
+            )
+            return primary_hash, legacy_hash
+
+    # For non-Google Drive files, use filename as before
+    primary_hash = generate_unique_identifier_hash(
+        DocumentType.FILE, filename, search_space_id
+    )
+    return primary_hash, None
+
+
+async def handle_existing_document_update(
+    session: AsyncSession,
+    existing_document: Document,
+    content_hash: str,
+    connector: dict | None,
+    filename: str,
+    primary_hash: str,
+) -> tuple[bool, Document | None]:
+    """
+    Handle update logic for an existing document.
+
+    Args:
+        session: Database session
+        existing_document: The existing document found in database
+        content_hash: Hash of the new content
+        connector: Optional connector info
+        filename: Current filename
+        primary_hash: The primary hash (file_id based for Google Drive)
+
+    Returns:
+        Tuple of (should_skip_processing, document_to_return)
+        - (True, document): Content unchanged, just return existing document
+        - (False, None): Content changed, need to re-process
+    """
+    # Check if this document needs hash migration (found via legacy hash)
+    if existing_document.unique_identifier_hash != primary_hash:
+        existing_document.unique_identifier_hash = primary_hash
+        logging.info(f"Migrated document to file_id-based identifier: {filename}")
+
+    # Check if content has changed
+    if existing_document.content_hash == content_hash:
+        # Content unchanged - check if we need to update metadata (e.g., filename changed)
+        if connector and connector.get("type") == DocumentType.GOOGLE_DRIVE_FILE:
+            connector_metadata = connector.get("metadata", {})
+            new_name = connector_metadata.get("google_drive_file_name")
+            # Check both possible keys for old name (FILE_NAME is used in stored documents)
+            doc_metadata = existing_document.document_metadata or {}
+            old_name = doc_metadata.get("FILE_NAME") or doc_metadata.get(
+                "google_drive_file_name"
+            )
+
+            if new_name and old_name and old_name != new_name:
+                # File was renamed - update title and metadata, skip expensive processing
+                from sqlalchemy.orm.attributes import flag_modified
+
+                existing_document.title = new_name
+                if not existing_document.document_metadata:
+                    existing_document.document_metadata = {}
+                existing_document.document_metadata["FILE_NAME"] = new_name
+                existing_document.document_metadata["google_drive_file_name"] = new_name
+                flag_modified(existing_document, "document_metadata")
+                await session.commit()
+                logging.info(
+                    f"File renamed in Google Drive: '{old_name}' → '{new_name}' (no re-processing needed)"
+                )
+
+        logging.info(f"Document for file {filename} unchanged. Skipping.")
+        return True, existing_document
+    else:
+        # Content has changed - need to re-process
+        logging.info(f"Content changed for file {filename}. Updating document.")
+        return False, None
+
+
+async def find_existing_document_with_migration(
+    session: AsyncSession,
+    primary_hash: str,
+    legacy_hash: str | None,
+    content_hash: str | None = None,
+) -> Document | None:
+    """
+    Find existing document, checking both new hash and legacy hash for migration,
+    with fallback to content_hash for cross-source deduplication.
+
+    Args:
+        session: Database session
+        primary_hash: The primary hash (file_id based for Google Drive)
+        legacy_hash: The legacy hash (filename based) for migration, or None
+        content_hash: The content hash for fallback deduplication, or None
+
+    Returns:
+        Existing document if found, None otherwise
+    """
+    # First check with primary hash (new method)
+    existing_document = await check_document_by_unique_identifier(session, primary_hash)
+
+    # If not found and we have a legacy hash, check with that (migration path)
+    if not existing_document and legacy_hash:
+        existing_document = await check_document_by_unique_identifier(
+            session, legacy_hash
+        )
+        if existing_document:
+            logging.info(
+                "Found legacy document (filename-based hash), will migrate to file_id-based hash"
+            )
+
+    # Fallback: check by content_hash to catch duplicates from different sources
+    # This prevents unique constraint violations when the same content exists
+    # under a different unique_identifier (e.g., manual upload vs Google Drive)
+    if not existing_document and content_hash:
+        existing_document = await check_duplicate_document(session, content_hash)
+        if existing_document:
+            logging.info(
+                f"Found duplicate content from different source (content_hash match). "
+                f"Original document ID: {existing_document.id}, type: {existing_document.document_type}"
+            )
+
+    return existing_document
+
+
 async def parse_with_llamacloud_retry(
     file_path: str,
     estimated_pages: int,
@@ -157,6 +313,7 @@ async def add_received_file_document_using_unstructured(
     unstructured_processed_elements: list[LangChainDocument],
     search_space_id: int,
     user_id: str,
+    connector: dict | None = None,
 ) -> Document | None:
     """
     Process and store a file document using Unstructured service.
@@ -167,6 +324,7 @@ async def add_received_file_document_using_unstructured(
         unstructured_processed_elements: Processed elements from Unstructured
         search_space_id: ID of the search space
         user_id: ID of the user
+        connector: Optional connector info for Google Drive files
 
     Returns:
         Document object if successful, None if failed
@@ -176,29 +334,32 @@ async def add_received_file_document_using_unstructured(
             unstructured_processed_elements
         )
 
-        # Generate unique identifier hash for this file
-        unique_identifier_hash = generate_unique_identifier_hash(
-            DocumentType.FILE, file_name, search_space_id
+        # Generate unique identifier hash (uses file_id for Google Drive, filename for others)
+        primary_hash, legacy_hash = get_google_drive_unique_identifier(
+            connector, file_name, search_space_id
         )
 
         # Generate content hash
         content_hash = generate_content_hash(file_in_markdown, search_space_id)
 
-        # Check if document with this unique identifier already exists
-        existing_document = await check_document_by_unique_identifier(
-            session, unique_identifier_hash
+        # Check if document exists (with migration support for Google Drive and content_hash fallback)
+        existing_document = await find_existing_document_with_migration(
+            session, primary_hash, legacy_hash, content_hash
         )
 
         if existing_document:
-            # Document exists - check if content has changed
-            if existing_document.content_hash == content_hash:
-                logging.info(f"Document for file {file_name} unchanged. Skipping.")
-                return existing_document
-            else:
-                # Content has changed - update the existing document
-                logging.info(
-                    f"Content changed for file {file_name}. Updating document."
-                )
+            # Handle existing document (rename detection, content change check)
+            should_skip, doc = await handle_existing_document_update(
+                session,
+                existing_document,
+                content_hash,
+                connector,
+                file_name,
+                primary_hash,
+            )
+            if should_skip:
+                return doc
+            # Content changed - continue to update
 
         # Get user's long context LLM (needed for both create and update)
         user_llm = await get_user_long_context_llm(session, user_id, search_space_id)
@@ -250,10 +411,15 @@ async def add_received_file_document_using_unstructured(
             document = existing_document
         else:
             # Create new document
+            # Determine document type based on connector
+            doc_type = DocumentType.FILE
+            if connector and connector.get("type") == DocumentType.GOOGLE_DRIVE_FILE:
+                doc_type = DocumentType.GOOGLE_DRIVE_FILE
+
             document = Document(
                 search_space_id=search_space_id,
                 title=file_name,
-                document_type=DocumentType.FILE,
+                document_type=doc_type,
                 document_metadata={
                     "FILE_NAME": file_name,
                     "ETL_SERVICE": "UNSTRUCTURED",
@@ -262,7 +428,7 @@ async def add_received_file_document_using_unstructured(
                 embedding=summary_embedding,
                 chunks=chunks,
                 content_hash=content_hash,
-                unique_identifier_hash=unique_identifier_hash,
+                unique_identifier_hash=primary_hash,
                 blocknote_document=blocknote_json,
                 content_needs_reindexing=False,
                 updated_at=get_current_timestamp(),
@@ -287,6 +453,7 @@ async def add_received_file_document_using_llamacloud(
     llamacloud_markdown_document: str,
     search_space_id: int,
     user_id: str,
+    connector: dict | None = None,
 ) -> Document | None:
     """
     Process and store document content parsed by LlamaCloud.
@@ -297,6 +464,7 @@ async def add_received_file_document_using_llamacloud(
         llamacloud_markdown_document: Markdown content from LlamaCloud parsing
         search_space_id: ID of the search space
         user_id: ID of the user
+        connector: Optional connector info for Google Drive files
 
     Returns:
         Document object if successful, None if failed
@@ -305,29 +473,32 @@ async def add_received_file_document_using_llamacloud(
         # Combine all markdown documents into one
         file_in_markdown = llamacloud_markdown_document
 
-        # Generate unique identifier hash for this file
-        unique_identifier_hash = generate_unique_identifier_hash(
-            DocumentType.FILE, file_name, search_space_id
+        # Generate unique identifier hash (uses file_id for Google Drive, filename for others)
+        primary_hash, legacy_hash = get_google_drive_unique_identifier(
+            connector, file_name, search_space_id
         )
 
         # Generate content hash
         content_hash = generate_content_hash(file_in_markdown, search_space_id)
 
-        # Check if document with this unique identifier already exists
-        existing_document = await check_document_by_unique_identifier(
-            session, unique_identifier_hash
+        # Check if document exists (with migration support for Google Drive and content_hash fallback)
+        existing_document = await find_existing_document_with_migration(
+            session, primary_hash, legacy_hash, content_hash
         )
 
         if existing_document:
-            # Document exists - check if content has changed
-            if existing_document.content_hash == content_hash:
-                logging.info(f"Document for file {file_name} unchanged. Skipping.")
-                return existing_document
-            else:
-                # Content has changed - update the existing document
-                logging.info(
-                    f"Content changed for file {file_name}. Updating document."
-                )
+            # Handle existing document (rename detection, content change check)
+            should_skip, doc = await handle_existing_document_update(
+                session,
+                existing_document,
+                content_hash,
+                connector,
+                file_name,
+                primary_hash,
+            )
+            if should_skip:
+                return doc
+            # Content changed - continue to update
 
         # Get user's long context LLM (needed for both create and update)
         user_llm = await get_user_long_context_llm(session, user_id, search_space_id)
@@ -379,10 +550,15 @@ async def add_received_file_document_using_llamacloud(
             document = existing_document
         else:
             # Create new document
+            # Determine document type based on connector
+            doc_type = DocumentType.FILE
+            if connector and connector.get("type") == DocumentType.GOOGLE_DRIVE_FILE:
+                doc_type = DocumentType.GOOGLE_DRIVE_FILE
+
             document = Document(
                 search_space_id=search_space_id,
                 title=file_name,
-                document_type=DocumentType.FILE,
+                document_type=doc_type,
                 document_metadata={
                     "FILE_NAME": file_name,
                     "ETL_SERVICE": "LLAMACLOUD",
@@ -391,7 +567,7 @@ async def add_received_file_document_using_llamacloud(
                 embedding=summary_embedding,
                 chunks=chunks,
                 content_hash=content_hash,
-                unique_identifier_hash=unique_identifier_hash,
+                unique_identifier_hash=primary_hash,
                 blocknote_document=blocknote_json,
                 content_needs_reindexing=False,
                 updated_at=get_current_timestamp(),
@@ -418,6 +594,7 @@ async def add_received_file_document_using_docling(
     docling_markdown_document: str,
     search_space_id: int,
     user_id: str,
+    connector: dict | None = None,
 ) -> Document | None:
     """
     Process and store document content parsed by Docling.
@@ -428,6 +605,7 @@ async def add_received_file_document_using_docling(
         docling_markdown_document: Markdown content from Docling parsing
         search_space_id: ID of the search space
         user_id: ID of the user
+        connector: Optional connector info for Google Drive files
 
     Returns:
         Document object if successful, None if failed
@@ -435,35 +613,38 @@ async def add_received_file_document_using_docling(
     try:
         file_in_markdown = docling_markdown_document
 
-        # Generate unique identifier hash for this file
-        unique_identifier_hash = generate_unique_identifier_hash(
-            DocumentType.FILE, file_name, search_space_id
+        # Generate unique identifier hash (uses file_id for Google Drive, filename for others)
+        primary_hash, legacy_hash = get_google_drive_unique_identifier(
+            connector, file_name, search_space_id
         )
 
         # Generate content hash
         content_hash = generate_content_hash(file_in_markdown, search_space_id)
 
-        # Check if document with this unique identifier already exists
-        existing_document = await check_document_by_unique_identifier(
-            session, unique_identifier_hash
+        # Check if document exists (with migration support for Google Drive and content_hash fallback)
+        existing_document = await find_existing_document_with_migration(
+            session, primary_hash, legacy_hash, content_hash
         )
 
         if existing_document:
-            # Document exists - check if content has changed
-            if existing_document.content_hash == content_hash:
-                logging.info(f"Document for file {file_name} unchanged. Skipping.")
-                return existing_document
-            else:
-                # Content has changed - update the existing document
-                logging.info(
-                    f"Content changed for file {file_name}. Updating document."
-                )
+            # Handle existing document (rename detection, content change check)
+            should_skip, doc = await handle_existing_document_update(
+                session,
+                existing_document,
+                content_hash,
+                connector,
+                file_name,
+                primary_hash,
+            )
+            if should_skip:
+                return doc
+            # Content changed - continue to update
 
         # Get user's long context LLM (needed for both create and update)
         user_llm = await get_user_long_context_llm(session, user_id, search_space_id)
         if not user_llm:
             raise RuntimeError(
-                f"No long context LLM configured for user {user_id} in search space {search_space_id}"
+                f"No long context LLM configured for user {user_id} in search_space {search_space_id}"
             )
 
         # Generate summary using chunked processing for large documents
@@ -533,10 +714,15 @@ async def add_received_file_document_using_docling(
             document = existing_document
         else:
             # Create new document
+            # Determine document type based on connector
+            doc_type = DocumentType.FILE
+            if connector and connector.get("type") == DocumentType.GOOGLE_DRIVE_FILE:
+                doc_type = DocumentType.GOOGLE_DRIVE_FILE
+
             document = Document(
                 search_space_id=search_space_id,
                 title=file_name,
-                document_type=DocumentType.FILE,
+                document_type=doc_type,
                 document_metadata={
                     "FILE_NAME": file_name,
                     "ETL_SERVICE": "DOCLING",
@@ -545,15 +731,15 @@ async def add_received_file_document_using_docling(
                 embedding=summary_embedding,
                 chunks=chunks,
                 content_hash=content_hash,
-                unique_identifier_hash=unique_identifier_hash,
+                unique_identifier_hash=primary_hash,
                 blocknote_document=blocknote_json,
                 content_needs_reindexing=False,
                 updated_at=get_current_timestamp(),
             )
 
-        session.add(document)
-        await session.commit()
-        await session.refresh(document)
+            session.add(document)
+            await session.commit()
+            await session.refresh(document)
 
         return document
     except SQLAlchemyError as db_error:
@@ -594,10 +780,23 @@ async def process_file_in_background(
     log_entry: Log,
     connector: dict
     | None = None,  # Optional: {"type": "GOOGLE_DRIVE_FILE", "metadata": {...}}
-):
+    notification: Notification
+    | None = None,  # Optional notification for progress updates
+) -> Document | None:
     try:
         # Check if the file is a markdown or text file
         if filename.lower().endswith((".md", ".markdown", ".txt")):
+            # Update notification: parsing stage
+            if notification:
+                await (
+                    NotificationService.document_processing.notify_processing_progress(
+                        session,
+                        notification,
+                        stage="parsing",
+                        stage_message="Reading file",
+                    )
+                )
+
             await task_logger.log_task_progress(
                 log_entry,
                 f"Processing markdown/text file: {filename}",
@@ -617,6 +816,14 @@ async def process_file_in_background(
                 print("Error deleting temp file", e)
                 pass
 
+            # Update notification: chunking stage
+            if notification:
+                await (
+                    NotificationService.document_processing.notify_processing_progress(
+                        session, notification, stage="chunking"
+                    )
+                )
+
             await task_logger.log_task_progress(
                 log_entry,
                 f"Creating document from markdown content: {filename}",
@@ -628,7 +835,7 @@ async def process_file_in_background(
 
             # Process markdown directly through specialized function
             result = await add_received_markdown_file_document(
-                session, filename, markdown_content, search_space_id, user_id
+                session, filename, markdown_content, search_space_id, user_id, connector
             )
 
             if connector:
@@ -644,17 +851,30 @@ async def process_file_in_background(
                         "file_type": "markdown",
                     },
                 )
+                return result
             else:
                 await task_logger.log_task_success(
                     log_entry,
                     f"Markdown file already exists (duplicate): {filename}",
                     {"duplicate_detected": True, "file_type": "markdown"},
                 )
+                return None
 
         # Check if the file is an audio file
         elif filename.lower().endswith(
             (".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".webm")
         ):
+            # Update notification: parsing stage (transcription)
+            if notification:
+                await (
+                    NotificationService.document_processing.notify_processing_progress(
+                        session,
+                        notification,
+                        stage="parsing",
+                        stage_message="Transcribing audio",
+                    )
+                )
+
             await task_logger.log_task_progress(
                 log_entry,
                 f"Processing audio file for transcription: {filename}",
@@ -738,6 +958,14 @@ async def process_file_in_background(
                 },
             )
 
+            # Update notification: chunking stage
+            if notification:
+                await (
+                    NotificationService.document_processing.notify_processing_progress(
+                        session, notification, stage="chunking"
+                    )
+                )
+
             # Clean up the temp file
             try:
                 os.unlink(file_path)
@@ -747,7 +975,7 @@ async def process_file_in_background(
 
             # Process transcription as markdown document
             result = await add_received_markdown_file_document(
-                session, filename, transcribed_text, search_space_id, user_id
+                session, filename, transcribed_text, search_space_id, user_id, connector
             )
 
             if connector:
@@ -765,12 +993,14 @@ async def process_file_in_background(
                         "stt_service": stt_service_type,
                     },
                 )
+                return result
             else:
                 await task_logger.log_task_success(
                     log_entry,
                     f"Audio file transcript already exists (duplicate): {filename}",
                     {"duplicate_detected": True, "file_type": "audio"},
                 )
+                return None
 
         else:
             # Import page limit service
@@ -835,6 +1065,15 @@ async def process_file_in_background(
                 ) from e
 
             if app_config.ETL_SERVICE == "UNSTRUCTURED":
+                # Update notification: parsing stage
+                if notification:
+                    await NotificationService.document_processing.notify_processing_progress(
+                        session,
+                        notification,
+                        stage="parsing",
+                        stage_message="Extracting content",
+                    )
+
                 await task_logger.log_task_progress(
                     log_entry,
                     f"Processing file with Unstructured ETL: {filename}",
@@ -860,6 +1099,12 @@ async def process_file_in_background(
 
                 docs = await loader.aload()
 
+                # Update notification: chunking stage
+                if notification:
+                    await NotificationService.document_processing.notify_processing_progress(
+                        session, notification, stage="chunking", chunks_count=len(docs)
+                    )
+
                 await task_logger.log_task_progress(
                     log_entry,
                     f"Unstructured ETL completed, creating document: {filename}",
@@ -895,7 +1140,7 @@ async def process_file_in_background(
 
                 # Pass the documents to the existing background task
                 result = await add_received_file_document_using_unstructured(
-                    session, filename, docs, search_space_id, user_id
+                    session, filename, docs, search_space_id, user_id, connector
                 )
 
                 if connector:
@@ -919,6 +1164,7 @@ async def process_file_in_background(
                             "pages_processed": final_page_count,
                         },
                     )
+                    return result
                 else:
                     await task_logger.log_task_success(
                         log_entry,
@@ -929,8 +1175,18 @@ async def process_file_in_background(
                             "etl_service": "UNSTRUCTURED",
                         },
                     )
+                    return None
 
             elif app_config.ETL_SERVICE == "LLAMACLOUD":
+                # Update notification: parsing stage
+                if notification:
+                    await NotificationService.document_processing.notify_processing_progress(
+                        session,
+                        notification,
+                        stage="parsing",
+                        stage_message="Extracting content",
+                    )
+
                 await task_logger.log_task_progress(
                     log_entry,
                     f"Processing file with LlamaCloud ETL: {filename}",
@@ -964,6 +1220,15 @@ async def process_file_in_background(
                     split_by_page=False
                 )
 
+                # Update notification: chunking stage
+                if notification:
+                    await NotificationService.document_processing.notify_processing_progress(
+                        session,
+                        notification,
+                        stage="chunking",
+                        chunks_count=len(markdown_documents),
+                    )
+
                 await task_logger.log_task_progress(
                     log_entry,
                     f"LlamaCloud parsing completed, creating documents: {filename}",
@@ -1023,6 +1288,7 @@ async def process_file_in_background(
                         llamacloud_markdown_document=markdown_content,
                         search_space_id=search_space_id,
                         user_id=user_id,
+                        connector=connector,
                     )
 
                     # Track if this document was successfully created
@@ -1056,6 +1322,7 @@ async def process_file_in_background(
                             "documents_count": len(markdown_documents),
                         },
                     )
+                    return last_created_doc
                 else:
                     # All documents were duplicates (markdown_documents was not empty, but all returned None)
                     await task_logger.log_task_success(
@@ -1068,8 +1335,18 @@ async def process_file_in_background(
                             "documents_count": len(markdown_documents),
                         },
                     )
+                    return None
 
             elif app_config.ETL_SERVICE == "DOCLING":
+                # Update notification: parsing stage
+                if notification:
+                    await NotificationService.document_processing.notify_processing_progress(
+                        session,
+                        notification,
+                        stage="parsing",
+                        stage_message="Extracting content",
+                    )
+
                 await task_logger.log_task_progress(
                     log_entry,
                     f"Processing file with Docling ETL: {filename}",
@@ -1152,6 +1429,12 @@ async def process_file_in_background(
                         },
                     )
 
+                # Update notification: chunking stage
+                if notification:
+                    await NotificationService.document_processing.notify_processing_progress(
+                        session, notification, stage="chunking"
+                    )
+
                 # Process the document using our Docling background task
                 doc_result = await add_received_file_document_using_docling(
                     session,
@@ -1159,6 +1442,7 @@ async def process_file_in_background(
                     docling_markdown_document=result["content"],
                     search_space_id=search_space_id,
                     user_id=user_id,
+                    connector=connector,
                 )
 
                 if doc_result:
@@ -1184,6 +1468,7 @@ async def process_file_in_background(
                             "pages_processed": final_page_count,
                         },
                     )
+                    return doc_result
                 else:
                     await task_logger.log_task_success(
                         log_entry,
@@ -1194,6 +1479,7 @@ async def process_file_in_background(
                             "etl_service": "DOCLING",
                         },
                     )
+                    return None
     except Exception as e:
         await session.rollback()
 

surfsense_backend/app/tasks/document_processors/markdown_processor.py

@@ -19,16 +19,157 @@ from app.utils.document_converters import (
 
 from .base import (
     check_document_by_unique_identifier,
+    check_duplicate_document,
     get_current_timestamp,
 )
 
 
+def _get_google_drive_unique_identifier(
+    connector: dict | None,
+    filename: str,
+    search_space_id: int,
+) -> tuple[str, str | None]:
+    """
+    Get unique identifier hash for a file, with special handling for Google Drive.
+
+    For Google Drive files, uses file_id as the unique identifier (doesn't change on rename).
+    For other files, uses filename.
+
+    Args:
+        connector: Optional connector info dict with type and metadata
+        filename: The filename (used for non-Google Drive files or as fallback)
+        search_space_id: The search space ID
+
+    Returns:
+        Tuple of (primary_hash, legacy_hash or None)
+    """
+    if connector and connector.get("type") == DocumentType.GOOGLE_DRIVE_FILE:
+        metadata = connector.get("metadata", {})
+        file_id = metadata.get("google_drive_file_id")
+
+        if file_id:
+            primary_hash = generate_unique_identifier_hash(
+                DocumentType.GOOGLE_DRIVE_FILE, file_id, search_space_id
+            )
+            legacy_hash = generate_unique_identifier_hash(
+                DocumentType.GOOGLE_DRIVE_FILE, filename, search_space_id
+            )
+            return primary_hash, legacy_hash
+
+    primary_hash = generate_unique_identifier_hash(
+        DocumentType.FILE, filename, search_space_id
+    )
+    return primary_hash, None
+
+
+async def _find_existing_document_with_migration(
+    session: AsyncSession,
+    primary_hash: str,
+    legacy_hash: str | None,
+    content_hash: str | None = None,
+) -> Document | None:
+    """
+    Find existing document, checking both new hash and legacy hash for migration,
+    with fallback to content_hash for cross-source deduplication.
+    """
+    existing_document = await check_document_by_unique_identifier(session, primary_hash)
+
+    if not existing_document and legacy_hash:
+        existing_document = await check_document_by_unique_identifier(
+            session, legacy_hash
+        )
+        if existing_document:
+            logging.info(
+                "Found legacy document (filename-based hash), will migrate to file_id-based hash"
+            )
+
+    # Fallback: check by content_hash to catch duplicates from different sources
+    if not existing_document and content_hash:
+        existing_document = await check_duplicate_document(session, content_hash)
+        if existing_document:
+            logging.info(
+                f"Found duplicate content from different source (content_hash match). "
+                f"Original document ID: {existing_document.id}, type: {existing_document.document_type}"
+            )
+
+    return existing_document
+
+
+async def _handle_existing_document_update(
+    session: AsyncSession,
+    existing_document: Document,
+    content_hash: str,
+    connector: dict | None,
+    filename: str,
+    primary_hash: str,
+    task_logger: TaskLoggingService,
+    log_entry,
+) -> tuple[bool, Document | None]:
+    """
+    Handle update logic for an existing document.
+
+    Returns:
+        Tuple of (should_skip_processing, document_to_return)
+    """
+    # Check if this document needs hash migration
+    if existing_document.unique_identifier_hash != primary_hash:
+        existing_document.unique_identifier_hash = primary_hash
+        logging.info(f"Migrated document to file_id-based identifier: {filename}")
+
+    # Check if content has changed
+    if existing_document.content_hash == content_hash:
+        # Content unchanged - check if we need to update metadata (e.g., filename changed)
+        if connector and connector.get("type") == DocumentType.GOOGLE_DRIVE_FILE:
+            connector_metadata = connector.get("metadata", {})
+            new_name = connector_metadata.get("google_drive_file_name")
+            # Check both possible keys for old name (FILE_NAME is used in stored documents)
+            doc_metadata = existing_document.document_metadata or {}
+            old_name = (
+                doc_metadata.get("FILE_NAME")
+                or doc_metadata.get("google_drive_file_name")
+                or doc_metadata.get("file_name")
+            )
+
+            if new_name and old_name and old_name != new_name:
+                # File was renamed - update title and metadata, skip expensive processing
+                from sqlalchemy.orm.attributes import flag_modified
+
+                existing_document.title = new_name
+                if not existing_document.document_metadata:
+                    existing_document.document_metadata = {}
+                existing_document.document_metadata["FILE_NAME"] = new_name
+                existing_document.document_metadata["file_name"] = new_name
+                existing_document.document_metadata["google_drive_file_name"] = new_name
+                flag_modified(existing_document, "document_metadata")
+                await session.commit()
+                logging.info(
+                    f"File renamed in Google Drive: '{old_name}' → '{new_name}' (no re-processing needed)"
+                )
+
+        await task_logger.log_task_success(
+            log_entry,
+            f"Markdown file document unchanged: {filename}",
+            {
+                "duplicate_detected": True,
+                "existing_document_id": existing_document.id,
+            },
+        )
+        logging.info(f"Document for markdown file {filename} unchanged. Skipping.")
+        return True, existing_document
+    else:
+        logging.info(
+            f"Content changed for markdown file {filename}. Updating document."
+        )
+        return False, None
+
+
 async def add_received_markdown_file_document(
     session: AsyncSession,
     file_name: str,
     file_in_markdown: str,
     search_space_id: int,
     user_id: str,
+    connector: dict | None = None,
 ) -> Document | None:
     """
     Process and store a markdown file document.
@@ -39,6 +180,7 @@ async def add_received_markdown_file_document(
         file_in_markdown: Content of the markdown file
         search_space_id: ID of the search space
         user_id: ID of the user
+        connector: Optional connector info for Google Drive files
 
     Returns:
         Document object if successful, None if failed
@@ -58,39 +200,34 @@ async def add_received_markdown_file_document(
     )
 
     try:
-        # Generate unique identifier hash for this markdown file
-        unique_identifier_hash = generate_unique_identifier_hash(
-            DocumentType.FILE, file_name, search_space_id
+        # Generate unique identifier hash (uses file_id for Google Drive, filename for others)
+        primary_hash, legacy_hash = _get_google_drive_unique_identifier(
+            connector, file_name, search_space_id
         )
 
         # Generate content hash
         content_hash = generate_content_hash(file_in_markdown, search_space_id)
 
-        # Check if document with this unique identifier already exists
-        existing_document = await check_document_by_unique_identifier(
-            session, unique_identifier_hash
+        # Check if document exists (with migration support for Google Drive and content_hash fallback)
+        existing_document = await _find_existing_document_with_migration(
+            session, primary_hash, legacy_hash, content_hash
         )
 
         if existing_document:
-            # Document exists - check if content has changed
-            if existing_document.content_hash == content_hash:
-                await task_logger.log_task_success(
-                    log_entry,
-                    f"Markdown file document unchanged: {file_name}",
-                    {
-                        "duplicate_detected": True,
-                        "existing_document_id": existing_document.id,
-                    },
-                )
-                logging.info(
-                    f"Document for markdown file {file_name} unchanged. Skipping."
-                )
-                return existing_document
-            else:
-                # Content has changed - update the existing document
-                logging.info(
-                    f"Content changed for markdown file {file_name}. Updating document."
-                )
+            # Handle existing document (rename detection, content change check)
+            should_skip, doc = await _handle_existing_document_update(
+                session,
+                existing_document,
+                content_hash,
+                connector,
+                file_name,
+                primary_hash,
+                task_logger,
+                log_entry,
+            )
+            if should_skip:
+                return doc
+            # Content changed - continue to update
 
         # Get user's long context LLM (needed for both create and update)
         user_llm = await get_user_long_context_llm(session, user_id, search_space_id)
@@ -139,10 +276,15 @@ async def add_received_markdown_file_document(
             document = existing_document
         else:
             # Create new document
+            # Determine document type based on connector
+            doc_type = DocumentType.FILE
+            if connector and connector.get("type") == DocumentType.GOOGLE_DRIVE_FILE:
+                doc_type = DocumentType.GOOGLE_DRIVE_FILE
+
             document = Document(
                 search_space_id=search_space_id,
                 title=file_name,
-                document_type=DocumentType.FILE,
+                document_type=doc_type,
                 document_metadata={
                     "FILE_NAME": file_name,
                 },
@@ -150,7 +292,7 @@ async def add_received_markdown_file_document(
                 embedding=summary_embedding,
                 chunks=chunks,
                 content_hash=content_hash,
-                unique_identifier_hash=unique_identifier_hash,
+                unique_identifier_hash=primary_hash,
                 blocknote_document=blocknote_json,
                 updated_at=get_current_timestamp(),
             )

surfsense_backend/scripts/docker/entrypoint.sh

@@ -11,6 +11,26 @@ cleanup() {
 
 trap cleanup SIGTERM SIGINT
 
+# Run database migrations with safeguards
+echo "Running database migrations..."
+# Wait for database to be ready (max 30 seconds)
+for i in {1..30}; do
+    if python -c "from app.db import engine; import asyncio; asyncio.run(engine.dispose())" 2>/dev/null; then
+        echo "Database is ready."
+        break
+    fi
+    echo "Waiting for database... ($i/30)"
+    sleep 1
+done
+
+# Run migrations with timeout (60 seconds max)
+if timeout 60 alembic upgrade head 2>&1; then
+    echo "Migrations completed successfully."
+else
+    echo "WARNING: Migration failed or timed out. Continuing anyway..."
+    echo "You may need to run migrations manually: alembic upgrade head"
+fi
+
 echo "Starting FastAPI Backend..."
 python main.py &
 backend_pid=$!