connector.py

from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import httpx

from utils.group_acl import unique_acl_principal_labels
from utils.logging_config import get_logger

from ..base import BaseConnector, ConnectorDocument, DocumentACL
from ..microsoft_graph_acl import (
    get_current_user_microsoft_group_roles,
    get_current_user_microsoft_principal_labels,
    get_current_user_microsoft_principals,
    get_oauth_access_token,
    microsoft_group_principal_label,
    microsoft_group_role,
    microsoft_user_principal,
    microsoft_user_principal_label,
)
from .oauth import SharePointOAuth

logger = get_logger(__name__)


class SharePointConnector(BaseConnector):
    """SharePoint connector using MSAL-based OAuth for authentication"""

    # Required BaseConnector class attributes
    CLIENT_ID_ENV_VAR = "MICROSOFT_GRAPH_OAUTH_CLIENT_ID"
    CLIENT_SECRET_ENV_VAR = "MICROSOFT_GRAPH_OAUTH_CLIENT_SECRET"  # pragma: allowlist secret

    # Connector metadata
    CONNECTOR_TYPE = "sharepoint"
    CONNECTOR_KIND = "oauth"
    CONNECTOR_NAME = "SharePoint"
    CONNECTOR_DESCRIPTION = "Add knowledge from SharePoint"
    CONNECTOR_ICON = "sharepoint"

    @classmethod
    def get_oauth_class(cls):
        from .oauth import SharePointOAuth

        return SharePointOAuth

    def __init__(self, config: dict[str, Any]):
        super().__init__(config)

        logger.debug(f"SharePoint connector __init__ called with config type: {type(config)}")
        logger.debug(f"SharePoint connector __init__ config value: {config}")

        # Ensure we always pass a valid config to the base class
        if config is None:
            logger.debug("Config was None, using empty dict")
            config = {}

        try:
            logger.debug("Calling super().__init__")
            super().__init__(config)  # Now safe to call with empty dict instead of None
            logger.debug("super().__init__ completed successfully")
        except Exception as e:
            logger.error(f"super().__init__ failed: {e}")
            raise

        # Initialize with defaults that allow the connector to be listed
        self.client_id = None
        self.client_secret = None
        self.tenant_id = config.get("tenant_id", "common")
        # Graph delta link for webhook change tracking (in-memory per instance)
        self._delta_link: str | None = None
        # base_url is the generic field name, sharepoint_url is kept for backward compatibility
        self.sharepoint_url = config.get("base_url") or config.get("sharepoint_url")
        logger.debug(
            f"SharePoint connector initialized with sharepoint_url from config: {self.sharepoint_url}"
        )
        self.redirect_uri = config.get("redirect_uri", "http://localhost")

        # Try to get credentials, but don't fail if they're missing
        try:
            logger.debug("Attempting to get client_id")
            self.client_id = self.get_client_id()
            logger.debug(f"Got client_id: {self.client_id is not None}")
        except Exception as e:
            logger.debug(f"Failed to get client_id: {e}")
            pass  # Credentials not available, that's OK for listing

        try:
            logger.debug("Attempting to get client_secret")
            self.client_secret = self.get_client_secret()
            logger.debug(f"Got client_secret: {self.client_secret is not None}")
        except Exception as e:
            logger.debug(f"Failed to get client_secret: {e}")
            pass  # Credentials not available, that's OK for listing

        # Token file setup - use data directory for persistence
        from config.paths import get_data_file

        token_file = config.get("token_file") or get_data_file("sharepoint_token.json")
        Path(token_file).parent.mkdir(parents=True, exist_ok=True)

        # Only initialize OAuth if we have credentials
        if self.client_id and self.client_secret:
            connection_id = config.get("connection_id", "default")

            # Use token_file from config if provided, otherwise generate one
            if config.get("token_file"):
                oauth_token_file = config["token_file"]
            else:
                oauth_token_file = get_data_file(f"sharepoint_token_{connection_id}.json")

            authority = (
                f"https://login.microsoftonline.com/{self.tenant_id}"
                if self.tenant_id != "common"
                else "https://login.microsoftonline.com/common"
            )

            self.oauth = SharePointOAuth(
                client_id=self.client_id,
                client_secret=self.client_secret,
                token_file=oauth_token_file,
                authority=authority,
            )
        else:
            self.oauth = None

        # Track subscription ID for webhooks
        self._subscription_id: str | None = None

        # Set by setup_subscription/renew_subscription; read by the
        # connection manager to persist
        self.webhook_resource_id: str | None = None
        self.webhook_expiration: str | None = None

        # Add Graph API defaults similar to Google Drive flags
        self._graph_api_version = "v1.0"
        self._default_params = {
            "$select": "id,name,size,lastModifiedDateTime,createdDateTime,webUrl,file,folder,@microsoft.graph.downloadUrl"
        }

        # Selective sync support (similar to Google Drive and OneDrive)
        self.cfg = type(
            "SharePointConfig",
            (),
            {
                "file_ids": config.get("file_ids")
                or config.get("selected_files")
                or config.get("selected_file_ids"),
                "folder_ids": config.get("folder_ids")
                or config.get("selected_folders")
                or config.get("selected_folder_ids"),
            },
        )()

        # Cache for file metadata including download URLs
        # This allows direct download without Graph API for sharing IDs
        self._file_infos: dict[str, dict[str, Any]] = {}

    @property
    def _graph_base_url(self) -> str:
        """Base URL for Microsoft Graph API calls"""
        return f"https://graph.microsoft.com/{self._graph_api_version}"

    @property
    def base_url(self) -> str | None:
        """Generic base URL property (returns sharepoint_url for SharePoint connector)"""
        return self.sharepoint_url

    @base_url.setter
    def base_url(self, value: str):
        """Set base URL (updates sharepoint_url internally)"""
        self.sharepoint_url = value

    async def get_current_user_group_roles(self) -> list[str]:
        """Return canonical group ACL roles for the connected Microsoft user."""
        return await get_current_user_microsoft_group_roles(
            self.oauth,
            self._graph_base_url,
            tenant_id=self.tenant_id,
        )

    async def get_current_user_principals(self) -> list[str]:
        """Return canonical user ACL principals for the connected Microsoft user."""
        return await get_current_user_microsoft_principals(
            self.oauth,
            self._graph_base_url,
            tenant_id=self.tenant_id,
        )

    async def get_current_user_principal_labels(self) -> list[dict[str, Any]]:
        """Return display labels for current Microsoft user/group ACL principals."""
        return await get_current_user_microsoft_principal_labels(
            self.oauth,
            self._graph_base_url,
            tenant_id=self.tenant_id,
        )

    def set_file_infos(self, file_infos: list[dict[str, Any]]) -> None:
        """
        Cache file metadata including download URLs for later use.
        This allows direct download without Graph API calls for sharing IDs.

        Args:
            file_infos: List of file info dicts with {id, name, mimeType, downloadUrl, size}
        """
        self._file_infos = {}
        for info in file_infos:
            file_id = info.get("id")
            if file_id:
                self._file_infos[file_id] = info
                if info.get("downloadUrl"):
                    logger.debug(f"Cached download URL for file {file_id}: {info.get('name')}")

    def get_cached_file_info(self, file_id: str) -> dict[str, Any] | None:
        """Get cached file info by ID."""
        return self._file_infos.get(file_id)

    def emit(self, doc: ConnectorDocument) -> None:
        """
        Emit a ConnectorDocument instance.
        """
        logger.debug(f"Emitting SharePoint document: {doc.id} ({doc.filename})")

    async def authenticate(self) -> bool:
        """Test authentication - BaseConnector interface"""
        logger.debug(f"SharePoint authenticate() called, oauth is None: {self.oauth is None}")
        try:
            if not self.oauth:
                logger.debug("SharePoint authentication failed: OAuth not initialized")
                self._authenticated = False
                return False

            logger.debug("Loading SharePoint credentials...")
            # Try to load existing credentials first
            load_result = await self.oauth.load_credentials()
            logger.debug(f"Load credentials result: {load_result}")

            logger.debug("Checking SharePoint authentication status...")
            authenticated = await self.oauth.is_authenticated()
            logger.debug(f"SharePoint is_authenticated result: {authenticated}")

            self._authenticated = authenticated
            return authenticated
        except Exception:
            logger.exception("[CONNECTOR] SharePoint authentication failed")
            self._authenticated = False
            return False

    def get_auth_url(self) -> str:
        """Get OAuth authorization URL"""
        if not self.oauth:
            raise RuntimeError("SharePoint OAuth not initialized - missing credentials")
        return self.oauth.create_authorization_url(self.redirect_uri)

    async def handle_oauth_callback(self, auth_code: str) -> dict[str, Any]:
        """Handle OAuth callback"""
        if not self.oauth:
            raise RuntimeError("SharePoint OAuth not initialized - missing credentials")
        try:
            success = await self.oauth.handle_authorization_callback(auth_code, self.redirect_uri)
            if success:
                self._authenticated = True

                # Auto-detect base URL from user's drive
                detected_url = await self._detect_base_url()
                if detected_url:
                    self.base_url = detected_url
                    logger.info(f"Auto-detected base URL: {detected_url}")

                return {"status": "success", "base_url": self.base_url}
            else:
                raise ValueError("OAuth callback failed")
        except Exception as e:
            logger.error(f"OAuth callback failed: {e}")
            raise

    async def _detect_base_url(self) -> str | None:
        """Override base class method to detect SharePoint URL"""
        return await self._detect_sharepoint_url()

    async def _detect_sharepoint_url(self) -> str | None:
        """Auto-detect SharePoint URL from Microsoft Graph API"""
        logger.info("_detect_sharepoint_url: Starting SharePoint URL detection")
        try:
            if not self.oauth:
                logger.warning("_detect_sharepoint_url: OAuth not initialized")
                return None

            access_token = self.oauth.get_access_token()
            logger.debug(
                f"_detect_sharepoint_url: Got access token (length: {len(access_token) if access_token else 0})"
            )

            headers = {
                "Authorization": f"Bearer {access_token}",
                "Content-Type": "application/json",
            }

            async with httpx.AsyncClient() as client:
                # Get user's default drive to extract SharePoint URL
                url = f"{self._graph_base_url}/me/drive"
                logger.info(f"_detect_sharepoint_url: Calling Graph API: {url}")

                response = await client.get(url, headers=headers, timeout=30.0)
                logger.info(
                    f"_detect_sharepoint_url: Graph API response status: {response.status_code}"
                )

                if response.status_code == 200:
                    data = response.json()
                    web_url = data.get("webUrl", "")
                    logger.info(f"_detect_sharepoint_url: webUrl from response: {web_url}")

                    # Extract the SharePoint domain from the webUrl

                    if web_url:
                        parsed = urlparse(web_url)
                        sharepoint_url = f"{parsed.scheme}://{parsed.netloc}"
                        logger.info(
                            f"_detect_sharepoint_url: Detected SharePoint URL: {sharepoint_url}"
                        )
                        return sharepoint_url
                    else:
                        logger.warning("_detect_sharepoint_url: webUrl is empty in response")
                else:
                    logger.warning(
                        "[CONNECTOR] SharePoint detect URL failed", status_code=response.status_code
                    )

        except Exception:
            logger.exception("[CONNECTOR] SharePoint URL detection failed")

        return None

    def sync_once(self) -> None:
        """
        Perform a one-shot sync of SharePoint files and emit documents.
        This method mirrors the Google Drive connector's sync_once functionality.
        """
        import asyncio

        async def _async_sync():
            try:
                # Get list of files
                file_list = await self.list_files(max_files=1000)  # Adjust as needed
                files = file_list.get("files", [])

                for file_info in files:
                    try:
                        file_id = file_info.get("id")
                        if not file_id:
                            continue

                        # Get full document content
                        doc = await self.get_file_content(file_id)
                        self.emit(doc)

                    except Exception as e:
                        logger.error(
                            f"Failed to sync SharePoint file {file_info.get('name', 'unknown')}: {e}"
                        )
                        continue

            except Exception as e:
                logger.error(f"SharePoint sync_once failed: {e}")
                raise

        # Run the async sync
        if hasattr(asyncio, "run"):
            asyncio.run(_async_sync())
        else:
            # Python < 3.7 compatibility
            loop = asyncio.get_event_loop()
            loop.run_until_complete(_async_sync())

    async def setup_subscription(self) -> str:
        """Set up real-time subscription for file changes - BaseConnector interface"""
        webhook_url = self.config.get("webhook_url")
        if not webhook_url:
            logger.warning("No webhook URL configured, skipping SharePoint subscription setup")
            return "no-webhook-configured"

        try:
            # Ensure we're authenticated
            if not await self.authenticate():
                raise RuntimeError("SharePoint authentication failed during subscription setup")

            token = self.oauth.get_access_token()

            # Microsoft Graph subscription for SharePoint site
            site_info = self._parse_sharepoint_url()
            if site_info:
                resource = (
                    f"sites/{site_info['host_name']}:/sites/{site_info['site_name']}:/drive/root"
                )
            else:
                resource = "/me/drive/root"

            subscription_data = {
                # Graph driveItem subscriptions only support "updated"; creates and
                # deletes still surface through the delta query the webhook triggers.
                "changeType": "updated",
                # webhook_url is already the full endpoint
                # ({WEBHOOK_BASE_URL}/connectors/sharepoint/webhook, set at connect time)
                "notificationUrl": webhook_url,
                "resource": resource,
                "expirationDateTime": self._get_subscription_expiry(),
                "clientState": f"sharepoint_{self.tenant_id}",
            }

            headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

            url = f"{self._graph_base_url}/subscriptions"

            async with httpx.AsyncClient() as client:
                response = await client.post(
                    url, json=subscription_data, headers=headers, timeout=30
                )
                if response.status_code >= 400:
                    logger.error(
                        f"Graph subscription request rejected: {response.status_code} {response.text}"
                    )
                response.raise_for_status()

                result = response.json()
                subscription_id = result.get("id")

                if subscription_id:
                    self._subscription_id = subscription_id
                    self.webhook_expiration = result.get("expirationDateTime")
                    logger.info(f"SharePoint subscription created: {subscription_id}")
                    return subscription_id
                else:
                    raise ValueError("No subscription ID returned from Microsoft Graph")

        except Exception as e:
            logger.error(f"Failed to setup SharePoint subscription: {e}")
            raise

    def _get_subscription_expiry(self) -> str:
        """Get subscription expiry time (max 3 days for Graph API)"""
        from datetime import datetime, timedelta

        expiry = datetime.now(UTC) + timedelta(days=3)  # 3 days max for Graph
        return expiry.strftime("%Y-%m-%dT%H:%M:%S.%fZ")

    def _parse_sharepoint_url(self) -> dict[str, str] | None:
        """Parse SharePoint URL to extract site information for Graph API"""
        if not self.sharepoint_url:
            return None

        try:
            parsed = urlparse(self.sharepoint_url)
            # Extract hostname and site name from URL like: https://contoso.sharepoint.com/sites/teamsite
            host_name = parsed.netloc
            path_parts = parsed.path.strip("/").split("/")

            if len(path_parts) >= 2 and path_parts[0] == "sites":
                site_name = path_parts[1]
                return {"host_name": host_name, "site_name": site_name}
        except Exception as e:
            logger.warning(f"Could not parse SharePoint URL {self.sharepoint_url}: {e}")

        return None

    async def list_files(
        self, page_token: str | None = None, max_files: int | None = None, **kwargs
    ) -> dict[str, Any]:
        """List all files using Microsoft Graph API - BaseConnector interface"""
        try:
            # Ensure authentication
            if not await self.authenticate():
                raise RuntimeError("SharePoint authentication failed during file listing")

            # If file_ids or folder_ids are specified in config, use selective sync
            if self.cfg.file_ids or self.cfg.folder_ids:
                return await self._list_selected_files()

            files = []
            max_files_value = max_files if max_files is not None else 100

            # Build Graph API URL for the site or fallback to user's OneDrive
            site_info = self._parse_sharepoint_url()
            if site_info:
                base_url = f"{self._graph_base_url}/sites/{site_info['host_name']}:/sites/{site_info['site_name']}:/drive/root/children"
            else:
                base_url = f"{self._graph_base_url}/me/drive/root/children"

            params = dict(self._default_params)
            params["$top"] = str(max_files_value)

            if page_token:
                params["$skiptoken"] = page_token

            response = await self._make_graph_request(base_url, params=params)
            data = response.json()

            items = data.get("value", [])
            for item in items:
                # Only include files, not folders
                if item.get("file"):
                    files.append(
                        {
                            "id": item.get("id", ""),
                            "name": item.get("name", ""),
                            "path": f"/drive/items/{item.get('id')}",
                            "size": int(item.get("size", 0)),
                            "modified": item.get("lastModifiedDateTime"),
                            "created": item.get("createdDateTime"),
                            "mime_type": item.get("file", {}).get(
                                "mimeType", self._get_mime_type(item.get("name", ""))
                            ),
                            "url": item.get("webUrl", ""),
                            "download_url": item.get("@microsoft.graph.downloadUrl"),
                        }
                    )

            # Check for next page
            next_page_token = None
            next_link = data.get("@odata.nextLink")
            if next_link:
                from urllib.parse import parse_qs, urlparse

                parsed = urlparse(next_link)
                query_params = parse_qs(parsed.query)
                if "$skiptoken" in query_params:
                    next_page_token = query_params["$skiptoken"][0]

            return {"files": files, "next_page_token": next_page_token}

        except Exception as e:
            logger.error(f"Failed to list SharePoint files: {e}")
            return {"files": [], "next_page_token": None}  # Return empty result instead of raising

    async def _extract_sharepoint_acl(self, file_id: str, file_metadata: dict) -> DocumentACL:
        """
        Extract ACL from SharePoint item.

        Queries Microsoft Graph API permissions endpoint to get allowed users and groups.

        Args:
            file_id: SharePoint item ID
            file_metadata: File metadata dict

        Returns:
            DocumentACL instance with extracted permissions
        """
        try:
            # Get access token - use same approach as _make_graph_request
            access_token = await get_oauth_access_token(self.oauth)

            if not access_token:
                logger.warning(f"No access token available for ACL extraction: {file_id}")
                return DocumentACL()

            # Determine the correct path for permissions API call. Mirror
            # _get_file_metadata_by_id: a composite "driveId!itemId" id must be
            # split into /drives/{driveId}/items/{itemId}. Using the composite id
            # verbatim against /drive/items/{id} yields a malformed URL → Graph
            # error → empty ACL, which is why shared-user updates never landed.
            if "!" in file_id and len(file_id.rsplit("!", 1)) == 2:
                drive_id, item_id = file_id.rsplit("!", 1)
                permissions_url = (
                    f"{self._graph_base_url}/drives/{drive_id}/items/{item_id}/permissions"
                )
            else:
                site_info = self._parse_sharepoint_url()
                if site_info:
                    permissions_url = f"{self._graph_base_url}/sites/{site_info['host_name']}:/sites/{site_info['site_name']}:/drive/items/{file_id}/permissions"
                else:
                    # Fallback to user drive
                    permissions_url = f"{self._graph_base_url}/me/drive/items/{file_id}/permissions"

            # Fetch permissions, following pagination so large share lists are
            # captured in full (not just the first page).
            permissions: list[dict[str, Any]] = []
            async with httpx.AsyncClient() as client:
                url: str | None = permissions_url
                while url:
                    response = await client.get(
                        url, headers={"Authorization": f"Bearer {access_token}"}
                    )
                    if response.status_code != 200:
                        logger.warning(
                            f"Failed to fetch permissions for {file_id}: {response.status_code}"
                        )
                        return DocumentACL()
                    page = response.json()
                    permissions.extend(page.get("value", []))
                    url = page.get("@odata.nextLink")

            permissions_data = {"value": permissions}

            allowed_users = []
            allowed_groups = []
            allowed_principals = []
            allowed_principal_labels = []
            owner = None

            for perm in permissions_data.get("value", []):
                roles = perm.get("roles", [])  # ["read", "write", "owner"]

                # Granted to user (grantedTo or grantedToV2)
                granted_to = perm.get("grantedToV2") or perm.get("grantedTo")
                if granted_to:
                    user_info = granted_to.get("user", {})
                    email = user_info.get("email")
                    display_name = user_info.get("displayName")
                    user_identifier = email or display_name
                    if user_identifier:
                        allowed_users.append(user_identifier)
                        if "owner" in roles:
                            owner = user_identifier
                    for identifier in (
                        user_info.get("id"),
                        user_info.get("userPrincipalName"),
                        email,
                    ):
                        user_principal = microsoft_user_principal(
                            identifier,
                            access_token=access_token,
                            tenant_id=self.tenant_id,
                        )
                        if user_principal:
                            allowed_principals.append(user_principal)
                            label = microsoft_user_principal_label(
                                identifier,
                                access_token=access_token,
                                tenant_id=self.tenant_id,
                                display_name=display_name or email,
                                email=email,
                                external_id=identifier,
                            )
                            if label:
                                allowed_principal_labels.append(label)
                    group_info = granted_to.get("group", {})
                    group_role = microsoft_group_role(
                        group_info.get("id"),
                        access_token=access_token,
                        tenant_id=self.tenant_id,
                    )
                    if group_role:
                        allowed_groups.append(group_role)
                        allowed_principals.append(group_role)
                        label = microsoft_group_principal_label(
                            group_info.get("id"),
                            access_token=access_token,
                            tenant_id=self.tenant_id,
                            display_name=group_info.get("displayName") or group_info.get("email"),
                            email=group_info.get("email"),
                        )
                        if label:
                            allowed_principal_labels.append(label)

                # Granted to identities (can include users and groups)
                if "grantedToIdentitiesV2" in perm or "grantedToIdentities" in perm:
                    identities = (
                        perm.get("grantedToIdentitiesV2") or perm.get("grantedToIdentities") or []
                    )
                    for identity in identities:
                        # User
                        if "user" in identity:
                            user_info = identity["user"]
                            email = user_info.get("email")
                            display_name = user_info.get("displayName")
                            user_identifier = email or display_name
                            if user_identifier:
                                allowed_users.append(user_identifier)
                                if "owner" in roles:
                                    owner = user_identifier
                            for identifier in (
                                user_info.get("id"),
                                user_info.get("userPrincipalName"),
                                email,
                            ):
                                user_principal = microsoft_user_principal(
                                    identifier,
                                    access_token=access_token,
                                    tenant_id=self.tenant_id,
                                )
                                if user_principal:
                                    allowed_principals.append(user_principal)

                        # Group
                        if "group" in identity:
                            group_info = identity["group"]
                            group_id = group_info.get("id")
                            group_role = microsoft_group_role(
                                group_id,
                                access_token=access_token,
                                tenant_id=self.tenant_id,
                            )
                            if group_role:
                                allowed_groups.append(group_role)
                                allowed_principals.append(group_role)
                                label = microsoft_group_principal_label(
                                    group_id,
                                    access_token=access_token,
                                    tenant_id=self.tenant_id,
                                    display_name=group_info.get("displayName")
                                    or group_info.get("email"),
                                    email=group_info.get("email"),
                                )
                                if label:
                                    allowed_principal_labels.append(label)

            return DocumentACL(
                owner=owner,
                allowed_users=allowed_users,
                allowed_groups=allowed_groups,
                allowed_principals=allowed_principals,
                allowed_principal_labels=unique_acl_principal_labels(allowed_principal_labels),
            )

        except Exception as e:
            logger.warning(f"Failed to extract ACL for SharePoint item {file_id}: {e}")
            return DocumentACL()

    async def get_file_content(self, file_id: str) -> ConnectorDocument:
        """Get file content and metadata - BaseConnector interface"""
        try:
            # Ensure authentication
            if not await self.authenticate():
                raise RuntimeError("SharePoint authentication failed during file content retrieval")

            # First, check for cached file info with download URL
            # This is used for SharePoint sharing IDs that can't be resolved via Graph API
            cached_info = self.get_cached_file_info(file_id)
            if cached_info and cached_info.get("downloadUrl"):
                logger.info(f"Using cached download URL for file {file_id}")
                content = await self._download_file_from_url(cached_info["downloadUrl"])

                # Extract ACL even for cached files
                acl = await self._extract_sharepoint_acl(file_id, cached_info)

                return ConnectorDocument(
                    id=file_id,
                    filename=cached_info.get("name", "Unknown"),
                    mimetype=cached_info.get("mimeType", "application/octet-stream"),
                    content=content,
                    source_url=cached_info.get("webUrl", ""),
                    acl=acl,
                    modified_time=datetime.now(),
                    created_time=datetime.now(),
                    metadata={
                        "sharepoint_path": "",
                        "sharepoint_url": self.sharepoint_url,
                        "size": cached_info.get("size", 0),
                    },
                )

            # Fall back to Graph API for regular file IDs
            file_metadata = await self._get_file_metadata_by_id(file_id)

            if not file_metadata:
                raise ValueError(f"File not found: {file_id}")

            # Download file content
            download_url = file_metadata.get("download_url")
            if download_url:
                content = await self._download_file_from_url(download_url)
            else:
                content = await self._download_file_content(file_id)

            # Extract ACL from SharePoint item
            acl = await self._extract_sharepoint_acl(file_id, file_metadata)

            # Parse dates
            modified_time = self._parse_graph_date(file_metadata.get("modified"))
            created_time = self._parse_graph_date(file_metadata.get("created"))

            return ConnectorDocument(
                id=file_id,
                filename=file_metadata.get("name", ""),
                mimetype=file_metadata.get("mime_type", "application/octet-stream"),
                content=content,
                source_url=file_metadata.get("url", ""),
                acl=acl,
                modified_time=modified_time,
                created_time=created_time,
                metadata={
                    "sharepoint_path": file_metadata.get("path", ""),
                    "sharepoint_url": self.sharepoint_url,
                    "size": file_metadata.get("size", 0),
                },
            )

        except Exception as e:
            logger.error(f"Failed to get SharePoint file content {file_id}: {e}")
            raise

    async def _get_file_metadata_by_id(self, file_id: str) -> dict[str, Any] | None:
        """Get file metadata by ID using Graph API"""
        try:
            # Check if ID contains '!' which indicates driveId!itemId format
            if "!" in file_id:
                parts = file_id.rsplit("!", 1)
                if len(parts) == 2:
                    drive_id, item_id = parts
                    url = f"{self._graph_base_url}/drives/{drive_id}/items/{item_id}"
                else:
                    url = f"{self._graph_base_url}/me/drive/items/{file_id}"
            else:
                site_info = self._parse_sharepoint_url()
                if site_info:
                    url = f"{self._graph_base_url}/sites/{site_info['host_name']}:/sites/{site_info['site_name']}:/drive/items/{file_id}"
                else:
                    url = f"{self._graph_base_url}/me/drive/items/{file_id}"

            params = dict(self._default_params)

            response = await self._make_graph_request(url, params=params)
            item = response.json()

            if item.get("file"):
                return {
                    "id": file_id,
                    "name": item.get("name", ""),
                    "path": f"/drive/items/{file_id}",
                    "size": int(item.get("size", 0)),
                    "modified": item.get("lastModifiedDateTime"),
                    "created": item.get("createdDateTime"),
                    "mime_type": item.get("file", {}).get(
                        "mimeType", self._get_mime_type(item.get("name", ""))
                    ),
                    "url": item.get("webUrl", ""),
                    "download_url": item.get("@microsoft.graph.downloadUrl"),
                }

            # Check if it's a folder
            if item.get("folder"):
                return {
                    "id": file_id,
                    "name": item.get("name", ""),
                    "isFolder": True,
                }

            return None

        except Exception as e:
            logger.error(
                f"Failed to get file metadata for {file_id}: {e}. "
                f"Site info: {self._parse_sharepoint_url()}, "
                f"SharePoint URL: {self.sharepoint_url}"
            )
            return None

    async def _download_file_content(self, file_id: str) -> bytes:
        """Download file content by file ID using Graph API"""
        try:
            # Check if ID contains '!' which indicates driveId!itemId format
            if "!" in file_id:
                parts = file_id.rsplit("!", 1)
                if len(parts) == 2:
                    drive_id, item_id = parts
                    if not item_id.startswith("s"):
                        url = f"{self._graph_base_url}/drives/{drive_id}/items/{item_id}/content"
                    else:
                        url = f"{self._graph_base_url}/me/drive/items/{file_id}/content"
                else:
                    url = f"{self._graph_base_url}/me/drive/items/{file_id}/content"
            else:
                site_info = self._parse_sharepoint_url()
                if site_info:
                    url = f"{self._graph_base_url}/sites/{site_info['host_name']}:/sites/{site_info['site_name']}:/drive/items/{file_id}/content"
                else:
                    url = f"{self._graph_base_url}/me/drive/items/{file_id}/content"

            token = self.oauth.get_access_token()
            headers = {"Authorization": f"Bearer {token}"}

            async with httpx.AsyncClient() as client:
                response = await client.get(url, headers=headers, timeout=60, follow_redirects=True)
                response.raise_for_status()
                return response.content

        except Exception as e:
            logger.error(f"Failed to download file content for {file_id}: {e}")
            raise

    async def _list_selected_files(self) -> dict[str, Any]:
        """List only selected files/folders (selective sync)."""
        files: list[dict[str, Any]] = []

        # Process selected file IDs
        if self.cfg.file_ids:
            for file_id in self.cfg.file_ids:
                try:
                    file_meta = await self._get_file_metadata_by_id(file_id)
                    if file_meta and not file_meta.get("isFolder", False):
                        files.append(file_meta)
                    elif file_meta and file_meta.get("isFolder", False):
                        # If it's a folder, expand its contents
                        folder_files = await self._list_folder_contents(file_id)
                        files.extend(folder_files)
                except Exception as e:
                    logger.warning(f"Failed to get file {file_id}: {e}")
                    continue

        # Process selected folder IDs
        if self.cfg.folder_ids:
            for folder_id in self.cfg.folder_ids:
                try:
                    folder_files = await self._list_folder_contents(folder_id)
                    files.extend(folder_files)
                except Exception as e:
                    logger.warning(f"Failed to list folder {folder_id}: {e}")
                    continue

        return {"files": files, "next_page_token": None}

    async def _list_folder_contents(self, folder_id: str) -> list[dict[str, Any]]:
        """List all files in a folder recursively."""
        files: list[dict[str, Any]] = []

        try:
            drive_id = None
            if "!" in folder_id:
                parts = folder_id.rsplit("!", 1)
                if len(parts) == 2:
                    potential_drive_id, item_id = parts
                    if not item_id.startswith("s"):
                        drive_id = potential_drive_id
                        url = f"{self._graph_base_url}/drives/{drive_id}/items/{item_id}/children"

            if not drive_id:
                site_info = self._parse_sharepoint_url()
                if site_info:
                    url = f"{self._graph_base_url}/sites/{site_info['host_name']}:/sites/{site_info['site_name']}:/drive/items/{folder_id}/children"
                else:
                    url = f"{self._graph_base_url}/me/drive/items/{folder_id}/children"

            params = dict(self._default_params)

            response = await self._make_graph_request(url, params=params)
            data = response.json()

            items = data.get("value", [])
            for item in items:
                parent_ref = item.get("parentReference", {})
                item_drive_id = parent_ref.get("driveId") or drive_id
                item_id = item.get("id")
                if item_id and "!" in item_id:
                    final_item_id = item_id
                else:
                    final_item_id = f"{item_drive_id}!{item_id}" if item_drive_id else item_id

                if item.get("file"):  # It's a file
                    file_meta = {
                        "id": final_item_id,
                        "name": item.get("name", ""),
                        "path": f"/drive/items/{item_id}",
                        "size": int(item.get("size") or 0),
                        "modified": item.get("lastModifiedDateTime"),
                        "created": item.get("createdDateTime"),
                        "mime_type": item.get("file", {}).get(
                            "mimeType", self._get_mime_type(item.get("name", ""))
                        ),
                        "url": item.get("webUrl", ""),
                        "download_url": item.get("@microsoft.graph.downloadUrl"),
                    }
                    files.append(file_meta)
                elif item.get("folder"):  # It's a subfolder, recurse
                    subfolder_files = await self._list_folder_contents(final_item_id)
                    files.extend(subfolder_files)
        except Exception as e:
            import traceback

            logger.error(
                f"Failed to list folder contents for {folder_id}: {e}\n{traceback.format_exc()}"
            )

        return files

    async def _download_file_from_url(self, download_url: str) -> bytes:
        """Download file content from direct download URL"""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(download_url, timeout=60, follow_redirects=True)
                response.raise_for_status()
                return response.content
        except Exception as e:
            logger.error(f"Failed to download from URL {download_url}: {e}")
            raise

    def _parse_graph_date(self, date_str: str | None) -> datetime:
        """Parse Microsoft Graph date string to datetime"""
        if not date_str:
            return datetime.now()

        try:
            if date_str.endswith("Z"):
                return datetime.fromisoformat(date_str[:-1]).replace(tzinfo=None)
            else:
                return datetime.fromisoformat(date_str.replace("T", " "))
        except (ValueError, AttributeError):
            return datetime.now()

    async def _make_graph_request(
        self,
        url: str,
        method: str = "GET",
        data: dict | None = None,
        params: dict | None = None,
    ) -> httpx.Response:
        """Make authenticated API request to Microsoft Graph"""
        token = self.oauth.get_access_token()
        headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

        async with httpx.AsyncClient() as client:
            if method.upper() == "GET":
                response = await client.get(url, headers=headers, params=params, timeout=30)
            elif method.upper() == "POST":
                response = await client.post(url, headers=headers, json=data, timeout=30)
            elif method.upper() == "DELETE":
                response = await client.delete(url, headers=headers, timeout=30)
            else:
                raise ValueError(f"Unsupported HTTP method: {method}")

            response.raise_for_status()
            return response

    def _get_mime_type(self, filename: str) -> str:
        """Get MIME type based on file extension"""
        import mimetypes

        mime_type, _ = mimetypes.guess_type(filename)
        return mime_type or "application/octet-stream"

    # Webhook methods - BaseConnector interface
    def handle_webhook_validation(
        self, request_method: str, headers: dict[str, str], query_params: dict[str, str]
    ) -> str | None:
        """Handle webhook validation (Graph API specific)"""
        if request_method == "POST" and "validationToken" in query_params:
            return query_params["validationToken"]
        return None

    def extract_webhook_channel_id(
        self, payload: dict[str, Any], headers: dict[str, str]
    ) -> str | None:
        """Extract channel/subscription ID from webhook payload"""
        notifications = payload.get("value", [])
        if notifications:
            return notifications[0].get("subscriptionId")
        return None

    @staticmethod
    def _delta_item_file_id(item: dict[str, Any]) -> str:
        """Return the composite ``{driveId}!{itemId}`` id used at ingest time.

        Selected-file listing stores ids as ``driveId!itemId`` (see
        _list_folder_contents), so the webhook delta must emit the same shape
        or the change can't be correlated with the indexed connector_file_id.
        """
        item_id = item.get("id", "")
        if item_id and "!" in item_id:
            return item_id
        drive_id = item.get("parentReference", {}).get("driveId")
        return f"{drive_id}!{item_id}" if drive_id else item_id

    async def handle_webhook(self, payload: dict[str, Any]) -> list[str]:
        """Handle webhook notification and return affected file IDs.

        Graph driveItem notifications never identify the changed items — the
        notification resource is the subscribed drive root — so run a delta
        query against the drive to discover what changed.
        """
        if not payload.get("value"):
            return []

        try:
            if not await self.authenticate():
                logger.error("SharePoint authentication failed during webhook handling")
                return []

            token = self.oauth.get_access_token()
            headers = {"Authorization": f"Bearer {token}"}

            site_info = self._parse_sharepoint_url()
            if site_info:
                resource = (
                    f"sites/{site_info['host_name']}:/sites/{site_info['site_name']}:/drive/root"
                )
            else:
                resource = "me/drive/root"

            # Without a stored delta link (first notification for this instance)
            # the delta query enumerates the whole drive, so only keep recently
            # modified files instead of re-syncing everything.
            first_sweep = self._delta_link is None
            url = self._delta_link or f"{self._graph_base_url}/{resource}/delta"
            cutoff = datetime.now(UTC) - timedelta(minutes=10)

            affected_files: list[str] = []
            async with httpx.AsyncClient() as client:
                while url:
                    response = await client.get(url, headers=headers, timeout=30)
                    response.raise_for_status()
                    data = response.json()

                    for item in data.get("value", []):
                        if "deleted" in item:
                            # Deleted at source: propagate the id so the processor
                            # runs its deleted-at-source cleanup
                            # (get_file_content -> 404 -> delete indexed chunks).
                            if "folder" not in item:
                                affected_files.append(self._delta_item_file_id(item))
                            continue
                        if "file" not in item:
                            continue
                        if first_sweep:
                            modified = item.get("lastModifiedDateTime")
                            if not modified:
                                continue
                            try:
                                modified_at = datetime.fromisoformat(
                                    modified.replace("Z", "+00:00")
                                )
                            except ValueError:
                                continue
                            if modified_at < cutoff:
                                continue
                        affected_files.append(self._delta_item_file_id(item))

                    delta_link = data.get("@odata.deltaLink")
                    if delta_link:
                        self._delta_link = delta_link
                    url = data.get("@odata.nextLink")

            return list(dict.fromkeys(affected_files))

        except Exception as e:
            logger.error(f"SharePoint webhook delta query failed: {e}")
            return []

    async def cleanup_subscription(self, subscription_id: str) -> bool:
        """Clean up subscription - BaseConnector interface"""
        if subscription_id == "no-webhook-configured":
            logger.info("No subscription to cleanup (webhook was not configured)")
            return True

        try:
            # Ensure authentication
            if not await self.authenticate():
                logger.error("SharePoint authentication failed during subscription cleanup")
                return False

            token = self.oauth.get_access_token()
            headers = {"Authorization": f"Bearer {token}"}

            url = f"{self._graph_base_url}/subscriptions/{subscription_id}"

            async with httpx.AsyncClient() as client:
                response = await client.delete(url, headers=headers, timeout=30)

                if response.status_code in [200, 204, 404]:
                    logger.info(
                        f"SharePoint subscription {subscription_id} cleaned up successfully"
                    )
                    return True
                else:
                    logger.warning(
                        f"Unexpected response cleaning up subscription: {response.status_code}"
                    )
                    return False

        except Exception as e:
            logger.error(f"Failed to cleanup SharePoint subscription {subscription_id}: {e}")
            return False

    async def renew_subscription(self, subscription_id: str) -> str | None:
        """Extend the Graph subscription in place (PATCH avoids re-validation).

        Returns the new expirationDateTime, or None to signal the caller to
        fall back to delete + recreate."""
        if subscription_id == "no-webhook-configured":
            return None

        try:
            if not await self.authenticate():
                logger.error("SharePoint authentication failed during subscription renewal")
                return None

            token = self.oauth.get_access_token()
            headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

            url = f"{self._graph_base_url}/subscriptions/{subscription_id}"
            body = {"expirationDateTime": self._get_subscription_expiry()}

            async with httpx.AsyncClient() as client:
                response = await client.patch(url, json=body, headers=headers, timeout=30)

                if response.status_code == 404:
                    # Subscription already expired/deleted at Graph; recreate.
                    logger.info(
                        f"SharePoint subscription {subscription_id} not found, will recreate"
                    )
                    return None
                if response.status_code not in [200, 201]:
                    logger.warning(
                        f"Unexpected response renewing SharePoint subscription: "
                        f"{response.status_code}"
                    )
                    return None

                expiration = response.json().get("expirationDateTime")
                self.webhook_expiration = expiration
                logger.info(f"SharePoint subscription {subscription_id} renewed until {expiration}")
                return expiration

        except Exception as e:
            logger.error(f"Failed to renew SharePoint subscription {subscription_id}: {e}")
            return None