GarminHealthLab/backend/services/garmin.py

"""
Garmin sync service.

Pulls daily summaries + activities through the `garminconnect` library and
upserts them. The library and real Garmin credentials are required to actually
run a sync; without them the endpoint reports a clear error instead of
crashing.

Garmin credentials: the app only stores a scrypt/PBKDF2 *hash* of the Garmin
password (so it cannot be recovered), therefore a live sync needs the
plaintext garminEmail/garminPassword supplied in the request body.

On the library's API — these were verified against garminconnect 0.2.8:
  * get_user_summary(cdate)                  -> one day of daily totals
  * get_sleep_data(cdate)                    -> sleep, NOT part of the summary
  * get_hrv_data(cdate)                      -> HRV, also separate
  * get_activities_by_date(start, end)       -> activities in a date range
  * get_activities(start, limit)             -> PAGINATION, not dates
The last two are easy to confuse: `get_activities` takes an offset and a count,
so passing it a date silently asks for activity number "2026-08-23".
"""
import datetime
import json
import logging
import os
import threading
import uuid

from db import execute, query_one, query_all
from config import DB_TYPE
from services import health
from services import garmin_extras as extras

# How many days back a sync reaches.
DEFAULT_SYNC_DAYS = int(os.environ.get("GARMIN_SYNC_DAYS") or 7)


def _set_sync_status(user_id, status, now, last_sync_time=None, **fields):
    """Write the status row; `now` becomes the new last_sync_time by default.

    Callers that merely *report* being blocked (a rate-limit refusal, a
    refused start) pass the previous last_sync_time explicitly: a sync that
    never happened must not move "上次同步" forward, or the UI reads "刚刚"
    while the data actually stopped syncing hours ago.
    """
    last = last_sync_time if last_sync_time is not None else now
    cols = ["user_id", "status", "last_sync_time"] + list(fields.keys())
    placeholders = ", ".join(["?"] * len(cols))
    if DB_TYPE == "mariadb":
        updates = ", ".join(f"{c}=VALUES({c})" for c in cols if c != "user_id")
        sql = (
            f"INSERT INTO sync_status ({', '.join(cols)}) VALUES ({placeholders}) "
            f"ON DUPLICATE KEY UPDATE {updates}"
        )
    else:
        updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "user_id")
        sql = (
            f"INSERT INTO sync_status ({', '.join(cols)}) VALUES ({placeholders}) "
            f"ON CONFLICT(user_id) DO UPDATE SET {updates}"
        )
    execute(sql, [user_id, status, last] + list(fields.values()))


def _log_sync_history(user_id, trigger, days, started_at, result):
    """Append one immutable row per sync attempt.

    `trigger` names what started it — "auto" (scheduler), "manual" (同步页
    开始同步) or "quick" (设置页立即同步) — so the 同步记录 screen can show
    the whole picture. A failed or refused sync is as much a record as a
    successful one; the user is usually looking at this list because one of
    those did not behave. A write failure must never break the sync it is
    reporting, so it is logged and swallowed.
    """
    try:
        finished = datetime.datetime.utcnow().isoformat(timespec="seconds")
        r = result or {}
        duration = 0
        try:
            s = datetime.datetime.fromisoformat(str(started_at).replace(" ", "T"))
            f = datetime.datetime.fromisoformat(finished)
            duration = max(0, int((f - s).total_seconds()))
        except (TypeError, ValueError):
            pass
        execute(
            "INSERT INTO sync_history (id, user_id, trigger_kind, status, days, "
            "records_synced, activities_synced, badges_synced, "
            "personal_records_synced, started_at, finished_at, "
            "duration_seconds, message) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            [
                str(uuid.uuid4()), user_id, trigger,
                r.get("status") or "error",
                days if days is not None else 0,
                r.get("recordsSynced") or 0,
                r.get("activitiesSynced") or 0,
                r.get("badgesSynced") or 0,
                r.get("personalRecordsSynced") or 0,
                started_at, finished, duration,
                (r.get("message") or "")[:500],
            ],
        )
    except Exception as e:  # noqa: BLE001 - history must never break a sync
        logging.getLogger(__name__).warning(
            "sync history write failed for %s: %s", user_id, e
        )


def get_sync_history(user_id, limit=50):
    """Most recent sync attempts for the user, newest first."""
    rows = query_all(
        "SELECT trigger_kind, status, days, records_synced, activities_synced, "
        "badges_synced, personal_records_synced, started_at, finished_at, "
        "duration_seconds, message "
        "FROM sync_history WHERE user_id = ? "
        "ORDER BY started_at DESC LIMIT ?",
        [user_id, limit],
    )
    return [
        {
            "triggerKind": r["trigger_kind"],
            "status": r["status"],
            "days": r["days"],
            "recordsSynced": r["records_synced"],
            "activitiesSynced": r["activities_synced"],
            "badgesSynced": r["badges_synced"],
            "personalRecordsSynced": r["personal_records_synced"],
            "startedAt": r["started_at"],
            "finishedAt": r["finished_at"],
            "durationSeconds": r["duration_seconds"],
            "message": r["message"],
        }
        for r in rows
    ]


def get_sync_status(user_id):
    row = query_one("SELECT * FROM sync_status WHERE user_id = ?", [user_id])
    # Count distinct days actually stored in the database for this user.
    count = query_one(
        "SELECT COUNT(DISTINCT date) AS cnt FROM health_data WHERE user_id = ?", [user_id]
    )
    total_days = count["cnt"] if count else 0
    if not row:
        return {
            "status": "idle",
            "lastSyncTime": None,
            "recordsSynced": 0,
            "totalDays": total_days,
            "lastError": None,
        }
    return {
        "status": row["status"],
        "lastSyncTime": row["last_sync_time"],
        "recordsSynced": row["records_synced"],
        "totalDays": total_days,
        "lastError": row["last_error"],
        "progressCurrent": row.get("progress_current"),
        "progressTotal": row.get("progress_total"),
        "startedAt": row.get("started_at"),
        "stage": row.get("stage"),
    }


def reset_stale_syncs():
    """Clear a "syncing" status left behind by a process that went away.

    The status lives in the database but the work lives in a thread. A restart
    (or a crash) takes the thread and leaves the row, so the UI shows a
    progress bar that will never move and refuses to start a new sync.
    """
    execute(
        "UPDATE sync_status SET status = 'idle', stage = NULL "
        "WHERE status = 'syncing'"
    )


class RateLimited(RuntimeError):
    """Garmin answered 429.

    It reaches us disguised: the body is the plain text "Rate limited", and
    garth feeds that to json.loads, so the exception surfaced as
    `JSONDecodeError: Expecting value: line 1 column 1` — which reads like a
    parsing bug rather than "stop asking". Naming it means the sync status
    says what is actually wrong.
    """


# Garmin answering 429 is what deepens the limit: the account once stayed
# stuck for days because every retry re-hit the throttle before its own
# (multi-hour) window closed. We therefore record a 24h cooldown when a 429
# actually arrives, so the UI and the sync history can say when to expect
# recovery. The cooldown is *information*, not a gate: since 2026-09-02 the
# local estimate is no longer used to refuse a sync before it starts — every
# entry point (manual, 立即同步 and the automatic scheduler alike) issues a
# real request and trusts Garmin's live answer. Only a real 429 writes a new
# cooldown; a stale local estimate must never keep a healthy account idle.
RATE_LIMIT_BACKOFF = datetime.timedelta(hours=24)
# In-process cache of the cooldown, kept in sync with the DB copy below and
# still the lever the tests reach for via _rate_limited_until.clear().
_rate_limited_until = {}


def _parse_dt(v):
    """Coerce a stored rate-limit time into a naive UTC datetime, or None."""
    if v is None:
        return None
    if isinstance(v, datetime.datetime):
        return v
    if isinstance(v, str):
        for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
            try:
                return datetime.datetime.strptime(v, fmt)
            except ValueError:
                continue
        try:
            return datetime.datetime.fromisoformat(v)
        except ValueError:
            return None
    return None


def rate_limited_until(user_id):
    """When the account is still cooling down, as a UTC datetime (or None).

    The cooldown lives in the database and is the single source of truth, so
    every gunicorn worker and a restart agree on it. A process-local dict caused
    a nasty "stuck forever" bug: a worker would remember a future deadline whose
    DB write had silently failed to persist, and go on blocking even after the
    real deadline had passed. We therefore trust the DB row, falling back to the
    in-memory cache only when no row exists yet.
    """
    try:
        row = query_one(
            "SELECT rate_limited_until FROM sync_status WHERE user_id = ?", [user_id]
        )
    except Exception:
        row = None
    db_val = _parse_dt(row["rate_limited_until"] if row else None)
    if db_val is not None:
        return db_val
    return _rate_limited_until.get(user_id)


def _note_rate_limit(user_id):
    """Record a rate-limit cooldown.

    A 429 means Garmin is throttling this account/IP, and every further request
    just extends the window — so we stand down for a long, fixed stretch
    (RATE_LIMIT_BACKOFF) rather than a short one that expires before Garmin's own
    throttle clears. The cooldown is persisted so every gunicorn worker and a
    restart agree on it.
    """
    now = datetime.datetime.utcnow()
    until = now + RATE_LIMIT_BACKOFF
    _rate_limited_until[user_id] = until
    try:
        # Persist alongside the current status so the cooldown survives across
        # workers and restarts.
        cur_status = (get_sync_status(user_id) or {}).get("status") or "idle"
        _set_sync_status(
            user_id, cur_status, now.isoformat(timespec="seconds"),
            rate_limited_until=until.isoformat(timespec="seconds"),
        )
    except Exception as e:  # no cover - surfaced so a persist failure is visible
        logging.getLogger(__name__).warning(
            "rate-limit cooldown failed to persist for %s: %s", user_id, e
        )


def _clear_rate_limit(user_id):
    """Drop a recorded cooldown after a sync that actually succeeded.

    A live success is proof Garmin stopped throttling, so the estimate has
    served its purpose; leaving a future deadline behind would just mislead
    the next diagnosis.
    """
    _rate_limited_until.pop(user_id, None)
    try:
        execute(
            "UPDATE sync_status SET rate_limited_until = NULL "
            "WHERE user_id = ?", [user_id],
        )
    except Exception:  # noqa: BLE001 - clearing is best-effort
        pass


def _rate_limit_block(user_id):
    """Human-readable recovery estimate from the *recorded* cooldown.

    Only meaningful right after a real 429 wrote a fresh cooldown (mid-run
    stand-down). It is not a pre-request gate — nothing consults the cooldown
    before calling Garmin any more.
    """
    until = rate_limited_until(user_id)
    if not until or datetime.datetime.utcnow() >= until:
        return None, None
    mins = max(1, int((until - datetime.datetime.utcnow()).total_seconds() // 60))
    return until, (
        f"Garmin 仍在限制请求频率，预计约 {mins} 分钟后自动恢复。"
        "已自动退避，请耐心等待——反复点击正是把限流撞得更深的原因，令牌本身没有失效。"
    )


def _is_rate_limited(e):
    """429 from Garmin, however it happens to be dressed.

    Usually it arrives as a JSONDecodeError, because garth calls .json() on a
    429 whose body is the plain text "Rate limited" — the status code is gone
    by the time the exception reaches us, and the message is the useless
    "Expecting value: line 1 column 1". What survives is the body itself, on
    the exception's `doc` attribute, so that is where to look.
    """
    response = getattr(e, "response", None)
    if response is not None and getattr(response, "status_code", None) == 429:
        return True

    # The 429 can be buried several layers down: garth/urllib3 fold the final
    # 429 into a RetryError whose message is 'too many 429 error responses' —
    # no status code survives and the words "rate limit" never appear. Walk the
    # cause chain and also accept the "429" marker itself.
    seen = set()
    cur = e
    while cur is not None and id(cur) not in seen:
        seen.add(id(cur))
        response = getattr(cur, "response", None)
        if response is not None and getattr(response, "status_code", None) == 429:
            return True
        text = f"{cur} {getattr(cur, 'doc', '') or ''}"
        if "429" in text or "rate limit" in text.lower():
            return True
        cur = getattr(cur, "__cause__", None) or getattr(cur, "__context__", None)
    return False


class MFARequired(RuntimeError):
    """Raised when a password login needs a code this process cannot obtain."""


# garth sends a browser User-Agent, which its SSO flow needs. The data API
# treats that same UA as a browser hitting it directly and answers every
# request with HTTP 200 and an empty array — no error, just no data. The
# official app's UA (and in fact any non-browser one) returns real data, so
# the header is swapped after login, before any API call.
API_USER_AGENT = "com.garmin.android.apps.connectmobile"


def _use_api_user_agent(client):
    try:
        client.garth.sess.headers["User-Agent"] = API_USER_AGENT
    except AttributeError:
        pass  # a stubbed client in tests has no session


def _is_cn():
    # Selects Garmin's China service, a separate backend with separate
    # accounts. This project tracks an international account.
    return (os.environ.get("GARMIN_IS_CN") or "").lower() in ("1", "true", "yes")


def _import_garmin():
    try:
        from garminconnect import Garmin
    except ImportError:
        raise RuntimeError(
            "GARMIN_LIB_MISSING: 请先运行 `pip install garminconnect` 以启用同步"
        )
    return Garmin


def load_token(user_id):
    row = query_one("SELECT token FROM garmin_tokens WHERE user_id = ?", [user_id])
    return row["token"] if row else None


def save_token(user_id, token, garmin_email=None):
    cols = ["user_id", "token", "garmin_email", "updated_at"]
    placeholders = ", ".join(["?"] * len(cols))
    if DB_TYPE == "mariadb":
        updates = ", ".join(f"{c}=VALUES({c})" for c in cols if c != "user_id")
        sql = (f"INSERT INTO garmin_tokens ({', '.join(cols)}) VALUES ({placeholders}) "
               f"ON DUPLICATE KEY UPDATE {updates}")
    else:
        updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "user_id")
        sql = (f"INSERT INTO garmin_tokens ({', '.join(cols)}) VALUES ({placeholders}) "
               f"ON CONFLICT(user_id) DO UPDATE SET {updates}")
    execute(sql, [user_id, token, garmin_email,
                  datetime.datetime.utcnow().isoformat(timespec="seconds")])
    # A re-bind means the old session is stale; the next call must build a
    # fresh one rather than keep using the session the old token minted.
    forget_client(user_id)


def has_token(user_id):
    return load_token(user_id) is not None


def get_remembered_email(user_id):
    """The Garmin email last used to sign in this account, if any.

    `garmin_tokens` is the live Garmin binding, so it is checked first. The
    fallback to `users.garmin_email` exists only for accounts that bound
    Garmin before this table did the remembering — that column has been
    unused by every path that binds a *new* account since auth-hub replaced
    local login, but dropping it would silently make those older accounts
    retype their Garmin email on every sync.
    """
    row = query_one("SELECT garmin_email FROM garmin_tokens WHERE user_id = ?", [user_id])
    if row and row.get("garmin_email"):
        return row["garmin_email"]
    row = query_one("SELECT garmin_email FROM users WHERE id = ?", [user_id])
    return (row or {}).get("garmin_email") or ""


def delete_token(user_id):
    """Forget the stored Garmin OAuth token.

    The next sync or login will have to re-authenticate and mint a fresh token.
    garmin_email on the user record is left in place so re-login only needs the
    password. The cached client — built from the old token's session — is dropped
    in the same step so a stale session can't keep being reused.
    """
    execute("DELETE FROM garmin_tokens WHERE user_id = ?", [user_id])
    forget_client(user_id)


# An authenticated client, reused across requests in this process.
#
# Building one costs ~11s against Garmin — loading the token, refreshing the
# OAuth2 grant and fetching the profile — which dwarfed the ~4s of actual data
# fetching behind an activity-detail request. The session is a requests.Session
# underneath, so it is reusable; it is dropped after CLIENT_TTL so a refreshed
# or revoked token is picked up rather than being cached indefinitely.
CLIENT_TTL_SECONDS = 900
_clients = {}
_clients_lock = threading.Lock()


def _cached_client(user_id):
    entry = _clients.get(user_id)
    if entry and (datetime.datetime.utcnow() - entry[1]).total_seconds() < CLIENT_TTL_SECONDS:
        return entry[0]
    return None


def _cache_client(user_id, client):
    if user_id:
        _clients[user_id] = (client, datetime.datetime.utcnow())


def forget_client(user_id):
    """Drop the cached session — call after re-binding an account."""
    _clients.pop(user_id, None)


def _connect(creds, user_id=None):
    """Obtain a logged-in Garmin client.

    Prefers stored OAuth tokens: an account with two-factor auth cannot be
    logged into from a web worker, because the library asks for the code on
    stdin and there is none (the failure surfaces as
    "EOFError: EOF when reading a line"). Tokens are minted once by
    `garmin_login.py`, which runs in a terminal where a code can be typed.
    """
    if user_id:
        with _clients_lock:
            cached = _cached_client(user_id)
        if cached is not None:
            return cached

    Garmin = _import_garmin()
    client = Garmin(is_cn=_is_cn())

    # Garmin SSO can be slow from some networks; the default 10s timeout in
    # the underlying garth library is too tight for the initial login.
    client.garth.configure(timeout=30)

    token = load_token(user_id) if user_id else None
    if token:
        client.garth.loads(token)
        _use_api_user_agent(client)

        # Only when it has actually expired. Refreshing on every connect spends
        # quota for nothing, and that is what walked the account into a 429.
        oauth2 = getattr(client.garth, "oauth2_token", None)
        if oauth2 is None or getattr(oauth2, "expired", True):
            try:
                client.garth.refresh_oauth2()
            except Exception as e:  # noqa: BLE001 - re-raised, just named better
                if _is_rate_limited(e):
                    _note_rate_limit(user_id)
                    raise RateLimited(
                        "Garmin 暂时限制了请求频率。这通常是短时间内连接过于频繁，"
                        "等待约半小时后会自动恢复，令牌本身没有失效。"
                    ) from e
                raise
        # garminconnect builds most of its URLs from display_name, so leaving
        # it unset sends every request to ".../None".
        client.display_name = client.garth.profile["displayName"]
        with _clients_lock:
            _cache_client(user_id, client)
        return client

    if not creds.get("garminPassword"):
        raise RuntimeError("缺少 Garmin 密码，且未找到已保存的登录令牌")

    client.username = creds["garminEmail"]
    client.password = creds["garminPassword"]
    try:
        client.login()
    except EOFError as e:
        # garth's default MFA prompt calls input(); under gunicorn stdin is
        # closed, so it raises EOFError rather than anything descriptive.
        raise MFARequired(
            "该 Garmin 账号开启了两步验证。请在「数据同步」页面用密码重新绑定，"
            "系统会提示你输入验证码。"
        ) from e
    _use_api_user_agent(client)
    with _clients_lock:
        _cache_client(user_id, client)
    return client


def describe(e):
    """A message that is never empty.

    Some exceptions carry no text at all — a bare `assert` raises
    AssertionError with str(e) == "" — and storing that produced a failed
    sync whose recorded reason was blank, which is undiagnosable.
    """
    text = str(e).strip()
    return f"{type(e).__name__}: {text}" if text else type(e).__name__


def _num(*values):
    """First value that is a usable number."""
    for v in values:
        if isinstance(v, (int, float)) and not isinstance(v, bool):
            return v
    return None


def _safe(fn, default=None):
    """Call an optional endpoint; a metric the device does not record must not
    abort the whole day."""
    try:
        return fn()
    except Exception:
        return default


def _first(seq):
    return seq[0] if isinstance(seq, list) and seq else {}


def _to_datetime(*values):
    """Normalise Garmin's several timestamp shapes into an ISO string.

    The same payload mixes ISO strings ("2019-10-13T10:10:12.0") with epoch
    milliseconds (1570961412000); handing the latter to a DATETIME column is
    rejected outright, so a personal record whose only timestamp was numeric
    failed the whole batch.
    """
    for v in values:
        if v is None or v == "":
            continue
        if isinstance(v, str):
            return v[:26]
        if isinstance(v, (int, float)) and not isinstance(v, bool):
            # Values past ~1e11 are milliseconds, below that seconds.
            seconds = v / 1000 if v > 1e11 else v
            try:
                return datetime.datetime.utcfromtimestamp(seconds).isoformat(
                    timespec="seconds"
                )
            except (ValueError, OverflowError, OSError):
                continue
    return None


def _extract_daily(client, date_str):
    """Everything Garmin exposes for one day.

    The daily summary is the bulk of it, but sleep, HRV, training readiness
    and endurance each live behind their own endpoint — none of them appear in
    get_user_summary. Each is fetched defensively so a metric this device does
    not record leaves a NULL instead of failing the day.
    """
    s = client.get_user_summary(date_str) or {}

    sleep_dto = (_safe(lambda: client.get_sleep_data(date_str)) or {}).get(
        "dailySleepDTO"
    ) or {}
    scores = sleep_dto.get("sleepScores") if isinstance(
        sleep_dto.get("sleepScores"), dict
    ) else {}
    sleep_seconds = _num(sleep_dto.get("sleepTimeSeconds"))

    hrv_summary = (_safe(lambda: client.get_hrv_data(date_str)) or {}).get(
        "hrvSummary"
    ) or {}

    readiness = _first(_safe(lambda: client.get_training_readiness(date_str), []))
    training = _safe(lambda: client.get_training_status(date_str), {}) or {}
    vo2 = (training.get("mostRecentVO2Max") or {}).get("generic") or {}
    endurance = _safe(lambda: client.get_endurance_score(date_str), {}) or {}

    def secs(key):
        return _num(sleep_dto.get(key))

    return {
        "date": date_str,
        # --- activity / energy ---
        "steps": _num(s.get("totalSteps")),
        "stepGoal": _num(s.get("dailyStepGoal")),
        "distanceMeters": _num(s.get("totalDistanceMeters")),
        "caloriesBurned": _num(s.get("totalKilocalories")),
        "activeCalories": _num(s.get("activeKilocalories")),
        "bmrCalories": _num(s.get("bmrKilocalories")),
        "floorsAscended": _num(s.get("floorsAscended")),
        "floorsDescended": _num(s.get("floorsDescended")),
        "intensityMinutes": (
            (_num(s.get("moderateIntensityMinutes")) or 0)
            + (_num(s.get("vigorousIntensityMinutes")) or 0)
        ) or None,
        "sedentarySeconds": _num(s.get("sedentarySeconds")),
        "activeSeconds": _num(s.get("activeSeconds")),
        # --- heart / stress ---
        "heartRate": _num(s.get("restingHeartRate"), s.get("averageHeartRate")),
        "heartRateMax": _num(s.get("maxHeartRate")),
        "heartRateMin": _num(s.get("minHeartRate")),
        "heartRateVariability": _num(
            hrv_summary.get("lastNightAvg"), hrv_summary.get("weeklyAvg")
        ),
        "stress": _num(s.get("averageStressLevel")),
        "stressMax": _num(s.get("maxStressLevel")),
        # --- body battery ---
        "bodyBatteryHigh": _num(s.get("bodyBatteryHighestValue")),
        "bodyBatteryLow": _num(s.get("bodyBatteryLowestValue")),
        "bodyBatteryCharged": _num(s.get("bodyBatteryChargedValue")),
        "bodyBatteryDrained": _num(s.get("bodyBatteryDrainedValue")),
        # --- breathing / blood oxygen ---
        "spo2Avg": _num(s.get("averageSpo2")),
        "spo2Min": _num(s.get("lowestSpo2")),
        "respirationAvg": _num(
            s.get("avgWakingRespirationValue"), s.get("latestRespirationValue")
        ),
        "respirationMin": _num(s.get("lowestRespirationValue")),
        "respirationMax": _num(s.get("highestRespirationValue")),
        # --- sleep ---
        "sleepDuration": round(sleep_seconds / 3600, 1) if sleep_seconds else None,
        "sleepQuality": _num((scores.get("overall") or {}).get("value")),
        "sleepDeepSeconds": secs("deepSleepSeconds"),
        "sleepLightSeconds": secs("lightSleepSeconds"),
        "sleepRemSeconds": secs("remSleepSeconds"),
        "sleepAwakeSeconds": secs("awakeSleepSeconds"),
        "sleepSpo2Avg": secs("averageSpO2Value"),
        "sleepRespirationAvg": secs("averageRespirationValue"),
        "sleepStressAvg": secs("avgSleepStress"),
        # --- training ---
        "trainingReadiness": _num(readiness.get("score")),
        "vo2max": _num(vo2.get("vo2MaxValue")),
        "enduranceScore": _num(endurance.get("overallScore")),
    }


def sync_badges(client, user_id):
    """Earned badges. Keyed by Garmin's badge id, so re-syncing updates."""
    badges = _safe(lambda: client.get_earned_badges(), []) or []
    stored = 0
    for b in badges:
        bid = b.get("badgeId")
        if bid is None:
            continue
        health.upsert_badge(user_id, {
            "id": str(bid),
            "badgeKey": b.get("badgeKey"),
            "name": b.get("badgeName"),
            "categoryId": _num(b.get("badgeCategoryId")),
            "difficultyId": _num(b.get("badgeDifficultyId")),
            "earnedDate": _to_datetime(b.get("badgeEarnedDate")),
            "earnedCount": _num(b.get("badgeEarnedNumber")),
            "points": _num(b.get("badgePoints")),
        })
        stored += 1
    return stored


def sync_personal_records(client, user_id):
    records = _safe(lambda: client.get_personal_record(), []) or []
    stored = 0
    for r in records:
        rid = r.get("id")
        if rid is None:
            continue
        health.upsert_personal_record(user_id, {
            "id": str(rid),
            "typeId": _num(r.get("typeId")),
            "activityId": r.get("activityId"),
            "activityName": r.get("activityName"),
            "activityType": r.get("activityType"),
            "value": _num(r.get("value")),
            # Prefer the pre-formatted strings; the bare fields are epoch ms.
            "achievedAt": _to_datetime(
                r.get("prStartTimeLocalFormatted"),
                r.get("prStartTimeGmtFormatted"),
                r.get("activityStartDateTimeLocalFormatted"),
                r.get("prStartTimeLocal"),
                r.get("prStartTimeGmt"),
            ),
        })
        stored += 1
    return stored


def _activity_end(start, duration_seconds):
    if not start or not duration_seconds:
        return start
    for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S.%f"):
        try:
            dt = datetime.datetime.strptime(start[:26], fmt)
            return (dt + datetime.timedelta(seconds=duration_seconds)).isoformat()
        except ValueError:
            continue
    return start


def _sync_activities(client, user_id, start_date, end_date):
    """Fetch the window's activities in one call and store the new ones."""
    activities = client.get_activities_by_date(start_date, end_date) or []
    stored = 0
    for a in activities:
        start = a.get("startTimeLocal") or a.get("startTime")
        activity_type = (
            (a.get("activityType") or {}).get("typeKey")
            if isinstance(a.get("activityType"), dict)
            else a.get("activityType")
        ) or "unknown"
        duration = _num(a.get("duration"))

        # Garmin activity ids are stable, so re-syncing a window must not
        # duplicate what is already stored.
        garmin_id = a.get("activityId")
        if garmin_id is not None:
            existing = query_one(
                "SELECT id FROM activities WHERE user_id = ? AND id = ?",
                [user_id, str(garmin_id)],
            )
            if existing:
                continue

        health.insert_activity(
            user_id,
            {
                "id": str(garmin_id) if garmin_id is not None else None,
                "activityType": activity_type,
                "startTime": start,
                "endTime": _activity_end(start, duration),
                "duration": duration,
                "distance": _num(a.get("distance")),
                "calories": _num(a.get("calories")),
                "heartRateAverage": _num(a.get("averageHR")),
                "heartRateMax": _num(a.get("maxHR")),
            },
        )
        stored += 1
    return stored


# --- one activity, in full ---------------------------------------------------

# Garmin will return thousands of samples per activity. A phone chart cannot
# draw more than a few hundred usefully, and the payload is stored as a row, so
# the series are thinned on the way in rather than on every read.
DETAIL_MAX_POINTS = 300

# Descriptor key -> the name the UI charts by. Anything not listed is dropped:
# the full descriptor set runs to dozens of fields, most of them empty.
SERIES_KEYS = {
    "directTimestamp": "timestamp",
    "sumElapsedDuration": "elapsed",
    "sumDuration": "duration",
    "sumDistance": "distance",
    "directHeartRate": "heartRate",
    "directSpeed": "speed",
    "directElevation": "elevation",
    "directRunCadence": "cadence",
    "directBikeCadence": "cadence",
    "directDoubleCadence": "cadence",
    "directPower": "power",
    "directAirTemperature": "temperature",
}


def _thin(values, limit=DETAIL_MAX_POINTS):
    """Evenly sample a list down to `limit` points, keeping first and last."""
    if len(values) <= limit:
        return values
    step = (len(values) - 1) / (limit - 1)
    return [values[int(round(i * step))] for i in range(limit)]


def _series_from_details(details):
    """Turn Garmin's column-store detail payload into per-metric arrays.

    The response is a descriptor list plus rows of parallel values, so every
    metric has to be read out by the index its descriptor names.
    """
    descriptors = details.get("metricDescriptors") or []
    rows = details.get("activityDetailMetrics") or []
    if not descriptors or not rows:
        return {}

    index = {}
    for d in descriptors:
        name = SERIES_KEYS.get(d.get("key"))
        if name and name not in index:
            index[name] = d.get("metricsIndex")

    rows = _thin(rows)
    out = {}
    for name, position in index.items():
        if position is None:
            continue
        column = []
        for row in rows:
            metrics = row.get("metrics") or []
            column.append(metrics[position] if position < len(metrics) else None)
        # A column of nothing but nulls is a sensor the watch does not have.
        if any(v is not None for v in column):
            out[name] = column
    return out


def _lap_rows(splits):
    laps = []
    for i, lap in enumerate((splits or {}).get("lapDTOs") or [], start=1):
        laps.append({
            "index": lap.get("lapIndex") or i,
            "duration": _num(lap.get("duration")),
            "movingDuration": _num(lap.get("movingDuration")),
            "distance": _num(lap.get("distance")),
            "averageSpeed": _num(lap.get("averageSpeed")),
            "maxSpeed": _num(lap.get("maxSpeed")),
            "calories": _num(lap.get("calories")),
            "averageHR": _num(lap.get("averageHR")),
            "maxHR": _num(lap.get("maxHR")),
            "elevationGain": _num(lap.get("elevationGain")),
            "elevationLoss": _num(lap.get("elevationLoss")),
        })
    return laps


def _hr_zones(zones):
    out = []
    for z in zones or []:
        out.append({
            "zone": z.get("zoneNumber"),
            "seconds": _num(z.get("secsInZone")) or 0,
            "lowBoundary": _num(z.get("zoneLowBoundary")),
        })
    return sorted(out, key=lambda z: z.get("zone") or 0)


def _build_detail(client, activity_id):
    """Assemble everything Garmin knows about one activity.

    Each call is wrapped: a watch without a barometer has no weather, a
    treadmill run has no gear, and a missing optional endpoint must leave the
    rest of the page intact rather than fail the request.
    """
    summary = _safe(lambda: client.get_activity_evaluation(activity_id), {}) or {}
    # 500 is already more samples than the 300 we keep, and asking for 2000
    # triples the payload for points that get thinned away anyway.
    details = _safe(
        lambda: client.get_activity_details(activity_id, maxchart=500, maxpoly=0), {}
    ) or {}

    return {
        "activityId": str(activity_id),
        "summary": summary.get("summaryDTO") or {},
        "activityName": summary.get("activityName"),
        "activityType": (summary.get("activityTypeDTO") or {}).get("typeKey"),
        "eventType": (summary.get("eventTypeDTO") or {}).get("typeKey"),
        "laps": _lap_rows(_safe(lambda: client.get_activity_splits(activity_id), {})),
        "hrZones": _hr_zones(
            _safe(lambda: client.get_activity_hr_in_timezones(activity_id), [])
        ),
        "weather": _safe(lambda: client.get_activity_weather(activity_id), {}) or {},
        "gear": _safe(lambda: client.get_activity_gear(activity_id), []) or [],
        "exerciseSets": (
            _safe(lambda: client.get_activity_exercise_sets(activity_id), {}) or {}
        ).get("exerciseSets") or [],
        "series": _series_from_details(details),
    }


def _store_detail(user_id, activity_id, detail):
    cols = ["activity_id", "user_id", "payload", "fetched_at"]
    values = [activity_id, user_id, json.dumps(detail, default=str),
              datetime.datetime.utcnow().isoformat(timespec="seconds")]
    placeholders = ", ".join(["?"] * len(cols))
    if DB_TYPE == "mariadb":
        updates = ", ".join(f"{c}=VALUES({c})" for c in cols if c != "activity_id")
        sql = (f"INSERT INTO activity_details ({', '.join(cols)}) "
               f"VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates}")
    else:
        updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "activity_id")
        sql = (f"INSERT INTO activity_details ({', '.join(cols)}) VALUES "
               f"({placeholders}) ON CONFLICT(activity_id) DO UPDATE SET {updates}")
    execute(sql, values)


def read_activity_detail(user_id, activity_id):
    """The stored detail for one activity, or None if it was never synced."""
    row = query_one(
        "SELECT payload FROM activity_details WHERE user_id = ? AND activity_id = ?",
        [user_id, str(activity_id)],
    )
    if not row or not row.get("payload"):
        return None
    try:
        return json.loads(row["payload"])
    except ValueError:
        # A truncated row is worth re-syncing, not worth crashing on.
        return None


def sync_activity_details(client, user_id, limit=None, on_progress=None):
    """Fetch and store the full detail for activities that lack one.

    Detail used to be fetched when the user opened an activity, which meant
    seven Garmin calls on the critical path of a tap: slow at best, and a
    timeout whenever the link was poor. It belongs in the sync, so the screen
    only ever reads the local database.
    """
    rows = query_all(
        "SELECT a.id FROM activities a "
        "LEFT JOIN activity_details d ON d.activity_id = a.id "
        "WHERE a.user_id = ? AND d.activity_id IS NULL "
        "ORDER BY a.start_time DESC",
        [user_id],
    )
    if limit:
        rows = rows[:limit]

    stored = 0
    for i, row in enumerate(rows):
        try:
            _store_detail(user_id, row["id"], _build_detail(client, row["id"]))
            stored += 1
        except Exception:  # noqa: BLE001 - one bad activity must not stop the rest
            continue
        if on_progress:
            on_progress(i + 1, len(rows))
    return stored


_backfill_progress = {}


def backfill_status(user_id):
    """Progress of the historical backfill for this account."""
    return _backfill_progress.get(user_id) or {
        "running": False, "stage": None, "done": 0, "total": 0, "error": None,
    }


def _set_backfill(user_id, **fields):
    state = dict(_backfill_progress.get(user_id) or {})
    state.update(fields)
    _backfill_progress[user_id] = state


def days_missing_series(user_id, limit=None):
    """Days that have a health row but no within-day curves stored."""
    rows = query_all(
        "SELECT h.date FROM health_data h "
        "LEFT JOIN daily_series s ON s.user_id = h.user_id AND s.date = h.date "
        "WHERE h.user_id = ? AND s.date IS NULL "
        "GROUP BY h.date ORDER BY h.date DESC",
        [user_id],
    )
    dates = [str(r["date"])[:10] for r in rows]
    return dates[:limit] if limit else dates


def start_backfill(user_id, limit=None):
    """Fill in everything the per-day sync leaves out, in the background.

    Two long jobs share one runner because they share a cause — an account
    whose history predates these features — and because the user should press
    one button, not two. Each activity costs several Garmin calls and each day
    of curves costs five, so this runs for minutes; the UI polls.
    """
    state = _backfill_progress.get(user_id)
    if state and state.get("running"):
        return state

    _set_backfill(user_id, running=True, stage="启动中", done=0, total=0, error=None)

    def run():
        try:
            client = _connect({}, user_id=user_id)

            _set_backfill(user_id, stage="运动详情", done=0, total=0)
            sync_activity_details(
                client, user_id, limit,
                on_progress=lambda d, n: _set_backfill(
                    user_id, stage="运动详情", done=d, total=n),
            )

            dates = days_missing_series(user_id, limit)
            _set_backfill(user_id, stage="每日曲线", done=0, total=len(dates))
            for i, date in enumerate(dates):
                try:
                    extras.sync_daily_series(client, user_id, date)
                except Exception:  # noqa: BLE001 - one day must not stop the rest
                    pass
                _set_backfill(user_id, stage="每日曲线", done=i + 1,
                              total=len(dates))

            _set_backfill(user_id, running=False, stage="完成", error=None)
        except Exception as e:  # noqa: BLE001 - reported through the status endpoint
            _set_backfill(user_id, running=False, stage=None, error=describe(e))

    threading.Thread(target=run, daemon=True, name=f"backfill-{user_id}").start()
    return backfill_status(user_id)


# What 全部历史 actually means. 730 was a made-up ceiling that silently
# truncated anyone with more than two years of Garmin history to two years.
# The real end of the data is found by walking back until the days stop
# containing anything (EMPTY_RUN_STOP below), so this is only a backstop.
MAX_HISTORY_DAYS = int(os.environ.get("GARMIN_MAX_HISTORY_DAYS") or 3650)

# A long backfill stops once this many consecutive fetched days come back with
# nothing at all: that is what reaching the start of the account looks like,
# and without it 全部历史 would spend thousands of requests on years that
# predate the watch. Long enough to ride out a season of not wearing it.
EMPTY_RUN_STOP = 120

# The most recent days are still being written to, so a backfill refetches
# them even when they are already stored. Everything older is skipped if it is
# already in the database — that is what makes a multi-year sync resumable
# after a rate limit instead of restarting from today every time.
ALWAYS_REFETCH_DAYS = 3


# Up to this many days, a sync also pulls each day's within-day curves inline.
# Beyond it the curves are left to the background backfill: five extra calls
# per day would turn a year's sync into an hour.
SERIES_INLINE_DAYS = 14


# Above this many days a sync is long enough that the caller must not block
# on it — a year takes roughly 20 minutes at ~3s per day.
BACKGROUND_THRESHOLD_DAYS = 14


def start_sync(user_id, creds, days=None):
    """Run a sync in the background and return immediately.

    Progress lands in sync_status, which the UI polls; a full backfill runs
    far longer than any sensible HTTP timeout.

    No local rate-limit estimate is consulted here: every manual start issues
    a real request and trusts Garmin's live answer. A real 429 writes the
    cooldown and the run stands down inside sync_data.
    """
    days = DEFAULT_SYNC_DAYS if days is None else days
    now = datetime.datetime.utcnow().isoformat(timespec="seconds")
    _set_sync_status(
        user_id, "syncing", now,
        records_synced=0, progress_current=0, progress_total=days,
        started_at=now, last_error=None,
    )
    thread = threading.Thread(
        target=sync_data, args=(user_id, creds, days), daemon=True
    )
    thread.start()
    return {"status": "syncing", "days": days}


def sync_data(user_id, creds, days=None, client=None, trigger="manual"):
    """Pull the last `days` days from Garmin Connect into the local database.

    `client` exists so tests can inject a stub instead of reaching Garmin.
    `trigger` ("auto" / "manual" / "quick") names what started this run and is
    recorded on the sync_history row every exit path appends.
    """
    days = DEFAULT_SYNC_DAYS if days is None else days
    if days == 0:
        days = MAX_HISTORY_DAYS  # 全部历史 → 走到数据尽头为止
    now = datetime.datetime.utcnow().isoformat(timespec="seconds")
    started = now

    def finish(result):
        """Record the attempt, then hand the result to the caller."""
        _log_sync_history(user_id, trigger, days, started, result)
        return result

    # No local cooldown gate here (2026-09-02): the recorded rate_limited_until
    # is an estimate, and refusing on it kept accounts idle after Garmin had
    # already recovered. Every run issues a real request; only a real 429
    # stands the run down (mid-run, below) and writes a fresh cooldown.
    _set_sync_status(
        user_id, "syncing", now,
        records_synced=0, progress_current=0, progress_total=days,
        stage="连接 Garmin",
    )

    try:
        client = client or _connect(creds, user_id)
    except RateLimited as e:
        # A genuine 429 at connect time: _connect already wrote the fresh
        # cooldown, so surface this as rate_limited (the UI draws a distinct
        # stand-down state for it) rather than a generic error.
        message = str(e)
        _set_sync_status(user_id, "rate_limited", now, records_synced=0,
                         last_error=message)
        return finish({
            "status": "rate_limited",
            "recordsSynced": 0,
            "message": message,
            "mfaRequired": False,
            "lastSyncTime": now,
        })
    except Exception as e:
        message = describe(e)
        _set_sync_status(user_id, "error", now, records_synced=0, last_error=message)
        return finish({
            "status": "error",
            "recordsSynced": 0,
            "message": message,
            "mfaRequired": isinstance(e, MFARequired),
            "lastSyncTime": now,
        })

    today = datetime.date.today()

    # -1 means "incremental sync": pick up from the latest date already in the
    # local database rather than pulling a fixed window.
    if days == -1:
        # The daily rows live in `health_data`; `health_daily` is the name of
        # the *writer* (health.upsert_health_daily), not of any table. Reading
        # it raised inside the background thread, so an incremental sync died
        # silently and left the status stuck on "syncing".
        row = query_one(
            "SELECT MAX(date) AS latest FROM health_data WHERE user_id = ?", (user_id,)
        )
        latest = row["latest"] if row and row.get("latest") else None
        if latest is None:
            days = DEFAULT_SYNC_DAYS  # first sync → fall back to default window
        else:
            days = max(1, (today - datetime.date.fromisoformat(latest)).days)
        # Rewrite the progress total now that we know the real day count.
        _set_sync_status(
            user_id, "syncing", now,
            records_synced=0, progress_current=0, progress_total=days,
            stage="连接 Garmin",
        )

    start_date = (today - datetime.timedelta(days=days - 1)).isoformat()

    # Days already stored, so a resumed backfill does not spend its whole
    # rate-limit budget re-fetching what it already has.
    stored = {
        r["date"] for r in query_all(
            "SELECT date FROM health_data WHERE user_id = ? AND date >= ?",
            (user_id, start_date),
        )
    }

    days_synced = 0
    days_attempted = 0
    empty_run = 0
    reached_the_start = False
    day_errors = []
    rate_limited_mid_run = None
    for i in range(days):
        date_str = (today - datetime.timedelta(days=i)).isoformat()
        if i >= ALWAYS_REFETCH_DAYS and date_str in stored:
            continue
        days_attempted += 1
        try:
            record = _extract_daily(client, date_str)
            record.update(extras.daily_extras(client, date_str))
        except Exception as e:
            # A 429 inside the day loop used to be filed as one more skipped
            # day, so a 730-day backfill kept hammering Garmin for another
            # 700 days and drove the throttle deeper. Stand down at the first
            # one and keep whatever was already stored.
            if isinstance(e, RateLimited) or _is_rate_limited(e):
                _note_rate_limit(user_id)
                rate_limited_mid_run = describe(e)
                break
            day_errors.append(f"{date_str}: {describe(e)}")
            continue
        # A day Garmin has no data for comes back all-None; storing it would
        # create an empty row that the metric endpoints then have to filter.
        if any(record[k] is not None for k in record if k != "date"):
            health.upsert_health_daily(user_id, record)
            days_synced += 1
            empty_run = 0
            # Within-day curves for short syncs only. A year-long backfill
            # would add five calls per day on top of everything else; those
            # days are filled by start_backfill instead.
            if days <= SERIES_INLINE_DAYS:
                try:
                    extras.sync_daily_series(client, user_id, date_str)
                except Exception as e:  # noqa: BLE001
                    day_errors.append(f"{date_str} series: {describe(e)}")

        else:
            # Walked back past the start of the account. Without this, 全部历史
            # would keep asking Garmin about years before the watch existed.
            empty_run += 1
            if days > 90 and empty_run >= EMPTY_RUN_STOP:
                reached_the_start = True
                break

        # A long backfill reports every fifth day — the write is cheap but not
        # free. A short one reports every day: at 7 days a "every 5th" cadence
        # meant the bar sat at 0 for most of the run and then vanished.
        if days <= 30 or (i + 1) % 5 == 0 or i + 1 == days:
            _set_sync_status(
                user_id, "syncing", now,
                records_synced=days_synced, progress_current=i + 1,
                progress_total=days, stage=f"每日数据 {date_str}",
            )

    if rate_limited_mid_run:
        until, msg = _rate_limit_block(user_id)
        message = msg or "Garmin 限制了请求频率，已自动退避。"
        _set_sync_status(
            user_id, "rate_limited", now, records_synced=days_synced,
            progress_current=days_synced, progress_total=days, stage=None,
            last_error=message,
        )
        return finish({
            "status": "rate_limited",
            "recordsSynced": days_synced,
            "message": f"已同步 {days_synced} 天后被 Garmin 限流：{message}",
            "retryAfterSeconds": (
                int((until - datetime.datetime.utcnow()).total_seconds()) if until else None
            ),
            "lastSyncTime": now,
        })

    activities_synced = 0
    _set_sync_status(user_id, "syncing", now, records_synced=days_synced,
                     progress_current=days, progress_total=days,
                     stage="运动记录")
    try:
        activities_synced = _sync_activities(
            client, user_id, start_date, today.isoformat()
        )
    except Exception as e:
        day_errors.append(f"activities: {describe(e)}")

    # Full detail for any activity that does not have it stored yet, so that
    # opening one later is a local read.
    details_synced = 0
    try:
        details_synced = sync_activity_details(
            client, user_id,
            on_progress=lambda d, n: _set_sync_status(
                user_id, "syncing", now, records_synced=days_synced,
                progress_current=days, progress_total=days,
                stage=f"运动详情 {d}/{n}"),
        )
    except Exception as e:
        day_errors.append(f"activity_details: {describe(e)}")

    # Everything else Garmin holds: body composition, blood pressure, race
    # predictions, challenges and devices. Account-wide, so once per sync.
    extra_counts = {}
    stage_names = {
        "bodyComposition": "身体成分", "bloodPressure": "血压",
        "racePredictions": "成绩预测", "challenges": "挑战赛", "devices": "设备",
    }
    for name, call in (
        ("bodyComposition",
         lambda: extras.sync_body_composition(client, user_id, start_date,
                                              today.isoformat())),
        ("bloodPressure",
         lambda: extras.sync_blood_pressure(client, user_id, start_date,
                                            today.isoformat())),
        ("racePredictions",
         lambda: extras.sync_race_predictions(client, user_id)),
        ("challenges", lambda: extras.sync_challenges(client, user_id)),
        ("devices", lambda: extras.sync_devices(client, user_id)),
    ):
        _set_sync_status(user_id, "syncing", now, records_synced=days_synced,
                         progress_current=days, progress_total=days,
                         stage=stage_names.get(name, name))
        try:
            extra_counts[name] = call()
        except Exception as e:  # noqa: BLE001 - one section must not fail the sync
            day_errors.append(f"{name}: {describe(e)}")

    # Badges and personal records are account-wide rather than per-day, so
    # they are fetched once per sync rather than inside the day loop.
    badges_synced = 0
    records_synced_pr = 0
    try:
        badges_synced = sync_badges(client, user_id)
    except Exception as e:
        day_errors.append(f"badges: {describe(e)}")
    try:
        records_synced_pr = sync_personal_records(client, user_id)
    except Exception as e:
        day_errors.append(f"personal_records: {describe(e)}")

    # Every single day failing means something systemic (expired session,
    # API change) — reporting that as a clean success would hide it.
    if days_synced == 0 and days_attempted > 0 and len(day_errors) >= days_attempted:
        message = "; ".join(day_errors[:3])
        _set_sync_status(user_id, "error", now, records_synced=0, last_error=message)
        return finish({"status": "error", "recordsSynced": 0,
                       "message": f"同步失败：{message}", "lastSyncTime": now})

    _set_sync_status(
        user_id, "idle", now, records_synced=days_synced,
        progress_current=days, progress_total=days, stage=None,
        last_error="; ".join(day_errors[:3]) if day_errors else None,
    )
    # A live success proves Garmin stopped throttling — retire any recorded
    # cooldown so it cannot mislead a later diagnosis.
    _clear_rate_limit(user_id)
    message = (
        f"同步完成，更新 {days_synced} 天数据、{activities_synced} 条运动记录"
        f"（含 {details_synced} 条详情）、"
        f"{badges_synced} 个奖励、{records_synced_pr} 项个人纪录"
    )
    if day_errors:
        message += f"（{len(day_errors)} 项跳过）"
    return finish({
        "status": "success",
        "recordsSynced": days_synced,
        "activitiesSynced": activities_synced,
        "badgesSynced": badges_synced,
        "personalRecordsSynced": records_synced_pr,
        "message": message,
        "lastSyncTime": now,
    })