Add Mastodon/ActivityPub connector for social media data indexing

Implement a read-only connector for Mastodon-compatible instances (Mastodon,
Pixelfed, BookWyrm) that indexes user's statuses, favourites, and bookmarks.

Features:
- MastodonConnector API client with pagination support
- HTML content cleaning and markdown formatting
- Incremental indexing using since_id tracking
- Authentication route for adding/testing connectors
- Search integration in researcher agent nodes
- Celery background task for non-blocking indexing

surfsense_backend/app/agents/researcher/nodes.py

@@ -216,6 +216,14 @@ async def search_single_connector(
                 top_k=top_k,
                 search_mode=search_mode,
             )
+        elif connector == "MASTODON_CONNECTOR":
+            source_object, chunks = await connector_service.search_mastodon(
+                user_query=reformulated_query,
+                user_id=user_id,
+                search_space_id=search_space_id,
+                top_k=top_k,
+                search_mode=search_mode,
+            )
     except Exception as e:
         logging.error("Error searching connector %s: %s", connector, traceback.format_exc())
         return (connector, None, [], str(e))
@@ -692,6 +700,7 @@ async def fetch_documents_by_ids(
                 "AIRTABLE_CONNECTOR": "Airtable (Selected)",
                 "LUMA_CONNECTOR": "Luma Events (Selected)",
                 "HOME_ASSISTANT_CONNECTOR": "Home Assistant (Selected)",
+                "MASTODON_CONNECTOR": "Mastodon (Selected)",
             }
 
             source_object = {

surfsense_backend/app/agents/researcher/utils.py

@@ -47,6 +47,7 @@ CONNECTOR_METADATA: dict[str, ConnectorMetadata] = {
     "LUMA_CONNECTOR": ConnectorMetadata("🎯", "Luma events", "Luma"),
     "ELASTICSEARCH_CONNECTOR": ConnectorMetadata("🔎", "Elasticsearch chunks", "Elasticsearch"),
     "HOME_ASSISTANT_CONNECTOR": ConnectorMetadata("🏠", "Home Assistant items", "Home Assistant"),
+    "MASTODON_CONNECTOR": ConnectorMetadata("🐘", "Mastodon posts", "Mastodon"),
 }
 
 # Default metadata for unknown connectors

surfsense_backend/app/connectors/mastodon_connector.py

@@ -0,0 +1,405 @@
+"""
+Mastodon/ActivityPub connector for fetching social media data.
+
+This connector interfaces with the Mastodon REST API to retrieve:
+- User's own posts (statuses)
+- Favorited posts
+- Bookmarked posts
+- Notifications
+
+Works with Mastodon, Pixelfed, and other Mastodon-compatible instances.
+"""
+
+import logging
+from datetime import datetime
+from typing import Any
+
+import aiohttp
+
+logger = logging.getLogger(__name__)
+
+
+class MastodonConnector:
+    """Connector for Mastodon and compatible ActivityPub platforms (Pixelfed, etc.)."""
+
+    def __init__(self, instance_url: str, access_token: str):
+        """
+        Initialize the Mastodon connector.
+
+        Args:
+            instance_url: Base URL of Mastodon instance (e.g., https://mastodon.social)
+            access_token: User access token for authentication
+        """
+        self.instance_url = instance_url.rstrip("/")
+        self.access_token = access_token
+        self.headers = {
+            "Authorization": f"Bearer {access_token}",
+            "Content-Type": "application/json",
+        }
+
+    async def _make_request(
+        self, endpoint: str, method: str = "GET", params: dict | None = None
+    ) -> tuple[Any | None, str | None]:
+        """
+        Make an authenticated request to Mastodon API.
+
+        Args:
+            endpoint: API endpoint (e.g., /api/v1/accounts/verify_credentials)
+            method: HTTP method
+            params: Optional query parameters
+
+        Returns:
+            Tuple of (response_data, error_message)
+        """
+        url = f"{self.instance_url}{endpoint}"
+
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.request(
+                    method, url, headers=self.headers, params=params, timeout=30
+                ) as response:
+                    if response.status == 200:
+                        return await response.json(), None
+                    elif response.status == 401:
+                        return None, "Invalid or expired access token"
+                    elif response.status == 403:
+                        return None, "Access forbidden - check token scopes"
+                    elif response.status == 404:
+                        return None, f"Endpoint not found: {endpoint}"
+                    elif response.status == 429:
+                        return None, "Rate limit exceeded - try again later"
+                    else:
+                        error_text = await response.text()
+                        return None, f"API error {response.status}: {error_text}"
+        except aiohttp.ClientConnectorError as e:
+            return None, f"Connection error: Unable to reach {self.instance_url}. {e!s}"
+        except TimeoutError:
+            return None, f"Request timeout connecting to {self.instance_url}"
+        except Exception as e:
+            return None, f"Unexpected error: {e!s}"
+
+    async def _paginate_request(
+        self, endpoint: str, params: dict | None = None, max_items: int = 200
+    ) -> tuple[list[dict], str | None]:
+        """
+        Make paginated requests to fetch all items.
+
+        Args:
+            endpoint: API endpoint
+            params: Optional query parameters
+            max_items: Maximum number of items to fetch
+
+        Returns:
+            Tuple of (items_list, error_message)
+        """
+        all_items = []
+        params = params or {}
+        params["limit"] = min(40, max_items)  # Mastodon default max per page
+
+        while len(all_items) < max_items:
+            result, error = await self._make_request(endpoint, params=params)
+            if error:
+                if all_items:
+                    # Return what we have so far
+                    return all_items, f"Partial fetch: {error}"
+                return [], error
+
+            if not result:
+                break
+
+            all_items.extend(result)
+
+            if len(result) < params["limit"]:
+                # No more items
+                break
+
+            # Get the last item's ID for pagination
+            if result:
+                params["max_id"] = result[-1]["id"]
+
+        return all_items[:max_items], None
+
+    async def verify_credentials(self) -> tuple[dict | None, str | None]:
+        """
+        Verify the access token and get account info.
+
+        Returns:
+            Tuple of (account_info, error_message)
+        """
+        return await self._make_request("/api/v1/accounts/verify_credentials")
+
+    async def test_connection(self) -> tuple[bool, str | None]:
+        """
+        Test the connection to the Mastodon instance.
+
+        Returns:
+            Tuple of (success, error_message)
+        """
+        result, error = await self.verify_credentials()
+        if error:
+            return False, error
+        return True, None
+
+    async def get_account_statuses(
+        self, account_id: str, max_items: int = 100, since_id: str | None = None
+    ) -> tuple[list[dict], str | None]:
+        """
+        Get statuses (posts) from a specific account.
+
+        Args:
+            account_id: The account ID
+            max_items: Maximum number of statuses to fetch
+            since_id: Only fetch statuses newer than this ID
+
+        Returns:
+            Tuple of (statuses_list, error_message)
+        """
+        params = {"exclude_replies": "false", "exclude_reblogs": "false"}
+        if since_id:
+            params["since_id"] = since_id
+
+        return await self._paginate_request(
+            f"/api/v1/accounts/{account_id}/statuses", params=params, max_items=max_items
+        )
+
+    async def get_own_statuses(
+        self, max_items: int = 100, since_id: str | None = None
+    ) -> tuple[list[dict], str | None]:
+        """
+        Get the authenticated user's own statuses.
+
+        Args:
+            max_items: Maximum number of statuses to fetch
+            since_id: Only fetch statuses newer than this ID
+
+        Returns:
+            Tuple of (statuses_list, error_message)
+        """
+        # First get account info
+        account, error = await self.verify_credentials()
+        if error:
+            return [], error
+
+        return await self.get_account_statuses(
+            account["id"], max_items=max_items, since_id=since_id
+        )
+
+    async def get_favourites(
+        self, max_items: int = 100
+    ) -> tuple[list[dict], str | None]:
+        """
+        Get the user's favorited statuses.
+
+        Returns:
+            Tuple of (favourites_list, error_message)
+        """
+        return await self._paginate_request(
+            "/api/v1/favourites", max_items=max_items
+        )
+
+    async def get_bookmarks(
+        self, max_items: int = 100
+    ) -> tuple[list[dict], str | None]:
+        """
+        Get the user's bookmarked statuses.
+
+        Returns:
+            Tuple of (bookmarks_list, error_message)
+        """
+        return await self._paginate_request(
+            "/api/v1/bookmarks", max_items=max_items
+        )
+
+    async def get_notifications(
+        self, max_items: int = 50, types: list[str] | None = None
+    ) -> tuple[list[dict], str | None]:
+        """
+        Get the user's notifications.
+
+        Args:
+            max_items: Maximum number of notifications to fetch
+            types: Filter by notification types (mention, favourite, reblog, follow, poll, etc.)
+
+        Returns:
+            Tuple of (notifications_list, error_message)
+        """
+        params = {}
+        if types:
+            params["types[]"] = types
+
+        return await self._paginate_request(
+            "/api/v1/notifications", params=params, max_items=max_items
+        )
+
+    async def get_instance_info(self) -> tuple[dict | None, str | None]:
+        """
+        Get information about the Mastodon instance.
+
+        Returns:
+            Tuple of (instance_info, error_message)
+        """
+        # Try v2 first, fall back to v1
+        result, error = await self._make_request("/api/v2/instance")
+        if error:
+            result, error = await self._make_request("/api/v1/instance")
+        return result, error
+
+    async def get_all_indexable_data(
+        self,
+        max_statuses: int = 100,
+        max_favourites: int = 50,
+        max_bookmarks: int = 50,
+        since_id: str | None = None,
+    ) -> tuple[list[dict], str | None]:
+        """
+        Get all data suitable for indexing.
+
+        This includes:
+        - User's own statuses
+        - Favorited statuses
+        - Bookmarked statuses
+
+        Returns:
+            Tuple of (indexable_items, error_message)
+        """
+        items = []
+        errors = []
+
+        # Get user's own statuses
+        statuses, error = await self.get_own_statuses(
+            max_items=max_statuses, since_id=since_id
+        )
+        if error:
+            errors.append(f"Statuses: {error}")
+        else:
+            for status in statuses:
+                items.append({
+                    "type": "status",
+                    "source": "own",
+                    "data": status,
+                })
+
+        # Get favourites
+        favourites, error = await self.get_favourites(max_items=max_favourites)
+        if error:
+            errors.append(f"Favourites: {error}")
+        else:
+            for status in favourites:
+                items.append({
+                    "type": "status",
+                    "source": "favourite",
+                    "data": status,
+                })
+
+        # Get bookmarks
+        bookmarks, error = await self.get_bookmarks(max_items=max_bookmarks)
+        if error:
+            errors.append(f"Bookmarks: {error}")
+        else:
+            for status in bookmarks:
+                items.append({
+                    "type": "status",
+                    "source": "bookmark",
+                    "data": status,
+                })
+
+        error_msg = "; ".join(errors) if errors else None
+        return items, error_msg
+
+    def format_status_to_markdown(self, item: dict) -> str:
+        """Format a status item to markdown."""
+        source = item.get("source", "unknown")
+        status = item.get("data", {})
+
+        # Extract status info
+        account = status.get("account", {})
+        username = account.get("acct", "unknown")
+        display_name = account.get("display_name", username)
+        content = status.get("content", "")
+        created_at = status.get("created_at", "")
+        url = status.get("url", "")
+        visibility = status.get("visibility", "public")
+
+        # Clean HTML content (basic cleaning)
+        import re
+        content_text = re.sub(r"<[^>]+>", "", content)
+        content_text = content_text.replace("&amp;", "&")
+        content_text = content_text.replace("&lt;", "<")
+        content_text = content_text.replace("&gt;", ">")
+        content_text = content_text.replace("&quot;", '"')
+        content_text = content_text.replace("&#39;", "'")
+
+        # Build markdown
+        source_label = {
+            "own": "My Post",
+            "favourite": "Favorited Post",
+            "bookmark": "Bookmarked Post",
+        }.get(source, "Post")
+
+        md = f"# {source_label}\n\n"
+        md += f"**Author:** {display_name} (@{username})\n"
+        md += f"**Date:** {created_at}\n"
+        md += f"**Visibility:** {visibility}\n"
+        if url:
+            md += f"**URL:** {url}\n"
+
+        # Engagement stats
+        reblogs = status.get("reblogs_count", 0)
+        favourites = status.get("favourites_count", 0)
+        replies = status.get("replies_count", 0)
+        if reblogs or favourites or replies:
+            md += f"**Engagement:** {reblogs} boosts, {favourites} favourites, {replies} replies\n"
+
+        md += f"\n## Content\n\n{content_text}\n"
+
+        # Media attachments
+        media = status.get("media_attachments", [])
+        if media:
+            md += "\n## Media\n\n"
+            for i, attachment in enumerate(media, 1):
+                media_type = attachment.get("type", "unknown")
+                media_url = attachment.get("url", "")
+                description = attachment.get("description", "No description")
+                md += f"{i}. [{media_type}]({media_url})"
+                if description:
+                    md += f" - {description}"
+                md += "\n"
+
+        # Tags/hashtags
+        tags = status.get("tags", [])
+        if tags:
+            tag_names = [f"#{tag.get('name', '')}" for tag in tags]
+            md += f"\n**Tags:** {' '.join(tag_names)}\n"
+
+        # Mentions
+        mentions = status.get("mentions", [])
+        if mentions:
+            mention_names = [f"@{m.get('acct', '')}" for m in mentions]
+            md += f"**Mentions:** {' '.join(mention_names)}\n"
+
+        # Poll if present
+        poll = status.get("poll")
+        if poll:
+            md += "\n## Poll\n\n"
+            for option in poll.get("options", []):
+                title = option.get("title", "")
+                votes = option.get("votes_count", 0)
+                md += f"- {title}: {votes} votes\n"
+
+        # Reblog info
+        reblog = status.get("reblog")
+        if reblog:
+            reblog_account = reblog.get("account", {})
+            md += f"\n*Boosted from @{reblog_account.get('acct', 'unknown')}*\n"
+
+        return md
+
+    def format_item_to_markdown(self, item: dict) -> str:
+        """Format any item to markdown based on its type."""
+        item_type = item.get("type", "")
+
+        if item_type == "status":
+            return self.format_status_to_markdown(item)
+        else:
+            # Generic formatting
+            return f"# {item.get('type', 'Unknown')}\n\n{item}"

surfsense_backend/app/db.py

@@ -51,6 +51,7 @@ class DocumentType(str, Enum):
     LUMA_CONNECTOR = "LUMA_CONNECTOR"
     ELASTICSEARCH_CONNECTOR = "ELASTICSEARCH_CONNECTOR"
     HOME_ASSISTANT_CONNECTOR = "HOME_ASSISTANT_CONNECTOR"
+    MASTODON_CONNECTOR = "MASTODON_CONNECTOR"
 
 
 class SearchSourceConnectorType(str, Enum):
@@ -73,6 +74,7 @@ class SearchSourceConnectorType(str, Enum):
     LUMA_CONNECTOR = "LUMA_CONNECTOR"
     ELASTICSEARCH_CONNECTOR = "ELASTICSEARCH_CONNECTOR"
     HOME_ASSISTANT_CONNECTOR = "HOME_ASSISTANT_CONNECTOR"
+    MASTODON_CONNECTOR = "MASTODON_CONNECTOR"
 
 
 class ChatType(str, Enum):

surfsense_backend/app/routes/__init__.py

@@ -15,6 +15,7 @@ from .home_assistant_add_connector_route import (
     router as home_assistant_add_connector_router,
 )
 from .llm_config_routes import router as llm_config_router
+from .mastodon_add_connector_route import router as mastodon_add_connector_router
 from .logs_routes import router as logs_router
 from .luma_add_connector_route import router as luma_add_connector_router
 from .podcasts_routes import router as podcasts_router
@@ -35,6 +36,7 @@ router.include_router(google_gmail_add_connector_router)
 router.include_router(airtable_add_connector_router)
 router.include_router(luma_add_connector_router)
 router.include_router(home_assistant_add_connector_router)
+router.include_router(mastodon_add_connector_router)
 router.include_router(llm_config_router)
 router.include_router(logs_router)
 router.include_router(site_configuration_router)

surfsense_backend/app/routes/mastodon_add_connector_route.py

@@ -0,0 +1,178 @@
+"""
+Mastodon/ActivityPub connector authentication route.
+
+This route allows users to add a Mastodon connector by providing
+their instance URL and access token.
+
+Works with Mastodon, Pixelfed, and other Mastodon-compatible instances.
+"""
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.connectors.mastodon_connector import MastodonConnector
+from app.db import SearchSourceConnector, SearchSourceConnectorType, User, get_async_session
+from app.users import current_active_user
+
+router = APIRouter()
+
+
+class MastodonConnectorRequest(BaseModel):
+    """Request model for adding a Mastodon connector."""
+
+    search_space_id: int
+    instance_url: str  # e.g., https://mastodon.social
+    access_token: str
+    name: str = "Mastodon"
+
+
+class MastodonConnectorResponse(BaseModel):
+    """Response model for Mastodon connector creation."""
+
+    id: int
+    name: str
+    connector_type: str
+    message: str
+    username: str
+
+
+@router.post("/auth/mastodon/add", response_model=MastodonConnectorResponse)
+async def add_mastodon_connector(
+    request: MastodonConnectorRequest,
+    user: User = Depends(current_active_user),
+    session: AsyncSession = Depends(get_async_session),
+):
+    """
+    Add a Mastodon/ActivityPub connector.
+
+    This endpoint allows users to connect their Mastodon, Pixelfed, or other
+    ActivityPub-compatible account by providing the instance URL and access token.
+
+    To create an access token:
+    1. Go to your Mastodon instance Settings → Development → New Application
+    2. Give it a name (e.g., "SurfSense")
+    3. Select read scopes: read:accounts, read:bookmarks, read:favourites, read:statuses
+    4. Create the application and copy the access token
+
+    Args:
+        request: The connector configuration
+        user: Current authenticated user
+        session: Database session
+
+    Returns:
+        MastodonConnectorResponse with connector details
+    """
+    # Validate and normalize URL
+    instance_url = request.instance_url.rstrip("/")
+    if not instance_url.startswith(("http://", "https://")):
+        instance_url = f"https://{instance_url}"
+
+    # Test connection to Mastodon
+    mastodon_client = MastodonConnector(
+        instance_url=instance_url,
+        access_token=request.access_token,
+    )
+
+    # Verify credentials and get account info
+    account, error = await mastodon_client.verify_credentials()
+    if error:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Failed to connect to Mastodon: {error}",
+        )
+
+    # Get username and display name
+    username = account.get("acct", "unknown")
+    display_name = account.get("display_name", username)
+
+    # Create connector config
+    connector_config = {
+        "INSTANCE_URL": instance_url,
+        "ACCESS_TOKEN": request.access_token,
+        "ACCOUNT_ID": account.get("id"),
+        "USERNAME": username,
+    }
+
+    # Use display name or provided name
+    connector_name = request.name if request.name != "Mastodon" else f"Mastodon (@{username})"
+
+    # Create the connector in the database
+    db_connector = SearchSourceConnector(
+        name=connector_name,
+        connector_type=SearchSourceConnectorType.MASTODON_CONNECTOR,
+        config=connector_config,
+        search_space_id=request.search_space_id,
+        user_id=str(user.id),
+        is_indexable=True,
+    )
+
+    session.add(db_connector)
+    await session.commit()
+    await session.refresh(db_connector)
+
+    return MastodonConnectorResponse(
+        id=db_connector.id,
+        name=db_connector.name,
+        connector_type=db_connector.connector_type.value,
+        message=f"Successfully connected to {instance_url} as @{username}",
+        username=username,
+    )
+
+
+@router.post("/auth/mastodon/test")
+async def test_mastodon_connection(
+    instance_url: str,
+    access_token: str,
+    user: User = Depends(current_active_user),
+):
+    """
+    Test connection to a Mastodon instance.
+
+    This endpoint allows users to verify their credentials before
+    creating a connector.
+
+    Args:
+        instance_url: Mastodon instance URL
+        access_token: User access token
+        user: Current authenticated user
+
+    Returns:
+        Connection status and account info
+    """
+    # Normalize URL
+    instance_url = instance_url.rstrip("/")
+    if not instance_url.startswith(("http://", "https://")):
+        instance_url = f"https://{instance_url}"
+
+    # Test connection
+    mastodon_client = MastodonConnector(
+        instance_url=instance_url,
+        access_token=access_token,
+    )
+
+    account, error = await mastodon_client.verify_credentials()
+    if error:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Connection failed: {error}",
+        )
+
+    # Get instance info
+    instance_info, _ = await mastodon_client.get_instance_info()
+
+    return {
+        "status": "success",
+        "message": "Successfully connected to Mastodon",
+        "account_info": {
+            "username": account.get("acct", "unknown"),
+            "display_name": account.get("display_name", ""),
+            "followers_count": account.get("followers_count", 0),
+            "following_count": account.get("following_count", 0),
+            "statuses_count": account.get("statuses_count", 0),
+        },
+        "instance_info": {
+            "title": instance_info.get("title", "Unknown") if instance_info else "Unknown",
+            "version": instance_info.get("version", "Unknown") if instance_info else "Unknown",
+        },
+    }

surfsense_backend/app/routes/search_source_connectors_routes.py

@@ -48,6 +48,7 @@ from app.tasks.connector_indexers import (
     index_jira_issues,
     index_linear_issues,
     index_luma_events,
+    index_mastodon_data,
     index_notion_pages,
     index_slack_messages,
 )
@@ -705,6 +706,22 @@ async def index_connector_content(
             )
             response_message = "Home Assistant indexing started in the background."
 
+        elif (
+            connector.connector_type
+            == SearchSourceConnectorType.MASTODON_CONNECTOR
+        ):
+            from app.tasks.celery_tasks.connector_tasks import (
+                index_mastodon_data_task,
+            )
+
+            logger.info(
+                f"Triggering Mastodon indexing for connector {connector_id} into search space {search_space_id}"
+            )
+            index_mastodon_data_task.delay(
+                connector_id, search_space_id, str(user.id), indexing_from, indexing_to
+            )
+            response_message = "Mastodon indexing started in the background."
+
         else:
             raise HTTPException(
                 status_code=400,
@@ -1578,3 +1595,41 @@ async def run_home_assistant_indexing(
             f"Critical error in run_home_assistant_indexing for connector {connector_id}: {e}",
             exc_info=True,
         )
+
+
+async def run_mastodon_indexing(
+    session: AsyncSession,
+    connector_id: int,
+    search_space_id: int,
+    user_id: str,
+    start_date: str,
+    end_date: str,
+):
+    """Runs the Mastodon indexing task and updates the timestamp."""
+    try:
+        indexed_count, error_message = await index_mastodon_data(
+            session,
+            connector_id,
+            search_space_id,
+            user_id,
+            start_date,
+            end_date,
+            update_last_indexed=False,
+        )
+        if error_message:
+            logger.error(
+                f"Mastodon indexing failed for connector {connector_id}: {error_message}"
+            )
+        else:
+            logger.info(
+                f"Mastodon indexing successful for connector {connector_id}. Indexed {indexed_count} items."
+            )
+            # Update the last indexed timestamp only on success
+            await update_connector_last_indexed(session, connector_id)
+            await session.commit()
+    except Exception as e:
+        await session.rollback()
+        logger.error(
+            f"Critical error in run_mastodon_indexing for connector {connector_id}: {e}",
+            exc_info=True,
+        )

surfsense_backend/app/services/connector_service.py

@@ -2645,3 +2645,93 @@ class ConnectorService:
         }
 
         return result_object, ha_chunks
+
+    async def search_mastodon(
+        self,
+        user_query: str,
+        user_id: str,
+        search_space_id: int,
+        top_k: int = 20,
+        search_mode: SearchMode = SearchMode.CHUNKS,
+    ) -> tuple:
+        """
+        Search for Mastodon documents and return both the source information and langchain documents
+
+        Args:
+            user_query: The user's query
+            user_id: The user's ID
+            search_space_id: The search space ID to search in
+            top_k: Maximum number of results to return
+            search_mode: Search mode (CHUNKS or DOCUMENTS)
+
+        Returns:
+            tuple: (sources_info, langchain_documents)
+        """
+        if search_mode == SearchMode.CHUNKS:
+            mastodon_chunks = await self.chunk_retriever.hybrid_search(
+                query_text=user_query,
+                top_k=top_k,
+                user_id=user_id,
+                search_space_id=search_space_id,
+                document_type="MASTODON_CONNECTOR",
+            )
+        elif search_mode == SearchMode.DOCUMENTS:
+            mastodon_chunks = await self.document_retriever.hybrid_search(
+                query_text=user_query,
+                top_k=top_k,
+                user_id=user_id,
+                search_space_id=search_space_id,
+                document_type="MASTODON_CONNECTOR",
+            )
+            mastodon_chunks = self._transform_document_results(mastodon_chunks)
+
+        if not mastodon_chunks:
+            return {
+                "id": 36,
+                "name": "Mastodon",
+                "type": "MASTODON_CONNECTOR",
+                "sources": [],
+            }, []
+
+        sources_list = []
+        async with self.counter_lock:
+            for _i, chunk in enumerate(mastodon_chunks):
+                document = chunk.get("document", {})
+                metadata = document.get("metadata", {})
+
+                status_id = metadata.get("status_id", "")
+                source = metadata.get("source", "")
+                username = metadata.get("username", "")
+                status_url = metadata.get("url", "")
+
+                title = document.get("title", "Mastodon Post")
+                description = chunk.get("content", "")[:150]
+                if len(description) == 150:
+                    description += "..."
+
+                favourites = metadata.get("favourites_count", 0)
+                reblogs = metadata.get("reblogs_count", 0)
+                if favourites or reblogs:
+                    description = f"[{favourites}❤️ {reblogs}🔁] {description}"
+
+                source_entry = {
+                    "id": chunk.get("chunk_id", self.source_id_counter),
+                    "title": title,
+                    "description": description,
+                    "url": status_url,
+                    "status_id": status_id,
+                    "source": source,
+                    "username": username,
+                }
+
+                self.source_id_counter += 1
+                sources_list.append(source_entry)
+
+        result_object = {
+            "id": 36,
+            "name": "Mastodon",
+            "type": "MASTODON_CONNECTOR",
+            "sources": sources_list,
+        }
+
+        return result_object, mastodon_chunks

surfsense_backend/app/tasks/celery_tasks/connector_tasks.py

@@ -643,3 +643,46 @@ async def _index_home_assistant_data(
         await run_home_assistant_indexing(
             session, connector_id, search_space_id, user_id, start_date, end_date
         )
+
+
+@celery_app.task(name="index_mastodon_data", bind=True)
+def index_mastodon_data_task(
+    self,
+    connector_id: int,
+    search_space_id: int,
+    user_id: str,
+    start_date: str,
+    end_date: str,
+):
+    """Celery task to index Mastodon data."""
+    import asyncio
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    try:
+        loop.run_until_complete(
+            _index_mastodon_data(
+                connector_id, search_space_id, user_id, start_date, end_date
+            )
+        )
+    finally:
+        loop.close()
+
+
+async def _index_mastodon_data(
+    connector_id: int,
+    search_space_id: int,
+    user_id: str,
+    start_date: str,
+    end_date: str,
+):
+    """Index Mastodon data with new session."""
+    from app.routes.search_source_connectors_routes import (
+        run_mastodon_indexing,
+    )
+
+    async with get_celery_session_maker()() as session:
+        await run_mastodon_indexing(
+            session, connector_id, search_space_id, user_id, start_date, end_date
+        )

surfsense_backend/app/tasks/connector_indexers/__init__.py

@@ -19,6 +19,7 @@ Available indexers:
 - Luma: Index events from Luma
 - Elasticsearch: Index documents from Elasticsearch instances
 - Home Assistant: Index automations, scripts, and events from Home Assistant
+- Mastodon: Index posts, favourites, and bookmarks from Mastodon/Pixelfed
 """
 
 # Communication platforms
@@ -40,6 +41,9 @@ from .jira_indexer import index_jira_issues
 from .linear_indexer import index_linear_issues
 from .luma_indexer import index_luma_events
 
+# Social media
+from .mastodon_indexer import index_mastodon_data
+
 # Documentation and knowledge management
 from .notion_indexer import index_notion_pages
 from .slack_indexer import index_slack_messages
@@ -65,4 +69,6 @@ __all__ = [  # noqa: RUF022
     "index_google_gmail_messages",
     # Smart home
     "index_home_assistant_data",
+    # Social media
+    "index_mastodon_data",
 ]

surfsense_backend/app/tasks/connector_indexers/mastodon_indexer.py

@@ -0,0 +1,320 @@
+"""
+Mastodon/ActivityPub connector indexer.
+"""
+
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import config
+from app.connectors.mastodon_connector import MastodonConnector
+from app.db import Document, DocumentType, SearchSourceConnectorType
+from app.services.task_logging_service import TaskLoggingService
+from app.utils.document_converters import (
+    create_document_chunks,
+    generate_content_hash,
+    generate_unique_identifier_hash,
+)
+
+from .base import (
+    check_document_by_unique_identifier,
+    get_connector_by_id,
+    logger,
+    update_connector_last_indexed,
+)
+
+
+async def index_mastodon_data(
+    session: AsyncSession,
+    connector_id: int,
+    search_space_id: int,
+    user_id: str,
+    start_date: str | None = None,
+    end_date: str | None = None,
+    update_last_indexed: bool = True,
+) -> tuple[int, str | None]:
+    """
+    Index Mastodon data including user's statuses, favourites, and bookmarks.
+
+    Args:
+        session: Database session
+        connector_id: ID of the Mastodon connector
+        search_space_id: ID of the search space to store documents in
+        user_id: User ID
+        start_date: Start date for indexing (not used for Mastodon, uses since_id instead)
+        end_date: End date for indexing (not used for Mastodon)
+        update_last_indexed: Whether to update the last_indexed_at timestamp (default: True)
+
+    Returns:
+        Tuple containing (number of documents indexed, error message or None)
+    """
+    task_logger = TaskLoggingService(session, search_space_id)
+
+    # Log task start
+    log_entry = await task_logger.log_task_start(
+        task_name="mastodon_indexing",
+        source="connector_indexing_task",
+        message=f"Starting Mastodon indexing for connector {connector_id}",
+        metadata={
+            "connector_id": connector_id,
+            "user_id": str(user_id),
+        },
+    )
+
+    try:
+        # Get the connector from the database
+        connector = await get_connector_by_id(
+            session, connector_id, SearchSourceConnectorType.MASTODON_CONNECTOR
+        )
+
+        if not connector:
+            await task_logger.log_task_failure(
+                log_entry,
+                f"Connector with ID {connector_id} not found",
+                "Connector not found",
+                {"error_type": "ConnectorNotFound"},
+            )
+            return 0, f"Connector with ID {connector_id} not found"
+
+        # Get Mastodon credentials from the connector config
+        instance_url = connector.config.get("INSTANCE_URL")
+        access_token = connector.config.get("ACCESS_TOKEN")
+
+        if not instance_url or not access_token:
+            await task_logger.log_task_failure(
+                log_entry,
+                f"Mastodon credentials not found in connector config for connector {connector_id}",
+                "Missing Mastodon credentials",
+                {"error_type": "MissingCredentials"},
+            )
+            return 0, "Mastodon credentials not found in connector config"
+
+        # Initialize Mastodon client
+        await task_logger.log_task_progress(
+            log_entry,
+            f"Initializing Mastodon client for connector {connector_id}",
+            {"stage": "client_initialization"},
+        )
+
+        mastodon_client = MastodonConnector(
+            instance_url=instance_url, access_token=access_token
+        )
+
+        # Test connection
+        connected, connection_error = await mastodon_client.test_connection()
+        if not connected:
+            await task_logger.log_task_failure(
+                log_entry,
+                f"Failed to connect to Mastodon: {connection_error}",
+                "Connection failed",
+                {"error_type": "ConnectionError"},
+            )
+            return 0, f"Failed to connect to Mastodon: {connection_error}"
+
+        # Get last indexed status ID for incremental indexing
+        since_id = connector.config.get("LAST_STATUS_ID")
+
+        await task_logger.log_task_progress(
+            log_entry,
+            f"Fetching Mastodon data (since_id: {since_id or 'None'})",
+            {"stage": "fetching_data", "since_id": since_id},
+        )
+
+        # Get all indexable data
+        items, fetch_error = await mastodon_client.get_all_indexable_data(
+            max_statuses=200,
+            max_favourites=100,
+            max_bookmarks=100,
+            since_id=since_id,
+        )
+
+        if fetch_error:
+            logger.warning(f"Some data fetch errors occurred: {fetch_error}")
+
+        if not items:
+            logger.info("No items found to index from Mastodon")
+            if update_last_indexed:
+                await update_connector_last_indexed(
+                    session, connector, update_last_indexed
+                )
+                await session.commit()
+
+            await task_logger.log_task_success(
+                log_entry,
+                "No new items found to index",
+                {"items_indexed": 0},
+            )
+            return 0, None
+
+        await task_logger.log_task_progress(
+            log_entry,
+            f"Processing {len(items)} items from Mastodon",
+            {"stage": "processing_items", "total_items": len(items)},
+        )
+
+        # Track the newest status ID for next incremental indexing
+        newest_status_id = None
+
+        # Process each item
+        documents_indexed = 0
+
+        for item in items:
+            try:
+                status = item.get("data", {})
+                source = item.get("source", "unknown")
+                status_id = status.get("id", "")
+
+                # Track newest ID for incremental indexing
+                if status_id and (not newest_status_id or status_id > newest_status_id):
+                    newest_status_id = status_id
+
+                # Generate unique identifier based on status ID and source
+                unique_id = f"{status_id}_{source}"
+                unique_identifier_hash = generate_unique_identifier_hash(unique_id)
+
+                # Check for existing document
+                existing_doc = await check_document_by_unique_identifier(
+                    session, unique_identifier_hash
+                )
+
+                # Format content to markdown
+                content = mastodon_client.format_item_to_markdown(item)
+                content_hash = generate_content_hash(content)
+
+                # Extract account info
+                account = status.get("account", {})
+                username = account.get("acct", "unknown")
+
+                # Build metadata
+                metadata = {
+                    "status_id": status_id,
+                    "source": source,
+                    "instance_url": instance_url,
+                    "username": username,
+                    "created_at": status.get("created_at", ""),
+                    "visibility": status.get("visibility", "public"),
+                    "url": status.get("url", ""),
+                    "reblogs_count": status.get("reblogs_count", 0),
+                    "favourites_count": status.get("favourites_count", 0),
+                    "replies_count": status.get("replies_count", 0),
+                }
+
+                # Add tags
+                tags = status.get("tags", [])
+                if tags:
+                    metadata["tags"] = [tag.get("name", "") for tag in tags]
+
+                # Generate title
+                content_preview = content[:50].replace("\n", " ").strip()
+                if len(content) > 50:
+                    content_preview += "..."
+
+                source_label = {
+                    "own": "Post",
+                    "favourite": "Favourite",
+                    "bookmark": "Bookmark",
+                }.get(source, "Post")
+                title = f"Mastodon {source_label}: {content_preview}"
+
+                if existing_doc:
+                    # Check if content has changed
+                    if existing_doc.content_hash == content_hash:
+                        logger.debug(f"Skipping unchanged item: {status_id}")
+                        continue
+
+                    # Update existing document
+                    existing_doc.title = title
+                    existing_doc.content = content
+                    existing_doc.content_hash = content_hash
+                    existing_doc.document_metadata = metadata
+
+                    # Delete old chunks and create new ones
+                    existing_doc.chunks.clear()
+
+                    # Create new chunks
+                    chunks = create_document_chunks(
+                        content,
+                        existing_doc,
+                        config.chunker_instance,
+                        config.embedding_model_instance,
+                    )
+                    for chunk in chunks:
+                        existing_doc.chunks.append(chunk)
+
+                    documents_indexed += 1
+                    logger.info(f"Updated existing Mastodon item: {status_id}")
+
+                else:
+                    # Create new document
+                    document = Document(
+                        title=title,
+                        document_type=DocumentType.MASTODON_CONNECTOR,
+                        document_metadata=metadata,
+                        content=content,
+                        content_hash=content_hash,
+                        unique_identifier_hash=unique_identifier_hash,
+                        search_space_id=search_space_id,
+                    )
+
+                    # Create chunks
+                    chunks = create_document_chunks(
+                        content,
+                        document,
+                        config.chunker_instance,
+                        config.embedding_model_instance,
+                    )
+                    for chunk in chunks:
+                        document.chunks.append(chunk)
+
+                    session.add(document)
+                    documents_indexed += 1
+                    logger.info(f"Indexed new Mastodon item: {status_id}")
+
+            except Exception as e:
+                logger.error(f"Error processing item {item.get('data', {}).get('id', 'unknown')}: {e!s}")
+                continue
+
+        # Update last indexed timestamp and newest status ID
+        if update_last_indexed:
+            await update_connector_last_indexed(session, connector, update_last_indexed)
+
+            # Store newest status ID for incremental indexing
+            if newest_status_id:
+                connector.config["LAST_STATUS_ID"] = newest_status_id
+
+        # Commit all changes
+        await session.commit()
+
+        await task_logger.log_task_success(
+            log_entry,
+            f"Successfully indexed {documents_indexed} Mastodon items",
+            {"items_indexed": documents_indexed, "newest_status_id": newest_status_id},
+        )
+
+        logger.info(f"Successfully indexed {documents_indexed} Mastodon items")
+        return documents_indexed, None
+
+    except SQLAlchemyError as e:
+        await session.rollback()
+        error_message = f"Database error during Mastodon indexing: {e!s}"
+        logger.error(error_message)
+
+        await task_logger.log_task_failure(
+            log_entry,
+            error_message,
+            "Database error",
+            {"error_type": "DatabaseError"},
+        )
+        return 0, error_message
+
+    except Exception as e:
+        await session.rollback()
+        error_message = f"Unexpected error during Mastodon indexing: {e!s}"
+        logger.error(error_message)
+
+        await task_logger.log_task_failure(
+            log_entry,
+            error_message,
+            "Unexpected error",
+            {"error_type": "UnexpectedError"},
+        )
+        return 0, error_message