smart-org/apps/api/app/application/services/knowledge_service.py

"""Knowledge Base use cases: ingest text, chunk it, and retrieve it safely.

Two things matter here:

* **Chunking is deterministic.** The same text always produces the same chunks,
  which makes ingestion idempotent-ish and the tests meaningful.
* **Retrieval is permission-aware.** Search is always scoped to one organization,
  and optionally narrowed to one project. An org-wide source is visible to every
  project; a project-scoped source is not visible from another project.

Embeddings are computed through ``VectorStorePort`` when a real adapter is
configured. With the mock adapter no vectors are produced and retrieval falls
back to keyword matching — the same fallback the plan calls for at MVP.
"""

from __future__ import annotations

from app.application.ports import VectorStorePort
from app.core.errors import NotFoundError, ValidationError
from app.domain.knowledge.models import (
    Document,
    DocumentChunk,
    EmbeddingJob,
    EmbeddingJobStatus,
    KnowledgeSource,
    RetrievalPolicy,
    SourceKind,
)
from app.infrastructure.repositories.knowledge import (
    DocumentChunkRepository,
    DocumentRepository,
    EmbeddingJobRepository,
    KnowledgeSourceRepository,
    RetrievalPolicyRepository,
)

#: Chunk size in words. Small enough to keep a retrieved passage focused.
CHUNK_WORDS = 180
CHUNK_OVERLAP_WORDS = 30
MAX_INGEST_CHARS = 400_000


def chunk_text(
    text: str, *, size: int = CHUNK_WORDS, overlap: int = CHUNK_OVERLAP_WORDS
) -> list[str]:
    """Split text into overlapping word windows.

    Pure and deterministic — no regex surprises, no dependence on punctuation.
    """
    words = text.split()
    if not words:
        return []
    if size <= 0:
        raise ValueError("chunk size must be positive")
    step = max(1, size - max(0, overlap))

    chunks: list[str] = []
    for start in range(0, len(words), step):
        window = words[start : start + size]
        if not window:
            break
        chunks.append(" ".join(window))
        if start + size >= len(words):
            break
    return chunks


class KnowledgeService:
    def __init__(
        self,
        *,
        sources: KnowledgeSourceRepository,
        documents: DocumentRepository,
        chunks: DocumentChunkRepository,
        jobs: EmbeddingJobRepository,
        policies: RetrievalPolicyRepository,
        vector_store: VectorStorePort,
    ) -> None:
        self.sources = sources
        self.documents = documents
        self.chunks = chunks
        self.jobs = jobs
        self.policies = policies
        self.vector_store = vector_store

    # -- ingestion -----------------------------------------------------------

    async def create_text_source(
        self,
        *,
        organization_id: str,
        name: str,
        content: str,
        project_id: str | None = None,
        title: str = "",
    ) -> tuple[KnowledgeSource, int]:
        if not name.strip():
            raise ValidationError("نام منبع دانش نمی‌تواند خالی باشد.")
        if not content.strip():
            raise ValidationError("محتوای منبع دانش نمی‌تواند خالی باشد.")
        if len(content) > MAX_INGEST_CHARS:
            raise ValidationError("حجم محتوا بیش از حد مجاز است.")

        source = await self.sources.create(
            KnowledgeSource(
                organization_id=organization_id,
                name=name.strip(),
                kind=SourceKind.TEXT,
                project_id=project_id,
            )
        )
        document = await self.documents.create(
            Document(source_id=source.id, title=title.strip() or source.name, content=content)
        )

        pieces = chunk_text(content)
        chunk_rows = [
            DocumentChunk(document_id=document.id, index=index, content=piece)
            for index, piece in enumerate(pieces)
        ]
        created = await self.chunks.create_many(chunk_rows)

        job = await self.jobs.create(EmbeddingJob(source_id=source.id))
        try:
            # The mock vector store is a no-op that accepts and discards; a real
            # adapter embeds here. Keyword search below works either way.
            await self.vector_store.upsert(
                organization_id=organization_id,
                chunks=[(chunk.id, chunk.content, []) for chunk in chunk_rows],
                metadata={"source_id": source.id, "project_id": project_id},
            )
            await self.jobs.set_status(job.id, EmbeddingJobStatus.DONE)
        except Exception as exc:  # noqa: BLE001 - embedding is best-effort at MVP
            # Ingestion still succeeded; only the vector index is degraded, and
            # keyword search keeps working. Record why rather than failing hard.
            await self.jobs.set_status(job.id, EmbeddingJobStatus.FAILED, error=str(exc))

        return source, created

    # -- reads ---------------------------------------------------------------

    async def list_sources(
        self, organization_id: str, *, limit: int = 100, offset: int = 0
    ) -> list[KnowledgeSource]:
        return await self.sources.list_for_org(organization_id, limit=limit, offset=offset)

    async def get_source_or_404(
        self, *, organization_id: str, source_id: str
    ) -> KnowledgeSource:
        source = await self.sources.get_for_org(organization_id, source_id)
        if source is None:
            raise NotFoundError("منبع دانش یافت نشد.")
        return source

    async def source_detail(self, *, source_id: str) -> dict:
        documents = await self.documents.list_for_source(source_id)
        return {
            "documents": documents,
            "chunk_count": await self.chunks.count_for_source(source_id),
            "job": await self.jobs.latest_for_source(source_id),
        }

    async def chunk_counts(self, source_ids: list[str]) -> dict[str, int]:
        """Chunk counts for many sources at once — see ``count_for_sources``."""
        return await self.chunks.count_for_sources(source_ids)

    async def search(
        self,
        *,
        organization_id: str,
        query: str,
        project_id: str | None = None,
        limit: int = 5,
    ) -> list[dict]:
        return await self.chunks.keyword_search(
            organization_id=organization_id,
            query=query,
            project_id=project_id,
            limit=limit,
        )

    async def delete_source(self, *, organization_id: str, source_id: str) -> None:
        source = await self.get_source_or_404(
            organization_id=organization_id, source_id=source_id
        )
        await self.sources.delete(source.id)

    # -- retrieval policies --------------------------------------------------

    async def list_policies(self, organization_id: str) -> list[RetrievalPolicy]:
        return await self.policies.list_for_org(organization_id)

    async def create_policy(
        self, *, organization_id: str, name: str, scope: dict | None = None
    ) -> RetrievalPolicy:
        if not name.strip():
            raise ValidationError("نام سیاست بازیابی نمی‌تواند خالی باشد.")
        return await self.policies.create(
            RetrievalPolicy(
                organization_id=organization_id, name=name.strip(), scope_json=dict(scope or {})
            )
        )