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 drives a poll loop that confirms a run actually
completed on MassPrice's side: after a run is confirmed sent, a one-off
APScheduler job (not a recurring cron job) fires VERIFY_POLL_MINUTES
later and enqueues a check (automation.check_completion()). If it's not
complete yet, another one-off job reschedules the next check the same
interval later — up to VERIFY_MAX_ATTEMPTS times, after which it's
marked "not_confirmed" for good (which can mean "still processing
somewhere past our patience", not necessarily "failed").
"""
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 often to re-check whether a sent run has completed on MassPrice's
# side, and how many times to try before giving up. MassPrice's own
# processing time varies a lot in practice (observed anywhere from ~1 to
# over 20 minutes, and its queue can back up further under load) — 5
# minutes x 12 attempts gives an hour of patience before a stuck/failed
# run is reported as "not_confirmed" instead of polling forever.
VERIFY_POLL_MINUTES = int(os.environ.get("VERIFY_POLL_MINUTES", "5"))
VERIFY_MAX_ATTEMPTS = int(os.environ.get("VERIFY_MAX_ATTEMPTS", "12"))


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, attempt = item
                _run_verification_check(entry_id, rule_name, watermark_id, attempt)
        except Exception:
            logger.exception("Unhandled error processing %r", item)
        finally:
            _work_queue.task_done()


def _run_verification_check(entry_id, rule_name, watermark_id, attempt):
    result, message = automation.check_completion(rule_name, watermark_id)
    if result == "complete":
        automation.update_verification(entry_id, "complete", "")
    elif attempt >= VERIFY_MAX_ATTEMPTS:
        elapsed = attempt * VERIFY_POLL_MINUTES
        automation.update_verification(
            entry_id,
            "not_confirmed",
            f"Still not confirmed after {attempt} checks over ~{elapsed} minutes "
            f"— last check: {message}",
        )
    else:
        # Not done yet, but not giving up — keep "pending", just record
        # the latest note, and queue the next check.
        automation.set_verification_message(entry_id, message)
        schedule_verification(entry_id, rule_name, watermark_id, attempt=attempt + 1)


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, attempt):
    _work_queue.put(("verify", entry_id, rule_name, watermark_id, attempt))


def schedule_verification(entry_id, rule_name, watermark_id, attempt=1):
    run_date = datetime.now(_TZ) + timedelta(minutes=VERIFY_POLL_MINUTES)
    _scheduler.add_job(
        enqueue_verify,
        trigger="date",
        run_date=run_date,
        args=[entry_id, rule_name, watermark_id, attempt],
        id=f"verify-{entry_id}-{attempt}",
        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's poll loop, 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