Massprice/scheduler.py

"""In-process cron scheduling for rule runs.

A single background worker thread owns every Playwright run — both
APScheduler-triggered runs and manual "Run Now" clicks are pushed onto
one queue.Queue so exactly one browser session is ever active at a time.
This must run in exactly one process; see gunicorn.conf.py's workers=1.

The same worker also handles delayed "verify_run" tasks: after a rule
run is confirmed sent, a one-off APScheduler job (not a recurring cron
job) fires VERIFY_DELAY_MINUTES later and enqueues a check of whether it
actually completed on MassPrice's side — see automation.verify_run().
"""
import logging
import os
import queue
import threading
from datetime import datetime, timedelta

import pytz
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger

import automation
import storage

logger = logging.getLogger("massprice.scheduler")

_work_queue: "queue.Queue" = queue.Queue()

# Resolved once and reused everywhere a timezone is needed. Must be
# passed explicitly at construction time, never left to be inferred —
# both BackgroundScheduler AND CronTrigger independently call tzlocal's
# get_localzone() in their own __init__ if not given one, which fails
# inside this container image for an arbitrary $TZ (no zoneinfo/tzdata
# wiring). Passing it to the scheduler alone is NOT enough — a
# CronTrigger built without its own explicit timezone still resolves
# get_localzone() itself, regardless of what the scheduler is configured
# with, and crashes the same way. Every trigger built below must pass
# timezone=_TZ explicitly.
_TZ = pytz.timezone(os.environ.get("TZ", "UTC"))
_scheduler = BackgroundScheduler(timezone=_TZ)
_started = False

VALID_DAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]

# How long to wait after a run is confirmed sent before checking whether
# it actually completed on MassPrice's side. MassPrice's own processing
# time varies a lot in practice (observed anywhere from ~1 to ~22
# minutes) — this is a reasonable middle ground, not a guarantee; a
# "not_confirmed" result can mean "still processing", not just "failed".
VERIFY_DELAY_MINUTES = int(os.environ.get("VERIFY_DELAY_MINUTES", "10"))


def _worker_loop():
    while True:
        item = _work_queue.get()
        kind = item[0]
        try:
            if kind == "run":
                _, rule_id, rule_name = item
                result = automation.run_rule(rule_id, rule_name)
                if result.get("will_verify"):
                    schedule_verification(
                        result["entry_id"], rule_name, result["watermark_activity_id"]
                    )
            elif kind == "verify":
                _, entry_id, rule_name, watermark_id = item
                automation.verify_run(entry_id, rule_name, watermark_id)
        except Exception:
            logger.exception("Unhandled error processing %r", item)
        finally:
            _work_queue.task_done()


def enqueue_run(rule_id, rule_name=None):
    _work_queue.put(("run", rule_id, rule_name))


def enqueue_verify(entry_id, rule_name, watermark_id):
    _work_queue.put(("verify", entry_id, rule_name, watermark_id))


def schedule_verification(entry_id, rule_name, watermark_id, delay_minutes=None):
    delay = VERIFY_DELAY_MINUTES if delay_minutes is None else delay_minutes
    run_date = datetime.now(_TZ) + timedelta(minutes=delay)
    _scheduler.add_job(
        enqueue_verify,
        trigger="date",
        run_date=run_date,
        args=[entry_id, rule_name, watermark_id],
        id=f"verify-{entry_id}",
        misfire_grace_time=600,
    )


def _job_id(rule_id, index):
    return f"rule-{rule_id}-{index}"


def rebuild_jobs():
    """Clear and re-add every recurring cron job from schedules.json.

    Only touches "rule-*" jobs — one-off "verify-*" jobs (see
    schedule_verification()) must survive a schedule edit elsewhere,
    since they're tracking an already-in-flight run, not a recurring
    schedule.
    """
    for job in _scheduler.get_jobs():
        if job.id.startswith("rule-"):
            job.remove()

    schedules = storage.load("schedules.json", {})
    for rule_id, rule in schedules.items():
        if not rule.get("enabled", True):
            continue
        for idx, entry in enumerate(rule.get("schedules", [])):
            days = [d for d in entry.get("days", []) if d in VALID_DAYS]
            time_str = entry.get("time")
            if not days or not time_str:
                continue
            hour, minute = time_str.split(":")
            trigger = CronTrigger(
                day_of_week=",".join(days),
                hour=int(hour),
                minute=int(minute),
                timezone=_TZ,
            )
            _scheduler.add_job(
                enqueue_run,
                trigger=trigger,
                args=[rule_id, rule.get("name")],
                id=_job_id(rule_id, idx),
                replace_existing=True,
                misfire_grace_time=300,
            )
    logger.info("Rebuilt %d scheduled job(s)", len(_scheduler.get_jobs()))


def start(tz=None):
    # tz is accepted for backwards-compat call sites but is a no-op now
    # that timezone is fixed at construction time above.
    global _started
    if _started:
        return
    rebuild_jobs()
    _scheduler.start()
    threading.Thread(target=_worker_loop, daemon=True).start()
    _started = True