notify-bridge/packages/server/src/notify_bridge_server/services/scheduler.py

"""APScheduler-based polling scheduler for trackers."""

from __future__ import annotations

import logging

from apscheduler.schedulers.asyncio import AsyncIOScheduler

_LOGGER = logging.getLogger(__name__)

_scheduler: AsyncIOScheduler | None = None


def get_scheduler() -> AsyncIOScheduler:
    global _scheduler
    if _scheduler is None:
        _scheduler = AsyncIOScheduler()
    return _scheduler


async def start_scheduler() -> None:
    scheduler = get_scheduler()
    if not scheduler.running:
        scheduler.start()
        _LOGGER.info("Scheduler started")

    await _load_tracker_jobs()

    # Start Telegram bot polling for bots with active command listeners
    from .telegram_poller import start_command_listener_polling
    await start_command_listener_polling()

    # Start debounced command auto-sync scheduler
    from .command_sync import start_sync_scheduler
    start_sync_scheduler()


async def _load_tracker_jobs() -> None:
    """Load enabled trackers and schedule polling jobs."""
    from sqlmodel import select
    from sqlmodel.ext.asyncio.session import AsyncSession
    from ..database.engine import get_engine
    from ..database.models import NotificationTracker, ServiceProvider as ServiceProviderModel

    engine = get_engine()
    scheduler = get_scheduler()

    async with AsyncSession(engine) as session:
        result = await session.exec(select(NotificationTracker).where(NotificationTracker.enabled == True))
        trackers = result.all()

        # Pre-load provider types for scheduler detection
        provider_types: dict[int, str] = {}
        for tracker in trackers:
            if tracker.provider_id not in provider_types:
                provider = await session.get(ServiceProviderModel, tracker.provider_id)
                if provider:
                    provider_types[tracker.provider_id] = provider.type

    for tracker in trackers:
        job_id = f"tracker_{tracker.id}"
        if scheduler.get_job(job_id):
            continue

        ptype = provider_types.get(tracker.provider_id, "")
        filters = tracker.filters or {}

        # Scheduler providers can use cron triggers
        if ptype == "scheduler" and filters.get("schedule_type") == "cron":
            cron_expr = filters.get("cron_expression", "")
            if cron_expr:
                try:
                    _add_cron_job(scheduler, job_id, tracker.id, cron_expr, tracker.name)
                    continue
                except Exception as e:
                    _LOGGER.error(
                        "Invalid cron expression for tracker %d (%s): %s — falling back to interval",
                        tracker.id, tracker.name, e,
                    )

        scheduler.add_job(
            _poll_tracker,
            "interval",
            seconds=tracker.scan_interval,
            id=job_id,
            args=[tracker.id],
            replace_existing=True,
        )
        _LOGGER.info("Scheduled tracker %d (%s) every %ds", tracker.id, tracker.name, tracker.scan_interval)


def _add_cron_job(
    scheduler: AsyncIOScheduler,
    job_id: str,
    tracker_id: int,
    cron_expression: str,
    tracker_name: str,
) -> None:
    """Add a cron-triggered job for a scheduler-type tracker."""
    from apscheduler.triggers.cron import CronTrigger
    trigger = CronTrigger.from_crontab(cron_expression)
    scheduler.add_job(
        _poll_tracker,
        trigger,
        id=job_id,
        args=[tracker_id],
        replace_existing=True,
    )
    _LOGGER.info("Scheduled tracker %d (%s) with cron: %s", tracker_id, tracker_name, cron_expression)


async def schedule_tracker(
    tracker_id: int,
    interval: int,
    cron_expression: str | None = None,
) -> None:
    """Add or update a scheduler job for a tracker."""
    scheduler = get_scheduler()
    job_id = f"tracker_{tracker_id}"

    # Remove existing job first to allow trigger type changes
    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)

    if cron_expression:
        try:
            _add_cron_job(scheduler, job_id, tracker_id, cron_expression, f"tracker-{tracker_id}")
            return
        except Exception as e:
            _LOGGER.error("Invalid cron for tracker %d: %s — using interval", tracker_id, e)

    scheduler.add_job(
        _poll_tracker,
        "interval",
        seconds=interval,
        id=job_id,
        args=[tracker_id],
        replace_existing=True,
    )
    _LOGGER.info("Scheduled tracker %d every %ds", tracker_id, interval)


async def unschedule_tracker(tracker_id: int) -> None:
    """Remove a scheduler job for a tracker."""
    scheduler = get_scheduler()
    job_id = f"tracker_{tracker_id}"
    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)
        _LOGGER.info("Unscheduled tracker %d", tracker_id)


async def _poll_tracker(tracker_id: int) -> None:
    """Poll a tracker for changes."""
    from .watcher import check_tracker
    try:
        await check_tracker(tracker_id)
    except Exception as e:
        _LOGGER.error("Error polling tracker %d: %s", tracker_id, e)