"""Background scheduler for connector sync jobs (APScheduler-based).
On app startup, discovers all connectors with status='connected' and schedules
periodic sync() calls based on their sync_frequency setting, plus a periodic
OAuth token-refresh check.
Multi-worker safety: Gunicorn runs multiple workers, each importing the app.
Only the worker that wins a non-blocking flock on a lockfile starts the
scheduler; the others skip it, so jobs never run in duplicate.
Sync frequency mapping:
real_time → every 5 minutes
hourly → every 60 minutes
daily → every 24 hours
"""
from __future__ import annotations
import fcntl
import logging
import os
import threading
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
logger = logging.getLogger(__name__)
# -- Configuration -----------------------------------------------------------
# Interval in seconds for each sync_frequency value
SYNC_INTERVALS: Dict[str, int] = {
"real_time": 5 * 60, # 5 minutes
"hourly": 60 * 60, # 1 hour
"daily": 24 * 60 * 60, # 24 hours
}
# Default interval for unknown frequencies
DEFAULT_INTERVAL = 60 * 60 # 1 hour
# Token refresh: check every 15 minutes, refresh tokens expiring within 30 minutes
TOKEN_REFRESH_INTERVAL = 15 * 60 # 15 minutes
TOKEN_REFRESH_BUFFER = 30 * 60 # refresh 30 minutes before expiry
# Leak detection: run all detectors for every active company every 6 hours
LEAK_DETECTION_INTERVAL = 6 * 60 * 60 # 6 hours
# Forecast sync: update actuals and auto-create future periods daily
FORECAST_SYNC_INTERVAL = 24 * 60 * 60 # 24 hours
# Ad optimization: run all active optimization rules for every company every 6 hours
OPTIMIZATION_INTERVAL = 6 * 60 * 60 # 6 hours
# Lockfile ensuring only one process (gunicorn worker) runs the scheduler
SCHEDULER_LOCKFILE = os.environ.get(
"SCHEDULER_LOCKFILE", "/tmp/command-sovereignty-scheduler.lock"
)
# -- Scheduler state ---------------------------------------------------------
_scheduler: "SyncScheduler | None" = None
_scheduler_lock = threading.Lock()
_lock_fd: Optional[int] = None # keep the flock fd alive for process lifetime
def _acquire_singleton_lock() -> bool:
"""Try to acquire the cross-process scheduler lock. Non-blocking."""
global _lock_fd
try:
fd = os.open(SCHEDULER_LOCKFILE, os.O_CREAT | os.O_RDWR, 0o600)
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
os.ftruncate(fd, 0)
os.write(fd, str(os.getpid()).encode())
_lock_fd = fd # hold open — lock released automatically on process exit
return True
except (OSError, BlockingIOError):
return False
# -- SyncScheduler class -----------------------------------------------------
class SyncScheduler:
"""APScheduler-based scheduler that runs connector sync jobs on a schedule."""
def __init__(self, app) -> None:
self._app = app
self._running = False
self._sched = BackgroundScheduler(
timezone="UTC",
job_defaults={
"coalesce": True, # collapse missed runs into one
"max_instances": 1, # never overlap the same job
"misfire_grace_time": 300,
},
)
def start(self) -> None:
"""Start the scheduler and schedule syncs for all connected connectors."""
if self._running:
logger.warning("SyncScheduler already running")
return
self._running = True
logger.info("SyncScheduler starting (APScheduler)...")
try:
self._sched.start()
with self._app.app_context():
self._schedule_all()
self._schedule_token_refresh()
self._schedule_leak_detection()
self._schedule_forecast_sync()
self._schedule_optimization()
except Exception:
logger.exception("Error during initial scheduler setup")
logger.info("SyncScheduler started")
def stop(self) -> None:
"""Stop the scheduler and all jobs."""
if not self._running:
return
self._running = False
try:
self._sched.shutdown(wait=False)
except Exception:
pass
logger.info("SyncScheduler stopped")
# -- Job registration -----------------------------------------------------
def _schedule_all(self) -> None:
"""Discover connected connectors and schedule their sync jobs."""
from app.models import Connector
connectors = Connector.query.filter_by(status="connected").all()
logger.info("Found %d connected connector(s) to schedule", len(connectors))
for connector in connectors:
self._schedule_connector(connector)
def _schedule_connector(self, connector: Any) -> None:
"""Schedule (or reschedule) a single connector's periodic sync job."""
interval = SYNC_INTERVALS.get(connector.sync_frequency, DEFAULT_INTERVAL)
job_id = f"sync_{connector.service}_{connector.id}"
self._sched.add_job(
self._run_sync,
trigger=IntervalTrigger(seconds=interval),
id=job_id,
replace_existing=True,
args=(connector.id, connector.service, connector.company_id),
)
logger.info(
"Scheduled %s sync for connector %s (interval=%ds, last_sync=%s)",
connector.service, connector.id[:8], interval,
connector.last_sync_at,
)
def unschedule_connector(self, connector: Any) -> None:
"""Remove a connector's sync job (e.g. after disconnect)."""
job_id = f"sync_{connector.service}_{connector.id}"
try:
self._sched.remove_job(job_id)
except Exception:
pass
def _schedule_token_refresh(self) -> None:
"""Schedule the periodic token refresh check."""
self._sched.add_job(
self._run_token_refresh,
trigger=IntervalTrigger(seconds=TOKEN_REFRESH_INTERVAL),
id="token_refresh",
replace_existing=True,
)
logger.info("Token refresh check scheduled (interval=%ds)", TOKEN_REFRESH_INTERVAL)
def _schedule_leak_detection(self) -> None:
"""Schedule the periodic revenue-leak detection scan (every 6 hours)."""
self._sched.add_job(
self._run_leak_detection,
trigger=IntervalTrigger(seconds=LEAK_DETECTION_INTERVAL),
id="leak_detection_scan",
replace_existing=True,
)
logger.info(
"Leak detection scan scheduled (interval=%ds)", LEAK_DETECTION_INTERVAL
)
def _schedule_forecast_sync(self) -> None:
"""Schedule the daily forecast actuals sync."""
self._sched.add_job(
self._run_forecast_sync,
trigger=IntervalTrigger(seconds=FORECAST_SYNC_INTERVAL),
id="forecast_sync",
replace_existing=True,
)
logger.info(
"Forecast sync scheduled (interval=%ds)", FORECAST_SYNC_INTERVAL
)
def _schedule_optimization(self) -> None:
"""Schedule the periodic ad optimization run (every 6 hours)."""
self._sched.add_job(
self._run_optimization,
trigger=IntervalTrigger(seconds=OPTIMIZATION_INTERVAL),
id="ad_optimization",
replace_existing=True,
)
logger.info(
"Ad optimization scheduled (interval=%ds)", OPTIMIZATION_INTERVAL
)
# -- Job bodies -------------------------------------------------------------
def _run_leak_detection(self) -> None:
"""Run all enabled leak detectors for every active company."""
logger.info("Running scheduled leak detection scan for all companies...")
with self._app.app_context():
try:
from app.models import Company
from app.services.leak_detectors import run_all_detectors
companies = Company.query.filter_by(is_deleted=False).all()
total_new = 0
total_updated = 0
total_alerts_sent = 0
for company in companies:
try:
result = run_all_detectors(company.id)
total_new += result.get("new_leaks", 0)
total_updated += result.get("updated_leaks", 0)
if result.get("errors"):
logger.warning(
"Leak scan errors for company %s: %s",
company.id[:8], result["errors"],
)
# Phase 4: send alerts for new leaks
try:
from app.services.leak_alerts import send_new_leak_alerts
alert_result = send_new_leak_alerts(
company.id,
result.get("leaks", []),
result,
)
sent = alert_result.get("slack_sent", 0)
total_alerts_sent += sent
if alert_result.get("slack_failed"):
logger.warning(
"Slack alert failed for company %s: %s",
company.id[:8],
alert_result["slack_failed"],
)
except Exception:
logger.exception(
"Alert delivery failed for company %s",
company.id[:8],
)
except Exception:
logger.exception(
"Leak detection scan failed for company %s",
company.id[:8],
)
logger.info(
"Leak detection scan complete: %d companies, %d new, "
"%d updated, %d alerts sent",
len(companies), total_new, total_updated, total_alerts_sent,
)
except Exception:
logger.exception("Leak detection scan encountered unexpected error")
def _run_forecast_sync(self) -> None:
"""Run forecast actuals sync for all active companies."""
logger.info("Running scheduled forecast sync for all companies...")
with self._app.app_context():
try:
from app.services.forecast_updater import run_all_forecasts
result = run_all_forecasts()
if result.get("errors"):
logger.warning(
"Forecast sync errors: %s", result["errors"],
)
logger.info(
"Forecast sync complete: %d companies, %d revenue, %d pipeline, "
"%d cost, %d future created",
result["companies_processed"],
result["total_revenue_updated"],
result["total_pipeline_updated"],
result["total_cost_updated"],
result["total_future_created"],
)
except Exception:
logger.exception("Forecast sync encountered unexpected error")
def _run_optimization(self) -> None:
"""Run ad optimization rules for every active company."""
logger.info("Running scheduled ad optimization for all companies...")
with self._app.app_context():
try:
from app.models import Company
from app.services.lead_gen.optimization_engine import OptimizationEngine
companies = Company.query.filter_by(is_deleted=False).all()
total_actions = 0
total_rules = 0
for company in companies:
try:
engine = OptimizationEngine(company_id=company.id)
result = engine.run_all_active()
total_rules += result.get("rules_run", 0)
total_actions += result.get("total_actions", 0)
if result.get("results"):
logger.info(
"Optimization for company %s: %d rules, %d actions",
company.id[:8],
result["rules_run"],
result["total_actions"],
)
except Exception:
logger.exception(
"Optimization failed for company %s",
company.id[:8],
)
logger.info(
"Ad optimization complete: %d companies, %d rules run, %d actions taken",
len(companies), total_rules, total_actions,
)
except Exception:
logger.exception("Ad optimization encountered unexpected error")
def _run_token_refresh(self) -> None:
"""Check all OAuth connectors for expiring tokens and refresh them."""
logger.info("Running token refresh check for all OAuth connectors...")
with self._app.app_context():
try:
from app.models import db, Connector
from app.connectors import OAuthConnector, get_connector_class, build_connector
import time as _time_module
now = int(_time_module.time())
connectors = Connector.query.filter_by(status="connected").all()
refreshed = 0
skipped = 0
for connector in connectors:
try:
cls = get_connector_class(connector.service)
if cls is None or not issubclass(cls, OAuthConnector):
skipped += 1
continue
config = connector.config
if not config:
skipped += 1
continue
# Check if token is about to expire
token_obtained_at = config.get("token_obtained_at", "0")
expires_in = config.get("expires_in", "0")
try:
obtained = int(token_obtained_at)
ttl = int(expires_in)
except (ValueError, TypeError):
skipped += 1
continue
if obtained == 0 or ttl == 0:
# No expiry tracking — skip (token may be permanent)
skipped += 1
continue
expires_at = obtained + ttl
time_until_expiry = expires_at - now
# Refresh if within buffer window or already expired
if time_until_expiry <= TOKEN_REFRESH_BUFFER:
refresh_token = config.get("refresh_token", "")
if not refresh_token:
logger.warning(
"Connector %s (%s) has no refresh_token, skipping",
connector.service, connector.id[:8],
)
skipped += 1
continue
logger.info(
"Refreshing token for %s (connector=%s) — %d seconds until expiry",
connector.service, connector.id[:8], time_until_expiry,
)
conn = build_connector(
connector.service,
connector.company_id,
config,
connector.id,
)
new_tokens = conn.refresh_access_token(refresh_token)
# Update config with new tokens
updated_config = dict(config)
updated_config.update(new_tokens)
updated_config["token_obtained_at"] = str(int(_time_module.time()))
connector.config = updated_config
db.session.commit()
refreshed += 1
logger.info(
"Token refreshed for %s (connector=%s, new TTL=%ds)",
connector.service, connector.id[:8],
int(new_tokens.get("expires_in", 0)),
)
else:
skipped += 1
except Exception:
logger.exception("Token refresh failed for connector %s", connector.id[:8])
logger.info(
"Token refresh check complete: %d refreshed, %d skipped", refreshed, skipped,
)
except Exception:
logger.exception("Token refresh check encountered unexpected error")
def _run_sync(self, connector_id: str, service: str, company_id: str) -> None:
"""Execute a sync for a connector inside an app context."""
logger.info("Running scheduled sync: %s (connector=%s)", service, connector_id[:8])
with self._app.app_context():
try:
from app.models import db, Connector
from app.connectors import build_connector
connector = db.session.get(Connector, connector_id)
if not connector:
logger.warning("Connector %s not found, removing job", connector_id[:8])
try:
self._sched.remove_job(f"sync_{service}_{connector_id}")
except Exception:
pass
return
if connector.status != "connected":
logger.info(
"Connector %s status=%s, skipping sync",
connector_id[:8], connector.status,
)
return
# Build connector instance (use .config property for decrypted value)
conn = build_connector(
service=service,
company_id=company_id,
config=connector.config,
connector_id=connector.id,
)
# Run sync
result = conn.sync()
# Update DB record
connector.last_sync_at = datetime.now(timezone.utc)
if result.get("status") == "success":
connector.status = "connected"
connector.error_message = ""
else:
connector.status = "error"
connector.error_message = result.get("error", "Sync failed")
db.session.commit()
logger.info(
"Scheduled sync complete: %s — status=%s, records=%d",
service, result.get("status"), result.get("record_count", 0),
)
except Exception:
logger.exception("Scheduled sync failed for connector %s", connector_id[:8])
try:
from app.models import db, Connector
db.session.rollback()
connector = db.session.get(Connector, connector_id)
if connector:
connector.status = "error"
connector.error_message = "Scheduled sync raised exception"
db.session.commit()
except Exception:
logger.exception("Failed to update connector error status in DB")
# -- Module-level helpers ----------------------------------------------------
def get_scheduler() -> SyncScheduler | None:
"""Get the global scheduler instance (None in workers that didn't win the lock)."""
with _scheduler_lock:
return _scheduler
def init_scheduler(app) -> None:
"""Initialize and start the sync scheduler within the app context.
Called from create_app() after the database is initialized. Only the
first process to grab the singleton lock actually starts the scheduler
(gunicorn runs multiple workers).
"""
global _scheduler
with _scheduler_lock:
if _scheduler is not None:
return
if os.environ.get("DISABLE_SCHEDULER", "").lower() in ("1", "true", "yes"):
logger.info("Scheduler disabled via DISABLE_SCHEDULER")
return
if not _acquire_singleton_lock():
logger.info(
"Scheduler lock held by another worker (pid lockfile: %s) — "
"skipping scheduler in this process", SCHEDULER_LOCKFILE,
)
return
_scheduler = SyncScheduler(app)
try:
_scheduler.start()
except Exception:
logger.exception("Failed to start SyncScheduler")
def trigger_immediate_sync(connector_id: str) -> None:
"""Trigger a sync for a specific connector immediately (non-blocking).
Useful for manual triggers or webhook callbacks. Works in any worker —
runs in a one-off daemon thread rather than through the scheduler.
"""
logger.info("Triggering immediate sync for connector %s", connector_id[:8])
from flask import current_app
app = current_app._get_current_object()
def _run():
with app.app_context():
from app.models import db, Connector
from app.connectors import build_connector
connector = db.session.get(Connector, connector_id)
if not connector:
logger.warning("Connector %s not found", connector_id[:8])
return
if connector.status == "syncing":
logger.info("Connector %s already syncing, skipping", connector_id[:8])
return
connector.status = "syncing"
db.session.commit()
try:
conn = build_connector(
service=connector.service,
company_id=connector.company_id,
config=connector.config,
connector_id=connector.id,
)
result = conn.sync()
connector.last_sync_at = datetime.now(timezone.utc)
if result.get("status") == "success":
connector.status = "connected"
connector.error_message = ""
else:
connector.status = "error"
connector.error_message = result.get("error", "Sync failed")
db.session.commit()
logger.info(
"Immediate sync complete: %s (records=%d)",
connector.service, result.get("record_count", 0),
)
except Exception:
logger.exception("Immediate sync failed for connector %s", connector_id[:8])
db.session.rollback()
connector = db.session.get(Connector, connector_id)
if connector:
connector.status = "error"
connector.error_message = "Sync raised exception"
db.session.commit()
thread = threading.Thread(target=_run, daemon=True)
thread.start()