feat(files): add source-file upload with MinIO storage and Postgres repositories

Why:
- Implements plan 001 Phase 3's upload orchestration.

Changes:
- ObjectStorage port and MinIO adapter, thread-offloaded per ADR-0017.
- Tenant-scoped repositories for api_keys, source_files, ingestion_jobs.
- upload_source_file implementing the two-transaction shape with
  (tenant_id, domain, content_sha256) idempotency.

Impact:
- This phase stores bytes only -- chunks_indexed is always 0 until
  Phase 4/5 add parsing/embedding.

src/application/files/__init__.py

@@ -0,0 +1,19 @@
+"""Source-file upload and status use cases (ADR-0008, ADR-0009, ADR-0017)."""
+
+from src.application.files.errors import FilesError, FileTooLargeError, InvalidUploadError
+from src.application.files.models import UploadResult, ValidatedUpload
+from src.application.files.status import FileStatusResult, get_file_status
+from src.application.files.upload import upload_source_file
+from src.application.files.validation import validate_and_hash_upload
+
+__all__ = [
+    "FileStatusResult",
+    "FileTooLargeError",
+    "FilesError",
+    "InvalidUploadError",
+    "UploadResult",
+    "ValidatedUpload",
+    "get_file_status",
+    "upload_source_file",
+    "validate_and_hash_upload",
+]

src/application/files/errors.py

@@ -0,0 +1,17 @@
+"""Upload-validation failures (ADR-0008). No HTTP knowledge here —
+`src/api/errors.py` maps these to status codes.
+"""
+
+
+class FilesError(Exception):
+    """Base class for file-upload failures."""
+
+
+class InvalidUploadError(FilesError):
+    """Missing domain, empty file, or content that doesn't match its
+    declared extension. Maps to `400`.
+    """
+
+
+class FileTooLargeError(FilesError):
+    """The upload exceeds `INGESTION_MAX_UPLOAD_SIZE_MB`. Maps to `413`."""

src/application/files/models.py

@@ -0,0 +1,26 @@
+"""Domain models for the upload use case (ADR-0008, ADR-0009)."""
+
+import uuid
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ValidatedUpload:
+    """The result of extension/content validation, before any I/O."""
+
+    source_type: str
+    content_type: str
+    content_sha256: str
+
+
+@dataclass(frozen=True)
+class UploadResult:
+    """What `upload_source_file` returns; the route maps this to `FileUploadResponse`."""
+
+    file_id: uuid.UUID
+    ingestion_job_id: uuid.UUID
+    status: str
+    chunks_indexed: int
+    is_new_attempt: bool
+    """`False` when an identical active upload already succeeded and no new
+    ingestion attempt was made (route returns `200`, not `201`)."""

src/application/files/status.py

@@ -0,0 +1,52 @@
+"""`GET /v1/files/{file_id}` read model (ADR-0008, ADR-0009)."""
+
+import uuid
+from dataclasses import dataclass
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo
+from src.infrastructure.postgres.repositories import source_files as source_files_repo
+
+
+@dataclass(frozen=True)
+class FileStatusResult:
+    file_id: uuid.UUID
+    source_filename: str
+    domain: str
+    status: str
+    ingestion_job_id: uuid.UUID | None
+    ingestion_status: str | None
+    chunks_indexed: int
+
+
+async def get_file_status(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    tenant_id: uuid.UUID,
+    source_file_id: uuid.UUID,
+) -> FileStatusResult | None:
+    """Returns `None` when the file doesn't exist under this tenant — the
+    route maps that to `404`, never `403` (ADR-0016: cross-tenant access
+    returns 404).
+    """
+    async with sessionmaker() as session:
+        source_file = await source_files_repo.get_by_id(
+            session, tenant_id=tenant_id, source_file_id=source_file_id
+        )
+        if source_file is None:
+            return None
+
+        latest_job = await jobs_repo.get_latest_for_source_file(
+            session, tenant_id=tenant_id, source_file_id=source_file.id
+        )
+
+        return FileStatusResult(
+            file_id=source_file.id,
+            source_filename=source_file.source_filename,
+            domain=source_file.domain,
+            status=source_file.status,
+            ingestion_job_id=latest_job.id if latest_job else None,
+            ingestion_status=latest_job.status if latest_job else None,
+            chunks_indexed=latest_job.points_created if latest_job else 0,
+        )

src/application/files/storage_keys.py

@@ -0,0 +1,11 @@
+"""Object-storage key derivation (ADR-0013).
+
+Object keys are internal identifiers, never the caller-supplied filename.
+Pure and synchronous.
+"""
+
+import uuid
+
+
+def source_file_object_key(tenant_id: uuid.UUID, source_file_id: uuid.UUID) -> str:
+    return f"tenants/{tenant_id}/source-files/{source_file_id}/original"

src/application/files/upload.py

@@ -0,0 +1,205 @@
+"""`POST /v1/files` orchestration: the ADR-0017 three-phase upload.
+
+This service owns two separate short-lived sessions/transactions rather than
+one request-scoped session, because the request is two units of work
+(ADR-0012, ADR-0017):
+
+    txn A (short):  source_files [+ ingestion_jobs(status='running')], commit
+    no txn:          store bytes in MinIO
+    txn B (short):  ingestion_jobs -> succeeded/failed, append event, commit
+
+No Postgres session is open during the MinIO write. A storage failure between
+txn A and txn B still leaves a durable, inspectable `failed` job — never a
+job stuck in `running`.
+
+Parsing/chunking/Qdrant indexing are Phase 4/5 work, not implemented here:
+this phase stores bytes only, so a successful job reports `chunks_indexed=0`.
+"""
+
+import uuid
+
+import structlog
+from anyio import CapacityLimiter, to_thread
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.context import AuthContext
+from src.application.files.errors import InvalidUploadError
+from src.application.files.models import UploadResult
+from src.application.files.storage_keys import source_file_object_key
+from src.application.files.validation import validate_and_hash_upload
+from src.application.ports.object_storage import ObjectStorage
+from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo
+from src.infrastructure.postgres.repositories import source_files as source_files_repo
+
+logger = structlog.get_logger(__name__)
+
+
+async def _mark_job_failed(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    tenant_id: uuid.UUID,
+    ingestion_job_id: uuid.UUID,
+    error_code: str,
+    error_message: str,
+) -> None:
+    async with sessionmaker() as session:
+        job = await jobs_repo.mark_terminal(
+            session,
+            tenant_id=tenant_id,
+            ingestion_job_id=ingestion_job_id,
+            status="failed",
+            error_code=error_code,
+            error_message=error_message,
+        )
+        if job is not None:
+            jobs_repo.append_event(
+                session,
+                tenant_id=tenant_id,
+                ingestion_job_id=ingestion_job_id,
+                level="error",
+                stage="received",
+                message=error_message,
+            )
+        await session.commit()
+
+
+async def upload_source_file(
+    *,
+    sessionmaker: async_sessionmaker[AsyncSession],
+    storage: ObjectStorage,
+    auth: AuthContext,
+    domain: str,
+    filename: str,
+    data: bytes,
+    max_upload_size_bytes: int,
+    chunking_strategy: str,
+    validation_limiter: CapacityLimiter,
+) -> UploadResult:
+    domain = domain.strip()
+    if not domain:
+        raise InvalidUploadError("domain is required")
+
+    validated = await to_thread.run_sync(
+        lambda: validate_and_hash_upload(
+            filename=filename, data=data, max_size_bytes=max_upload_size_bytes
+        ),
+        limiter=validation_limiter,
+    )
+
+    async with sessionmaker() as session:
+        existing = await source_files_repo.find_active_by_content_hash(
+            session,
+            tenant_id=auth.tenant_id,
+            domain=domain,
+            content_sha256=validated.content_sha256,
+        )
+
+        if existing is not None:
+            latest_job = await jobs_repo.get_latest_for_source_file(
+                session, tenant_id=auth.tenant_id, source_file_id=existing.id
+            )
+            if latest_job is not None and latest_job.status == "succeeded":
+                logger.info(
+                    "files.upload.duplicate",
+                    tenant_id=str(auth.tenant_id),
+                    file_id=str(existing.id),
+                )
+                return UploadResult(
+                    file_id=existing.id,
+                    ingestion_job_id=latest_job.id,
+                    status=latest_job.status,
+                    chunks_indexed=latest_job.points_created,
+                    is_new_attempt=False,
+                )
+            source_file_id = existing.id
+            object_key = existing.storage_uri or source_file_object_key(
+                auth.tenant_id, source_file_id
+            )
+        else:
+            source_file_id = uuid.uuid4()
+            object_key = source_file_object_key(auth.tenant_id, source_file_id)
+            source_files_repo.create(
+                session,
+                source_file_id=source_file_id,
+                tenant_id=auth.tenant_id,
+                domain=domain,
+                source_filename=filename,
+                source_type=validated.source_type,
+                content_sha256=validated.content_sha256,
+                byte_size=len(data),
+                storage_uri=object_key,
+                created_by_api_key_id=auth.api_key_id,
+            )
+            # `ingestion_jobs.source_file_id` FKs to this row; flush so the
+            # insert below sees it, since the two mapped classes carry no
+            # ORM relationship for the unit of work to order by itself.
+            await session.flush()
+
+        job = jobs_repo.create_running(
+            session,
+            tenant_id=auth.tenant_id,
+            source_file_id=source_file_id,
+            requested_by_api_key_id=auth.api_key_id,
+            chunking_strategy=chunking_strategy,
+        )
+        jobs_repo.append_event(
+            session,
+            tenant_id=auth.tenant_id,
+            ingestion_job_id=job.id,
+            level="info",
+            stage="received",
+            message="upload accepted, storing object",
+        )
+        await session.commit()
+        ingestion_job_id = job.id
+
+    # Phase 2: no Postgres session open across this work (ADR-0017).
+    try:
+        await storage.put_object(key=object_key, data=data, content_type=validated.content_type)
+    except Exception as exc:
+        logger.warning(
+            "files.upload.storage_failed",
+            tenant_id=str(auth.tenant_id),
+            file_id=str(source_file_id),
+            ingestion_job_id=str(ingestion_job_id),
+        )
+        await _mark_job_failed(
+            sessionmaker,
+            tenant_id=auth.tenant_id,
+            ingestion_job_id=ingestion_job_id,
+            error_code="storage_upload_failed",
+            error_message=f"failed to store object: {exc}",
+        )
+        raise
+
+    async with sessionmaker() as session:
+        await jobs_repo.mark_terminal(
+            session,
+            tenant_id=auth.tenant_id,
+            ingestion_job_id=ingestion_job_id,
+            status="succeeded",
+            points_created=0,
+        )
+        jobs_repo.append_event(
+            session,
+            tenant_id=auth.tenant_id,
+            ingestion_job_id=ingestion_job_id,
+            level="info",
+            stage="completed",
+            message="object stored; parsing/embedding/indexing not yet implemented",
+        )
+        await session.commit()
+
+    logger.info(
+        "files.upload.succeeded",
+        tenant_id=str(auth.tenant_id),
+        file_id=str(source_file_id),
+        ingestion_job_id=str(ingestion_job_id),
+    )
+    return UploadResult(
+        file_id=source_file_id,
+        ingestion_job_id=ingestion_job_id,
+        status="succeeded",
+        chunks_indexed=0,
+        is_new_attempt=True,
+    )

src/application/files/validation.py

@@ -0,0 +1,55 @@
+"""Upload extension/content-type/size validation (ADR-0008).
+
+Pure and synchronous: no I/O. `content_sha256` computation lives here too —
+hashing is blocking CPU work (ADR-0017), so the caller runs this whole
+function through `anyio.to_thread.run_sync` with the ingestion
+`CapacityLimiter`, the same rule applied to parsing/chunking.
+"""
+
+import hashlib
+
+from src.application.files.errors import FileTooLargeError, InvalidUploadError
+from src.application.files.models import ValidatedUpload
+from src.application.ingestion.errors import UnsupportedSourceTypeError
+
+_CONTENT_TYPES = {
+    "csv": "text/csv",
+    "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+}
+_OOXML_MAGIC = b"PK\x03\x04"
+
+
+def _source_type_from_filename(filename: str) -> str:
+    suffix = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
+    if suffix == "doc":
+        raise UnsupportedSourceTypeError(
+            "legacy .doc is not ingestible until an out-of-process conversion "
+            "service exists (ADR-0018)"
+        )
+    if suffix not in _CONTENT_TYPES:
+        raise UnsupportedSourceTypeError(f"'.{suffix}' is not an ingestible file type")
+    return suffix
+
+
+def validate_and_hash_upload(*, filename: str, data: bytes, max_size_bytes: int) -> ValidatedUpload:
+    source_type = _source_type_from_filename(filename)
+
+    if not data:
+        raise InvalidUploadError("uploaded file is empty")
+    if len(data) > max_size_bytes:
+        raise FileTooLargeError(
+            f"upload is {len(data)} bytes, over the {max_size_bytes}-byte limit"
+        )
+
+    is_ooxml = data[:4] == _OOXML_MAGIC
+    if source_type in ("docx", "xlsx") and not is_ooxml:
+        raise InvalidUploadError(f"content does not match the declared .{source_type} extension")
+    if source_type == "csv" and is_ooxml:
+        raise InvalidUploadError("content does not match the declared .csv extension")
+
+    return ValidatedUpload(
+        source_type=source_type,
+        content_type=_CONTENT_TYPES[source_type],
+        content_sha256=hashlib.sha256(data).hexdigest(),
+    )

src/application/ports/__init__.py

@@ -0,0 +1,7 @@
+"""Narrow contracts for external side effects (ADR-0015).
+
+Ports exist for external side effects/persistence that need a swappable or
+fake-able boundary — not as a blanket wrapper around every database access.
+`object_storage.py` is one: MinIO is a real external system with its own
+failure modes, and ADR-0016 requires a hand-written fake for it in tests.
+"""

src/application/ports/object_storage.py

@@ -0,0 +1,14 @@
+"""The object-storage port (ADR-0013).
+
+`src/infrastructure/minio/storage.py` is the production adapter; tests use a
+hand-written fake (ADR-0016). Application code depends on this Protocol, not
+on the `minio` SDK.
+"""
+
+from typing import Protocol
+
+
+class ObjectStorage(Protocol):
+    async def put_object(self, *, key: str, data: bytes, content_type: str) -> None:
+        """Store `data` privately under `key`. Overwrites an existing object."""
+        ...

src/infrastructure/minio/storage.py

@@ -0,0 +1,33 @@
+"""MinIO adapter for the `ObjectStorage` port (ADR-0013, ADR-0017).
+
+The `minio` SDK is synchronous, so every call runs through
+`anyio.to_thread.run_sync` bounded by the ingestion `CapacityLimiter` — the
+same rule ADR-0017 applies to parsing/chunking. Calling the SDK directly from
+`async def` would block every concurrent request in the process.
+"""
+
+import io
+from functools import partial
+
+from anyio import CapacityLimiter, to_thread
+from minio import Minio
+
+
+class MinioObjectStorage:
+    def __init__(self, client: Minio, *, bucket: str, limiter: CapacityLimiter) -> None:
+        self._client = client
+        self._bucket = bucket
+        self._limiter = limiter
+
+    async def put_object(self, *, key: str, data: bytes, content_type: str) -> None:
+        await to_thread.run_sync(
+            partial(
+                self._client.put_object,
+                self._bucket,
+                key,
+                io.BytesIO(data),
+                length=len(data),
+                content_type=content_type,
+            ),
+            limiter=self._limiter,
+        )

src/infrastructure/postgres/repositories/api_keys.py

@@ -0,0 +1,17 @@
+"""API-key lookups (ADR-0008, ADR-0009).
+
+Plain functions over an `AsyncSession` the caller owns. No function here
+commits, rolls back, or closes the session (ADR-0012). Secret comparison
+happens in `src/application/auth`, not here — this module only fetches rows
+by their non-secret `key_prefix`.
+"""
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.models.api_key import ApiKey
+
+
+async def get_by_prefix(session: AsyncSession, key_prefix: str) -> ApiKey | None:
+    result = await session.execute(select(ApiKey).where(ApiKey.key_prefix == key_prefix))
+    return result.scalar_one_or_none()

src/infrastructure/postgres/repositories/ingestion_jobs.py

@@ -0,0 +1,119 @@
+"""`ingestion_jobs`/`ingestion_job_events` persistence (ADR-0009, ADR-0017).
+
+Plain functions over an `AsyncSession` the caller owns. No function here
+commits, rolls back, or closes the session (ADR-0012) — the two-transaction
+shape in `src/application/files/upload.py` depends on that. Every read is
+tenant-scoped by a required `tenant_id` argument.
+"""
+
+import uuid
+from datetime import UTC, datetime
+
+from sqlalchemy import desc, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.models.ingestion_job import IngestionJob
+from src.infrastructure.postgres.models.ingestion_job_event import IngestionJobEvent
+
+
+async def get_by_id(
+    session: AsyncSession, *, tenant_id: uuid.UUID, ingestion_job_id: uuid.UUID
+) -> IngestionJob | None:
+    result = await session.execute(
+        select(IngestionJob).where(
+            IngestionJob.id == ingestion_job_id, IngestionJob.tenant_id == tenant_id
+        )
+    )
+    return result.scalar_one_or_none()
+
+
+async def get_latest_for_source_file(
+    session: AsyncSession, *, tenant_id: uuid.UUID, source_file_id: uuid.UUID
+) -> IngestionJob | None:
+    result = await session.execute(
+        select(IngestionJob)
+        .where(
+            IngestionJob.tenant_id == tenant_id,
+            IngestionJob.source_file_id == source_file_id,
+        )
+        .order_by(desc(IngestionJob.created_at))
+        .limit(1)
+    )
+    return result.scalar_one_or_none()
+
+
+def create_running(
+    session: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    source_file_id: uuid.UUID,
+    requested_by_api_key_id: uuid.UUID | None,
+    chunking_strategy: str,
+) -> IngestionJob:
+    job = IngestionJob(
+        id=uuid.uuid4(),
+        tenant_id=tenant_id,
+        source_file_id=source_file_id,
+        requested_by_api_key_id=requested_by_api_key_id,
+        status="running",
+        chunking_strategy=chunking_strategy,
+        started_at=datetime.now(UTC),
+    )
+    session.add(job)
+    return job
+
+
+async def mark_terminal(
+    session: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    ingestion_job_id: uuid.UUID,
+    status: str,
+    points_created: int = 0,
+    points_updated: int = 0,
+    points_soft_deleted: int = 0,
+    points_skipped: int = 0,
+    error_code: str | None = None,
+    error_message: str | None = None,
+) -> IngestionJob | None:
+    """Move a job from `running` to a terminal status.
+
+    Never reads/writes a job whose current status is already terminal — a
+    terminal job must not transition back to `running` or to a different
+    terminal status (ADR-0017).
+    """
+    job = await get_by_id(session, tenant_id=tenant_id, ingestion_job_id=ingestion_job_id)
+    if job is None or job.status != "running":
+        return None
+    job.status = status
+    job.completed_at = datetime.now(UTC)
+    job.points_created = points_created
+    job.points_updated = points_updated
+    job.points_soft_deleted = points_soft_deleted
+    job.points_skipped = points_skipped
+    job.error_code = error_code
+    job.error_message = error_message
+    return job
+
+
+def append_event(
+    session: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    ingestion_job_id: uuid.UUID,
+    level: str,
+    stage: str,
+    message: str,
+    details: dict[str, object] | None = None,
+) -> IngestionJobEvent:
+    event = IngestionJobEvent(
+        id=uuid.uuid4(),
+        tenant_id=tenant_id,
+        ingestion_job_id=ingestion_job_id,
+        level=level,
+        stage=stage,
+        message=message,
+        details=details or {},
+    )
+    session.add(event)
+    return event

src/infrastructure/postgres/repositories/source_files.py

@@ -0,0 +1,69 @@
+"""`source_files` persistence (ADR-0009).
+
+Plain functions over an `AsyncSession` the caller owns. No function here
+commits, rolls back, or closes the session (ADR-0012). Every read is
+tenant-scoped by a required `tenant_id` argument, so a missing filter is a
+signature error rather than a cross-tenant leak.
+"""
+
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.models.source_file import SourceFile
+
+
+async def get_by_id(
+    session: AsyncSession, *, tenant_id: uuid.UUID, source_file_id: uuid.UUID
+) -> SourceFile | None:
+    result = await session.execute(
+        select(SourceFile).where(SourceFile.id == source_file_id, SourceFile.tenant_id == tenant_id)
+    )
+    return result.scalar_one_or_none()
+
+
+async def find_active_by_content_hash(
+    session: AsyncSession, *, tenant_id: uuid.UUID, domain: str, content_sha256: str
+) -> SourceFile | None:
+    result = await session.execute(
+        select(SourceFile).where(
+            SourceFile.tenant_id == tenant_id,
+            SourceFile.domain == domain,
+            SourceFile.content_sha256 == content_sha256,
+            SourceFile.status == "active",
+        )
+    )
+    return result.scalar_one_or_none()
+
+
+def create(
+    session: AsyncSession,
+    *,
+    source_file_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+    domain: str,
+    source_filename: str,
+    source_type: str,
+    content_sha256: str,
+    byte_size: int,
+    storage_uri: str,
+    created_by_api_key_id: uuid.UUID | None,
+) -> SourceFile:
+    """`source_file_id` is caller-generated: the upload service derives the
+    MinIO object key from it before this row exists, so the id has to be
+    chosen up front rather than assigned by the database.
+    """
+    source_file = SourceFile(
+        id=source_file_id,
+        tenant_id=tenant_id,
+        domain=domain,
+        source_filename=source_filename,
+        source_type=source_type,
+        content_sha256=content_sha256,
+        byte_size=byte_size,
+        storage_uri=storage_uri,
+        created_by_api_key_id=created_by_api_key_id,
+    )
+    session.add(source_file)
+    return source_file

src/infrastructure/postgres/repositories/tenants.py

@@ -0,0 +1,15 @@
+"""Tenant lookups (ADR-0009).
+
+Plain functions over an `AsyncSession` the caller owns. No function here
+commits, rolls back, or closes the session (ADR-0012).
+"""
+
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.models.tenant import Tenant
+
+
+async def get_by_id(session: AsyncSession, tenant_id: uuid.UUID) -> Tenant | None:
+    return await session.get(Tenant, tenant_id)