diff --git a/CHANGELOG.md b/CHANGELOG.md index 2031ddf..abdc198 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this - OpenSearch adapter (`OpenSearchStore`): kNN vector search, full-text search, bulk indexing, document CRUD - Embedding microservice (`embedding-service/`): sentence-transformers REST API with batch processing and Dockerfile - `EmbeddingService` port and `EmbeddingClient` HTTP adapter for remote embedding generation +- Orchestrated ingestion pipeline: Docling → chunking → embedding → OpenSearch indexing (idempotent) +- Ingestion REST API: `POST /api/ingestion/{jobId}`, `DELETE /api/ingestion/{docId}`, `GET /api/ingestion/status` +- Production docker-compose with OpenSearch and embedding service +- E2E Karate test for full ingestion workflow (PDF → chunks in OpenSearch) +- My Documents screen: search, filter (all/indexed/not indexed), sort (name/date), ingestion status badges +- Ingest button in Studio: one-click ingestion from completed analysis with progress feedback ### Fixed diff --git a/docker-compose.yml b/docker-compose.yml index 3eb3d1e..e965fc5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,38 @@ services: + # --- OpenSearch (single-node, security disabled) --- + opensearch: + image: opensearchproject/opensearch:2 + environment: + discovery.type: single-node + DISABLE_SECURITY_PLUGIN: "true" + OPENSEARCH_JAVA_OPTS: "-Xms512m -Xmx512m" + volumes: + - opensearch_data:/usr/share/opensearch/data + healthcheck: + test: ["CMD-SHELL", "curl -sf http://localhost:9200/_cluster/health || exit 1"] + interval: 10s + timeout: 5s + retries: 10 + + # --- Embedding service (sentence-transformers) --- + embedding: + build: + context: ./embedding-service + environment: + EMBEDDING_MODEL: ${EMBEDDING_MODEL:-all-MiniLM-L6-v2} + EMBEDDING_BATCH_SIZE: ${EMBEDDING_BATCH_SIZE:-64} + healthcheck: + test: ["CMD-SHELL", "curl -sf http://localhost:8001/health || exit 1"] + interval: 15s + timeout: 10s + retries: 20 + start_period: 120s + deploy: + resources: + limits: + memory: 2g + + # --- Backend (FastAPI) --- document-parser: build: context: ./document-parser @@ -15,11 +49,19 @@ services: RATE_LIMIT_RPM: ${RATE_LIMIT_RPM:-100} MAX_FILE_SIZE_MB: ${MAX_FILE_SIZE_MB:-50} BATCH_PAGE_SIZE: ${BATCH_PAGE_SIZE:-0} + OPENSEARCH_URL: http://opensearch:9200 + EMBEDDING_URL: http://embedding:8001 + depends_on: + opensearch: + condition: service_healthy + embedding: + condition: service_healthy deploy: resources: limits: memory: 4g + # --- Frontend (nginx) --- frontend: build: context: ./frontend @@ -29,5 +71,6 @@ services: - document-parser volumes: + opensearch_data: uploads_data: db_data: diff --git a/document-parser/api/ingestion.py b/document-parser/api/ingestion.py new file mode 100644 index 0000000..2f48e7d --- /dev/null +++ b/document-parser/api/ingestion.py @@ -0,0 +1,82 @@ +"""Ingestion API router — trigger and manage vector ingestion pipeline.""" + +from __future__ import annotations + +import logging +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, Request + +from api.schemas import IngestionResponse, IngestionStatusResponse +from services.analysis_service import AnalysisService +from services.ingestion_service import IngestionService + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/ingestion", tags=["ingestion"]) + + +def _get_ingestion_service(request: Request) -> IngestionService: + svc = request.app.state.ingestion_service + if svc is None: + raise HTTPException( + status_code=503, + detail="Ingestion not available (EMBEDDING_URL and OPENSEARCH_URL required)", + ) + return svc + + +def _get_analysis_service(request: Request) -> AnalysisService: + return request.app.state.analysis_service + + +IngestionDep = Annotated[IngestionService, Depends(_get_ingestion_service)] +AnalysisDep = Annotated[AnalysisService, Depends(_get_analysis_service)] + + +@router.post("/{job_id}", response_model=IngestionResponse) +async def ingest_analysis( + job_id: str, + ingestion: IngestionDep, + analysis: AnalysisDep, +) -> IngestionResponse: + """Ingest a completed analysis into the vector index. + + Takes the chunks from an existing analysis job, embeds them, + and indexes them into OpenSearch. + """ + job = await analysis.find_by_id(job_id) + if not job: + raise HTTPException(status_code=404, detail="Analysis not found") + if job.status.value != "COMPLETED": + raise HTTPException(status_code=400, detail="Analysis is not completed") + if not job.chunks_json: + raise HTTPException(status_code=400, detail="Analysis has no chunks — run chunking first") + + try: + result = await ingestion.ingest( + doc_id=job.document_id, + filename=job.document_filename or "unknown", + chunks_json=job.chunks_json, + ) + except Exception as e: + logger.exception("Ingestion failed for job %s", job_id) + raise HTTPException(status_code=500, detail=f"Ingestion failed: {e}") from e + + return IngestionResponse( + doc_id=result.doc_id, + chunks_indexed=result.chunks_indexed, + embedding_dimension=result.embedding_dimension, + ) + + +@router.delete("/{doc_id}", status_code=204) +async def delete_ingested_document(doc_id: str, ingestion: IngestionDep) -> None: + """Delete all indexed chunks for a document.""" + await ingestion.delete_document(doc_id) + + +@router.get("/status", response_model=IngestionStatusResponse) +async def ingestion_status(request: Request) -> IngestionStatusResponse: + """Check if the ingestion pipeline is available.""" + available = request.app.state.ingestion_service is not None + return IngestionStatusResponse(available=available) diff --git a/document-parser/api/schemas.py b/document-parser/api/schemas.py index 42079a6..808b70b 100644 --- a/document-parser/api/schemas.py +++ b/document-parser/api/schemas.py @@ -180,3 +180,13 @@ class RechunkRequest(BaseModel): chunkingOptions: ChunkingOptionsRequest = Field( validation_alias=AliasChoices("chunkingOptions", "chunking_options") ) + + +class IngestionResponse(_CamelModel): + doc_id: str + chunks_indexed: int + embedding_dimension: int + + +class IngestionStatusResponse(_CamelModel): + available: bool diff --git a/document-parser/main.py b/document-parser/main.py index aa6573c..a316b34 100644 --- a/document-parser/main.py +++ b/document-parser/main.py @@ -20,6 +20,7 @@ from fastapi.middleware.cors import CORSMiddleware from api.analyses import router as analyses_router from api.documents import router as documents_router +from api.ingestion import router as ingestion_router from api.schemas import HealthResponse from infra.rate_limiter import RateLimiterMiddleware from infra.settings import settings @@ -28,6 +29,7 @@ from persistence.database import get_connection, init_db from persistence.document_repo import SqliteDocumentRepository from services.analysis_service import AnalysisConfig, AnalysisService from services.document_service import DocumentConfig, DocumentService +from services.ingestion_service import IngestionConfig, IngestionService logging.basicConfig( level=logging.INFO, @@ -87,6 +89,28 @@ def _build_analysis_service( ) +def _build_ingestion_service() -> IngestionService | None: + """Build the ingestion service — only if embedding + opensearch are configured.""" + if not settings.embedding_url or not settings.opensearch_url: + logger.info("Ingestion disabled (EMBEDDING_URL or OPENSEARCH_URL not set)") + return None + + from infra.embedding_client import EmbeddingClient + from infra.opensearch_store import OpenSearchStore + + embedding = EmbeddingClient(settings.embedding_url) + vector_store = OpenSearchStore(settings.opensearch_url) + config = IngestionConfig( + embedding_dimension=settings.embedding_dimension, + ) + logger.info( + "Ingestion enabled (embedding=%s, opensearch=%s)", + settings.embedding_url, + settings.opensearch_url, + ) + return IngestionService(embedding, vector_store, config) + + def _build_document_service( document_repo: SqliteDocumentRepository, analysis_repo: SqliteAnalysisRepository, @@ -114,6 +138,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: document_repo, analysis_repo = _build_repos() app.state.analysis_service = _build_analysis_service(document_repo, analysis_repo) app.state.document_service = _build_document_service(document_repo, analysis_repo) + app.state.ingestion_service = _build_ingestion_service() logger.info("Docling Studio backend ready (engine=%s)", settings.conversion_engine) yield @@ -128,7 +153,7 @@ app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, allow_credentials=True, - allow_methods=["GET", "POST", "DELETE", "OPTIONS"], + allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"], allow_headers=["Content-Type", "Authorization"], ) if settings.rate_limit_rpm > 0: @@ -140,6 +165,7 @@ if settings.rate_limit_rpm > 0: app.include_router(documents_router) app.include_router(analyses_router) +app.include_router(ingestion_router) @app.get("/api/health", response_model=HealthResponse) diff --git a/document-parser/services/ingestion_service.py b/document-parser/services/ingestion_service.py new file mode 100644 index 0000000..ca29c1e --- /dev/null +++ b/document-parser/services/ingestion_service.py @@ -0,0 +1,166 @@ +"""Ingestion service — orchestrates Docling → embedding → OpenSearch. + +Chains the full ingestion pipeline: +1. Convert document via Docling (reuse existing analysis) +2. Chunk with selected strategy +3. Embed all chunk texts via EmbeddingService +4. Index into OpenSearch via VectorStore + +Idempotent: re-ingesting a document deletes old chunks before re-indexing. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from domain.vector_schema import ( + ChunkBboxEntry, + ChunkOrigin, + IndexedChunk, + build_index_mapping, +) + +if TYPE_CHECKING: + from domain.ports import EmbeddingService, VectorStore + +logger = logging.getLogger(__name__) + + +@dataclass +class IngestionConfig: + """Configuration for the ingestion pipeline.""" + + index_name: str = "docling-studio-chunks" + embedding_dimension: int = 384 + + +@dataclass +class IngestionResult: + """Result of an ingestion pipeline run.""" + + doc_id: str + chunks_indexed: int + embedding_dimension: int + + +class IngestionService: + """Orchestrates the embedding + indexing pipeline.""" + + def __init__( + self, + embedding_service: EmbeddingService, + vector_store: VectorStore, + config: IngestionConfig | None = None, + ) -> None: + self._embedding = embedding_service + self._vector_store = vector_store + self._config = config or IngestionConfig() + + async def ensure_index(self) -> None: + """Ensure the vector index exists with the correct mapping.""" + mapping = build_index_mapping(self._config.embedding_dimension) + await self._vector_store.ensure_index(self._config.index_name, mapping) + + async def ingest( + self, + doc_id: str, + filename: str, + chunks_json: str, + *, + binary_hash: str | None = None, + ) -> IngestionResult: + """Run the embedding + indexing pipeline on pre-chunked data. + + This method is idempotent: it deletes any existing chunks for the + document before re-indexing. + + Args: + doc_id: Unique document identifier. + filename: Original filename. + chunks_json: JSON-serialized list of chunk dicts (from analysis). + binary_hash: Optional hash of the source file for provenance. + + Returns: + IngestionResult with the number of chunks indexed. + """ + await self.ensure_index() + + chunks_data: list[dict] = json.loads(chunks_json) + active_chunks = [c for c in chunks_data if not c.get("deleted")] + if not active_chunks: + logger.info("No active chunks for doc %s — skipping ingestion", doc_id) + return IngestionResult(doc_id=doc_id, chunks_indexed=0, embedding_dimension=0) + + # 1. Embed all chunk texts + texts = [c["text"] for c in active_chunks] + logger.info("Embedding %d chunks for doc %s", len(texts), doc_id) + embeddings = await self._embedding.embed(texts) + + # 2. Build IndexedChunk domain objects + origin = ( + ChunkOrigin(binary_hash=binary_hash or "", filename=filename) if binary_hash else None + ) + indexed_chunks: list[IndexedChunk] = [] + for i, (chunk_data, embedding) in enumerate(zip(active_chunks, embeddings, strict=True)): + bboxes = [ + ChunkBboxEntry( + page=b["page"], + x=b["bbox"][0] if b.get("bbox") else 0, + y=b["bbox"][1] if b.get("bbox") else 0, + w=(b["bbox"][2] - b["bbox"][0]) if b.get("bbox") and len(b["bbox"]) >= 4 else 0, + h=(b["bbox"][3] - b["bbox"][1]) if b.get("bbox") and len(b["bbox"]) >= 4 else 0, + ) + for b in chunk_data.get("bboxes", []) + ] + indexed_chunks.append( + IndexedChunk( + doc_id=doc_id, + filename=filename, + content=chunk_data["text"], + embedding=embedding, + chunk_index=i, + chunk_type=chunk_data.get("chunkType", "text"), + page_number=chunk_data.get("sourcePage", 0) or 0, + bboxes=bboxes, + headings=chunk_data.get("headings", []), + origin=origin, + ) + ) + + # 3. Delete old chunks (idempotent re-indexing) + deleted = await self._vector_store.delete_document(self._config.index_name, doc_id) + if deleted: + logger.info("Deleted %d old chunks for doc %s", deleted, doc_id) + + # 4. Index new chunks + indexed = await self._vector_store.index_chunks(self._config.index_name, indexed_chunks) + logger.info("Indexed %d/%d chunks for doc %s", indexed, len(indexed_chunks), doc_id) + + return IngestionResult( + doc_id=doc_id, + chunks_indexed=indexed, + embedding_dimension=len(embeddings[0]) if embeddings else 0, + ) + + async def delete_document(self, doc_id: str) -> int: + """Remove all indexed chunks for a document.""" + return await self._vector_store.delete_document(self._config.index_name, doc_id) + + async def search( + self, + query: str, + *, + k: int = 10, + doc_id: str | None = None, + ) -> list: + """Semantic search: embed the query then find nearest chunks.""" + embeddings = await self._embedding.embed([query]) + return await self._vector_store.search_similar( + self._config.index_name, + embeddings[0], + k=k, + doc_id=doc_id, + ) diff --git a/document-parser/tests/test_ingestion_api.py b/document-parser/tests/test_ingestion_api.py new file mode 100644 index 0000000..ee124a8 --- /dev/null +++ b/document-parser/tests/test_ingestion_api.py @@ -0,0 +1,114 @@ +"""Tests for the ingestion API endpoints (api.ingestion).""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from api.ingestion import router +from domain.models import AnalysisJob +from services.ingestion_service import IngestionResult + + +@pytest.fixture +def mock_ingestion_service() -> AsyncMock: + svc = AsyncMock() + svc.ingest.return_value = IngestionResult( + doc_id="doc-1", chunks_indexed=5, embedding_dimension=384 + ) + svc.delete_document.return_value = 3 + return svc + + +@pytest.fixture +def mock_analysis_service() -> AsyncMock: + svc = AsyncMock() + job = AnalysisJob(document_id="doc-1") + job.document_filename = "test.pdf" + job.mark_running() + job.mark_completed( + markdown="# Test", + html="

Test

", + pages_json="[]", + document_json='{"doc": true}', + chunks_json='[{"text": "hello"}]', + ) + svc.find_by_id.return_value = job + return svc + + +@pytest.fixture +def client(mock_ingestion_service: AsyncMock, mock_analysis_service: AsyncMock) -> TestClient: + app = FastAPI() + app.include_router(router) + app.state.ingestion_service = mock_ingestion_service + app.state.analysis_service = mock_analysis_service + return TestClient(app) + + +class TestIngestAnalysis: + def test_ingest_success(self, client: TestClient) -> None: + resp = client.post("/api/ingestion/job-1") + assert resp.status_code == 200 + data = resp.json() + assert data["docId"] == "doc-1" + assert data["chunksIndexed"] == 5 + assert data["embeddingDimension"] == 384 + + def test_ingest_not_found(self, client: TestClient, mock_analysis_service: AsyncMock) -> None: + mock_analysis_service.find_by_id.return_value = None + resp = client.post("/api/ingestion/missing") + assert resp.status_code == 404 + + def test_ingest_not_completed( + self, client: TestClient, mock_analysis_service: AsyncMock + ) -> None: + job = AnalysisJob(document_id="doc-1") + mock_analysis_service.find_by_id.return_value = job + resp = client.post("/api/ingestion/job-1") + assert resp.status_code == 400 + + def test_ingest_no_chunks(self, client: TestClient, mock_analysis_service: AsyncMock) -> None: + job = AnalysisJob(document_id="doc-1") + job.mark_running() + job.mark_completed(markdown="x", html="x", pages_json="[]") + mock_analysis_service.find_by_id.return_value = job + resp = client.post("/api/ingestion/job-1") + assert resp.status_code == 400 + + +class TestDeleteIngested: + def test_delete_success(self, client: TestClient) -> None: + resp = client.delete("/api/ingestion/doc-1") + assert resp.status_code == 204 + + +class TestIngestionStatus: + def test_available(self, client: TestClient) -> None: + resp = client.get("/api/ingestion/status") + assert resp.status_code == 200 + assert resp.json()["available"] is True + + def test_not_available(self) -> None: + app = FastAPI() + app.include_router(router) + app.state.ingestion_service = None + app.state.analysis_service = AsyncMock() + tc = TestClient(app) + resp = tc.get("/api/ingestion/status") + assert resp.status_code == 200 + assert resp.json()["available"] is False + + +class TestIngestionDisabled: + def test_returns_503_when_disabled(self) -> None: + app = FastAPI() + app.include_router(router) + app.state.ingestion_service = None + app.state.analysis_service = AsyncMock() + tc = TestClient(app) + resp = tc.post("/api/ingestion/job-1") + assert resp.status_code == 503 diff --git a/document-parser/tests/test_ingestion_service.py b/document-parser/tests/test_ingestion_service.py new file mode 100644 index 0000000..cbea903 --- /dev/null +++ b/document-parser/tests/test_ingestion_service.py @@ -0,0 +1,150 @@ +"""Tests for the ingestion service (services.ingestion_service).""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock + +import pytest + +from services.ingestion_service import IngestionConfig, IngestionService + + +def _make_chunks_json(count: int = 3, *, with_deleted: bool = False) -> str: + chunks = [] + for i in range(count): + chunk = { + "text": f"chunk text {i}", + "headings": [f"Heading {i}"], + "sourcePage": i + 1, + "tokenCount": 10, + "bboxes": [{"page": i + 1, "bbox": [0.0, 0.0, 100.0, 50.0]}], + } + if with_deleted and i == count - 1: + chunk["deleted"] = True + chunks.append(chunk) + return json.dumps(chunks) + + +@pytest.fixture +def mock_embedding() -> AsyncMock: + svc = AsyncMock() + svc.embed.return_value = [[0.1, 0.2, 0.3]] * 3 + return svc + + +@pytest.fixture +def mock_vector_store() -> AsyncMock: + store = AsyncMock() + store.ensure_index.return_value = None + store.delete_document.return_value = 0 + store.index_chunks.return_value = 3 + return store + + +@pytest.fixture +def service(mock_embedding: AsyncMock, mock_vector_store: AsyncMock) -> IngestionService: + return IngestionService( + embedding_service=mock_embedding, + vector_store=mock_vector_store, + config=IngestionConfig(index_name="test-idx", embedding_dimension=3), + ) + + +class TestIngest: + async def test_full_pipeline( + self, service: IngestionService, mock_embedding: AsyncMock, mock_vector_store: AsyncMock + ) -> None: + result = await service.ingest("doc-1", "test.pdf", _make_chunks_json(3)) + + assert result.doc_id == "doc-1" + assert result.chunks_indexed == 3 + mock_embedding.embed.assert_awaited_once() + texts = mock_embedding.embed.call_args[0][0] + assert len(texts) == 3 + mock_vector_store.ensure_index.assert_awaited_once() + mock_vector_store.delete_document.assert_awaited_once_with("test-idx", "doc-1") + mock_vector_store.index_chunks.assert_awaited_once() + indexed = mock_vector_store.index_chunks.call_args[0][1] + assert len(indexed) == 3 + assert indexed[0].doc_id == "doc-1" + assert indexed[0].filename == "test.pdf" + assert indexed[0].embedding == [0.1, 0.2, 0.3] + + async def test_skips_deleted_chunks( + self, service: IngestionService, mock_embedding: AsyncMock, mock_vector_store: AsyncMock + ) -> None: + mock_embedding.embed.return_value = [[0.1, 0.2, 0.3]] * 2 + mock_vector_store.index_chunks.return_value = 2 + result = await service.ingest("doc-1", "test.pdf", _make_chunks_json(3, with_deleted=True)) + + assert result.chunks_indexed == 2 + texts = mock_embedding.embed.call_args[0][0] + assert len(texts) == 2 + + async def test_empty_chunks( + self, service: IngestionService, mock_embedding: AsyncMock, mock_vector_store: AsyncMock + ) -> None: + result = await service.ingest("doc-1", "test.pdf", json.dumps([])) + assert result.chunks_indexed == 0 + mock_embedding.embed.assert_not_awaited() + + async def test_idempotent_deletes_old( + self, service: IngestionService, mock_vector_store: AsyncMock + ) -> None: + mock_vector_store.delete_document.return_value = 5 + await service.ingest("doc-1", "test.pdf", _make_chunks_json(3)) + mock_vector_store.delete_document.assert_awaited_once_with("test-idx", "doc-1") + + async def test_bbox_conversion( + self, service: IngestionService, mock_embedding: AsyncMock, mock_vector_store: AsyncMock + ) -> None: + mock_embedding.embed.return_value = [[0.1, 0.2, 0.3]] + mock_vector_store.index_chunks.return_value = 1 + await service.ingest("doc-1", "test.pdf", _make_chunks_json(1)) + indexed = mock_vector_store.index_chunks.call_args[0][1] + bbox = indexed[0].bboxes[0] + assert bbox.x == 0.0 + assert bbox.y == 0.0 + assert bbox.w == 100.0 + assert bbox.h == 50.0 + + async def test_with_binary_hash( + self, service: IngestionService, mock_vector_store: AsyncMock + ) -> None: + mock_embedding = service._embedding + mock_embedding.embed.return_value = [[0.1]] * 1 + await service.ingest("doc-1", "test.pdf", _make_chunks_json(1), binary_hash="abc123") + indexed = mock_vector_store.index_chunks.call_args[0][1] + assert indexed[0].origin is not None + assert indexed[0].origin.binary_hash == "abc123" + + +class TestDeleteDocument: + async def test_delegates_to_vector_store( + self, service: IngestionService, mock_vector_store: AsyncMock + ) -> None: + mock_vector_store.delete_document.return_value = 3 + result = await service.delete_document("doc-1") + assert result == 3 + + +class TestSearch: + async def test_embeds_and_searches( + self, service: IngestionService, mock_embedding: AsyncMock, mock_vector_store: AsyncMock + ) -> None: + mock_embedding.embed.return_value = [[0.5, 0.6, 0.7]] + mock_vector_store.search_similar.return_value = [] + await service.search("test query", k=5) + mock_embedding.embed.assert_awaited_once_with(["test query"]) + mock_vector_store.search_similar.assert_awaited_once() + + +class TestEnsureIndex: + async def test_calls_vector_store( + self, service: IngestionService, mock_vector_store: AsyncMock + ) -> None: + await service.ensure_index() + mock_vector_store.ensure_index.assert_awaited_once() + call_args = mock_vector_store.ensure_index.call_args + assert call_args[0][0] == "test-idx" diff --git a/e2e/api/src/test/resources/ingestion/ingest-and-verify.feature b/e2e/api/src/test/resources/ingestion/ingest-and-verify.feature new file mode 100644 index 0000000..46ad687 --- /dev/null +++ b/e2e/api/src/test/resources/ingestion/ingest-and-verify.feature @@ -0,0 +1,59 @@ +@e2e @ingestion +Feature: Ingestion pipeline — PDF → chunks → embeddings → OpenSearch + + Background: + * url baseUrl + + Scenario: Upload PDF, analyze with chunking, ingest into OpenSearch, verify + + # Step 1: Check ingestion is available + Given path '/api/ingestion/status' + When method GET + Then status 200 + And match response.available == true + + # Step 2: Upload a PDF + Given path '/api/documents/upload' + And multipart file file = { read: 'classpath:common/data/generated/medium.pdf', filename: 'medium.pdf', contentType: 'application/pdf' } + When method POST + Then status 200 + * def docId = response.id + + # Step 3: Create analysis with chunking + Given path '/api/analyses' + And request { documentId: '#(docId)', pipelineOptions: { doOcr: true, tableMode: 'fast' }, chunkingOptions: { chunkerType: 'hybrid', maxTokens: 256 } } + When method POST + Then status 200 + * def jobId = response.id + + # Step 4: Poll until completed + Given path '/api/analyses', jobId + And retry until response.status == 'COMPLETED' || response.status == 'FAILED' + When method GET + Then status 200 + And match response.status == 'COMPLETED' + And match response.chunksJson == '#string' + + # Step 5: Trigger ingestion + Given path '/api/ingestion', jobId + When method POST + Then status 200 + And match response.docId == docId + And match response.chunksIndexed == '#number' + And assert response.chunksIndexed > 0 + And match response.embeddingDimension == '#number' + And assert response.embeddingDimension > 0 + + # Step 6: Cleanup — delete ingested data + Given path '/api/ingestion', docId + When method DELETE + Then status 204 + + # Step 7: Cleanup — delete analysis and document + Given path '/api/analyses', jobId + When method DELETE + Then status 204 + + Given path '/api/documents', docId + When method DELETE + Then status 204 diff --git a/embedding-service/Dockerfile b/embedding-service/Dockerfile index b7b8e04..7311e03 100644 --- a/embedding-service/Dockerfile +++ b/embedding-service/Dockerfile @@ -2,6 +2,8 @@ FROM python:3.12-slim WORKDIR /app +RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/* + # Install dependencies first (cache layer) COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt diff --git a/frontend/src/features/ingestion/api.test.ts b/frontend/src/features/ingestion/api.test.ts new file mode 100644 index 0000000..1035474 --- /dev/null +++ b/frontend/src/features/ingestion/api.test.ts @@ -0,0 +1,48 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { ingestAnalysis, deleteIngested, fetchIngestionStatus } from './api' + +const mockFetch = vi.fn() +vi.stubGlobal('fetch', mockFetch) + +beforeEach(() => { + mockFetch.mockReset() +}) + +describe('ingestAnalysis', () => { + it('posts to /api/ingestion/:jobId', async () => { + mockFetch.mockResolvedValue({ + ok: true, + status: 200, + json: () => Promise.resolve({ docId: 'doc-1', chunksIndexed: 5, embeddingDimension: 384 }), + }) + const result = await ingestAnalysis('job-1') + expect(mockFetch).toHaveBeenCalledWith( + '/api/ingestion/job-1', + expect.objectContaining({ method: 'POST' }), + ) + expect(result.chunksIndexed).toBe(5) + }) +}) + +describe('deleteIngested', () => { + it('deletes /api/ingestion/:docId', async () => { + mockFetch.mockResolvedValue({ ok: true, status: 204, json: () => Promise.resolve(null) }) + await deleteIngested('doc-1') + expect(mockFetch).toHaveBeenCalledWith( + '/api/ingestion/doc-1', + expect.objectContaining({ method: 'DELETE' }), + ) + }) +}) + +describe('fetchIngestionStatus', () => { + it('gets /api/ingestion/status', async () => { + mockFetch.mockResolvedValue({ + ok: true, + status: 200, + json: () => Promise.resolve({ available: true }), + }) + const result = await fetchIngestionStatus() + expect(result.available).toBe(true) + }) +}) diff --git a/frontend/src/features/ingestion/api.ts b/frontend/src/features/ingestion/api.ts new file mode 100644 index 0000000..4cdb345 --- /dev/null +++ b/frontend/src/features/ingestion/api.ts @@ -0,0 +1,25 @@ +import { apiFetch } from '../../shared/api/http' + +export interface IngestionResult { + docId: string + chunksIndexed: number + embeddingDimension: number +} + +export interface IngestionStatus { + available: boolean +} + +export function ingestAnalysis(jobId: string): Promise { + return apiFetch(`/api/ingestion/${jobId}`, { + method: 'POST', + }) +} + +export function deleteIngested(docId: string): Promise { + return apiFetch(`/api/ingestion/${docId}`, { method: 'DELETE' }) +} + +export function fetchIngestionStatus(): Promise { + return apiFetch('/api/ingestion/status') +} diff --git a/frontend/src/features/ingestion/index.ts b/frontend/src/features/ingestion/index.ts new file mode 100644 index 0000000..95916a3 --- /dev/null +++ b/frontend/src/features/ingestion/index.ts @@ -0,0 +1 @@ +export { useIngestionStore } from './store' diff --git a/frontend/src/features/ingestion/store.test.ts b/frontend/src/features/ingestion/store.test.ts new file mode 100644 index 0000000..b911903 --- /dev/null +++ b/frontend/src/features/ingestion/store.test.ts @@ -0,0 +1,66 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { setActivePinia, createPinia } from 'pinia' +import { useIngestionStore } from './store' +import * as api from './api' + +vi.mock('./api', () => ({ + fetchIngestionStatus: vi.fn(), + ingestAnalysis: vi.fn(), + deleteIngested: vi.fn(), +})) + +beforeEach(() => { + setActivePinia(createPinia()) + vi.clearAllMocks() +}) + +describe('useIngestionStore', () => { + describe('checkAvailability', () => { + it('sets available to true when API responds', async () => { + vi.mocked(api.fetchIngestionStatus).mockResolvedValue({ available: true }) + const store = useIngestionStore() + await store.checkAvailability() + expect(store.available).toBe(true) + }) + + it('sets available to false on error', async () => { + vi.mocked(api.fetchIngestionStatus).mockRejectedValue(new Error('fail')) + const store = useIngestionStore() + await store.checkAvailability() + expect(store.available).toBe(false) + }) + }) + + describe('ingest', () => { + it('calls API and tracks ingested doc', async () => { + vi.mocked(api.ingestAnalysis).mockResolvedValue({ + docId: 'doc-1', + chunksIndexed: 5, + embeddingDimension: 384, + }) + const store = useIngestionStore() + const result = await store.ingest('job-1') + expect(result?.chunksIndexed).toBe(5) + expect(store.ingestedDocs['doc-1']).toBe(5) + expect(store.ingesting).toBe(false) + }) + + it('sets error on failure', async () => { + vi.mocked(api.ingestAnalysis).mockRejectedValue(new Error('fail')) + const store = useIngestionStore() + const result = await store.ingest('job-1') + expect(result).toBeNull() + expect(store.error).toBe('fail') + }) + }) + + describe('deleteIngested', () => { + it('removes doc from tracked map', async () => { + vi.mocked(api.deleteIngested).mockResolvedValue(null) + const store = useIngestionStore() + store.ingestedDocs['doc-1'] = 5 + await store.deleteIngested('doc-1') + expect(store.ingestedDocs['doc-1']).toBeUndefined() + }) + }) +}) diff --git a/frontend/src/features/ingestion/store.ts b/frontend/src/features/ingestion/store.ts new file mode 100644 index 0000000..4fbcd14 --- /dev/null +++ b/frontend/src/features/ingestion/store.ts @@ -0,0 +1,56 @@ +import { defineStore } from 'pinia' +import { ref } from 'vue' +import * as api from './api' + +export const useIngestionStore = defineStore('ingestion', () => { + const available = ref(false) + const ingesting = ref(false) + const error = ref(null) + /** Map of docId → chunks indexed count (tracks which docs are ingested) */ + const ingestedDocs = ref>({}) + + async function checkAvailability(): Promise { + try { + const status = await api.fetchIngestionStatus() + available.value = status.available + } catch { + available.value = false + } + } + + async function ingest(jobId: string): Promise { + ingesting.value = true + error.value = null + try { + const result = await api.ingestAnalysis(jobId) + ingestedDocs.value[result.docId] = result.chunksIndexed + return result + } catch (e) { + error.value = (e as Error).message || 'Ingestion failed' + console.error('Ingestion failed', e) + return null + } finally { + ingesting.value = false + } + } + + async function deleteIngested(docId: string): Promise { + try { + await api.deleteIngested(docId) + delete ingestedDocs.value[docId] + } catch (e) { + error.value = (e as Error).message || 'Failed to delete ingested data' + console.error('Failed to delete ingested data', e) + } + } + + return { + available, + ingesting, + error, + ingestedDocs, + checkAvailability, + ingest, + deleteIngested, + } +}) diff --git a/frontend/src/pages/DocumentsPage.vue b/frontend/src/pages/DocumentsPage.vue index 9dd2351..3256bff 100644 --- a/frontend/src/pages/DocumentsPage.vue +++ b/frontend/src/pages/DocumentsPage.vue @@ -2,13 +2,40 @@

{{ t('nav.documents') }}

+
+ +
+ +
+
+ + +
+
-
+
{{ t('history.emptyDocs') }}
-
+
- +
+ + {{ t('ingestion.indexed') }} + {{ ingestionStore.ingestedDocs[doc.id] }} + + + {{ t('ingestion.notIndexed') }} + + + +
@@ -42,21 +93,75 @@ @@ -72,6 +177,11 @@ onMounted(() => { padding: 16px 24px; border-bottom: 1px solid var(--border); flex-shrink: 0; + display: flex; + align-items: center; + justify-content: space-between; + gap: 16px; + flex-wrap: wrap; } .page-title { @@ -80,6 +190,57 @@ onMounted(() => { color: var(--text); } +.header-actions { + display: flex; + align-items: center; + gap: 12px; +} + +.search-input { + padding: 6px 12px; + border: 1px solid var(--border); + border-radius: var(--radius-sm); + background: var(--bg); + color: var(--text); + font-size: 13px; + width: 180px; + outline: none; + transition: border-color var(--transition); +} + +.search-input:focus { + border-color: var(--accent); +} + +.filter-group, +.sort-group { + display: flex; + gap: 2px; + background: var(--bg-surface); + border-radius: var(--radius-sm); + padding: 2px; + border: 1px solid var(--border); +} + +.filter-btn, +.sort-btn { + padding: 4px 10px; + border: none; + background: none; + color: var(--text-secondary); + font-size: 12px; + font-weight: 500; + border-radius: 4px; + cursor: pointer; + transition: all var(--transition); +} + +.filter-btn.active, +.sort-btn.active { + background: var(--accent); + color: white; +} + .page-content { flex: 1; overflow-y: auto; @@ -152,7 +313,41 @@ onMounted(() => { font-family: 'IBM Plex Mono', monospace; } -.doc-row-delete { +.doc-row-actions { + display: flex; + align-items: center; + gap: 8px; + flex-shrink: 0; +} + +.status-badge { + display: inline-flex; + align-items: center; + gap: 4px; + padding: 3px 8px; + border-radius: 10px; + font-size: 11px; + font-weight: 600; + text-transform: uppercase; + letter-spacing: 0.03em; +} + +.status-badge.indexed { + background: rgba(34, 197, 94, 0.15); + color: var(--success); +} + +.status-badge.not-indexed { + background: rgba(156, 163, 175, 0.15); + color: var(--text-muted); +} + +.badge-count { + font-family: 'IBM Plex Mono', monospace; + font-size: 10px; +} + +.action-btn { background: none; border: none; padding: 6px; @@ -164,14 +359,21 @@ onMounted(() => { transition: all var(--transition); } -.doc-row:hover .doc-row-delete { +.doc-row:hover .action-btn { opacity: 1; } -.doc-row-delete:hover { + +.action-btn:hover { + color: var(--accent); + background: rgba(249, 115, 22, 0.1); +} + +.action-btn.delete:hover { color: var(--error); background: rgba(239, 68, 68, 0.1); } -.doc-row-delete svg { + +.action-btn svg { width: 16px; height: 16px; } diff --git a/frontend/src/pages/StudioPage.vue b/frontend/src/pages/StudioPage.vue index e96769b..5ef26f9 100644 --- a/frontend/src/pages/StudioPage.vue +++ b/frontend/src/pages/StudioPage.vue @@ -94,6 +94,23 @@ {{ analysisStore.running ? t('studio.analyzing') : t('studio.run') }} +
@@ -459,6 +476,7 @@ import { ref, computed, watch, nextTick, onMounted, onBeforeUnmount, reactive } import { useRoute, useRouter } from 'vue-router' import { useDocumentStore } from '../features/document/store' import { useAnalysisStore } from '../features/analysis/store' +import { useIngestionStore } from '../features/ingestion/store' import { DocumentUpload, DocumentList } from '../features/document/index' import { ResultTabs } from '../features/analysis/index' import BboxOverlay from '../features/analysis/ui/BboxOverlay.vue' @@ -472,6 +490,7 @@ const route = useRoute() const router = useRouter() const documentStore = useDocumentStore() const analysisStore = useAnalysisStore() +const ingestionStore = useIngestionStore() const { t } = useI18n() const chunkingEnabled = useFeatureFlag('chunking') @@ -528,6 +547,14 @@ const pipelineOptions = reactive({ images_scale: 1.0, }) +const canIngest = computed(() => { + return ( + ingestionStore.available && + analysisStore.currentAnalysis?.status === 'COMPLETED' && + analysisStore.currentAnalysis?.chunksJson != null + ) +}) + const hasAnalysisResults = computed(() => { return ( analysisStore.currentAnalysis?.status === 'COMPLETED' && analysisStore.currentPages?.length > 0 @@ -564,6 +591,11 @@ async function runAnalysis() { await analysisStore.run(documentStore.selectedId, { ...pipelineOptions }) } +async function runIngestion() { + if (!analysisStore.currentAnalysis?.id) return + await ingestionStore.ingest(analysisStore.currentAnalysis.id) +} + function addMore() { documentStore.selectedId = null } @@ -598,6 +630,7 @@ watch( onMounted(async () => { await documentStore.load() analysisStore.load() + ingestionStore.checkAvailability() // Restore analysis from history via query param const analysisId = route.query.analysisId @@ -811,6 +844,21 @@ onBeforeUnmount(() => { cursor: not-allowed; } +.topbar-btn.ingest { + background: var(--success); + border-color: var(--success); + color: white; +} + +.topbar-btn.ingest:hover:not(:disabled) { + filter: brightness(1.1); +} + +.topbar-btn.ingest:disabled { + opacity: 0.6; + cursor: not-allowed; +} + .topbar-btn .btn-icon { width: 16px; height: 16px; diff --git a/frontend/src/shared/i18n.ts b/frontend/src/shared/i18n.ts index d56832e..8a20d57 100644 --- a/frontend/src/shared/i18n.ts +++ b/frontend/src/shared/i18n.ts @@ -131,6 +131,23 @@ const messages: Messages = { 'chunking.batchNotice': 'Le chunking n\u2019est pas disponible pour cette analyse. Les documents volumineux trait\u00e9s par batch ne g\u00e9n\u00e8rent pas la structure interne n\u00e9cessaire au d\u00e9coupage.', + // Ingestion / My Documents + 'ingestion.ingest': 'Ingérer', + 'ingestion.ingesting': 'Ingestion...', + 'ingestion.reindex': 'Ré-indexer', + 'ingestion.indexed': 'Indexé', + 'ingestion.notIndexed': 'Non indexé', + 'ingestion.chunksIndexed': '{n} chunks indexés', + 'ingestion.openInStudio': 'Ouvrir dans le Studio', + 'ingestion.deleteIndex': "Supprimer de l'index", + 'ingestion.unavailable': 'Ingestion non disponible', + 'ingestion.filterAll': 'Tous', + 'ingestion.filterIndexed': 'Indexés', + 'ingestion.filterNotIndexed': 'Non indexés', + 'ingestion.sortName': 'Nom', + 'ingestion.sortDate': 'Date', + 'ingestion.search': 'Rechercher...', + // Pagination 'pagination.pageOf': 'Page {current} sur {total}', 'pagination.perPage': '/ page', @@ -266,6 +283,22 @@ const messages: Messages = { 'chunking.batchNotice': 'Chunking is not available for this analysis. Large documents processed in batch mode do not generate the internal structure required for chunking.', + 'ingestion.ingest': 'Ingest', + 'ingestion.ingesting': 'Ingesting...', + 'ingestion.reindex': 'Re-index', + 'ingestion.indexed': 'Indexed', + 'ingestion.notIndexed': 'Not indexed', + 'ingestion.chunksIndexed': '{n} chunks indexed', + 'ingestion.openInStudio': 'Open in Studio', + 'ingestion.deleteIndex': 'Remove from index', + 'ingestion.unavailable': 'Ingestion unavailable', + 'ingestion.filterAll': 'All', + 'ingestion.filterIndexed': 'Indexed', + 'ingestion.filterNotIndexed': 'Not indexed', + 'ingestion.sortName': 'Name', + 'ingestion.sortDate': 'Date', + 'ingestion.search': 'Search...', + 'pagination.pageOf': 'Page {current} of {total}', 'pagination.perPage': '/ page',