src/services/documents.py

import hashlib
import importlib
import json
import unicodedata
import zipfile
from dataclasses import dataclass
from datetime import UTC, datetime
from io import BytesIO
from pathlib import Path, PurePath
from secrets import token_urlsafe
from uuid import uuid4
from xml.etree import ElementTree

from pydantic import ValidationError
from sqlalchemy import select
from sqlalchemy.orm import Session

from src.config import Settings, get_settings
from src.models.auth import AuthorityLevel, SourceDocumentStatus, SourceScope
from src.models.schemas import (
    LEGAL_SOURCE_TYPE_LABELS_VI,
    DocumentListItem,
    DocumentMetadata,
    DocumentType,
    EffectiveStatus,
    IngestionStatus,
    LegalChunk,
    LegalDocument,
    LegalParseFlag,
    LegalSourceType,
    Purpose,
    RetentionPolicy,
    ScopedDocumentItem,
    utc_now_iso,
)
from src.services.citations import CitationService
from src.services.db import DocumentChunkRecord, new_id, utc_now
from src.services.db import DocumentRecord as DbDocumentRecord
from src.services.embeddings import has_real_api_key
from src.services.legal_chunking import LegalChunker

ALLOWED_UPLOAD_EXTENSIONS = {".pdf", ".docx", ".txt"}
HTML_EXTENSION = ".html"
PDF_EXTENSION = ".pdf"
LEGAL_SOURCE_TYPE_BY_LABEL = {
    "hien phap": LegalSourceType.HIEN_PHAP,
    "bo luat": LegalSourceType.BO_LUAT,
    "luat": LegalSourceType.LUAT,
    "phap lenh": LegalSourceType.PHAP_LENH,
    "lenh": LegalSourceType.LENH,
    "nghi quyet": LegalSourceType.NGHI_QUYET,
    "nghi quyet lien tich": LegalSourceType.NGHI_QUYET_LIEN_TICH,
    "nghi dinh": LegalSourceType.NGHI_DINH,
    "quyet dinh": LegalSourceType.QUYET_DINH,
    "chi thi": LegalSourceType.CHI_THI,
    "thong tu": LegalSourceType.THONG_TU,
    "thong tu lien tich": LegalSourceType.THONG_TU_LIEN_TICH,
    "van ban hop nhat": LegalSourceType.VAN_BAN_HOP_NHAT,
    "dieu uoc quoc te": LegalSourceType.DIEU_UOC_QUOC_TE,
    "cong van": LegalSourceType.CONG_VAN,
    "cong dien": LegalSourceType.CONG_DIEN,
    "thong bao": LegalSourceType.THONG_BAO,
    "quy dinh": LegalSourceType.QUY_DINH,
    "quy che": LegalSourceType.QUY_CHE,
    "dieu le": LegalSourceType.DIEU_LE,
    "huong dan": LegalSourceType.HUONG_DAN,
    "khac": LegalSourceType.OTHER,
}


class DocumentServiceError(Exception):
    status_code = 400
    code = "validation_error"
    message = "Document request is invalid."

    def __init__(self, message: str | None = None, details: dict | None = None) -> None:
        super().__init__(message or self.message)
        self.message = message or self.message
        self.details = details or {}


class EmptyFileError(DocumentServiceError):
    status_code = 400
    code = "empty_file"
    message = "Uploaded file is empty."


class FileTooLargeError(DocumentServiceError):
    status_code = 413
    code = "file_too_large"
    message = "Uploaded file exceeds the configured size limit."


class UnsupportedFileTypeError(DocumentServiceError):
    status_code = 415
    code = "unsupported_file_type"
    message = "Uploaded file type is not supported."


class DocumentStoreUnavailableError(DocumentServiceError):
    status_code = 503
    code = "document_store_unavailable"
    message = "Document store is unavailable."


class IngestionUnavailableError(DocumentServiceError):
    status_code = 503
    code = "ingestion_unavailable"
    message = "Document ingestion is unavailable."


class MissingExtractionDependencyError(DocumentServiceError):
    status_code = 503
    code = "missing_extraction_dependency"
    message = "Required document extraction dependency is unavailable."


class EmptyExtractionError(DocumentServiceError):
    status_code = 422
    code = "empty_extraction"
    message = "Document extraction returned empty text."


class DocumentNotFoundError(DocumentServiceError):
    status_code = 404
    code = "not_found"
    message = "Document was not found."


@dataclass(frozen=True)
class DocumentRecord:
    document_id: str
    filename: str
    document_type: DocumentType
    purpose: Purpose
    retention_policy: RetentionPolicy
    ingestion_status: IngestionStatus
    metadata: DocumentMetadata
    created_at: str


@dataclass(frozen=True)
class DocumentUploadInput:
    filename: str
    content: bytes
    document_type: DocumentType
    purpose: Purpose
    retention_policy: RetentionPolicy
    title: str | None
    metadata: DocumentMetadata


@dataclass(frozen=True)
class ExtractionResult:
    text: str
    file_type: str
    method: str
    warnings: list[str]


@dataclass(frozen=True)
class DocumentIngestionResult:
    upload_record: DocumentRecord
    extraction: ExtractionResult
    document: LegalDocument
    chunks: list[LegalChunk]
    citation_previews: list[str]
    warnings: list[str]


@dataclass(frozen=True)
class DocumentListFilters:
    page: int
    page_size: int
    document_type: DocumentType | None = None
    purpose: Purpose | None = None
    ingestion_status: IngestionStatus | None = None
    effective_status: EffectiveStatus | None = None
    q: str | None = None


@dataclass(frozen=True)
class ScopedUploadContext:
    source_scope: SourceScope
    session_id: str | None = None
    owner_user_id: str | None = None
    org_id: str | None = None
    uploaded_by: str | None = None
    authority_level: AuthorityLevel = AuthorityLevel.UPLOADED_CONTEXT
    document_status: SourceDocumentStatus = SourceDocumentStatus.ACTIVE


def normalize_legal_source_label(value: str) -> str:
    normalized = unicodedata.normalize("NFKD", value)
    ascii_value = "".join(char for char in normalized if not unicodedata.combining(char))
    ascii_value = ascii_value.replace("đ", "d").replace("Đ", "D")
    return " ".join(ascii_value.casefold().strip().replace("_", " ").split())


def legal_source_type_label_vi(value: LegalSourceType) -> str:
    return LEGAL_SOURCE_TYPE_LABELS_VI[value]


def legal_source_type_from_source_value(value: str) -> LegalSourceType:
    slug_match = next((item for item in LegalSourceType if item.value == value), None)
    if slug_match is not None:
        return slug_match
    return LEGAL_SOURCE_TYPE_BY_LABEL.get(normalize_legal_source_label(value), LegalSourceType.OTHER)


def parse_upload_metadata(metadata_json: str | None) -> dict:
    if metadata_json in {None, ""}:
        return {}
    try:
        parsed = json.loads(metadata_json)
    except json.JSONDecodeError as exc:
        raise DocumentServiceError(
            message="Metadata must be a valid JSON object.",
            details={"metadata": "invalid_json"},
        ) from exc
    if not isinstance(parsed, dict):
        raise DocumentServiceError(
            message="Metadata must be a valid JSON object.",
            details={"metadata": "expected_object"},
        )
    return parsed


def ensure_upload_directories(settings: Settings) -> None:
    """Safely ensure configured upload directories exist.
    Only creates directories if upload_storage_backend is local_disk.
    Does not delete existing uploads.
    """
    if settings.upload_storage_backend != "local_disk":
        return
    for path_str in [settings.upload_dir, settings.guest_upload_dir, settings.org_upload_dir]:
        if path_str:
            Path(path_str).mkdir(parents=True, exist_ok=True)


def is_safe_upload_path(path: Path | str, settings: Settings) -> bool:
    """Check if the resolved path is inside the allowed upload directories to prevent path traversal."""
    try:
        resolved_path = Path(path).resolve()
        allowed_bases = [
            Path(settings.upload_dir).resolve(),
            Path(settings.guest_upload_dir).resolve(),
            Path(settings.org_upload_dir).resolve(),
        ]
        if settings.app_env == "test":
            import tempfile
            allowed_bases.append(Path(tempfile.gettempdir()).resolve())
        return any(
            resolved_path == base or base in resolved_path.parents
            for base in allowed_bases
            if base
        )
    except Exception:
        return False


def build_document_metadata(raw_metadata: dict, title: str | None) -> DocumentMetadata:
    metadata = dict(raw_metadata)
    if title is not None:
        metadata["title"] = title

    raw_loai_van_ban = metadata.pop("loai_van_ban", None) or metadata.get("raw_loai_van_ban")
    if raw_loai_van_ban is not None:
        metadata["raw_loai_van_ban"] = raw_loai_van_ban
        metadata["legal_source_type"] = legal_source_type_from_source_value(str(raw_loai_van_ban))

    try:
        document_metadata = DocumentMetadata.model_validate(metadata)
    except ValidationError as exc:
        raise DocumentServiceError(
            message="Document metadata validation failed.",
            details={"errors": exc.errors(include_input=False, include_url=False)},
        ) from exc

    if document_metadata.legal_source_type is not None:
        document_metadata.legal_source_type_label_vi = legal_source_type_label_vi(
            document_metadata.legal_source_type
        )
    return document_metadata


def extract_document_text_from_path(path: Path | str) -> ExtractionResult:
    source_path = Path(path)
    suffix = source_path.suffix.casefold()
    if suffix == PDF_EXTENSION:
        return _extract_pdf_text_from_path(source_path)
    if suffix == ".txt":
        return _extract_plain_text_from_path(source_path)
    if suffix == ".docx":
        return _extract_docx_text_from_path(source_path)
    raise UnsupportedFileTypeError(message=f"No extraction adapter is available for {suffix!r}.")


def extract_document_text_from_bytes(content: bytes, *, filename: str) -> ExtractionResult:
    suffix = PurePath(filename).suffix.casefold()
    if suffix == PDF_EXTENSION:
        return _extract_pdf_text_from_bytes(content)
    if suffix == ".txt":
        return _extract_plain_text_from_bytes(content, file_type="txt")
    if suffix == ".docx":
        return _extract_docx_text_from_bytes(content)
    raise UnsupportedFileTypeError(message=f"No extraction adapter is available for {suffix!r}.")


def ingest_upload_to_legal_chunks(
    *,
    upload: DocumentUploadInput,
    max_upload_bytes: int,
    ingestion_service: "InMemoryDocumentIngestionService | None" = None,
    source_uri: str | None = None,
    chunker: LegalChunker | None = None,
    citation_service: CitationService | None = None,
    include_citation_previews: bool = True,
    fail_on_empty: bool = False,
) -> DocumentIngestionResult:
    service = ingestion_service or InMemoryDocumentIngestionService()
    upload_record = service.upload_document(upload, max_upload_bytes=max_upload_bytes)
    extraction = extract_document_text_from_bytes(upload.content, filename=upload.filename)
    text = extraction.text.strip()

    document = _legal_document_from_upload(
        upload_record=upload_record,
        upload=upload,
        extraction=extraction,
        source_uri=source_uri or upload_record.filename,
        ready=bool(text),
    )

    if not text:
        warnings = [*extraction.warnings, "Skipped chunking because extraction returned empty text."]
        if fail_on_empty:
            raise EmptyExtractionError(details={"warnings": warnings})
        return DocumentIngestionResult(
            upload_record=upload_record,
            extraction=extraction,
            document=document,
            chunks=[],
            citation_previews=[],
            warnings=warnings,
        )

    chunks = (chunker or LegalChunker()).chunk_text(
        document_id=document.document_id,
        text=extraction.text,
    )
    previews = []
    if include_citation_previews:
        service_for_citations = citation_service or CitationService()
        previews = [
            service_for_citations.build_label(document=document, chunk=chunk) for chunk in chunks
        ]
    return DocumentIngestionResult(
        upload_record=upload_record,
        extraction=extraction,
        document=document,
        chunks=chunks,
        citation_previews=previews,
        warnings=list(extraction.warnings),
    )


def _legal_document_from_upload(
    *,
    upload_record: DocumentRecord,
    upload: DocumentUploadInput,
    extraction: ExtractionResult,
    source_uri: str,
    ready: bool,
) -> LegalDocument:
    metadata = upload.metadata
    parse_flags = []
    if not any(
        [
            metadata.document_number,
            metadata.legal_source_type,
            metadata.authority,
            metadata.effective_date,
        ]
    ):
        parse_flags.append(LegalParseFlag.METADATA_MISSING)

    return LegalDocument(
        document_id=upload_record.document_id,
        title=metadata.title or upload.title or upload_record.filename,
        document_type=upload_record.document_type,
        source_uri=source_uri,
        ingestion_status=IngestionStatus.READY if ready else IngestionStatus.FAILED,
        ingested_at=datetime.now(UTC),
        effective_status=metadata.effective_status or EffectiveStatus.UNKNOWN,
        document_number=metadata.document_number,
        legal_source_type=metadata.legal_source_type,
        authority=metadata.authority,
        metadata={
            "filename": upload_record.filename,
            "file_size_bytes": len(upload.content),
            "file_type": extraction.file_type,
            "extraction_method": extraction.method,
            "upload_ingestion_status": upload_record.ingestion_status.value,
            "metadata_source": "provided-or-metadata-missing",
        },
        parse_flags=parse_flags,
    )


def _extract_plain_text_from_path(path: Path) -> ExtractionResult:
    try:
        text = path.read_text(encoding="utf-8-sig")
        method = "Path.read_text(utf-8-sig)"
    except UnicodeDecodeError:
        text = path.read_text(encoding="utf-8")
        method = "Path.read_text(utf-8)"
    return ExtractionResult(
        text=_normalize_text_newlines(text),
        file_type=path.suffix.casefold().lstrip("."),
        method=method,
        warnings=[] if text.strip() else ["Text extraction returned empty text."],
    )


def _extract_plain_text_from_bytes(content: bytes, *, file_type: str) -> ExtractionResult:
    try:
        text = content.decode("utf-8-sig")
        method = "bytes.decode(utf-8-sig)"
    except UnicodeDecodeError:
        text = content.decode("utf-8")
        method = "bytes.decode(utf-8)"
    return ExtractionResult(
        text=_normalize_text_newlines(text),
        file_type=file_type,
        method=method,
        warnings=[] if text.strip() else ["Text extraction returned empty text."],
    )


def _extract_pdf_text_from_path(path: Path) -> ExtractionResult:
    data = path.read_bytes()
    return _extract_pdf_text(data, source=str(path))


def _extract_pdf_text_from_bytes(content: bytes) -> ExtractionResult:
    return _extract_pdf_text(content, source=None)


def _extract_pdf_text(content: bytes, *, source: str | None) -> ExtractionResult:
    pypdf = _optional_import("pypdf")
    if pypdf is not None:
        reader_input = source if source is not None else BytesIO(content)
        reader = pypdf.PdfReader(reader_input)
        pages = [(page.extract_text() or "") for page in reader.pages]
        text = "\n\n".join(page.strip() for page in pages if page.strip())
        return ExtractionResult(
            text=text,
            file_type="pdf",
            method="pypdf.PdfReader",
            warnings=[] if text else ["PDF extraction returned empty text."],
        )

    pypdf2 = _optional_import("PyPDF2")
    if pypdf2 is not None:
        reader_input = source if source is not None else BytesIO(content)
        reader = pypdf2.PdfReader(reader_input)
        pages = [(page.extract_text() or "") for page in reader.pages]
        text = "\n\n".join(page.strip() for page in pages if page.strip())
        return ExtractionResult(
            text=text,
            file_type="pdf",
            method="PyPDF2.PdfReader",
            warnings=[] if text else ["PDF extraction returned empty text."],
        )

    fitz = _optional_import("fitz")
    if fitz is None:
        raise MissingExtractionDependencyError(
            message=(
                "PDF extraction dependency is missing. Install one parser, for example: "
                "python -m pip install pypdf"
            )
        )

    if source is not None:
        document = fitz.open(source)
    else:
        document = fitz.open(stream=content, filetype="pdf")
    with document:
        pages = [page.get_text("text") or "" for page in document]
    text = "\n\n".join(page.strip() for page in pages if page.strip())
    return ExtractionResult(
        text=text,
        file_type="pdf",
        method="PyMuPDF(fitz)",
        warnings=[] if text else ["PDF extraction returned empty text."],
    )


def _extract_docx_text_from_path(path: Path) -> ExtractionResult:
    data = path.read_bytes()
    docx = _optional_import("docx")
    if docx is not None:
        document = docx.Document(str(path))
        return _python_docx_result(document)
    return _extract_docx_text_with_stdlib(data)


def _extract_docx_text_from_bytes(content: bytes) -> ExtractionResult:
    docx = _optional_import("docx")
    if docx is not None:
        document = docx.Document(BytesIO(content))
        return _python_docx_result(document)
    return _extract_docx_text_with_stdlib(content)


def _python_docx_result(document) -> ExtractionResult:
    text = "\n".join(paragraph.text for paragraph in document.paragraphs).strip()
    return ExtractionResult(
        text=text,
        file_type="docx",
        method="python-docx",
        warnings=[] if text else ["DOCX extraction returned empty text."],
    )


def _extract_docx_text_with_stdlib(content: bytes) -> ExtractionResult:
    try:
        with zipfile.ZipFile(BytesIO(content)) as archive:
            document_xml = archive.read("word/document.xml")
    except (KeyError, zipfile.BadZipFile) as exc:
        raise MissingExtractionDependencyError(
            message=(
                "DOCX extraction failed because the file is not a readable DOCX package. "
                "Install python-docx for broader DOCX compatibility: "
                "python -m pip install python-docx"
            )
        ) from exc

    root = ElementTree.fromstring(document_xml)
    namespace = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
    paragraphs = []
    for paragraph in root.findall(".//w:p", namespace):
        pieces = [
            node.text or ""
            for node in paragraph.findall(".//w:t", namespace)
        ]
        text = "".join(pieces).strip()
        if text:
            paragraphs.append(text)
    text = _normalize_text_newlines("\n".join(paragraphs))
    return ExtractionResult(
        text=text,
        file_type="docx",
        method="stdlib zipfile+xml",
        warnings=[] if text else ["DOCX extraction returned empty text."],
    )


def _optional_import(module_name: str):
    try:
        return importlib.import_module(module_name)
    except ImportError:
        return None


def _normalize_text_newlines(text: str) -> str:
    return text.replace("\r\n", "\n").replace("\r", "\n")


class InMemoryDocumentIngestionService:
    def __init__(self) -> None:
        self._documents: list[DocumentRecord] = []

    def reset(self) -> None:
        self._documents.clear()

    def upload_document(
        self,
        upload: DocumentUploadInput,
        max_upload_bytes: int,
    ) -> DocumentRecord:
        self._validate_upload(upload, max_upload_bytes)
        record = DocumentRecord(
            document_id=f"doc_{uuid4().hex}",
            filename=PurePath(upload.filename).name,
            document_type=upload.document_type,
            purpose=upload.purpose,
            retention_policy=upload.retention_policy,
            ingestion_status=IngestionStatus.QUEUED,
            metadata=upload.metadata,
            created_at=utc_now_iso(),
        )
        self._documents.append(record)
        return record

    def list_documents(self, filters: DocumentListFilters) -> tuple[list[DocumentListItem], int]:
        filtered = list(self._documents)
        if filters.document_type is not None:
            filtered = [item for item in filtered if item.document_type == filters.document_type]
        if filters.purpose is not None:
            filtered = [item for item in filtered if item.purpose == filters.purpose]
        if filters.ingestion_status is not None:
            filtered = [
                item for item in filtered if item.ingestion_status == filters.ingestion_status
            ]
        if filters.effective_status is not None:
            filtered = [
                item
                for item in filtered
                if item.metadata.effective_status == filters.effective_status
            ]
        if filters.q:
            query = filters.q.casefold()
            filtered = [
                item
                for item in filtered
                if query in (item.metadata.title or "").casefold()
                or query in item.filename.casefold()
                or query in (item.metadata.document_number or "").casefold()
            ]

        total = len(filtered)
        start = (filters.page - 1) * filters.page_size
        end = start + filters.page_size
        return [self._to_list_item(item) for item in filtered[start:end]], total

    def _validate_upload(self, upload: DocumentUploadInput, max_upload_bytes: int) -> None:
        filename = PurePath(upload.filename).name
        extension = PurePath(filename).suffix.casefold()
        if not upload.content:
            raise EmptyFileError()
        if len(upload.content) > max_upload_bytes:
            raise FileTooLargeError(details={"max_upload_bytes": max_upload_bytes})
        if extension in ALLOWED_UPLOAD_EXTENSIONS:
            return
        if extension == HTML_EXTENSION:
            approved_import = bool(upload.metadata.model_extra.get("approved_corpus_import"))
            if upload.purpose == Purpose.CORPUS and approved_import:
                return
        raise UnsupportedFileTypeError()

    @staticmethod
    def _to_list_item(record: DocumentRecord) -> DocumentListItem:
        return DocumentListItem(
            document_id=record.document_id,
            title=record.metadata.title or record.filename,
            filename=record.filename,
            document_type=record.document_type,
            purpose=record.purpose,
            retention_policy=record.retention_policy,
            ingestion_status=record.ingestion_status,
            effective_status=record.metadata.effective_status,
            created_at=record.created_at,
        )


document_ingestion_service = InMemoryDocumentIngestionService()


PRIVATE_COLLECTION_BY_SCOPE = {
    SourceScope.GUEST_SESSION: "guest_sessions",
    SourceScope.ORG_PRIVATE: "org_documents",
}


def _private_collection_for_scope(scope: SourceScope) -> str:
    try:
        if scope == SourceScope.ORG_PRIVATE:
            from src.config import get_settings
            return get_settings().private_chroma_collection_name
        return PRIVATE_COLLECTION_BY_SCOPE[scope]
    except KeyError as exc:
        raise DocumentServiceError(
            "Unsupported private document scope. Personal/user_private libraries are disabled.",
            details={"source_scope": scope.value},
        ) from exc


class DocumentLibraryService:
    def __init__(
        self,
        db: Session,
        settings: Settings | None = None,
    ) -> None:
        self.db = db
        self.settings = settings or get_settings()

    def ensure_guest_session_id(self, session_id: str | None) -> str:
        cleaned = str(session_id or "").strip()
        if cleaned:
            return cleaned
        return f"guest_{token_urlsafe(32)}"

    def upload_scoped_document(
        self,
        upload: DocumentUploadInput,
        *,
        context: ScopedUploadContext,
        mime_type: str = "",
    ) -> tuple[DbDocumentRecord, int, list[str]]:
        if not self.settings.enable_private_document_library:
            raise DocumentStoreUnavailableError("Private document libraries are disabled.")

        storage_path = self._write_upload_file(upload, context)
        source_uri = f"private://documents/{context.source_scope.value}/{Path(storage_path).name}"
        ingestion = ingest_upload_to_legal_chunks(
            upload=upload,
            max_upload_bytes=self.settings.max_upload_bytes,
            source_uri=source_uri,
            fail_on_empty=True,
        )
        self._apply_private_metadata(
            ingestion.document,
            ingestion.chunks,
            context=context,
            filename=ingestion.upload_record.filename,
            mime_type=mime_type,
        )
        collection_name = _private_collection_for_scope(context.source_scope)

        # Ingestion Flow: Save metadata + chunk_text to Postgres first
        record = DbDocumentRecord(
            id=ingestion.document.document_id,
            source_scope=context.source_scope.value,
            owner_user_id=context.owner_user_id,
            org_id=context.org_id,
            session_id=context.session_id,
            uploaded_by=context.uploaded_by,
            title=ingestion.document.title,
            filename=ingestion.upload_record.filename,
            mime_type=mime_type,
            storage_path=storage_path,
            source_uri=source_uri,
            authority_level=context.authority_level.value,
            document_status=SourceDocumentStatus.PENDING_INDEX.value,
            document_type=upload.document_type.value,
            purpose=upload.purpose.value,
            retention_policy=upload.retention_policy.value,
            effective_date=upload.metadata.effective_date,
        )
        self.db.add(record)
        for chunk in ingestion.chunks:
            self.db.add(
                DocumentChunkRecord(
                    id=new_id("dchk"),
                    document_id=record.id,
                    chunk_id=chunk.chunk_id,
                    chunk_index=chunk.chunk_index,
                    chroma_collection=collection_name,
                    chroma_id=chunk.chunk_id,
                    token_count=chunk.token_count,
                    chunk_text=chunk.content,
                    metadata_json=chunk.metadata,
                    citation_label=None,
                    citation_debug=None,
                    content_hash=hashlib.sha256(chunk.content.encode("utf-8")).hexdigest(),
                )
            )
        self.db.commit()
        self.db.refresh(record)

        # Call retrieval service /index
        from src.services.retrieval_service_client import RetrievalServiceClient
        chunks_payload = [
            {
                "chunk_id": chunk.chunk_id,
                "chunk_index": chunk.chunk_index,
                "text": chunk.content,
                "metadata": chunk.metadata,
            }
            for chunk in ingestion.chunks
        ]
        client = RetrievalServiceClient()
        success = client.index_chunks(
            document_id=record.id,
            organization_id=context.org_id or "",
            collection_name=collection_name,
            chunks=chunks_payload,
        )

        if success:
            record.document_status = SourceDocumentStatus.ACTIVE.value
            for chunk_rec in record.chunks:
                meta = dict(chunk_rec.metadata_json) if chunk_rec.metadata_json else {}
                meta["document_status"] = SourceDocumentStatus.ACTIVE.value
                chunk_rec.metadata_json = meta
        else:
            record.document_status = SourceDocumentStatus.INDEXING_FAILED.value
            for chunk_rec in record.chunks:
                meta = dict(chunk_rec.metadata_json) if chunk_rec.metadata_json else {}
                meta["document_status"] = SourceDocumentStatus.INDEXING_FAILED.value
                chunk_rec.metadata_json = meta

        self.db.add(record)
        self.db.commit()
        self.db.refresh(record)

        return record, len(ingestion.chunks), ingestion.warnings

    def list_documents(self, context: ScopedUploadContext) -> list[DbDocumentRecord]:
        statement = self._scoped_statement(context).order_by(DbDocumentRecord.created_at.desc())
        return list(self.db.scalars(statement).all())

    def get_document(self, document_id: str, context: ScopedUploadContext) -> DbDocumentRecord:
        record = self.db.scalar(
            self._scoped_statement(context).where(DbDocumentRecord.id == document_id)
        )
        if record is None:
            raise DocumentNotFoundError()
        return record

    def delete_document(self, document_id: str, context: ScopedUploadContext) -> DbDocumentRecord:
        record = self.get_document(document_id, context)
        record.deleted_at = utc_now()
        record.document_status = SourceDocumentStatus.OUTDATED.value
        self.db.add(record)
        self._delete_vectors(record)

        # Best-effort physical file deletion
        if record.storage_path:
            try:
                if is_safe_upload_path(record.storage_path, self.settings):
                    file_path = Path(record.storage_path).resolve()
                    if file_path.is_file():
                        file_path.unlink()
                else:
                    import logging
                    logging.getLogger(__name__).warning(
                        "Skipped physical file deletion for document %s: path %s is outside allowed upload directories.",
                        record.id,
                        record.storage_path,
                    )
            except Exception as exc:
                import logging
                logging.getLogger(__name__).warning(
                    "Best-effort physical file deletion failed for document %s (path=%s): %s",
                    record.id,
                    record.storage_path,
                    exc,
                )

        self.db.commit()
        self.db.refresh(record)
        return record

    @staticmethod
    def to_item(record: DbDocumentRecord) -> ScopedDocumentItem:
        return ScopedDocumentItem(
            document_id=record.id,
            source_scope=SourceScope(record.source_scope),
            title=record.title,
            filename=record.filename,
            mime_type=record.mime_type,
            authority_level=AuthorityLevel(record.authority_level),
            document_status=SourceDocumentStatus(record.document_status),
            document_type=DocumentType(record.document_type),
            purpose=Purpose(record.purpose),
            retention_policy=RetentionPolicy(record.retention_policy),
            created_at=record.created_at.isoformat(),
            updated_at=record.updated_at.isoformat(),
            effective_date=record.effective_date,
        )

    def _scoped_statement(self, context: ScopedUploadContext):
        statement = select(DbDocumentRecord).where(
            DbDocumentRecord.source_scope == context.source_scope.value,
            DbDocumentRecord.deleted_at.is_(None),
        )
        if context.source_scope == SourceScope.GUEST_SESSION:
            statement = statement.where(DbDocumentRecord.session_id == context.session_id)
        elif context.source_scope == SourceScope.USER_PRIVATE:
            statement = statement.where(DbDocumentRecord.owner_user_id == context.owner_user_id)
        elif context.source_scope == SourceScope.ORG_PRIVATE:
            statement = statement.where(DbDocumentRecord.org_id == context.org_id)
        return statement

    def _write_upload_file(self, upload: DocumentUploadInput, context: ScopedUploadContext) -> str:
        directory = self._upload_directory(context)
        directory.mkdir(parents=True, exist_ok=True)
        safe_name = PurePath(upload.filename).name or "upload.bin"
        path = directory / f"{uuid4().hex}_{safe_name}"
        path.write_bytes(upload.content)
        return str(path)

    def _upload_directory(self, context: ScopedUploadContext) -> Path:
        if context.source_scope == SourceScope.GUEST_SESSION:
            return Path(self.settings.guest_upload_dir) / str(context.session_id)
        if context.source_scope == SourceScope.ORG_PRIVATE:
            return Path(self.settings.org_upload_dir) / str(context.org_id)
        raise DocumentServiceError("Unsupported document scope.")

    def _apply_private_metadata(
        self,
        document: LegalDocument,
        chunks: list[LegalChunk],
        *,
        context: ScopedUploadContext,
        filename: str,
        mime_type: str,
    ) -> None:
        org_id = context.org_id or ""
        document_id = document.document_id
        source_uri = document.source_uri or f"org://{org_id}/documents/{document_id}"

        metadata = {
            "source_scope": context.source_scope.value,
            "owner_user_id": context.owner_user_id or "",
            "org_id": org_id,
            "organization_id": org_id,
            "session_id": context.session_id or "",
            "uploaded_by": context.uploaded_by or "",
            "filename": filename,
            "mime_type": mime_type,
            "authority_level": context.authority_level.value,
            "document_status": context.document_status.value,
            "document_id": document_id,
            "source_uri": source_uri,
        }
        document.metadata.update(metadata)
        if not document.source_uri:
            document.source_uri = source_uri

        for chunk in chunks:
            chunk.metadata.update(metadata)
            chunk.metadata["chunk_id"] = chunk.chunk_id
            chunk.metadata["chunk_index"] = chunk.chunk_index

    def _index_private_chunks(
        self,
        *,
        document: LegalDocument,
        chunks: list[LegalChunk],
        collection_name: str,
    ) -> None:
        from src.services.corpus_indexing import ChromaIndexAdapter, build_indexable_records

        explicit_provider = self.settings.embedding_provider
        allow_fake_embeddings = (
            explicit_provider == "auto"
            and not (
                has_real_api_key(self.settings.openai_api_key)
                or has_real_api_key(self.settings.google_api_key)
            )
        )

        adapter = ChromaIndexAdapter(
            persist_dir=self.settings.private_chroma_persist_dir,
            collection_name=collection_name,
            embedding_provider=explicit_provider,
            allow_fake_embeddings=allow_fake_embeddings,
        )
        try:
            adapter.upsert(build_indexable_records(document, chunks))
        except Exception as exc:
            raise DocumentStoreUnavailableError(
                "Could not index uploaded document in Chroma. "
                "Check EMBEDDING_PROVIDER and rebuild the private Chroma collection if needed.",
                details={"collection_name": collection_name, "reason": str(exc)},
            ) from exc

    def _delete_vectors(self, record: DbDocumentRecord) -> None:
        collection_name = _private_collection_for_scope(SourceScope(record.source_scope))
        try:
            from src.services.retrieval_service_client import RetrievalServiceClient
            client = RetrievalServiceClient()
            client.delete_document(record.id, collection_name)
        except Exception:
            return