async def ingest_chunks_ingress(
        self,
        document_id: UUID,
        metadata: Optional[dict],
        chunks: list[RawChunk],
        user: User,
        *args: Any,
        **kwargs: Any,
    ) -> DocumentResponse:
        """Directly ingest user-provided text chunks (rather than from a
        file)."""
        if not chunks:
            raise R2RException(
                status_code=400, message="No chunks provided for ingestion."
            )
        metadata = metadata or {}
        version = STARTING_VERSION

        document_info = self._create_document_info_from_chunks(
            document_id,
            user,
            chunks,
            metadata,
            version,
        )

        existing_document_info = (
            await self.providers.database.documents_handler.get_documents_overview(
                offset=0,
                limit=100,
                filter_user_ids=[user.id],
                filter_document_ids=[document_id],
            )
        )["results"]
        if len(existing_document_info) > 0:
            existing_doc = existing_document_info[0]
            if existing_doc.ingestion_status != IngestionStatus.FAILED:
                raise R2RException(
                    status_code=409,
                    message=(
                        f"Document {document_id} was already ingested "
                        "and is not in a failed state."
                    ),
                )

        await self.providers.database.documents_handler.upsert_documents_overview(
            document_info
        )
        return document_info
Directly ingest user-provided text chunks (rather than from a
        file).
py/core/main/services/ingestion_service.py