Merge branch health-check into alt

Dockerfile

@@ -28,5 +28,10 @@ USER appuser
 # Expose a volume mount for logs ― Hugging Face Spaces requires specifically /data.
 VOLUME /data
 
+# Temporary block for the health server fix:
+COPY scripts/run.sh ./scripts/
+COPY temporary_health_check_server.py ./
+CMD ["./scripts/run.sh"]
+
 # Run the application.
-CMD ["python", "-m", "ctp_slack_bot.app"]
+#CMD ["python", "-m", "ctp_slack_bot.app"]

README.md

@@ -7,6 +7,7 @@ sdk: docker
 pinned: false
 license: mit
 short_description: Spring 2025 CTP Slack Bot RAG system
+app_port: 8080
 ---
 
 

scripts/run.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+parent_path=$(cd "$(dirname "${BASH_SOURCE[0]}")"; pwd -P)
+
+cd "${parent_path}/.."
+
+python "temporary_health_check_server.py" & python -m ctp_slack_bot.app

src/ctp_slack_bot/containers.py

@@ -7,7 +7,7 @@ from slack_bolt.async_app import AsyncApp
 
 from ctp_slack_bot.core.config import Settings
 from ctp_slack_bot.db.mongo_db import MongoDBResource
-from ctp_slack_bot.db.repositories.mongo_db_vectorized_chunk_repository import MongoVectorizedChunkRepository
+from ctp_slack_bot.db.repositories.mongo_db_vectorized_chunk_repository import MongoVectorizedChunkRepositoryResource
 from ctp_slack_bot.mime_type_handlers.base import MimeTypeHandlerMeta
 from ctp_slack_bot.services.answer_retrieval_service import AnswerRetrievalService
 from ctp_slack_bot.services.content_ingestion_service import ContentIngestionService
@@ -34,13 +34,13 @@ def __load_plugins(plugin_dir) -> None:
 __load_plugins("ctp_slack_bot/mime_type_handlers")
 
 
-class Container(DeclarativeContainer):
+class Container(DeclarativeContainer): # TODO: audit for potential async-related bugs.
     settings = Singleton(Settings)
     event_brokerage_service = Singleton(EventBrokerageService)
     schedule_service = Resource(ScheduleServiceResource, settings=settings)
     mongo_db = Resource(MongoDBResource, settings=settings) # TODO: generalize to any database.
-    vectorized_chunk_repository = Singleton(MongoVectorizedChunkRepository, mongo_db=mongo_db)
-    vector_database_service = Singleton(VectorDatabaseService, settings=settings, mongo_db=mongo_db)
+    vectorized_chunk_repository = Resource(MongoVectorizedChunkRepositoryResource, settings=settings, mongo_db=mongo_db)
+    vector_database_service = Singleton(VectorDatabaseService, settings=settings, vectorized_chunk_repository=vectorized_chunk_repository)
     embeddings_model_service = Singleton(EmbeddingsModelService, settings=settings)
     vectorization_service = Singleton(VectorizationService, settings=settings, embeddings_model_service=embeddings_model_service)
     content_ingestion_service = Singleton(ContentIngestionService, settings=settings, event_brokerage_service=event_brokerage_service, vector_database_service=vector_database_service, vectorization_service=vectorization_service)

src/ctp_slack_bot/db/mongo_db.py

@@ -1,15 +1,14 @@
-from asyncio import create_task
 from dependency_injector.resources import AsyncResource
-from motor.motor_asyncio import AsyncIOMotorClient
+from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
 from pymongo.errors import ConnectionFailure, ServerSelectionTimeoutError
-from pymongo.operations import SearchIndexModel
 from loguru import logger
 from pydantic import BaseModel, PrivateAttr
-from typing import Any, Dict, Optional, Self
+from typing import Any, Dict, Self
 
 from ctp_slack_bot.core.config import Settings
 from ctp_slack_bot.utils import sanitize_mongo_db_uri
 
+
 class MongoDB(BaseModel):
     """
     MongoDB connection manager using Motor for async operations.
@@ -19,6 +18,7 @@ class MongoDB(BaseModel):
     _db: PrivateAttr = PrivateAttr()
 
     class Config:
+        frozen=True
         arbitrary_types_allowed = True
 
     def __init__(self: Self, **data: Dict[str, Any]) -> None:
@@ -31,7 +31,7 @@ class MongoDB(BaseModel):
             connection_string = self.settings.MONGODB_URI.get_secret_value()
             logger.debug("Connecting to MongoDB using URI: {}", sanitize_mongo_db_uri(connection_string))
 
-            # Create client with appropriate settings
+            # Create client with appropriate settings.
             self._client = AsyncIOMotorClient(
                 connection_string,
                 serverSelectionTimeoutMS=5000,
@@ -42,7 +42,7 @@ class MongoDB(BaseModel):
                 w="majority"
             )
 
-            # Set database
+            # Get the database name.
             db_name = self.settings.MONGODB_NAME
 
             self._db = self._client[db_name]
@@ -54,116 +54,50 @@ class MongoDB(BaseModel):
             self._db = None
             raise
 
-    @property
-    def client(self: Self) -> AsyncIOMotorClient:
-        """Get the MongoDB client instance."""
-        if not hasattr(self, '_client') or self._client is None:
-            logger.warning("MongoDB client not initialized. Attempting to initialize…")
-            self.connect()
-            if not hasattr(self, '_client') or self._client is None:
-                raise ConnectionError("Failed to initialize MongoDB client.")
-        return self._client
-
-    @property
-    def db(self: Self) -> Any:
-        """Get the MongoDB database instance."""
-        if not hasattr(self, '_db') or self._db is None:
-            logger.warning("MongoDB database not initialized. Attempting to initialize client…")
-            self.connect()
-            if not hasattr(self, '_db') or self._db is None:
-                raise ConnectionError("Failed to initialize MongoDB database.")
-        return self._db
-
     async def ping(self: Self) -> bool:
         """Check if MongoDB connection is alive."""
         try:
-            # Get client to ensure we're connected
-            client = self.client
-
-            # Try a simple ping command
-            await client.admin.command('ping')
+            await self._client.admin.command("ping")
             logger.debug("MongoDB connection is active!")
             return True
         except (ConnectionFailure, ServerSelectionTimeoutError) as e:
             logger.error("MongoDB connection failed: {}", e)
-            return False
         except Exception as e:
             logger.error("Unexpected error during MongoDB ping: {}", e)
-            return False
+        return False
 
-    async def get_collection(self: Self, name: str) -> Any:
+    async def get_collection(self: Self, name: str) -> AsyncIOMotorCollection:
         """
-        Get a collection by name with validation.
-        Creates the collection if it doesn't exist.
+        Get a collection by name or creates it if it doesn’t exist.
         """
-        # First ensure we can connect at all
+        # First ensure we can connect at all.
         if not await self.ping():
             logger.error("Cannot get collection '{}' because a MongoDB connection is not available.", name)
             raise ConnectionError("MongoDB connection is not available.")
 
         try:
-            # Get all collection names to check if this one exists
+            # Get all collection names to check if this one exists.
             logger.debug("Checking if collection '{}' exists…", name)
-            collection_names = await self.db.list_collection_names()
+            collection_names = await self._db.list_collection_names()
             
             if name not in collection_names:
                 logger.info("Collection '{}' does not exist. Creating it…", name)
-                # Create the collection
-                await self.db.create_collection(name)
+
+                # Create the collection.
+                await self._db.create_collection(name)
                 logger.debug("Successfully created collection: {}", name)
             else:
                 logger.debug("Collection '{}' already exists!", name)
 
-            # Get and return the collection
-            collection = self.db[name]
+            # Get and return the collection.
+            collection = self._db[name]
             return collection
         except Exception as e:
             logger.error("Error accessing collection '{}': {}", name, e)
             raise
 
-    async def create_indexes(self: Self, collection_name: str) -> None:
-        """
-        Create a vector search index on a collection.
-        
-        Args:
-            collection_name: Name of the collection
-        """
-        collection = await self.get_collection(collection_name)
-
-        try:
-            # Create search index model using MongoDB's recommended approach
-            search_index_model = SearchIndexModel(
-                definition={
-                    "fields": [
-                        {
-                            "type": "vector",
-                            "path": "embedding",
-                            "numDimensions": self.settings.VECTOR_DIMENSION,
-                            "similarity": "cosine",
-                            "quantization": "scalar"
-                        }
-                    ]
-                },
-                name=f"{collection_name}_vector_index",
-                type="vectorSearch"
-            )
-
-            # Create the search index using the motor collection
-            result = await collection.create_search_index(search_index_model)
-            logger.info("Vector search index '{}' created for collection {}.", result, collection_name)
-
-        except Exception as e:
-            if "command not found" in str(e).lower():
-                logger.warning("Vector search not supported by this MongoDB instance. Some functionality may be limited.")
-                # Create a fallback standard index on embedding field
-                await collection.create_index("embedding")
-                logger.info("Created standard index on 'embedding' field as fallback.")
-            else:
-                logger.error("Failed to create vector index: {}", e)
-                raise
-
-    async def close(self: Self) -> None:
-        """Close MongoDB connection."""
+    def close(self: Self) -> None:
+        """Close the MongoDB connection."""
         if self._client:
             self._client.close()
             logger.info("Closed MongoDB connection.")
@@ -193,6 +127,6 @@ class MongoDBResource(AsyncResource):
     async def shutdown(self: Self, mongo_db: MongoDB) -> None:
         """Close MongoDB connection on shutdown."""
         try:
-            await mongo_db.close()
+            mongo_db.close()
         except Exception as e:
             logger.error("Error closing MongoDB connection: {}", e)

src/ctp_slack_bot/db/repositories/__init__.py

@@ -1,2 +1,3 @@
 from ctp_slack_bot.db.repositories.mongo_db_vectorized_chunk_repository import MongoVectorizedChunkRepository
 from ctp_slack_bot.db.repositories.vectorized_chunk_repository import VectorizedChunkRepository
+from ctp_slack_bot.db.repositories.vector_repository_base import VectorRepositoryBase

src/ctp_slack_bot/db/repositories/mongo_db_vectorized_chunk_repository.py

@@ -1,65 +1,161 @@
-from typing import List, Optional, Dict, Any
-import pymongo
-from bson import ObjectId
+from dependency_injector.resources import AsyncResource
+from loguru import logger
+from pymongo import ASCENDING, ReturnDocument
+from typing import Any, Collection, Dict, Iterable, Mapping, Optional, Self, Sequence, Set
 
-from ctp_slack_bot.db import MongoDB
+from ctp_slack_bot.core import Settings
+from ctp_slack_bot.models import Chunk, VectorizedChunk, VectorQuery
+from ctp_slack_bot.db.mongo_db import MongoDB
 from ctp_slack_bot.db.repositories.vectorized_chunk_repository import VectorizedChunkRepository
-from ctp_slack_bot.models.base import VectorizedChunk
-
-class MongoVectorizedChunkRepository(VectorizedChunkRepository):
-    """MongoDB implementation of VectorizedChunkRepository."""
-    
-    def __init__(self, mongo_db: MongoDB):
-        self.mongo_db = mongo_db
-        self.collection = self.mongo_db.db.get_collection("vectorized_chunks")
-        
-        # Create indexes for efficient queries
-        self.collection.create_index("chunk_id")
-        self.collection.create_index("parent_id")
-        
-    async def find_by_id(self, id: str) -> Optional[VectorizedChunk]:
-        doc = await self.collection.find_one({"_id": ObjectId(id)})
-        return self._map_to_entity(doc) if doc else None
-    
-    async def find_all(self) -> List[VectorizedChunk]:
-        cursor = self.collection.find({})
-        return [self._map_to_entity(doc) async for doc in cursor]
-    
-    async def find_by_parent_id(self, parent_id: str) -> List[VectorizedChunk]:
+from ctp_slack_bot.db.repositories.vector_repository_base import VectorRepositoryBase
+
+
+class MongoVectorizedChunkRepository(VectorRepositoryBase, VectorizedChunkRepository):
+    """MongoDB implementation of VectorizedChunkRepository"""
+
+    def __init__(self: Self, **data: Dict[str, Any]) -> None:
+        super().__init__(**data)
+        logger.debug("Created {}", self.__class__.__name__)
+
+    async def count_by_id(self: Self, parent_id: str, chunk_id: Optional[str] = None) -> int:
+        if chunk_id is None:
+            return await self.collection.count_documents({"parent_id": parent_id})
+        else:
+            return await self.collection.count_documents({"parent_id": parent_id, "chunk_id": chunk_id})
+
+    async def find_all(self: Self) -> Collection[VectorizedChunk]:
+        cursor = self.collection.find()
+        return [VectorizedChunk(**document) async for document in cursor] # TODO: mutable until async support is extended to tuples
+
+    async def find_by_metadata(self: Self, metadata_query: Mapping[str, Any]) -> Collection[VectorizedChunk]:
+        query = {f"metadata.{key}": value for key, value in metadata_query.items()}
+        cursor = self.collection.find(query)
+        return [VectorizedChunk(**document) async for document in cursor] # TODO: mutable until async support is extended to tuples
+
+    async def find_by_parent_id(self: Self, parent_id: str) -> Collection[VectorizedChunk]:
         cursor = self.collection.find({"parent_id": parent_id})
-        return [self._map_to_entity(doc) async for doc in cursor]
-    
-    async def save(self, chunk: VectorizedChunk) -> VectorizedChunk:
-        doc = self._map_to_document(chunk)
+        return [VectorizedChunk(**document) async for document in cursor] # TODO: mutable until async support is extended to tuples
+
+    async def find_by_parent_and_chunk_ids(self: Self, parent_id: str, chunk_id: str) -> Optional[VectorizedChunk]:
+        document = await self.collection.find_one({"parent_id": parent_id, "chunk_id": chunk_id})
+        return VectorizedChunk(**document) if document else None
+
+    async def find_by_vector(self: Self, query_embedding: Sequence[float], k: int = 5, score_threshold: float = 0.7) -> Sequence[VectorizedChunk]:
+        pipeline = [
+            {
+                "$vectorSearch": {
+                    "index": "vector_index",
+                    "path": "embedding",
+                    "queryVector": query_embedding,
+                    "numCandidates": k * 2,
+                    "limit": k,
+                    "score": {"$meta": "vectorSearchScore"}
+                }
+            },
+            {"$match": {"score": {"$gte": score_threshold}}}
+        ]
+        cursor = self.collection.aggregate(pipeline)
+        return [VectorizedChunk(**document) async for document in cursor] # TODO: mutable until async support is extended to tuples
+
+    async def find_by_vector(self: Self, query: VectorQuery) -> Sequence[Chunk]:
+        """
+        Query the vector database for similar documents.
         
-        if "_id" in doc and doc["_id"]:
-            # Update existing document
-            await self.collection.replace_one({"_id": doc["_id"]}, doc)
+        Args:
+            query: VectorQuery object with search parameters
+     
+        Returns:
+            Sequence[Chunk]: List of similar chunks
+        """
+        # Build aggregation pipeline for vector search using official MongoDB format.
+        pipeline = [
+            {
+                "$vectorSearch": {
+                    "index": f"{self.collection.name}_vector_index",
+                    "path": "embedding",
+                    "queryVector": query.query_embeddings,
+                    "numCandidates": query.k * 10,
+                    "limit": query.k
+                }
+            },
+            {
+                "$project": {
+                    "text": 1,
+                    "metadata": 1,
+                    "parent_id": 1,
+                    "chunk_id": 1,
+                    "score": { "$meta": "vectorSearchScore" }
+                }
+            },
+            {
+                "$match": {
+                    "score": { "$gte": query.score_threshold }
+                }
+            }
+        ]
+        if query.filter_metadata: # Add metadata filters if provided.
+            metadata_filter = {f"metadata.{key}": value for key, value in query.filter_metadata.items()}
+            pipeline.insert(1, {"$match": metadata_filter})
+
+        # Execute the vector search pipeline.
+        results = await self.collection.aggregate(pipeline).to_list(length=query.k)
+
+        # Convert results to Chunk objects ― don’t care about the embeddings.
+        return tuple(Chunk(text=result["text"],
+                           parent_id=result["parent_id"],
+                           chunk_id=result["chunk_id"],
+                           metadata={**result["metadata"], "similarity_score": result.get("score", 0)})
+                     for result
+                     in results)
+
+    async def insert_one(self, chunk: VectorizedChunk) -> str:
+        document = chunk.model_dump()
+        result = await self.collection.insert_one(document)
+        return str(result.inserted_id)
+
+    async def insert_many(self, chunks: Iterable[VectorizedChunk]) -> Set[str]:
+        documents = [chunk.model_dump() for chunk in chunks]
+        result = await self.collection.insert_many(documents)
+        return frozenset(map(str, result.inserted_ids))
+
+    async def replace_all(self: Self, chunks: Iterable[VectorizedChunk]) -> Set[str]:
+        parent_ids = set()
+        documents = []
+        for chunk in chunks:
+            parent_ids.add(chunk.parent_id)
+            documents.append(chunk.model_dump())
+        async with await self.collection.database.client.start_session() as session:
+            async with session.start_transaction():
+                delete_result = await self.collection.delete_many({"parent_id": {"$in": tuple(parent_ids)}}, session=session)
+                insert_result = await self.collection.insert_many(documents, session=session)
+                return frozenset(map(str, insert_result.inserted_ids))
+
+    async def replace_one(self: Self, chunk: VectorizedChunk) -> str:
+        result = await self.collection.find_one_and_replace(
+            {"parent_id": chunk.parent_id, "chunk_id": chunk.chunk_id},
+            chunk.model_dump(),
+            upsert=True,
+            return_document=ReturnDocument.AFTER
+        )
+        return result["_id"]
+
+    async def delete(self: Self, parent_id: str, chunk_id: Optional[str] = None) -> int:
+        if chunk_id is not None:
+            result = await self.collection.delete_one({"parent_id": parent_id, "chunk_id": chunk_id})
         else:
-            # Insert new document
-            result = await self.collection.insert_one(doc)
-            doc["_id"] = result.inserted_id
-            
-        return self._map_to_entity(doc)
-    
-    async def delete(self, id: str) -> bool:
-        result = await self.collection.delete_one({"_id": ObjectId(id)})
-        return result.deleted_count > 0
-    
-    async def find_by_metadata(self, metadata_query: Dict[str, Any]) -> List[VectorizedChunk]:
-        # Convert the metadata query to MongoDB query format
-        query = {f"metadata.{k}": v for k, v in metadata_query.items()}
-        cursor = self.collection.find(query)
-        return [self._map_to_entity(doc) async for doc in cursor]
-    
-    def _map_to_document(self, chunk: VectorizedChunk) -> Dict[str, Any]:
-        """Convert a VectorizedChunk to a MongoDB document."""
-        doc = chunk.model_dump()
-        # Handle any special conversions needed
-        return doc
-    
-    def _map_to_entity(self, doc: Dict[str, Any]) -> VectorizedChunk:
-        """Convert a MongoDB document to a VectorizedChunk."""
-        if "_id" in doc:
-            doc["id"] = str(doc.pop("_id"))
-        return VectorizedChunk(**doc)
+            result = await self.collection.delete_many({"parent_id": parent_id})
+        return result.deleted_count
+
+    async def ensure_indices_exist(self: Self) -> None:
+        await super().ensure_indices_exist()
+        index_name = "parent_chunk_unique"
+        existing_indices = await self.collection.index_information()
+        if index_name not in existing_indices:
+            await self.collection.create_index([("parent_id", ASCENDING), ("chunk_id", ASCENDING)], unique=True, name=index_name)
+
+class MongoVectorizedChunkRepositoryResource(AsyncResource):
+    async def init(self: Self, settings: Settings, mongo_db: MongoDB) -> MongoVectorizedChunkRepository:
+        vectorized_chunk_collection = await mongo_db.get_collection("vectorized_chunks")
+        vectorized_chunk_repository = MongoVectorizedChunkRepository(settings=settings, collection=vectorized_chunk_collection)
+        await vectorized_chunk_repository.ensure_indices_exist()
+        return vectorized_chunk_repository

src/ctp_slack_bot/db/repositories/vector_repository_base.py

@@ -0,0 +1,62 @@
+from abc import ABC
+from loguru import logger
+from motor.motor_asyncio import AsyncIOMotorCollection
+from pydantic import BaseModel
+from pymongo.operations import SearchIndexModel
+from typing import Self
+
+from ctp_slack_bot.core import Settings
+
+class VectorRepositoryBase(ABC, BaseModel):
+    """MongoDB implementation of VectorizedChunkRepository"""
+
+    settings: Settings
+    collection: AsyncIOMotorCollection
+
+    class Config:
+        frozen=True
+        arbitrary_types_allowed = True
+
+    async def ensure_indices_exist(self: Self) -> None:
+        """Ensure that indices exist."""
+        await self.ensure_search_index_exists() 
+
+    async def ensure_search_index_exists(self: Self) -> None:
+        """
+        Ensure that a vector search index exists.
+        """
+        index_name = f"{self.collection.name}_vector_index"
+        try:
+            existing_indexes = [index["name"] async for index in self.collection.list_search_indexes()]
+            logger.debug("{} existing indices were found: {}", len(existing_indexes), existing_indexes)
+            if index_name in existing_indexes:
+                logger.debug("Index '{}' already exists; duplicate index will not be created.", index_name)
+                return
+
+            # Create search index model using MongoDB's recommended approach.
+            search_index_model = SearchIndexModel(
+                definition={
+                    "fields": [
+                        {
+                            "type": "vector",
+                            "path": "embedding",
+                            "numDimensions": self.settings.VECTOR_DIMENSION,
+                            "similarity": "cosine",
+                            "quantization": "scalar"
+                        }
+                    ]
+                },
+                name=index_name,
+                type="vectorSearch"
+            )
+            result = await self.collection.create_search_index(search_index_model)
+            logger.info("Vector search index '{}' created for collection {}.", result, self.collection.name)
+        except Exception as e:
+            if "command not found" in str(e).lower():
+                logger.warning("Vector search not supported by this MongoDB instance. Some functionality may be limited.")
+                # Create a fallback standard index on embedding field.
+                await self.collection.create_index("embedding")
+                logger.info("Created standard index on 'embedding' field as fallback.")
+            else:
+                logger.error("Failed to create vector index: {}", e)
+                raise

src/ctp_slack_bot/db/repositories/vectorized_chunk_repository.py

@@ -1,30 +1,52 @@
-from typing import List, Optional, Dict, Any
+from abc import ABC, abstractmethod
+from pydantic import BaseModel
+from typing import Any, Collection, Iterable, Mapping, Optional, Self, Sequence, Set
 
-from ctp_slack_bot.models.base import VectorizedChunk
+from ctp_slack_bot.models import Chunk, VectorizedChunk, VectorQuery
 
-class VectorizedChunkRepository:
+class VectorizedChunkRepository(ABC, BaseModel):
     """Repository interface for VectorizedChunk entities."""
 
-    async def find_by_id(self, id: str) -> Optional[VectorizedChunk]:
-        """Find a chunk by its ID."""
+    @abstractmethod
+    async def count_by_id(self: Self, parent_id: str, chunk_id: Optional[str] = None) -> int:
         pass
 
-    async def find_all(self) -> List[VectorizedChunk]:
-        """Find all chunks."""
+    @abstractmethod
+    async def find_all(self: Self) -> Collection[VectorizedChunk]:
         pass
 
-    async def find_by_parent_id(self, parent_id: str) -> List[VectorizedChunk]:
-        """Find chunks by parent document ID."""
+    @abstractmethod
+    async def find_by_metadata(self: Self, metadata_query: Mapping[str, Any]) -> Collection[VectorizedChunk]:
         pass
 
-    async def save(self, chunk: VectorizedChunk) -> VectorizedChunk:
-        """Save a chunk to the database."""
+    @abstractmethod
+    async def find_by_parent_id(self: Self, parent_id: str) -> Collection[VectorizedChunk]:
         pass
 
-    async def delete(self, id: str) -> bool:
-        """Delete a chunk by its ID."""
+    @abstractmethod
+    async def find_by_parent_and_chunk_ids(self: Self, parent_id: str, chunk_id: str) -> Optional[VectorizedChunk]:
         pass
 
-    async def find_by_metadata(self, metadata_query: Dict[str, Any]) -> List[VectorizedChunk]:
-        """Find chunks by metadata criteria."""
+    @abstractmethod
+    async def find_by_vector(self: Self, query: VectorQuery) -> Sequence[Chunk]:
+        pass
+
+    @abstractmethod
+    async def insert_one(self, chunk: VectorizedChunk) -> str:
+        pass
+
+    @abstractmethod
+    async def insert_many(self, chunks: Iterable[VectorizedChunk]) -> Set[str]:
+        pass
+
+    @abstractmethod
+    async def replace_all(self: Self, chunks: Iterable[VectorizedChunk]) -> Set[str]:
+        pass
+
+    @abstractmethod
+    async def replace_one(self: Self, chunk: VectorizedChunk) -> str:
+        pass
+
+    @abstractmethod
+    async def delete(self: Self, parent_id: str, chunk_id: Optional[str] = None) -> int:
         pass

src/ctp_slack_bot/models/base.py

@@ -1,6 +1,8 @@
 from abc import ABC, abstractmethod
-from pydantic import BaseModel, ConfigDict, Field
-from typing import Any, final, Mapping, Self, Sequence, Optional
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+from typing import Any, final, Mapping, Optional, Self
+
+from ctp_slack_bot.utils import to_deep_immutable
 
 
 class Chunk(BaseModel):
@@ -9,10 +11,15 @@ class Chunk(BaseModel):
     text: str                   # The text representation
     parent_id: str              # The source content’s identity
     chunk_id: str               # This chunk’s identity—unique within the source content
-    metadata: Mapping[str, Any]
+    metadata: Mapping[str, Any] = Field(default_factory=dict)
 
     model_config = ConfigDict(frozen=True)
 
+    @field_validator('metadata')
+    @classmethod
+    def __make_metadata_readonly(cls, value: Mapping[str, Any]) -> Mapping[str, Any]:
+        return to_deep_immutable(value)
+
 
 @final
 class VectorQuery(BaseModel):
@@ -25,19 +32,24 @@ class VectorQuery(BaseModel):
         filter_metadata: Optional filters for metadata fields
     """
 
-    query_embeddings: Sequence[float]
+    query_embeddings: tuple[float, ...]
     k: int
     score_threshold: float = Field(default=0.7)
-    filter_metadata: Optional[Mapping[str, Any]] = None
+    filter_metadata: Mapping[str, Any] = Field(default_factory=dict)
 
     model_config = ConfigDict(frozen=True)
 
+    @field_validator('filter_metadata')
+    @classmethod
+    def __make_metadata_readonly(cls, value: Mapping[str, Any]) -> Mapping[str, Any]:
+        return to_deep_immutable(value)
+
     
 @final
 class VectorizedChunk(Chunk):
     """A class representing a vectorized chunk of content."""
 
-    embedding: Sequence[float] # The vector representation
+    embedding: tuple[float, ...] # The vector representation
 
 
 class Content(ABC, BaseModel):
@@ -50,7 +62,7 @@ class Content(ABC, BaseModel):
         pass
 
     @abstractmethod
-    def get_chunks(self: Self) -> Sequence[Chunk]:
+    def get_chunks(self: Self) -> tuple[Chunk, ...]:
         pass
 
     @abstractmethod

src/ctp_slack_bot/models/slack.py

@@ -2,7 +2,7 @@ from datetime import datetime
 from json import dumps
 from pydantic import BaseModel, ConfigDict, PositiveInt, PrivateAttr
 from types import MappingProxyType
-from typing import Any, Dict, Literal, Mapping, Optional, Self, Sequence
+from typing import Any, Literal, Mapping, Optional, Self
 
 from ctp_slack_bot.models.base import Chunk, Content
 
@@ -23,7 +23,7 @@ class SlackEvent(BaseModel):
     type: str
     event_id: str
     event_time: int
-    authed_users: Sequence[str]
+    authed_users: tuple[str, ...]
 
     model_config = ConfigDict(frozen=True) 
 
@@ -40,7 +40,7 @@ class SlackReaction(BaseModel):
 
     name: str
     count: PositiveInt
-    users: Sequence[str]
+    users: tuple[str, ...]
 
     model_config = ConfigDict(frozen=True)
 
@@ -61,14 +61,14 @@ class SlackMessage(Content):
     deleted_ts: Optional[str] = None
     hidden: bool = False
     is_starred: Optional[bool] = None
-    pinned_to: Optional[Sequence[str]] = None
-    reactions: Optional[Sequence[SlackReaction]] = None
+    pinned_to: Optional[tuple[str, ...]] = None
+    reactions: Optional[tuple[SlackReaction, ...]] = None
 
     def get_id(self: Self) -> str:
         """Unique identifier for this message."""
         return f"slack-message:{self.channel}:{self.ts}"
 
-    def get_chunks(self: Self) -> Sequence[Chunk]:
+    def get_chunks(self: Self) -> tuple[Chunk]:
         return (Chunk(text=self.text, parent_id=self.get_id(), chunk_id="", metadata=self.get_metadata()), )
 
     def get_metadata(self: Self) -> Mapping[str, Any]:

src/ctp_slack_bot/models/webvtt.py

@@ -1,14 +1,13 @@
 from datetime import datetime, timedelta
 from io import BytesIO
-from itertools import starmap
-from json import dumps
 from more_itertools import windowed
-from pydantic import BaseModel, ConfigDict, Field, PositiveInt, PrivateAttr
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 from types import MappingProxyType
-from typing import Any, Dict, Literal, Mapping, Optional, Self, Sequence
+from typing import Any, Literal, Mapping, Optional, Self
 from webvtt import Caption, WebVTT
 
 from ctp_slack_bot.models.base import Chunk, Content
+from ctp_slack_bot.utils import to_deep_immutable
 
 
 CHUNK_FRAMES_OVERLAP = 1
@@ -45,12 +44,12 @@ class WebVTTContent(Content):
     id: str
     metadata: Mapping[str, Any] = Field(default_factory=dict)
     start_time: Optional[datetime]
-    frames: Sequence[WebVTTFrame]
+    frames: tuple[WebVTTFrame, ...]
 
     def get_id(self: Self) -> str:
         return self.id
 
-    def get_chunks(self: Self) -> Sequence[Chunk]:
+    def get_chunks(self: Self) -> tuple[Chunk]:
         windows = (tuple(filter(None, window))
                    for window
                    in windowed(self.frames, CHUNK_FRAMES_WINDOW, step=CHUNK_FRAMES_WINDOW-CHUNK_FRAMES_OVERLAP))
@@ -62,7 +61,7 @@ class WebVTTContent(Content):
                            metadata={
                                "start": self.start_time + frames[0].start if self.start_time else None,
                                "end": self.start_time + frames[-1].end if self.start_time else None,
-                               "speakers": tuple(frame.speaker for frame in frames if frame.speaker)
+                               "speakers": (frame.speaker for frame in frames if frame.speaker)
                            })
                      for frames
                      in windows)

src/ctp_slack_bot/services/content_ingestion_service.py

@@ -30,9 +30,9 @@ class ContentIngestionService(BaseModel):
 
     async def process_incoming_content(self: Self, content: Content) -> None:
         logger.debug("Content ingestion service received content with metadata: {}", content.get_metadata())
-        # if self.vector_database_service.has_content(content.get_id()) # TODO
-        #    logger.debug("Ignored content with ID {} because it already exists in the database.", content.get_id())
-        #    return
+        if self.vector_database_service.content_exists(content.get_id()):
+           logger.debug("Ignored content with identifier, {}, because it already exists in the database.", content.get_id())
+           return
         chunks = content.get_chunks()
         await self.__vectorize_and_store_chunks_in_database(chunks)
         logger.debug("Stored {} vectorized chunk(s) in the database.", len(chunks))
@@ -44,6 +44,5 @@ class ContentIngestionService(BaseModel):
         logger.debug("Stored {} vectorized chunk(s) in the database.", len(chunks))
 
     async def __vectorize_and_store_chunks_in_database(self: Self, chunks: Sequence[Chunk]) -> None:
-        vectorized_chunks = self.vectorization_service.vectorize(chunks) # TODO
-        await self.vector_database_service.store(vectorized_chunks) # TODO
-
+        vectorized_chunks = self.vectorization_service.vectorize(chunks)
+        await self.vector_database_service.store(vectorized_chunks)

src/ctp_slack_bot/services/context_retrieval_service.py

@@ -34,33 +34,23 @@ class ContextRetrievalService(BaseModel):
         Returns:
             Sequence[Chunk]: List of retrieved context items with similarity scores
         """
-        # Extract chunks from the message
-        message_chunks = message.get_chunks()
-        
-        # Vectorize the chunks
-        vectorized_chunks = self.vectorization_service.vectorize(message_chunks)
-        
-        # Create vector query using the first chunk's embedding (typically there's only one chunk for a message)
-        if not vectorized_chunks:
-            logger.warning("No vectorized chunks were created for message")
-            return []
-            
+        message_chunks = message.get_chunks() # Guaranteed to have exactly 1 chunk
+
+        try:
+            vectorized_message_chunks = self.vectorization_service.vectorize(message_chunks)
+        except Exception as e:
+            logger.error("An error occurred while vectorizing the question, “{}”: {}", message.text, e)
+
         query = VectorQuery(
-            query_embeddings=vectorized_chunks[0].embedding,
+            query_embeddings=vectorized_message_chunks[0].embedding,
             k=self.settings.TOP_K_MATCHES,
             score_threshold=self.settings.SCORE_THRESHOLD,
-            filter_metadata=None  # Can be expanded to include filters based on message metadata
+            filter_metadata={} # Can be expanded to include filters based on message metadata
         )
-        
-        # Perform similarity search
+
         try:
-            results = await self.vector_database_service.search_by_similarity(query)
-            # logger.info(f"Retrieved {len(results)} context chunks for query")
+            results = await self.vector_database_service.find_by_vector(query)
             return results
         except Exception as e:
-            logger.error(f"Error retrieving context: {str(e)}")
-            return []
-        
-        # test return statement
-        # return (VectorizedChunk(text="Mock context chunk", parent_id="lol", chunk_id="no", metadata={}, embedding=tuple()),
-        #         VectorizedChunk(text="Moar mock context chunk", parent_id="lol", chunk_id="wut", metadata={}, embedding=tuple()))
+            logger.error("An error occurred while searching the vector database for context: {}", e)
+            return ()

src/ctp_slack_bot/services/vector_database_service.py

@@ -1,17 +1,18 @@
 from loguru import logger
 from pydantic import BaseModel
-from typing import Any, Collection, Dict, List, Optional, Self, Sequence
+from typing import Iterable, Optional, Self, Sequence
 
 from ctp_slack_bot.core import Settings
-from ctp_slack_bot.db import MongoDB
+from ctp_slack_bot.db.repositories import VectorizedChunkRepository
 from ctp_slack_bot.models import Chunk, VectorizedChunk, VectorQuery
 
 class VectorDatabaseService(BaseModel): # TODO: this should not rely specifically on MongoDB.
     """
     Service for storing and retrieving vector embeddings from MongoDB.
     """
+
     settings: Settings
-    mongo_db: MongoDB
+    vectorized_chunk_repository: VectorizedChunkRepository
 
     class Config:
         frozen=True
@@ -19,157 +20,48 @@ class VectorDatabaseService(BaseModel): # TODO: this should not rely specificall
     def __init__(self: Self, **data) -> None:
         super().__init__(**data)
         logger.debug("Created {}", self.__class__.__name__)
-
-    async def store(self: Self, chunks: Collection[VectorizedChunk]) -> None:
-        """
-        Stores vectorized chunks and their embedding vectors in the database.
-        
-        Args:
-            chunks: Collection of VectorizedChunk objects to store
-            
-        Returns: None
-        """
-        if not chunks:
-            logger.debug("No chunks to store")
-            return
-            
-        try:
-            # Get the vector collection - this will create it if it doesn't exist
-            logger.debug("Getting vectors collection for storing {} chunks", len(chunks))
-            vector_collection = await self.mongo_db.get_collection("vectors")
-            
-            # Ensure vector search index exists
-            logger.debug("Creating vector search index for vectors collection")
-            await self.mongo_db.create_indexes("vectors")
-            
-            # Create documents to store, ensuring compatibility with BSON
-            documents = []
-            for chunk in chunks:
-                # Convert embedding to standard list format (important for BSON compatibility)
-                embedding = list(chunk.embedding) if not isinstance(chunk.embedding, list) else chunk.embedding
-                
-                # Build document with proper structure
-                document = {
-                    "text": chunk.text,
-                    "embedding": embedding,
-                    "metadata": chunk.metadata,
-                    "parent_id": chunk.parent_id,
-                    "chunk_id": chunk.chunk_id
-                }
-                documents.append(document)
-            
-            # Insert into collection as a batch
-            logger.debug("Inserting {} documents into vectors collection", len(documents))
-            result = await vector_collection.insert_many(documents)
-            logger.info("Stored {} vector chunks in database", len(result.inserted_ids))
-            
-        except Exception as e:
-            logger.error("Error storing vector embeddings: {}", str(e))
-            # Include more diagnostic information
-            logger.debug("MongoDB connection info: URI defined: {}, DB name: {}", 
-                         bool(self.settings.MONGODB_URI), self.settings.MONGODB_NAME)
-            raise
     
-    async def content_exists(self: Self, key: str)-> bool: # TODO: implement this.
+    async def content_exists(self: Self, parent_id: str, chunk_id: Optional[str] = None)-> bool:
         """
-        Check if content exists in the database.
+        Check if the content identified by the parent and optionally the chunk identifiers exist in the database.
         
         Args:
-            key: The key to check for content existence
+            parent_id: the identifier of the source content
+            chunk_id:  the identifier of the chunk within the source content
         """
-        pass
+        matching_chunk_count = await self.vectorized_chunk_repository.count_by_id(parent_id, chunk_id)
+        return 0 < matching_chunk_count
 
-    async def search_by_similarity(self: Self, query: VectorQuery) -> Sequence[Chunk]:
+    async def find_by_vector(self: Self, query: VectorQuery) -> Sequence[Chunk]:
         """
-        Query the vector database for similar documents.
-        
+        Query the vector database for similar chunks.
+
         Args:
-            query: VectorQuery object with search parameters
+            query: the query criteria
      
         Returns:
-            Sequence[Chunk]: List of similar chunks
+            Sequence[Chunk]: an ordered collection of similar chunks
+        """
+        try:
+            result = await self.vectorized_chunk_repository.find_by_vector(query)
+            logger.debug("Found {} chunks in the database by similarity search.", len(result))
+            return result
+        except Exception as e:
+            logger.error("Error finding chunks by vector: {}", str(e))
+            raise
+
+    async def store(self: Self, chunks: Iterable[VectorizedChunk]) -> None:
+        """
+        Stores vectorized chunks and their embedding vectors in the database.
+
+        Args:
+            chunks: a collection of vectorized chunks to store
+
+        Returns: None
         """
         try:
-            # Get the vector collection
-            logger.debug("Getting vectors collection for similarity search")
-            vector_collection = await self.mongo_db.get_collection("vectors")
-            
-            # Build aggregation pipeline for vector search using official MongoDB format
-            logger.debug("Building vector search pipeline with query embedding dimension: {}", len(query.query_embeddings))
-            pipeline = [
-                {
-                    "$vectorSearch": {
-                        "index": "vectors_vector_index",
-                        "path": "embedding",
-                        "queryVector": query.query_embeddings, #list(query.query_embeddings),
-                        "numCandidates": query.k * 10,
-                        "limit": query.k
-                    }
-                },
-                {
-                    "$project": {
-                        "text": 1,
-                        "metadata": 1,
-                        "parent_id": 1,
-                        "chunk_id": 1,
-                        "score": { "$meta": "vectorSearchScore" }
-                    }
-                }
-            ]
-            
-            # Add metadata filters if provided
-            if query.filter_metadata:
-                metadata_filter = {f"metadata.{k}": v for k, v in query.filter_metadata.items()}
-                pipeline.insert(1, {"$match": metadata_filter})
-                logger.debug("Added metadata filters to search: {}", query.filter_metadata)
-            
-            # Add score threshold filter if needed
-            if query.score_threshold > 0:
-                pipeline.append({
-                    "$match": {
-                        "score": { "$gte": query.score_threshold }
-                    }
-                })
-                logger.debug("Added score threshold filter: {}", query.score_threshold)
-            
-            try:
-                # Execute the vector search pipeline
-                logger.debug("Executing vector search pipeline")
-                results = await vector_collection.aggregate(pipeline).to_list(length=query.k)
-                logger.debug("Vector search returned {} results", len(results))
-            except Exception as e:
-                logger.warning("Vector search failed: {}. Falling back to basic text search.", str(e))
-                # Fall back to basic filtering with limit
-                query_filter = {}
-                if query.filter_metadata:
-                    query_filter.update({f"metadata.{k}": v for k, v in query.filter_metadata.items()})
-                
-                logger.debug("Executing fallback basic search with filter: {}", query_filter)
-                results = await vector_collection.find(query_filter).limit(query.k).to_list(length=query.k)
-                logger.debug("Fallback search returned {} results", len(results))
-            
-            # Convert results to Chunk objects
-            chunks = []
-            for result in results:
-                chunk = Chunk(
-                    text=result["text"],
-                    parent_id=result["parent_id"],
-                    chunk_id=result["chunk_id"],
-                    metadata={
-                        **result["metadata"],
-                        "similarity_score": result.get("score", 0)
-                    }
-                )
-                chunks.append(chunk)
-            
-            logger.info("Found {} similar chunks with similarity search", len(chunks))
-            return chunks
-            
+            inserted_ids = await self.vectorized_chunk_repository.insert_many(chunks)
+            logger.debug("Stored {} vectorized chunks in the database.", len(inserted_ids))
         except Exception as e:
-            logger.error("Error in similarity search: {}", str(e))
-            # Include additional diagnostic information
-            logger.debug("MongoDB connection info: URI defined: {}, DB name: {}", 
-                         bool(self.settings.MONGODB_URI), self.settings.MONGODB_NAME)
-            logger.debug("Query details: k={}, dimension={}", 
-                         query.k, len(query.query_embeddings) if query.query_embeddings else "None")
+            logger.error("Error storing vectorized chunks: {}", str(e))
             raise

src/ctp_slack_bot/utils/__init__.py

@@ -1 +1,2 @@
+from ctp_slack_bot.utils.immutable import to_deep_immutable
 from ctp_slack_bot.utils.secret_stripper import sanitize_mongo_db_uri

src/ctp_slack_bot/utils/immutable.py

@@ -0,0 +1,22 @@
+from types import MappingProxyType
+from collections.abc import Iterable, Mapping, Sequence, Set
+from typing import Any
+
+
+def to_deep_immutable(obj: Any):
+    """Recursively convert mutable containers to immutable equivalents."""
+
+    # Handle mappings (dict-like).
+    if isinstance(obj, Mapping):
+        return MappingProxyType({to_deep_immutable(key): to_deep_immutable(value) for key, value in obj.items()})
+    
+    # Handle sets.
+    if isinstance(obj, Set):
+        return frozenset(to_deep_immutable(item) for item in obj)
+    
+    # Handle sequences (list/tuple-like).
+    if isinstance(obj, (Iterable, Sequence)) and not isinstance(obj, (str, bytes)):
+        return tuple(to_deep_immutable(item) for item in obj)
+    
+    # Return anything else as-is.
+    return obj

temporary_health_check_server.py

@@ -0,0 +1,11 @@
+from aiohttp import web
+
+async def aliveness_handler(request):
+    return web.Response(text="Server is alive and kicking!")
+
+app = web.Application()
+app.router.add_get('/', aliveness_handler)
+app.router.add_get('/health', aliveness_handler)
+
+if __name__ == "__main__":
+    web.run_app(app, host='0.0.0.0', port=8080)