NewWebConfigurator/src/db_manager.py

"""
db_manager.py — Configuration Storage Layer (Multi-Backend)

Storage priority:
  1. sqlite3 Python module (preferred, full dpworldapp compat)
  2. sqlite3 CLI via subprocess (dpworldapp compat on minimal Python)
  3. JSON file fallback (no dpworldapp compat, temp only)
"""

import os
import sys
import json
import subprocess
import shutil
import threading
import time
from datetime import datetime, timezone

try:
    import sqlite3
    HAS_SQLITE = True
except ImportError:
    HAS_SQLITE = False

DB_PATH_DEFAULT = os.path.join(os.path.expanduser("~"), "db", "dynamic_data.db")

# save_config retries transient SQLite "database is locked" errors: a competing
# writer can briefly hold the lock even with PRAGMA busy_timeout set. Total
# attempts, and the pause between them.
_WRITE_MAX_ATTEMPTS = 3
_WRITE_RETRY_DELAY_S = 0.1


def _has_sqlite3_cli():
    """Check if sqlite3 command-line tool is available."""
    return shutil.which("sqlite3") is not None


class DBManager:
    """Configuration storage manager with 3-tier backend."""

    # Whitelist of allowed config keys (SQL injection defense for CLI backend)
    ALLOWED_KEYS = frozenset({'device_config', 'protocol_config', 'log_config', 'net_config',
                              'ap_config', 'uplink_config'})  # uplink: Telemetry Uplink (web-owned)

    def __init__(self, db_path=None, backend=None):
        self.db_path = db_path or os.environ.get("DB_PATH", DB_PATH_DEFAULT)
        self._lock = threading.Lock()

        if backend is not None:
            # Explicit backend override (used in tests and special deployments)
            self._backend = backend
            if backend == "json":
                self._json_dir = os.path.dirname(self.db_path)
                os.makedirs(self._json_dir, exist_ok=True)
            print(f"[DB] Backend override: {backend} ({self.db_path})")
        elif HAS_SQLITE:
            self._backend = "python"
            print(f"[DB] Using Python sqlite3: {self.db_path}")
        elif _has_sqlite3_cli():
            self._backend = "cli"
            print(f"[DB] Using sqlite3 CLI: {self.db_path}")
        else:
            self._backend = "json"
            self._json_dir = os.path.dirname(self.db_path)
            os.makedirs(self._json_dir, exist_ok=True)
            print(f"[DB] Fallback to JSON files: {self._json_dir}")

    # ─── Python sqlite3 module ───────────────────────────────────

    def _connect(self):
        conn = sqlite3.connect(self.db_path)
        conn.execute("PRAGMA busy_timeout = 5000")
        conn.execute("PRAGMA journal_mode = WAL")
        conn.row_factory = sqlite3.Row
        return conn

    def _validate_key(self, key):
        """Validate config key against whitelist to prevent SQL injection."""
        if not isinstance(key, str):
            raise ValueError(f"Invalid config key type: {type(key)}")
        if key not in self.ALLOWED_KEYS:
            raise ValueError(f"Invalid config key: {key}")

    # ─── sqlite3 CLI (subprocess) ────────────────────────────────

    def _cli_exec(self, sql):
        """Execute SQL via sqlite3 CLI and return stdout."""
        os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
        result = subprocess.run(
            ["sqlite3", self.db_path],
            input=sql,
            capture_output=True,
            text=True,
            timeout=10
        )
        if result.returncode != 0:
            raise RuntimeError(f"sqlite3 CLI error: {result.stderr.strip()}")
        return result.stdout.strip()

    def _cli_query(self, sql):
        """Execute query via CLI and return result string."""
        return self._cli_exec(sql)

    # ─── JSON file helpers ───────────────────────────────────────

    def _json_path(self, key):
        return os.path.join(self._json_dir, f"{key}.json")

    def _read_json_file(self, key):
        path = self._json_path(key)
        if not os.path.exists(path):
            return None
        try:
            with open(path, "r", encoding="utf-8") as f:
                return json.load(f)
        except (json.JSONDecodeError, IOError):
            return None

    def _write_json_file(self, key, data):
        path = self._json_path(key)
        tmp_path = path + ".tmp"
        json_str = json.dumps(data, ensure_ascii=False, indent=2)
        with open(tmp_path, "w", encoding="utf-8") as f:
            f.write(json_str)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp_path, path)

    # ─── Public API ──────────────────────────────────────────────

    def ensure_tables(self):
        if self._backend == "python":
            with self._lock:
                conn = self._connect()
                try:
                    conn.execute("""
                        CREATE TABLE IF NOT EXISTS board_config (
                            key TEXT PRIMARY KEY NOT NULL,
                            value TEXT NOT NULL
                        )
                    """)
                    conn.execute("""
                        CREATE TABLE IF NOT EXISTS event_history (
                            id INTEGER PRIMARY KEY AUTOINCREMENT,
                            data TEXT NOT NULL
                        )
                    """)
                    conn.commit()
                finally:
                    conn.close()
        elif self._backend == "cli":
            with self._lock:
                self._cli_exec(
                    "CREATE TABLE IF NOT EXISTS board_config "
                    "(key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL);\n"
                    "CREATE TABLE IF NOT EXISTS event_history "
                    "(id INTEGER PRIMARY KEY AUTOINCREMENT, data TEXT NOT NULL);"
                )
        else:
            os.makedirs(self._json_dir, exist_ok=True)

    def get_config(self, key):
        self._validate_key(key)
        if self._backend == "python":
            with self._lock:
                conn = self._connect()
                try:
                    cursor = conn.execute(
                        "SELECT value FROM board_config WHERE key = ?", (key,)
                    )
                    row = cursor.fetchone()
                    if row is None:
                        return None
                    try:
                        return json.loads(row["value"])
                    except json.JSONDecodeError:
                        print(f"[DB] WARNING: malformed JSON in board_config key={key!r}; returning None", file=sys.stderr)
                        return None
                finally:
                    conn.close()
        elif self._backend == "cli":
            with self._lock:
                raw = self._cli_query(
                    f"SELECT value FROM board_config WHERE key = '{key}';"
                )
            if not raw:
                return None
            try:
                return json.loads(raw)
            except json.JSONDecodeError:
                print(f"[DB] WARNING: malformed JSON in board_config key={key!r}; returning None", file=sys.stderr)
                return None
        else:
            with self._lock:
                return self._read_json_file(key)

    def save_config(self, key, data):
        self._validate_key(key)
        json_str = json.dumps(data, ensure_ascii=False, indent=2)

        if self._backend == "python":
            last_error = None
            for attempt in range(_WRITE_MAX_ATTEMPTS):
                # #37 fix: acquire the lock only around the actual DB op; the retry
                # backoff sleep below runs OUTSIDE the lock so a contended write
                # does not stall every other DB thread for the backoff window.
                with self._lock:
                    conn = self._connect()
                    try:
                        conn.execute(
                            "INSERT OR REPLACE INTO board_config (key, value) VALUES (?, ?)",
                            (key, json_str)
                        )
                        conn.commit()
                        return
                    except sqlite3.OperationalError as e:
                        # Retry only the transient lock; re-raise anything else.
                        if "database is locked" not in str(e).lower():
                            raise
                        last_error = e
                    finally:
                        conn.close()
                if attempt < _WRITE_MAX_ATTEMPTS - 1:
                    time.sleep(_WRITE_RETRY_DELAY_S)  # outside the lock
            raise last_error
        elif self._backend == "cli":
            # Escape single quotes in JSON for SQL
            escaped = json_str.replace("'", "''")
            with self._lock:
                self._cli_exec(
                    f"INSERT OR REPLACE INTO board_config (key, value) "
                    f"VALUES ('{key}', '{escaped}');"
                )
        else:
            with self._lock:
                self._write_json_file(key, data)

    def update_config(self, key, mutator, default=None):
        """
        Atomic read-modify-write of board_config entry.

        Args:
            key: board_config key (must be in ALLOWED_KEYS)
            mutator: callable(current_value) -> new_value. current_value is
                     the JSON-deserialized current value (or `default` if absent).
                     Must NOT mutate input — return a new dict.
            default: value to pass to mutator when key is missing (default: None)

        Returns:
            The new value (after mutator) — i.e., what was actually saved.
            If mutator returns None the key is deleted and None is returned.

        Atomic guarantees:
            - python backend: BEGIN IMMEDIATE + SELECT + UPDATE + COMMIT within
              one connection. Other writers block; lock retries with exponential
              backoff (same _WRITE_MAX_ATTEMPTS pattern).
            - cli backend: serialized by _lock (cli is sequential).
            - json backend: lock + read + mutator + atomic write (tempfile+rename).
        """
        self._validate_key(key)

        if self._backend == "python":
            last_error = None
            for attempt in range(_WRITE_MAX_ATTEMPTS):
                # #37 fix: lock only around the atomic RMW; the retry backoff sleep
                # below runs OUTSIDE the lock so a contended write does not stall
                # every other DB thread for the backoff window.
                with self._lock:
                    conn = self._connect()
                    try:
                        # BEGIN IMMEDIATE — acquires write lock immediately,
                        # blocking other writers but allowing concurrent reads.
                        conn.execute("BEGIN IMMEDIATE")
                        cursor = conn.execute(
                            "SELECT value FROM board_config WHERE key = ?", (key,)
                        )
                        row = cursor.fetchone()
                        if row is None:
                            current = default
                        else:
                            try:
                                current = json.loads(row["value"])
                            except json.JSONDecodeError:
                                print(f"[DB] WARNING: malformed JSON in board_config key={key!r}; using default", file=sys.stderr)
                                current = default
                        new_value = mutator(current)
                        if new_value is None:
                            # mutator returned None → delete key
                            conn.execute(
                                "DELETE FROM board_config WHERE key = ?", (key,)
                            )
                        else:
                            json_str = json.dumps(new_value, ensure_ascii=False, indent=2)
                            conn.execute(
                                "INSERT OR REPLACE INTO board_config (key, value) VALUES (?, ?)",
                                (key, json_str)
                            )
                        conn.commit()
                        return new_value
                    except sqlite3.OperationalError as e:
                        if "database is locked" not in str(e).lower():
                            raise
                        last_error = e
                        try:
                            conn.rollback()
                        except Exception:
                            pass
                    finally:
                        conn.close()
                if attempt < _WRITE_MAX_ATTEMPTS - 1:
                    time.sleep(_WRITE_RETRY_DELAY_S)  # outside the lock
            raise last_error
        elif self._backend == "cli":
            with self._lock:
                raw = self._cli_query(
                    f"SELECT value FROM board_config WHERE key = '{key}';"
                )
                if not raw:
                    current = default
                else:
                    try:
                        current = json.loads(raw)
                    except json.JSONDecodeError:
                        print(f"[DB] WARNING: malformed JSON in board_config key={key!r}; using default", file=sys.stderr)
                        current = default
                new_value = mutator(current)
                if new_value is None:
                    self._cli_exec(f"DELETE FROM board_config WHERE key = '{key}';")
                else:
                    json_str = json.dumps(new_value, ensure_ascii=False, indent=2)
                    escaped = json_str.replace("'", "''")
                    self._cli_exec(
                        f"INSERT OR REPLACE INTO board_config (key, value) "
                        f"VALUES ('{key}', '{escaped}');"
                    )
                return new_value
        else:  # json backend
            with self._lock:
                current = self._read_json_file(key)
                if current is None:
                    current = default
                new_value = mutator(current)
                if new_value is None:
                    path = self._json_path(key)
                    if os.path.exists(path):
                        os.remove(path)
                else:
                    self._write_json_file(key, new_value)
                return new_value

    def config_exists(self, key):
        self._validate_key(key)
        if self._backend == "python":
            with self._lock:
                conn = self._connect()
                try:
                    cursor = conn.execute(
                        "SELECT 1 FROM board_config WHERE key = ? LIMIT 1", (key,)
                    )
                    return cursor.fetchone() is not None
                finally:
                    conn.close()
        elif self._backend == "cli":
            with self._lock:
                raw = self._cli_query(
                    f"SELECT 1 FROM board_config WHERE key = '{key}' LIMIT 1;"
                )
            return bool(raw)
        else:
            with self._lock:
                return os.path.exists(self._json_path(key))

    def config_initialized(self, key):
        """True iff the key exists AND holds a non-empty dict. An empty {} counts as
        NOT initialized — it is the residual state the seed-ordering bug produces
        (web/old-code created an empty/partial row before dpworldapp seeded it).
        Single source of truth for 'absent' across migrations / GET 204 / POST 409 /
        apply DB-write guard (2026-06-29 strengthened merge)."""
        v = self.get_config(key)
        return isinstance(v, dict) and bool(v)

    # ─── Schema meta (migration flags) ──────────────────────────

    def _ensure_schema_meta_table(self) -> None:
        # _lock is acquired here independently; callers must NOT hold the
        # lock when calling this method (sequential acquisition only).
        with self._lock:
            conn = self._connect()
            try:
                conn.execute("""
                    CREATE TABLE IF NOT EXISTS schema_meta (
                        key TEXT PRIMARY KEY,
                        value TEXT NOT NULL,
                        applied_at TEXT NOT NULL
                    )
                """)
                conn.commit()
            finally:
                conn.close()

    def get_schema_meta(self, key: str):
        self._ensure_schema_meta_table()
        with self._lock:
            conn = self._connect()
            try:
                row = conn.execute(
                    "SELECT value FROM schema_meta WHERE key = ?", (key,)
                ).fetchone()
                return row[0] if row else None
            finally:
                conn.close()

    def set_schema_meta(self, key: str, value: str) -> None:
        self._ensure_schema_meta_table()
        now = datetime.now(timezone.utc).isoformat()
        with self._lock:
            conn = self._connect()
            try:
                conn.execute("""
                    INSERT INTO schema_meta(key, value, applied_at)
                    VALUES (?, ?, ?)
                    ON CONFLICT(key) DO UPDATE SET value=excluded.value,
                                                    applied_at=excluded.applied_at
                """, (key, value, now))
                conn.commit()
            finally:
                conn.close()

    def get_raw_json(self, key):
        self._validate_key(key)
        if self._backend == "python":
            with self._lock:
                conn = self._connect()
                try:
                    cursor = conn.execute(
                        "SELECT value FROM board_config WHERE key = ?", (key,)
                    )
                    row = cursor.fetchone()
                    return row["value"] if row else None
                finally:
                    conn.close()
        elif self._backend == "cli":
            with self._lock:
                raw = self._cli_query(
                    f"SELECT value FROM board_config WHERE key = '{key}';"
                )
            return raw if raw else None
        else:
            with self._lock:
                data = self._read_json_file(key)
            return json.dumps(data, ensure_ascii=False, indent=2) if data else None