"""
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 os
import threading

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, **fields):
    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, now] + list(fields.values()))


def get_sync_status(user_id):
    row = query_one("SELECT * FROM sync_status WHERE user_id = ?", [user_id])
    if not row:
        return {
            "status": "idle",
            "lastSyncTime": None,
            "recordsSynced": 0,
            "lastError": None,
        }
    return {
        "status": row["status"],
        "lastSyncTime": row["last_sync_time"],
        "recordsSynced": row["records_synced"],
        "lastError": row["last_error"],
        "progressCurrent": row.get("progress_current"),
        "progressTotal": row.get("progress_total"),
        "startedAt": row.get("started_at"),
    }


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


# 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())

    token = load_token(user_id) if user_id else None
    if token:
        client.garth.loads(token)
        _use_api_user_agent(client)
        # Proves the token still works, and refreshes it if near expiry.
        client.garth.refresh_oauth2()
        # 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)


# 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.
    """
    days = days or DEFAULT_SYNC_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):
    """Pull the last `days` days from Garmin Connect into the local database.

    `client` exists so tests can inject a stub instead of reaching Garmin.
    """
    days = days or DEFAULT_SYNC_DAYS
    now = datetime.datetime.utcnow().isoformat(timespec="seconds")
    _set_sync_status(
        user_id, "syncing", now,
        records_synced=0, progress_current=0, progress_total=days,
    )

    try:
        client = client or _connect(creds, user_id)
    except Exception as e:
        message = describe(e)
        _set_sync_status(user_id, "error", now, records_synced=0, last_error=message)
        return {
            "status": "error",
            "recordsSynced": 0,
            "message": message,
            "mfaRequired": isinstance(e, MFARequired),
            "lastSyncTime": now,
        }

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

    days_synced = 0
    day_errors = []
    for i in range(days):
        date_str = (today - datetime.timedelta(days=i)).isoformat()
        try:
            record = _extract_daily(client, date_str)
            record.update(extras.daily_extras(client, date_str))
        except Exception as e:
            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
            # 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)}")

        # Reported every few days rather than every day: the write is cheap
        # but not free, and the UI polls on a 2s cadence anyway.
        if (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,
            )

    activities_synced = 0
    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)
    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 = {}
    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)),
    ):
        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 len(day_errors) >= days:
        message = "; ".join(day_errors[:3])
        _set_sync_status(user_id, "error", now, records_synced=0, last_error=message)
        return {"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,
        last_error="; ".join(day_errors[:3]) if day_errors else None,
    )
    message = (
        f"同步完成，更新 {days_synced} 天数据、{activities_synced} 条运动记录"
        f"（含 {details_synced} 条详情）、"
        f"{badges_synced} 个奖励、{records_synced_pr} 项个人纪录"
    )
    if day_errors:
        message += f"（{len(day_errors)} 项跳过）"
    return {
        "status": "success",
        "recordsSynced": days_synced,
        "activitiesSynced": activities_synced,
        "badgesSynced": badges_synced,
        "personalRecordsSynced": records_synced_pr,
        "message": message,
        "lastSyncTime": now,
    }