Refactor #6

README.md

@@ -101,7 +101,7 @@ Not every file or folder is listed, but the important stuff is here.
             * `google_drive_service.py`: interfaces with Google Drive
             * `language_model_service.py`: answers questions using relevant context
             * `question_dispatch_service.py`: listens for questions and retrieves relevant context to get answers
-            * `schedule_service.py`: runs background jobs
+            * `task_service.py`: runs periodic background tasks
             * `slack_service.py`: handles events from Slack and sends back responses
             * `vectorization_service.py`: converts chunks into chunks with embeddings
         * `tasks/`: scheduled tasks to run in the background

src/ctp_slack_bot/app.py

@@ -7,8 +7,11 @@ from containers import Container
 from core.logging import setup_logging
 
 
-async def handle_shutdown_signal() -> None:
+async def handle_shutdown_signal(*args) -> None:
     logger.info("Received shutdown signal.")
+    for arg in args:
+        await arg()
+    logger.info("Executed shutdown tasks.")
     for task in all_tasks():
         if task is not current_task() and not task.done():
             task.cancel()
@@ -16,9 +19,9 @@ async def handle_shutdown_signal() -> None:
     logger.info("Cancelled all tasks.")
 
 
-def create_shutdown_signal_handler() -> Callable[[], None]:
+def create_shutdown_signal_handler(*args) -> Callable[[], None]:
     def shutdown_signal_handler() -> None:
-        create_task(handle_shutdown_signal())
+        create_task(handle_shutdown_signal(*args))
     return shutdown_signal_handler
 
 
@@ -32,29 +35,27 @@ async def main() -> None:
     container.wire(packages=["ctp_slack_bot"])
     logger.debug("Created dependency injection container with providers: {}", '; '.join(container.providers))
 
-    # Initialize/instantiate services which should be active from the beginning.
-    container.content_ingestion_service()
-    container.question_dispatch_service()
+    # Initialize/instantiate services which should be available from the start.
+    await container.content_ingestion_service()
+    await container.question_dispatch_service()
     http_server = await container.http_server()
-    socket_mode_handler = await container.socket_mode_handler()
-    container.schedule_service()
+    slack_service = await container.slack_service()
+    task_service = await container.task_service()
     logger.debug("Initialized services.")
 
     # Install the shutdown signal handler.
-    shutdown_signal_handler = create_shutdown_signal_handler()
+    shutdown_signal_handler = create_shutdown_signal_handler(http_server.stop, slack_service.stop, task_service.stop)
     loop = get_running_loop()
     loop.add_signal_handler(SIGINT, shutdown_signal_handler)
     loop.add_signal_handler(SIGTERM, shutdown_signal_handler)
 
     # Start the HTTP server and Slack socket mode handler in the background; clean up resources when shut down.
     try:
-        logger.info("Starting HTTP server and Slack Socket Mode handler…")
-        await gather(http_server.start(), socket_mode_handler.start_async())
+        logger.info("Starting services…")
+        await gather(http_server.start(), slack_service.start(), task_service.start())
     except CancelledError:
         logger.info("Shutting down application…")
     finally:
-        await socket_mode_handler.close_async()
-        logger.info("Stopped Slack Socket Mode handler.")
         await container.shutdown_resources()
 
 

src/ctp_slack_bot/containers.py

@@ -1,13 +1,13 @@
+from asyncio import iscoroutine, isfuture
+from dependency_injector import providers
 from dependency_injector.containers import DeclarativeContainer
-from dependency_injector.providers import Coroutine, Callable, Dict, List, Resource, Self, Singleton
+from dependency_injector.providers import Callable, Configuration, Dict, List, Resource, Singleton
 from importlib import import_module
 from itertools import chain
 from openai import AsyncOpenAI
 from pkgutil import iter_modules
-from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
-from slack_bolt.async_app import AsyncApp
 from types import ModuleType
-from typing import Sequence
+from typing import Any, Iterator, Sequence
 
 from ctp_slack_bot.controllers import ControllerBase, ControllerRegistry
 from ctp_slack_bot.core import Settings
@@ -16,7 +16,7 @@ from ctp_slack_bot.db.repositories.mongo_db_vectorized_chunk_repository import M
 from ctp_slack_bot.mime_type_handlers import MimeTypeHandlerRegistry
 from ctp_slack_bot.services.answer_retrieval_service import AnswerRetrievalService
 from ctp_slack_bot.services.application_health_service import ApplicationHealthService
-from ctp_slack_bot.services.content_ingestion_service import ContentIngestionService
+from ctp_slack_bot.services.content_ingestion_service import ContentIngestionServiceResource
 from ctp_slack_bot.services.context_retrieval_service import ContextRetrievalService
 from ctp_slack_bot.services.embeddings_model_service import EmbeddingsModelService
 from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
@@ -24,26 +24,38 @@ from ctp_slack_bot.services.google_drive_service import GoogleDriveService
 from ctp_slack_bot.services.http_client_service import HTTPClientServiceResource
 from ctp_slack_bot.services.http_server_service import HTTPServerResource
 from ctp_slack_bot.services.language_model_service import LanguageModelService
-from ctp_slack_bot.services.question_dispatch_service import QuestionDispatchService
-from ctp_slack_bot.services.schedule_service import ScheduleServiceResource
+from ctp_slack_bot.services.question_dispatch_service import QuestionDispatchServiceResource
 from ctp_slack_bot.services.slack_service import SlackServiceResource
+from ctp_slack_bot.services.task_service import TaskServiceResource
 from ctp_slack_bot.services.vectorization_service import VectorizationService
 
 
+async def _await_or_return(value):
+    if iscoroutine(value) or isfuture(value):
+        return await value
+    return value
+
+
 class Container(DeclarativeContainer): # TODO: audit for potential async-related bugs.
-    async def __get_http_controllers(dependency_container: Self) -> Sequence[ControllerBase]:
-        return [controller_class(**{dependency_name: await dependency()
-                                    for dependency_name, dependency
-                                    in dependency_container.providers.items()
-                                    if dependency_name != "__self__" and dependency_name in controller_class.model_fields})
-                for controller_class
-                in ControllerRegistry.get_registry()] # TODO: mutable until async support is extended to tuples
+    async def __get_http_controller_providers(container) -> Sequence[ControllerBase]:
+        return [controller_class(**{dependency_name: await _await_or_return(container.providers[dependency_name]())
+                                    for dependency_name
+                                    in controller_class.model_fields.keys() & container.providers.keys()})
+                for controller_class in ControllerRegistry.get_registry()]
+
+    def __iter_mime_type_handler_providers() -> Iterator[tuple[str, Singleton]]:
+        handler_provider_map = {}
+        for mime_type, handler in MimeTypeHandlerRegistry.get_registry().items():
+            if handler in handler_provider_map:
+                provider = handler_provider_map[handler]
+            else:
+                provider = Singleton(handler)
+                handler_provider_map[handler] = provider
+            yield (mime_type, provider)
 
-    __self__                    = Self()
+    __self__                    = providers.Self()
     settings                    = Singleton(Settings)
     event_brokerage_service     = Singleton(EventBrokerageService)
-    schedule_service            = Resource (ScheduleServiceResource,
-                                            settings=settings)
     http_client                 = Resource (HTTPClientServiceResource)
     mongo_db                    = Resource (MongoDBResource,
                                             settings=settings)
@@ -58,7 +70,7 @@ class Container(DeclarativeContainer): # TODO: audit for potential async-related
     vectorization_service       = Singleton(VectorizationService,
                                             settings=settings,
                                             embeddings_model_service=embeddings_model_service)
-    content_ingestion_service   = Singleton(ContentIngestionService,
+    content_ingestion_service   = Resource (ContentIngestionServiceResource,
                                             settings=settings,
                                             event_brokerage_service=event_brokerage_service,
                                             vectorized_chunk_repository=vectorized_chunk_repository,
@@ -74,25 +86,18 @@ class Container(DeclarativeContainer): # TODO: audit for potential async-related
                                             settings=settings,
                                             event_brokerage_service=event_brokerage_service,
                                             language_model_service=language_model_service)
-    question_dispatch_service   = Singleton(QuestionDispatchService,
+    question_dispatch_service   = Resource (QuestionDispatchServiceResource,
                                             settings=settings,
                                             event_brokerage_service=event_brokerage_service,
-                                            content_ingestion_service=content_ingestion_service,
                                             context_retrieval_service=context_retrieval_service,
                                             answer_retrieval_service=answer_retrieval_service)
-    slack_bolt_app              = Singleton(lambda settings: AsyncApp(token=settings.slack_bot_token.get_secret_value()),
-                                            settings)
     slack_service               = Resource (SlackServiceResource,
+                                            settings=settings,
                                             event_brokerage_service=event_brokerage_service,
-                                            http_client=http_client,
-                                            slack_bolt_app=slack_bolt_app)
-    socket_mode_handler         = Singleton(lambda _, app, settings: AsyncSocketModeHandler(app, settings.slack_app_token.get_secret_value()),
-                                            slack_service,
-                                            slack_bolt_app,
-                                            settings)
-    mime_type_handlers          = Dict     ({mime_type: Singleton(handler)
-                                             for mime_type, handler
-                                             in MimeTypeHandlerRegistry.get_registry().items()})
+                                            http_client=http_client)
+    mime_type_handlers          = Dict     ({mime_type: handler_provider
+                                             for mime_type, handler_provider
+                                             in __iter_mime_type_handler_providers()})
     google_drive_service        = Singleton(GoogleDriveService,
                                             settings=settings)
     # file_monitor_service        = Singleton(FileMonitorService,
@@ -101,8 +106,8 @@ class Container(DeclarativeContainer): # TODO: audit for potential async-related
     #                                         mime_type_handler_factory=mime_type_handler_factory)
     application_health_service  = Singleton(ApplicationHealthService,
                                             services=List(mongo_db, slack_service))
-    http_controllers            = Callable (__get_http_controllers,
-                                            dependency_container=__self__)
+    task_service                = Resource (TaskServiceResource,
+                                            settings=settings)
     http_server                 = Resource (HTTPServerResource,
                                             settings=settings,
-                                            controllers=http_controllers)
+                                            controllers=Callable(__get_http_controller_providers, __self__))

src/ctp_slack_bot/controllers/__init__.py

@@ -1,2 +1,2 @@
 from .application_health_controller import ApplicationHealthController
-from .base import ControllerBase, ControllerRegistry, Route
+from .base import controller, ControllerBase, ControllerRegistry, delete, get, patch, post, put, Route, route

src/ctp_slack_bot/controllers/application_health_controller.py

@@ -2,19 +2,21 @@ from aiohttp.web import json_response, Request, Response
 from pydantic import ConfigDict
 from typing import Self
 
-from .base import ControllerBase, ControllerRegistry, Route
+from .base import ControllerBase, controller, get
 from ctp_slack_bot.services import ApplicationHealthService
 
 
-@ControllerRegistry.register()
+@controller("/health")
 class ApplicationHealthController(ControllerBase):
     """
     Application health reporting endpoints.
     """
 
+    model_config = ConfigDict(frozen=True)
+
     application_health_service: ApplicationHealthService
 
-    @Route.get("/health")
+    @get("")
     async def get_health(self: Self, request: Request) -> Response:
         health_statuses = await self.application_health_service.get_health()
         return json_response(dict(health_statuses), status=200 if all(health_statuses.values()) else 503)

src/ctp_slack_bot/controllers/base.py

@@ -1,9 +1,10 @@
+from abc import abstractmethod
 from aiohttp.web import Request, Response
-from functools import partial
+from functools import partial, wraps
 from importlib import import_module
 from inspect import getmembers, ismethod
 from pydantic import BaseModel, ConfigDict
-from typing import Awaitable, Callable, ClassVar, Mapping, Self, Sequence, TypeVar
+from typing import Awaitable, Callable, ClassVar, Collection, Mapping, Optional, overload, ParamSpec, Self, Sequence, TypeVar
 
 from ctp_slack_bot.core import ApplicationComponentBase
 
@@ -18,21 +19,20 @@ class Route(BaseModel):
     path: str
     handler: AsyncHandler
 
-    @staticmethod
-    def get(path: str) -> Callable[[AsyncHandler], AsyncHandler]:
-        def decorator(function: AsyncHandler) -> AsyncHandler:
-            function._http_method = "GET"
-            function._http_path = path
-            return function
-        return decorator
-
 
 class ControllerBase(ApplicationComponentBase):
 
     def get_routes(self: Self) -> Sequence[Route]:
-        return tuple(Route(method=method._http_method, path=method._http_path, handler=method)
+        return tuple(Route(method=method._http_method,
+                           path="/".join(filter(None, (self.prefix, method._http_path))),
+                           handler=method)
                      for name, method in getmembers(self, predicate=ismethod)
-                     if name != 'get_routes' and hasattr(method, "_http_method") and hasattr(method, "_http_path"))
+                     if name != 'get_routes' and name != 'prefix' and hasattr(method, "_http_method") and hasattr(method, "_http_path"))
+
+    @property
+    @abstractmethod
+    def prefix(self: Self) -> str:
+        pass
 
 
 T = TypeVar('T', bound=ControllerBase)
@@ -40,16 +40,50 @@ T = TypeVar('T', bound=ControllerBase)
 
 class ControllerRegistry:
 
-    _registry: ClassVar[list[T]] = []
+    __registry: ClassVar[list[T]] = []
 
     @classmethod
-    def get_registry(cls) -> Mapping[T, Sequence[Route]]:
+    def get_registry(cls) -> Collection[T]:
         import_module(__package__)
-        return tuple(cls._registry)
+        return tuple(cls.__registry)
 
     @classmethod
-    def register(cls):
-        def decorator(controller_cls: T):
-            cls._registry.append(controller_cls)
-            return controller_cls
-        return decorator
+    def register(cls, controller_cls: T) -> None:
+        cls.__registry.append(controller_cls)
+
+
+@overload
+def controller(cls: T) -> T: ...
+
+@overload
+def controller(prefix: str = "/") -> Callable[[T], T]: ...
+
+def controller(cls_or_prefix=None):
+    def implement_prefix_property_and_register_controller(cls: T, prefix: Optional[str] = "/") -> T:
+        def prefix_getter(self: T) -> str:
+            return prefix
+        setattr(cls, 'prefix', property(prefix_getter))
+        if hasattr(cls, '__abstractmethods__'):
+            cls.__abstractmethods__ = frozenset(method for method in cls.__abstractmethods__ if method != 'prefix')
+        ControllerRegistry.register(cls)
+        return cls
+    if isinstance(cls_or_prefix, type):
+        return implement_prefix_property_and_register_controller(cls_or_prefix)
+    def decorator(cls: T) -> T:
+        return implement_prefix_property_and_register_controller(cls, cls_or_prefix)
+    return decorator
+
+
+def route(method: str, path: str = "") -> Callable[[AsyncHandler], AsyncHandler]:
+    def decorator(function: AsyncHandler) -> AsyncHandler:
+        function._http_method = method
+        function._http_path = path
+        return function
+    return decorator
+
+
+get = partial(route, "GET")
+post = partial(route, "POST")
+put = partial(route, "PUT")
+delete = partial(route, "DELETE")
+patch = partial(route, "PATCH")

src/ctp_slack_bot/db/mongo_db.py

@@ -65,30 +65,16 @@ class MongoDB(HealthReportingApplicationComponentBase):
         """
         Get a collection by name or creates it if it doesn’t exist.
         """
-        # 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.
-            logger.debug("Checking if collection '{}' exists…", name)
-            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)
-                logger.debug("Successfully created collection: {}", name)
+            if name not in await self._db.list_collection_names():
+                collection = await self._db.create_collection(name)
+                logger.debug("Created previously nonexistent collection, {}.", name)
             else:
-                logger.debug("Collection '{}' already exists!", name)
-
-            # Get and return the collection.
-            collection = self._db[name]
+                collection = self._db[name]
+                logger.debug("Retrieved collection, {}.", name)
             return collection
         except Exception as e:
-            logger.error("Error accessing collection '{}': {}", name, e)
+            logger.error("Error accessing collection, {}: {}", name, e)
             raise e
 
     def close(self: Self) -> None:
@@ -117,19 +103,11 @@ class MongoDBResource(AsyncResource):
 
     async def _test_connection(self: Self, mongo_db: MongoDB) -> None:
         """Test MongoDB connection and log the result."""
-        try:
-            is_connected = await mongo_db.ping()
-            if is_connected:
-                logger.info("MongoDB connection test successful!")
-            else:
-                logger.error("MongoDB connection test failed!")
-        except Exception as e:
-            logger.error("Error testing MongoDB connection: {}", e)
-            raise e
+        if await mongo_db.ping():
+            logger.info("MongoDB connection test successful!")
+        else:
+            logger.error("MongoDB connection test failed!")
 
     async def shutdown(self: Self, mongo_db: MongoDB) -> None:
         """Close MongoDB connection on shutdown."""
-        try:
-            mongo_db.close()
-        except Exception as e:
-            logger.error("Error closing MongoDB connection: {}", e)
+        mongo_db.close()

src/ctp_slack_bot/mime_type_handlers/__init__.py

@@ -1,2 +1,2 @@
-from .base import MimeTypeHandler, MimeTypeHandlerRegistry
+from .base import MimeTypeHandler, mime_type_handler, MimeTypeHandlerRegistry
 from .text.vtt import WebVTTMimeTypeHandler

src/ctp_slack_bot/mime_type_handlers/base.py

@@ -1,7 +1,7 @@
-from abc import ABC, abstractmethod
+from abc import abstractmethod
 from importlib import import_module
 from types import MappingProxyType
-from typing import Any, ClassVar, Mapping, Optional
+from typing import Any, Callable, ClassVar, Mapping, Optional, overload, Set, TypeVar
 
 from ctp_slack_bot.core import ApplicationComponentBase
 from ctp_slack_bot.models import Content
@@ -14,20 +14,38 @@ class MimeTypeHandler(ApplicationComponentBase):
         pass
 
 
+T = TypeVar('T', bound=MimeTypeHandler)
+
+
 class MimeTypeHandlerRegistry:
 
-    _registry: ClassVar[dict[str, MimeTypeHandler]] = {}
+    __registry: ClassVar[dict[str, T]] = {}
 
     @classmethod
-    def get_registry(cls) -> Mapping[str, MimeTypeHandler]:
+    def get_registry(cls) -> Mapping[str, T]:
         import_module(__package__)
-        return MappingProxyType(cls._registry)
+        return MappingProxyType(cls.__registry)
 
     @classmethod
-    def register(cls, mime_type: str):
-        def decorator(handler_cls: MimeTypeHandler):
-            if mime_type in cls._registry:
+    def register(cls, mime_types: Set[str], handler_cls: T):
+        for mime_type in mime_types:
+            if mime_type in cls.__registry:
                 raise ValueError(f"The MIME type, {mime_type}, is already registered.")
-            cls._registry[mime_type] = handler_cls
-            return handler_cls
-        return decorator
+            cls.__registry[mime_type] = handler_cls
+
+
+@overload
+def mime_type_handler(cls: T) -> T: ...
+
+@overload
+def mime_type_handler(mime_types: Optional[Set[str] | str] = None) -> Callable[[T], T]: ...
+
+def mime_type_handler(cls_or_mime_types=None):
+    def register_mime_type_handler(cls: T, mime_types: Optional[Set[str]] = None) -> T:
+        MimeTypeHandlerRegistry.register({mime_types} if isinstance(mime_types, str) else mime_types, cls)
+        return cls
+    if isinstance(cls_or_mime_types, type):
+        return register_mime_type_handler(cls_or_mime_types)
+    def decorator(cls: T) -> T:
+        return register_mime_type_handler(cls, cls_or_mime_types)
+    return decorator

src/ctp_slack_bot/mime_type_handlers/text/vtt.py

@@ -6,11 +6,11 @@ from types import MappingProxyType
 from typing import Any, ClassVar, Mapping, Optional, Self
 from webvtt import WebVTT
 
-from ctp_slack_bot.mime_type_handlers.base import MimeTypeHandler, MimeTypeHandlerRegistry
+from ctp_slack_bot.mime_type_handlers.base import MimeTypeHandler, mime_type_handler
 from ctp_slack_bot.models import Content, WebVTTContent, WebVTTFrame
 
 
-@MimeTypeHandlerRegistry.register("text/vtt")
+@mime_type_handler("text/vtt")
 class WebVTTMimeTypeHandler(MimeTypeHandler):
 
     model_config = ConfigDict(frozen=True)

src/ctp_slack_bot/services/__init__.py

@@ -8,6 +8,6 @@ from .google_drive_service import GoogleDriveService
 from .http_server_service import HTTPServer
 from .language_model_service import LanguageModelService
 from .question_dispatch_service import QuestionDispatchService
-from. schedule_service import ScheduleService
 from .slack_service import SlackService
+from. task_service import TaskService
 from .vectorization_service import VectorizationService

src/ctp_slack_bot/services/content_ingestion_service.py

@@ -1,3 +1,4 @@
+from dependency_injector.resources import AsyncResource
 from loguru import logger
 from pydantic import ConfigDict
 from typing import Any, Self, Sequence, Set
@@ -18,15 +19,9 @@ class ContentIngestionService(ApplicationComponentBase):
     model_config = ConfigDict(frozen=True)
 
     settings: Settings
-    event_brokerage_service: EventBrokerageService
     vectorized_chunk_repository: VectorizedChunkRepository
     vectorization_service: VectorizationService
 
-    def model_post_init(self: Self, context: Any, /) -> None:
-        super().model_post_init(context)
-        self.event_brokerage_service.subscribe(EventType.INCOMING_CONTENT, self.process_incoming_content)
-        # self.event_brokerage_service.subscribe(EventType.INCOMING_SLACK_MESSAGE, self.process_incoming_slack_message)
-
     async def process_incoming_content(self: Self, content: Content) -> None:
         logger.debug("Content ingestion service received content with metadata: {}", content.get_metadata())
         if self.vectorized_chunk_repository.count_by_id(content.get_id()):
@@ -49,3 +44,11 @@ class ContentIngestionService(ApplicationComponentBase):
     @property
     def name(self: Self) -> str:
         return "content_ingestion_service"
+
+
+class ContentIngestionServiceResource(AsyncResource):
+    async def init(self: Self, settings: Settings, event_brokerage_service: EventBrokerageService, vectorized_chunk_repository: VectorizedChunkRepository, vectorization_service: VectorizationService) -> ContentIngestionService:
+        content_ingestion_service = ContentIngestionService(settings=settings, vectorized_chunk_repository=vectorized_chunk_repository, vectorization_service=vectorization_service)
+        await event_brokerage_service.subscribe(EventType.INCOMING_CONTENT, content_ingestion_service.process_incoming_content)
+        # await event_brokerage_service.subscribe(EventType.INCOMING_SLACK_MESSAGE, content_ingestion_service.process_incoming_slack_message)
+        return content_ingestion_service

src/ctp_slack_bot/services/event_brokerage_service.py

@@ -1,4 +1,4 @@
-from asyncio import create_task, iscoroutinefunction, to_thread
+from asyncio import create_task, iscoroutinefunction, Lock, to_thread
 from collections import defaultdict
 from loguru import logger
 from pydantic import ConfigDict, PrivateAttr
@@ -15,18 +15,19 @@ class EventBrokerageService(ApplicationComponentBase):
 
     model_config = ConfigDict(frozen=True)
 
-    _subscribers: MutableMapping[EventType, list[Callable]] = PrivateAttr(default_factory=lambda: defaultdict(list))
+    __write_lock: Lock = PrivateAttr(default_factory=Lock)
+    __subscribers: MutableMapping[EventType, tuple[Callable]] = PrivateAttr(default_factory=lambda: defaultdict(tuple))
 
-    def subscribe(self: Self, type: EventType, callback: Callable) -> None:
+    async def subscribe(self: Self, type: EventType, callback: Callable) -> None:
         """Subscribe to an event type with a callback function."""
-        logger.debug("1 new subscriber is listening for {} events.", type)
-        subscribers = self._subscribers[type]
-        subscribers.append(callback)
-        logger.debug("Event type {} has {} subscriber(s).", type, len(subscribers))
+        async with self.__write_lock:
+            subscribers = self.__subscribers[type]
+            self.__subscribers[type] = subscribers + (callback, )
+            logger.debug("One new subscriber was added for event type {} ({} subscriber(s) in total).", type, len(subscribers))
 
     async def publish(self: Self, type: EventType, data: Any = None) -> None:
         """Publish an event with optional data to all subscribers."""
-        subscribers = self._subscribers[type]
+        subscribers = self.__subscribers[type]
         if not subscribers:
             logger.debug("No subscribers handle event {}: {}", type, len(subscribers), data)
             return

src/ctp_slack_bot/services/google_drive_service.py

@@ -39,7 +39,6 @@ class GoogleDriveService(ApplicationComponentBase):
             "token_uri": self.settings.google_token_uri,
         }, scopes=["https://www.googleapis.com/auth/drive"])
         self._google_drive_client = build('drive', 'v3', credentials=credentials)
-        logger.info(type(self._google_drive_client))
 
     def _resolve_folder_id(self: Self, folder_path: str) -> Optional[str]:
         """Resolve a folder path to a Google Drive ID."""

src/ctp_slack_bot/services/question_dispatch_service.py

@@ -1,3 +1,4 @@
+from dependency_injector.resources import AsyncResource
 from loguru import logger
 from pydantic import ConfigDict
 from typing import Any, Self
@@ -18,15 +19,10 @@ class QuestionDispatchService(ApplicationComponentBase):
     model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)
 
     settings: Settings
-    event_brokerage_service: EventBrokerageService
     context_retrieval_service: ContextRetrievalService
     answer_retrieval_service: AnswerRetrievalService
 
-    def model_post_init(self: Self, context: Any, /) -> None:
-        super().model_post_init(context)
-        self.event_brokerage_service.subscribe(EventType.INCOMING_SLACK_MESSAGE, self.__process_incoming_slack_message)
-
-    async def __process_incoming_slack_message(self: Self, message: SlackMessage) -> None:
+    async def process_incoming_slack_message(self: Self, message: SlackMessage) -> None:
         if message.subtype != 'bot_message':
             logger.debug("Question dispatch service received an answerable question: {}", message.text)
             context = await self.context_retrieval_service.get_context(message)
@@ -35,3 +31,10 @@ class QuestionDispatchService(ApplicationComponentBase):
     @property
     def name(self: Self) -> str:
         return "question_dispatch_service"
+
+
+class QuestionDispatchServiceResource(AsyncResource):
+    async def init(self: Self, settings: Settings, event_brokerage_service: EventBrokerageService, context_retrieval_service: ContextRetrievalService, answer_retrieval_service: AnswerRetrievalService) -> QuestionDispatchService:
+        question_dispatch_service = QuestionDispatchService(settings=settings, context_retrieval_service=context_retrieval_service, answer_retrieval_service=answer_retrieval_service)
+        await event_brokerage_service.subscribe(EventType.INCOMING_SLACK_MESSAGE, question_dispatch_service.process_incoming_slack_message)
+        return question_dispatch_service

src/ctp_slack_bot/services/slack_service.py

@@ -2,14 +2,15 @@ from dependency_injector.resources import AsyncResource
 from httpx import AsyncClient
 from loguru import logger
 from openai import OpenAI
-from pydantic import ConfigDict
+from pydantic import ConfigDict, PrivateAttr
 from re import compile as compile_re, Pattern
-from slack_sdk.errors import SlackApiError
+from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
 from slack_bolt.async_app import AsyncApp
+from slack_sdk.errors import SlackApiError
 from slack_sdk.web.async_slack_response import AsyncSlackResponse
 from typing import Any, ClassVar, Mapping, MutableMapping, Optional, Self, Set
 
-from ctp_slack_bot.core import HealthReportingApplicationComponentBase
+from ctp_slack_bot.core import HealthReportingApplicationComponentBase, Settings
 from ctp_slack_bot.enums import EventType
 from ctp_slack_bot.models import SlackMessage, SlackResponse
 from .event_brokerage_service import EventBrokerageService
@@ -25,17 +26,54 @@ class SlackService(HealthReportingApplicationComponentBase):
     _SLACK_USER_ID_PATTERN: ClassVar[Pattern] = compile_re(r"U\d+")
     _SLACK_USER_MENTION_PATTERN: ClassVar[Pattern] = compile_re(r"<@(U[A-Z0-9]+)>")
 
+    settings: Settings
     event_brokerage_service: EventBrokerageService
     http_client: AsyncClient
     slack_bolt_app: AsyncApp
-    id_name_map: MutableMapping[str, str] # TODO: Spin message processing out into its own service.
+    socket_mode_handler: AsyncSocketModeHandler
+    _id_name_map: MutableMapping[str, str] = PrivateAttr(default={}) # TODO: Spin message processing out into its own service.
 
-    def initialize(self: Self) -> None:
-        self.event_brokerage_service.subscribe(EventType.OUTGOING_SLACK_RESPONSE, self.send_message)
+    async def initialize(self: Self) -> None:
+        async def get_users_list():
+            cursor = None
+            while True:
+                try:
+                    response = await self.slack_bolt_app.client.users_list(cursor=cursor, limit=200)
+                except SlackApiError as e:
+                    logger.warning("Could not get a list of users: {}", e)
+                    break
+                match response:
+                    case AsyncSlackResponse(status_code=200, data={"ok": True, "members": users}):
+                        for user in users:
+                            yield user
+                        match response.data:
+                            case {"response_metadata": {"next_cursor": cursor}} if cursor:
+                                continue
+                    case AsyncSlackResponse(status_code=status_code) if status_code != 200:
+                        logger.warning("Could not get a list of users: response status {}", status_code)
+                    case AsyncSlackResponse(data={"ok": False}):
+                        logger.warning("Could not get a list of users: non-OK response")
+                    case _:
+                        logger.warning("Could not get a list of users.")
+                break
+        id_name_map = {user["id"]: self._resolve_user_name(user)
+                       async for user
+                       in get_users_list()}
+        self._id_name_map.update(id_name_map)
+        logger.debug("Obtained a list of {} user name(s) for the workspace: {}", len(id_name_map), id_name_map)
+
+        await self.event_brokerage_service.subscribe(EventType.OUTGOING_SLACK_RESPONSE, self.send_message)
         self.slack_bolt_app.event("message")(self._handle_message_event)
         self.slack_bolt_app.event("app_mention")(self._handle_app_mention_event)
         logger.debug("Registered 2 handlers for Slack Bolt message and app mention events.")
 
+    async def start(self: Self) -> None:
+        await self.socket_mode_handler.start_async()
+
+    async def stop(self: Self) -> None:
+        await self.socket_mode_handler.close_async()
+        logger.info("Stopped Slack Bolt socket mode handler and Slack service.")
+
     async def send_message(self: Self, message: SlackResponse) -> None:
         await self.slack_bolt_app.client.chat_postMessage(channel=message.channel, text=message.text, thread_ts=message.thread_ts)
 
@@ -64,16 +102,16 @@ class SlackService(HealthReportingApplicationComponentBase):
         )
 
     async def _ensure_ids_in_id_name_map(self: Self, ids: Set[str]) -> None:
-        unknown_ids = ids - self.id_name_map.keys()
+        unknown_ids = ids - self._id_name_map.keys()
         if len(unknown_ids) == 0:
             return
         async with TaskGroup() as task_group:
             update_tasks = {unknown_id: task_group.create_task(self._look_up_name(unknown_id)) for unknown_id in unknown_ids}
-        self.id_name_map.update({id: task.result() for id, task in update_tasks.items() if task.result()})
+        self._id_name_map.update({id: task.result() for id, task in update_tasks.items() if task.result()})
 
     async def _get_name(self: Self, id: str) -> str:
         await self._ensure_ids_in_id_name_map({id})
-        return self.id_name_map.get(id, id)
+        return self._id_name_map.get(id, id)
 
     async def _handle_message_event(self: Self, body: Mapping[str, Any]) -> None:
         logger.debug("Ignored regular message: {}", body.get("event", {}).get("text"))
@@ -113,46 +151,13 @@ class SlackService(HealthReportingApplicationComponentBase):
             start, end = match.span()
             parts.append(text[previous_end:start])
             user_id = match.group(1)
-            parts.append(f"@{self.id_name_map.get(user_id, user_id)}")
+            parts.append(f"@{self._id_name_map.get(user_id, user_id)}")
             previous_end = end
         parts.append(text[previous_end:])
         return ''.join(parts)
 
-
-class SlackServiceResource(AsyncResource):
-    async def init(self: Self, event_brokerage_service: EventBrokerageService, http_client: AsyncClient, slack_bolt_app: AsyncApp) -> SlackService:
-        async def get_users_list():
-            cursor = None
-            while True:
-                try:
-                    response = await slack_bolt_app.client.users_list(cursor=cursor, limit=200)
-                except SlackApiError as e:
-                    logger.warning("Could not get a list of users: {}", e)
-                    break
-                match response:
-                    case AsyncSlackResponse(status_code=200, data={"ok": True, "members": users}):
-                        for user in users:
-                            yield user
-                        match response.data:
-                            case {"response_metadata": {"next_cursor": cursor}} if cursor:
-                                continue
-                    case AsyncSlackResponse(status_code=status_code) if status_code != 200:
-                        logger.warning("Could not get a list of users: response status {}", status_code)
-                    case AsyncSlackResponse(data={"ok": False}):
-                        logger.warning("Could not get a list of users: non-OK response")
-                    case _:
-                        logger.warning("Could not get a list of users.")
-                break
-        id_name_map = {user["id"]: self._get_name(user)
-                       async for user
-                       in get_users_list()}
-        logger.debug("Obtained a list of {} user name(s) for the workspace: {}", len(id_name_map), id_name_map)
-        slack_service = SlackService(event_brokerage_service=event_brokerage_service, http_client=http_client, slack_bolt_app=slack_bolt_app, id_name_map=id_name_map)
-        slack_service.initialize()
-        return slack_service
-
-    @classmethod
-    def _get_name(cls, user: Mapping[str, Any]):
+    @staticmethod
+    def _resolve_user_name(user: Mapping[str, Any]):
         match user:
             case {"real_name": real_name}:
                 return real_name
@@ -160,3 +165,12 @@ class SlackServiceResource(AsyncResource):
                 return display_name
             case {"name": name}:
                 return name
+
+
+class SlackServiceResource(AsyncResource):
+    async def init(self: Self, settings: Settings, event_brokerage_service: EventBrokerageService, http_client: AsyncClient) -> SlackService:
+        slack_bolt_app = AsyncApp(token=settings.slack_bot_token.get_secret_value())
+        socket_mode_handler = AsyncSocketModeHandler(slack_bolt_app, settings.slack_app_token.get_secret_value())
+        slack_service = SlackService(settings=settings, event_brokerage_service=event_brokerage_service, http_client=http_client, slack_bolt_app=slack_bolt_app, socket_mode_handler=socket_mode_handler)
+        await slack_service.initialize()
+        return slack_service

src/ctp_slack_bot/services/{schedule_service.py → task_service.py}

@@ -1,17 +1,16 @@
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
-from asyncio import create_task, iscoroutinefunction, to_thread
 from datetime import datetime
-from dependency_injector.resources import Resource
+from dependency_injector.resources import AsyncResource
 from loguru import logger
 from pydantic import ConfigDict
 from pytz import timezone
-from typing import Any, Optional, Self
+from typing import Any, Self
 
 from ctp_slack_bot.core import ApplicationComponentBase, Settings
 
 
-class ScheduleService(ApplicationComponentBase):
+class TaskService(ApplicationComponentBase):
     """
     Service for running scheduled tasks.
     """
@@ -43,28 +42,25 @@ class ScheduleService(ApplicationComponentBase):
         # )
         pass
 
-    def start(self: Self) -> None:
+    async def start(self: Self) -> None:
+        logger.info("Starting scheduler…")
         self._scheduler.start()
 
-    def stop(self: Self) -> None:
+    async def stop(self: Self) -> None:
         if self._scheduler.running:
             self._scheduler.shutdown()
+            logger.info("Stopped scheduler.")
         else:
             logger.debug("The scheduler is not running. There is no scheduler to shut down.")
 
     @property
     def name(self: Self) -> str:
-        return "schedule_service"
+        return "task_service"
 
 
-class ScheduleServiceResource(Resource):
-    def init(self: Self, settings: Settings) -> ScheduleService:
-        logger.info("Starting scheduler…")
-        schedule_service = ScheduleService(settings=settings)
-        schedule_service.start()
-        return schedule_service
+class TaskServiceResource(AsyncResource):
+    async def init(self: Self, settings: Settings) -> TaskService:
+        return TaskService(settings=settings)
 
-    def shutdown(self: Self, schedule_service: ScheduleService) -> None:
-        """Stop scheduler on shutdown."""
-        schedule_service.stop()
-        logger.info("Stopped scheduler.")
+    async def shutdown(self: Self, task_service: TaskService) -> None:
+        await task_service.stop()