Architecture Overview

┌─────────────────────────────────────────────────────────────────┐
│ QDrant Loader │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ CLI │ │ MCP Server │ │ Config │ │
│ │ Interface │ │ (Separate) │ │ Manager │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │ │ │
│ └─────────────────┼─────────────────┘ │
│ │ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Async Ingestion Pipeline │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Data │ │ File │ │ Content │ │ │
│ │ │ Connectors │ │ Converters │ │ Processors │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ │ │ │ │ │ │
│ │ └─────────────────┼─────────────────┘ │ │
│ │ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Embedding │ │ State │ │ QDrant │ │ │
│ │ │ Service │ │ Manager │ │ Manager │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
└───────────────────────────┼────────────────────────────────────┘ │
┌───────────────────────────┼────────────────────────────────────┐
│ External Services │
├───────────────────────────┼────────────────────────────────────┤
│ │ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ QDrant │ │ OpenAI │ │ Data │ │
│ │ Database │ │ API │ │ Sources │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘

from abc import ABC, abstractmethod
from qdrant_loader.config.source_config import SourceConfig
from qdrant_loader.core.document import Document
from qdrant_loader.core.file_conversion import FileConversionConfig

class BaseConnector(ABC):
    def __init__(self, config: SourceConfig): ...
    async def __aenter__(self): ...
    async def __aexit__(self, exc_type, exc_val, exc_tb): ...
    def set_file_conversion_config(self, cfg: FileConversionConfig) -> None: ...
    @abstractmethod
    async def get_documents(self) -> list[Document]: ...

┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Data │───▶│ File │───▶│ Content │───▶│ Embedding │
│ Connector │ │ Converter │ │ Processor │ │ Service │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ▼ ▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Raw Data │ │ Text │ │ Chunks │ │ Vectors │
│ + Metadata │ │ + Metadata │ │ + Metadata │ │ + Metadata │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ ▼ ┌─────────────┐ │ QDrant │ │ Manager │ └─────────────┘ │ ▼ ┌─────────────┐ │ QDrant │ │ Database │ └─────────────┘

┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Query │───▶│ Embedding │───▶│ QDrant │───▶│ Results │
│ (Text) │ │ Service │ │ Search │ │ + Metadata │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ▼ ▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ User Query │ │ Query Vector│ │ Similarity │ │ Ranked │
│ │ │ │ │ Scores │ │ Results │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘

# Actual connector instantiation in PipelineOrchestrator
async def _collect_documents_from_sources(
    self, filtered_config: SourcesConfig, project_id: str | None = None
) -> list[Document]:
    """Collect documents from all configured sources."""
    documents = []

    # Process each source type with direct connector instantiation
    if filtered_config.confluence:
        confluence_docs = await self.components.source_processor.process_source_type(
            filtered_config.confluence, ConfluenceConnector, "Confluence"
        )
        documents.extend(confluence_docs)

    if filtered_config.git:
        git_docs = await self.components.source_processor.process_source_type(
            filtered_config.git, GitConnector, "Git"
        )
        documents.extend(git_docs)

    if filtered_config.jira:
        jira_docs = await self.components.source_processor.process_source_type(
            filtered_config.jira, JiraConnector, "Jira"
        )
        documents.extend(jira_docs)

    if filtered_config.publicdocs:
        publicdocs_docs = await self.components.source_processor.process_source_type(
            filtered_config.publicdocs, PublicDocsConnector, "PublicDocs"
        )
        documents.extend(publicdocs_docs)

    if filtered_config.localfile:
        localfile_docs = await self.components.source_processor.process_source_type(
            filtered_config.localfile, LocalFileConnector, "LocalFile"
        )
        documents.extend(localfile_docs)

    return documents

class StateManager:
    """Manages state for document ingestion."""

    def __init__(self, config: StateManagementConfig):
        self.config = config
        self._engine = None
        self._session_factory = None

    async def initialize(self):
        """Initialize the database schema and connection."""
        db_url = self.config.database_path
        if not db_url.startswith("sqlite:///"):
            db_url = f"sqlite:///{db_url}"

        self._engine = create_async_engine(f"sqlite+aiosqlite:///{db_file}")
        self._session_factory = async_sessionmaker(bind=self._engine)

        # Initialize schema
        async with self._engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

async def update_document_state(
    self, document: Document, project_id: str | None = None
) -> DocumentStateRecord:
    """Update document state for change detection."""
    content_hash = hashlib.sha256(document.content.encode()).hexdigest()

    # Check if document exists and has changed
    existing = await self.get_document_state_record(
        document.source_type, document.source, document.id, project_id
    )

    if existing and existing.content_hash == content_hash:
        # No changes detected
        return existing

    # Update or create new state record
    # ... implementation details

class AsyncIngestionPipeline:
    """Main async ingestion pipeline."""

    async def process_documents(
        self,
        project_id: str | None = None,
        source_type: str | None = None,
        source: str | None = None,
    ) -> None:
        """Process documents asynchronously."""
        # Async document collection and processing
        async with self.state_manager:
            documents = await self.orchestrator.process_documents(
                project_id=project_id,
                source_type=source_type,
                source=source,
            )

class QdrantManager:
    """Manages QDrant operations with batching."""

    async def upsert_points(self, points: list[dict]) -> None:
        """Upsert points in batches."""
        batch_size = self.batch_size  # Configurable batch size

        for i in range(0, len(points), batch_size):
            batch = points[i:i + batch_size]
            await self._upsert_batch(batch)

class ConfluenceConnector(BaseConnector):
    """Confluence connector with authentication."""

    def _setup_authentication(self):
        """Set up authentication based on deployment type."""
        if self.config.deployment_type == ConfluenceDeploymentType.CLOUD:
            self.session.auth = HTTPBasicAuth(self.config.email, self.config.token)
        else:
            self.session.headers.update({"Authorization": f"Bearer {self.config.token}"})