"""Nexus — Schedule Runner Background Task Every 60 seconds: check PushSchedule cron expressions, trigger due schedules. """ import asyncio import json import logging from datetime import datetime, timezone from server.infrastructure.database.session import AsyncSessionLocal from server.domain.models import PushSchedule logger = logging.getLogger("nexus.schedule_runner") SCHEDULE_CHECK_INTERVAL = 60 # seconds def _cron_match(cron_expr: str, now: datetime) -> bool: """Check if current time matches a 5-field cron expression (min hour dom month dow). Supports: * (any), */N (step), specific values, comma-separated lists. """ parts = cron_expr.strip().split() if len(parts) != 5: return False fields = [ (now.minute, range(0, 60)), (now.hour, range(0, 24)), (now.day, range(1, 32)), (now.month, range(1, 13)), ((now.weekday() + 1) % 7, range(0, 7)), # cron: 0=Sunday, Python weekday(): 0=Monday ] for (current_val, _), field_expr in zip(fields, parts): if not _field_match(field_expr, current_val): return False return True def _field_match(expr: str, value: int) -> bool: """Match a single cron field expression against a value. Supports: * (any), */N (step), specific values, comma-separated lists, and range (N-M). Each comma-separated part is tried independently; the function returns True as soon as any part matches. """ for part in expr.split(","): if part == "*": return True if part.startswith("*/"): step = int(part[2:]) if step == 0: continue # */0 is invalid; skip rather than ZeroDivisionError if value % step == 0: return True continue if "-" in part: low, high = part.split("-", 1) if int(low) <= value <= int(high): return True else: # cron: 7 also means Sunday (=0) if value == int(part) or (int(part) == 7 and value == 0): return True return False async def schedule_runner_loop(): """Background task: check schedules every 60s, trigger due cron jobs.""" logger.info("Schedule runner loop started (interval: 60s)") while True: await asyncio.sleep(SCHEDULE_CHECK_INTERVAL) try: async with AsyncSessionLocal() as session: from sqlalchemy import select result = await session.execute( select(PushSchedule).where(PushSchedule.enabled == True) ) schedules = result.scalars().all() now = datetime.now(timezone.utc) triggered = 0 for schedule in schedules: try: run_mode = getattr(schedule, 'run_mode', None) or 'cron' # ── One-time schedule: check fire_at ── if run_mode == 'once': fire_at = getattr(schedule, 'fire_at', None) if not fire_at: continue now_naive = now.replace(tzinfo=None) fire_naive = fire_at.replace(tzinfo=None) if fire_at.tzinfo else fire_at if fire_naive > now_naive: continue # not time yet if schedule.last_run_at: continue # already ran else: # ── Cron schedule: check cron expression ── cron_expr = getattr(schedule, 'cron_expr', None) if not cron_expr or not _cron_match(cron_expr, now): continue if schedule.last_run_at: last_run = schedule.last_run_at now_naive = now.replace(tzinfo=None) if last_run.tzinfo is not None: last_run = last_run.replace(tzinfo=None) seconds_since = (now_naive - last_run).total_seconds() if seconds_since < SCHEDULE_CHECK_INTERVAL: continue # Parse server_ids from JSON try: server_ids = json.loads(schedule.server_ids) except (json.JSONDecodeError, TypeError): logger.error(f"Schedule {schedule.id}: invalid server_ids JSON") continue if not server_ids: continue # Mark as running BEFORE execution to prevent double-trigger # (if execution takes longer than SCHEDULE_CHECK_INTERVAL) schedule.last_run_at = now if run_mode == 'once': schedule.enabled = False await session.commit() schedule_type = getattr(schedule, 'schedule_type', 'push') or 'push' if schedule_type == 'script': # ── Script execution schedule ── from server.application.services.script_service import ScriptService from server.infrastructure.database.script_repo import ( ScriptRepositoryImpl, ScriptExecutionRepositoryImpl, ) from server.infrastructure.database.db_credential_repo import DbCredentialRepositoryImpl from server.infrastructure.database.server_repo import ServerRepositoryImpl from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl cmd = getattr(schedule, 'script_content', None) or '' script_id = getattr(schedule, 'script_id', None) # If no inline command, load from script library if not cmd and script_id: script = await ScriptRepositoryImpl(session).get_by_id(script_id) if script: cmd = script.content or '' if not cmd: logger.error(f"Schedule '{schedule.name}': no command or script content, skipping") continue svc = ScriptService( script_repo=ScriptRepositoryImpl(session), execution_repo=ScriptExecutionRepositoryImpl(session), credential_repo=DbCredentialRepositoryImpl(session), server_repo=ServerRepositoryImpl(session), audit_repo=AuditLogRepositoryImpl(session), ) timeout = getattr(schedule, 'exec_timeout', None) or 60 long_task = bool(getattr(schedule, 'long_task', False)) execution = await svc.execute_command( command=cmd, server_ids=server_ids, script_id=script_id, timeout=timeout, operator=f"schedule:{schedule.name}", long_task=long_task, ) logger.info( f"Schedule '{schedule.name}' (script) triggered: " f"execution_id={execution.id} on {len(server_ids)} servers" ) else: # ── File push schedule (original behaviour) ── from server.application.services.sync_engine_v2 import SyncEngineV2 from server.infrastructure.database.server_repo import ServerRepositoryImpl from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl from server.infrastructure.database.push_schedule_repo import PushRetryJobRepositoryImpl engine = SyncEngineV2( server_repo=ServerRepositoryImpl(session), sync_log_repo=SyncLogRepositoryImpl(session), audit_repo=AuditLogRepositoryImpl(session), retry_repo=PushRetryJobRepositoryImpl(session), ) result = await engine.sync_files( server_ids=server_ids, source_path=schedule.source_path or '/tmp', sync_mode=getattr(schedule, 'sync_mode', None) or "incremental", trigger_type="schedule", operator=f"schedule:{schedule.name}", ) logger.info( f"Schedule '{schedule.name}' (push) triggered: " f"{result['completed']}/{result['total']} succeeded" ) # last_run_at + enabled already set before execution (prevent double-trigger) await session.commit() triggered += 1 except Exception as e: logger.error(f"Schedule {schedule.id} execution failed: {e}") if triggered: logger.info(f"Schedule runner: {triggered} schedules triggered") except Exception as e: logger.error(f"Schedule runner check failed: {e}")