From ff8a3a1009611b93b9b071bded3cb7b7580e3348 Mon Sep 17 00:00:00 2001
From: Jeroen Schweitzer <jpmschweitzer@gmail.com>
Date: Sun, 7 Dec 2025 23:13:07 +0100
Subject: [PATCH] feat(scheduler): add task executors for common operations
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add three built-in task executors for various automation tasks.

Executors:
1. example_executor - Simple test implementation with configurable message and delay
2. doc_sync_executor - Mirror documentation from upstream Git repos to Gitea
3. config_backup_executor - Backup Docker configs and data directories

doc_sync_executor features:
- Clones upstream repository (GitHub, GitLab, etc.)
- Supports full repository mirroring or selective path syncing
- Pushes to Gitea with authentication
- Creates date-tagged snapshots (YYYY-MM-DD)
- Generates .SYNC_INFO.md with sync metadata

config_backup_executor features:
- Backs up multiple source paths with exclusion patterns
- Optional compression (tar.gz)
- Retention policy (days-based cleanup)
- Timestamped backups

Pre-configured tasks:
- backup_docker_configs_daily (priority 20, daily 03:05)
- sync_fastapi_docs_monthly (priority 60, 11th @ 04:00)
- sync_ollama_docs_monthly (priority 60, 12th @ 04:00)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
---
 services/scheduler/src/executors/__init__.py  |  23 ++
 .../src/executors/config_backup_executor.py   | 159 +++++++++++
 .../src/executors/doc_sync_executor.py        | 259 ++++++++++++++++++
 .../src/executors/example_executor.py         |  42 +++
 4 files changed, 483 insertions(+)
 create mode 100644 services/scheduler/src/executors/__init__.py
 create mode 100644 services/scheduler/src/executors/config_backup_executor.py
 create mode 100644 services/scheduler/src/executors/doc_sync_executor.py
 create mode 100644 services/scheduler/src/executors/example_executor.py

diff --git a/services/scheduler/src/executors/__init__.py b/services/scheduler/src/executors/__init__.py
new file mode 100644
index 0000000..743cf48
--- /dev/null
+++ b/services/scheduler/src/executors/__init__.py
@@ -0,0 +1,23 @@
+"""
+Executor modules for The Scheduler.
+
+Each executor must implement an async execute(config, settings) function.
+
+Example:
+    async def execute(config: dict, settings: Settings) -> str:
+        '''
+        Perform the task.
+
+        Args:
+            config: Task-specific configuration from scheduled_tasks.config
+            settings: Global scheduler settings
+
+        Returns:
+            Output message (success)
+
+        Raises:
+            Exception: On failure (will be logged as error)
+        '''
+        # Your task logic here
+        return "Task completed successfully"
+"""
diff --git a/services/scheduler/src/executors/config_backup_executor.py b/services/scheduler/src/executors/config_backup_executor.py
new file mode 100644
index 0000000..ef61728
--- /dev/null
+++ b/services/scheduler/src/executors/config_backup_executor.py
@@ -0,0 +1,159 @@
+"""
+Config Backup Executor
+Backs up Docker container configs and host-based service configs.
+Replicates functionality of maintenance container's backup-configs.sh
+"""
+import asyncio
+import logging
+import tarfile
+import tempfile
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import List, Dict, Any
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Execute config backup task.
+
+    Config schema:
+        {
+            "sources": [
+                {
+                    "path": "/data/docker-data",
+                    "name": "docker-data",
+                    "excludes": ["*/cache/*", "*/temp/*", "*.log"]
+                }
+            ],
+            "backup_dir": "/backups/docker-configs",
+            "retention_days": 30,
+            "compress": true
+        }
+
+    Args:
+        config: Backup configuration
+        settings: Global scheduler settings
+
+    Returns:
+        Summary of backup operation
+
+    Raises:
+        Exception: On backup failure
+    """
+    sources = config.get('sources', [])
+    backup_dir = Path(config.get('backup_dir', '/backups/docker-configs'))
+    retention_days = config.get('retention_days', 30)
+    compress = config.get('compress', True)
+
+    if not sources:
+        raise ValueError("No backup sources configured")
+
+    timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
+    backup_filename = f"docker-configs-{timestamp}.tar.gz"
+    backup_file = backup_dir / backup_filename
+
+    logger.info(f"Starting Docker configs backup: {backup_filename}")
+
+    # Create backup directory
+    backup_dir.mkdir(parents=True, exist_ok=True)
+
+    # Create temporary directory for staging
+    with tempfile.TemporaryDirectory(prefix='backup-') as temp_dir:
+        temp_path = Path(temp_dir)
+        results = []
+
+        # Backup each source
+        for source in sources:
+            source_path = Path(source['path'])
+            source_name = source['name']
+            excludes = source.get('excludes', [])
+
+            if not source_path.exists():
+                logger.warning(f"Source path does not exist: {source_path}")
+                continue
+
+            logger.info(f"Backing up {source_name} from {source_path}")
+
+            # Create tar for this source
+            source_tar = temp_path / f"{source_name}.tar.gz"
+
+            def tar_filter(tarinfo):
+                """Filter function to exclude patterns."""
+                for pattern in excludes:
+                    # Simple pattern matching (could be enhanced with fnmatch)
+                    if pattern.replace('*/', '').replace('/*', '') in tarinfo.name:
+                        logger.debug(f"Excluding: {tarinfo.name}")
+                        return None
+                return tarinfo
+
+            with tarfile.open(source_tar, 'w:gz') as tar:
+                tar.add(
+                    source_path,
+                    arcname=source_name,
+                    filter=tar_filter,
+                    recursive=True
+                )
+
+            source_size = source_tar.stat().st_size / (1024 * 1024)  # MB
+            results.append(f"{source_name}: {source_size:.2f}MB")
+            logger.info(f"Backed up {source_name}: {source_size:.2f}MB")
+
+        # Combine all source backups into final archive
+        logger.info("Creating combined backup archive...")
+        with tarfile.open(backup_file, 'w:gz') as final_tar:
+            for item in temp_path.glob('*.tar.gz'):
+                final_tar.add(item, arcname=item.name)
+
+    # Verify backup created
+    if not backup_file.exists():
+        raise Exception("Backup file was not created")
+
+    backup_size = backup_file.stat().st_size / (1024 * 1024)  # MB
+    logger.info(f"Backup created successfully: {backup_size:.2f}MB")
+
+    # Clean up old backups
+    await cleanup_old_backups(backup_dir, retention_days)
+
+    # Count remaining backups
+    backup_count = len(list(backup_dir.glob('docker-configs-*.tar.gz')))
+    total_size = sum(f.stat().st_size for f in backup_dir.glob('docker-configs-*.tar.gz'))
+    total_size_mb = total_size / (1024 * 1024)
+
+    output = (
+        f"Backup completed: {backup_filename} ({backup_size:.2f}MB). "
+        f"Sources: {', '.join(results)}. "
+        f"Retention: {backup_count} backups, {total_size_mb:.2f}MB total."
+    )
+
+    logger.info(output)
+    return output
+
+
+async def cleanup_old_backups(backup_dir: Path, retention_days: int):
+    """Remove backups older than retention period."""
+    cutoff_date = datetime.now() - timedelta(days=retention_days)
+    removed_count = 0
+    removed_size = 0
+
+    logger.info(f"Cleaning up backups older than {retention_days} days...")
+
+    for backup_file in backup_dir.glob('docker-configs-*.tar.gz'):
+        # Get file modification time
+        file_mtime = datetime.fromtimestamp(backup_file.stat().st_mtime)
+
+        if file_mtime < cutoff_date:
+            file_size = backup_file.stat().st_size
+            logger.info(f"Removing old backup: {backup_file.name} (from {file_mtime:%Y-%m-%d})")
+            backup_file.unlink()
+            removed_count += 1
+            removed_size += file_size
+
+    if removed_count > 0:
+        removed_size_mb = removed_size / (1024 * 1024)
+        logger.info(f"Removed {removed_count} old backups, freed {removed_size_mb:.2f}MB")
+    else:
+        logger.info("No old backups to remove")
diff --git a/services/scheduler/src/executors/doc_sync_executor.py b/services/scheduler/src/executors/doc_sync_executor.py
new file mode 100644
index 0000000..0a8f5d4
--- /dev/null
+++ b/services/scheduler/src/executors/doc_sync_executor.py
@@ -0,0 +1,259 @@
+"""
+Documentation Sync Executor
+Syncs documentation from upstream Git repositories to Gitea.
+Clones source repos, extracts docs directories, pushes to Gitea.
+"""
+import asyncio
+import logging
+import shutil
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Sync documentation from upstream Git repo to Gitea.
+
+    Config schema:
+        {
+            "project": "fastapi",
+            "upstream_repo": "https://github.com/tiangolo/fastapi.git",
+            "docs_paths": ["/docs", "/docs_src"],
+            "gitea_repo": "library/docs-fastapi",
+            "gitea_url": "http://gitea:3000",
+            "branch": "main"
+        }
+
+    Args:
+        config: Documentation sync configuration
+        settings: Global scheduler settings
+
+    Returns:
+        Summary of sync operation
+
+    Raises:
+        Exception: On sync failure
+    """
+    project = config.get('project')
+    upstream_repo = config.get('upstream_repo')
+    docs_paths = config.get('docs_paths', ['/docs'])
+    gitea_repo = config.get('gitea_repo')
+    gitea_url = config.get('gitea_url', settings.gitea_url)
+    branch = config.get('branch', 'main')
+
+    if not all([project, upstream_repo, gitea_repo]):
+        raise ValueError("Missing required config: project, upstream_repo, or gitea_repo")
+
+    logger.info(f"Starting doc sync for {project}")
+    logger.info(f"Upstream: {upstream_repo}")
+    logger.info(f"Gitea: {gitea_repo}")
+
+    # Work directory
+    work_dir = Path(f"/app/task-data/doc-sync/{project}")
+    upstream_dir = work_dir / "upstream"
+    gitea_dir = work_dir / "gitea"
+
+    try:
+        # Clean work directory
+        if work_dir.exists():
+            logger.info(f"Cleaning work directory: {work_dir}")
+            shutil.rmtree(work_dir)
+        work_dir.mkdir(parents=True)
+
+        # Clone upstream repo (shallow clone for speed)
+        logger.info(f"Cloning upstream repo...")
+        await _run_command([
+            'git', 'clone',
+            '--depth', '1',
+            '--branch', branch,
+            upstream_repo,
+            str(upstream_dir)
+        ])
+
+        # Get upstream version/commit
+        upstream_commit = await _get_git_commit(upstream_dir)
+        upstream_date = datetime.now().strftime('%Y-%m-%d')
+        logger.info(f"Upstream commit: {upstream_commit[:8]}")
+
+        # Clone Gitea repo (or create if doesn't exist)
+        # Build authenticated URL for Gitea
+        gitea_url_clean = gitea_url.replace('http://', '').replace('https://', '')
+        gitea_clone_url = f"http://{settings.gitea_user}:{settings.gitea_password}@{gitea_url_clean}/{gitea_repo}.git"
+        # Log without credentials
+        logger.info(f"Cloning Gitea repo: {gitea_url}/{gitea_repo}.git")
+
+        # Try to clone, if fails create new repo
+        try:
+            await _run_command([
+                'git', 'clone',
+                gitea_clone_url,
+                str(gitea_dir)
+            ])
+        except Exception as e:
+            logger.warning(f"Gitea repo doesn't exist, will create: {e}")
+            gitea_dir.mkdir(parents=True)
+            await _run_command(['git', 'init'], cwd=gitea_dir)
+            await _run_command(['git', 'checkout', '-b', branch], cwd=gitea_dir)
+
+            # Set remote
+            await _run_command([
+                'git', 'remote', 'add', 'origin',
+                gitea_clone_url
+            ], cwd=gitea_dir)
+
+        # Clear existing content in Gitea repo (except .git)
+        for item in gitea_dir.iterdir():
+            if item.name != '.git':
+                if item.is_dir():
+                    shutil.rmtree(item)
+                else:
+                    item.unlink()
+
+        # Determine what to copy based on docs_paths
+        copied_paths = []
+
+        # If docs_paths is empty or contains "." or "/", sync entire repo
+        if not docs_paths or any(p in [".", "/", ""] for p in docs_paths):
+            logger.info(f"Copying entire repository...")
+            for item in upstream_dir.iterdir():
+                if item.name != '.git':
+                    dest = gitea_dir / item.name
+                    if item.is_dir():
+                        shutil.copytree(item, dest)
+                    else:
+                        shutil.copy2(item, dest)
+            copied_paths = ["entire repository"]
+        else:
+            # Copy only specified paths
+            logger.info(f"Copying specific paths: {docs_paths}")
+            for doc_path in docs_paths:
+                source = upstream_dir / doc_path.lstrip('/')
+                if source.exists():
+                    dest = gitea_dir / source.name
+                    logger.info(f"Copying {source.name}...")
+                    if source.is_dir():
+                        shutil.copytree(source, dest)
+                    else:
+                        shutil.copy2(source, dest)
+                    copied_paths.append(source.name)
+                else:
+                    logger.warning(f"Path not found in upstream: {doc_path}")
+
+            if not copied_paths:
+                raise Exception("No documentation paths were copied")
+
+        # Create .SYNC_INFO.md with metadata (don't overwrite README.md from upstream)
+        sync_info_path = gitea_dir / ".SYNC_INFO.md"
+        content_desc = "Complete repository mirror" if "entire repository" in copied_paths else f"Paths: {', '.join(copied_paths)}"
+        sync_info_content = f"""# Sync Information
+
+This is a mirror of the {project.title()} repository.
+
+**Synced from:** {upstream_repo}
+**Branch:** {branch}
+**Commit:** {upstream_commit}
+**Sync Date:** {upstream_date}
+**Content:** {content_desc}
+
+---
+
+This repository is automatically synced monthly by The Scheduler.
+For the latest updates, visit the official repository.
+"""
+        sync_info_path.write_text(sync_info_content)
+
+        # Git add, commit, push
+        await _run_command(['git', 'add', '.'], cwd=gitea_dir)
+
+        # Check if there are changes
+        status = await _run_command(
+            ['git', 'status', '--porcelain'],
+            cwd=gitea_dir,
+            capture=True
+        )
+
+        if not status.strip():
+            logger.info("No changes detected, skipping commit")
+            return f"Documentation already up to date (commit: {upstream_commit[:8]})"
+
+        # Commit changes
+        commit_msg = f"Sync {project} docs from {upstream_commit[:8]} on {upstream_date}"
+        await _run_command([
+            'git', 'commit',
+            '-m', commit_msg
+        ], cwd=gitea_dir)
+
+        # Tag with date
+        tag = f"sync-{upstream_date}"
+        await _run_command([
+            'git', 'tag', '-f', tag,
+            '-m', f"Documentation snapshot {upstream_date}"
+        ], cwd=gitea_dir)
+
+        # Push to Gitea
+        logger.info("Pushing to Gitea...")
+        await _run_command([
+            'git', 'push', 'origin', branch, '--force'
+        ], cwd=gitea_dir)
+
+        await _run_command([
+            'git', 'push', 'origin', tag, '--force'
+        ], cwd=gitea_dir)
+
+        # Cleanup
+        logger.info("Cleaning up work directory...")
+        shutil.rmtree(work_dir)
+
+        result = (
+            f"Successfully synced {project} documentation. "
+            f"Copied: {', '.join(copied_paths)}. "
+            f"Upstream commit: {upstream_commit[:8]}. "
+            f"Tagged: {tag}"
+        )
+        logger.info(result)
+        return result
+
+    except Exception as e:
+        # Cleanup on error
+        if work_dir.exists():
+            shutil.rmtree(work_dir)
+        raise
+
+
+async def _run_command(
+    cmd: List[str],
+    cwd: Optional[Path] = None,
+    capture: bool = False
+) -> str:
+    """Run shell command asynchronously."""
+    logger.debug(f"Running: {' '.join(cmd)} (cwd: {cwd})")
+
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        cwd=cwd,
+        stdout=asyncio.subprocess.PIPE if capture else asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE
+    )
+
+    stdout, stderr = await proc.communicate()
+
+    if proc.returncode != 0:
+        error_msg = stderr.decode() if stderr else "Unknown error"
+        raise Exception(f"Command failed: {' '.join(cmd)}\n{error_msg}")
+
+    return stdout.decode() if capture else ""
+
+
+async def _get_git_commit(repo_dir: Path) -> str:
+    """Get current git commit hash."""
+    output = await _run_command(
+        ['git', 'rev-parse', 'HEAD'],
+        cwd=repo_dir,
+        capture=True
+    )
+    return output.strip()
diff --git a/services/scheduler/src/executors/example_executor.py b/services/scheduler/src/executors/example_executor.py
new file mode 100644
index 0000000..3d4b62d
--- /dev/null
+++ b/services/scheduler/src/executors/example_executor.py
@@ -0,0 +1,42 @@
+"""
+Example executor demonstrating the pattern.
+Shows how to write task executors for The Scheduler.
+"""
+import asyncio
+import logging
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """
+    Example task executor.
+
+    Args:
+        config: Task configuration from scheduled_tasks.config JSONB field
+                Example: {"message": "Hello", "delay_seconds": 2}
+        settings: Global scheduler settings (database, API keys, etc.)
+
+    Returns:
+        Output message describing what was done
+
+    Raises:
+        Exception: On failure (will be caught and logged by executor framework)
+    """
+    message = config.get('message', 'No message configured')
+    delay = config.get('delay_seconds', 1)
+
+    logger.info(f"Example executor starting: {message}")
+
+    # Simulate some work
+    await asyncio.sleep(delay)
+
+    # You can access settings
+    logger.info(f"Using database: {settings.postgres_db}")
+
+    # Return success message
+    output = f"Executed example task: {message} (took {delay}s)"
+    logger.info(output)
+
+    return output