GarminHealthLab/backend/services/scheduler.py

"""
Background scheduler.

Keeps the local database close to Garmin without the user having to press
anything: every interval it pulls the last couple of days for each account
that has stored OAuth tokens.

Two things make this fiddly in this deployment, and both are handled here:

* **Several workers.** gunicorn runs more than one process, and each would
  otherwise start its own timer and sync the same account concurrently. A row
  in `job_locks` is claimed before any work starts, so exactly one worker runs
  a given tick.
* **Restarts.** The thread dies with its worker. The lock records when the job
  last completed, so a freshly started worker picks the schedule back up
  rather than either skipping an interval or immediately re-running.
"""
import datetime
import os
import threading
import time

from config import DB_TYPE
from db import execute, query_one, query_all
from services import analysis as analysis_svc
from services import garmin as garmin_svc
from services import settings as settings_svc

JOB_NAME = "garmin_auto_sync"

# How often to pull, and how far back. Two days rather than one: the current
# day is still being written to, and a day can arrive late.
INTERVAL_SECONDS = int(os.environ.get("AUTO_SYNC_INTERVAL_SECONDS") or 3600)
SYNC_DAYS = int(os.environ.get("AUTO_SYNC_DAYS") or 2)
ENABLED = (os.environ.get("AUTO_SYNC") or "true").lower() not in ("0", "false", "no")

# The loop wakes on this cadence; whether an account is actually due is then
# decided per account from its own 同步频率 setting. A single global interval
# would mean one user's choice of 30 minutes silently applied to everyone.
TICK_SECONDS = 300

# A claim older than this is treated as abandoned — the worker holding it died
# mid-run, and without expiry the job would never run again.
CLAIM_TIMEOUT_SECONDS = 1800

_started = False
_lock = threading.Lock()


def _now():
    return datetime.datetime.utcnow()


def _iso(dt):
    return dt.isoformat(timespec="seconds")


def _parse(value):
    if not value:
        return None
    try:
        return datetime.datetime.fromisoformat(str(value).replace(" ", "T"))
    except ValueError:
        return None


def claim(name=JOB_NAME, interval=INTERVAL_SECONDS):
    """Take the job if it is due and nobody else holds it.

    Returns True when this process should do the work.
    """
    holder = f"{os.getpid()}"
    now = _now()
    row = query_one("SELECT * FROM job_locks WHERE name = ?", [name])

    if row:
        last_run = _parse(row.get("last_run_at"))
        if last_run and (now - last_run).total_seconds() < interval:
            return False
        claimed = _parse(row.get("claimed_at"))
        if claimed and (now - claimed).total_seconds() < CLAIM_TIMEOUT_SECONDS:
            return False

    if DB_TYPE == "mariadb":
        sql = ("INSERT INTO job_locks (name, holder, claimed_at) VALUES (?, ?, ?) "
               "ON DUPLICATE KEY UPDATE holder=VALUES(holder), claimed_at=VALUES(claimed_at)")
    else:
        sql = ("INSERT INTO job_locks (name, holder, claimed_at) VALUES (?, ?, ?) "
               "ON CONFLICT(name) DO UPDATE SET holder=excluded.holder, "
               "claimed_at=excluded.claimed_at")
    execute(sql, [name, holder, _iso(now)])

    # Re-read: if another worker claimed between our check and our write, its
    # holder is the one now recorded and we must stand down.
    check = query_one("SELECT holder FROM job_locks WHERE name = ?", [name])
    return bool(check and check.get("holder") == holder)


def release(name=JOB_NAME, ran=True):
    if ran:
        execute(
            "UPDATE job_locks SET claimed_at = NULL, last_run_at = ? WHERE name = ?",
            [_iso(_now()), name],
        )
    else:
        execute("UPDATE job_locks SET claimed_at = NULL WHERE name = ?", [name])


def due_at(user_id):
    """When this account may next be synced automatically, or None if never.

    None means auto-sync is switched off for them; a time in the past means
    they are due now.
    """
    prefs = settings_svc.get_raw(user_id)
    if not prefs.get("auto_sync"):
        return None
    minutes = prefs.get("auto_sync_minutes") or (INTERVAL_SECONDS // 60)
    last = _parse((garmin_svc.get_sync_status(user_id) or {}).get("lastSyncTime"))
    if not last:
        return _now() - datetime.timedelta(seconds=1)
    return last + datetime.timedelta(minutes=minutes)


def sync_all_accounts(days=None, respect_schedule=False):
    """Sync every account that has a stored token. Returns a per-user result.

    `respect_schedule` is what the background loop passes: it skips accounts
    that have auto-sync off or that were synced recently enough. A direct call
    (a manual "sync everything") leaves it False and syncs unconditionally.

    The window is `SYNC_DAYS` unless the caller overrides it. Auto-sync is
    deliberately blind to the user's manual sync range: reading it here is what
    let a half-hourly tick re-pull 730 days and hold the account in a 429 loop,
    and it also made one setting mean two different things in two places.
    Anything the user chooses by hand goes through `/garmin/sync` instead.
    """
    rows = query_all("SELECT user_id FROM garmin_tokens")
    results = []
    for row in rows:
        uid = row["user_id"]
        try:
            if respect_schedule:
                due = due_at(uid)
                if due is None:
                    results.append({"user": uid, "status": "skipped",
                                    "reason": "auto-sync off"})
                    continue
                if due > _now():
                    results.append({"user": uid, "status": "skipped",
                                    "reason": "not due"})
                    continue
            d = SYNC_DAYS if days is None else days
            # Never poke Garmin while it is rate-limiting us — that is exactly
            # what keeps the limit alive. Respect the persisted cooldown and sit
            # this tick out.
            blocked = garmin_svc.rate_limited_until(uid)
            if blocked and blocked > _now():
                results.append({
                    "user": uid, "status": "skipped", "reason": "rate-limited",
                    "retryAfterSeconds": int((blocked - _now()).total_seconds()),
                })
                continue
            out = garmin_svc.sync_data(uid, {}, days=d, trigger="auto")
            results.append({"user": uid, "status": out.get("status"),
                            "records": out.get("recordsSynced")})
            # New data invalidates every stored insight, so queue them all at
            # background priority. They may take a while; anything the user
            # opens meanwhile jumps ahead of them.
            if out.get("recordsSynced"):
                try:
                    analysis_svc.prefetch_insights(uid)
                except Exception as e:  # noqa: BLE001 - never fail a sync over this
                    print(f"[scheduler] prefetch failed for {uid}: {e}")
        except Exception as e:  # noqa: BLE001 - one account must not stop the rest
            results.append({"user": uid, "status": "error", "error": str(e)[:200]})
    return results


def _loop():
    while True:
        try:
            if claim(interval=TICK_SECONDS):
                try:
                    sync_all_accounts(respect_schedule=True)
                finally:
                    release()
        except Exception as e:  # noqa: BLE001 - the loop must outlive any single failure
            print(f"[scheduler] tick failed: {e}")
        # Checked more often than the interval so a worker that starts late
        # still picks the job up promptly rather than waiting a full hour.
        time.sleep(TICK_SECONDS)


def start():
    """Start the scheduler thread once per process."""
    global _started
    if not ENABLED:
        print("[scheduler] disabled by AUTO_SYNC")
        return
    with _lock:
        if _started:
            return
        _started = True
    threading.Thread(target=_loop, daemon=True, name="auto-sync").start()
    print(f"[scheduler] auto-sync every {INTERVAL_SECONDS}s, {SYNC_DAYS} day(s) back")


def status(user_id=None):
    """Scheduler state, and — when a user is named — their own next due time."""
    row = query_one("SELECT * FROM job_locks WHERE name = ?", [JOB_NAME])
    last = _parse(row.get("last_run_at")) if row else None

    out = {
        "enabled": ENABLED,
        "intervalSeconds": INTERVAL_SECONDS,
        "tickSeconds": TICK_SECONDS,
        "days": SYNC_DAYS,
        "lastRunAt": _iso(last) if last else None,
        "nextRunAt": _iso(last + datetime.timedelta(seconds=TICK_SECONDS)) if last else None,
        "running": bool(row and row.get("claimed_at")),
    }

    if user_id:
        prefs = settings_svc.get_raw(user_id)
        due = due_at(user_id)
        out["account"] = {
            "autoSync": bool(prefs.get("auto_sync")),
            "intervalMinutes": prefs.get("auto_sync_minutes"),
            "dueAt": _iso(due) if due else None,
        }
    return out