alembic/app/services/scheduler_service.py

import time

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from sqlalchemy import select

from app.db import SessionLocal
from app.models import Playlist, ScheduledJob
from app.services import dedup_review_service, genre_review_service, pipeline_runner
from app.settings import settings

TIMEZONE = "America/Edmonton"

# ---------------------------------------------------------------------------
# Maintenance jobs, ported 1:1 from /etc/cron.d/sldl-maintenance. Each entry
# is (job_key, cron kwargs for CronTrigger, callable). Dedup deliberately has
# no --apply here (dry-run only on schedule) -- a deliberate behavior change
# from the old blind `--apply` cron, matching the false-negative-biased
# dedup preference; applying deletions requires a reviewed confirm through
# the dedup UI (services/dedup_review_service.py).
#
# NOTE on job store choice: APScheduler's SQLAlchemyJobStore pickles job
# functions to persist them, and closures returned from a factory (like
# _lib()/_beet() below) are NOT picklable -- `add_job` would fail at
# runtime. MemoryJobStore (the default) keeps live objects in-process
# instead, which is fine here because register_all_jobs() rebuilds the
# entire schedule from code + the DB on every startup anyway. Persisted
# enable/disable state (so a paused job stays paused across a restart)
# lives in the app's own `scheduled_jobs` table, not in APScheduler's store.
# ---------------------------------------------------------------------------


def _lib(job_key: str, script_name: str, args: list[str] | None = None, timeout: float | None = None):
    async def _run(triggered_by: str = "schedule"):
        return await pipeline_runner.run_lib_script(job_key, script_name, args, triggered_by=triggered_by)

    return _run


def _beet(job_key: str, args: list[str]):
    async def _run(triggered_by: str = "schedule"):
        return await pipeline_runner.run_beet(job_key, args, triggered_by=triggered_by)

    return _run


async def _dedup_scan(triggered_by: str = "schedule"):
    """Dry-run only, ever, on the schedule -- see module docstring. Populates
    dedup_candidates for review; deletion only ever happens through a
    confirmed dedup_review_service.confirm_and_apply() call from the UI."""
    return await dedup_review_service.scan(triggered_by=triggered_by)


async def _genre_run(triggered_by: str = "schedule"):
    """Keeps its current auto-apply behavior (unlike dedup) -- genre writes
    are low-risk/reversible and GENRE_LOCK-protected -- but now goes through
    genre_review_service so every run's diff is persisted for after-the-fact
    review instead of only living in a log file."""
    return await genre_review_service.run(apply=True, force=True, triggered_by=triggered_by)


async def _log_rotation(triggered_by: str = "schedule"):
    """Replaces `find /var/log/sldl -name '*.log' -mtime +30 -delete`."""
    cutoff = time.time() - 30 * 86400
    removed = 0
    if settings.logs_dir.exists():
        for path in settings.logs_dir.rglob("*.log"):
            if path.is_file() and path.stat().st_mtime < cutoff:
                path.unlink(missing_ok=True)
                removed += 1
    return removed


MAINTENANCE_JOBS: dict[str, tuple[dict, callable]] = {
    # ==== Daily ====
    "maintenance:sync_bandcamp": (
        dict(minute=0, hour=8),
        _lib("maintenance:sync_bandcamp", "sync-bandcamp.sh"),
    ),
    "maintenance:normalize_casing": (
        dict(minute=52, hour=8),
        _lib("maintenance:normalize_casing", "normalize-artist-casing.py", ["--apply"]),
    ),
    "maintenance:dedup": (
        dict(minute=55, hour=8),
        _dedup_scan,  # no --apply, ever, on schedule -- see _dedup_scan docstring
    ),
    "maintenance:gen_djmix_playlist": (
        dict(minute=57, hour=8),
        _lib("maintenance:gen_djmix_playlist", "gen-djmix-playlist.sh"),
    ),
    "maintenance:gen_vgm_playlist": (
        dict(minute=58, hour=8),
        _lib("maintenance:gen_vgm_playlist", "gen-vgm-playlist.sh"),
    ),
    "maintenance:navidrome_scan": (
        dict(minute=0, hour=9),
        _lib("maintenance:navidrome_scan", "navidrome-scan.sh"),
    ),
    "maintenance:export_laptop_playlists": (
        dict(minute=5, hour=9),
        _lib("maintenance:export_laptop_playlists", "export-laptop-playlists.py"),
    ),
    "maintenance:enrich_buy_url": (
        dict(minute=10, hour=9),
        _lib("maintenance:enrich_buy_url", "enrich-buy-url.py", ["--apply"]),
    ),
    "maintenance:build_fingerprint_index": (
        dict(minute=25, hour=9),
        _lib("maintenance:build_fingerprint_index", "build-fingerprint-index.py", ["--workers", "8"]),
    ),
    "maintenance:pipeline_status_report": (
        dict(minute=30, hour=9),
        _lib("maintenance:pipeline_status_report", "pipeline-status.sh"),
    ),
    "maintenance:log_rotation": (
        dict(minute=0, hour=0),
        _log_rotation,
    ),
    # ==== Weekly (Sunday) ====
    "maintenance:strip_mb_tags": (
        dict(minute=30, hour=8, day_of_week="sun"),
        _lib("maintenance:strip_mb_tags", "strip-mb-tags.sh"),
    ),
    "maintenance:strip_watermark_art": (
        dict(minute=35, hour=8, day_of_week="sun"),
        _lib("maintenance:strip_watermark_art", "strip-watermark-art.py", ["--apply"]),
    ),
    "maintenance:scrub_watermark_text": (
        dict(minute=40, hour=8, day_of_week="sun"),
        _lib("maintenance:scrub_watermark_text", "scrub-watermark-text.py", ["--apply"]),
    ),
    "maintenance:clean_sldl_index": (
        dict(minute=45, hour=8, day_of_week="sun"),
        _lib("maintenance:clean_sldl_index", "clean-sldl-index.py", ["--apply"]),
    ),
    "maintenance:clear_bad_genres": (
        dict(minute=48, hour=8, day_of_week="sun"),
        _lib("maintenance:clear_bad_genres", "clear-bad-genres.py", ["--apply"]),
    ),
    "maintenance:spotify_genre": (
        dict(minute=50, hour=8, day_of_week="sun"),
        _genre_run,  # goes through genre_review_service -- see its docstring
    ),
    "maintenance:beets_update_sync": (
        dict(minute=53, hour=8, day_of_week="sun"),
        _beet("maintenance:beets_update_sync", ["update"]),
    ),
    # ==== Monthly ====
    "maintenance:upgrade_mp3_to_flac": (
        dict(minute=0, hour=10, day=1),
        _lib("maintenance:upgrade_mp3_to_flac", "upgrade-mp3-to-flac.sh"),
    ),
}

# One short line per job for the /jobs page -- what it actually does, not
# just its key. Kept separate from MAINTENANCE_JOBS so the schedule/function
# mapping above stays easy to scan.
MAINTENANCE_JOB_DESCRIPTIONS: dict[str, str] = {
    "maintenance:sync_bandcamp": "Pull new Bandcamp purchases and import them.",
    "maintenance:normalize_casing": "Fix artist-name casing to match your canonical list.",
    "maintenance:dedup": "Scan for duplicate tracks (dry-run only; confirm deletes in Dedup).",
    "maintenance:gen_djmix_playlist": "Rebuild the DJ-mix M3U from configured albums.",
    "maintenance:gen_vgm_playlist": "Rebuild the video-game-soundtrack M3U.",
    "maintenance:navidrome_scan": "Trigger a Navidrome library rescan (share-health gated).",
    "maintenance:export_laptop_playlists": "Export dj-* Navidrome playlists for the laptop.",
    "maintenance:enrich_buy_url": "Look up buy links for tracks without one yet.",
    "maintenance:build_fingerprint_index": "Refresh the acoustic fingerprint index.",
    "maintenance:pipeline_status_report": "Write the daily health snapshot / Telegram digest.",
    "maintenance:log_rotation": "Delete job logs older than 30 days.",
    "maintenance:strip_mb_tags": "Strip stale MusicBrainz tags that fragment Navidrome albums.",
    "maintenance:strip_watermark_art": "Remove Soulseek-uploader watermark cover art.",
    "maintenance:scrub_watermark_text": "Remove Soulseek-uploader watermark text frames.",
    "maintenance:clean_sldl_index": "Prune stale entries from each playlist's download index.",
    "maintenance:clear_bad_genres": "Blank malformed/junk GENRE tags before Spotify refill.",
    "maintenance:spotify_genre": "Refill GENRE tags from Spotify per-artist data.",
    "maintenance:beets_update_sync": "Sync the beets DB to on-disk tag changes.",
    "maintenance:upgrade_mp3_to_flac": "Look for FLAC replacements for MP3 tracks (monthly).",
}


# Module-level singleton so other services (playlist_service, future
# routers) can reach the live scheduler without main.py threading it through
# every function call. None until main.py's lifespan starts it.
_instance: AsyncIOScheduler | None = None


def get_scheduler() -> AsyncIOScheduler | None:
    return _instance


def create_scheduler() -> AsyncIOScheduler:
    global _instance
    # Default job store (MemoryJobStore) -- see module docstring for why
    # SQLAlchemyJobStore doesn't work here.
    _instance = AsyncIOScheduler(timezone=TIMEZONE)
    return _instance


def _ensure_scheduled_job_rows(db) -> None:
    """Make sure every MAINTENANCE_JOBS entry has a scheduled_jobs row,
    seeding cron_expr from the hardcoded default and enabled=True the first
    time. Subsequent runs never overwrite an existing row -- this table is
    the persisted source of truth for enable/disable once it exists."""
    existing = {row.job_key for row in db.execute(select(ScheduledJob)).scalars()}
    now = time.time()
    for job_key, (cron_kwargs, _runner) in MAINTENANCE_JOBS.items():
        if job_key in existing:
            continue
        cron_expr = "{minute} {hour} {day} {month} {day_of_week}".format(
            minute=cron_kwargs.get("minute", "*"),
            hour=cron_kwargs.get("hour", "*"),
            day=cron_kwargs.get("day", "*"),
            month=cron_kwargs.get("month", "*"),
            day_of_week=cron_kwargs.get("day_of_week", "*"),
        )
        db.add(
            ScheduledJob(
                job_key=job_key,
                job_type="maintenance",
                playlist_id=None,
                cron_expr=cron_expr,
                enabled=True,
            )
        )
    db.commit()
    _ = now  # unused, kept for symmetry with playlists' created_at/updated_at style


def _playlist_job_key(name: str) -> str:
    return f"playlist:{name}"


async def _run_playlist_job(name: str, no_m3u: bool, triggered_by: str = "schedule"):
    return await pipeline_runner.run_playlist(name, no_m3u=no_m3u, triggered_by=triggered_by)


def sync_playlist_jobs(scheduler: AsyncIOScheduler) -> None:
    """Converge the scheduler's registered playlist:<name> jobs with the
    playlists table. Call after any playlist create/update/delete -- this
    is what makes add/remove-in-the-UI take effect without a redeploy."""
    db = SessionLocal()
    try:
        playlists = list(db.execute(select(Playlist)).scalars())
    finally:
        db.close()

    desired_ids = {
        _playlist_job_key(p.name) for p in playlists if p.cron_expr and p.active
    }
    existing_ids = {j.id for j in scheduler.get_jobs() if j.id.startswith("playlist:")}

    for job_id in existing_ids - desired_ids:
        scheduler.remove_job(job_id)

    for playlist in playlists:
        job_id = _playlist_job_key(playlist.name)
        if job_id not in desired_ids:
            continue
        trigger = CronTrigger.from_crontab(playlist.cron_expr, timezone=TIMEZONE)
        scheduler.add_job(
            _run_playlist_job,
            trigger=trigger,
            id=job_id,
            args=[playlist.name, playlist.no_m3u],
            replace_existing=True,
            misfire_grace_time=3600,
        )


def register_all_jobs(scheduler: AsyncIOScheduler) -> None:
    """Rebuild the entire in-memory schedule from code (MAINTENANCE_JOBS) +
    the DB (scheduled_jobs.enabled, playlists). Safe to call at every
    startup -- MemoryJobStore has nothing to reconcile against."""
    db = SessionLocal()
    try:
        _ensure_scheduled_job_rows(db)
        enabled_by_key = {
            row.job_key: row.enabled for row in db.execute(select(ScheduledJob)).scalars()
        }
    finally:
        db.close()

    for job_key, (cron_kwargs, runner) in MAINTENANCE_JOBS.items():
        if not enabled_by_key.get(job_key, True):
            continue
        scheduler.add_job(
            runner,
            trigger=CronTrigger(**cron_kwargs, timezone=TIMEZONE),
            id=job_key,
            replace_existing=True,
            misfire_grace_time=3600,
        )

    sync_playlist_jobs(scheduler)


async def trigger_now(scheduler: AsyncIOScheduler, job_key: str) -> None:
    """'Run now' support: bypass the schedule and run this job's function
    immediately, still through pipeline_runner's lock/logging/job_runs
    recording (triggered_by='manual'). Works even for a currently-disabled
    job (disabled only means "don't run on schedule")."""
    job = scheduler.get_job(job_key)
    if job is None:
        raise ValueError(f"no such registered job: {job_key}")
    func = job.func
    args = list(job.args)
    kwargs = dict(job.kwargs)
    kwargs["triggered_by"] = "manual"
    await func(*args, **kwargs)


def set_maintenance_enabled(scheduler: AsyncIOScheduler, job_key: str, enabled: bool) -> None:
    """Pause/resume a maintenance job AND persist the choice to
    scheduled_jobs.enabled so it survives a restart -- this is the actual
    'pause an individual job without a redeploy' mechanism."""
    if job_key not in MAINTENANCE_JOBS:
        raise ValueError(f"not a maintenance job: {job_key}")

    db = SessionLocal()
    try:
        row = db.execute(
            select(ScheduledJob).where(ScheduledJob.job_key == job_key)
        ).scalar_one_or_none()
        if row is not None:
            row.enabled = enabled
            db.commit()
    finally:
        db.close()

    if enabled:
        cron_kwargs, runner = MAINTENANCE_JOBS[job_key]
        scheduler.add_job(
            runner,
            trigger=CronTrigger(**cron_kwargs, timezone=TIMEZONE),
            id=job_key,
            replace_existing=True,
            misfire_grace_time=3600,
        )
    else:
        if scheduler.get_job(job_key) is not None:
            scheduler.remove_job(job_key)