# 001. Ingestion vertical-slice implementation plan ## Purpose This plan turns the accepted architectural direction in the ADRs into the first working product slice: a tenant-scoped CSV upload is stored in MinIO, represented by durable Postgres records, dispatched through RabbitMQ using a transactional outbox, processed by a separate worker, and indexed as Qdrant points. This is an implementation plan, not an Architecture Decision Record. ADRs explain why major technologies and boundaries were chosen; this document defines the order, scope, and verification criteria for implementing them. ## Architecture baseline The first vertical slice uses these responsibilities: | System | Responsibility | |---|---| | FastAPI | HTTP boundary, validation, authentication, tenant derivation, and job creation. | | Postgres | Tenant/auth data, source-file metadata, ingestion job state/progress, audit, and transactional outbox events. | | MinIO | Private source-file bytes and retained derived ingestion blobs. | | RabbitMQ | Durable delivery of ingestion and maintenance work. | | Ingestion worker | Parsing, chunking, embedding, ingestion-generated Chunk/Point CRUD, and job status updates. | | Qdrant | Tenant-filtered generated chunks and their vectors/payloads. | The controlling ADRs are: - [ADR-0008](../adr/0008-rest-api-and-fastapi-boundary.md): FastAPI REST boundary and job-shaped file ingestion contract. - [ADR-0009](../adr/0009-postgres-sqlalchemy-alembic-schema.md): Postgres source files, jobs, audit, and migration conventions. - [ADR-0012](../adr/0012-application-resource-lifetime-and-dependency-ownership.md): resource lifetime, dependency injection, and explicit transaction ownership. - [ADR-0013](../adr/0013-s3-compatible-object-storage-with-minio.md): MinIO object storage boundary. - [ADR-0014](../adr/0014-durable-job-dispatch-with-rabbitmq.md): RabbitMQ, transactional outbox, separate workers, and worker-owned ingestion Chunk/Point CRUD. The cited ADRs are currently proposed. Treat them as the implementation baseline only after the project owner accepts them; code should not silently diverge from them. ## First release scope ### In scope - `POST /v1/files` for authenticated tenant-scoped **CSV** upload. - File validation, size limits, content hashing, and streaming upload to MinIO. - Alembic-managed Postgres schema for the minimal tenant/auth, source file, ingestion job, job event, and outbox records needed by this slice. - Transactional outbox publication of `ingestion.job.created` to RabbitMQ. - A separate ingestion worker process with a durable RabbitMQ consumer. - CSV parsing and deterministic chunk creation. - Tenant-filtered Qdrant point upserts using deterministic point identifiers. - Job status/progress persistence and `GET /v1/files/{file_id}` status lookup. - Structured correlation logging at HTTP, outbox, and worker ingress. - Automated tests for the critical state transitions, redelivery, and tenant boundaries. ### Explicitly out of scope - XLSX, DOCX, and legacy DOC ingestion. - The conversational LangGraph API and SSE streaming. - Final reranker selection, GPU deployment, or unresolved model licensing from ADR-0005. - Direct `/v1/points` CRUD endpoints beyond the reusable service layer required by ingestion. - Full tenant erasure and hard-deletion workflow. - A public download API or presigned object URLs. - Exactly-once end-to-end processing. The worker must instead be safe under at-least-once delivery. ## Required invariants The implementation must preserve these rules from the ADRs: 1. `tenant_id` is derived from trusted authentication context; it is never accepted from the upload body, query parameters, MinIO metadata, or a broker message as authority. 2. MinIO stores bytes; Postgres stores metadata, lifecycle state, job progress, audit records, and dispatch intent. 3. Broker messages contain stable IDs and correlation metadata only. They never contain file bytes, extracted text, chunks, embeddings, secrets, raw prompts, or raw model output. 4. The worker reloads the job and source-file records from Postgres before doing tenant-scoped work. 5. The worker, not the HTTP publisher, performs parsing, chunking, embedding, and generated Qdrant Chunk/Point CRUD. 6. Qdrant reads and mutations are tenant-filtered. Ingestion-generated point IDs are deterministic so retrying a job does not create duplicate logical chunks. 7. The worker acknowledges a RabbitMQ message only after it has durably persisted the applicable Postgres progress/final state. 8. Application clients are built at FastAPI or worker-process startup and closed at shutdown. No mutable clients are opened as import-time globals. ## Required decisions before implementing the affected phase The first implementation should use these defaults unless a later ADR changes them: ### Source-file idempotency and replacement - Use `(tenant_id, domain, content_sha256)` to recognize identical uploads. - An identical active upload should return the existing source-file/job reference rather than create a duplicate ingestion. - A changed upload creates a new ingestion job. Existing active Qdrant points are replaced only after the new job completes successfully, so a failed re-ingestion does not remove a working index. - Preserve the original filename in Postgres metadata. MinIO object keys remain internal ID-based paths. This policy should be made explicit in ADR-0009 before implementing re-ingestion rather than becoming an accidental repository behavior. ### File deletion For the first release, file deletion should be soft and job-shaped: 1. mark the source file as deletion requested/soft deleted in Postgres; 2. write a maintenance outbox event; 3. have a worker soft-delete the related Qdrant points; 4. retain the MinIO object until an explicit retention or hard-erasure workflow. Hard deletion requires a later retention/erasure implementation covering MinIO, Qdrant, and the relevant Postgres data. ### Broker operations Before deploying an environment, define and document: - RabbitMQ exchange, queue, binding, and dead-letter-exchange configuration; - queue name, prefetch, manual-ack policy, and DLX retry/backoff; - worker concurrency and resource limits; - outbox polling/publish interval and stuck-event alert threshold; - how failed jobs are inspected, retried, and cancelled. These are deployment/runbook settings, not new ADRs unless they change the reliability guarantee or system boundary. ## Build order ### Phase 1: Foundation and local dependencies 1. Add typed configuration in `src/config.py` for Postgres, MinIO, RabbitMQ, Qdrant, application limits, and logging. 2. Populate `.env.example` with non-secret local-development configuration. 3. Add application Docker Compose services for Postgres, MinIO, RabbitMQ, and Qdrant. Keep application MinIO buckets/credentials separate from Langfuse infrastructure. 4. Add direct Python dependencies and lock them with `uv`: SQLAlchemy async/Postgres driver, MinIO/S3 client, aio-pika, Qdrant client, and structured logging dependencies chosen by ADR-0011. 5. Add the pytest foundation from ADR-0016: async test configuration, boundary markers, and support for dependency-injected fakes. Add a lifespan smoke test before creating external clients. 6. Create FastAPI lifespan setup and typed dependency helpers without creating schema at startup. **Exit criteria:** local infrastructure starts; readiness checks can report each required dependency; clients are opened/closed by process owners; fast unit tests run without Docker or live providers. ### Phase 2: Database, migrations, and domain contracts 1. Define SQLAlchemy models and Alembic migrations for the minimum required tables: `tenants`, `api_keys`, `source_files`, `ingestion_jobs`, `ingestion_job_events`, and `outbox_events`. 2. Define Pydantic request/response/message schemas, including a versioned `ingestion.job.created` message containing `event_id`, `tenant_id`, `file_id`, `ingestion_job_id`, `request_id`, and `api_key_id`. 3. Implement explicit repositories/services with a request/job-lifetime `AsyncSession`; routes/services own commit/rollback boundaries as specified by ADR-0012. 4. Write the empty-database migration test before each schema revision. Create Testcontainers-based Postgres fixtures for tenants, hashed API keys, database sessions, and migrations. Do not use `create_all()` in test fixtures. **Exit criteria:** migrations create the schema from an empty database; application startup performs no DDL; schema and repository tests verify tenant-scoped reads/writes and valid job transitions. ### Phase 3: MinIO upload and durable job creation 1. Implement API-key authentication and `AuthContext` tenant derivation. 2. Implement `POST /v1/files` for CSV only, including streaming-size controls, file-type validation, SHA-256 calculation, and a private MinIO upload using an internal object key. 3. In one Postgres transaction, persist `source_files`, create `ingestion_jobs(status='queued')`, and write the corresponding unpublished `outbox_events` row. 4. Return `202 Accepted` with `file_id`, `ingestion_job_id`, and `queued` status. 5. Implement `GET /v1/files/{file_id}` with tenant filtering and a public status response that does not expose raw storage credentials or internal artifacts. 6. Add cleanup/compensation handling for a MinIO upload that succeeds while the database transaction fails. 7. Add unit/API tests for trusted tenant derivation, CSV validation, idempotency, `202 Accepted`, and tenant-scoped status. Add MinIO adapter integration tests for server-derived private object paths and compensation behavior. **Exit criteria:** an authenticated CSV upload creates a private object, a queued job, and an unpublished outbox event; a tenant cannot retrieve another tenant's file status; the HTTP path does not publish directly to RabbitMQ. ### Phase 4: RabbitMQ and outbox publisher 1. Provision/verify the application-owned RabbitMQ topic exchange, queue, and binding configuration through deployment/bootstrap code rather than route startup side effects. 2. Implement an outbox-publisher process that safely claims unpublished events, publishes them with the stable event id using publisher confirms, and records success/failure attempts. 3. RabbitMQ has no broker-native message de-duplication; retain idempotency in all consumers as the sole guard against duplicate processing. 4. Add monitoring/logging for publish attempts, unpublished-event age, and repeated failures. 5. Add unit tests for event claiming and retryable failures, then Testcontainers RabbitMQ integration tests for durable publication, restart/retry, duplicate publication, and message metadata. **Exit criteria:** an outbox event becomes a durable RabbitMQ message after a publisher restart; retrying publication cannot create duplicate application work. ### Phase 5: Ingestion worker and Qdrant Chunk/Point CRUD 1. Create a separate worker entrypoint with its own application-lifetime database, MinIO, RabbitMQ, Qdrant, model, and logging clients. 2. Consume `ingestion.job.created`; reload and validate Postgres records before fetching the MinIO object. 3. Transition the job from `queued` to `running` conditionally, append progress events, parse CSV, create deterministic chunks, and upsert tenant-scoped Qdrant points. 4. Mark the job `succeeded` with counters or `failed` with a safe error summary; acknowledge only after final/progress state is persisted. 5. Make a repeated delivery of the same `ingestion_job_id` safe: no duplicate logical chunks, no incorrect counters, and no transition from a terminal state back to `running`. 6. Add unit tests for deterministic CSV chunks, point IDs, and terminal job transitions. Add Testcontainers Qdrant and RabbitMQ integration tests for tenant-filtered upserts, acknowledgement after durable state, redelivery, and parser/Qdrant failure handling. **Exit criteria:** a successful upload reaches `succeeded`, and its points are retrievable only under the owning tenant's Qdrant filter. Forced worker failure and message redelivery produce a correct final job state. ### Phase 6: Operations, integration tests, and documentation 1. Add an operator runbook covering local startup, migrations, MinIO bucket setup, RabbitMQ exchange/queue setup, web/worker/outbox commands, and job replay. 2. Add a serialized Compose-based operational smoke test where the web process, outbox publisher, and worker run independently for upload through indexed points. Testcontainers remains the standard pytest mechanism for individual adapter integration tests. 3. Add end-to-end tests for duplicate upload, duplicate RabbitMQ delivery, outbox publisher crash/restart, worker crash/restart, tenant isolation, and failed parser/Qdrant behavior. 4. Add health/readiness checks that distinguish process health from dependency readiness. 5. Update the README with local-start instructions and links to ADRs, this plan, and the operations runbook. **Exit criteria:** a new developer can start the stack, apply migrations, upload a CSV, observe the job through completion, and understand how to investigate or retry a failure. ## Definition of done for the vertical slice The first slice is done when the following path works in local Compose and is covered by automated tests: ```text POST /v1/files (authenticated CSV upload) -> raw bytes stored privately in MinIO -> source file, queued job, and outbox event committed in Postgres -> outbox publisher writes a durable RabbitMQ message -> ingestion worker processes and indexes deterministic Qdrant points -> Postgres records progress and terminal job status -> GET /v1/files/{file_id} reports that status within the owning tenant only ``` The next implementation work after this slice is direct Point CRUD, retrieval, and then the LangGraph conversational flow. Do not couple those later milestones to the initial ingestion path unless they are needed to preserve one of the invariants above.