diff --git a/backend/app.py b/backend/app.py index b6f134a..ab87eed 100644 --- a/backend/app.py +++ b/backend/app.py @@ -12,6 +12,7 @@ from flask_cors import CORS import db from config import CORS_ORIGINS, PORT, STATIC_DIR from routes import auth, garmin, health, analysis +from services import scheduler def create_app(): @@ -21,6 +22,11 @@ def create_app(): # Create tables once at startup (idempotent). db.init_db() + # Keeps the database current without the user pressing anything. Safe to + # call in every worker: the job is claimed through the database, so only + # one of them actually runs a given tick. + scheduler.start() + # In production the built React app is served by this same process, so the # deployment is a single port with no reverse proxy to configure. In # development STATIC_DIR does not exist and the CRA dev server serves the diff --git a/backend/db.py b/backend/db.py index 85a353a..54e075d 100644 --- a/backend/db.py +++ b/backend/db.py @@ -128,6 +128,17 @@ CREATE TABLE IF NOT EXISTS personal_records ( FOREIGN KEY (user_id) REFERENCES users(id) ); +-- Coordination for work that must happen once per interval regardless of how +-- many gunicorn workers are running. A worker claims a job by writing its +-- row, and the others see a fresh claim and stand down. Without this the +-- hourly sync would fire once per worker. +CREATE TABLE IF NOT EXISTS job_locks ( + name VARCHAR(64) PRIMARY KEY, + holder VARCHAR(64), + claimed_at DATETIME, + last_run_at DATETIME +); + -- Rendezvous for the interactive MFA login. -- garth asks for the code through a *blocking* callback, so the login parks in -- a background thread while the code arrives in a separate HTTP request that @@ -334,14 +345,25 @@ def _migrate(cur): # --- Public API ------------------------------------------------------------- +def _statements(schema): + """Split a schema script into statements. + + Comments are stripped first: splitting the raw text on ';' would cut a + comment that happens to contain one in half and hand the remainder to the + database as SQL. + """ + lines = [ln for ln in schema.splitlines() if not ln.strip().startswith("--")] + for stmt in "\n".join(lines).split(";"): + stmt = stmt.strip() + if stmt: + yield stmt + + def init_db(): conn = _connect() try: cur = conn.cursor() - for stmt in SCHEMA.split(";"): - stmt = stmt.strip() - if not stmt: - continue + for stmt in _statements(SCHEMA): cur.execute(_adapt_sql(stmt)) _migrate(cur) finally: diff --git a/backend/routes/garmin.py b/backend/routes/garmin.py index 51b03fb..fa741a6 100644 --- a/backend/routes/garmin.py +++ b/backend/routes/garmin.py @@ -5,6 +5,7 @@ from auth import require_auth from db import query_one from services import garmin as garmin_svc from services import garmin_auth +from services import scheduler bp = Blueprint("garmin", __name__) @@ -119,3 +120,31 @@ def cancel_login(): @require_auth def status(): return jsonify(garmin_svc.get_sync_status(g.user_id)) + + +@bp.route("/sync-latest", methods=["POST"]) +@require_auth +def sync_latest(): + """Pull just the last couple of days. + + Separate from /sync because it is fast enough to wait for (a few seconds + rather than minutes), so the UI can report the result directly instead of + handing back a job to poll. + """ + if not garmin_svc.has_token(g.user_id): + return jsonify({"error": "尚未绑定 Garmin 账号"}), 400 + + days = request.get_json(silent=True) or {} + try: + window = max(1, min(int(days.get("days", scheduler.SYNC_DAYS)), 7)) + except (TypeError, ValueError): + window = scheduler.SYNC_DAYS + + return jsonify(garmin_svc.sync_data(g.user_id, {}, days=window)) + + +@bp.route("/auto-sync", methods=["GET"]) +@require_auth +def auto_sync_status(): + """When the scheduler last ran and when it runs next.""" + return jsonify(scheduler.status()) diff --git a/backend/services/scheduler.py b/backend/services/scheduler.py new file mode 100644 index 0000000..e2d8ec5 --- /dev/null +++ b/backend/services/scheduler.py @@ -0,0 +1,157 @@ +""" +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 garmin as garmin_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") + +# 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 sync_all_accounts(days=None): + """Sync every account that has a stored token. Returns a per-user result.""" + days = days or SYNC_DAYS + rows = query_all("SELECT user_id FROM garmin_tokens") + results = [] + for row in rows: + uid = row["user_id"] + try: + out = garmin_svc.sync_data(uid, {}, days=days) + results.append({"user": uid, "status": out.get("status"), + "records": out.get("recordsSynced")}) + 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(): + try: + sync_all_accounts() + 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(min(300, INTERVAL_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(): + row = query_one("SELECT * FROM job_locks WHERE name = ?", [JOB_NAME]) + last = _parse(row.get("last_run_at")) if row else None + return { + "enabled": ENABLED, + "intervalSeconds": INTERVAL_SECONDS, + "days": SYNC_DAYS, + "lastRunAt": _iso(last) if last else None, + "nextRunAt": _iso(last + datetime.timedelta(seconds=INTERVAL_SECONDS)) if last else None, + "running": bool(row and row.get("claimed_at")), + } diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py new file mode 100644 index 0000000..9fb8e40 --- /dev/null +++ b/backend/tests/test_scheduler.py @@ -0,0 +1,194 @@ +""" +Unit tests for the auto-sync scheduler. + +The interesting behaviour is the claim: gunicorn runs several workers, each of +which starts its own timer, so without coordination one hourly tick would fire +a sync per worker. +""" +import datetime + +import pytest + +from services import scheduler + + +@pytest.fixture(autouse=True) +def _clean(db): + db.execute("DELETE FROM job_locks") + + +def set_last_run(db, when): + db.execute( + "UPDATE job_locks SET last_run_at = ? WHERE name = ?", + [when.isoformat(timespec="seconds"), scheduler.JOB_NAME], + ) + + +class TestClaim: + def test_first_caller_gets_the_job(self, db): + assert scheduler.claim() is True + + def test_second_caller_is_turned_away_while_the_first_holds_it(self, db, monkeypatch): + monkeypatch.setattr(scheduler.os, "getpid", lambda: 111) + assert scheduler.claim() is True + + # A different worker, same instant. + monkeypatch.setattr(scheduler.os, "getpid", lambda: 222) + assert scheduler.claim() is False, "two workers must not run the same tick" + + def test_not_due_again_within_the_interval(self, db): + scheduler.claim() + scheduler.release() + assert scheduler.claim() is False + + def test_due_again_after_the_interval(self, db): + scheduler.claim() + scheduler.release() + set_last_run(db, datetime.datetime.utcnow() - datetime.timedelta(seconds=7200)) + assert scheduler.claim() is True + + def test_abandoned_claim_expires(self, db, monkeypatch): + """A worker that dies mid-run must not block the job forever.""" + monkeypatch.setattr(scheduler.os, "getpid", lambda: 111) + scheduler.claim() + + stale = datetime.datetime.utcnow() - datetime.timedelta( + seconds=scheduler.CLAIM_TIMEOUT_SECONDS + 60 + ) + db.execute( + "UPDATE job_locks SET claimed_at = ? WHERE name = ?", + [stale.isoformat(timespec="seconds"), scheduler.JOB_NAME], + ) + + monkeypatch.setattr(scheduler.os, "getpid", lambda: 222) + assert scheduler.claim() is True + + def test_release_without_running_leaves_it_due(self, db): + scheduler.claim() + scheduler.release(ran=False) + assert scheduler.claim() is True + + def test_release_after_running_records_the_time(self, db): + scheduler.claim() + scheduler.release() + assert scheduler.status()["lastRunAt"] is not None + + +class TestSyncAllAccounts: + def test_no_accounts_is_a_no_op(self, db, user): + assert scheduler.sync_all_accounts() == [] + + def test_syncs_every_account_holding_a_token(self, db, user, monkeypatch, client): + from services import garmin as garmin_svc + + other = client.post( + "/api/auth/register", + json={"email": "b@example.com", "garminEmail": "bg@example.com", + "garminPassword": "pw123456"}, + ).get_json() + garmin_svc.save_token(user["id"], "t1", "a@example.com") + garmin_svc.save_token(other["id"], "t2", "b@example.com") + + seen = [] + monkeypatch.setattr( + garmin_svc, "sync_data", + lambda uid, creds, days=None, client=None: seen.append(uid) + or {"status": "success", "recordsSynced": days}, + ) + + results = scheduler.sync_all_accounts(days=2) + assert set(seen) == {user["id"], other["id"]} + assert all(r["status"] == "success" for r in results) + + def test_one_failing_account_does_not_stop_the_others( + self, db, user, monkeypatch, client + ): + from services import garmin as garmin_svc + + other = client.post( + "/api/auth/register", + json={"email": "c@example.com", "garminEmail": "cg@example.com", + "garminPassword": "pw123456"}, + ).get_json() + garmin_svc.save_token(user["id"], "t1") + garmin_svc.save_token(other["id"], "t2") + + def flaky(uid, creds, days=None, client=None): + if uid == user["id"]: + raise RuntimeError("token expired") + return {"status": "success", "recordsSynced": 2} + + monkeypatch.setattr(garmin_svc, "sync_data", flaky) + results = scheduler.sync_all_accounts() + + assert len(results) == 2 + assert {r["status"] for r in results} == {"error", "success"} + + def test_accounts_without_a_token_are_skipped(self, db, user, monkeypatch): + from services import garmin as garmin_svc + called = [] + monkeypatch.setattr( + garmin_svc, "sync_data", + lambda *a, **k: called.append(1) or {"status": "success"}, + ) + scheduler.sync_all_accounts() + assert called == [], "an account with no token cannot be synced" + + +class TestStatus: + def test_reports_configuration(self, db): + s = scheduler.status() + assert s["intervalSeconds"] == scheduler.INTERVAL_SECONDS + assert s["days"] == scheduler.SYNC_DAYS + + def test_next_run_follows_the_last(self, db): + scheduler.claim() + scheduler.release() + s = scheduler.status() + assert s["lastRunAt"] and s["nextRunAt"] + assert s["nextRunAt"] > s["lastRunAt"] + + def test_no_run_yet(self, db): + s = scheduler.status() + assert s["lastRunAt"] is None + assert s["nextRunAt"] is None + + +class TestEndpoints: + def test_sync_latest_requires_auth(self, client): + assert client.post("/api/garmin/sync-latest", json={}).status_code == 401 + + def test_sync_latest_needs_a_bound_account(self, client, auth): + r = client.post("/api/garmin/sync-latest", headers=auth, json={}) + assert r.status_code == 400 + assert "绑定" in r.get_json()["error"] + + def test_sync_latest_runs_inline(self, client, auth, user, db, monkeypatch): + from services import garmin as garmin_svc + garmin_svc.save_token(user["id"], "blob") + monkeypatch.setattr( + garmin_svc, "sync_data", + lambda uid, creds, days=None, client=None: { + "status": "success", "recordsSynced": days, "message": "ok", + }, + ) + r = client.post("/api/garmin/sync-latest", headers=auth, json={"days": 3}) + assert r.status_code == 200 + assert r.get_json()["recordsSynced"] == 3 + + def test_sync_latest_window_is_clamped(self, client, auth, user, db, monkeypatch): + from services import garmin as garmin_svc + garmin_svc.save_token(user["id"], "blob") + monkeypatch.setattr( + garmin_svc, "sync_data", + lambda uid, creds, days=None, client=None: { + "status": "success", "recordsSynced": days, + }, + ) + r = client.post("/api/garmin/sync-latest", headers=auth, json={"days": 999}) + assert r.get_json()["recordsSynced"] == 7 + + def test_auto_sync_status_endpoint(self, client, auth): + r = client.get("/api/garmin/auto-sync", headers=auth) + assert r.status_code == 200 + assert "intervalSeconds" in r.get_json()