=== Ingesting a directory

Point a pipeline at a folder and the files in it become a knowledge base. No route, no ingestion code:

[source,properties]
----
quarkus.camel.langchain4j.ingest.products.source.directory=/var/data/product-docs
quarkus.camel.langchain4j.ingest.products.embedding-store=products
quarkus.camel.langchain4j.ingest.products.embedding-model=my-model
----

Each file is read as UTF-8 text — there is no format parsing, so convert a PDF or DOCX before it reaches the pipeline — split into overlapping segments (`max-segment-size`, `max-overlap-size`), embedded in batches and written to the store; a document is held in memory whole while it is split. `embedding-store` and `embedding-model` name CDI beans and may be omitted when the application has exactly one of each. Every segment carries `camel_quarkus_pipeline` and `camel_quarkus_document_id` metadata, so retrieval can cite which document an answer came from. Apart from `enabled` and the `source.*` settings shown here, properties are fixed at build time. A pipeline declared through runtime properties alone — nothing but a `source.directory`, say — is invisible to build-time validation; its checks, including the clash with an equally named `@Ingest` pipeline, report at startup instead.

[NOTE]
====
This experimental extension ingests what it is given and keeps no record of it: a document ingested twice leaves two copies in the store, a restart re-reads the whole directory (the duplicate register is in-memory, sized for 100,000 files), and a polled consumer re-reads its source on every poll. Keeping the store in step with a changing source — skipping unchanged documents, replacing edited ones, removing deleted ones — needs that record and arrives with the synchronising engine in a later release.
====

=== Other sources, declared in Java

Any Camel consumer can feed a pipeline — the roughly 300 components, each with its own options and its own documentation. Such a pipeline is declared in Java with `@Ingest` and the Camel Endpoint DSL:

[source,java]
----
@Ingest("events")
IngestPipeline events() {
    return IngestPipeline.from(Source.endpoint(dsl -> dsl.kafka("ingest-events").groupId("ingest"))
                    .documentId("CamelKafkaKey"))
            .embeddingStore("events");
}
----

Typing `dsl.` lists a factory for every component, each completing its own typed options. The method runs once at startup; it must return `IngestPipeline`, take no parameters and use a name no configured pipeline uses. `enabled=false` in configuration switches a Java-declared pipeline off. A component missing from the classpath fails at startup with an error naming the extension artifact that provides it — the DSL compiles regardless, since its factories all ship in one artifact.

Ingestion needs a stable id per document, and where it lives is the consumer's business: `documentId` names the header — the record key `CamelKafkaKey` above, `CamelAwsS3Key` for S3 — or gives a simple-language expression. Without it, the pipeline expects the `CamelIngestDocumentId` header and fails the exchange when it is absent. Mind each component's own defaults, too: the `aws2-s3` consumer *deletes objects after reading them* unless `deleteAfterRead(false)` is set — a knowledge base reads its source, it does not consume it.

After such a pipeline ingests a document, the exchange body is replaced with the `IngestResult`, so a request-reply caller receives the outcome of its call.

The same pipeline can be declared purely in properties: `source.uri` takes the consumer URI as written, `source.document-id` the header name — or a simple-language expression written `+$simple{...}+`, the one form MicroProfile Config leaves untouched. The URI is fixed at build time by design — a runtime-overridable consumer URI would be arbitrary component invocation — while property placeholders inside it still resolve at startup, keeping credentials and endpoints runtime configuration. Treat runtime configuration as the trust boundary it is: the directory and the id expression decide what the process reads into an often external store.

[source,properties]
----
quarkus.camel.langchain4j.ingest.s3docs.source.uri=aws2-s3://product-docs?region=eu-west-1&deleteAfterRead=false
quarkus.camel.langchain4j.ingest.s3docs.source.document-id=CamelAwsS3Key
quarkus.camel.langchain4j.ingest.s3docs.embedding-store=products
----

=== When ingestion fails

A failure while splitting, embedding or storing — a rate-limited model, an unreachable store — propagates to the consumer; there is no dead-letter channel in this increment. For a directory pipeline the file stays where it is and is retried on the next poll, because the duplicate-protection key is only committed on success — which also means a permanently failing file is retried forever, loudly. For a consumer-fed pipeline the component's own error handling applies: a request-reply caller receives the exception, while a Kafka consumer with default settings logs the failure and commits the offset, so the record is *dropped* — and since this engine keeps no record either, nothing remembers it. A record whose configured `document-id` resolves to nothing (a Kafka record without a key, say) fails the same way, one exchange at a time.
