fix(app_enabled): centralize app_enabled check

to reduce requests to nextcloud server

Signed-off-by: Marcel Klehr <mklehr@gmx.net>

Author: marcelklehr

context_chat_backend/controller.py

@@ -2,6 +2,8 @@
 # SPDX-FileCopyrightText: 2023 Nextcloud GmbH and Nextcloud contributors
 # SPDX-License-Identifier: AGPL-3.0-or-later
 #
+import time
+
 from nc_py_api.ex_app.providers.task_processing import TaskProcessingProvider
 
 # isort: off
@@ -61,7 +63,20 @@
 		'revision': '607a30d783dfa663caf39e06633721c8d4cfcd7e',
 	}
 } if __download_models_from_hf else {}
+
+
 app_enabled = threading.Event()
+last_enabled_check: int|None = None
+def get_enabled_state() -> bool:
+	global last_enabled_check
+	if last_enabled_check is None or time.time() - last_enabled_check > 30:
+		nc = NextcloudApp()
+		if nc.enabled_state:
+			app_enabled.set()
+		else:
+			app_enabled.clear()
+		last_enabled_check = time.time()
+	return app_enabled.is_set()
 
 def enabled_handler(enabled: bool, nc: NextcloudApp | AsyncNextcloudApp) -> str:
 	try:
@@ -87,7 +102,7 @@ def enabled_handler(enabled: bool, nc: NextcloudApp | AsyncNextcloudApp) -> str:
 			if THREAD_STOP_EVENT.is_set():
 				# If the threads were previously stopped, we start them again
 				# otherwise the lifecycle handler has already started them
-				start_bg_threads(app_config)
+				start_bg_threads(app_config, get_enabled_state)
 				THREAD_STOP_EVENT.clear()
 		else:
 			app_enabled.clear()
@@ -103,9 +118,8 @@ def enabled_handler(enabled: bool, nc: NextcloudApp | AsyncNextcloudApp) -> str:
 @asynccontextmanager
 async def lifespan(app: FastAPI):
 	set_handlers(app, enabled_handler, models_to_fetch=models_to_fetch, trigger_handler=trigger_handler)
-	start_bg_threads(app_config)
-	nc = NextcloudApp()
-	logger.info(f'App enable state at startup: {nc.enabled_state}')
+	start_bg_threads(app_config, get_enabled_state)
+	logger.info(f'App enable state at startup: {get_enabled_state()}')
 	yield
 	vectordb_loader.offload()
 	wait_for_bg_threads()
@@ -192,7 +206,7 @@ def decorator(func: Callable):
 		@wraps(func)
 		def wrapper(*args, **kwargs):
 			disable_aaa = app.extra['CONFIG'].disable_aaa
-			if not disable_aaa and not app_enabled.is_set():
+			if not disable_aaa and not get_enabled_state():
 				return JSONResponse('Context Chat is disabled, enable it from AppAPI to use it.', 503)
 
 			return func(*args, **kwargs)
@@ -213,7 +227,7 @@ def _(request: Request):
 
 @app.get('/enabled')
 def _():
-	return JSONResponse(content={'enabled': app_enabled.is_set()}, status_code=200)
+	return JSONResponse(content={'enabled': get_enabled_state()}, status_code=200)
 
 
 @app.post('/countIndexedDocuments')

context_chat_backend/task_fetcher.py

@@ -10,7 +10,7 @@
 from contextlib import suppress
 from enum import Enum
 from threading import Event, Thread
-from time import sleep, time
+from time import sleep
 from typing import Any
 
 import niquests
@@ -69,7 +69,7 @@ class ThreadType(Enum):
 	REQUEST_PROCESSING = 'request_processing'
 
 
-def files_indexing_thread(app_config: TConfig) -> None:
+def files_indexing_thread(app_config: TConfig, get_enabled_state) -> None:
 	try:
 		network_em = NetworkEmbeddings(app_config)
 		vectordb_loader = VectorDBLoader(app_config)
@@ -138,18 +138,12 @@ def _load_sources(source_items: Mapping[int, SourceItem | ReceivedFileItem]) ->
 	LOGGER.info(f'Using {file_parsing_cpu_count} parallel file parsing workers')
 
 	nc = NextcloudApp()
-	last_enabled_check = time()
-	enabled_state = nc.enabled_state
 	while True:
 		if THREAD_STOP_EVENT.is_set():
 			LOGGER.info('Files indexing thread is stopping due to stop event being set')
 			return
 
-		if time() - last_enabled_check > 30:  # check enabled state every 30 seconds
-			enabled_state = nc.enabled_state
-			last_enabled_check = time()
-
-		if not enabled_state:
+		if not get_enabled_state():
 			LOGGER.info('App is disabled, files indexing thread will sleep until next enabled state check')
 			sleep(POLLING_COOLDOWN)
 			continue
@@ -284,26 +278,20 @@ def _load_sources(source_items: Mapping[int, SourceItem | ReceivedFileItem]) ->
 
 
 
-def updates_processing_thread(app_config: TConfig) -> None:
+def updates_processing_thread(app_config: TConfig, get_enabled_state) -> None:
 	try:
 		vectordb_loader = VectorDBLoader(app_config)
 	except LoaderException as e:
 		LOGGER.error('Error initializing vector DB loader, files indexing thread will not start:', exc_info=e)
 		return
 
 	nc = NextcloudApp()
-	enabled_state = nc.enabled_state
-	last_enabled_check = time()
 	while True:
 		if THREAD_STOP_EVENT.is_set():
 			LOGGER.info('Updates processing thread is stopping due to stop event being set')
 			return
 
-		if time() - last_enabled_check > 30:  # check enabled state every 30 seconds
-			enabled_state = nc.enabled_state
-			last_enabled_check = time()
-
-		if not enabled_state:
+		if not get_enabled_state():
 			sleep(POLLING_COOLDOWN)
 			continue
 
@@ -475,7 +463,7 @@ def resolve_scope_list(source_ids: list[str], userId: str) -> list[str]:
 	return ScopeList.model_validate(data).source_ids
 
 
-def request_processing_thread(app_config: TConfig) -> None:
+def request_processing_thread(app_config: TConfig, get_enabled_state) -> None:
 	LOGGER.info('Starting request processing thread')
 
 	try:
@@ -487,8 +475,6 @@ def request_processing_thread(app_config: TConfig) -> None:
 		return
 
 	nc = NextcloudApp()
-	enabled_state = nc.enabled_state
-	last_enabled_check = time()
 	llm: LLM = llm_loader.load()
 
 	while True:
@@ -500,11 +486,7 @@ def request_processing_thread(app_config: TConfig) -> None:
 			sleep(POLLING_COOLDOWN)
 			continue
 
-		if time() - last_enabled_check > 30:  # check enabled state every 30 seconds
-			enabled_state = nc.enabled_state
-			last_enabled_check = time()
-
-		if not enabled_state:
+		if not get_enabled_state():
 			sleep(POLLING_COOLDOWN)
 			continue
 
@@ -726,7 +708,7 @@ def process_search_task(
 	)
 
 
-def start_bg_threads(app_config: TConfig):
+def start_bg_threads(app_config: TConfig, get_enabled_state):
 	if APP_ROLE == AppRole.INDEXING or APP_ROLE == AppRole.NORMAL:
 		if (
 			ThreadType.FILES_INDEXING in THREADS
@@ -738,12 +720,12 @@ def start_bg_threads(app_config: TConfig):
 		THREAD_STOP_EVENT.clear()
 		THREADS[ThreadType.FILES_INDEXING] = Thread(
 			target=files_indexing_thread,
-			args=(app_config,),
+			args=(app_config,get_enabled_state),
 			name='FilesIndexingThread',
 		)
 		THREADS[ThreadType.UPDATES_PROCESSING] = Thread(
 			target=updates_processing_thread,
-			args=(app_config,),
+			args=(app_config,get_enabled_state),
 			name='UpdatesProcessingThread',
 		)
 		THREADS[ThreadType.FILES_INDEXING].start()
@@ -757,7 +739,7 @@ def start_bg_threads(app_config: TConfig):
 		THREAD_STOP_EVENT.clear()
 		THREADS[ThreadType.REQUEST_PROCESSING] = Thread(
 			target=request_processing_thread,
-			args=(app_config,),
+			args=(app_config,get_enabled_state),
 			name='RequestProcessingThread',
 		)
 		THREADS[ThreadType.REQUEST_PROCESSING].start()

context_chat_backend/utils.py

@@ -9,6 +9,7 @@
 import os
 import signal
 import sys
+import time
 import traceback
 from collections.abc import Callable
 from contextlib import suppress
@@ -18,7 +19,6 @@
 from typing import Any, TypeGuard, TypeVar
 
 from fastapi.responses import JSONResponse as FastAPIJSONResponse
-
 from .types import AppRole, TConfig, TEmbeddingAuthApiKey, TEmbeddingAuthBasic, TEmbeddingConfig
 
 T = TypeVar('T')
@@ -313,4 +313,4 @@ def get_app_role() -> AppRole:
 	if role not in ['indexing', 'rp']:
 		_logger.warning(f'Invalid app role: {role}, defaulting to all roles')
 		return AppRole.NORMAL
-	return AppRole(role)
+	return AppRole(role)