""" 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) 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