diff --git a/surfsense_backend/app/celery_app.py b/surfsense_backend/app/celery_app.py index 8858c2619..b77f5698e 100644 --- a/surfsense_backend/app/celery_app.py +++ b/surfsense_backend/app/celery_app.py @@ -79,6 +79,7 @@ celery_app = Celery( "app.tasks.celery_tasks.schedule_checker_task", "app.tasks.celery_tasks.blocknote_migration_tasks", "app.tasks.celery_tasks.document_reindex_tasks", + "app.tasks.celery_tasks.stale_notification_cleanup_task", ], ) @@ -121,4 +122,14 @@ celery_app.conf.beat_schedule = { "expires": 30, # Task expires after 30 seconds if not picked up }, }, + # Cleanup stale connector indexing notifications every 5 minutes + # This detects tasks that crashed or timed out without proper cleanup + # and marks their notifications as failed so users don't see perpetual "syncing" + "cleanup-stale-indexing-notifications": { + "task": "cleanup_stale_indexing_notifications", + "schedule": crontab(minute="*/5"), # Every 5 minutes + "options": { + "expires": 60, # Task expires after 60 seconds if not picked up + }, + }, } diff --git a/surfsense_backend/app/connectors/composio_gmail_connector.py b/surfsense_backend/app/connectors/composio_gmail_connector.py index 953e2e8fc..d3a0d344b 100644 --- a/surfsense_backend/app/connectors/composio_gmail_connector.py +++ b/surfsense_backend/app/connectors/composio_gmail_connector.py @@ -5,6 +5,8 @@ Provides Gmail specific methods for data retrieval and indexing via Composio. """ import logging +import time +from collections.abc import Awaitable, Callable from datetime import UTC, datetime from typing import Any @@ -26,6 +28,10 @@ from app.utils.document_converters import ( generate_unique_identifier_hash, ) +# Heartbeat configuration +HeartbeatCallbackType = Callable[[int], Awaitable[None]] +HEARTBEAT_INTERVAL_SECONDS = 30 + logger = logging.getLogger(__name__) @@ -427,6 +433,7 @@ async def index_composio_gmail( log_entry, update_last_indexed: bool = True, max_items: int = 1000, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str]: """Index Gmail messages via Composio with pagination and incremental processing.""" try: @@ -471,8 +478,16 @@ async def index_composio_gmail( total_documents_skipped = 0 total_messages_fetched = 0 result_size_estimate = None # Will be set from first API response + last_heartbeat_time = time.time() while total_messages_fetched < max_items: + # Send heartbeat periodically to indicate task is still alive + if on_heartbeat_callback: + current_time = time.time() + if current_time - last_heartbeat_time >= HEARTBEAT_INTERVAL_SECONDS: + await on_heartbeat_callback(total_documents_indexed) + last_heartbeat_time = current_time + # Calculate how many messages to fetch in this batch remaining = max_items - total_messages_fetched current_batch_size = min(batch_size, remaining) diff --git a/surfsense_backend/app/connectors/composio_google_calendar_connector.py b/surfsense_backend/app/connectors/composio_google_calendar_connector.py index ec5b22b7f..4302e479b 100644 --- a/surfsense_backend/app/connectors/composio_google_calendar_connector.py +++ b/surfsense_backend/app/connectors/composio_google_calendar_connector.py @@ -5,6 +5,8 @@ Provides Google Calendar specific methods for data retrieval and indexing via Co """ import logging +import time +from collections.abc import Awaitable, Callable from datetime import UTC, datetime from typing import Any @@ -29,6 +31,10 @@ from app.utils.document_converters import ( generate_unique_identifier_hash, ) +# Heartbeat configuration +HeartbeatCallbackType = Callable[[int], Awaitable[None]] +HEARTBEAT_INTERVAL_SECONDS = 30 + logger = logging.getLogger(__name__) @@ -191,6 +197,7 @@ async def index_composio_google_calendar( log_entry, update_last_indexed: bool = True, max_items: int = 2500, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str]: """Index Google Calendar events via Composio.""" try: @@ -262,8 +269,15 @@ async def index_composio_google_calendar( duplicate_content_count = ( 0 # Track events skipped due to duplicate content_hash ) + last_heartbeat_time = time.time() for event in events: + # Send heartbeat periodically to indicate task is still alive + if on_heartbeat_callback: + current_time = time.time() + if current_time - last_heartbeat_time >= HEARTBEAT_INTERVAL_SECONDS: + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = current_time try: # Handle both standard Google API and potential Composio variations event_id = event.get("id", "") or event.get("eventId", "") diff --git a/surfsense_backend/app/connectors/composio_google_drive_connector.py b/surfsense_backend/app/connectors/composio_google_drive_connector.py index 5b8c4b993..364712215 100644 --- a/surfsense_backend/app/connectors/composio_google_drive_connector.py +++ b/surfsense_backend/app/connectors/composio_google_drive_connector.py @@ -9,6 +9,8 @@ import json import logging import os import tempfile +import time +from collections.abc import Awaitable, Callable from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -29,6 +31,10 @@ from app.utils.document_converters import ( generate_unique_identifier_hash, ) +# Heartbeat configuration +HeartbeatCallbackType = Callable[[int], Awaitable[None]] +HEARTBEAT_INTERVAL_SECONDS = 30 + logger = logging.getLogger(__name__) @@ -552,7 +558,9 @@ def generate_indexing_settings_hash( "include_subfolders": indexing_options.get("include_subfolders", True), "max_files_per_folder": indexing_options.get("max_files_per_folder", 100), } - return hashlib.md5(json.dumps(settings, sort_keys=True).encode()).hexdigest() + return hashlib.md5( + json.dumps(settings, sort_keys=True).encode(), usedforsecurity=False + ).hexdigest() async def index_composio_google_drive( @@ -565,6 +573,7 @@ async def index_composio_google_drive( log_entry, update_last_indexed: bool = True, max_items: int = 1000, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, int, str | None]: """Index Google Drive files via Composio with delta sync support. @@ -652,6 +661,7 @@ async def index_composio_google_drive( max_items=max_items, task_logger=task_logger, log_entry=log_entry, + on_heartbeat_callback=on_heartbeat_callback, ) else: logger.info( @@ -684,6 +694,7 @@ async def index_composio_google_drive( max_items=max_items, task_logger=task_logger, log_entry=log_entry, + on_heartbeat_callback=on_heartbeat_callback, ) # Get new page token for next sync (always update after successful sync) @@ -765,6 +776,7 @@ async def _index_composio_drive_delta_sync( max_items: int, task_logger: TaskLoggingService, log_entry, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, int, list[str]]: """Index Google Drive files using delta sync (only changed files). @@ -774,6 +786,7 @@ async def _index_composio_drive_delta_sync( documents_indexed = 0 documents_skipped = 0 processing_errors = [] + last_heartbeat_time = time.time() # Fetch all changes with pagination all_changes = [] @@ -804,6 +817,13 @@ async def _index_composio_drive_delta_sync( logger.info(f"Processing {len(all_changes)} changes from delta sync") for change in all_changes[:max_items]: + # Send heartbeat periodically to indicate task is still alive + if on_heartbeat_callback: + current_time = time.time() + if current_time - last_heartbeat_time >= HEARTBEAT_INTERVAL_SECONDS: + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = current_time + try: # Handle removed files is_removed = change.get("removed", False) @@ -886,11 +906,13 @@ async def _index_composio_drive_full_scan( max_items: int, task_logger: TaskLoggingService, log_entry, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, int, list[str]]: """Index Google Drive files using full scan (first sync or when no delta token).""" documents_indexed = 0 documents_skipped = 0 processing_errors = [] + last_heartbeat_time = time.time() all_files = [] @@ -1001,6 +1023,13 @@ async def _index_composio_drive_full_scan( ) for file_info in all_files: + # Send heartbeat periodically to indicate task is still alive + if on_heartbeat_callback: + current_time = time.time() + if current_time - last_heartbeat_time >= HEARTBEAT_INTERVAL_SECONDS: + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = current_time + try: # Handle both standard Google API and potential Composio variations file_id = file_info.get("id", "") or file_info.get("fileId", "") diff --git a/surfsense_backend/app/connectors/discord_connector.py b/surfsense_backend/app/connectors/discord_connector.py index 1e12cb9a4..7a9e8d5dd 100644 --- a/surfsense_backend/app/connectors/discord_connector.py +++ b/surfsense_backend/app/connectors/discord_connector.py @@ -61,6 +61,9 @@ class DiscordConnector(commands.Bot): self.token = None self._bot_task = None # Holds the async bot task self._is_running = False # Flag to track if the bot is running + self._start_called_event = ( + asyncio.Event() + ) # Event to signal when start() is called # Event to confirm bot is ready @self.event @@ -226,6 +229,9 @@ class DiscordConnector(commands.Bot): ) return + # Signal that we're about to call start() - this allows _wait_until_ready() to proceed + self._start_called_event.set() + await self.start(self.token) logger.info("Discord bot started successfully.") except discord.LoginFailure: @@ -260,6 +266,9 @@ class DiscordConnector(commands.Bot): else: logger.info("Bot is not running or already disconnected.") + # Reset the start event so the connector can be reused + self._start_called_event.clear() + def set_token(self, token: str) -> None: """ Set the discord bot token (for backward compatibility). @@ -277,10 +286,16 @@ class DiscordConnector(commands.Bot): """Helper to wait until the bot is connected and ready.""" logger.info("Waiting for the bot to be ready...") - # Give the event loop a chance to switch to the bot's startup task. - # This allows self.start() to begin initializing the client. - # Terrible solution, but necessary to avoid blocking the event loop. - await asyncio.sleep(1) # Yield control to the event loop + # Wait for start_bot() to actually call self.start() + # This ensures we don't call wait_until_ready() before the client is initialized + try: + await asyncio.wait_for(self._start_called_event.wait(), timeout=30.0) + logger.info("Bot start() has been called, now waiting for ready state...") + except TimeoutError: + logger.error("start_bot() did not call start() within 30 seconds") + raise RuntimeError( + "Discord client failed to initialize - start() was never called" + ) from None try: await asyncio.wait_for(self.wait_until_ready(), timeout=60.0) diff --git a/surfsense_backend/app/connectors/google_calendar_connector.py b/surfsense_backend/app/connectors/google_calendar_connector.py index 7e24f3642..4681251ad 100644 --- a/surfsense_backend/app/connectors/google_calendar_connector.py +++ b/surfsense_backend/app/connectors/google_calendar_connector.py @@ -252,12 +252,16 @@ class GoogleCalendarConnector: if dt_start.tzinfo is None: dt_start = dt_start.replace(hour=0, minute=0, second=0, tzinfo=pytz.UTC) else: - dt_start = dt_start.astimezone(pytz.UTC).replace(hour=0, minute=0, second=0) + dt_start = dt_start.astimezone(pytz.UTC).replace( + hour=0, minute=0, second=0 + ) if dt_end.tzinfo is None: dt_end = dt_end.replace(hour=23, minute=59, second=59, tzinfo=pytz.UTC) else: - dt_end = dt_end.astimezone(pytz.UTC).replace(hour=23, minute=59, second=59) + dt_end = dt_end.astimezone(pytz.UTC).replace( + hour=23, minute=59, second=59 + ) if dt_start >= dt_end: return [], ( diff --git a/surfsense_backend/app/connectors/google_drive/credentials.py b/surfsense_backend/app/connectors/google_drive/credentials.py index 7e6335f6d..5b1900ab2 100644 --- a/surfsense_backend/app/connectors/google_drive/credentials.py +++ b/surfsense_backend/app/connectors/google_drive/credentials.py @@ -132,6 +132,15 @@ async def get_valid_credentials( await session.commit() except Exception as e: + error_str = str(e) + # Check if this is an invalid_grant error (token expired/revoked) + if ( + "invalid_grant" in error_str.lower() + or "token has been expired or revoked" in error_str.lower() + ): + raise Exception( + "Google Drive authentication failed. Please re-authenticate." + ) from e raise Exception(f"Failed to refresh Google OAuth credentials: {e!s}") from e return credentials diff --git a/surfsense_backend/app/routes/airtable_add_connector_route.py b/surfsense_backend/app/routes/airtable_add_connector_route.py index 64fa104d8..423d61fb2 100644 --- a/surfsense_backend/app/routes/airtable_add_connector_route.py +++ b/surfsense_backend/app/routes/airtable_add_connector_route.py @@ -442,11 +442,24 @@ async def refresh_airtable_token( if token_response.status_code != 200: error_detail = token_response.text + error_code = "" try: error_json = token_response.json() error_detail = error_json.get("error_description", error_detail) + error_code = error_json.get("error", "") except Exception: pass + # Check if this is a token expiration/revocation error + error_lower = (error_detail + error_code).lower() + if ( + "invalid_grant" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="Airtable authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Token refresh failed: {error_detail}" ) diff --git a/surfsense_backend/app/routes/clickup_add_connector_route.py b/surfsense_backend/app/routes/clickup_add_connector_route.py index f962f65fb..1b2e6795d 100644 --- a/surfsense_backend/app/routes/clickup_add_connector_route.py +++ b/surfsense_backend/app/routes/clickup_add_connector_route.py @@ -417,6 +417,17 @@ async def refresh_clickup_token( error_detail = error_json.get("error", error_detail) except Exception: pass + # Check if this is a token expiration/revocation error + error_lower = error_detail.lower() + if ( + "invalid_grant" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="ClickUp authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Token refresh failed: {error_detail}" ) diff --git a/surfsense_backend/app/routes/confluence_add_connector_route.py b/surfsense_backend/app/routes/confluence_add_connector_route.py index 6c5830b17..24e0f858a 100644 --- a/surfsense_backend/app/routes/confluence_add_connector_route.py +++ b/surfsense_backend/app/routes/confluence_add_connector_route.py @@ -428,13 +428,26 @@ async def refresh_confluence_token( if token_response.status_code != 200: error_detail = token_response.text + error_code = "" try: error_json = token_response.json() error_detail = error_json.get( "error_description", error_json.get("error", error_detail) ) + error_code = error_json.get("error", "") except Exception: pass + # Check if this is a token expiration/revocation error + error_lower = (error_detail + error_code).lower() + if ( + "invalid_grant" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="Confluence authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Token refresh failed: {error_detail}" ) diff --git a/surfsense_backend/app/routes/discord_add_connector_route.py b/surfsense_backend/app/routes/discord_add_connector_route.py index 1d8b40fcf..e49acf30b 100644 --- a/surfsense_backend/app/routes/discord_add_connector_route.py +++ b/surfsense_backend/app/routes/discord_add_connector_route.py @@ -46,6 +46,11 @@ SCOPES = [ "guilds.members.read", # Read member information ] +# Discord permission bits +VIEW_CHANNEL = 1 << 10 # 1024 +READ_MESSAGE_HISTORY = 1 << 16 # 65536 +ADMINISTRATOR = 1 << 3 # 8 + # Initialize security utilities _state_manager = None _token_encryption = None @@ -531,3 +536,296 @@ async def refresh_discord_token( raise HTTPException( status_code=500, detail=f"Failed to refresh Discord tokens: {e!s}" ) from e + + +def _compute_channel_permissions( + base_permissions: int, + bot_role_ids: set[str], + bot_user_id: str | None, + channel_overwrites: list[dict], + guild_id: str, +) -> int: + """ + Compute effective permissions for a channel based on role permissions and overwrites. + + Discord permission computation follows this order (per official docs): + 1. Start with base permissions from roles + 2. Apply @everyone role overwrites (deny, then allow) + 3. Apply role-specific overwrites (deny, then allow) + 4. Apply member-specific overwrites (deny, then allow) + + Args: + base_permissions: Combined permissions from all bot roles + bot_role_ids: Set of role IDs the bot has + bot_user_id: The bot's user ID for member-specific overwrites + channel_overwrites: List of permission overwrites for the channel + guild_id: Guild ID (same as @everyone role ID) + + Returns: + Computed permission integer + """ + permissions = base_permissions + + # Permission overwrites are applied in order: @everyone, roles, member + everyone_allow = 0 + everyone_deny = 0 + role_allow = 0 + role_deny = 0 + member_allow = 0 + member_deny = 0 + + for overwrite in channel_overwrites: + overwrite_id = overwrite.get("id") + overwrite_type = overwrite.get("type") # 0 = role, 1 = member + allow = int(overwrite.get("allow", 0)) + deny = int(overwrite.get("deny", 0)) + + if overwrite_type == 0: # Role overwrite + if overwrite_id == guild_id: # @everyone role + everyone_allow = allow + everyone_deny = deny + elif overwrite_id in bot_role_ids: + role_allow |= allow + role_deny |= deny + elif overwrite_type == 1 and bot_user_id and overwrite_id == bot_user_id: + # Member-specific overwrite for the bot + member_allow = allow + member_deny = deny + + # Apply in order per Discord docs: + # 1. @everyone deny, then allow + permissions &= ~everyone_deny + permissions |= everyone_allow + # 2. Role deny, then allow + permissions &= ~role_deny + permissions |= role_allow + # 3. Member deny, then allow (applied LAST, highest priority) + permissions &= ~member_deny + permissions |= member_allow + + return permissions + + +@router.get("/discord/connector/{connector_id}/channels", response_model=None) +async def get_discord_channels( + connector_id: int, + session: AsyncSession = Depends(get_async_session), + user: User = Depends(current_active_user), +): + """ + Get list of Discord text channels for a connector with permission info. + + Uses Discord's HTTP REST API directly instead of WebSocket bot connection. + Computes effective permissions to determine if bot can read message history. + + Args: + connector_id: The Discord connector ID + session: Database session + user: Current authenticated user + + Returns: + List of channels with id, name, type, position, category_id, and can_index fields + """ + from sqlalchemy import select + + try: + # Get connector and verify ownership + result = await session.execute( + select(SearchSourceConnector).where( + SearchSourceConnector.id == connector_id, + SearchSourceConnector.user_id == user.id, + SearchSourceConnector.connector_type + == SearchSourceConnectorType.DISCORD_CONNECTOR, + ) + ) + connector = result.scalar_one_or_none() + + if not connector: + raise HTTPException( + status_code=404, + detail="Discord connector not found or access denied", + ) + + # Get credentials and decrypt bot token + credentials = DiscordAuthCredentialsBase.from_dict(connector.config) + token_encryption = get_token_encryption() + is_encrypted = connector.config.get("_token_encrypted", False) + + bot_token = credentials.bot_token + if is_encrypted and bot_token: + try: + bot_token = token_encryption.decrypt_token(bot_token) + except Exception as e: + logger.error(f"Failed to decrypt bot token: {e!s}") + raise HTTPException( + status_code=500, detail="Failed to decrypt stored bot token" + ) from e + + if not bot_token: + raise HTTPException( + status_code=400, + detail="No bot token available. Please re-authenticate.", + ) + + # Get guild_id from connector config + guild_id = connector.config.get("guild_id") + if not guild_id: + raise HTTPException( + status_code=400, + detail="No guild_id associated with this connector. Please reconnect the Discord server.", + ) + + headers = {"Authorization": f"Bot {bot_token}"} + + async with httpx.AsyncClient() as client: + # Fetch bot's user info to get bot user ID + bot_user_response = await client.get( + "https://discord.com/api/v10/users/@me", + headers=headers, + timeout=30.0, + ) + + if bot_user_response.status_code != 200: + logger.warning( + f"Failed to fetch bot user info: {bot_user_response.text}" + ) + bot_user_id = None + else: + bot_user_id = bot_user_response.json().get("id") + + # Fetch guild info to get roles + guild_response = await client.get( + f"https://discord.com/api/v10/guilds/{guild_id}", + headers=headers, + timeout=30.0, + ) + + if guild_response.status_code != 200: + raise HTTPException( + status_code=guild_response.status_code, + detail="Failed to fetch guild information", + ) + + guild_data = guild_response.json() + guild_roles = {role["id"]: role for role in guild_data.get("roles", [])} + + # Fetch bot's member info to get its roles + bot_member_response = await client.get( + f"https://discord.com/api/v10/guilds/{guild_id}/members/{bot_user_id}", + headers=headers, + timeout=30.0, + ) + + if bot_member_response.status_code != 200: + logger.warning( + f"Failed to fetch bot member info: {bot_member_response.text}" + ) + bot_role_ids = {guild_id} # At minimum, bot has @everyone role + base_permissions = int( + guild_roles.get(guild_id, {}).get("permissions", 0) + ) + else: + bot_member_data = bot_member_response.json() + bot_role_ids = set(bot_member_data.get("roles", [])) + bot_role_ids.add(guild_id) # @everyone role is always included + + # Compute base permissions from all bot roles + base_permissions = 0 + for role_id in bot_role_ids: + if role_id in guild_roles: + role_perms = int(guild_roles[role_id].get("permissions", 0)) + base_permissions |= role_perms + + # Check if bot has administrator permission (bypasses all checks) + is_admin = (base_permissions & ADMINISTRATOR) == ADMINISTRATOR + + # Fetch channels + channels_response = await client.get( + f"https://discord.com/api/v10/guilds/{guild_id}/channels", + headers=headers, + timeout=30.0, + ) + + if channels_response.status_code == 403: + raise HTTPException( + status_code=403, + detail="Bot does not have permission to view channels in this server. Please ensure the bot has the 'View Channels' permission.", + ) + elif channels_response.status_code == 404: + raise HTTPException( + status_code=404, + detail="Discord server not found. The bot may have been removed from the server.", + ) + elif channels_response.status_code != 200: + error_detail = channels_response.text + try: + error_json = channels_response.json() + error_detail = error_json.get("message", error_detail) + except Exception: + pass + raise HTTPException( + status_code=channels_response.status_code, + detail=f"Failed to fetch Discord channels: {error_detail}", + ) + + channels_data = channels_response.json() + + # Discord channel types: + # 0 = GUILD_TEXT, 2 = GUILD_VOICE, 4 = GUILD_CATEGORY, 5 = GUILD_ANNOUNCEMENT + # We want text channels (type 0) and announcement channels (type 5) + text_channel_types = {0, 5} + + text_channels = [] + for ch in channels_data: + if ch.get("type") in text_channel_types: + # Compute effective permissions for this channel + if is_admin: + # Administrators bypass all permission checks + can_index = True + else: + channel_overwrites = ch.get("permission_overwrites", []) + effective_perms = _compute_channel_permissions( + base_permissions, + bot_role_ids, + bot_user_id, + channel_overwrites, + guild_id, + ) + + # Bot can index if it has both VIEW_CHANNEL and READ_MESSAGE_HISTORY + has_view = (effective_perms & VIEW_CHANNEL) == VIEW_CHANNEL + has_read_history = ( + effective_perms & READ_MESSAGE_HISTORY + ) == READ_MESSAGE_HISTORY + can_index = has_view and has_read_history + + text_channels.append( + { + "id": ch["id"], + "name": ch["name"], + "type": "text" if ch["type"] == 0 else "announcement", + "position": ch.get("position", 0), + "category_id": ch.get("parent_id"), + "can_index": can_index, + } + ) + + # Sort by position + text_channels.sort(key=lambda x: x["position"]) + + logger.info( + f"Fetched {len(text_channels)} text channels for Discord connector {connector_id}" + ) + + return text_channels + + except HTTPException: + raise + except Exception as e: + logger.error( + f"Failed to get Discord channels for connector {connector_id}: {e!s}", + exc_info=True, + ) + raise HTTPException( + status_code=500, detail=f"Failed to get Discord channels: {e!s}" + ) from e diff --git a/surfsense_backend/app/routes/jira_add_connector_route.py b/surfsense_backend/app/routes/jira_add_connector_route.py index fb66f4da7..58903606a 100644 --- a/surfsense_backend/app/routes/jira_add_connector_route.py +++ b/surfsense_backend/app/routes/jira_add_connector_route.py @@ -446,13 +446,26 @@ async def refresh_jira_token( if token_response.status_code != 200: error_detail = token_response.text + error_code = "" try: error_json = token_response.json() error_detail = error_json.get( "error_description", error_json.get("error", error_detail) ) + error_code = error_json.get("error", "") except Exception: pass + # Check if this is a token expiration/revocation error + error_lower = (error_detail + error_code).lower() + if ( + "invalid_grant" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="Jira authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Token refresh failed: {error_detail}" ) diff --git a/surfsense_backend/app/routes/linear_add_connector_route.py b/surfsense_backend/app/routes/linear_add_connector_route.py index fc9501bfb..dd5f7443c 100644 --- a/surfsense_backend/app/routes/linear_add_connector_route.py +++ b/surfsense_backend/app/routes/linear_add_connector_route.py @@ -403,11 +403,24 @@ async def refresh_linear_token( if token_response.status_code != 200: error_detail = token_response.text + error_code = "" try: error_json = token_response.json() error_detail = error_json.get("error_description", error_detail) + error_code = error_json.get("error", "") except Exception: pass + # Check if this is a token expiration/revocation error + error_lower = (error_detail + error_code).lower() + if ( + "invalid_grant" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="Linear authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Token refresh failed: {error_detail}" ) diff --git a/surfsense_backend/app/routes/notion_add_connector_route.py b/surfsense_backend/app/routes/notion_add_connector_route.py index aac821793..81017af50 100644 --- a/surfsense_backend/app/routes/notion_add_connector_route.py +++ b/surfsense_backend/app/routes/notion_add_connector_route.py @@ -407,11 +407,24 @@ async def refresh_notion_token( if token_response.status_code != 200: error_detail = token_response.text + error_code = "" try: error_json = token_response.json() error_detail = error_json.get("error_description", error_detail) + error_code = error_json.get("error", "") except Exception: pass + # Check if this is a token expiration/revocation error + error_lower = (error_detail + error_code).lower() + if ( + "invalid_grant" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="Notion authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Token refresh failed: {error_detail}" ) diff --git a/surfsense_backend/app/routes/search_source_connectors_routes.py b/surfsense_backend/app/routes/search_source_connectors_routes.py index a27c2125c..3a937653d 100644 --- a/surfsense_backend/app/routes/search_source_connectors_routes.py +++ b/surfsense_backend/app/routes/search_source_connectors_routes.py @@ -19,10 +19,12 @@ Non-OAuth connectors (BookStack, GitHub, etc.) are limited to one per search spa """ import logging +import os from datetime import UTC, datetime, timedelta from typing import Any import pytz +import redis from dateutil.parser import isoparse from fastapi import APIRouter, Body, Depends, HTTPException, Query from pydantic import BaseModel, Field, ValidationError @@ -78,6 +80,27 @@ from app.utils.rbac import check_permission # Set up logging logger = logging.getLogger(__name__) +# Redis client for heartbeat tracking +_heartbeat_redis_client: redis.Redis | None = None + +# Redis key TTL - notification is stale if no heartbeat in this time +HEARTBEAT_TTL_SECONDS = 120 # 2 minutes + + +def get_heartbeat_redis_client() -> redis.Redis: + """Get or create Redis client for heartbeat tracking.""" + global _heartbeat_redis_client + if _heartbeat_redis_client is None: + redis_url = os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0") + _heartbeat_redis_client = redis.from_url(redis_url, decode_responses=True) + return _heartbeat_redis_client + + +def _get_heartbeat_key(notification_id: int) -> str: + """Generate Redis key for notification heartbeat.""" + return f"indexing:heartbeat:{notification_id}" + + router = APIRouter() @@ -1137,6 +1160,7 @@ async def run_slack_indexing( end_date=end_date, indexing_function=index_slack_messages, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1150,6 +1174,7 @@ async def _run_indexing_with_notifications( indexing_function, update_timestamp_func=None, supports_retry_callback: bool = False, + supports_heartbeat_callback: bool = False, ): """ Generic helper to run indexing with real-time notifications. @@ -1164,11 +1189,14 @@ async def _run_indexing_with_notifications( indexing_function: Async function that performs the indexing update_timestamp_func: Optional function to update connector timestamp supports_retry_callback: Whether the indexing function supports on_retry_callback + supports_heartbeat_callback: Whether the indexing function supports on_heartbeat_callback """ from uuid import UUID + from celery.exceptions import SoftTimeLimitExceeded + notification = None - # Track indexed count for retry notifications + # Track indexed count for retry notifications and heartbeat current_indexed_count = 0 try: @@ -1195,6 +1223,16 @@ async def _run_indexing_with_notifications( ) ) + # Set initial Redis heartbeat for stale detection + if notification: + try: + heartbeat_key = _get_heartbeat_key(notification.id) + get_heartbeat_redis_client().setex( + heartbeat_key, HEARTBEAT_TTL_SECONDS, "0" + ) + except Exception as e: + logger.warning(f"Failed to set initial Redis heartbeat: {e}") + # Update notification to fetching stage if notification: await NotificationService.connector_indexing.notify_indexing_progress( @@ -1227,6 +1265,40 @@ async def _run_indexing_with_notifications( # Don't let notification errors break the indexing logger.warning(f"Failed to update retry notification: {e}") + # Create heartbeat callback for connectors that support it + # This updates the notification periodically during long-running indexing loops + # to prevent the task from appearing stuck if the worker crashes + async def on_heartbeat_callback(indexed_count: int) -> None: + """Callback to update notification during indexing (heartbeat).""" + nonlocal notification, current_indexed_count + current_indexed_count = indexed_count + if notification: + try: + # Set Redis heartbeat key with TTL (fast, for stale detection) + heartbeat_key = _get_heartbeat_key(notification.id) + get_heartbeat_redis_client().setex( + heartbeat_key, HEARTBEAT_TTL_SECONDS, str(indexed_count) + ) + except Exception as e: + # Don't let Redis errors break the indexing + logger.warning(f"Failed to set Redis heartbeat: {e}") + + try: + # Still update DB notification for progress display + await session.refresh(notification) + await ( + NotificationService.connector_indexing.notify_indexing_progress( + session=session, + notification=notification, + indexed_count=indexed_count, + stage="processing", + ) + ) + await session.commit() + except Exception as e: + # Don't let notification errors break the indexing + logger.warning(f"Failed to update heartbeat notification: {e}") + # Build kwargs for indexing function indexing_kwargs = { "session": session, @@ -1242,6 +1314,10 @@ async def _run_indexing_with_notifications( if supports_retry_callback: indexing_kwargs["on_retry_callback"] = on_retry_callback + # Add heartbeat callback for connectors that support it + if supports_heartbeat_callback: + indexing_kwargs["on_heartbeat_callback"] = on_heartbeat_callback + # Run the indexing function # Some indexers return (indexed, error), others return (indexed, skipped, error) result = await indexing_function(**indexing_kwargs) @@ -1398,6 +1474,32 @@ async def _run_indexing_with_notifications( await ( session.commit() ) # Commit to ensure Electric SQL syncs the notification update + except SoftTimeLimitExceeded: + # Celery soft time limit was reached - task is about to be killed + # Gracefully save progress and mark as interrupted + logger.warning( + f"Soft time limit reached for connector {connector_id}. " + f"Saving partial progress: {current_indexed_count} items indexed." + ) + + if notification: + try: + await session.refresh(notification) + await NotificationService.connector_indexing.notify_indexing_completed( + session=session, + notification=notification, + indexed_count=current_indexed_count, + error_message="Time limit reached. Partial sync completed. Please run again for remaining items.", + is_warning=True, # Mark as warning since partial data was indexed + ) + await session.commit() + except Exception as notif_error: + logger.error( + f"Failed to update notification on soft timeout: {notif_error!s}" + ) + + # Re-raise so Celery knows the task was terminated + raise except Exception as e: logger.error(f"Error in indexing task: {e!s}", exc_info=True) @@ -1409,12 +1511,20 @@ async def _run_indexing_with_notifications( await NotificationService.connector_indexing.notify_indexing_completed( session=session, notification=notification, - indexed_count=0, + indexed_count=current_indexed_count, # Use tracked count, not 0 error_message=str(e), skipped_count=None, # Unknown on exception ) except Exception as notif_error: logger.error(f"Failed to update notification: {notif_error!s}") + finally: + # Clean up Redis heartbeat key when task completes (success or failure) + if notification: + try: + heartbeat_key = _get_heartbeat_key(notification.id) + get_heartbeat_redis_client().delete(heartbeat_key) + except Exception: + pass # Ignore cleanup errors - key will expire anyway async def run_notion_indexing_with_new_session( @@ -1439,6 +1549,7 @@ async def run_notion_indexing_with_new_session( indexing_function=index_notion_pages, update_timestamp_func=_update_connector_timestamp_by_id, supports_retry_callback=True, # Notion connector supports retry notifications + supports_heartbeat_callback=True, # Notion connector supports heartbeat notifications ) @@ -1471,6 +1582,7 @@ async def run_notion_indexing( indexing_function=index_notion_pages, update_timestamp_func=_update_connector_timestamp_by_id, supports_retry_callback=True, # Notion connector supports retry notifications + supports_heartbeat_callback=True, # Notion connector supports heartbeat notifications ) @@ -1521,6 +1633,7 @@ async def run_github_indexing( end_date=end_date, indexing_function=index_github_repos, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1571,6 +1684,7 @@ async def run_linear_indexing( end_date=end_date, indexing_function=index_linear_issues, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1620,6 +1734,7 @@ async def run_discord_indexing( end_date=end_date, indexing_function=index_discord_messages, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1670,6 +1785,7 @@ async def run_teams_indexing( end_date=end_date, indexing_function=index_teams_messages, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1720,6 +1836,7 @@ async def run_jira_indexing( end_date=end_date, indexing_function=index_jira_issues, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1772,6 +1889,7 @@ async def run_confluence_indexing( end_date=end_date, indexing_function=index_confluence_pages, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1822,6 +1940,7 @@ async def run_clickup_indexing( end_date=end_date, indexing_function=index_clickup_tasks, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1872,6 +1991,7 @@ async def run_airtable_indexing( end_date=end_date, indexing_function=index_airtable_records, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1924,6 +2044,7 @@ async def run_google_calendar_indexing( end_date=end_date, indexing_function=index_google_calendar_events, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -1998,6 +2119,7 @@ async def run_google_gmail_indexing( end_date=end_date, indexing_function=gmail_indexing_wrapper, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -2206,6 +2328,7 @@ async def run_luma_indexing( end_date=end_date, indexing_function=index_luma_events, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -2257,6 +2380,7 @@ async def run_elasticsearch_indexing( end_date=end_date, indexing_function=index_elasticsearch_documents, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -2306,6 +2430,7 @@ async def run_web_page_indexing( end_date=end_date, indexing_function=index_crawled_urls, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -2360,6 +2485,7 @@ async def run_bookstack_indexing( end_date=end_date, indexing_function=index_bookstack_pages, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -2412,6 +2538,7 @@ async def run_obsidian_indexing( end_date=end_date, indexing_function=index_obsidian_vault, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) @@ -2465,6 +2592,7 @@ async def run_composio_indexing( end_date=end_date, indexing_function=index_composio_connector, update_timestamp_func=_update_connector_timestamp_by_id, + supports_heartbeat_callback=True, ) diff --git a/surfsense_backend/app/routes/slack_add_connector_route.py b/surfsense_backend/app/routes/slack_add_connector_route.py index 62d2ccaaa..0cbfdef44 100644 --- a/surfsense_backend/app/routes/slack_add_connector_route.py +++ b/surfsense_backend/app/routes/slack_add_connector_route.py @@ -6,6 +6,7 @@ Handles OAuth 2.0 authentication flow for Slack connector. import logging from datetime import UTC, datetime, timedelta +from typing import Any from uuid import UUID import httpx @@ -14,6 +15,7 @@ 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 ( @@ -418,6 +420,19 @@ async def refresh_slack_token( error_detail = error_json.get("error", error_detail) except Exception: pass + # Check if this is a token expiration/revocation error + error_lower = error_detail.lower() + if ( + "invalid_grant" in error_lower + or "invalid_auth" in error_lower + or "token_revoked" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="Slack authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Token refresh failed: {error_detail}" ) @@ -427,6 +442,20 @@ async def refresh_slack_token( # Slack OAuth v2 returns success status in the JSON if not token_json.get("ok", False): error_msg = token_json.get("error", "Unknown error") + # Check if this is a token expiration/revocation error + error_lower = error_msg.lower() + if ( + "invalid_grant" in error_lower + or "invalid_auth" in error_lower + or "invalid_refresh_token" in error_lower + or "token_revoked" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="Slack authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Slack OAuth refresh error: {error_msg}" ) @@ -490,3 +519,88 @@ async def refresh_slack_token( raise HTTPException( status_code=500, detail=f"Failed to refresh Slack token: {e!s}" ) from e + + +@router.get("/slack/connector/{connector_id}/channels") +async def get_slack_channels( + connector_id: int, + session: AsyncSession = Depends(get_async_session), + user: User = Depends(current_active_user), +) -> list[dict[str, Any]]: + """ + Get list of Slack channels with bot membership status. + + This endpoint fetches all channels the bot can see and indicates + whether the bot is a member of each channel (required for accessing messages). + + Args: + connector_id: The Slack connector ID + session: Database session + user: Current authenticated user + + Returns: + List of channels with id, name, is_private, and is_member fields + """ + try: + # Get the connector and verify ownership + result = await session.execute( + select(SearchSourceConnector).where( + SearchSourceConnector.id == connector_id, + SearchSourceConnector.user_id == user.id, + SearchSourceConnector.connector_type + == SearchSourceConnectorType.SLACK_CONNECTOR, + ) + ) + connector = result.scalar_one_or_none() + + if not connector: + raise HTTPException( + status_code=404, + detail="Slack connector not found or access denied", + ) + + # Get credentials and decrypt bot token + credentials = SlackAuthCredentialsBase.from_dict(connector.config) + token_encryption = get_token_encryption() + is_encrypted = connector.config.get("_token_encrypted", False) + + bot_token = credentials.bot_token + if is_encrypted and bot_token: + try: + bot_token = token_encryption.decrypt_token(bot_token) + except Exception as e: + logger.error(f"Failed to decrypt bot token: {e!s}") + raise HTTPException( + status_code=500, detail="Failed to decrypt stored bot token" + ) from e + + if not bot_token: + raise HTTPException( + status_code=400, + detail="No bot token available. Please re-authenticate.", + ) + + # Import SlackHistory here to avoid circular imports + from app.connectors.slack_history import SlackHistory + + # Create Slack client with direct token (simple pattern for quick operations) + slack_client = SlackHistory(token=bot_token) + + channels = await slack_client.get_all_channels(include_private=True) + + logger.info( + f"Fetched {len(channels)} channels for Slack connector {connector_id}" + ) + + return channels + + except HTTPException: + raise + except Exception as e: + logger.error( + f"Failed to get Slack channels for connector {connector_id}: {e!s}", + exc_info=True, + ) + raise HTTPException( + status_code=500, detail=f"Failed to get Slack channels: {e!s}" + ) from e diff --git a/surfsense_backend/app/routes/teams_add_connector_route.py b/surfsense_backend/app/routes/teams_add_connector_route.py index 9ce84e171..77ce4965e 100644 --- a/surfsense_backend/app/routes/teams_add_connector_route.py +++ b/surfsense_backend/app/routes/teams_add_connector_route.py @@ -420,11 +420,24 @@ async def refresh_teams_token( if token_response.status_code != 200: error_detail = token_response.text + error_code = "" try: error_json = token_response.json() error_detail = error_json.get("error_description", error_detail) + error_code = error_json.get("error", "") except Exception: pass + # Check if this is a token expiration/revocation error + error_lower = (error_detail + error_code).lower() + if ( + "invalid_grant" in error_lower + or "expired" in error_lower + or "revoked" in error_lower + ): + raise HTTPException( + status_code=401, + detail="Microsoft Teams authentication failed. Please re-authenticate.", + ) raise HTTPException( status_code=400, detail=f"Token refresh failed: {error_detail}" ) diff --git a/surfsense_backend/app/tasks/celery_tasks/stale_notification_cleanup_task.py b/surfsense_backend/app/tasks/celery_tasks/stale_notification_cleanup_task.py new file mode 100644 index 000000000..e77b3225e --- /dev/null +++ b/surfsense_backend/app/tasks/celery_tasks/stale_notification_cleanup_task.py @@ -0,0 +1,164 @@ +"""Celery task to detect and mark stale connector indexing notifications as failed. + +This task runs periodically (every 5 minutes by default) to find notifications +that are stuck in "in_progress" status but don't have an active Redis heartbeat key. +These are marked as "failed" to prevent the frontend from showing a perpetual "syncing" state. + +Detection mechanism: +- Active indexing tasks set a Redis key with TTL (2 minutes) as a heartbeat +- If the task crashes, the Redis key expires automatically +- This cleanup task checks for in-progress notifications without a Redis heartbeat key +- Such notifications are marked as failed with O(1) batch UPDATE +""" + +import json +import logging +import os +from datetime import UTC, datetime + +import redis +from sqlalchemy import and_, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.future import select +from sqlalchemy.pool import NullPool + +from app.celery_app import celery_app +from app.config import config +from app.db import Notification + +logger = logging.getLogger(__name__) + +# Redis client for checking heartbeats +_redis_client: redis.Redis | None = None + + +def get_redis_client() -> redis.Redis: + """Get or create Redis client for heartbeat checking.""" + global _redis_client + if _redis_client is None: + redis_url = os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0") + _redis_client = redis.from_url(redis_url, decode_responses=True) + return _redis_client + + +def _get_heartbeat_key(notification_id: int) -> str: + """Generate Redis key for notification heartbeat.""" + return f"indexing:heartbeat:{notification_id}" + + +def get_celery_session_maker(): + """Create async session maker for Celery tasks.""" + engine = create_async_engine( + config.DATABASE_URL, + poolclass=NullPool, + echo=False, + ) + return async_sessionmaker(engine, expire_on_commit=False) + + +@celery_app.task(name="cleanup_stale_indexing_notifications") +def cleanup_stale_indexing_notifications_task(): + """ + Check for stale connector indexing notifications and mark them as failed. + + This task finds notifications that: + - Have type = 'connector_indexing' + - Have metadata.status = 'in_progress' + - Do NOT have a corresponding Redis heartbeat key (meaning task crashed) + + And marks them as failed with O(1) batch UPDATE. + """ + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + loop.run_until_complete(_cleanup_stale_notifications()) + finally: + loop.close() + + +async def _cleanup_stale_notifications(): + """Find and mark stale connector indexing notifications as failed. + + Uses Redis TTL-based detection: + 1. Find all in-progress notifications + 2. Check which ones are missing their Redis heartbeat key + 3. Mark those as failed with O(1) batch UPDATE using JSONB || operator + """ + async with get_celery_session_maker()() as session: + try: + # Find all in-progress connector indexing notifications + result = await session.execute( + select(Notification.id).where( + and_( + Notification.type == "connector_indexing", + Notification.notification_metadata["status"].astext + == "in_progress", + ) + ) + ) + in_progress_ids = [row[0] for row in result.fetchall()] + + if not in_progress_ids: + logger.debug("No in-progress connector indexing notifications found") + return + + # Check which ones are missing heartbeat keys in Redis + redis_client = get_redis_client() + stale_notification_ids = [] + + for notification_id in in_progress_ids: + heartbeat_key = _get_heartbeat_key(notification_id) + if not redis_client.exists(heartbeat_key): + stale_notification_ids.append(notification_id) + + if not stale_notification_ids: + logger.debug( + f"All {len(in_progress_ids)} in-progress notifications have active Redis heartbeats" + ) + return + + logger.warning( + f"Found {len(stale_notification_ids)} stale connector indexing notifications " + f"(no Redis heartbeat key): {stale_notification_ids}" + ) + + # O(1) Batch UPDATE using JSONB || operator + # This merges the update data into existing notification_metadata + # Also updates title and message for proper UI display + error_message = ( + "Something went wrong while syncing your content. Please retry." + ) + + update_data = { + "status": "failed", + "completed_at": datetime.now(UTC).isoformat(), + "error_message": error_message, + "sync_stage": "failed", + } + + await session.execute( + text(""" + UPDATE notifications + SET metadata = metadata || CAST(:update_json AS jsonb), + title = 'Failed: ' || COALESCE(metadata->>'connector_name', 'Connector'), + message = :display_message + WHERE id = ANY(:ids) + """), + { + "update_json": json.dumps(update_data), + "display_message": f"{error_message}", + "ids": stale_notification_ids, + }, + ) + + await session.commit() + logger.info( + f"Successfully marked {len(stale_notification_ids)} stale notifications as failed (batch UPDATE)" + ) + + except Exception as e: + logger.error(f"Error cleaning up stale notifications: {e!s}", exc_info=True) + await session.rollback() diff --git a/surfsense_backend/app/tasks/composio_indexer.py b/surfsense_backend/app/tasks/composio_indexer.py index ffc4a1f27..0518ad2a6 100644 --- a/surfsense_backend/app/tasks/composio_indexer.py +++ b/surfsense_backend/app/tasks/composio_indexer.py @@ -9,6 +9,7 @@ to avoid circular import issues with the connector_indexers package. """ import logging +from collections.abc import Awaitable, Callable from importlib import import_module from sqlalchemy.exc import SQLAlchemyError @@ -22,6 +23,9 @@ from app.db import ( from app.services.composio_service import INDEXABLE_TOOLKITS, TOOLKIT_TO_INDEXER from app.services.task_logging_service import TaskLoggingService +# Type alias for heartbeat callback function +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + # Set up logging logger = logging.getLogger(__name__) @@ -86,6 +90,7 @@ async def index_composio_connector( end_date: str | None = None, update_last_indexed: bool = True, max_items: int = 1000, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, int, str | None]: """ Index content from a Composio connector. @@ -102,6 +107,7 @@ async def index_composio_connector( end_date: End date for filtering (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp max_items: Maximum number of items to fetch + on_heartbeat_callback: Optional callback to report progress for heartbeat updates Returns: Tuple of (number_of_indexed_items, number_of_skipped_items, error_message or None) @@ -180,6 +186,7 @@ async def index_composio_connector( "log_entry": log_entry, "update_last_indexed": update_last_indexed, "max_items": max_items, + "on_heartbeat_callback": on_heartbeat_callback, } # Add date params for toolkits that support them diff --git a/surfsense_backend/app/tasks/connector_indexers/airtable_indexer.py b/surfsense_backend/app/tasks/connector_indexers/airtable_indexer.py index 6bb62d716..3bcf95d6a 100644 --- a/surfsense_backend/app/tasks/connector_indexers/airtable_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/airtable_indexer.py @@ -2,6 +2,9 @@ Airtable connector indexer. """ +import time +from collections.abc import Awaitable, Callable + from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession @@ -27,6 +30,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_airtable_records( session: AsyncSession, @@ -37,6 +46,7 @@ async def index_airtable_records( end_date: str | None = None, max_records: int = 2500, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Airtable records for a given connector. @@ -50,6 +60,7 @@ async def index_airtable_records( end_date: End date for filtering records (YYYY-MM-DD) max_records: Maximum number of records to fetch per table update_last_indexed: Whether to update the last_indexed_at timestamp + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple of (number_of_documents_processed, error_message) @@ -127,8 +138,20 @@ async def index_airtable_records( logger.info(f"Found {len(bases)} Airtable bases to process") + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + total_documents_indexed = 0 + # Process each base for base in bases: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) + >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(total_documents_indexed) + last_heartbeat_time = time.time() base_id = base.get("id") base_name = base.get("name", "Unknown Base") @@ -204,6 +227,15 @@ async def index_airtable_records( documents_skipped = 0 # Process each record for record in records: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) + >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(total_documents_indexed) + last_heartbeat_time = time.time() + try: # Generate markdown content markdown_content = ( diff --git a/surfsense_backend/app/tasks/connector_indexers/bookstack_indexer.py b/surfsense_backend/app/tasks/connector_indexers/bookstack_indexer.py index e183ab333..d726e5d95 100644 --- a/surfsense_backend/app/tasks/connector_indexers/bookstack_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/bookstack_indexer.py @@ -2,6 +2,8 @@ BookStack connector indexer. """ +import time +from collections.abc import Awaitable, Callable from datetime import datetime from sqlalchemy.exc import SQLAlchemyError @@ -29,6 +31,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_bookstack_pages( session: AsyncSession, @@ -38,6 +46,7 @@ async def index_bookstack_pages( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index BookStack pages. @@ -50,6 +59,7 @@ async def index_bookstack_pages( start_date: Start date for indexing (YYYY-MM-DD format) end_date: End date for indexing (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -179,7 +189,17 @@ async def index_bookstack_pages( skipped_pages = [] documents_skipped = 0 + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for page in pages: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() try: page_id = page.get("id") page_name = page.get("name", "") diff --git a/surfsense_backend/app/tasks/connector_indexers/clickup_indexer.py b/surfsense_backend/app/tasks/connector_indexers/clickup_indexer.py index 887c3e2e5..e7e8b23e5 100644 --- a/surfsense_backend/app/tasks/connector_indexers/clickup_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/clickup_indexer.py @@ -3,6 +3,8 @@ ClickUp connector indexer. """ import contextlib +import time +from collections.abc import Awaitable, Callable from datetime import datetime from sqlalchemy.exc import SQLAlchemyError @@ -29,6 +31,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_clickup_tasks( session: AsyncSession, @@ -38,6 +46,7 @@ async def index_clickup_tasks( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index tasks from ClickUp workspace. @@ -50,6 +59,7 @@ async def index_clickup_tasks( start_date: Start date for filtering tasks (YYYY-MM-DD format) end_date: End date for filtering tasks (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple of (number of indexed tasks, error message if any) @@ -132,6 +142,9 @@ async def index_clickup_tasks( documents_indexed = 0 documents_skipped = 0 + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + # Iterate workspaces and fetch tasks for workspace in workspaces: workspace_id = workspace.get("id") @@ -170,6 +183,15 @@ async def index_clickup_tasks( ) for task in tasks: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) + >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() + try: task_id = task.get("id") task_name = task.get("name", "Untitled Task") diff --git a/surfsense_backend/app/tasks/connector_indexers/confluence_indexer.py b/surfsense_backend/app/tasks/connector_indexers/confluence_indexer.py index 5673839bb..2f20472d2 100644 --- a/surfsense_backend/app/tasks/connector_indexers/confluence_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/confluence_indexer.py @@ -3,6 +3,8 @@ Confluence connector indexer. """ import contextlib +import time +from collections.abc import Awaitable, Callable from datetime import datetime from sqlalchemy.exc import SQLAlchemyError @@ -30,6 +32,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_confluence_pages( session: AsyncSession, @@ -39,6 +47,7 @@ async def index_confluence_pages( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Confluence pages and comments. @@ -51,6 +60,7 @@ async def index_confluence_pages( start_date: Start date for indexing (YYYY-MM-DD format) end_date: End date for indexing (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -175,7 +185,17 @@ async def index_confluence_pages( skipped_pages = [] documents_skipped = 0 + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for page in pages: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() try: page_id = page.get("id") page_title = page.get("title", "") diff --git a/surfsense_backend/app/tasks/connector_indexers/discord_indexer.py b/surfsense_backend/app/tasks/connector_indexers/discord_indexer.py index 9e401b335..a70bc42d4 100644 --- a/surfsense_backend/app/tasks/connector_indexers/discord_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/discord_indexer.py @@ -3,6 +3,8 @@ Discord connector indexer. """ import asyncio +import time +from collections.abc import Awaitable, Callable from datetime import UTC, datetime, timedelta from sqlalchemy.exc import SQLAlchemyError @@ -28,6 +30,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds - update notification every 30 seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_discord_messages( session: AsyncSession, @@ -37,6 +45,7 @@ async def index_discord_messages( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Discord messages from all accessible channels. @@ -49,6 +58,8 @@ async def index_discord_messages( start_date: Start date for indexing (YYYY-MM-DD format) end_date: End date for indexing (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. + Called periodically with (indexed_count) to prevent task appearing stuck. Returns: Tuple containing (number of documents indexed, error message or None) @@ -281,6 +292,9 @@ async def index_discord_messages( documents_skipped = 0 skipped_channels: list[str] = [] + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + # Process each guild and channel await task_logger.log_task_progress( log_entry, @@ -290,6 +304,14 @@ async def index_discord_messages( try: for guild in guilds: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) + >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() guild_id = guild["id"] guild_name = guild["name"] logger.info(f"Processing guild: {guild_name} ({guild_id})") diff --git a/surfsense_backend/app/tasks/connector_indexers/elasticsearch_indexer.py b/surfsense_backend/app/tasks/connector_indexers/elasticsearch_indexer.py index 6a18af83b..8fbba6463 100644 --- a/surfsense_backend/app/tasks/connector_indexers/elasticsearch_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/elasticsearch_indexer.py @@ -4,6 +4,8 @@ Elasticsearch indexer for SurfSense import json import logging +import time +from collections.abc import Awaitable, Callable from datetime import UTC, datetime from typing import Any @@ -25,6 +27,12 @@ from .base import ( get_current_timestamp, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + logger = logging.getLogger(__name__) @@ -36,6 +44,7 @@ async def index_elasticsearch_documents( start_date: str, end_date: str, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index documents from Elasticsearch into SurfSense @@ -48,6 +57,7 @@ async def index_elasticsearch_documents( start_date: Start date for indexing (not used for Elasticsearch, kept for compatibility) end_date: End date for indexing (not used for Elasticsearch, kept for compatibility) update_last_indexed: Whether to update the last indexed timestamp + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple of (number of documents processed, error message if any) @@ -155,6 +165,9 @@ async def index_elasticsearch_documents( documents_processed = 0 + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + try: await task_logger.log_task_progress( log_entry, @@ -172,6 +185,15 @@ async def index_elasticsearch_documents( size=min(max_documents, 100), # Scroll in batches fields=config.get("ELASTICSEARCH_FIELDS"), ): + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) + >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_processed) + last_heartbeat_time = time.time() + if documents_processed >= max_documents: break diff --git a/surfsense_backend/app/tasks/connector_indexers/github_indexer.py b/surfsense_backend/app/tasks/connector_indexers/github_indexer.py index fb6989bb9..b01d235cf 100644 --- a/surfsense_backend/app/tasks/connector_indexers/github_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/github_indexer.py @@ -5,6 +5,8 @@ This indexer processes entire repository digests in one pass, dramatically reducing LLM API calls compared to the previous file-by-file approach. """ +import time +from collections.abc import Awaitable, Callable from datetime import UTC, datetime from sqlalchemy.exc import SQLAlchemyError @@ -30,6 +32,12 @@ from .base import ( logger, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds - update notification every 30 seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + # Maximum tokens for a single digest before splitting # Most LLMs can handle 128k+ tokens now, but we'll be conservative MAX_DIGEST_CHARS = 500_000 # ~125k tokens @@ -43,6 +51,7 @@ async def index_github_repos( start_date: str | None = None, # Ignored - GitHub indexes full repo snapshots end_date: str | None = None, # Ignored - GitHub indexes full repo snapshots update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index GitHub repositories using gitingest for efficient processing. @@ -62,6 +71,7 @@ async def index_github_repos( start_date: Ignored - kept for API compatibility end_date: Ignored - kept for API compatibility update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -168,7 +178,18 @@ async def index_github_repos( f"Starting gitingest indexing for {len(repo_full_names_to_index)} repositories." ) + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + documents_indexed = 0 + for repo_full_name in repo_full_names_to_index: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() if not repo_full_name or not isinstance(repo_full_name, str): logger.warning(f"Skipping invalid repository entry: {repo_full_name}") continue diff --git a/surfsense_backend/app/tasks/connector_indexers/google_calendar_indexer.py b/surfsense_backend/app/tasks/connector_indexers/google_calendar_indexer.py index 81d33b5e2..f64a7a5c3 100644 --- a/surfsense_backend/app/tasks/connector_indexers/google_calendar_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/google_calendar_indexer.py @@ -2,10 +2,10 @@ Google Calendar connector indexer. """ +import time +from collections.abc import Awaitable, Callable from datetime import datetime, timedelta -import pytz -from dateutil.parser import isoparse from google.oauth2.credentials import Credentials from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession @@ -30,6 +30,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_google_calendar_events( session: AsyncSession, @@ -39,6 +45,7 @@ async def index_google_calendar_events( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Google Calendar events. @@ -52,6 +59,7 @@ async def index_google_calendar_events( end_date: End date for indexing (YYYY-MM-DD format). Can be in the future to index upcoming events. Defaults to today if not provided. update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -281,7 +289,17 @@ async def index_google_calendar_events( 0 # Track events skipped due to duplicate content_hash ) + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for event in events: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() try: event_id = event.get("id") event_summary = event.get("summary", "No Title") diff --git a/surfsense_backend/app/tasks/connector_indexers/google_drive_indexer.py b/surfsense_backend/app/tasks/connector_indexers/google_drive_indexer.py index f50e149d3..3cd59674e 100644 --- a/surfsense_backend/app/tasks/connector_indexers/google_drive_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/google_drive_indexer.py @@ -1,6 +1,8 @@ """Google Drive indexer using Surfsense file processors.""" import logging +import time +from collections.abc import Awaitable, Callable from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession @@ -24,6 +26,12 @@ from app.tasks.connector_indexers.base import ( ) from app.utils.document_converters import generate_unique_identifier_hash +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + logger = logging.getLogger(__name__) @@ -38,6 +46,7 @@ async def index_google_drive_files( update_last_indexed: bool = True, max_files: int = 500, include_subfolders: bool = False, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Google Drive files for a specific connector. @@ -53,6 +62,7 @@ async def index_google_drive_files( 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 + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple of (number_of_indexed_files, error_message) @@ -147,6 +157,7 @@ async def index_google_drive_files( log_entry=log_entry, max_files=max_files, include_subfolders=include_subfolders, + on_heartbeat_callback=on_heartbeat_callback, ) else: logger.info(f"Using full scan for connector {connector_id}") @@ -163,6 +174,7 @@ async def index_google_drive_files( log_entry=log_entry, max_files=max_files, include_subfolders=include_subfolders, + on_heartbeat_callback=on_heartbeat_callback, ) documents_indexed, documents_skipped = result @@ -383,6 +395,7 @@ async def _index_full_scan( log_entry: any, max_files: int, include_subfolders: bool = False, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, int]: """Perform full scan indexing of a folder.""" await task_logger.log_task_progress( @@ -399,10 +412,20 @@ async def _index_full_scan( documents_skipped = 0 files_processed = 0 + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + # Queue of folders to process: (folder_id, folder_name) folders_to_process = [(folder_id, folder_name)] while folders_to_process and files_processed < max_files: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() 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 @@ -485,6 +508,7 @@ async def _index_with_delta_sync( log_entry: any, max_files: int, include_subfolders: bool = False, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, int]: """Perform delta sync indexing using change tracking. @@ -515,7 +539,17 @@ async def _index_with_delta_sync( documents_skipped = 0 files_processed = 0 + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for change in changes: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() if files_processed >= max_files: break diff --git a/surfsense_backend/app/tasks/connector_indexers/google_gmail_indexer.py b/surfsense_backend/app/tasks/connector_indexers/google_gmail_indexer.py index e599abd22..45ce91c6f 100644 --- a/surfsense_backend/app/tasks/connector_indexers/google_gmail_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/google_gmail_indexer.py @@ -2,6 +2,8 @@ Google Gmail connector indexer. """ +import time +from collections.abc import Awaitable, Callable from datetime import datetime from google.oauth2.credentials import Credentials @@ -33,6 +35,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_google_gmail_messages( session: AsyncSession, @@ -43,6 +51,7 @@ async def index_google_gmail_messages( end_date: str | None = None, update_last_indexed: bool = True, max_messages: int = 1000, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str]: """ Index Gmail messages for a specific connector. @@ -56,6 +65,7 @@ async def index_google_gmail_messages( end_date: End date for filtering messages (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) max_messages: Maximum number of messages to fetch (default: 100) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple of (number_of_indexed_messages, status_message) @@ -212,7 +222,18 @@ async def index_google_gmail_messages( documents_indexed = 0 skipped_messages = [] documents_skipped = 0 + + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for message in messages: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() try: # Extract message information message_id = message.get("id", "") diff --git a/surfsense_backend/app/tasks/connector_indexers/jira_indexer.py b/surfsense_backend/app/tasks/connector_indexers/jira_indexer.py index d6095d20e..acee74192 100644 --- a/surfsense_backend/app/tasks/connector_indexers/jira_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/jira_indexer.py @@ -3,6 +3,8 @@ Jira connector indexer. """ import contextlib +import time +from collections.abc import Awaitable, Callable from datetime import datetime from sqlalchemy.exc import SQLAlchemyError @@ -30,6 +32,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds - update notification every 30 seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_jira_issues( session: AsyncSession, @@ -39,6 +47,7 @@ async def index_jira_issues( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Jira issues and comments. @@ -51,6 +60,7 @@ async def index_jira_issues( start_date: Start date for indexing (YYYY-MM-DD format) end_date: End date for indexing (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -169,7 +179,17 @@ async def index_jira_issues( skipped_issues = [] documents_skipped = 0 + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for issue in issues: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() try: issue_id = issue.get("key") issue_identifier = issue.get("key", "") diff --git a/surfsense_backend/app/tasks/connector_indexers/linear_indexer.py b/surfsense_backend/app/tasks/connector_indexers/linear_indexer.py index d00a39160..fc4ae5f18 100644 --- a/surfsense_backend/app/tasks/connector_indexers/linear_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/linear_indexer.py @@ -2,6 +2,8 @@ Linear connector indexer. """ +import time +from collections.abc import Awaitable, Callable from datetime import datetime from sqlalchemy.exc import SQLAlchemyError @@ -29,6 +31,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds - update notification every 30 seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_linear_issues( session: AsyncSession, @@ -38,6 +46,7 @@ async def index_linear_issues( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Linear issues and comments. @@ -50,6 +59,7 @@ async def index_linear_issues( start_date: Start date for indexing (YYYY-MM-DD format) end_date: End date for indexing (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -188,6 +198,9 @@ async def index_linear_issues( documents_skipped = 0 skipped_issues = [] + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + await task_logger.log_task_progress( log_entry, f"Starting to process {len(issues)} Linear issues", @@ -196,6 +209,14 @@ async def index_linear_issues( # Process each issue for issue in issues: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() + try: issue_id = issue.get("id", "") issue_identifier = issue.get("identifier", "") diff --git a/surfsense_backend/app/tasks/connector_indexers/luma_indexer.py b/surfsense_backend/app/tasks/connector_indexers/luma_indexer.py index 59890dbe4..a18abf8ae 100644 --- a/surfsense_backend/app/tasks/connector_indexers/luma_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/luma_indexer.py @@ -2,6 +2,8 @@ Luma connector indexer. """ +import time +from collections.abc import Awaitable, Callable from datetime import datetime, timedelta from sqlalchemy.exc import SQLAlchemyError @@ -28,6 +30,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_luma_events( session: AsyncSession, @@ -37,6 +45,7 @@ async def index_luma_events( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Luma events. @@ -50,6 +59,7 @@ async def index_luma_events( end_date: End date for indexing (YYYY-MM-DD format). Can be in the future to index upcoming events. Defaults to today if not provided. update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -221,7 +231,17 @@ async def index_luma_events( documents_skipped = 0 skipped_events = [] + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for event in events: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() try: # Luma event structure fields - events have nested 'event' field event_data = event.get("event", {}) diff --git a/surfsense_backend/app/tasks/connector_indexers/notion_indexer.py b/surfsense_backend/app/tasks/connector_indexers/notion_indexer.py index a65bf84a7..52622471a 100644 --- a/surfsense_backend/app/tasks/connector_indexers/notion_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/notion_indexer.py @@ -2,6 +2,7 @@ Notion connector indexer. """ +import time from collections.abc import Awaitable, Callable from datetime import datetime @@ -34,6 +35,13 @@ from .base import ( # Signature: async callback(retry_reason, attempt, max_attempts, wait_seconds) -> None RetryCallbackType = Callable[[str, int, int, float], Awaitable[None]] +# Type alias for heartbeat callback +# Signature: async callback(indexed_count) -> None +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds - update notification every 30 seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_notion_pages( session: AsyncSession, @@ -44,6 +52,7 @@ async def index_notion_pages( end_date: str | None = None, update_last_indexed: bool = True, on_retry_callback: RetryCallbackType | None = None, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Notion pages from all accessible pages. @@ -59,6 +68,8 @@ async def index_notion_pages( on_retry_callback: Optional callback for retry progress notifications. Signature: async callback(retry_reason, attempt, max_attempts, wait_seconds) retry_reason is one of: 'rate_limit', 'server_error', 'timeout' + on_heartbeat_callback: Optional callback to update notification during long-running indexing. + Called periodically with (indexed_count) to prevent task appearing stuck. Returns: Tuple containing (number of documents indexed, error message or None) @@ -211,6 +222,9 @@ async def index_notion_pages( documents_skipped = 0 skipped_pages = [] + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + await task_logger.log_task_progress( log_entry, f"Starting to process {len(pages)} Notion pages", @@ -219,6 +233,14 @@ async def index_notion_pages( # Process each page for page in pages: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() + try: page_id = page.get("page_id") page_title = page.get("title", f"Untitled page ({page_id})") diff --git a/surfsense_backend/app/tasks/connector_indexers/obsidian_indexer.py b/surfsense_backend/app/tasks/connector_indexers/obsidian_indexer.py index a603d3fba..a8cd78cc9 100644 --- a/surfsense_backend/app/tasks/connector_indexers/obsidian_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/obsidian_indexer.py @@ -7,6 +7,8 @@ This connector is only available in self-hosted mode. import os import re +import time +from collections.abc import Awaitable, Callable from datetime import UTC, datetime from pathlib import Path @@ -35,6 +37,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + def parse_frontmatter(content: str) -> tuple[dict | None, str]: """ @@ -152,6 +160,7 @@ async def index_obsidian_vault( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index notes from a local Obsidian vault. @@ -167,6 +176,7 @@ async def index_obsidian_vault( start_date: Start date for filtering (YYYY-MM-DD format) - optional end_date: End date for filtering (YYYY-MM-DD format) - optional update_last_indexed: Whether to update the last_indexed_at timestamp + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -305,7 +315,17 @@ async def index_obsidian_vault( indexed_count = 0 skipped_count = 0 + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for file_info in files: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(indexed_count) + last_heartbeat_time = time.time() try: file_path = file_info["path"] relative_path = file_info["relative_path"] diff --git a/surfsense_backend/app/tasks/connector_indexers/slack_indexer.py b/surfsense_backend/app/tasks/connector_indexers/slack_indexer.py index f244c97f8..5923c8089 100644 --- a/surfsense_backend/app/tasks/connector_indexers/slack_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/slack_indexer.py @@ -2,6 +2,8 @@ Slack connector indexer. """ +import time +from collections.abc import Awaitable, Callable from datetime import datetime from slack_sdk.errors import SlackApiError @@ -29,6 +31,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds - update notification every 30 seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_slack_messages( session: AsyncSession, @@ -38,6 +46,7 @@ async def index_slack_messages( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Slack messages from all accessible channels. @@ -50,6 +59,8 @@ async def index_slack_messages( start_date: Start date for indexing (YYYY-MM-DD format) end_date: End date for indexing (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. + Called periodically with (indexed_count) to prevent task appearing stuck. Returns: Tuple containing (number of documents indexed, error message or None) @@ -164,6 +175,9 @@ async def index_slack_messages( documents_skipped = 0 skipped_channels = [] + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + await task_logger.log_task_progress( log_entry, f"Starting to process {len(channels)} Slack channels", @@ -172,6 +186,13 @@ async def index_slack_messages( # Process each channel for channel_obj in channels: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() channel_id = channel_obj["id"] channel_name = channel_obj["name"] is_private = channel_obj["is_private"] diff --git a/surfsense_backend/app/tasks/connector_indexers/teams_indexer.py b/surfsense_backend/app/tasks/connector_indexers/teams_indexer.py index 66b709ddc..162509a1e 100644 --- a/surfsense_backend/app/tasks/connector_indexers/teams_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/teams_indexer.py @@ -2,6 +2,8 @@ Microsoft Teams connector indexer. """ +import time +from collections.abc import Awaitable, Callable from datetime import UTC from sqlalchemy.exc import SQLAlchemyError @@ -28,6 +30,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds - update notification every 30 seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_teams_messages( session: AsyncSession, @@ -37,6 +45,7 @@ async def index_teams_messages( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index Microsoft Teams messages from all accessible teams and channels. @@ -49,6 +58,8 @@ async def index_teams_messages( start_date: Start date for indexing (YYYY-MM-DD format) end_date: End date for indexing (YYYY-MM-DD format) update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. + Called periodically with (indexed_count) to prevent task appearing stuck. Returns: Tuple containing (number of documents indexed, error message or None) @@ -161,6 +172,9 @@ async def index_teams_messages( documents_skipped = 0 skipped_channels = [] + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + await task_logger.log_task_progress( log_entry, f"Starting to process {len(teams)} Teams", @@ -185,6 +199,14 @@ async def index_teams_messages( # Process each team for team in teams: + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() + team_id = team.get("id") team_name = team.get("displayName", "Unknown Team") diff --git a/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py b/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py index 0c63fd2f0..ac16ecde6 100644 --- a/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py +++ b/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py @@ -2,6 +2,8 @@ Webcrawler connector indexer. """ +import time +from collections.abc import Awaitable, Callable from datetime import datetime from sqlalchemy.exc import SQLAlchemyError @@ -29,6 +31,12 @@ from .base import ( update_connector_last_indexed, ) +# Type hint for heartbeat callback +HeartbeatCallbackType = Callable[[int], Awaitable[None]] + +# Heartbeat interval in seconds +HEARTBEAT_INTERVAL_SECONDS = 30 + async def index_crawled_urls( session: AsyncSession, @@ -38,6 +46,7 @@ async def index_crawled_urls( start_date: str | None = None, end_date: str | None = None, update_last_indexed: bool = True, + on_heartbeat_callback: HeartbeatCallbackType | None = None, ) -> tuple[int, str | None]: """ Index web page URLs. @@ -50,6 +59,7 @@ async def index_crawled_urls( start_date: Start date for filtering (YYYY-MM-DD format) - optional end_date: End date for filtering (YYYY-MM-DD format) - optional update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + on_heartbeat_callback: Optional callback to update notification during long-running indexing. Returns: Tuple containing (number of documents indexed, error message or None) @@ -140,7 +150,17 @@ async def index_crawled_urls( documents_skipped = 0 failed_urls = [] + # Heartbeat tracking - update notification periodically to prevent appearing stuck + last_heartbeat_time = time.time() + for idx, url in enumerate(urls, 1): + # Check if it's time for a heartbeat update + if ( + on_heartbeat_callback + and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS + ): + await on_heartbeat_callback(documents_indexed) + last_heartbeat_time = time.time() try: logger.info(f"Processing URL {idx}/{len(urls)}: {url}") diff --git a/surfsense_web/components/assistant-ui/connector-popup.tsx b/surfsense_web/components/assistant-ui/connector-popup.tsx index 20d76faa2..7184d393f 100644 --- a/surfsense_web/components/assistant-ui/connector-popup.tsx +++ b/surfsense_web/components/assistant-ui/connector-popup.tsx @@ -240,6 +240,10 @@ export const ConnectorIndicator: FC = () => { ...editingConnector, config: connectorConfig || editingConnector.config, name: editingConnector.name, + // Sync last_indexed_at with live data from Electric SQL for real-time updates + last_indexed_at: + (connectors as SearchSourceConnector[]).find((c) => c.id === editingConnector.id) + ?.last_indexed_at ?? editingConnector.last_indexed_at, }} startDate={startDate} endDate={endDate} diff --git a/surfsense_web/components/assistant-ui/connector-popup/components/date-range-selector.tsx b/surfsense_web/components/assistant-ui/connector-popup/components/date-range-selector.tsx index 7490aa959..9d7044bb4 100644 --- a/surfsense_web/components/assistant-ui/connector-popup/components/date-range-selector.tsx +++ b/surfsense_web/components/assistant-ui/connector-popup/components/date-range-selector.tsx @@ -7,6 +7,7 @@ import { Button } from "@/components/ui/button"; import { Calendar } from "@/components/ui/calendar"; import { Label } from "@/components/ui/label"; import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover"; +import { formatRelativeDate } from "@/lib/format-date"; import { cn } from "@/lib/utils"; interface DateRangeSelectorProps { @@ -26,19 +27,10 @@ export const DateRangeSelector: FC = ({ allowFutureDates = false, lastIndexedAt, }) => { - // Get the placeholder text for start date based on whether connector was previously indexed - const getStartDatePlaceholder = () => { - if (lastIndexedAt) { - const date = new Date(lastIndexedAt); - const currentYear = new Date().getFullYear(); - const indexedYear = date.getFullYear(); - // Show year only if different from current year - const formatStr = indexedYear === currentYear ? "MMM d, HH:mm" : "MMM d, yyyy HH:mm"; - const formattedDate = format(date, formatStr); - return `Since (${formattedDate})`; - } - return "Default (1 year ago)"; - }; + const startDatePlaceholder = lastIndexedAt + ? `From ${formatRelativeDate(lastIndexedAt)}` + : "Default (1 year)"; + const handleLast30Days = () => { const today = new Date(); onStartDateChange(subDays(today, 30)); @@ -88,7 +80,7 @@ export const DateRangeSelector: FC = ({ )} > - {startDate ? format(startDate, "PPP") : getStartDatePlaceholder()} + {startDate ? format(startDate, "PPP") : startDatePlaceholder} diff --git a/surfsense_web/components/assistant-ui/connector-popup/connector-configs/components/discord-config.tsx b/surfsense_web/components/assistant-ui/connector-popup/connector-configs/components/discord-config.tsx index dd4c89c8e..f782a6f4d 100644 --- a/surfsense_web/components/assistant-ui/connector-popup/connector-configs/components/discord-config.tsx +++ b/surfsense_web/components/assistant-ui/connector-popup/connector-configs/components/discord-config.tsx @@ -1,29 +1,188 @@ "use client"; -import { Info } from "lucide-react"; -import type { FC } from "react"; +import { AlertCircle, CheckCircle2, Hash, Info, Megaphone, RefreshCw } from "lucide-react"; +import { type FC, useCallback, useEffect, useState } from "react"; +import { Button } from "@/components/ui/button"; +import { Spinner } from "@/components/ui/spinner"; +import { connectorsApiService, type DiscordChannel } from "@/lib/apis/connectors-api.service"; +import { cn } from "@/lib/utils"; import type { ConnectorConfigProps } from "../index"; export interface DiscordConfigProps extends ConnectorConfigProps { onNameChange?: (name: string) => void; } -export const DiscordConfig: FC = () => { +export const DiscordConfig: FC = ({ connector }) => { + const [channels, setChannels] = useState([]); + const [isLoading, setIsLoading] = useState(false); + const [error, setError] = useState(null); + const [lastFetched, setLastFetched] = useState(null); + + const fetchChannels = useCallback(async () => { + if (!connector?.id) return; + + setIsLoading(true); + setError(null); + + try { + const data = await connectorsApiService.getDiscordChannels(connector.id); + setChannels(data); + setLastFetched(new Date()); + } catch (err) { + console.error("Failed to fetch Discord channels:", err); + setError(err instanceof Error ? err.message : "Failed to fetch channels"); + } finally { + setIsLoading(false); + } + }, [connector?.id]); + + // Fetch channels on mount + useEffect(() => { + fetchChannels(); + }, [fetchChannels]); + + // Auto-refresh when user returns to tab + useEffect(() => { + const handleVisibilityChange = () => { + if (document.visibilityState === "visible" && connector?.id) { + fetchChannels(); + } + }; + + document.addEventListener("visibilitychange", handleVisibilityChange); + return () => document.removeEventListener("visibilitychange", handleVisibilityChange); + }, [connector?.id, fetchChannels]); + + // Separate channels by indexing capability + const readyToIndex = channels.filter((ch) => ch.can_index); + const needsPermissions = channels.filter((ch) => !ch.can_index); + + // Format last fetched time + const formatLastFetched = () => { + if (!lastFetched) return null; + const now = new Date(); + const diffMs = now.getTime() - lastFetched.getTime(); + const diffSecs = Math.floor(diffMs / 1000); + const diffMins = Math.floor(diffSecs / 60); + + if (diffSecs < 60) return "just now"; + if (diffMins === 1) return "1 minute ago"; + if (diffMins < 60) return `${diffMins} minutes ago`; + return lastFetched.toLocaleTimeString(); + }; + return (
+ {/* Info box */}
-

Add Bot to Servers

- Before indexing, make sure the Discord bot has been added to the servers (guilds) you - want to index. The bot can only access messages from servers it's been added to. Use the - OAuth authorization flow to add the bot to your servers. + The bot needs "Read Message History" permission to index channels. Ask a + server admin to grant this permission for channels shown below.

+ + {/* Channels Section */} +
+
+
+

Channel Access

+
+
+ {lastFetched && ( + {formatLastFetched()} + )} + +
+
+ + {error && ( +
+ {error} +
+ )} + + {isLoading && channels.length === 0 ? ( +
+ + Loading channels +
+ ) : channels.length === 0 && !error ? ( +
+ No channels found. Make sure the bot has been added to your Discord server with proper + permissions. +
+ ) : ( +
+ {/* Ready to index */} + {readyToIndex.length > 0 && ( +
0 && "border-b border-border")}> +
+ + Ready to index + + {readyToIndex.length} {readyToIndex.length === 1 ? "channel" : "channels"} + +
+
+ {readyToIndex.map((channel) => ( + + ))} +
+
+ )} + + {/* Needs permissions */} + {needsPermissions.length > 0 && ( +
+
+ + Grant permissions to index + + {needsPermissions.length}{" "} + {needsPermissions.length === 1 ? "channel" : "channels"} + +
+
+ {needsPermissions.map((channel) => ( + + ))} +
+
+ )} +
+ )} +
+
+ ); +}; + +interface ChannelPillProps { + channel: DiscordChannel; +} + +const ChannelPill: FC = ({ channel }) => { + return ( +
+ {channel.type === "announcement" ? ( + + ) : ( + + )} + {channel.name}
); }; diff --git a/surfsense_web/components/assistant-ui/connector-popup/connector-configs/components/slack-config.tsx b/surfsense_web/components/assistant-ui/connector-popup/connector-configs/components/slack-config.tsx index 58293c4de..ff01ac96a 100644 --- a/surfsense_web/components/assistant-ui/connector-popup/connector-configs/components/slack-config.tsx +++ b/surfsense_web/components/assistant-ui/connector-popup/connector-configs/components/slack-config.tsx @@ -1,16 +1,79 @@ "use client"; -import { Info } from "lucide-react"; -import type { FC } from "react"; +import { AlertCircle, CheckCircle2, Hash, Info, Lock, RefreshCw } from "lucide-react"; +import { type FC, useCallback, useEffect, useState } from "react"; +import { Button } from "@/components/ui/button"; +import { Spinner } from "@/components/ui/spinner"; +import { connectorsApiService, type SlackChannel } from "@/lib/apis/connectors-api.service"; +import { cn } from "@/lib/utils"; import type { ConnectorConfigProps } from "../index"; export interface SlackConfigProps extends ConnectorConfigProps { onNameChange?: (name: string) => void; } -export const SlackConfig: FC = () => { +export const SlackConfig: FC = ({ connector }) => { + const [channels, setChannels] = useState([]); + const [isLoading, setIsLoading] = useState(false); + const [error, setError] = useState(null); + const [lastFetched, setLastFetched] = useState(null); + + const fetchChannels = useCallback(async () => { + if (!connector?.id) return; + + setIsLoading(true); + setError(null); + + try { + const data = await connectorsApiService.getSlackChannels(connector.id); + setChannels(data); + setLastFetched(new Date()); + } catch (err) { + console.error("Failed to fetch Slack channels:", err); + setError(err instanceof Error ? err.message : "Failed to fetch channels"); + } finally { + setIsLoading(false); + } + }, [connector?.id]); + + // Fetch channels on mount + useEffect(() => { + fetchChannels(); + }, [fetchChannels]); + + // Auto-refresh when user returns to tab + useEffect(() => { + const handleVisibilityChange = () => { + if (document.visibilityState === "visible" && connector?.id) { + fetchChannels(); + } + }; + + document.addEventListener("visibilitychange", handleVisibilityChange); + return () => document.removeEventListener("visibilitychange", handleVisibilityChange); + }, [connector?.id, fetchChannels]); + + // Separate channels by bot membership + const channelsWithBot = channels.filter((ch) => ch.is_member); + const channelsWithoutBot = channels.filter((ch) => !ch.is_member); + + // Format last fetched time + const formatLastFetched = () => { + if (!lastFetched) return null; + const now = new Date(); + const diffMs = now.getTime() - lastFetched.getTime(); + const diffSecs = Math.floor(diffMs / 1000); + const diffMins = Math.floor(diffSecs / 60); + + if (diffSecs < 60) return "just now"; + if (diffMins === 1) return "1 minute ago"; + if (diffMins < 60) return `${diffMins} minutes ago`; + return lastFetched.toLocaleTimeString(); + }; + return (
+ {/* Info box */}
@@ -25,6 +88,103 @@ export const SlackConfig: FC = () => {

+ + {/* Channels Section */} +
+
+
+

Channel Access

+
+
+ {lastFetched && ( + {formatLastFetched()} + )} + +
+
+ + {error && ( +
+ {error} +
+ )} + + {isLoading && channels.length === 0 ? ( +
+ + Loading channels +
+ ) : channels.length === 0 && !error ? ( +
+ No channels found. Make sure the bot has been added to your Slack workspace. +
+ ) : ( +
+ {/* Channels with bot access */} + {channelsWithBot.length > 0 && ( +
0 && "border-b border-border")}> +
+ + Ready to index + + {channelsWithBot.length} {channelsWithBot.length === 1 ? "channel" : "channels"} + +
+
+ {channelsWithBot.map((channel) => ( + + ))} +
+
+ )} + + {/* Channels without bot access */} + {channelsWithoutBot.length > 0 && ( +
+
+ + Add bot to index + + {channelsWithoutBot.length}{" "} + {channelsWithoutBot.length === 1 ? "channel" : "channels"} + +
+
+ {channelsWithoutBot.map((channel) => ( + + ))} +
+
+ )} +
+ )} +
+
+ ); +}; + +interface ChannelPillProps { + channel: SlackChannel; +} + +const ChannelPill: FC = ({ channel }) => { + return ( +
+ {channel.is_private ? ( + + ) : ( + + )} + {channel.name}
); }; diff --git a/surfsense_web/components/assistant-ui/connector-popup/hooks/use-indexing-connectors.ts b/surfsense_web/components/assistant-ui/connector-popup/hooks/use-indexing-connectors.ts index 19741e020..5783540d8 100644 --- a/surfsense_web/components/assistant-ui/connector-popup/hooks/use-indexing-connectors.ts +++ b/surfsense_web/components/assistant-ui/connector-popup/hooks/use-indexing-connectors.ts @@ -5,6 +5,44 @@ import type { SearchSourceConnector } from "@/contracts/types/connector.types"; import type { InboxItem } from "@/contracts/types/inbox.types"; import { isConnectorIndexingMetadata } from "@/contracts/types/inbox.types"; +/** + * Timeout thresholds for stuck task detection + * + * These align with the backend Celery configuration: + * - HARD_TIMEOUT: 8 hours (task_time_limit=28800 in Celery) + * Any task running longer than this is definitely dead. + * + * - STALE_THRESHOLD: 15 minutes without notification updates + * If heartbeats are being sent every 30s, missing 15+ minutes of updates + * indicates the task has likely crashed or the worker is down. + */ +const HARD_TIMEOUT_MS = 8 * 60 * 60 * 1000; // 8 hours in milliseconds +const STALE_THRESHOLD_MS = 15 * 60 * 1000; // 15 minutes in milliseconds + +/** + * Check if a notification is stale (no updates for too long) + * @param updatedAt - ISO timestamp of last notification update + * @returns true if the notification hasn't been updated recently + */ +function isNotificationStale(updatedAt: string | null | undefined): boolean { + if (!updatedAt) return false; + const lastUpdate = new Date(updatedAt).getTime(); + const now = Date.now(); + return now - lastUpdate > STALE_THRESHOLD_MS; +} + +/** + * Check if a task has exceeded the hard timeout (definitely dead) + * @param startedAt - ISO timestamp when the task started + * @returns true if the task has been running longer than the hard limit + */ +function isTaskTimedOut(startedAt: string | null | undefined): boolean { + if (!startedAt) return false; + const startTime = new Date(startedAt).getTime(); + const now = Date.now(); + return now - startTime > HARD_TIMEOUT_MS; +} + /** * Hook to track which connectors are currently indexing using local state. * @@ -13,6 +51,8 @@ import { isConnectorIndexingMetadata } from "@/contracts/types/inbox.types"; * 2. Detecting in_progress notifications from Electric SQL to restore state after remounts * 3. Clearing indexing state when notifications become completed or failed * 4. Clearing indexing state when Electric SQL detects last_indexed_at changed + * 5. Detecting stale/stuck tasks that haven't updated in 15+ minutes + * 6. Detecting hard timeout (8h) - tasks that definitely cannot still be running * * The actual `last_indexed_at` value comes from Electric SQL/PGlite, not local state. */ @@ -57,6 +97,7 @@ export function useIndexingConnectors( // Detect notification status changes and update indexing state accordingly // This restores spinner state after component remounts and handles all status transitions + // Also detects stale/stuck tasks that haven't been updated in a while useEffect(() => { if (!inboxItems || inboxItems.length === 0) return; @@ -71,11 +112,26 @@ export function useIndexingConnectors( const metadata = isConnectorIndexingMetadata(item.metadata) ? item.metadata : null; if (!metadata) continue; - // If status is "in_progress", add connector to indexing set + // If status is "in_progress", check if it's actually still running if (metadata.status === "in_progress") { - if (!newIndexingIds.has(metadata.connector_id)) { - newIndexingIds.add(metadata.connector_id); - hasChanges = true; + // Check for hard timeout (8h) - task is definitely dead + const timedOut = isTaskTimedOut(metadata.started_at); + + // Check for stale notification (15min without updates) - task likely crashed + const stale = isNotificationStale(item.updated_at); + + if (timedOut || stale) { + // Task is stuck - don't show as indexing + if (newIndexingIds.has(metadata.connector_id)) { + newIndexingIds.delete(metadata.connector_id); + hasChanges = true; + } + } else { + // Task appears to be genuinely running + if (!newIndexingIds.has(metadata.connector_id)) { + newIndexingIds.add(metadata.connector_id); + hasChanges = true; + } } } // If status is "completed" or "failed", remove connector from indexing set diff --git a/surfsense_web/components/assistant-ui/connector-popup/views/connector-accounts-list-view.tsx b/surfsense_web/components/assistant-ui/connector-popup/views/connector-accounts-list-view.tsx index bf151ab43..2f0e0eb5e 100644 --- a/surfsense_web/components/assistant-ui/connector-popup/views/connector-accounts-list-view.tsx +++ b/surfsense_web/components/assistant-ui/connector-popup/views/connector-accounts-list-view.tsx @@ -1,6 +1,5 @@ "use client"; -import { differenceInDays, differenceInMinutes, format, isToday, isYesterday } from "date-fns"; import { ArrowLeft, Plus, Server } from "lucide-react"; import type { FC } from "react"; import { Button } from "@/components/ui/button"; @@ -8,6 +7,7 @@ import { Spinner } from "@/components/ui/spinner"; import { EnumConnectorName } from "@/contracts/enums/connector"; import { getConnectorIcon } from "@/contracts/enums/connectorIcons"; import type { SearchSourceConnector } from "@/contracts/types/connector.types"; +import { formatRelativeDate } from "@/lib/format-date"; import { cn } from "@/lib/utils"; import { useConnectorStatus } from "../hooks/use-connector-status"; import { getConnectorDisplayName } from "../tabs/all-connectors-tab"; @@ -32,38 +32,6 @@ function isIndexableConnector(connectorType: string): boolean { return !nonIndexableTypes.includes(connectorType); } -/** - * Format last indexed date with contextual messages - */ -function formatLastIndexedDate(dateString: string): string { - const date = new Date(dateString); - const now = new Date(); - const minutesAgo = differenceInMinutes(now, date); - const daysAgo = differenceInDays(now, date); - - if (minutesAgo < 1) { - return "Just now"; - } - - if (minutesAgo < 60) { - return `${minutesAgo} ${minutesAgo === 1 ? "minute" : "minutes"} ago`; - } - - if (isToday(date)) { - return `Today at ${format(date, "h:mm a")}`; - } - - if (isYesterday(date)) { - return `Yesterday at ${format(date, "h:mm a")}`; - } - - if (daysAgo < 7) { - return `${daysAgo} ${daysAgo === 1 ? "day" : "days"} ago`; - } - - return format(date, "MMM d, yyyy"); -} - export const ConnectorAccountsListView: FC = ({ connectorType, connectorTitle, @@ -215,7 +183,7 @@ export const ConnectorAccountsListView: FC = ({

{isIndexableConnector(connector.connector_type) ? connector.last_indexed_at - ? `Last indexed: ${formatLastIndexedDate(connector.last_indexed_at)}` + ? `Last indexed: ${formatRelativeDate(connector.last_indexed_at)}` : "Never indexed" : "Active"}

diff --git a/surfsense_web/components/layout/ui/sidebar/Sidebar.tsx b/surfsense_web/components/layout/ui/sidebar/Sidebar.tsx index 4a587cd58..db04bf6dc 100644 --- a/surfsense_web/components/layout/ui/sidebar/Sidebar.tsx +++ b/surfsense_web/components/layout/ui/sidebar/Sidebar.tsx @@ -1,6 +1,6 @@ "use client"; -import { FolderOpen, MessageSquare, PenSquare } from "lucide-react"; +import { FolderOpen, PenSquare } from "lucide-react"; import { useTranslations } from "next-intl"; import { Button } from "@/components/ui/button"; import { Tooltip, TooltipContent, TooltipTrigger } from "@/components/ui/tooltip"; @@ -122,33 +122,15 @@ export function Sidebar({ {/* Chat sections - fills available space */} {isCollapsed ? ( -
- {(chats.length > 0 || sharedChats.length > 0) && ( - - - - - - {t("chats")} ({chats.length + sharedChats.length}) - - - )} -
+
) : (
- {/* Shared Chats Section - takes half the space */} + {/* Shared Chats Section - takes only space needed, max 50% */} @@ -170,9 +152,9 @@ export function Sidebar({ } > {sharedChats.length > 0 ? ( -
+
4 ? "pb-8" : ""}`} + className={`flex flex-col gap-0.5 max-h-full overflow-y-auto scrollbar-thin scrollbar-thumb-muted-foreground/20 scrollbar-track-transparent ${sharedChats.length > 4 ? "pb-8" : ""}`} > {sharedChats.slice(0, 20).map((chat) => ( - {/* Private Chats Section - takes half the space */} + {/* Private Chats Section - fills remaining space */}
@@ -56,12 +61,8 @@ export function SidebarSection({ )}
- -
+ +
{children}
diff --git a/surfsense_web/contracts/types/connector.types.ts b/surfsense_web/contracts/types/connector.types.ts index 5082fe49c..a7760745d 100644 --- a/surfsense_web/contracts/types/connector.types.ts +++ b/surfsense_web/contracts/types/connector.types.ts @@ -203,6 +203,46 @@ export const listGoogleDriveFoldersResponse = z.object({ items: z.array(googleDriveItem), }); +/** + * Slack channel with bot membership status + */ +export const slackChannel = z.object({ + id: z.string(), + name: z.string(), + is_private: z.boolean(), + is_member: z.boolean(), +}); + +/** + * List Slack channels + */ +export const listSlackChannelsRequest = z.object({ + connector_id: z.number(), +}); + +export const listSlackChannelsResponse = z.array(slackChannel); + +/** + * Discord channel with indexing permission info + */ +export const discordChannel = z.object({ + id: z.string(), + name: z.string(), + type: z.enum(["text", "announcement"]), + position: z.number(), + category_id: z.string().nullable().optional(), + can_index: z.boolean(), +}); + +/** + * List Discord channels + */ +export const listDiscordChannelsRequest = z.object({ + connector_id: z.number(), +}); + +export const listDiscordChannelsResponse = z.array(discordChannel); + // Inferred types export type SearchSourceConnectorType = z.infer; export type SearchSourceConnector = z.infer; @@ -223,3 +263,9 @@ export type ListGitHubRepositoriesResponse = z.infer; export type ListGoogleDriveFoldersResponse = z.infer; export type GoogleDriveItem = z.infer; +export type SlackChannel = z.infer; +export type ListSlackChannelsRequest = z.infer; +export type ListSlackChannelsResponse = z.infer; +export type DiscordChannel = z.infer; +export type ListDiscordChannelsRequest = z.infer; +export type ListDiscordChannelsResponse = z.infer; diff --git a/surfsense_web/lib/apis/connectors-api.service.ts b/surfsense_web/lib/apis/connectors-api.service.ts index 10e08dc71..45898b762 100644 --- a/surfsense_web/lib/apis/connectors-api.service.ts +++ b/surfsense_web/lib/apis/connectors-api.service.ts @@ -5,6 +5,7 @@ import { type DeleteConnectorRequest, deleteConnectorRequest, deleteConnectorResponse, + type DiscordChannel, type GetConnectorRequest, type GetConnectorsRequest, getConnectorRequest, @@ -16,10 +17,13 @@ import { indexConnectorResponse, type ListGitHubRepositoriesRequest, type ListGoogleDriveFoldersRequest, + listDiscordChannelsResponse, listGitHubRepositoriesRequest, listGitHubRepositoriesResponse, listGoogleDriveFoldersRequest, listGoogleDriveFoldersResponse, + listSlackChannelsResponse, + type SlackChannel, type UpdateConnectorRequest, updateConnectorRequest, updateConnectorResponse, @@ -335,6 +339,36 @@ class ConnectorsApiService { } ); }; + + // ============================================================================= + // Slack Connector Methods + // ============================================================================= + + /** + * Get Slack channels with bot membership status + */ + getSlackChannels = async (connectorId: number) => { + return baseApiService.get( + `/api/v1/slack/connector/${connectorId}/channels`, + listSlackChannelsResponse + ); + }; + + // ============================================================================= + // Discord Connector Methods + // ============================================================================= + + /** + * Get Discord text channels for a connector + */ + getDiscordChannels = async (connectorId: number) => { + return baseApiService.get( + `/api/v1/discord/connector/${connectorId}/channels`, + listDiscordChannelsResponse + ); + }; } +export type { SlackChannel, DiscordChannel }; + export const connectorsApiService = new ConnectorsApiService(); diff --git a/surfsense_web/lib/electric/client.ts b/surfsense_web/lib/electric/client.ts index d25e268be..177a66d28 100644 --- a/surfsense_web/lib/electric/client.ts +++ b/surfsense_web/lib/electric/client.ts @@ -55,7 +55,8 @@ const pendingSyncs = new Map>(); // Version for sync state - increment this to force fresh sync when Electric config changes // v2: user-specific database architecture // v3: consistent cutoff date for sync+queries, visibility refresh support -const SYNC_VERSION = 3; +// v4: heartbeat-based stale notification detection with updated_at tracking +const SYNC_VERSION = 4; // Database name prefix for identifying SurfSense databases const DB_PREFIX = "surfsense-"; diff --git a/surfsense_web/lib/format-date.ts b/surfsense_web/lib/format-date.ts new file mode 100644 index 000000000..ee60d113d --- /dev/null +++ b/surfsense_web/lib/format-date.ts @@ -0,0 +1,24 @@ +import { differenceInDays, differenceInMinutes, format, isToday, isYesterday } from "date-fns"; + +/** + * Format a date string as a human-readable relative time + * - < 1 min: "Just now" + * - < 60 min: "15m ago" + * - Today: "Today, 2:30 PM" + * - Yesterday: "Yesterday, 2:30 PM" + * - < 7 days: "3d ago" + * - Older: "Jan 15, 2026" + */ +export function formatRelativeDate(dateString: string): string { + const date = new Date(dateString); + const now = new Date(); + const minutesAgo = differenceInMinutes(now, date); + const daysAgo = differenceInDays(now, date); + + if (minutesAgo < 1) return "Just now"; + if (minutesAgo < 60) return `${minutesAgo}m ago`; + if (isToday(date)) return `Today, ${format(date, "h:mm a")}`; + if (isYesterday(date)) return `Yesterday, ${format(date, "h:mm a")}`; + if (daysAgo < 7) return `${daysAgo}d ago`; + return format(date, "MMM d, yyyy"); +}