GarminHealthLab/backend/db.py

"""
Pluggable data layer for Garmin Health Lab.

Supports both SQLite (stdlib, local dev) and MariaDB (PyMySQL, NAS production)
through a single unified API:

    init_db()                 -> create tables if missing
    execute(sql, params)      -> INSERT/UPDATE/DELETE, returns {id, changes}
    query_one(sql, params)    -> one row as dict or None
    query_all(sql, params)    -> list of row dicts

Both backends accept `?` placeholders; the SQL is translated to `%s` for
MariaDB automatically. Upserts must use backend-specific SQL (see services).
"""
import os
import sqlite3
import threading
import queue
import datetime

from config import (
    DB_TYPE,
    SQLITE_PATH,
    MARIADB_SOCKET,
    MARIADB_HOST,
    MARIADB_PORT,
    MARIADB_USER,
    MARIADB_PASSWORD,
    MARIADB_DATABASE,
)

SCHEMA = """
CREATE TABLE IF NOT EXISTS users (
  id VARCHAR(64) PRIMARY KEY,
  email VARCHAR(255) NOT NULL UNIQUE,
  garmin_email VARCHAR(255) NOT NULL,
  garmin_password_hash TEXT NOT NULL,
  jwt_token TEXT,
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS health_data (
  id VARCHAR(64) PRIMARY KEY,
  user_id VARCHAR(64) NOT NULL,
  date DATE NOT NULL,
  steps INT,
  heart_rate INT,
  heart_rate_variability DOUBLE,
  blood_pressure_systolic INT,
  blood_pressure_diastolic INT,
  sleep_duration INT,
  sleep_quality DOUBLE,
  stress INT,
  calories_burned DOUBLE,
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  UNIQUE(user_id, date),
  FOREIGN KEY (user_id) REFERENCES users(id)
);

CREATE TABLE IF NOT EXISTS activities (
  id VARCHAR(64) PRIMARY KEY,
  user_id VARCHAR(64) NOT NULL,
  activity_type VARCHAR(255) NOT NULL,
  start_time DATETIME NOT NULL,
  end_time DATETIME NOT NULL,
  duration INT,
  distance DOUBLE,
  calories DOUBLE,
  heart_rate_average INT,
  heart_rate_max INT,
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  FOREIGN KEY (user_id) REFERENCES users(id)
);

CREATE TABLE IF NOT EXISTS sync_status (
  user_id VARCHAR(64) PRIMARY KEY,
  last_sync_time DATETIME,
  status VARCHAR(32) DEFAULT 'idle',
  last_error TEXT,
  records_synced INT DEFAULT 0,
  updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  FOREIGN KEY (user_id) REFERENCES users(id)
);

-- Garmin OAuth tokens, obtained once through an interactive login.
-- Garmin accounts with two-factor auth cannot be logged into unattended: the
-- library asks for an MFA code on stdin, which a gunicorn worker does not
-- have. Storing the resulting tokens lets every later sync skip the login
-- entirely (they stay valid for roughly a year).
CREATE TABLE IF NOT EXISTS garmin_tokens (
  user_id VARCHAR(64) PRIMARY KEY,
  token TEXT NOT NULL,
  garmin_email VARCHAR(255),
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  FOREIGN KEY (user_id) REFERENCES users(id)
);

-- Badges earned on Garmin Connect ("奖励"). Keyed by Garmin's own badge id so
-- a re-sync updates rather than duplicates.
CREATE TABLE IF NOT EXISTS badges (
  id VARCHAR(64) NOT NULL,
  user_id VARCHAR(64) NOT NULL,
  badge_key VARCHAR(128),
  name VARCHAR(255),
  category_id INT,
  difficulty_id INT,
  earned_date DATETIME,
  earned_count INT,
  points INT,
  PRIMARY KEY (user_id, id),
  FOREIGN KEY (user_id) REFERENCES users(id)
);

-- Personal records (个人纪录), e.g. fastest 5k, longest run.
CREATE TABLE IF NOT EXISTS personal_records (
  id VARCHAR(64) NOT NULL,
  user_id VARCHAR(64) NOT NULL,
  type_id INT,
  activity_id VARCHAR(64),
  activity_name VARCHAR(255),
  activity_type VARCHAR(64),
  value DOUBLE,
  achieved_at DATETIME,
  PRIMARY KEY (user_id, id),
  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
-- may land on a different gunicorn worker. The handoff therefore goes through
-- the database rather than process memory.
-- Holds no password: that stays in the waiting thread's memory only.
CREATE TABLE IF NOT EXISTS garmin_mfa_sessions (
  id VARCHAR(64) PRIMARY KEY,
  user_id VARCHAR(64) NOT NULL,
  status VARCHAR(32) NOT NULL,
  code VARCHAR(16),
  error TEXT,
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  FOREIGN KEY (user_id) REFERENCES users(id)
);

-- One cached LLM answer per user. Generating one takes minutes against a
-- large reasoning model, which is far too slow to sit in a page load, so the
-- result is stored and reused until the underlying data changes.
-- `fingerprint` identifies the health data the advice was derived from.
CREATE TABLE IF NOT EXISTS ai_recommendations (
  user_id VARCHAR(64) PRIMARY KEY,
  fingerprint VARCHAR(64) NOT NULL,
  model VARCHAR(64),
  upstream VARCHAR(64),
  days INT,
  payload TEXT NOT NULL,
  created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
  FOREIGN KEY (user_id) REFERENCES users(id)
);
"""

# --- MariaDB pool (lazy) ----------------------------------------------------
_mariadb_pool = None
_pool_lock = threading.Lock()


def _new_mariadb_conn():
    import pymysql
    from pymysql.cursors import DictCursor

    kwargs = dict(
        user=MARIADB_USER,
        password=MARIADB_PASSWORD,
        database=MARIADB_DATABASE,
        charset="utf8mb4",
        autocommit=True,
        cursorclass=DictCursor,
        connect_timeout=10,
    )
    if MARIADB_SOCKET:
        kwargs["unix_socket"] = MARIADB_SOCKET
    else:
        kwargs["host"] = MARIADB_HOST
        kwargs["port"] = MARIADB_PORT
    return pymysql.connect(**kwargs)


def _mariadb_acquire():
    global _mariadb_pool
    if _mariadb_pool is None:
        with _pool_lock:
            if _mariadb_pool is None:
                _mariadb_pool = queue.Queue(maxsize=10)
                for _ in range(10):
                    _mariadb_pool.put(_new_mariadb_conn())
    try:
        return _mariadb_pool.get(block=False)
    except queue.Empty:
        return _new_mariadb_conn()


def _mariadb_release(conn):
    try:
        conn.ping(reconnect=False)
        _mariadb_pool.put(conn)
    except Exception:
        try:
            conn.close()
        except Exception:
            pass


# --- SQLite connection ------------------------------------------------------
def _sqlite_connect():
    data_dir = os.path.dirname(SQLITE_PATH)
    if data_dir and not os.path.exists(data_dir):
        os.makedirs(data_dir, exist_ok=True)
    conn = sqlite3.connect(SQLITE_PATH, isolation_level=None)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def _connect():
    if DB_TYPE == "mariadb":
        return _mariadb_acquire()
    return _sqlite_connect()


def _disconnect(conn):
    if DB_TYPE == "mariadb":
        _mariadb_release(conn)
    else:
        conn.close()


def _adapt_sql(sql):
    # pymysql uses %s placeholders; sqlite3 uses ?. Business code writes ?.
    return sql.replace("?", "%s") if DB_TYPE == "mariadb" else sql


def _serialize(value):
    if value is None:
        return None
    if isinstance(value, (datetime.datetime, datetime.date)):
        return value.isoformat()
    return value


def _row_to_dict(row):
    if row is None:
        return None
    if isinstance(row, dict):
        return {k: _serialize(v) for k, v in row.items()}
    return {k: _serialize(row[k]) for k in row.keys()}


# Columns added after the first release. `CREATE TABLE IF NOT EXISTS` does
# nothing to a table that already exists, so new metrics need an explicit
# additive migration or they silently never appear in production.
MIGRATIONS = {
    "sync_status": [
        # A full backfill runs for many minutes, so the UI needs to show how
        # far along it is rather than an indefinite spinner.
        ("progress_current", "INT"),
        ("progress_total", "INT"),
        ("started_at", "DATETIME"),
    ],
    "health_data": [
        # activity / energy
        ("distance_meters", "DOUBLE"),
        ("active_calories", "DOUBLE"),
        ("bmr_calories", "DOUBLE"),
        ("floors_ascended", "DOUBLE"),
        ("floors_descended", "DOUBLE"),
        ("intensity_minutes", "INT"),
        ("step_goal", "INT"),
        ("sedentary_seconds", "INT"),
        ("active_seconds", "INT"),
        # heart / stress
        ("heart_rate_max", "INT"),
        ("heart_rate_min", "INT"),
        ("stress_max", "INT"),
        # body battery
        ("body_battery_high", "INT"),
        ("body_battery_low", "INT"),
        ("body_battery_charged", "INT"),
        ("body_battery_drained", "INT"),
        # breathing / blood oxygen
        ("spo2_avg", "DOUBLE"),
        ("spo2_min", "INT"),
        ("respiration_avg", "DOUBLE"),
        ("respiration_min", "DOUBLE"),
        ("respiration_max", "DOUBLE"),
        # sleep detail
        ("sleep_deep_seconds", "INT"),
        ("sleep_light_seconds", "INT"),
        ("sleep_rem_seconds", "INT"),
        ("sleep_awake_seconds", "INT"),
        ("sleep_spo2_avg", "DOUBLE"),
        ("sleep_respiration_avg", "DOUBLE"),
        ("sleep_stress_avg", "DOUBLE"),
        # training
        ("training_readiness", "INT"),
        ("vo2max", "DOUBLE"),
        ("endurance_score", "INT"),
    ],
}


def _existing_columns(cur, table):
    if DB_TYPE == "mariadb":
        cur.execute(
            "SELECT COLUMN_NAME FROM information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s",
            [table],
        )
        return {r["COLUMN_NAME"] if isinstance(r, dict) else r[0] for r in cur.fetchall()}
    cur.execute(f"PRAGMA table_info({table})")
    return {row[1] for row in cur.fetchall()}


def _migrate(cur):
    for table, columns in MIGRATIONS.items():
        present = _existing_columns(cur, table)
        for name, coltype in columns:
            if name in present:
                continue
            # SQLite has no "ADD COLUMN IF NOT EXISTS"; the membership check
            # above is what keeps this idempotent on both backends.
            cur.execute(f"ALTER TABLE {table} ADD COLUMN {name} {coltype}")


# --- 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 _statements(SCHEMA):
            cur.execute(_adapt_sql(stmt))
        _migrate(cur)
    finally:
        _disconnect(conn)


def execute(sql, params=None):
    params = params or []
    conn = _connect()
    try:
        cur = conn.cursor()
        cur.execute(_adapt_sql(sql), params)
        return {"id": cur.lastrowid, "changes": cur.rowcount}
    finally:
        _disconnect(conn)


def query_one(sql, params=None):
    params = params or []
    conn = _connect()
    try:
        cur = conn.cursor()
        cur.execute(_adapt_sql(sql), params)
        return _row_to_dict(cur.fetchone())
    finally:
        _disconnect(conn)


def query_all(sql, params=None):
    params = params or []
    conn = _connect()
    try:
        cur = conn.cursor()
        cur.execute(_adapt_sql(sql), params)
        return [_row_to_dict(r) for r in cur.fetchall()]
    finally:
        _disconnect(conn)