Massprice/scheduler.py

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

A single background worker thread owns *every* Playwright session in
this process — scheduled runs, manual "Run Now" clicks, "Refresh Rules"
and cookie imports all go through submit() onto one queue.Queue, so
exactly one browser is ever open at a time. Nothing may call into
automation.py directly from a Flask request thread: with gunicorn's
threads=4 that would open a second Chromium alongside a scheduled run,
and both would write /data/auth_state.json (which Playwright writes
itself, non-atomically) — a torn write there reads back as "session
expired" and costs a pointless cookie re-import.

Request threads that need an answer use run_browser_task(), which waits
on the queued job for REQUEST_WAIT_SECONDS and then gives up *waiting*
without cancelling the work. That bound matters: gunicorn kills a worker
whose request exceeds its `timeout` (120s), and since the APScheduler
jobstore and this queue live in that worker's memory, being killed drops
every pending verification poll on the floor.

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 concurrent.futures import Future
from concurrent.futures import TimeoutError as FutureTimeoutError
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 (app.py also
# imports this to render stored UTC timestamps for display). 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_NAME = os.environ.get("TZ") or "UTC"
try:
    TZ = pytz.timezone(_TZ_NAME)
except pytz.UnknownTimeZoneError:
    # A typo in TZ would otherwise raise at import and crash-loop the
    # container with a traceback that doesn't obviously point at .env.
    logger.error("Unknown TZ %r — falling back to UTC", _TZ_NAME)
    TZ = pytz.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"))

# How long a web request will wait for its queued browser work before
# answering "still working". Must stay comfortably under gunicorn's
# `timeout` (120s in gunicorn.conf.py) — the point is to end the request
# ourselves rather than let gunicorn kill the worker and take the
# scheduler and every pending poll down with it.
REQUEST_WAIT_SECONDS = int(os.environ.get("REQUEST_WAIT_SECONDS", "90"))


class BrowserBusy(Exception):
    """Queued browser work didn't finish inside REQUEST_WAIT_SECONDS.
    The work is still queued and will complete — this only means we
    stopped waiting on it inside the HTTP request."""


def submit(fn, *args):
    """Hand browser work to the single worker thread; returns a Future.

    Every Playwright session in this process goes through here — see
    this module's docstring for why calling automation.py from a request
    thread is not allowed.
    """
    future = Future()
    _work_queue.put((fn, args, future))
    return future


def run_browser_task(fn, *args):
    """submit() + wait for the result, for request handlers that need
    one. Re-raises whatever the work raised; raises BrowserBusy if it's
    still going when our patience runs out."""
    try:
        return submit(fn, *args).result(timeout=REQUEST_WAIT_SECONDS)
    except FutureTimeoutError:
        raise BrowserBusy(
            "Still working on it — a rule run is probably ahead of this in the "
            "queue. It'll finish in the background; reload in a moment."
        )


def _worker_loop():
    while True:
        fn, args, future = _work_queue.get()
        if not future.set_running_or_notify_cancel():
            _work_queue.task_done()
            continue
        try:
            future.set_result(fn(*args))
        except Exception as exc:
            # Recorded on the Future for whoever is waiting, and logged
            # here too — nothing waits on a scheduled run's Future.
            future.set_exception(exc)
            logger.exception("Queued %s failed", getattr(fn, "__name__", fn))
        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 _run_and_verify(rule_id, rule_name):
    result = automation.run_rule(rule_id, rule_name)
    if result.get("will_verify"):
        schedule_verification(result["entry_id"], rule_name, result["watermark_activity_id"])
    return result


def enqueue_run(rule_id, rule_name=None):
    """Fire-and-forget — the cron trigger and the "Run Now" button both
    land here, and neither waits for the run to finish."""
    return submit(_run_and_verify, rule_id, rule_name)


def enqueue_verify(entry_id, rule_name, watermark_id, attempt):
    return submit(_run_verification_check, entry_id, rule_name, watermark_id, attempt)


def refresh_rules():
    """Re-scrape the live rule list on the worker thread. Called from a
    request handler, so it goes through run_browser_task() rather than
    calling automation.list_rules() inline."""
    return run_browser_task(automation.list_rules)


def import_session(raw_text):
    """Validate and save a pasted cookie export on the worker thread."""
    return run_browser_task(automation.import_cookies, raw_text)


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 _parse_entry(entry):
    """Normalize one saved schedule entry into (days, hour, minute), or
    None if it isn't usable. Accepts both "HH:MM" and "HH:MM:SS" — an
    <input type="time"> emits seconds in some browsers, and the old
    two-value unpack raised ValueError on those, aborting the whole
    rebuild partway through (see rebuild_jobs)."""
    if not isinstance(entry, dict):
        return None
    days = [d for d in entry.get("days") or [] if d in VALID_DAYS]
    time_str = entry.get("time")
    if not days or not isinstance(time_str, str):
        return None
    parts = time_str.split(":")
    if len(parts) < 2:
        return None
    try:
        hour, minute = int(parts[0]), int(parts[1])
    except ValueError:
        return None
    if not (0 <= hour <= 23 and 0 <= minute <= 59):
        return None
    return days, hour, minute


def is_valid_entry(entry):
    """Public form of _parse_entry() for app.py's request validation, so
    the webui rejects an unusable entry at save time instead of storing
    one that every rebuild silently skips."""
    return _parse_entry(entry) is not None


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.

    Every entry is parsed and added inside its own guard: this runs
    *after* the removal loop below has already cleared the existing
    cron jobs, so one malformed entry escaping as an exception would
    leave the scheduler holding no jobs at all — every schedule silently
    dead until the next restart.
    """
    for job in _scheduler.get_jobs():
        if job.id.startswith("rule-"):
            job.remove()

    schedules = storage.load("schedules.json", {})
    skipped = 0
    for rule_id, rule in schedules.items():
        if not isinstance(rule, dict) or not rule.get("enabled", True):
            continue
        for idx, entry in enumerate(rule.get("schedules") or []):
            parsed = _parse_entry(entry)
            if parsed is None:
                skipped += 1
                logger.warning("Skipping unusable schedule entry %r on rule %s", entry, rule_id)
                continue
            days, hour, minute = parsed
            try:
                trigger = CronTrigger(
                    day_of_week=",".join(days),
                    hour=hour,
                    minute=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,
                )
            except Exception:
                skipped += 1
                logger.exception("Failed to schedule entry %r on rule %s", entry, rule_id)
    logger.info("Rebuilt %d scheduled job(s), skipped %d", len(_scheduler.get_jobs()), skipped)


def start():
    global _started
    if _started:
        return
    rebuild_jobs()
    _scheduler.start()
    threading.Thread(target=_worker_loop, daemon=True).start()
    _started = True