feat: Integrate Electric SQL for real-time notifications and enhance PostgreSQL configuration

- Added Electric SQL service to docker-compose for real-time data synchronization.
- Introduced PostgreSQL configuration for logical replication and performance tuning.
- Created scripts for initializing Electric SQL user and electrifying tables.
- Implemented notification model and service in the backend.
- Developed ElectricProvider and useNotifications hook in the frontend for managing notifications.
- Updated environment variables and package dependencies for Electric SQL integration.

surfsense_backend/alembic/versions/60_add_notifications_table.py

@@ -0,0 +1,51 @@
+"""Add notifications table
+
+Revision ID: 60
+Revises: 59
+"""
+from collections.abc import Sequence
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "60"
+down_revision: str | None = "59"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema - add notifications table."""
+    # 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
+    op.create_index("ix_notifications_user_id", "notifications", ["user_id"])
+    op.create_index("ix_notifications_read", "notifications", ["read"])
+    op.create_index("ix_notifications_created_at", "notifications", ["created_at"])
+    op.create_index("ix_notifications_user_read", "notifications", ["user_id", "read"])
+
+
+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/app/db.py

@@ -492,6 +492,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",
@@ -629,6 +635,25 @@ 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)  # 'document_processed', 'connector_indexed', 'user_mentioned', 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={})
+
+    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.
@@ -773,6 +798,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(
@@ -799,6 +830,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(

surfsense_backend/app/services/notification_service.py

@@ -0,0 +1,140 @@
+"""Service for creating and managing notifications."""
+
+import logging
+from typing import Any
+from uuid import UUID
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db import Notification
+
+logger = logging.getLogger(__name__)
+
+
+class NotificationService:
+    """Service for creating notifications that sync via Electric SQL."""
+
+    @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_processed', 'connector_indexed')
+            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
+
+    @staticmethod
+    async def create_document_processed_notification(
+        session: AsyncSession,
+        user_id: UUID,
+        document_id: int,
+        document_title: str,
+        status: str,
+        search_space_id: int,
+    ) -> Notification:
+        """
+        Create notification when document processing completes.
+
+        Args:
+            session: Database session
+            user_id: User to notify
+            document_id: ID of the processed document
+            document_title: Title of the document
+            status: Processing status ('SUCCESS', 'FAILED')
+            search_space_id: Search space ID
+
+        Returns:
+            Notification: The created notification
+        """
+        status_lower = status.lower()
+        title = f"Document processed: {document_title}"
+        message = f'Your document "{document_title}" has been {status_lower}.'
+
+        return await NotificationService.create_notification(
+            session=session,
+            user_id=user_id,
+            notification_type="document_processed",
+            title=title,
+            message=message,
+            search_space_id=search_space_id,
+            notification_metadata={
+                "document_id": document_id,
+                "status": status,
+            },
+        )
+
+    @staticmethod
+    async def create_connector_indexed_notification(
+        session: AsyncSession,
+        user_id: UUID,
+        connector_name: str,
+        connector_type: str,
+        status: str,
+        search_space_id: int,
+        indexed_count: int | None = None,
+    ) -> Notification:
+        """
+        Create notification when connector indexing completes.
+
+        Args:
+            session: Database session
+            user_id: User to notify
+            connector_name: Name of the connector
+            connector_type: Type of connector
+            status: Indexing status ('SUCCESS', 'FAILED')
+            search_space_id: Search space ID
+            indexed_count: Number of items indexed (optional)
+
+        Returns:
+            Notification: The created notification
+        """
+        status_lower = status.lower()
+        title = f"Connector indexed: {connector_name}"
+        message = f'Your connector "{connector_name}" has finished indexing ({status_lower}).'
+        if indexed_count is not None:
+            message += f" {indexed_count} items indexed."
+
+        return await NotificationService.create_notification(
+            session=session,
+            user_id=user_id,
+            notification_type="connector_indexed",
+            title=title,
+            message=message,
+            search_space_id=search_space_id,
+            notification_metadata={
+                "connector_name": connector_name,
+                "connector_type": connector_type,
+                "status": status,
+                "indexed_count": indexed_count,
+            },
+        )