refactor(indexing): add a queue for indexing operations (#237)

Author: philwinder

src/kodit/app.py

@@ -9,6 +9,7 @@
 
 from kodit._version import version
 from kodit.application.services.auto_indexing_service import AutoIndexingService
+from kodit.application.services.indexing_worker_service import IndexingWorkerService
 from kodit.application.services.sync_scheduler import SyncSchedulerService
 from kodit.config import AppContext
 from kodit.infrastructure.api.v1.routers import indexes_router, search_router
@@ -28,9 +29,16 @@ async def app_lifespan(_: FastAPI) -> AsyncIterator[AppLifespanState]:
 
     # App context has already been configured by the CLI.
     app_context = AppContext()
+    db = await app_context.get_db()
+
+    # Start the queue worker service
+    _indexing_worker_service = IndexingWorkerService(
+        app_context=app_context,
+        session_factory=db.session_factory,
+    )
+    await _indexing_worker_service.start()
 
     # Start auto-indexing service
-    db = await app_context.get_db()
     _auto_indexing_service = AutoIndexingService(
         app_context=app_context,
         session_factory=db.session_factory,
@@ -40,7 +48,6 @@ async def app_lifespan(_: FastAPI) -> AsyncIterator[AppLifespanState]:
     # Start sync scheduler service
     if app_context.periodic_sync.enabled:
         _sync_scheduler_service = SyncSchedulerService(
-            app_context=app_context,
             session_factory=db.session_factory,
         )
         _sync_scheduler_service.start_periodic_sync(
@@ -54,6 +61,8 @@ async def app_lifespan(_: FastAPI) -> AsyncIterator[AppLifespanState]:
         await _sync_scheduler_service.stop_periodic_sync()
     if _auto_indexing_service:
         await _auto_indexing_service.stop()
+    if _indexing_worker_service:
+        await _indexing_worker_service.stop()
 
 
 # See https://gofastmcp.com/integrations/fastapi#mounting-an-mcp-server

src/kodit/application/services/auto_indexing_service.py

@@ -1,6 +1,7 @@
 """Service for automatically indexing configured sources."""
 
 import asyncio
+import warnings
 from collections.abc import Callable
 from contextlib import suppress
 
@@ -10,8 +11,10 @@
 from kodit.application.factories.code_indexing_factory import (
     create_code_indexing_application_service,
 )
+from kodit.application.services.queue_service import QueueService
 from kodit.config import AppContext
-from kodit.infrastructure.ui.progress import create_log_progress_callback
+from kodit.domain.entities import Task
+from kodit.domain.value_objects import QueuePriority
 
 
 class AutoIndexingService:
@@ -37,13 +40,20 @@ async def start_background_indexing(self) -> None:
             self.log.info("Auto-indexing is disabled (no sources configured)")
             return
 
+        warnings.warn(
+            "Auto-indexing is deprecated and will be removed in a future version, please use the API to index sources.",  # noqa: E501
+            DeprecationWarning,
+            stacklevel=2,
+        )
+
         auto_sources = [source.uri for source in self.app_context.auto_indexing.sources]
         self.log.info("Starting background indexing", num_sources=len(auto_sources))
         self._indexing_task = asyncio.create_task(self._index_sources(auto_sources))
 
     async def _index_sources(self, sources: list[str]) -> None:
         """Index all configured sources in the background."""
         async with self.session_factory() as session:
+            queue_service = QueueService(session=session)
             service = create_code_indexing_application_service(
                 app_context=self.app_context,
                 session=session,
@@ -56,18 +66,17 @@ async def _index_sources(self, sources: list[str]) -> None:
                         self.log.info("Index already exists, skipping", source=source)
                         continue
 
-                    self.log.info("Auto-indexing source", source=source)
+                    self.log.info("Adding auto-indexing task to queue", source=source)
 
                     # Create index
                     index = await service.create_index_from_uri(source)
 
-                    # Run indexing (without progress callback for background mode)
-                    await service.run_index(
-                        index, progress_callback=create_log_progress_callback()
+                    await queue_service.enqueue_task(
+                        Task.create_index_update_task(
+                            index.id, QueuePriority.BACKGROUND
+                        )
                     )
 
-                    self.log.info("Successfully auto-indexed source", source=source)
-
                 except Exception as exc:
                     self.log.exception(
                         "Failed to auto-index source", source=source, error=str(exc)

src/kodit/application/services/indexing_worker_service.py

@@ -0,0 +1,154 @@
+"""Service for processing indexing tasks from the database queue."""
+
+import asyncio
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import suppress
+from datetime import UTC, datetime
+
+import structlog
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from kodit.application.factories.code_indexing_factory import (
+    create_code_indexing_application_service,
+)
+from kodit.config import AppContext
+from kodit.domain.entities import Task
+from kodit.domain.value_objects import TaskType
+from kodit.infrastructure.sqlalchemy.task_repository import SqlAlchemyTaskRepository
+
+
+class IndexingWorkerService:
+    """Service for processing indexing tasks from the database queue.
+
+    This worker polls the database for pending tasks and processes the heavy
+    indexing work in separate threads to prevent blocking API responsiveness.
+    """
+
+    def __init__(
+        self,
+        app_context: AppContext,
+        session_factory: Callable[[], AsyncSession],
+    ) -> None:
+        """Initialize the indexing worker service."""
+        self.app_context = app_context
+        self.session_factory = session_factory
+        self._worker_task: asyncio.Task | None = None
+        self._shutdown_event = asyncio.Event()
+        self._executor = ThreadPoolExecutor(
+            max_workers=1, thread_name_prefix="indexing-worker"
+        )
+        self.log = structlog.get_logger(__name__)
+
+    async def start(self) -> None:
+        """Start the worker to process the queue."""
+        self._running = True
+
+        # Start single worker task
+        self._worker_task = asyncio.create_task(self._worker_loop())
+
+        self.log.info(
+            "Indexing worker started",
+        )
+
+    async def stop(self) -> None:
+        """Stop the worker gracefully."""
+        self.log.info("Stopping indexing worker")
+        self._shutdown_event.set()
+
+        if self._worker_task and not self._worker_task.done():
+            self._worker_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await self._worker_task
+
+        # Shutdown the thread pool executor
+        self._executor.shutdown(wait=True)
+
+        self.log.info("Indexing worker stopped")
+
+    async def _worker_loop(self) -> None:
+        self.log.debug("Worker loop started")
+
+        while not self._shutdown_event.is_set():
+            try:
+                async with self.session_factory() as session:
+                    repo = SqlAlchemyTaskRepository(session)
+                    task = await repo.take()
+                    await session.commit()
+
+                # If there's a task, process it in a new thread
+                if task:
+                    await asyncio.get_event_loop().run_in_executor(
+                        self._executor, self._process_task, task
+                    )
+                    continue
+
+                # If no task, sleep for a bit
+                await asyncio.sleep(1)
+                continue
+
+            except Exception as e:
+                self.log.exception(
+                    "Error processing task",
+                    error=str(e),
+                )
+                continue
+
+        self.log.info("Worker loop stopped")
+
+    def _process_task(self, task: Task) -> None:
+        """Process a single task."""
+        self.log.info(
+            "Processing task",
+            task_id=task.id,
+            task_type=task.type.value,
+        )
+
+        start_time = datetime.now(UTC)
+
+        # Create a new event loop for this thread
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        try:
+            # Process based on task type (currently only INDEX_UPDATE is supported)
+            if task.type is TaskType.INDEX_UPDATE:
+                loop.run_until_complete(self._process_index_update(task))
+            else:
+                self.log.warning(
+                    "Unknown task type",
+                    task_id=task.id,
+                    task_type=task.type,
+                )
+                return
+        finally:
+            loop.close()
+
+        duration = (datetime.now(UTC) - start_time).total_seconds()
+        self.log.info(
+            "Task completed successfully",
+            task_id=task.id,
+            duration_seconds=duration,
+        )
+
+    async def _process_index_update(self, task: Task) -> None:
+        """Process index update/sync task."""
+        index_id = task.payload.get("index_id")
+        if not index_id:
+            raise ValueError("Missing index_id in task payload")
+
+        # Create a fresh database connection for this thread's event loop
+        db = await self.app_context.new_db(run_migrations=True)
+        try:
+            async with db.session_factory() as session:
+                service = create_code_indexing_application_service(
+                    app_context=self.app_context,
+                    session=session,
+                )
+                index = await service.index_repository.get(index_id)
+                if not index:
+                    raise ValueError(f"Index not found: {index_id}")
+
+                await service.run_index(index)
+        finally:
+            await db.close()

src/kodit/application/services/queue_service.py

@@ -0,0 +1,52 @@
+"""Queue service for managing tasks."""
+
+import structlog
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from kodit.domain.entities import Task
+from kodit.domain.value_objects import TaskType
+from kodit.infrastructure.sqlalchemy.task_repository import SqlAlchemyTaskRepository
+
+
+class QueueService:
+    """Service for queue operations using database persistence.
+
+    This service provides the main interface for enqueuing and managing tasks.
+    It uses the existing Task entity in the database with a flexible JSON payload.
+    """
+
+    def __init__(
+        self,
+        session: AsyncSession,
+    ) -> None:
+        """Initialize the queue service."""
+        self.session = session
+        self.log = structlog.get_logger(__name__)
+
+    async def enqueue_task(self, task: Task) -> None:
+        """Queue a task in the database."""
+        repo = SqlAlchemyTaskRepository(self.session)
+
+        # See if task already exists
+        db_task = await repo.get(task.id)
+        if db_task:
+            # Task already exists, update priority
+            db_task.priority = task.priority
+            await repo.update(db_task)
+            self.log.info("Task updated", task_id=task.id, task_type=task.type)
+        else:
+            # Otherwise, add task
+            await repo.add(task)
+            self.log.info(
+                "Task queued",
+                task_id=task.id,
+                task_type=task.type,
+                payload=task.payload,
+            )
+
+        await self.session.commit()
+
+    async def list_tasks(self, task_type: TaskType | None = None) -> list[Task]:
+        """List all tasks in the queue."""
+        repo = SqlAlchemyTaskRepository(self.session)
+        return await repo.list(task_type)