"""Scheduled background jobs. Registers periodic tasks on an APScheduler ``BackgroundScheduler``: * **MITRE sync** — every 24 hours (see :func:`sync_mitre`) * **Intel scan** — every 7 days (see :func:`scan_intel`) Each job manages its own database session (created on entry, closed in ``finally``) so it is fully independent from FastAPI's request-scoped sessions. """ import logging from datetime import datetime, timedelta, timezone from apscheduler.schedulers.background import BackgroundScheduler from app.database import SessionLocal from app.services.mitre_sync_service import sync_mitre from app.services.intel_service import scan_intel from app.services.notification_service import cleanup_old_notifications from app.services.snapshot_service import create_snapshot, cleanup_old_snapshots from app.services.campaign_scheduler_service import check_and_run_recurring_campaigns from app.jobs.jira_sync_job import sync_all_jira_links from app.services.osint_enrichment_service import enrich_all_techniques from app.services.stale_detection_service import detect_stale_coverage from app.jobs.retention_job import run_retention_job logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Module-level scheduler instance # --------------------------------------------------------------------------- scheduler = BackgroundScheduler() # --------------------------------------------------------------------------- # Job functions # --------------------------------------------------------------------------- def _run_mitre_sync() -> None: """Execute a MITRE sync inside its own DB session.""" logger.info("Scheduled MITRE sync job starting...") db = SessionLocal() try: summary = sync_mitre(db) logger.info("Scheduled MITRE sync job finished — %s", summary) except Exception: logger.exception("Scheduled MITRE sync job failed") finally: db.close() def _run_notification_cleanup() -> None: """Clean up old read notifications.""" logger.info("Scheduled notification cleanup job starting...") db = SessionLocal() try: deleted = cleanup_old_notifications(db, days=90) logger.info("Notification cleanup finished — deleted %d old notifications", deleted) except Exception: logger.exception("Notification cleanup job failed") finally: db.close() def _run_weekly_snapshot() -> None: """Create a weekly coverage snapshot and clean up old ones.""" logger.info("Scheduled weekly snapshot job starting...") db = SessionLocal() try: snapshot = create_snapshot(db, name="Auto-weekly") logger.info( "Weekly snapshot created — score %.1f, %d techniques", snapshot.organization_score, snapshot.total_techniques, ) deleted = cleanup_old_snapshots(db, keep_last=52) if deleted: logger.info("Cleaned up %d old snapshots", deleted) except Exception: logger.exception("Weekly snapshot job failed") finally: db.close() def _run_recurring_campaigns() -> None: """Check and run any due recurring campaigns.""" logger.info("Scheduled recurring campaigns check starting...") db = SessionLocal() try: spawned = check_and_run_recurring_campaigns(db) logger.info("Recurring campaigns check finished — spawned %d campaigns", spawned) except Exception: logger.exception("Recurring campaigns check failed") finally: db.close() def _run_intel_scan() -> None: """Execute an intel scan inside its own DB session.""" logger.info("Scheduled intel scan job starting...") db = SessionLocal() try: summary = scan_intel(db) logger.info("Scheduled intel scan job finished — %s", summary) except Exception: logger.exception("Scheduled intel scan job failed") finally: db.close() def _run_osint_enrichment() -> None: """Execute weekly OSINT enrichment inside its own DB session.""" logger.info("Scheduled OSINT enrichment job starting...") db = SessionLocal() try: total = enrich_all_techniques(db) logger.info("OSINT enrichment finished — %d new items", total) except Exception: logger.exception("OSINT enrichment job failed") finally: db.close() _FREQUENCY_INTERVALS: dict[str, timedelta] = { "daily": timedelta(days=1), "weekly": timedelta(weeks=1), "monthly": timedelta(days=30), } def _run_data_sources_sync() -> None: """Check all enabled data sources and sync those that are overdue.""" logger.info("Scheduled data sources sync check starting...") db = SessionLocal() try: from app.models.data_source import DataSource from app.services.data_source_service import sync_source now = datetime.now(timezone.utc) sources = ( db.query(DataSource) .filter(DataSource.is_enabled == True) # noqa: E712 .all() ) synced = 0 for ds in sources: freq = ds.sync_frequency if not freq or freq == "manual": continue interval = _FREQUENCY_INTERVALS.get(freq) if interval is None: continue last = ds.last_sync_at if last is None: # Never synced — run it now overdue = True else: # Make last timezone-aware if needed if last.tzinfo is None: last = last.replace(tzinfo=timezone.utc) overdue = now - last >= interval if overdue: logger.info( "Data source '%s' is overdue (freq=%s, last=%s) — syncing", ds.name, freq, last, ) try: sync_source(db, str(ds.id)) synced += 1 except Exception: logger.exception("Failed to sync data source '%s'", ds.name) logger.info("Data sources sync check finished — %d source(s) synced", synced) except Exception: logger.exception("Data sources sync check failed") finally: db.close() def _run_stale_detection() -> None: """Execute daily stale coverage detection inside its own DB session.""" logger.info("Scheduled stale coverage detection starting...") db = SessionLocal() try: count = detect_stale_coverage(db) logger.info("Stale detection finished — %d techniques flagged", count) except Exception: logger.exception("Stale coverage detection job failed") finally: db.close() # --------------------------------------------------------------------------- # Scheduler bootstrap # --------------------------------------------------------------------------- def start_scheduler() -> None: """Register all periodic jobs and start the background scheduler. Jobs registered: * ``mitre_sync`` — every **24 hours** * ``intel_scan`` — every **7 days** Neither job fires immediately on startup. """ scheduler.add_job( _run_mitre_sync, trigger="interval", hours=24, id="mitre_sync", name="MITRE ATT&CK sync (every 24h)", replace_existing=True, ) scheduler.add_job( _run_intel_scan, trigger="interval", weeks=1, id="intel_scan", name="Intel scan (every 7d)", replace_existing=True, ) scheduler.add_job( _run_notification_cleanup, trigger="interval", hours=24, id="notification_cleanup", name="Notification cleanup (daily)", replace_existing=True, ) scheduler.add_job( _run_weekly_snapshot, trigger="cron", day_of_week="sun", hour=0, minute=0, id="weekly_snapshot", name="Weekly coverage snapshot (Sundays 00:00)", replace_existing=True, ) scheduler.add_job( _run_recurring_campaigns, trigger="interval", hours=24, id="recurring_campaigns", name="Recurring campaigns check (daily)", replace_existing=True, ) scheduler.add_job( sync_all_jira_links, trigger="interval", hours=1, id="jira_sync", name="Jira link sync (hourly)", replace_existing=True, ) scheduler.add_job( _run_osint_enrichment, trigger="interval", weeks=1, id="osint_enrichment", name="OSINT enrichment (weekly)", replace_existing=True, ) scheduler.add_job( _run_stale_detection, trigger="interval", hours=24, id="stale_detection", name="Stale coverage detection (daily)", replace_existing=True, ) scheduler.add_job( run_retention_job, trigger="interval", hours=24, id="retention_policies", name="Data retention policies (daily)", replace_existing=True, ) scheduler.add_job( _run_data_sources_sync, trigger="interval", hours=6, id="data_sources_sync", name="Data sources auto-sync (every 6h)", replace_existing=True, ) scheduler.start() logger.info( "Background scheduler started — mitre_sync (24h), intel_scan (7d), " "notification_cleanup (24h), weekly_snapshot (Sundays 00:00), " "recurring_campaigns (daily), jira_sync (1h), " "osint_enrichment (weekly), stale_detection (daily), " "retention_policies (daily), data_sources_sync (6h)" )