feat: implement Confluence OAuth integration and connector routes

- Added support for Confluence OAuth with new environment variables for client ID, client secret, and redirect URI.
- Implemented Confluence connector routes for OAuth flow, including authorization and callback handling.
- Enhanced Confluence connector to support both OAuth 2.0 and legacy API token authentication methods.
- Updated Confluence indexing logic to utilize OAuth credentials with auto-refresh capabilities.
- Removed outdated Confluence UI components and adjusted frontend logic to reflect the new integration.

surfsense_backend/app/config/__init__.py

@@ -95,10 +95,11 @@ class Config:
     NOTION_CLIENT_SECRET = os.getenv("NOTION_CLIENT_SECRET")
     NOTION_REDIRECT_URI = os.getenv("NOTION_REDIRECT_URI")
 
-    # Jira OAuth
-    JIRA_CLIENT_ID = os.getenv("JIRA_CLIENT_ID")
-    JIRA_CLIENT_SECRET = os.getenv("JIRA_CLIENT_SECRET")
+    # Atlassian OAuth (shared for Jira and Confluence)
+    ATLASSIAN_CLIENT_ID = os.getenv("ATLASSIAN_CLIENT_ID")
+    ATLASSIAN_CLIENT_SECRET = os.getenv("ATLASSIAN_CLIENT_SECRET")
     JIRA_REDIRECT_URI = os.getenv("JIRA_REDIRECT_URI")
+    CONFLUENCE_REDIRECT_URI = os.getenv("CONFLUENCE_REDIRECT_URI")
 
     # Linear OAuth
     LINEAR_CLIENT_ID = os.getenv("LINEAR_CLIENT_ID")

surfsense_backend/app/connectors/confluence_history.py

@@ -0,0 +1,488 @@
+"""
+Confluence OAuth Connector.
+
+Handles OAuth-based authentication and token refresh for Confluence API access.
+"""
+
+import logging
+from typing import Any
+
+import httpx
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+
+from app.config import config
+from app.db import SearchSourceConnector
+from app.routes.confluence_add_connector_route import refresh_confluence_token
+from app.schemas.atlassian_auth_credentials import AtlassianAuthCredentialsBase
+from app.utils.oauth_security import TokenEncryption
+
+logger = logging.getLogger(__name__)
+
+
+class ConfluenceHistoryConnector:
+    """
+    Confluence connector with OAuth support and automatic token refresh.
+    
+    This connector uses OAuth 2.0 access tokens to authenticate with the
+    Confluence API. It automatically refreshes expired tokens when needed.
+    """
+
+    def __init__(
+        self,
+        session: AsyncSession,
+        connector_id: int,
+        credentials: AtlassianAuthCredentialsBase | None = None,
+    ):
+        """
+        Initialize the ConfluenceHistoryConnector with auto-refresh capability.
+
+        Args:
+            session: Database session for updating connector
+            connector_id: Connector ID for direct updates
+            credentials: Confluence OAuth credentials (optional, will be loaded from DB if not provided)
+        """
+        self._session = session
+        self._connector_id = connector_id
+        self._credentials = credentials
+        self._cloud_id: str | None = None
+        self._base_url: str | None = None
+        self._http_client: httpx.AsyncClient | None = None
+
+    async def _get_valid_token(self) -> str:
+        """
+        Get valid Confluence access token, refreshing if needed.
+
+        Returns:
+            Valid access token
+
+        Raises:
+            ValueError: If credentials are missing or invalid
+            Exception: If token refresh fails
+        """
+        # Load credentials from DB if not provided
+        if self._credentials is None:
+            result = await self._session.execute(
+                select(SearchSourceConnector).filter(
+                    SearchSourceConnector.id == self._connector_id
+                )
+            )
+            connector = result.scalars().first()
+
+            if not connector:
+                raise ValueError(f"Connector {self._connector_id} not found")
+
+            config_data = connector.config.copy()
+
+            # Decrypt credentials if they are encrypted
+            token_encrypted = config_data.get("_token_encrypted", False)
+            if token_encrypted and config.SECRET_KEY:
+                try:
+                    token_encryption = TokenEncryption(config.SECRET_KEY)
+
+                    # Decrypt sensitive fields
+                    if config_data.get("access_token"):
+                        config_data["access_token"] = token_encryption.decrypt_token(
+                            config_data["access_token"]
+                        )
+                    if config_data.get("refresh_token"):
+                        config_data["refresh_token"] = token_encryption.decrypt_token(
+                            config_data["refresh_token"]
+                        )
+
+                    logger.info(
+                        f"Decrypted Confluence credentials for connector {self._connector_id}"
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Failed to decrypt Confluence credentials for connector {self._connector_id}: {e!s}"
+                    )
+                    raise ValueError(
+                        f"Failed to decrypt Confluence credentials: {e!s}"
+                    ) from e
+
+            try:
+                self._credentials = AtlassianAuthCredentialsBase.from_dict(config_data)
+                # Store cloud_id and base_url for API calls (with backward compatibility for site_url)
+                self._cloud_id = config_data.get("cloud_id")
+                self._base_url = config_data.get("base_url") or config_data.get("site_url")
+            except Exception as e:
+                raise ValueError(f"Invalid Confluence credentials: {e!s}") from e
+
+        # Check if token is expired and refreshable
+        if self._credentials.is_expired and self._credentials.is_refreshable:
+            try:
+                logger.info(
+                    f"Confluence token expired for connector {self._connector_id}, refreshing..."
+                )
+
+                # Get connector for refresh
+                result = await self._session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.id == self._connector_id
+                    )
+                )
+                connector = result.scalars().first()
+
+                if not connector:
+                    raise RuntimeError(
+                        f"Connector {self._connector_id} not found; cannot refresh token."
+                    )
+
+                # Refresh token
+                connector = await refresh_confluence_token(self._session, connector)
+
+                # Reload credentials after refresh
+                config_data = connector.config.copy()
+                token_encrypted = config_data.get("_token_encrypted", False)
+                if token_encrypted and config.SECRET_KEY:
+                    token_encryption = TokenEncryption(config.SECRET_KEY)
+                    if config_data.get("access_token"):
+                        config_data["access_token"] = token_encryption.decrypt_token(
+                            config_data["access_token"]
+                        )
+                    if config_data.get("refresh_token"):
+                        config_data["refresh_token"] = token_encryption.decrypt_token(
+                            config_data["refresh_token"]
+                        )
+
+                self._credentials = AtlassianAuthCredentialsBase.from_dict(config_data)
+                self._cloud_id = config_data.get("cloud_id")
+                # Handle backward compatibility: check both base_url and site_url
+                self._base_url = config_data.get("base_url") or config_data.get("site_url")
+
+                # Invalidate cached client so it's recreated with new token
+                if self._http_client:
+                    await self._http_client.aclose()
+                    self._http_client = None
+
+                logger.info(
+                    f"Successfully refreshed Confluence token for connector {self._connector_id}"
+                )
+            except Exception as e:
+                logger.error(
+                    f"Failed to refresh Confluence token for connector {self._connector_id}: {e!s}"
+                )
+                raise Exception(
+                    f"Failed to refresh Confluence OAuth credentials: {e!s}"
+                ) from e
+
+        return self._credentials.access_token
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        """
+        Get or create HTTP client with valid token.
+
+        Returns:
+            httpx.AsyncClient instance
+        """
+        if self._http_client is None:
+            self._http_client = httpx.AsyncClient(timeout=30.0)
+        return self._http_client
+
+    async def _get_base_url(self) -> str:
+        """
+        Get the base URL for Confluence API calls.
+
+        Returns:
+            Base URL string
+        """
+        if not self._cloud_id:
+            raise ValueError("Cloud ID not available. Cannot construct API URL.")
+        
+        # Use the Atlassian API format: https://api.atlassian.com/ex/confluence/{cloudid}
+        return f"https://api.atlassian.com/ex/confluence/{self._cloud_id}"
+
+    async def _make_api_request(
+        self, endpoint: str, params: dict[str, Any] | None = None
+    ) -> dict[str, Any]:
+        """
+        Make a request to the Confluence API.
+
+        Args:
+            endpoint: API endpoint (without base URL)
+            params: Query parameters for the request (optional)
+
+        Returns:
+            Response data from the API
+
+        Raises:
+            ValueError: If credentials have not been set
+            Exception: If the API request fails
+        """
+        token = await self._get_valid_token()
+        base_url = await self._get_base_url()
+        client = await self._get_client()
+
+        url = f"{base_url}/wiki/api/v2/{endpoint}"
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}",
+            "Accept": "application/json",
+        }
+
+        try:
+            response = await client.get(url, headers=headers, params=params)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            # Enhanced error logging to see the actual error
+            error_detail = {
+                "status_code": e.response.status_code,
+                "url": str(e.request.url),
+                "response_text": e.response.text,
+                "headers": dict(e.response.headers),
+            }
+            logger.error(f"Confluence API HTTP error: {error_detail}")
+            raise Exception(
+                f"Confluence API request failed (HTTP {e.response.status_code}): {e.response.text}"
+            ) from e
+        except httpx.RequestError as e:
+            logger.error(f"Confluence API request error: {e!s}", exc_info=True)
+            raise Exception(f"Confluence API request failed: {e!s}") from e
+
+    async def get_all_spaces(self) -> list[dict[str, Any]]:
+        """
+        Fetch all spaces from Confluence.
+
+        Returns:
+            List of space objects
+
+        Raises:
+            ValueError: If credentials have not been set
+            Exception: If the API request fails
+        """
+        params = {
+            "limit": 100,
+        }
+
+        all_spaces = []
+        cursor = None
+
+        while True:
+            if cursor:
+                params["cursor"] = cursor
+
+            result = await self._make_api_request("spaces", params)
+
+            if not isinstance(result, dict) or "results" not in result:
+                raise Exception("Invalid response from Confluence API")
+
+            spaces = result["results"]
+            all_spaces.extend(spaces)
+
+            # Check if there are more spaces to fetch
+            links = result.get("_links", {})
+            if "next" not in links:
+                break
+
+            # Extract cursor from next link if available
+            next_link = links["next"]
+            if "cursor=" in next_link:
+                cursor = next_link.split("cursor=")[1].split("&")[0]
+            else:
+                break
+
+        return all_spaces
+
+    async def get_pages_in_space(
+        self, space_id: str, include_body: bool = True
+    ) -> list[dict[str, Any]]:
+        """
+        Fetch all pages in a specific space.
+
+        Args:
+            space_id: The ID of the space to fetch pages from
+            include_body: Whether to include page body content
+
+        Returns:
+            List of page objects
+
+        Raises:
+            ValueError: If credentials have not been set
+            Exception: If the API request fails
+        """
+        params = {
+            "limit": 100,
+        }
+
+        if include_body:
+            params["body-format"] = "storage"
+
+        all_pages = []
+        cursor = None
+
+        while True:
+            if cursor:
+                params["cursor"] = cursor
+
+            result = await self._make_api_request(f"spaces/{space_id}/pages", params)
+
+            if not isinstance(result, dict) or "results" not in result:
+                raise Exception("Invalid response from Confluence API")
+
+            pages = result["results"]
+            all_pages.extend(pages)
+
+            # Check if there are more pages to fetch
+            links = result.get("_links", {})
+            if "next" not in links:
+                break
+
+            # Extract cursor from next link if available
+            next_link = links["next"]
+            if "cursor=" in next_link:
+                cursor = next_link.split("cursor=")[1].split("&")[0]
+            else:
+                break
+
+        return all_pages
+
+    async def get_page_comments(self, page_id: str) -> list[dict[str, Any]]:
+        """
+        Fetch all comments for a specific page (both footer and inline comments).
+
+        Args:
+            page_id: The ID of the page to fetch comments from
+
+        Returns:
+            List of comment objects
+
+        Raises:
+            ValueError: If credentials have not been set
+            Exception: If the API request fails
+        """
+        all_comments = []
+
+        # Get footer comments
+        footer_comments = await self._get_comments_for_page(page_id, "footer-comments")
+        all_comments.extend(footer_comments)
+
+        # Get inline comments
+        inline_comments = await self._get_comments_for_page(page_id, "inline-comments")
+        all_comments.extend(inline_comments)
+
+        return all_comments
+
+    async def _get_comments_for_page(
+        self, page_id: str, comment_type: str
+    ) -> list[dict[str, Any]]:
+        """
+        Helper method to fetch comments of a specific type for a page.
+
+        Args:
+            page_id: The ID of the page
+            comment_type: Type of comments ('footer-comments' or 'inline-comments')
+
+        Returns:
+            List of comment objects
+        """
+        params = {
+            "limit": 100,
+            "body-format": "storage",
+        }
+
+        all_comments = []
+        cursor = None
+
+        while True:
+            if cursor:
+                params["cursor"] = cursor
+
+            result = await self._make_api_request(f"pages/{page_id}/{comment_type}", params)
+
+            if not isinstance(result, dict) or "results" not in result:
+                break  # No comments or invalid response
+
+            comments = result["results"]
+            all_comments.extend(comments)
+
+            # Check if there are more comments to fetch
+            links = result.get("_links", {})
+            if "next" not in links:
+                break
+
+            # Extract cursor from next link if available
+            next_link = links["next"]
+            if "cursor=" in next_link:
+                cursor = next_link.split("cursor=")[1].split("&")[0]
+            else:
+                break
+
+        return all_comments
+
+    async def get_pages_by_date_range(
+        self,
+        start_date: str,
+        end_date: str,
+        space_ids: list[str] | None = None,
+        include_comments: bool = True,
+    ) -> tuple[list[dict[str, Any]], str | None]:
+        """
+        Fetch pages within a date range, optionally filtered by spaces.
+
+        Args:
+            start_date: Start date in YYYY-MM-DD format
+            end_date: End date in YYYY-MM-DD format (inclusive)
+            space_ids: Optional list of space IDs to filter pages
+            include_comments: Whether to include comments for each page
+
+        Returns:
+            Tuple containing (pages list with comments, error message or None)
+        """
+        try:
+            all_pages = []
+
+            if space_ids:
+                # Fetch pages from specific spaces
+                for space_id in space_ids:
+                    pages = await self.get_pages_in_space(space_id, include_body=True)
+                    all_pages.extend(pages)
+            else:
+                # Fetch all pages (this might be expensive for large instances)
+                params = {
+                    "limit": 100,
+                    "body-format": "storage",
+                }
+
+                cursor = None
+                while True:
+                    if cursor:
+                        params["cursor"] = cursor
+
+                    result = await self._make_api_request("pages", params)
+                    if not isinstance(result, dict) or "results" not in result:
+                        break
+
+                    pages = result["results"]
+                    all_pages.extend(pages)
+
+                    links = result.get("_links", {})
+                    if "next" not in links:
+                        break
+
+                    next_link = links["next"]
+                    if "cursor=" in next_link:
+                        cursor = next_link.split("cursor=")[1].split("&")[0]
+                    else:
+                        break
+
+            return all_pages, None
+
+        except Exception as e:
+            return [], f"Error fetching pages: {e!s}"
+
+    async def close(self):
+        """Close the HTTP client connection."""
+        if self._http_client:
+            await self._http_client.aclose()
+            self._http_client = None
+
+    async def __aenter__(self):
+        """Async context manager entry."""
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit."""
+        await self.close()
+

surfsense_backend/app/routes/__init__.py

@@ -17,6 +17,7 @@ from .google_gmail_add_connector_route import (
     router as google_gmail_add_connector_router,
 )
 from .jira_add_connector_route import router as jira_add_connector_router
+from .confluence_add_connector_route import router as confluence_add_connector_router
 from .linear_add_connector_route import router as linear_add_connector_router
 from .logs_routes import router as logs_router
 from .luma_add_connector_route import router as luma_add_connector_router
@@ -50,6 +51,7 @@ router.include_router(notion_add_connector_router)
 router.include_router(slack_add_connector_router)
 router.include_router(discord_add_connector_router)
 router.include_router(jira_add_connector_router)
+router.include_router(confluence_add_connector_router)
 router.include_router(new_llm_config_router)  # LLM configs with prompt configuration
 router.include_router(logs_router)
 router.include_router(circleback_webhook_router)  # Circleback meeting webhooks

surfsense_backend/app/routes/confluence_add_connector_route.py

@@ -0,0 +1,473 @@
+"""
+Confluence Connector OAuth Routes.
+
+Handles OAuth 2.0 authentication flow for Confluence connector.
+"""
+
+import logging
+from datetime import UTC, datetime, timedelta
+from uuid import UUID
+
+import httpx
+from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.responses import RedirectResponse
+from pydantic import ValidationError
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+
+from app.config import config
+from app.db import (
+    SearchSourceConnector,
+    SearchSourceConnectorType,
+    User,
+    get_async_session,
+)
+from app.schemas.atlassian_auth_credentials import AtlassianAuthCredentialsBase
+from app.users import current_active_user
+from app.utils.oauth_security import OAuthStateManager, TokenEncryption
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# Atlassian OAuth endpoints
+AUTHORIZATION_URL = "https://auth.atlassian.com/authorize"
+TOKEN_URL = "https://auth.atlassian.com/oauth/token"
+RESOURCES_URL = "https://api.atlassian.com/oauth/token/accessible-resources"
+
+# OAuth scopes for Confluence
+SCOPES = [
+    "read:confluence-content.all",
+    "read:confluence-space.summary",
+    "read:confluence-user",
+    "read:space:confluence",
+    "read:page:confluence",
+    "read:comment:confluence",
+    "offline_access",  # Required for refresh tokens
+]
+
+# Initialize security utilities
+_state_manager = None
+_token_encryption = None
+
+
+def get_state_manager() -> OAuthStateManager:
+    """Get or create OAuth state manager instance."""
+    global _state_manager
+    if _state_manager is None:
+        if not config.SECRET_KEY:
+            raise ValueError("SECRET_KEY must be set for OAuth security")
+        _state_manager = OAuthStateManager(config.SECRET_KEY)
+    return _state_manager
+
+
+def get_token_encryption() -> TokenEncryption:
+    """Get or create token encryption instance."""
+    global _token_encryption
+    if _token_encryption is None:
+        if not config.SECRET_KEY:
+            raise ValueError("SECRET_KEY must be set for token encryption")
+        _token_encryption = TokenEncryption(config.SECRET_KEY)
+    return _token_encryption
+
+
+@router.get("/auth/confluence/connector/add")
+async def connect_confluence(space_id: int, user: User = Depends(current_active_user)):
+    """
+    Initiate Confluence OAuth flow.
+
+    Args:
+        space_id: The search space ID
+        user: Current authenticated user
+
+    Returns:
+        Authorization URL for redirect
+    """
+    try:
+        if not space_id:
+            raise HTTPException(status_code=400, detail="space_id is required")
+
+        if not config.ATLASSIAN_CLIENT_ID:
+            raise HTTPException(status_code=500, detail="Atlassian OAuth not configured.")
+
+        if not config.SECRET_KEY:
+            raise HTTPException(
+                status_code=500, detail="SECRET_KEY not configured for OAuth security."
+            )
+
+        # Generate secure state parameter with HMAC signature
+        state_manager = get_state_manager()
+        state_encoded = state_manager.generate_secure_state(space_id, user.id)
+
+        # Build authorization URL
+        from urllib.parse import urlencode
+
+        auth_params = {
+            "audience": "api.atlassian.com",
+            "client_id": config.ATLASSIAN_CLIENT_ID,
+            "scope": " ".join(SCOPES),
+            "redirect_uri": config.CONFLUENCE_REDIRECT_URI,
+            "state": state_encoded,
+            "response_type": "code",
+            "prompt": "consent",
+        }
+
+        auth_url = f"{AUTHORIZATION_URL}?{urlencode(auth_params)}"
+
+        logger.info(f"Generated Confluence OAuth URL for user {user.id}, space {space_id}")
+        return {"auth_url": auth_url}
+
+    except Exception as e:
+        logger.error(f"Failed to initiate Confluence OAuth: {e!s}", exc_info=True)
+        raise HTTPException(
+            status_code=500, detail=f"Failed to initiate Confluence OAuth: {e!s}"
+        ) from e
+
+
+@router.get("/auth/confluence/connector/callback")
+async def confluence_callback(
+    request: Request,
+    code: str | None = None,
+    error: str | None = None,
+    state: str | None = None,
+    session: AsyncSession = Depends(get_async_session),
+):
+    """
+    Handle Confluence OAuth callback.
+
+    Args:
+        request: FastAPI request object
+        code: Authorization code from Atlassian (if user granted access)
+        error: Error code from Atlassian (if user denied access or error occurred)
+        state: State parameter containing user/space info
+        session: Database session
+
+    Returns:
+        Redirect response to frontend
+    """
+    try:
+        # Handle OAuth errors (e.g., user denied access)
+        if error:
+            logger.warning(f"Confluence OAuth error: {error}")
+            # Try to decode state to get space_id for redirect, but don't fail if it's invalid
+            space_id = None
+            if state:
+                try:
+                    state_manager = get_state_manager()
+                    data = state_manager.validate_state(state)
+                    space_id = data.get("space_id")
+                except Exception:
+                    # If state is invalid, we'll redirect without space_id
+                    logger.warning("Failed to validate state in error handler")
+
+            # Redirect to frontend with error parameter
+            if space_id:
+                return RedirectResponse(
+                    url=f"{config.NEXT_FRONTEND_URL}/dashboard/{space_id}/new-chat?modal=connectors&tab=all&error=confluence_oauth_denied"
+                )
+            else:
+                return RedirectResponse(
+                    url=f"{config.NEXT_FRONTEND_URL}/dashboard?error=confluence_oauth_denied"
+                )
+
+        # Validate required parameters for successful flow
+        if not code:
+            raise HTTPException(status_code=400, detail="Missing authorization code")
+        if not state:
+            raise HTTPException(status_code=400, detail="Missing state parameter")
+
+        # Validate and decode state with signature verification
+        state_manager = get_state_manager()
+        try:
+            data = state_manager.validate_state(state)
+        except HTTPException:
+            raise
+        except Exception as e:
+            raise HTTPException(
+                status_code=400, detail=f"Invalid state parameter: {e!s}"
+            ) from e
+
+        user_id = UUID(data["user_id"])
+        space_id = data["space_id"]
+
+        # Validate redirect URI (security: ensure it matches configured value)
+        if not config.CONFLUENCE_REDIRECT_URI:
+            raise HTTPException(
+                status_code=500, detail="CONFLUENCE_REDIRECT_URI not configured"
+            )
+
+        # Exchange authorization code for access token
+        token_data = {
+            "grant_type": "authorization_code",
+            "client_id": config.ATLASSIAN_CLIENT_ID,
+            "client_secret": config.ATLASSIAN_CLIENT_SECRET,
+            "code": code,
+            "redirect_uri": config.CONFLUENCE_REDIRECT_URI,
+        }
+
+        async with httpx.AsyncClient() as client:
+            token_response = await client.post(
+                TOKEN_URL,
+                json=token_data,
+                headers={"Content-Type": "application/json"},
+                timeout=30.0,
+            )
+
+        if token_response.status_code != 200:
+            error_detail = token_response.text
+            try:
+                error_json = token_response.json()
+                error_detail = error_json.get("error_description", error_json.get("error", error_detail))
+            except Exception:
+                pass
+            raise HTTPException(
+                status_code=400, detail=f"Token exchange failed: {error_detail}"
+            )
+
+        token_json = token_response.json()
+
+        access_token = token_json.get("access_token")
+        refresh_token = token_json.get("refresh_token")
+        if not access_token:
+            raise HTTPException(
+                status_code=400, detail="No access token received from Atlassian"
+            )
+
+        # Get accessible resources to find Confluence cloud ID and site URL
+        async with httpx.AsyncClient() as client:
+            resources_response = await client.get(
+                RESOURCES_URL,
+                headers={"Authorization": f"Bearer {access_token}"},
+                timeout=30.0,
+            )
+
+        cloud_id = None
+        site_url = None
+        if resources_response.status_code == 200:
+            resources = resources_response.json()
+            # Find Confluence resource
+            for resource in resources:
+                if resource.get("id") and resource.get("name"):
+                    cloud_id = resource.get("id")
+                    site_url = resource.get("url")
+                    break
+
+        if not cloud_id:
+            logger.warning("Could not determine Confluence cloud ID from accessible resources")
+
+        # Calculate expiration time (UTC, tz-aware)
+        expires_at = None
+        expires_in = token_json.get("expires_in")
+        if expires_in:
+            now_utc = datetime.now(UTC)
+            expires_at = now_utc + timedelta(seconds=int(expires_in))
+
+        # Encrypt sensitive tokens before storing
+        token_encryption = get_token_encryption()
+
+        # Store the encrypted tokens and metadata in connector config
+        connector_config = {
+            "access_token": token_encryption.encrypt_token(access_token),
+            "refresh_token": token_encryption.encrypt_token(refresh_token)
+            if refresh_token
+            else None,
+            "token_type": token_json.get("token_type", "Bearer"),
+            "expires_in": expires_in,
+            "expires_at": expires_at.isoformat() if expires_at else None,
+            "scope": token_json.get("scope"),
+            "cloud_id": cloud_id,
+            "base_url": site_url,  # Store as base_url to match shared schema
+            # Mark that tokens are encrypted for backward compatibility
+            "_token_encrypted": True,
+        }
+
+        # Check if connector already exists for this search space and user
+        existing_connector_result = await session.execute(
+            select(SearchSourceConnector).filter(
+                SearchSourceConnector.search_space_id == space_id,
+                SearchSourceConnector.user_id == user_id,
+                SearchSourceConnector.connector_type
+                == SearchSourceConnectorType.CONFLUENCE_CONNECTOR,
+            )
+        )
+        existing_connector = existing_connector_result.scalars().first()
+
+        if existing_connector:
+            # Update existing connector
+            existing_connector.config = connector_config
+            existing_connector.name = "Confluence Connector"
+            existing_connector.is_indexable = True
+            logger.info(
+                f"Updated existing Confluence connector for user {user_id} in space {space_id}"
+            )
+        else:
+            # Create new connector
+            new_connector = SearchSourceConnector(
+                name="Confluence Connector",
+                connector_type=SearchSourceConnectorType.CONFLUENCE_CONNECTOR,
+                is_indexable=True,
+                config=connector_config,
+                search_space_id=space_id,
+                user_id=user_id,
+            )
+            session.add(new_connector)
+            logger.info(
+                f"Created new Confluence connector for user {user_id} in space {space_id}"
+            )
+
+        try:
+            await session.commit()
+            logger.info(f"Successfully saved Confluence connector for user {user_id}")
+
+            # Redirect to the frontend with success params
+            return RedirectResponse(
+                url=f"{config.NEXT_FRONTEND_URL}/dashboard/{space_id}/new-chat?modal=connectors&tab=all&success=true&connector=confluence-connector"
+            )
+
+        except ValidationError as e:
+            await session.rollback()
+            raise HTTPException(
+                status_code=422, detail=f"Validation error: {e!s}"
+            ) from e
+        except IntegrityError as e:
+            await session.rollback()
+            raise HTTPException(
+                status_code=409,
+                detail=f"Integrity error: A connector with this type already exists. {e!s}",
+            ) from e
+        except Exception as e:
+            logger.error(f"Failed to create search source connector: {e!s}")
+            await session.rollback()
+            raise HTTPException(
+                status_code=500,
+                detail=f"Failed to create search source connector: {e!s}",
+            ) from e
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Failed to complete Confluence OAuth: {e!s}", exc_info=True)
+        raise HTTPException(
+            status_code=500, detail=f"Failed to complete Confluence OAuth: {e!s}"
+        ) from e
+
+
+async def refresh_confluence_token(
+    session: AsyncSession, connector: SearchSourceConnector
+) -> SearchSourceConnector:
+    """
+    Refresh the Confluence access token for a connector.
+
+    Args:
+        session: Database session
+        connector: Confluence connector to refresh
+
+    Returns:
+        Updated connector object
+    """
+    try:
+        logger.info(f"Refreshing Confluence token for connector {connector.id}")
+
+        credentials = AtlassianAuthCredentialsBase.from_dict(connector.config)
+
+        # Decrypt tokens if they are encrypted
+        token_encryption = get_token_encryption()
+        is_encrypted = connector.config.get("_token_encrypted", False)
+
+        refresh_token = credentials.refresh_token
+        if is_encrypted and refresh_token:
+            try:
+                refresh_token = token_encryption.decrypt_token(refresh_token)
+            except Exception as e:
+                logger.error(f"Failed to decrypt refresh token: {e!s}")
+                raise HTTPException(
+                    status_code=500, detail="Failed to decrypt stored refresh token"
+                ) from e
+
+        if not refresh_token:
+            raise HTTPException(
+                status_code=400,
+                detail="No refresh token available. Please re-authenticate.",
+            )
+
+        # Prepare token refresh data
+        refresh_data = {
+            "grant_type": "refresh_token",
+            "client_id": config.ATLASSIAN_CLIENT_ID,
+            "client_secret": config.ATLASSIAN_CLIENT_SECRET,
+            "refresh_token": refresh_token,
+        }
+
+        async with httpx.AsyncClient() as client:
+            token_response = await client.post(
+                TOKEN_URL,
+                json=refresh_data,
+                headers={"Content-Type": "application/json"},
+                timeout=30.0,
+            )
+
+        if token_response.status_code != 200:
+            error_detail = token_response.text
+            try:
+                error_json = token_response.json()
+                error_detail = error_json.get("error_description", error_json.get("error", error_detail))
+            except Exception:
+                pass
+            raise HTTPException(
+                status_code=400, detail=f"Token refresh failed: {error_detail}"
+            )
+
+        token_json = token_response.json()
+
+        # Calculate expiration time (UTC, tz-aware)
+        expires_at = None
+        expires_in = token_json.get("expires_in")
+        if expires_in:
+            now_utc = datetime.now(UTC)
+            expires_at = now_utc + timedelta(seconds=int(expires_in))
+
+        # Encrypt new tokens before storing
+        access_token = token_json.get("access_token")
+        new_refresh_token = token_json.get("refresh_token")
+
+        if not access_token:
+            raise HTTPException(
+                status_code=400, detail="No access token received from Confluence refresh"
+            )
+
+        # Update credentials object with encrypted tokens
+        credentials.access_token = token_encryption.encrypt_token(access_token)
+        if new_refresh_token:
+            credentials.refresh_token = token_encryption.encrypt_token(
+                new_refresh_token
+            )
+        credentials.expires_in = expires_in
+        credentials.expires_at = expires_at
+        credentials.scope = token_json.get("scope")
+
+        # Preserve cloud_id and base_url (with backward compatibility for site_url)
+        if not credentials.cloud_id:
+            credentials.cloud_id = connector.config.get("cloud_id")
+        if not credentials.base_url:
+            # Check both base_url and site_url for backward compatibility
+            credentials.base_url = connector.config.get("base_url") or connector.config.get("site_url")
+
+        # Update connector config with encrypted tokens
+        credentials_dict = credentials.to_dict()
+        credentials_dict["_token_encrypted"] = True
+        connector.config = credentials_dict
+        await session.commit()
+        await session.refresh(connector)
+
+        logger.info(f"Successfully refreshed Confluence token for connector {connector.id}")
+
+        return connector
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Failed to refresh Confluence token: {e!s}", exc_info=True)
+        raise HTTPException(
+            status_code=500, detail=f"Failed to refresh Confluence token: {e!s}"
+        ) from e
+

surfsense_backend/app/routes/jira_add_connector_route.py

@@ -86,8 +86,8 @@ async def connect_jira(space_id: int, user: User = Depends(current_active_user))
         if not space_id:
             raise HTTPException(status_code=400, detail="space_id is required")
 
-        if not config.JIRA_CLIENT_ID:
-            raise HTTPException(status_code=500, detail="Jira OAuth not configured.")
+        if not config.ATLASSIAN_CLIENT_ID:
+            raise HTTPException(status_code=500, detail="Atlassian OAuth not configured.")
 
         if not config.SECRET_KEY:
             raise HTTPException(
@@ -103,7 +103,7 @@ async def connect_jira(space_id: int, user: User = Depends(current_active_user))
 
         auth_params = {
             "audience": "api.atlassian.com",
-            "client_id": config.JIRA_CLIENT_ID,
+            "client_id": config.ATLASSIAN_CLIENT_ID,
             "scope": " ".join(SCOPES),
             "redirect_uri": config.JIRA_REDIRECT_URI,
             "state": state_encoded,
@@ -198,8 +198,8 @@ async def jira_callback(
         # Exchange authorization code for access token
         token_data = {
             "grant_type": "authorization_code",
-            "client_id": config.JIRA_CLIENT_ID,
-            "client_secret": config.JIRA_CLIENT_SECRET,
+            "client_id": config.ATLASSIAN_CLIENT_ID,
+            "client_secret": config.ATLASSIAN_CLIENT_SECRET,
             "code": code,
             "redirect_uri": config.JIRA_REDIRECT_URI,
         }
@@ -417,8 +417,8 @@ async def refresh_jira_token(
         # Prepare token refresh data
         refresh_data = {
             "grant_type": "refresh_token",
-            "client_id": config.JIRA_CLIENT_ID,
-            "client_secret": config.JIRA_CLIENT_SECRET,
+            "client_id": config.ATLASSIAN_CLIENT_ID,
+            "client_secret": config.ATLASSIAN_CLIENT_SECRET,
             "refresh_token": refresh_token,
         }
 

surfsense_backend/app/tasks/connector_indexers/confluence_indexer.py

@@ -8,7 +8,7 @@ from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import config
-from app.connectors.confluence_connector import ConfluenceConnector
+from app.connectors.confluence_history import ConfluenceHistoryConnector
 from app.db import Document, DocumentType, SearchSourceConnectorType
 from app.services.llm_service import get_user_long_context_llm
 from app.services.task_logging_service import TaskLoggingService
@@ -83,31 +83,16 @@ async def index_confluence_pages(
             )
             return 0, f"Connector with ID {connector_id} not found"
 
-        # Get the Confluence credentials from the connector config
-        confluence_email = connector.config.get("CONFLUENCE_EMAIL")
-        confluence_api_token = connector.config.get("CONFLUENCE_API_TOKEN")
-        confluence_base_url = connector.config.get("CONFLUENCE_BASE_URL")
-
-        if not confluence_email or not confluence_api_token or not confluence_base_url:
-            await task_logger.log_task_failure(
-                log_entry,
-                f"Confluence credentials not found in connector config for connector {connector_id}",
-                "Missing Confluence credentials",
-                {"error_type": "MissingCredentials"},
-            )
-            return 0, "Confluence credentials not found in connector config"
-
-        # Initialize Confluence client
+        # Initialize Confluence OAuth client
         await task_logger.log_task_progress(
             log_entry,
-            f"Initializing Confluence client for connector {connector_id}",
+            f"Initializing Confluence OAuth client for connector {connector_id}",
             {"stage": "client_initialization"},
         )
 
-        confluence_client = ConfluenceConnector(
-            base_url=confluence_base_url,
-            email=confluence_email,
-            api_token=confluence_api_token,
+        confluence_client: ConfluenceHistoryConnector | None = ConfluenceHistoryConnector(
+            session=session,
+            connector_id=connector_id,
         )
 
         # Calculate date range
@@ -127,7 +112,7 @@ async def index_confluence_pages(
 
         # Get pages within date range
         try:
-            pages, error = confluence_client.get_pages_by_date_range(
+            pages, error = await confluence_client.get_pages_by_date_range(
                 start_date=start_date_str, end_date=end_date_str, include_comments=True
             )
 
@@ -153,6 +138,12 @@ async def index_confluence_pages(
                         f"No Confluence pages found in date range {start_date_str} to {end_date_str}",
                         {"pages_found": 0},
                     )
+                    # Close client before returning
+                    if confluence_client:
+                        try:
+                            await confluence_client.close()
+                        except Exception:
+                            pass
                     return 0, None
                 else:
                     await task_logger.log_task_failure(
@@ -161,12 +152,24 @@ async def index_confluence_pages(
                         "API Error",
                         {"error_type": "APIError"},
                     )
+                    # Close client on error
+                    if confluence_client:
+                        try:
+                            await confluence_client.close()
+                        except Exception:
+                            pass
                     return 0, f"Failed to get Confluence pages: {error}"
 
             logger.info(f"Retrieved {len(pages)} pages from Confluence API")
 
         except Exception as e:
             logger.error(f"Error fetching Confluence pages: {e!s}", exc_info=True)
+            # Close client on error
+            if confluence_client:
+                try:
+                    await confluence_client.close()
+                except Exception:
+                    pass
             return 0, f"Error fetching Confluence pages: {e!s}"
 
         # Process and index each page
@@ -418,6 +421,11 @@ async def index_confluence_pages(
         logger.info(
             f"Confluence indexing completed: {documents_indexed} new pages, {documents_skipped} skipped"
         )
+        
+        # Close the client connection
+        if confluence_client:
+            await confluence_client.close()
+        
         return (
             total_processed,
             None,
@@ -425,6 +433,12 @@ async def index_confluence_pages(
 
     except SQLAlchemyError as db_error:
         await session.rollback()
+        # Close client if it exists
+        if confluence_client:
+            try:
+                await confluence_client.close()
+            except Exception:
+                pass
         await task_logger.log_task_failure(
             log_entry,
             f"Database error during Confluence indexing for connector {connector_id}",
@@ -435,6 +449,12 @@ async def index_confluence_pages(
         return 0, f"Database error: {db_error!s}"
     except Exception as e:
         await session.rollback()
+        # Close client if it exists
+        if confluence_client:
+            try:
+                await confluence_client.close()
+            except Exception:
+                pass
         await task_logger.log_task_failure(
             log_entry,
             f"Failed to index Confluence pages for connector {connector_id}",

surfsense_backend/app/utils/validators.py

@@ -545,21 +545,12 @@ def validate_connector_config(
         #         "JIRA_BASE_URL": lambda: validate_url_field("JIRA_BASE_URL", "JIRA"),
         #     },
         # },
-        "CONFLUENCE_CONNECTOR": {
-            "required": [
-                "CONFLUENCE_BASE_URL",
-                "CONFLUENCE_EMAIL",
-                "CONFLUENCE_API_TOKEN",
-            ],
-            "validators": {
-                "CONFLUENCE_EMAIL": lambda: validate_email_field(
-                    "CONFLUENCE_EMAIL", "Confluence"
-                ),
-                "CONFLUENCE_BASE_URL": lambda: validate_url_field(
-                    "CONFLUENCE_BASE_URL", "Confluence"
-                ),
-            },
-        },
+        # "CONFLUENCE_CONNECTOR": {
+        #     "required": [
+        #         "access_token",
+        #     ],
+        #     "validators": {},
+        # },
         "CLICKUP_CONNECTOR": {"required": ["CLICKUP_API_TOKEN"], "validators": {}},
         # "GOOGLE_CALENDAR_CONNECTOR": {
         #     "required": ["token", "refresh_token", "token_uri", "client_id", "expiry", "scopes", "client_secret"],