From ca5bb8fe93a157bc77bc2b22175127b017ef8bf9 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Mon, 5 Jan 2026 16:18:15 +0100 Subject: [PATCH] refactor --- .../templates/failed-notifications.html | 3 +- .../{task_queue.py => task_queue/__init__.py} | 443 +++++------------- .../notification/task_queue/base.py | 127 +++++ .../notification/task_queue/file_storage.py | 270 +++++++++++ .../notification/task_queue/redis_storage.py | 204 ++++++++ .../notification/task_queue/sqlite_storage.py | 239 ++++++++++ .../static/images/notification-fail.svg | 10 + .../tests/unit/test_huey_filestorage.py | 161 +++++++ 8 files changed, 1127 insertions(+), 330 deletions(-) rename changedetectionio/notification/{task_queue.py => task_queue/__init__.py} (73%) create mode 100644 changedetectionio/notification/task_queue/base.py create mode 100644 changedetectionio/notification/task_queue/file_storage.py create mode 100644 changedetectionio/notification/task_queue/redis_storage.py create mode 100644 changedetectionio/notification/task_queue/sqlite_storage.py create mode 100644 changedetectionio/static/images/notification-fail.svg create mode 100644 changedetectionio/tests/unit/test_huey_filestorage.py diff --git a/changedetectionio/blueprint/settings/templates/failed-notifications.html b/changedetectionio/blueprint/settings/templates/failed-notifications.html index aa7e57fe1..4326232fa 100644 --- a/changedetectionio/blueprint/settings/templates/failed-notifications.html +++ b/changedetectionio/blueprint/settings/templates/failed-notifications.html @@ -153,7 +153,8 @@ You can fix the notification settings (e.g., SMTP server) and retry them manually below.

-
diff --git a/changedetectionio/notification/task_queue.py b/changedetectionio/notification/task_queue/__init__.py similarity index 73% rename from changedetectionio/notification/task_queue.py rename to changedetectionio/notification/task_queue/__init__.py index d408c7f8b..3cdd5b01b 100644 --- a/changedetectionio/notification/task_queue.py +++ b/changedetectionio/notification/task_queue/__init__.py @@ -97,6 +97,40 @@ def get_retry_config(): } +# ============================================================================ +# Storage Backend Abstraction - Import Task Manager Classes +# ============================================================================ + +from .base import HueyTaskManager +from .file_storage import FileStorageTaskManager +from .sqlite_storage import SqliteStorageTaskManager +from .redis_storage import RedisStorageTaskManager + + +def _get_task_manager(): + """ + Factory function to get the appropriate task manager for the current storage backend. + + Returns: + HueyTaskManager: Concrete task manager instance for the storage backend + """ + if huey is None: + return None + + storage_type = type(huey.storage).__name__ + + if storage_type == 'FileStorage': + storage_path = getattr(huey.storage, 'path', None) + return FileStorageTaskManager(huey.storage, storage_path) + elif storage_type in ['SqliteStorage', 'SqliteHuey']: + return SqliteStorageTaskManager(huey.storage) + elif storage_type in ['RedisStorage', 'RedisHuey']: + return RedisStorageTaskManager(huey.storage) + else: + logger.warning(f"Unknown storage type {storage_type}, operations may fail") + return None + + # Global Huey instance (initialized later with proper datastore path) huey = None @@ -181,89 +215,18 @@ def init_huey(datastore_path): return huey -def _count_storage_items(storage, storage_type): +def _count_storage_items(): """ - Count items in Huey storage (queue + schedule) based on storage backend type. - - Args: - storage: Huey storage instance - storage_type: Type name string (e.g., 'FileStorage', 'SqliteStorage', 'RedisStorage') + Count items in Huey storage (queue + schedule) using task manager. Returns: Tuple of (queue_count, schedule_count) """ - queue_count = 0 - schedule_count = 0 + task_manager = _get_task_manager() + if task_manager is None: + return 0, 0 - import os - - if storage_type == 'FileStorage': - # FileStorage: Walk file directories - try: - if hasattr(storage, 'path'): - # Count queue files - queue_dir = os.path.join(storage.path, 'queue') - if os.path.exists(queue_dir): - for root, dirs, files in os.walk(queue_dir): - queue_count += len([f for f in files if not f.startswith('.')]) - - # Count schedule files - schedule_dir = os.path.join(storage.path, 'schedule') - if os.path.exists(schedule_dir): - for root, dirs, files in os.walk(schedule_dir): - schedule_count += len([f for f in files if not f.startswith('.')]) - except Exception as e: - logger.debug(f"FileStorage count error: {e}") - - elif storage_type in ['SqliteStorage', 'SqliteHuey']: - # SqliteStorage: Query database tables - try: - import sqlite3 - if hasattr(storage, 'filename'): - conn = sqlite3.connect(storage.filename) - cursor = conn.cursor() - - # Count queue - cursor.execute("SELECT COUNT(*) FROM queue") - queue_count = cursor.fetchone()[0] - - # Count schedule - cursor.execute("SELECT COUNT(*) FROM schedule") - schedule_count = cursor.fetchone()[0] - - conn.close() - except Exception as e: - logger.debug(f"SqliteStorage count error: {e}") - - elif storage_type in ['RedisStorage', 'RedisHuey']: - # RedisStorage: Use Redis commands - try: - if hasattr(storage, 'conn'): - # Queue is a list - queue_count = storage.conn.llen(f"{storage.name}:queue") - - # Schedule is a sorted set - schedule_count = storage.conn.zcard(f"{storage.name}:schedule") - except Exception as e: - logger.debug(f"RedisStorage count error: {e}") - - else: - # Unknown storage type - try generic attributes - try: - if hasattr(storage, 'queue_size'): - queue_count = storage.queue_size() - elif hasattr(storage, 'queue'): - queue_count = len(storage.queue) - except Exception: - pass - - try: - if hasattr(storage, 'schedule'): - schedule_count = len(storage.schedule) - except Exception: - pass - - return queue_count, schedule_count + return task_manager.count_storage_items() def get_pending_notifications_count(): @@ -283,11 +246,8 @@ def get_pending_notifications_count(): return 0 try: - # Detect storage backend type - storage_type = type(huey.storage).__name__ - - # Get counts using backend-specific logic - queue_count, schedule_count = _count_storage_items(huey.storage, storage_type) + # Get counts using task manager (polymorphic, backend-agnostic) + queue_count, schedule_count = _count_storage_items() total_count = queue_count + schedule_count @@ -427,6 +387,22 @@ def get_pending_notifications(limit=50): return pending +def _enumerate_results(): + """ + Enumerate all results from Huey's result store. + + Uses polymorphic task manager to handle storage backend differences. + + Returns: + dict: {task_id: result_data} for all stored results + """ + task_manager = _get_task_manager() + if task_manager is None: + return {} + + return task_manager.enumerate_results() + + def get_last_successful_notification(): """ Get the most recent successful notification for reference. @@ -483,81 +459,11 @@ def get_failed_notifications(limit=100, max_age_days=30): import time try: - # Query Huey's result storage for failed tasks - # Different storage backends work differently + # Query Huey's result storage for failed tasks using backend-agnostic helper cutoff_time = time.time() - (max_age_days * 86400) - results = {} - - # Try to get results - method varies by storage backend - try: - # SqliteHuey/RedisHuey have result_store.flush() - results = huey.storage.result_store.flush() - except AttributeError: - # FileStorage doesn't have result_store.flush() - # Need to enumerate result files directly from filesystem - import os - import pickle - - try: - # FileStorage stores results as pickled files in subdirectories - # Path structure: {storage.path}/results/{hash_subdir}/... - storage_path = huey.storage.path - results_dir = os.path.join(storage_path, 'results') - - if os.path.exists(results_dir): - # Walk through all subdirectories to find result files - for root, dirs, files in os.walk(results_dir): - for filename in files: - if filename.startswith('.'): - continue - - filepath = os.path.join(root, filename) - try: - # Read and unpickle the result - # Huey FileStorage format: 4-byte length + task_id + pickled data - with open(filepath, 'rb') as f: - # Read the task ID header (length-prefixed) - task_id_len_bytes = f.read(4) - if len(task_id_len_bytes) < 4: - raise EOFError("Incomplete header") - task_id_len = struct.unpack('>I', task_id_len_bytes)[0] - task_id_bytes = f.read(task_id_len) - if len(task_id_bytes) < task_id_len: - raise EOFError("Incomplete task ID") - task_id = task_id_bytes.decode('utf-8') - - # Now unpickle the result data - result_data = pickle.load(f) - results[task_id] = result_data - except (pickle.UnpicklingError, EOFError) as e: - # Corrupted or incomplete result file - # This can happen if: - # - Process crashed during write - # - Disk full - # - Leftover from interrupted shutdown - file_size = os.path.getsize(filepath) - logger.warning(f"Corrupted result file {filename} ({file_size} bytes) - likely from interrupted write. Moving to lost-found.") - try: - # Move to lost-found directory instead of deleting - import shutil - lost_found_dir = os.path.join(storage_path, 'lost-found', 'results') - os.makedirs(lost_found_dir, exist_ok=True) - - # Add timestamp to filename to avoid collisions - import time - timestamp = int(time.time()) - lost_found_path = os.path.join(lost_found_dir, f"{filename}.{timestamp}.corrupted") - - shutil.move(filepath, lost_found_path) - logger.info(f"Moved corrupted file to {lost_found_path}") - except Exception as move_err: - logger.error(f"Unable to move corrupted file to lost-found: {move_err}") - except Exception as e: - logger.debug(f"Unable to read result file {filename}: {e}") - # Note: Not logging when results_dir doesn't exist - this is normal when no failures yet - except Exception as e: - logger.debug(f"Unable to enumerate FileStorage results: {e}") + # Use helper function that works with all storage backends + results = _enumerate_results() # Import Huey's Error class for checking failed tasks from huey.utils import Error as HueyError @@ -629,6 +535,23 @@ def get_failed_notifications(limit=100, max_age_days=30): return failed_tasks +def _delete_result(task_id): + """ + Delete a result from Huey's result store using task manager. + + Args: + task_id: Task ID to delete result for + + Returns: + True if deleted successfully, False otherwise + """ + task_manager = _get_task_manager() + if task_manager is None: + return False + + return task_manager.delete_result(task_id) + + def retry_failed_notification(task_id): """ Retry a failed notification by task ID. @@ -662,8 +585,8 @@ def retry_failed_notification(task_id): # which will store new metadata for the new task queue_notification(notification_data) - # Remove from dead letter queue (it will go back if it fails again) - huey.storage.delete(task_id) + # Remove from dead letter queue using backend-appropriate method + _delete_result(task_id) # Clean up old metadata _delete_task_metadata(task_id) @@ -699,8 +622,9 @@ def retry_all_failed_notifications(): try: from huey.utils import Error as HueyError - # Get all failed tasks - results = huey.storage.result_store.flush() + # Use helper function to get all results from backend-agnostic storage + # This works with FileStorage (default), SqliteStorage, and RedisStorage + results = _enumerate_results() for task_id, result in results.items(): if isinstance(result, (Exception, HueyError)): @@ -964,60 +888,31 @@ def send_notification_task(n_object_dict): # Decorator will be applied after huey is initialized # This is set up in init_huey_task() def _store_task_metadata(task_id, n_object_dict): - """Store notification metadata for a task so we can retrieve it later when task fails.""" - if not huey or not hasattr(huey.storage, 'path'): - return + """Store notification metadata using task manager.""" + task_manager = _get_task_manager() + if task_manager is None: + return False - try: - import json - metadata_dir = os.path.join(huey.storage.path, 'task_metadata') - os.makedirs(metadata_dir, exist_ok=True) - - metadata_file = os.path.join(metadata_dir, f"{task_id}.json") - metadata = { - 'task_id': task_id, - 'timestamp': time.time(), - 'notification_data': n_object_dict - } - - with open(metadata_file, 'w') as f: - json.dump(metadata, f, indent=2) - except Exception as e: - logger.debug(f"Unable to store task metadata: {e}") + metadata = {'notification_data': n_object_dict} + return task_manager.store_task_metadata(task_id, metadata) def _get_task_metadata(task_id): - """Retrieve notification metadata for a task ID.""" - if not huey or not hasattr(huey.storage, 'path'): + """Retrieve notification metadata using task manager.""" + task_manager = _get_task_manager() + if task_manager is None: return None - try: - import json - metadata_dir = os.path.join(huey.storage.path, 'task_metadata') - metadata_file = os.path.join(metadata_dir, f"{task_id}.json") - - if os.path.exists(metadata_file): - with open(metadata_file, 'r') as f: - return json.load(f) - except Exception as e: - logger.debug(f"Unable to load task metadata for {task_id}: {e}") - - return None + return task_manager.get_task_metadata(task_id) def _delete_task_metadata(task_id): - """Delete task metadata file (cleanup after success or manual deletion).""" - if not huey or not hasattr(huey.storage, 'path'): - return + """Delete task metadata using task manager.""" + task_manager = _get_task_manager() + if task_manager is None: + return False - try: - metadata_dir = os.path.join(huey.storage.path, 'task_metadata') - metadata_file = os.path.join(metadata_dir, f"{task_id}.json") - - if os.path.exists(metadata_file): - os.remove(metadata_file) - except Exception as e: - logger.debug(f"Unable to delete task metadata for {task_id}: {e}") + return task_manager.delete_task_metadata(task_id) def queue_notification(n_object_dict): @@ -1078,114 +973,19 @@ def clear_all_notifications(): - Schedule (retrying/delayed notifications) - Results (failed notifications) - Retry attempt files + - Task metadata Returns: Dict with counts of cleared items """ - if huey is None: + task_manager = _get_task_manager() + if task_manager is None: return {'error': 'Huey not initialized'} - import os - import shutil - - cleared = { - 'queue': 0, - 'schedule': 0, - 'results': 0, - 'retry_attempts': 0, - 'task_metadata': 0 - } - try: - storage_type = type(huey.storage).__name__ - - if storage_type == 'FileStorage' and hasattr(huey.storage, 'path'): - # FileStorage: Delete directory contents - storage_path = huey.storage.path - - # Clear queue - queue_dir = os.path.join(storage_path, 'queue') - if os.path.exists(queue_dir): - for root, dirs, files in os.walk(queue_dir): - for f in files: - if not f.startswith('.'): - os.remove(os.path.join(root, f)) - cleared['queue'] += 1 - - # Clear schedule - schedule_dir = os.path.join(storage_path, 'schedule') - if os.path.exists(schedule_dir): - for root, dirs, files in os.walk(schedule_dir): - for f in files: - if not f.startswith('.'): - os.remove(os.path.join(root, f)) - cleared['schedule'] += 1 - - # Clear results - results_dir = os.path.join(storage_path, 'results') - if os.path.exists(results_dir): - for root, dirs, files in os.walk(results_dir): - for f in files: - if not f.startswith('.'): - os.remove(os.path.join(root, f)) - cleared['results'] += 1 - - # Clear retry attempts - attempts_dir = os.path.join(storage_path, 'retry_attempts') - if os.path.exists(attempts_dir): - for f in os.listdir(attempts_dir): - if f.endswith('.json'): - os.remove(os.path.join(attempts_dir, f)) - cleared['retry_attempts'] += 1 - - # Clear task metadata - metadata_dir = os.path.join(storage_path, 'task_metadata') - if os.path.exists(metadata_dir): - for f in os.listdir(metadata_dir): - if f.endswith('.json'): - os.remove(os.path.join(metadata_dir, f)) - cleared['task_metadata'] += 1 - - elif storage_type in ['SqliteStorage', 'SqliteHuey'] and hasattr(huey.storage, 'filename'): - # SqliteStorage: Delete from tables - import sqlite3 - conn = sqlite3.connect(huey.storage.filename) - cursor = conn.cursor() - - cursor.execute("DELETE FROM queue") - cleared['queue'] = cursor.rowcount - - cursor.execute("DELETE FROM schedule") - cleared['schedule'] = cursor.rowcount - - cursor.execute("DELETE FROM results") - cleared['results'] = cursor.rowcount - - conn.commit() - conn.close() - - elif storage_type in ['RedisStorage', 'RedisHuey'] and hasattr(huey.storage, 'conn'): - # RedisStorage: Delete keys - name = huey.storage.name - - # Clear queue (list) - cleared['queue'] = huey.storage.conn.llen(f"{name}:queue") - huey.storage.conn.delete(f"{name}:queue") - - # Clear schedule (sorted set) - cleared['schedule'] = huey.storage.conn.zcard(f"{name}:schedule") - huey.storage.conn.delete(f"{name}:schedule") - - # Clear results (hash or keys) - # Note: This depends on how Huey stores results in Redis - result_keys = huey.storage.conn.keys(f"{name}:result:*") - if result_keys: - cleared['results'] = len(result_keys) - huey.storage.conn.delete(*result_keys) - + cleared = task_manager.clear_all_notifications() logger.warning(f"Cleared all notifications: {cleared}") return cleared - except Exception as e: logger.error(f"Error clearing notifications: {e}", exc_info=True) return {'error': str(e)} @@ -1201,50 +1001,35 @@ def cleanup_old_failed_notifications(max_age_days=30): max_age_days: Delete failed notifications older than this (default: 30 days) Returns: - Number of old failed notifications deleted + Number of old retry attempts deleted """ if huey is None: return 0 import time - import os - deleted_count = 0 try: - # Use get_failed_notifications with auto-cleanup to handle this - # It already has logic to delete old failed notifications - # We just call it and let it do the cleanup cutoff_time = time.time() - (max_age_days * 86400) - # FileStorage and other backends handle result storage differently + # Trigger cleanup of old failed notifications # The get_failed_notifications function already handles cleanup - # So we just trigger it here get_failed_notifications(limit=1000, max_age_days=max_age_days) - # Also clean up old retry attempt files - if hasattr(huey.storage, 'path'): - attempts_dir = os.path.join(huey.storage.path, 'retry_attempts') - if os.path.exists(attempts_dir): - for filename in os.listdir(attempts_dir): - if filename.endswith('.json'): - filepath = os.path.join(attempts_dir, filename) - try: - file_mtime = os.path.getmtime(filepath) - if file_mtime < cutoff_time: - os.remove(filepath) - deleted_count += 1 - except Exception as fe: - logger.debug(f"Unable to delete old retry attempt file {filename}: {fe}") - - if deleted_count > 0: - logger.info(f"Cleaned up {deleted_count} old retry attempt files (older than {max_age_days} days)") + # Clean up old retry attempts using task manager + task_manager = _get_task_manager() + if task_manager: + deleted_count = task_manager.cleanup_old_retry_attempts(cutoff_time) + if deleted_count > 0: + logger.info(f"Cleaned up {deleted_count} old retry attempt files (older than {max_age_days} days)") + else: + deleted_count = 0 logger.info(f"Completed cleanup check for failed notifications older than {max_age_days} days") + return deleted_count except Exception as e: logger.debug(f"Unable to cleanup old failed notifications: {e}") - - return deleted_count + return 0 def start_huey_consumer(): diff --git a/changedetectionio/notification/task_queue/base.py b/changedetectionio/notification/task_queue/base.py new file mode 100644 index 000000000..aca0b1bde --- /dev/null +++ b/changedetectionio/notification/task_queue/base.py @@ -0,0 +1,127 @@ +""" +Abstract base class for Huey storage backend task managers. +""" + +from abc import ABC, abstractmethod + + +class HueyTaskManager(ABC): + """ + Abstract base class for Huey storage backend operations. + + Provides a polymorphic interface for storage-specific operations like: + - Enumerating results (failed notifications) + - Deleting results + - Counting pending notifications + - Clearing all notifications + + Each storage backend (FileStorage, SqliteStorage, RedisStorage) has its own + concrete implementation that knows how to interact with that specific storage. + """ + + def __init__(self, storage, storage_path=None): + """ + Initialize task manager with storage instance. + + Args: + storage: Huey storage instance + storage_path: Optional path for file-based storage + """ + self.storage = storage + self.storage_path = storage_path + + @abstractmethod + def enumerate_results(self): + """ + Enumerate all results from storage. + + Returns: + dict: {task_id: result_data} for all stored results + """ + pass + + @abstractmethod + def delete_result(self, task_id): + """ + Delete a result from storage. + + Args: + task_id: Task ID to delete + + Returns: + bool: True if deleted successfully, False otherwise + """ + pass + + @abstractmethod + def count_storage_items(self): + """ + Count items in storage (queue + schedule). + + Returns: + tuple: (queue_count, schedule_count) + """ + pass + + @abstractmethod + def clear_all_notifications(self): + """ + Clear all notifications (queue, schedule, results, metadata). + + Returns: + dict: Counts of cleared items by type + """ + pass + + @abstractmethod + def store_task_metadata(self, task_id, metadata): + """ + Store task metadata for later retrieval. + + Args: + task_id: Task ID + metadata: Metadata dictionary to store + + Returns: + bool: True if stored successfully, False otherwise + """ + pass + + @abstractmethod + def get_task_metadata(self, task_id): + """ + Retrieve task metadata. + + Args: + task_id: Task ID + + Returns: + dict: Metadata dictionary or None if not found + """ + pass + + @abstractmethod + def delete_task_metadata(self, task_id): + """ + Delete task metadata. + + Args: + task_id: Task ID + + Returns: + bool: True if deleted successfully, False otherwise + """ + pass + + @abstractmethod + def cleanup_old_retry_attempts(self, cutoff_time): + """ + Clean up retry attempt records older than cutoff_time. + + Args: + cutoff_time: Unix timestamp - delete records older than this + + Returns: + int: Number of retry attempts deleted + """ + pass diff --git a/changedetectionio/notification/task_queue/file_storage.py b/changedetectionio/notification/task_queue/file_storage.py new file mode 100644 index 000000000..7a352d71e --- /dev/null +++ b/changedetectionio/notification/task_queue/file_storage.py @@ -0,0 +1,270 @@ +""" +FileStorage backend task manager for Huey notifications. + +This is the default backend, optimized for NAS/CIFS compatibility. +""" + +from loguru import logger + +from .base import HueyTaskManager + +import os + +class FileStorageTaskManager(HueyTaskManager): + """Task manager for FileStorage backend (default, NAS-safe).""" + + def enumerate_results(self): + """Enumerate results by walking filesystem directories.""" + import os + import pickle + import struct + import time + + results = {} + + if not self.storage_path: + return results + + results_dir = os.path.join(self.storage_path, 'results') + + if not os.path.exists(results_dir): + return results + + # Walk through all subdirectories to find result files + for root, dirs, files in os.walk(results_dir): + for filename in files: + if filename.startswith('.'): + continue + + filepath = os.path.join(root, filename) + try: + # Read and unpickle the result + # Huey FileStorage format: 4-byte length + task_id + pickled data + with open(filepath, 'rb') as f: + # Read the task ID header (length-prefixed) + task_id_len_bytes = f.read(4) + if len(task_id_len_bytes) < 4: + raise EOFError("Incomplete header") + task_id_len = struct.unpack('>I', task_id_len_bytes)[0] + task_id_bytes = f.read(task_id_len) + if len(task_id_bytes) < task_id_len: + raise EOFError("Incomplete task ID") + task_id = task_id_bytes.decode('utf-8') + + # Now unpickle the result data + result_data = pickle.load(f) + results[task_id] = result_data + except (pickle.UnpicklingError, EOFError) as e: + # Corrupted or incomplete result file + file_size = os.path.getsize(filepath) + logger.warning(f"Corrupted result file {filename} ({file_size} bytes) - moving to lost-found.") + try: + import shutil + lost_found_dir = os.path.join(self.storage_path, 'lost-found', 'results') + os.makedirs(lost_found_dir, exist_ok=True) + + timestamp = int(time.time()) + lost_found_path = os.path.join(lost_found_dir, f"{filename}.{timestamp}.corrupted") + + shutil.move(filepath, lost_found_path) + logger.info(f"Moved corrupted file to {lost_found_path}") + except Exception as move_err: + logger.error(f"Unable to move corrupted file: {move_err}") + except Exception as e: + logger.debug(f"Unable to read result file {filename}: {e}") + + return results + + def delete_result(self, task_id): + """Delete result file from filesystem.""" + import hashlib + + if not self.storage_path: + return False + + results_dir = os.path.join(self.storage_path, 'results') + + # Huey uses MD5 hash to create subdirectories + task_id_bytes = task_id.encode('utf-8') + hex_hash = hashlib.md5(task_id_bytes).hexdigest() + + # FileStorage creates subdirectories based on first 2 chars of hash + subdir = hex_hash[:2] + result_file = os.path.join(results_dir, subdir, task_id) + + if os.path.exists(result_file): + os.remove(result_file) + logger.debug(f"Deleted result file for task {task_id}") + return True + else: + logger.debug(f"Result file not found for task {task_id}") + return False + + def count_storage_items(self): + """Count items by walking filesystem directories.""" + queue_count = 0 + schedule_count = 0 + + if not self.storage_path: + return queue_count, schedule_count + + try: + # Count queue files + queue_dir = os.path.join(self.storage_path, 'queue') + if os.path.exists(queue_dir): + for root, dirs, files in os.walk(queue_dir): + queue_count += len([f for f in files if not f.startswith('.')]) + + # Count schedule files + schedule_dir = os.path.join(self.storage_path, 'schedule') + if os.path.exists(schedule_dir): + for root, dirs, files in os.walk(schedule_dir): + schedule_count += len([f for f in files if not f.startswith('.')]) + except Exception as e: + logger.debug(f"FileStorage count error: {e}") + + return queue_count, schedule_count + + def clear_all_notifications(self): + """Clear all notification files from filesystem.""" + cleared = { + 'queue': 0, + 'schedule': 0, + 'results': 0, + 'retry_attempts': 0, + 'task_metadata': 0 + } + + if not self.storage_path: + return cleared + + # Clear queue + queue_dir = os.path.join(self.storage_path, 'queue') + if os.path.exists(queue_dir): + for root, dirs, files in os.walk(queue_dir): + for f in files: + if not f.startswith('.'): + os.remove(os.path.join(root, f)) + cleared['queue'] += 1 + + # Clear schedule + schedule_dir = os.path.join(self.storage_path, 'schedule') + if os.path.exists(schedule_dir): + for root, dirs, files in os.walk(schedule_dir): + for f in files: + if not f.startswith('.'): + os.remove(os.path.join(root, f)) + cleared['schedule'] += 1 + + # Clear results + results_dir = os.path.join(self.storage_path, 'results') + if os.path.exists(results_dir): + for root, dirs, files in os.walk(results_dir): + for f in files: + if not f.startswith('.'): + os.remove(os.path.join(root, f)) + cleared['results'] += 1 + + # Clear retry attempts + attempts_dir = os.path.join(self.storage_path, 'retry_attempts') + if os.path.exists(attempts_dir): + for f in os.listdir(attempts_dir): + if f.endswith('.json'): + os.remove(os.path.join(attempts_dir, f)) + cleared['retry_attempts'] += 1 + + # Clear task metadata + metadata_dir = os.path.join(self.storage_path, 'task_metadata') + if os.path.exists(metadata_dir): + for f in os.listdir(metadata_dir): + if f.endswith('.json'): + os.remove(os.path.join(metadata_dir, f)) + cleared['task_metadata'] += 1 + + return cleared + + def store_task_metadata(self, task_id, metadata): + """Store task metadata as JSON file.""" + import json + import time + + if not self.storage_path: + return False + + try: + metadata_dir = os.path.join(self.storage_path, 'task_metadata') + os.makedirs(metadata_dir, exist_ok=True) + + metadata_file = os.path.join(metadata_dir, f"{task_id}.json") + metadata_with_id = { + 'task_id': task_id, + 'timestamp': time.time(), + **metadata + } + + with open(metadata_file, 'w') as f: + json.dump(metadata_with_id, f, indent=2) + return True + except Exception as e: + logger.debug(f"Unable to store task metadata: {e}") + return False + + def get_task_metadata(self, task_id): + """Retrieve task metadata from JSON file.""" + import json + + if not self.storage_path: + return None + + try: + metadata_dir = os.path.join(self.storage_path, 'task_metadata') + metadata_file = os.path.join(metadata_dir, f"{task_id}.json") + + if os.path.exists(metadata_file): + with open(metadata_file, 'r') as f: + return json.load(f) + except Exception as e: + logger.debug(f"Unable to load task metadata for {task_id}: {e}") + + return None + + def delete_task_metadata(self, task_id): + """Delete task metadata JSON file.""" + if not self.storage_path: + return False + + try: + metadata_dir = os.path.join(self.storage_path, 'task_metadata') + metadata_file = os.path.join(metadata_dir, f"{task_id}.json") + + if os.path.exists(metadata_file): + os.remove(metadata_file) + return True + return False + except Exception as e: + logger.debug(f"Unable to delete task metadata for {task_id}: {e}") + return False + + def cleanup_old_retry_attempts(self, cutoff_time): + """Clean up old retry attempt files from filesystem.""" + if not self.storage_path: + return 0 + + deleted_count = 0 + try: + attempts_dir = os.path.join(self.storage_path, 'retry_attempts') + if os.path.exists(attempts_dir): + for filename in os.listdir(attempts_dir): + if filename.endswith('.json'): + filepath = os.path.join(attempts_dir, filename) + try: + file_mtime = os.path.getmtime(filepath) + if file_mtime < cutoff_time: + os.remove(filepath) + deleted_count += 1 + except Exception as fe: + logger.debug(f"Unable to delete old retry attempt file {filename}: {fe}") + except Exception as e: + logger.debug(f"Error cleaning up old retry attempts: {e}") + + return deleted_count diff --git a/changedetectionio/notification/task_queue/redis_storage.py b/changedetectionio/notification/task_queue/redis_storage.py new file mode 100644 index 000000000..ccd2458b4 --- /dev/null +++ b/changedetectionio/notification/task_queue/redis_storage.py @@ -0,0 +1,204 @@ +""" +RedisStorage backend task manager for Huey notifications. + +For distributed deployments with Redis. +""" + +from loguru import logger + +from .base import HueyTaskManager + + +class RedisStorageTaskManager(HueyTaskManager): + """Task manager for RedisStorage backend (distributed deployments).""" + + def enumerate_results(self): + import pickle + """Enumerate results using Redis commands.""" + results = {} + + if not hasattr(self.storage, 'conn'): + return results + + try: + # Redis stores results with keys like "{name}:result:{task_id}" + name = self.storage.name + pattern = f"{name}:result:*" + + # Get all result keys + result_keys = self.storage.conn.keys(pattern) + + for key in result_keys: + # Extract task_id from key + task_id = key.decode('utf-8').split(':')[-1] + + # Get result data + result_data = self.storage.conn.get(key) + if result_data: + results[task_id] = pickle.loads(result_data) + except Exception as e: + logger.error(f"Error enumerating Redis results: {e}") + + return results + + def delete_result(self, task_id): + """Delete result from Redis.""" + if not hasattr(self.storage, 'conn'): + return False + + try: + name = self.storage.name + result_key = f"{name}:result:{task_id}" + deleted = self.storage.conn.delete(result_key) > 0 + logger.debug(f"Deleted result from Redis for task {task_id}: {deleted}") + return deleted + except Exception as e: + logger.error(f"Error deleting Redis result: {e}") + return False + + def count_storage_items(self): + """Count items using Redis commands.""" + queue_count = 0 + schedule_count = 0 + + if not hasattr(self.storage, 'conn'): + return queue_count, schedule_count + + try: + name = self.storage.name + + # Queue is a list + queue_count = self.storage.conn.llen(f"{name}:queue") + + # Schedule is a sorted set + schedule_count = self.storage.conn.zcard(f"{name}:schedule") + except Exception as e: + logger.debug(f"Redis count error: {e}") + + return queue_count, schedule_count + + def clear_all_notifications(self): + """Clear all notifications from Redis.""" + cleared = { + 'queue': 0, + 'schedule': 0, + 'results': 0, + 'retry_attempts': 0, + 'task_metadata': 0 + } + + if not hasattr(self.storage, 'conn'): + return cleared + + try: + name = self.storage.name + + # Clear queue (list) + cleared['queue'] = self.storage.conn.llen(f"{name}:queue") + self.storage.conn.delete(f"{name}:queue") + + # Clear schedule (sorted set) + cleared['schedule'] = self.storage.conn.zcard(f"{name}:schedule") + self.storage.conn.delete(f"{name}:schedule") + + # Clear results (keys) + result_keys = self.storage.conn.keys(f"{name}:result:*") + if result_keys: + cleared['results'] = len(result_keys) + self.storage.conn.delete(*result_keys) + except Exception as e: + logger.error(f"Error clearing Redis notifications: {e}") + + return cleared + + def store_task_metadata(self, task_id, metadata): + """Store task metadata in Redis.""" + import json + import time + + if not hasattr(self.storage, 'conn'): + return False + + try: + name = self.storage.name + metadata_key = f"{name}:metadata:{task_id}" + + metadata_with_id = { + 'task_id': task_id, + 'timestamp': time.time(), + **metadata + } + + self.storage.conn.set(metadata_key, json.dumps(metadata_with_id)) + return True + except Exception as e: + logger.error(f"Error storing Redis task metadata: {e}") + return False + + def get_task_metadata(self, task_id): + """Retrieve task metadata from Redis.""" + import json + + if not hasattr(self.storage, 'conn'): + return None + + try: + name = self.storage.name + metadata_key = f"{name}:metadata:{task_id}" + + data = self.storage.conn.get(metadata_key) + if data: + return json.loads(data.decode('utf-8') if isinstance(data, bytes) else data) + return None + except Exception as e: + logger.debug(f"Error retrieving Redis task metadata: {e}") + return None + + def delete_task_metadata(self, task_id): + """Delete task metadata from Redis.""" + if not hasattr(self.storage, 'conn'): + return False + + try: + name = self.storage.name + metadata_key = f"{name}:metadata:{task_id}" + deleted = self.storage.conn.delete(metadata_key) > 0 + return deleted + except Exception as e: + logger.debug(f"Error deleting Redis task metadata: {e}") + return False + + def cleanup_old_retry_attempts(self, cutoff_time): + """Clean up old retry attempts from Redis.""" + if not hasattr(self.storage, 'conn'): + return 0 + + deleted_count = 0 + try: + name = self.storage.name + pattern = f"{name}:retry_attempts:*" + + # Get all retry attempt keys + retry_keys = self.storage.conn.keys(pattern) + + for key in retry_keys: + try: + # Get the timestamp from the key's data + data = self.storage.conn.get(key) + if data: + import json + attempt_data = json.loads(data.decode('utf-8') if isinstance(data, bytes) else data) + timestamp = attempt_data.get('timestamp', 0) + + if timestamp < cutoff_time: + self.storage.conn.delete(key) + deleted_count += 1 + except Exception as ke: + logger.debug(f"Error checking retry attempt key: {ke}") + + if deleted_count > 0: + logger.info(f"Cleaned up {deleted_count} old retry attempts from Redis") + except Exception as e: + logger.debug(f"Error cleaning up old Redis retry attempts: {e}") + + return deleted_count diff --git a/changedetectionio/notification/task_queue/sqlite_storage.py b/changedetectionio/notification/task_queue/sqlite_storage.py new file mode 100644 index 000000000..988c51ce2 --- /dev/null +++ b/changedetectionio/notification/task_queue/sqlite_storage.py @@ -0,0 +1,239 @@ +""" +SQLiteStorage backend task manager for Huey notifications. + +WARNING: Only use on local disk storage, NOT on NFS/CIFS network storage! +""" + +from loguru import logger + +from .base import HueyTaskManager + + +class SqliteStorageTaskManager(HueyTaskManager): + """Task manager for SqliteStorage backend (local disk only).""" + + def enumerate_results(self): + import pickle + import sqlite3 + """Enumerate results by querying SQLite database.""" + results = {} + + if not hasattr(self.storage, 'filename'): + return results + + try: + conn = sqlite3.connect(self.storage.filename) + cursor = conn.cursor() + + # Query all results from database + cursor.execute("SELECT key, value FROM results") + for row in cursor.fetchall(): + task_id = row[0] + result_data = pickle.loads(row[1]) + results[task_id] = result_data + + conn.close() + except Exception as e: + logger.error(f"Error enumerating SQLite results: {e}") + + return results + + def delete_result(self, task_id): + """Delete result from SQLite database.""" + if not hasattr(self.storage, 'filename'): + return False + import sqlite3 + try: + conn = sqlite3.connect(self.storage.filename) + cursor = conn.cursor() + cursor.execute("DELETE FROM results WHERE key = ?", (task_id,)) + conn.commit() + deleted = cursor.rowcount > 0 + conn.close() + logger.debug(f"Deleted result from SQLite for task {task_id}: {deleted}") + return deleted + except Exception as e: + logger.error(f"Error deleting SQLite result: {e}") + return False + + def count_storage_items(self): + """Count items by querying SQLite database.""" + queue_count = 0 + schedule_count = 0 + + if not hasattr(self.storage, 'filename'): + return queue_count, schedule_count + import sqlite3 + try: + conn = sqlite3.connect(self.storage.filename) + cursor = conn.cursor() + + cursor.execute("SELECT COUNT(*) FROM queue") + queue_count = cursor.fetchone()[0] + + cursor.execute("SELECT COUNT(*) FROM schedule") + schedule_count = cursor.fetchone()[0] + + conn.close() + except Exception as e: + logger.debug(f"SQLite count error: {e}") + + return queue_count, schedule_count + + def clear_all_notifications(self): + """Clear all notifications from SQLite database.""" + cleared = { + 'queue': 0, + 'schedule': 0, + 'results': 0, + 'retry_attempts': 0, + 'task_metadata': 0 + } + + if not hasattr(self.storage, 'filename'): + return cleared + import sqlite3 + try: + conn = sqlite3.connect(self.storage.filename) + cursor = conn.cursor() + + cursor.execute("DELETE FROM queue") + cleared['queue'] = cursor.rowcount + + cursor.execute("DELETE FROM schedule") + cleared['schedule'] = cursor.rowcount + + cursor.execute("DELETE FROM results") + cleared['results'] = cursor.rowcount + + conn.commit() + conn.close() + except Exception as e: + logger.error(f"Error clearing SQLite notifications: {e}") + + return cleared + + def store_task_metadata(self, task_id, metadata): + """Store task metadata in SQLite database.""" + import sqlite3 + import json + import time + + if not hasattr(self.storage, 'filename'): + return False + + try: + conn = sqlite3.connect(self.storage.filename) + cursor = conn.cursor() + + # Create table if it doesn't exist + cursor.execute(""" + CREATE TABLE IF NOT EXISTS task_metadata ( + task_id TEXT PRIMARY KEY, + timestamp REAL, + metadata TEXT + ) + """) + + metadata_with_id = { + 'task_id': task_id, + 'timestamp': time.time(), + **metadata + } + + cursor.execute( + "INSERT OR REPLACE INTO task_metadata (task_id, timestamp, metadata) VALUES (?, ?, ?)", + (task_id, time.time(), json.dumps(metadata_with_id)) + ) + conn.commit() + conn.close() + return True + except Exception as e: + logger.error(f"Error storing SQLite task metadata: {e}") + return False + + def get_task_metadata(self, task_id): + """Retrieve task metadata from SQLite database.""" + import sqlite3 + import json + + if not hasattr(self.storage, 'filename'): + return None + + try: + conn = sqlite3.connect(self.storage.filename) + cursor = conn.cursor() + + # Check if table exists + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='task_metadata'") + if not cursor.fetchone(): + conn.close() + return None + + cursor.execute("SELECT metadata FROM task_metadata WHERE task_id = ?", (task_id,)) + row = cursor.fetchone() + conn.close() + + if row: + return json.loads(row[0]) + return None + except Exception as e: + logger.debug(f"Error retrieving SQLite task metadata: {e}") + return None + + def delete_task_metadata(self, task_id): + """Delete task metadata from SQLite database.""" + import sqlite3 + + if not hasattr(self.storage, 'filename'): + return False + + try: + conn = sqlite3.connect(self.storage.filename) + cursor = conn.cursor() + + # Check if table exists + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='task_metadata'") + if not cursor.fetchone(): + conn.close() + return False + + cursor.execute("DELETE FROM task_metadata WHERE task_id = ?", (task_id,)) + conn.commit() + deleted = cursor.rowcount > 0 + conn.close() + return deleted + except Exception as e: + logger.debug(f"Error deleting SQLite task metadata: {e}") + return False + + def cleanup_old_retry_attempts(self, cutoff_time): + """Clean up old retry attempts from SQLite database.""" + import sqlite3 + + if not hasattr(self.storage, 'filename'): + return 0 + + deleted_count = 0 + try: + conn = sqlite3.connect(self.storage.filename) + cursor = conn.cursor() + + # Check if retry_attempts table exists + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='retry_attempts'") + if not cursor.fetchone(): + conn.close() + return 0 + + # Delete old retry attempts + cursor.execute("DELETE FROM retry_attempts WHERE timestamp < ?", (cutoff_time,)) + deleted_count = cursor.rowcount + conn.commit() + conn.close() + + if deleted_count > 0: + logger.info(f"Cleaned up {deleted_count} old retry attempts from SQLite") + except Exception as e: + logger.debug(f"Error cleaning up old SQLite retry attempts: {e}") + + return deleted_count diff --git a/changedetectionio/static/images/notification-fail.svg b/changedetectionio/static/images/notification-fail.svg new file mode 100644 index 000000000..1d7d9d096 --- /dev/null +++ b/changedetectionio/static/images/notification-fail.svg @@ -0,0 +1,10 @@ + + + + + + + ! + diff --git a/changedetectionio/tests/unit/test_huey_filestorage.py b/changedetectionio/tests/unit/test_huey_filestorage.py new file mode 100644 index 000000000..52004010e --- /dev/null +++ b/changedetectionio/tests/unit/test_huey_filestorage.py @@ -0,0 +1,161 @@ +""" +Unit tests for Huey FileStorage task manager. + +Tests the basic functionality of the FileStorage task manager without requiring +a full Huey instance or changedetection.io app. +""" + +import pytest +import tempfile +import shutil +import os +from changedetectionio.notification.task_queue.file_storage import FileStorageTaskManager + + +class MockStorage: + """Mock storage object to simulate Huey FileStorage.""" + def __init__(self, path): + self.path = path + + +@pytest.fixture +def temp_storage_dir(): + """Create a temporary directory for testing.""" + temp_dir = tempfile.mkdtemp() + yield temp_dir + shutil.rmtree(temp_dir, ignore_errors=True) + + +@pytest.fixture +def task_manager(temp_storage_dir): + """Create a FileStorageTaskManager instance for testing.""" + mock_storage = MockStorage(temp_storage_dir) + return FileStorageTaskManager(mock_storage, temp_storage_dir) + + +class TestFileStorageTaskManager: + """Tests for FileStorageTaskManager basic functionality.""" + + def test_store_and_get_metadata(self, task_manager): + """Test storing and retrieving task metadata.""" + task_id = "test-task-123" + # Use realistic notification data structure matching actual app usage + metadata = { + 'notification_data': { + 'watch_url': 'https://example.com/test', + 'uuid': 'test-watch-uuid-123', + 'current_snapshot': 'Test content snapshot', + 'diff': '+ New content added\n- Old content removed', + 'diff_clean': 'New content added\nOld content removed', + 'triggered_text': 'price: $99.99', + 'notification_urls': ['mailto://test@example.com'], + 'notification_title': 'Change detected on example.com', + 'notification_body': 'The page has changed', + 'notification_format': 'HTML' + } + } + + # Store metadata + result = task_manager.store_task_metadata(task_id, metadata) + assert result is True, "Should successfully store metadata" + + # Retrieve metadata + retrieved = task_manager.get_task_metadata(task_id) + assert retrieved is not None, "Should retrieve stored metadata" + assert retrieved['task_id'] == task_id + assert 'timestamp' in retrieved + assert retrieved['notification_data'] == metadata['notification_data'] + + def test_delete_metadata(self, task_manager): + """Test deleting task metadata.""" + task_id = "test-task-456" + metadata = {'notification_data': {'test': 'data'}} + + # Store then delete + task_manager.store_task_metadata(task_id, metadata) + result = task_manager.delete_task_metadata(task_id) + assert result is True, "Should successfully delete metadata" + + # Verify it's gone + retrieved = task_manager.get_task_metadata(task_id) + assert retrieved is None, "Metadata should be deleted" + + def test_delete_nonexistent_metadata(self, task_manager): + """Test deleting metadata that doesn't exist.""" + result = task_manager.delete_task_metadata("nonexistent-task") + assert result is False, "Should return False for nonexistent metadata" + + def test_get_nonexistent_metadata(self, task_manager): + """Test retrieving metadata that doesn't exist.""" + retrieved = task_manager.get_task_metadata("nonexistent-task") + assert retrieved is None, "Should return None for nonexistent metadata" + + def test_count_storage_items_empty(self, task_manager): + """Test counting storage items when empty.""" + queue_count, schedule_count = task_manager.count_storage_items() + assert queue_count == 0, "Empty queue should have 0 items" + assert schedule_count == 0, "Empty schedule should have 0 items" + + def test_count_storage_items_with_files(self, task_manager, temp_storage_dir): + """Test counting storage items with files present.""" + # Create some queue files + queue_dir = os.path.join(temp_storage_dir, 'queue') + os.makedirs(queue_dir, exist_ok=True) + + for i in range(3): + with open(os.path.join(queue_dir, f"task-{i}"), 'w') as f: + f.write("test") + + # Create some schedule files + schedule_dir = os.path.join(temp_storage_dir, 'schedule') + os.makedirs(schedule_dir, exist_ok=True) + + for i in range(2): + with open(os.path.join(schedule_dir, f"scheduled-{i}"), 'w') as f: + f.write("test") + + queue_count, schedule_count = task_manager.count_storage_items() + assert queue_count == 3, "Should count 3 queue items" + assert schedule_count == 2, "Should count 2 schedule items" + + def test_clear_all_notifications(self, task_manager, temp_storage_dir): + """Test clearing all notifications.""" + # Create test files in various directories + for subdir in ['queue', 'schedule', 'results']: + dir_path = os.path.join(temp_storage_dir, subdir) + os.makedirs(dir_path, exist_ok=True) + with open(os.path.join(dir_path, 'test-file'), 'w') as f: + f.write("test") + + # Create metadata files + metadata_dir = os.path.join(temp_storage_dir, 'task_metadata') + os.makedirs(metadata_dir, exist_ok=True) + with open(os.path.join(metadata_dir, 'test-task.json'), 'w') as f: + f.write('{"test": "data"}') + + # Clear all + cleared = task_manager.clear_all_notifications() + + assert cleared['queue'] == 1, "Should clear 1 queue file" + assert cleared['schedule'] == 1, "Should clear 1 schedule file" + assert cleared['results'] == 1, "Should clear 1 result file" + assert cleared['task_metadata'] == 1, "Should clear 1 metadata file" + + def test_metadata_file_structure(self, task_manager, temp_storage_dir): + """Test that metadata files are created in the correct structure.""" + task_id = "test-structure-789" + metadata = {'notification_data': {'test': 'value'}} + + task_manager.store_task_metadata(task_id, metadata) + + # Check file exists in correct location + expected_path = os.path.join(temp_storage_dir, 'task_metadata', f"{task_id}.json") + assert os.path.exists(expected_path), f"Metadata file should exist at {expected_path}" + + # Check file contains valid JSON + import json + with open(expected_path, 'r') as f: + data = json.load(f) + assert data['task_id'] == task_id + assert 'timestamp' in data + assert 'notification_data' in data