async def create_document(
            file: Optional[UploadFile] = File(
                None,
                description="The file to ingest. Exactly one of file, raw_text, or chunks must be provided.",
            ),
            raw_text: Optional[str] = Form(
                None,
                description="Raw text content to ingest. Exactly one of file, raw_text, or chunks must be provided.",
            ),
            chunks: Optional[Json[list[str]]] = Form(
                None,
                description="Pre-processed text chunks to ingest. Exactly one of file, raw_text, or chunks must be provided.",
            ),
            id: Optional[UUID] = Form(
                None,
                description="The ID of the document. If not provided, a new ID will be generated.",
            ),
            collection_ids: Optional[Json[list[UUID]]] = Form(
                None,
                description="Collection IDs to associate with the document. If none are provided, the document will be assigned to the user's default collection.",
            ),
            metadata: Optional[Json[dict]] = Form(
                None,
                description="Metadata to associate with the document, such as title, description, or custom fields.",
            ),
            ingestion_mode: IngestionMode = Form(
                default=IngestionMode.custom,
                description=(
                    "Ingestion modes:\n"
                    "- `hi-res`: Thorough ingestion with full summaries and enrichment.\n"
                    "- `ocr`: OCR via Mistral and full summaries.\n"
                    "- `fast`: Quick ingestion with minimal enrichment and no summaries.\n"
                    "- `custom`: Full control via `ingestion_config`.\n\n"
                    "If `filters` or `limit` (in `ingestion_config`) are provided alongside `hi-res` or `fast`, "
                    "they will override the default settings for that mode."
                ),
            ),
            ingestion_config: Optional[Json[IngestionConfig]] = Form(
                None,
                description="An optional dictionary to override the default chunking configuration for the ingestion process. If not provided, the system will use the default server-side chunking configuration.",
            ),
            run_with_orchestration: Optional[bool] = Form(
                True,
                description="Whether or not ingestion runs with orchestration, default is `True`. When set to `False`, the ingestion process will run synchronous and directly return the result.",
            ),
            auth_user=Depends(self.providers.auth.auth_wrapper()),
        ) -> WrappedIngestionResponse:
            """
            Creates a new Document object from an input file, text content, or chunks. The chosen `ingestion_mode` determines
            how the ingestion process is configured:

            **Ingestion Modes:**
            - `hi-res`: Comprehensive parsing and enrichment, including summaries and possibly more thorough parsing.
            - `fast`: Speed-focused ingestion that skips certain enrichment steps like summaries.
            - `custom`: Provide a full `ingestion_config` to customize the entire ingestion process.

            Either a file or text content must be provided, but not both. Documents are shared through `Collections` which allow for tightly specified cross-user interactions.

            The ingestion process runs asynchronously and its progress can be tracked using the returned
            task_id.
            """
            if not auth_user.is_superuser:
                user_document_count = (
                    await self.services.management.documents_overview(
                        user_ids=[auth_user.id],
                        offset=0,
                        limit=1,
                    )
                )["total_entries"]
                user_max_documents = (
                    await self.services.management.get_user_max_documents(
                        auth_user.id
                    )
                )

                if user_document_count >= user_max_documents:
                    raise R2RException(
                        status_code=403,
                        message=f"User has reached the maximum number of documents allowed ({user_max_documents}).",
                    )

                # Get chunks using the vector handler's list_chunks method
                user_chunk_count = (
                    await self.services.ingestion.list_chunks(
                        filters={"owner_id": {"$eq": str(auth_user.id)}},
                        offset=0,
                        limit=1,
                    )
                )["total_entries"]
                user_max_chunks = (
                    await self.services.management.get_user_max_chunks(
                        auth_user.id
                    )
                )
                if user_chunk_count >= user_max_chunks:
                    raise R2RException(
                        status_code=403,
                        message=f"User has reached the maximum number of chunks allowed ({user_max_chunks}).",
                    )

                user_collections_count = (
                    await self.services.management.collections_overview(
                        user_ids=[auth_user.id],
                        offset=0,
                        limit=1,
                    )
                )["total_entries"]
                user_max_collections = (
                    await self.services.management.get_user_max_collections(
                        auth_user.id
                    )
                )
                if user_collections_count >= user_max_collections:  # type: ignore
                    raise R2RException(
                        status_code=403,
                        message=f"User has reached the maximum number of collections allowed ({user_max_collections}).",
                    )

            effective_ingestion_config = self._prepare_ingestion_config(
                ingestion_mode=ingestion_mode,
                ingestion_config=ingestion_config,
            )
            if not file and not raw_text and not chunks:
                raise R2RException(
                    status_code=422,
                    message="Either a `file`, `raw_text`, or `chunks` must be provided.",
                )
            if (
                (file and raw_text)
                or (file and chunks)
                or (raw_text and chunks)
            ):
                raise R2RException(
                    status_code=422,
                    message="Only one of `file`, `raw_text`, or `chunks` may be provided.",
                )
            # Check if the user is a superuser
            metadata = metadata or {}

            if chunks:
                if len(chunks) == 0:
                    raise R2RException("Empty list of chunks provided", 400)

                if len(chunks) > MAX_CHUNKS_PER_REQUEST:
                    raise R2RException(
                        f"Maximum of {MAX_CHUNKS_PER_REQUEST} chunks per request",
                        400,
                    )

                document_id = id or generate_document_id(
                    "".join(chunks), auth_user.id
                )

                # FIXME: Metadata doesn't seem to be getting passed through
                raw_chunks_for_doc = [
                    UnprocessedChunk(
                        text=chunk,
                        metadata=metadata,
                        id=generate_id(),
                    )
                    for chunk in chunks
                ]

                # Prepare workflow input
                workflow_input = {
                    "document_id": str(document_id),
                    "chunks": [
                        chunk.model_dump(mode="json")
                        for chunk in raw_chunks_for_doc
                    ],
                    "collection_ids": (
                        [str(cid) for cid in collection_ids]
                        if collection_ids
                        else None
                    ),
                    "metadata": metadata,  # Base metadata for the document
                    "user": auth_user.model_dump_json(),
                    "ingestion_config": effective_ingestion_config.model_dump(
                        mode="json"
                    ),
                }

                if run_with_orchestration:
                    try:
                        # Run ingestion with orchestration
                        raw_message = (
                            await self.providers.orchestration.run_workflow(
                                "ingest-chunks",
                                {"request": workflow_input},
                                options={
                                    "additional_metadata": {
                                        "document_id": str(document_id),
                                    }
                                },
                            )
                        )
                        raw_message["document_id"] = str(document_id)
                        return raw_message  # type: ignore
                    except Exception as e:  # TODO: Need to find specific errors that we should be excepting (gRPC most likely?)
                        logger.error(
                            f"Error running orchestrated ingestion: {e} \n\nAttempting to run without orchestration."
                        )

                logger.info("Running chunk ingestion without orchestration.")
                from core.main.orchestration import simple_ingestion_factory

                simple_ingestor = simple_ingestion_factory(
                    self.services.ingestion
                )
                await simple_ingestor["ingest-chunks"](workflow_input)

                return {  # type: ignore
                    "message": "Document created and ingested successfully.",
                    "document_id": str(document_id),
                    "task_id": None,
                }

            else:
                if file:
                    file_data = await self._process_file(file)

                    if not file.filename:
                        raise R2RException(
                            status_code=422,
                            message="Uploaded file must have a filename.",
                        )

                    file_ext = file.filename.split(".")[
                        -1
                    ]  # e.g. "pdf", "txt"
                    max_allowed_size = await self.services.management.get_max_upload_size_by_type(
                        user_id=auth_user.id, file_type_or_ext=file_ext
                    )

                    content_length = file_data["content_length"]

                    if content_length > max_allowed_size:
                        raise R2RException(
                            status_code=413,  # HTTP 413: Payload Too Large
                            message=(
                                f"File size exceeds maximum of {max_allowed_size} bytes "
                                f"for extension '{file_ext}'."
                            ),
                        )

                    file_content = BytesIO(
                        base64.b64decode(file_data["content"])
                    )

                    file_data.pop("content", None)
                    document_id = id or generate_document_id(
                        file_data["filename"], auth_user.id
                    )
                elif raw_text:
                    content_length = len(raw_text)
                    file_content = BytesIO(raw_text.encode("utf-8"))
                    document_id = id or generate_document_id(
                        raw_text, auth_user.id
                    )
                    title = metadata.get("title", None)
                    title = title + ".txt" if title else None
                    file_data = {
                        "filename": title or "N/A",
                        "content_type": "text/plain",
                    }
                else:
                    raise R2RException(
                        status_code=422,
                        message="Either a file or content must be provided.",
                    )

            workflow_input = {
                "file_data": file_data,
                "document_id": str(document_id),
                "collection_ids": (
                    [str(cid) for cid in collection_ids]
                    if collection_ids
                    else None
                ),
                "metadata": metadata,
                "ingestion_config": effective_ingestion_config.model_dump(
                    mode="json"
                ),
                "user": auth_user.model_dump_json(),
                "size_in_bytes": content_length,
                "version": "v0",
            }

            file_name = file_data["filename"]
            await self.providers.file.store_file(
                document_id,
                file_name,
                file_content,
                file_data["content_type"],
            )

            await self.services.ingestion.ingest_file_ingress(
                file_data=workflow_input["file_data"],
                user=auth_user,
                document_id=workflow_input["document_id"],
                size_in_bytes=workflow_input["size_in_bytes"],
                metadata=workflow_input["metadata"],
                version=workflow_input["version"],
            )

            if run_with_orchestration:
                try:
                    # TODO - Modify create_chunks so that we can add chunks to existing document

                    workflow_result: dict[
                        str, str | None
                    ] = await self.providers.orchestration.run_workflow(  # type: ignore
                        "ingest-files",
                        {"request": workflow_input},
                        options={
                            "additional_metadata": {
                                "document_id": str(document_id),
                            }
                        },
                    )
                    workflow_result["document_id"] = str(document_id)
                    return workflow_result  # type: ignore
                except Exception as e:  # TODO: Need to find specific error (gRPC most likely?)
                    logger.error(
                        f"Error running orchestrated ingestion: {e} \n\nAttempting to run without orchestration."
                    )
            logger.info(
                f"Running ingestion without orchestration for file {file_name} and document_id {document_id}."
            )
            # TODO - Clean up implementation logic here to be more explicitly `synchronous`
            from core.main.orchestration import simple_ingestion_factory

            simple_ingestor = simple_ingestion_factory(self.services.ingestion)
            await simple_ingestor["ingest-files"](workflow_input)
            return {  # type: ignore
                "message": "Document created and ingested successfully.",
                "document_id": str(document_id),
                "task_id": None,
            }

            Creates a new Document object from an input file, text content, or chunks. The chosen `ingestion_mode` determines
            how the ingestion process is configured:

            **Ingestion Modes:**
            - `hi-res`: Comprehensive parsing and enrichment, including summaries and possibly more thorough parsing.
            - `fast`: Speed-focused ingestion that skips certain enrichment steps like summaries.
            - `custom`: Provide a full `ingestion_config` to customize the entire ingestion process.

            Either a file or text content must be provided, but not both. Documents are shared through `Collections` which allow for tightly specified cross-user interactions.

            The ingestion process runs asynchronously and its progress can be tracked using the returned
            task_id.
            
py/core/main/api/v3/documents_router.py