from __future__ import annotations

import json
import hashlib
import hmac
import math
import secrets
import time
import uuid
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterator

import psycopg

from app.core.config import get_settings


ALL_PERMISSIONS = [
    "dashboard",
    "fine-tune",
    "model-eval",
    "model-inference",
    "model-manage",
    "dataset",
    "data-process",
    "data-convert",
    "compute",
    "hardware",
    "logs",
    "user-settings",
]


def utcnow() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def parse_time(value: str | None) -> datetime | None:
    if not value:
        return None
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def json_loads(value: str | None, default: Any) -> Any:
    if not value:
        return default
    return json.loads(value)


def json_dumps(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))


def new_id(prefix: str) -> str:
    return f"{prefix}_{uuid.uuid4().hex[:12]}"


PASSWORD_HASH_ITERATIONS = 390_000


def hash_password(password: str) -> str:
    salt = secrets.token_hex(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), PASSWORD_HASH_ITERATIONS)
    return f"pbkdf2_sha256${PASSWORD_HASH_ITERATIONS}${salt}${digest.hex()}"


def verify_password(password: str, stored: str) -> tuple[bool, bool]:
    if not stored.startswith("pbkdf2_sha256$"):
        return hmac.compare_digest(password, stored), True
    try:
        _, iterations, salt, expected = stored.split("$", 3)
        digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), int(iterations)).hex()
        return hmac.compare_digest(digest, expected), False
    except ValueError:
        return False, False


def _psycopg_url(database_url: str) -> str:
    return database_url.replace("postgresql+psycopg://", "postgresql://")


def _pg_sql(sql: str) -> str:
    return sql.replace("?", "%s")


class PgRow(dict):
    def __init__(self, columns: list[str], values: tuple[Any, ...]) -> None:
        super().__init__(zip(columns, values))
        self._values = values

    def __getitem__(self, key: str | int) -> Any:
        if isinstance(key, int):
            return self._values[key]
        return super().__getitem__(key)


class PgCursor:
    def __init__(self, cursor: psycopg.Cursor[Any]) -> None:
        self.cursor = cursor

    def execute(self, sql: str, params: tuple[Any, ...] | list[Any] | None = None) -> "PgCursor":
        self.cursor.execute(_pg_sql(sql), params)
        return self

    def fetchone(self) -> PgRow | None:
        row = self.cursor.fetchone()
        if row is None:
            return None
        return PgRow(self._columns(), tuple(row))

    def fetchall(self) -> list[PgRow]:
        columns = self._columns()
        return [PgRow(columns, tuple(row)) for row in self.cursor.fetchall()]

    def _columns(self) -> list[str]:
        return [col.name for col in self.cursor.description or []]


class PgConnection:
    def __init__(self, conn: psycopg.Connection[Any]) -> None:
        self.conn = conn

    def execute(self, sql: str, params: tuple[Any, ...] | list[Any] | None = None) -> PgCursor:
        cursor = PgCursor(self.conn.cursor())
        return cursor.execute(sql, params)

    def executemany(self, sql: str, params_seq: list[tuple[Any, ...]] | list[list[Any]]) -> None:
        with self.conn.cursor() as cursor:
            cursor.executemany(_pg_sql(sql), params_seq)

    def executescript(self, sql: str) -> None:
        with self.conn.cursor() as cursor:
            for statement in sql.split(";"):
                statement = statement.strip()
                if statement:
                    cursor.execute(statement)

    def commit(self) -> None:
        self.conn.commit()

    def rollback(self) -> None:
        self.conn.rollback()

    def close(self) -> None:
        self.conn.close()


class PlatformStore:
    """PostgreSQL-backed store for the first runnable platform version.

    This store mirrors the API-facing subset needed by the first system
    iteration while using the same PostgreSQL dependency as later production
    development.
    """

    def __init__(self, database_url: str | None = None) -> None:
        settings = get_settings()
        self.database_url = _psycopg_url(database_url or settings.database_url)
        self.ensure_schema()
        self.ensure_seed_data()

    @contextmanager
    def connect(self) -> Iterator["PgConnection"]:
        raw_conn = psycopg.connect(self.database_url)
        conn = PgConnection(raw_conn)
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def ensure_schema(self) -> None:
        schema_path = Path(__file__).with_name("sql") / "001_platform_runtime.sql"
        with self.connect() as conn:
            conn.executescript(schema_path.read_text(encoding="utf-8"))
            columns = conn.execute(
                "SELECT column_name FROM information_schema.columns WHERE table_name='users'"
            ).fetchall()
            column_names = {row["column_name"] for row in columns}
            if "password" in column_names and "password_hash" not in column_names:
                conn.execute("ALTER TABLE users RENAME COLUMN password TO password_hash")

    def ensure_seed_data(self) -> None:
        with self.connect() as conn:
            if conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] > 0:
                return

            now = utcnow()
            users = [
                ("u_admin", "admin", "admin123", "Platform Admin", "admin", "active", ALL_PERMISSIONS, 1),
                (
                    "u_operator",
                    "operator",
                    "operator123",
                    "Platform Operator",
                    "operator",
                    "active",
                    [p for p in ALL_PERMISSIONS if p != "user-settings"],
                    0,
                ),
            ]
            conn.executemany(
                """
                INSERT INTO users
                (id, username, password_hash, display_name, role, status, permissions, create_time, protected)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                [(u[0], u[1], hash_password(u[2]), u[3], u[4], u[5], json_dumps(u[6]), now, u[7]) for u in users],
            )

    def _duration(self, start_time: str | None, end_time: str | None = None) -> str:
        start = parse_time(start_time)
        if not start:
            return ""
        end = parse_time(end_time) or datetime.now(timezone.utc)
        seconds = max(0, int((end - start).total_seconds()))
        minutes, sec = divmod(seconds, 60)
        hours, minutes = divmod(minutes, 60)
        if hours:
            return f"{hours}h {minutes}m {sec}s"
        if minutes:
            return f"{minutes}m {sec}s"
        return f"{sec}s"

    def refresh_runtime_state(self) -> None:
        if get_settings().compute_mode != "simulator":
            return

        with self.connect() as conn:
            rows = conn.execute(
                "SELECT * FROM fine_tune_tasks WHERE status IN ('syncing','queued','running')"
            ).fetchall()
            now_dt = datetime.now(timezone.utc)
            for row in rows:
                start = parse_time(row["start_time"])
                if not start:
                    continue
                age = max(0, int((now_dt - start).total_seconds()))
                if age < 4:
                    status, progress = "syncing", 8 + age
                elif age < 8:
                    status, progress = "queued", 18 + age
                elif age < 70:
                    status = "running"
                    progress = min(96, 25 + int((age - 8) / 62 * 70))
                else:
                    status, progress = "completed", 100

                payload = json_loads(row["payload"], {})
                payload.update(
                    {
                        "status": status,
                        "progress": progress,
                        "train_duration": self._duration(row["start_time"], utcnow() if status == "completed" else None),
                    }
                )
                completed_at = row["completed_at"] or (utcnow() if status == "completed" else None)
                conn.execute(
                    """
                    UPDATE fine_tune_tasks
                    SET status=?, progress=?, payload=?, completed_at=?
                    WHERE id=?
                    """,
                    (status, progress, json_dumps(payload), completed_at, row["id"]),
                )
                if status == "completed":
                    self._ensure_trained_model(conn, payload)

            sync_rows = conn.execute(
                "SELECT * FROM resource_sync_jobs WHERE status IN ('pending','running')"
            ).fetchall()
            for row in sync_rows:
                created = parse_time(row["create_time"])
                age = int((now_dt - created).total_seconds()) if created else 0
                status = "completed" if age >= 6 else "running"
                progress = 100 if status == "completed" else min(95, 15 + age * 12)
                completed_at = row["completed_at"] or (utcnow() if status == "completed" else None)
                conn.execute(
                    "UPDATE resource_sync_jobs SET status=?, progress=?, completed_at=? WHERE id=?",
                    (status, progress, completed_at, row["id"]),
                )

    def _ensure_trained_model(self, conn: PgConnection, task: dict[str, Any]) -> None:
        name = task.get("output_model_name") or f"{task['name']}-lora"
        exists = conn.execute("SELECT id FROM trained_models WHERE name=?", (name,)).fetchone()
        if exists:
            return
        model = conn.execute("SELECT path FROM models WHERE id=?", (task.get("base_model"),)).fetchone()
        conn.execute(
            """
            INSERT INTO trained_models
            (id, name, train_methods, base_model_path, create_time, merged, merging, merged_path)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                new_id("tm"),
                name,
                json_dumps([{"name": task.get("train_method", "lora")}]),
                model["path"] if model else "",
                utcnow(),
                0,
                0,
                f"/data/yg-ft/outputs/{task['name']}/adapter",
            ),
        )

    def users(self) -> list[dict[str, Any]]:
        with self.connect() as conn:
            rows = conn.execute("SELECT * FROM users ORDER BY create_time").fetchall()
            return [self._user(row) for row in rows]

    def login(self, username: str, password: str) -> dict[str, Any] | None:
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
            if not row or row["status"] != "active":
                return None
            matched, legacy_plaintext = verify_password(password, row["password_hash"])
            if not matched:
                return None
            last_login = utcnow()
            if legacy_plaintext:
                conn.execute(
                    "UPDATE users SET password_hash=?, last_login=? WHERE id=?",
                    (hash_password(password), last_login, row["id"]),
                )
            else:
                conn.execute("UPDATE users SET last_login=? WHERE id=?", (last_login, row["id"]))
            data = self._user(row)
            data["last_login"] = last_login
            return data

    def create_user(self, payload: dict[str, Any]) -> dict[str, Any]:
        user_id = new_id("u")
        permissions = payload.get("permissions") or (ALL_PERMISSIONS if payload.get("role") == "admin" else ["dashboard"])
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO users
                (id, username, password_hash, display_name, role, status, permissions, create_time, protected)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
                """,
                (
                    user_id,
                    payload["username"],
                    hash_password(payload.get("password", "platform123")),
                    payload.get("display_name") or payload["username"],
                    payload.get("role", "viewer"),
                    payload.get("status", "active"),
                    json_dumps(permissions),
                    utcnow(),
                ),
            )
            return self._user(conn.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone())

    def update_user(self, user_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
            if not row:
                raise KeyError(user_id)
            values = {
                "role": payload.get("role", row["role"]),
                "status": payload.get("status", row["status"]),
                "permissions": json_dumps(payload.get("permissions", json_loads(row["permissions"], []))),
            }
            conn.execute(
                "UPDATE users SET role=?, status=?, permissions=? WHERE id=?",
                (values["role"], values["status"], values["permissions"], user_id),
            )
            return self._user(conn.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone())

    def delete_user(self, user_id: str) -> None:
        with self.connect() as conn:
            row = conn.execute("SELECT protected FROM users WHERE id=?", (user_id,)).fetchone()
            if not row:
                raise KeyError(user_id)
            if row["protected"]:
                raise ValueError("protected user cannot be deleted")
            conn.execute("DELETE FROM users WHERE id=?", (user_id,))

    def _user(self, row: PgRow) -> dict[str, Any]:
        return {
            "id": row["id"],
            "username": row["username"],
            "display_name": row["display_name"],
            "role": row["role"],
            "status": row["status"],
            "permissions": json_loads(row["permissions"], []),
            "create_time": row["create_time"],
            "last_login": row["last_login"],
            "protected": bool(row["protected"]),
        }

    def models(self) -> list[dict[str, Any]]:
        with self.connect() as conn:
            return [dict(row) for row in conn.execute("SELECT * FROM models ORDER BY create_time DESC").fetchall()]

    def model(self, model_id: str) -> dict[str, Any]:
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM models WHERE id=?", (model_id,)).fetchone()
            if not row:
                raise KeyError(model_id)
            return dict(row)

    def model_by_name(self, name: str) -> dict[str, Any]:
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM models WHERE name=?", (name,)).fetchone()
            if not row:
                raise KeyError(name)
            return dict(row)

    def create_model(self, payload: dict[str, Any]) -> dict[str, Any]:
        model_id = payload.get("id") or new_id("m")
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO models
                (id, name, type, purpose, model_source, description, path, api_url, api_key, online_model_name, create_time)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    model_id,
                    payload["name"],
                    payload.get("type", "LLM"),
                    payload.get("purpose", "training"),
                    payload.get("model_source", "local"),
                    payload.get("description"),
                    payload.get("path"),
                    payload.get("api_url"),
                    payload.get("api_key"),
                    payload.get("online_model_name"),
                    utcnow(),
                ),
            )
            return self.model(model_id)

    def update_model(self, model_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        current = self.model(model_id)
        merged = {**current, **payload}
        with self.connect() as conn:
            conn.execute(
                """
                UPDATE models
                SET name=?, type=?, purpose=?, model_source=?, description=?, path=?, api_url=?, api_key=?, online_model_name=?
                WHERE id=?
                """,
                (
                    merged["name"],
                    merged.get("type", "LLM"),
                    merged.get("purpose", "training"),
                    merged.get("model_source", "local"),
                    merged.get("description"),
                    merged.get("path"),
                    merged.get("api_url"),
                    merged.get("api_key"),
                    merged.get("online_model_name"),
                    model_id,
                ),
            )
            return self.model(model_id)

    def delete_model(self, model_id: str) -> None:
        with self.connect() as conn:
            conn.execute("DELETE FROM models WHERE id=?", (model_id,))

    def trained_models(self) -> list[dict[str, Any]]:
        self.refresh_runtime_state()
        with self.connect() as conn:
            rows = conn.execute("SELECT * FROM trained_models ORDER BY create_time DESC").fetchall()
            return [
                {
                    **dict(row),
                    "train_methods": json_loads(row["train_methods"], []),
                    "merged": bool(row["merged"]),
                    "merging": bool(row["merging"]),
                }
                for row in rows
            ]

    def datasets(self) -> list[dict[str, Any]]:
        with self.connect() as conn:
            rows = conn.execute("SELECT * FROM datasets ORDER BY create_time DESC").fetchall()
            return [self._dataset(conn, row) for row in rows]

    def dataset(self, dataset_id: str) -> dict[str, Any]:
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM datasets WHERE id=?", (dataset_id,)).fetchone()
            if not row:
                raise KeyError(dataset_id)
            return self._dataset(conn, row)

    def _dataset(self, conn: PgConnection, row: PgRow) -> dict[str, Any]:
        files = conn.execute(
            "SELECT id, name, size, active_version_id, create_time FROM dataset_files WHERE dataset_id=? ORDER BY create_time",
            (row["id"],),
        ).fetchall()
        return {
            **dict(row),
            "files": [
                {
                    "id": f["id"],
                    "name": f["name"],
                    "size": f["size"],
                    "active_version_id": f["active_version_id"],
                    "create_time": f["create_time"],
                }
                for f in files
            ],
        }

    def create_dataset(self, payload: dict[str, Any]) -> dict[str, Any]:
        dataset_id = payload.get("id") or new_id("ds")
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO datasets
                (id, name, type, storage_type, source, task_id, size, count, description, create_time)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    dataset_id,
                    payload["name"],
                    payload.get("type", "train"),
                    payload.get("storage_type", "local"),
                    payload.get("source", "upload"),
                    payload.get("task_id"),
                    payload.get("size", "0 KB"),
                    payload.get("count", 0),
                    payload.get("description"),
                    utcnow(),
                ),
            )
            return self._dataset(conn, conn.execute("SELECT * FROM datasets WHERE id=?", (dataset_id,)).fetchone())

    def update_dataset(self, dataset_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        current = self.dataset(dataset_id)
        merged = {**current, **payload}
        with self.connect() as conn:
            conn.execute(
                """
                UPDATE datasets
                SET name=?, type=?, storage_type=?, source=?, task_id=?, size=?, count=?, description=?
                WHERE id=?
                """,
                (
                    merged["name"],
                    merged.get("type", "train"),
                    merged.get("storage_type", "local"),
                    merged.get("source", "upload"),
                    merged.get("task_id"),
                    merged.get("size", "0 KB"),
                    merged.get("count", 0),
                    merged.get("description"),
                    dataset_id,
                ),
            )
            return self._dataset(conn, conn.execute("SELECT * FROM datasets WHERE id=?", (dataset_id,)).fetchone())

    def delete_dataset(self, dataset_id: str) -> None:
        with self.connect() as conn:
            conn.execute("DELETE FROM dataset_files WHERE dataset_id=?", (dataset_id,))
            conn.execute("DELETE FROM datasets WHERE id=?", (dataset_id,))

    def add_dataset_file(self, conn: PgConnection, dataset_id: str, name: str, content: str) -> dict[str, Any]:
        now = utcnow()
        file_id = new_id("file")
        version_id = f"{file_id}_v1"
        size = f"{max(1, len(content.encode('utf-8')) // 1024)} KB"
        conn.execute(
            """
            INSERT INTO dataset_files
            (id, dataset_id, name, size, content, active_version_id, versions, create_time)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                file_id,
                dataset_id,
                name,
                size,
                content,
                version_id,
                json_dumps([{"id": version_id, "version": 1, "create_time": now, "description": "uploaded"}]),
                now,
            ),
        )
        count = len([line for line in content.splitlines() if line.strip()])
        conn.execute(
            "UPDATE datasets SET count=count+?, size=? WHERE id=?",
            (count, size, dataset_id),
        )
        return {"id": file_id, "name": name, "size": size}

    def dataset_file(self, file_id: str) -> PgRow:
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM dataset_files WHERE id=?", (file_id,)).fetchone()
            if not row:
                raise KeyError(file_id)
            return row

    def file_versions(self, file_id: str) -> dict[str, Any]:
        row = self.dataset_file(file_id)
        versions = json_loads(row["versions"], [])
        return {
            "versions": versions,
            "active_version_id": row["active_version_id"],
            "next_version_number": len(versions) + 1,
        }

    def create_file_version(self, file_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM dataset_files WHERE id=?", (file_id,)).fetchone()
            if not row:
                raise KeyError(file_id)
            versions = json_loads(row["versions"], [])
            version = {
                "id": f"{file_id}_v{len(versions) + 1}",
                "version": len(versions) + 1,
                "create_time": utcnow(),
                "description": payload.get("description", "online edit"),
            }
            versions.append(version)
            conn.execute(
                "UPDATE dataset_files SET content=?, active_version_id=?, versions=? WHERE id=?",
                (payload.get("content", ""), version["id"], json_dumps(versions), file_id),
            )
            return {"version": version, "content": payload.get("content", "")}

    def activate_file_version(self, file_id: str, version_id: str) -> dict[str, Any]:
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM dataset_files WHERE id=?", (file_id,)).fetchone()
            if not row:
                raise KeyError(file_id)
            versions = json_loads(row["versions"], [])
            version = next((item for item in versions if item["id"] == version_id), None)
            if not version:
                raise KeyError(version_id)
            conn.execute("UPDATE dataset_files SET active_version_id=? WHERE id=?", (version_id, file_id))
            return {"version": version, "content": row["content"]}

    def tasks(self) -> list[dict[str, Any]]:
        self.refresh_runtime_state()
        with self.connect() as conn:
            rows = conn.execute("SELECT * FROM fine_tune_tasks ORDER BY create_time DESC").fetchall()
            return [self._task(row) for row in rows]

    def task(self, task_id: str) -> dict[str, Any]:
        self.refresh_runtime_state()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM fine_tune_tasks WHERE id=?", (task_id,)).fetchone()
            if not row:
                raise KeyError(task_id)
            return self._task(row)

    def _task(self, row: PgRow) -> dict[str, Any]:
        payload = json_loads(row["payload"], {})
        payload.update(
            {
                "id": row["id"],
                "status": row["status"],
                "progress": row["progress"],
                "process_id": row["process_id"],
                "create_time": row["create_time"],
                "gpus": json_loads(row["gpus"], payload.get("gpus", [])),
                "train_duration": self._duration(row["start_time"], row["completed_at"]) if row["start_time"] else "",
                "compute_node_id": row["compute_node_id"],
                "sync_job_id": row["sync_job_id"],
            }
        )
        return payload

    def create_task(self, payload: dict[str, Any]) -> dict[str, Any]:
        task_id = str(payload.get("task_id") or payload.get("id") or new_id("ft"))
        name = payload.get("name") or f"fine-tune-{task_id[-6:]}"
        base_model = payload.get("base_model") or payload.get("base_model_id")
        train_dataset_id = payload.get("train_dataset_id")
        if not base_model:
            raise ValueError("base_model or base_model_id is required")
        if not train_dataset_id:
            raise ValueError("train_dataset_id is required")
        now = utcnow()
        task = {
            "id": task_id,
            "name": name,
            "description": payload.get("description", ""),
            "status": "pending",
            "train_type": payload.get("train_type", "SFT"),
            "train_method": payload.get("train_method", "lora"),
            "template": payload.get("template", "qwen"),
            "base_model": base_model,
            "train_dataset_id": train_dataset_id,
            "auto_merge": bool(payload.get("auto_merge", False)),
            "output_model_name": payload.get("output_model_name") or f"{name}-lora",
            "gpus": payload.get("gpus") or [],
            "batch_size": payload.get("batch_size", 2),
            "learning_rate": payload.get("learning_rate", 0.0002),
            "n_epochs": payload.get("n_epochs", 3),
            "save_steps": payload.get("save_steps", 50),
            "lr_scheduler_type": payload.get("lr_scheduler_type", "cosine"),
            "max_length": payload.get("max_length", 2048),
            "warmup_ratio": payload.get("warmup_ratio", 0.03),
            "weight_decay": payload.get("weight_decay", 0.01),
            "lora_alpha": payload.get("lora_alpha", 16),
            "lora_dropout": payload.get("lora_dropout", 0.05),
            "lora_rank": payload.get("lora_rank", 8),
            "quantization_bit": payload.get("quantization_bit", 4),
            "export_quantized": bool(payload.get("export_quantized", False)),
            "quant_method": payload.get("quant_method", "bnb"),
            "quant_bits": payload.get("quant_bits", 4),
            "quant_group_size": payload.get("quant_group_size", 128),
            "export_format": payload.get("export_format", "safetensors"),
            "progress": 0,
            "process_id": None,
            "train_duration": "",
            "create_time": now,
        }
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO fine_tune_tasks
                (id, name, payload, status, progress, process_id, create_time, gpus)
                VALUES (?, ?, ?, 'pending', 0, NULL, ?, ?)
                """,
                (task_id, name, json_dumps(task), now, json_dumps(task["gpus"])),
            )
        return task

    def update_task(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        current = self.task(task_id)
        merged = {**current, **payload, "id": task_id}
        with self.connect() as conn:
            conn.execute(
                "UPDATE fine_tune_tasks SET name=?, payload=?, gpus=? WHERE id=?",
                (merged["name"], json_dumps(merged), json_dumps(merged.get("gpus", [])), task_id),
            )
        return self.task(task_id)

    def start_task(self, payload: dict[str, Any]) -> dict[str, Any]:
        task_id = str(payload.get("task_id") or payload.get("id"))
        current = self.task(task_id)
        merged = {**current, **payload, "id": task_id, "status": "syncing", "progress": 8}
        node = self.schedule_node(payload)
        selected_gpus = payload.get("gpus") or merged.get("gpus") or [0]
        process_id = int(43000 + (time.time() % 10000))
        sync_job_id = self.create_sync_job(node["id"], current)
        with self.connect() as conn:
            conn.execute(
                """
                UPDATE fine_tune_tasks
                SET payload=?, status='syncing', progress=8, process_id=?, start_time=?,
                    compute_node_id=?, gpus=?, sync_job_id=?
                WHERE id=?
                """,
                (
                    json_dumps({**merged, "process_id": process_id, "gpus": selected_gpus}),
                    process_id,
                    utcnow(),
                    node["id"],
                    json_dumps(selected_gpus),
                    sync_job_id,
                    task_id,
                ),
            )
        return self.task(task_id)

    def stop_task(self, task_id: str) -> dict[str, Any]:
        task = self.task(task_id)
        task.update({"status": "failed", "progress": min(task.get("progress", 0), 99)})
        with self.connect() as conn:
            conn.execute(
                "UPDATE fine_tune_tasks SET status='failed', payload=?, completed_at=? WHERE id=?",
                (json_dumps(task), utcnow(), task_id),
            )
        return self.task(task_id)

    def delete_task(self, task_id: str) -> None:
        with self.connect() as conn:
            conn.execute("DELETE FROM fine_tune_tasks WHERE id=?", (task_id,))

    def schedule_node(self, payload: dict[str, Any]) -> dict[str, Any]:
        requested = payload.get("requested_node_id") or payload.get("compute_node_id")
        nodes = self.compute_nodes()
        candidates = [
            n
            for n in nodes
            if n["enabled"] and n["scheduler_status"] == "online" and n["current_running_jobs"] < n["max_parallel_jobs"]
        ]
        if requested:
            selected = next((n for n in candidates if n["id"] == requested), None)
            if selected:
                return selected
        if not candidates:
            raise RuntimeError("no available compute node")
        return sorted(candidates, key=lambda n: (-n["scheduler_weight"], n["current_running_jobs"], n["code"]))[0]

    def create_sync_job(self, node_id: str, task: dict[str, Any]) -> str:
        sync_id = new_id("sync")
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO resource_sync_jobs
                (id, target_node_id, resources, status, progress, create_time)
                VALUES (?, ?, ?, 'pending', 0, ?)
                """,
                (
                    sync_id,
                    node_id,
                    json_dumps(
                        [
                            {"resource_type": "model", "resource_id": task.get("base_model")},
                            {"resource_type": "dataset", "resource_id": task.get("train_dataset_id")},
                        ]
                    ),
                    utcnow(),
                ),
            )
        return sync_id

    def progress(self, task_id: str) -> dict[str, Any]:
        task = self.task(task_id)
        status = task.get("status", "pending")
        labels = {
            "pending": "waiting for start",
            "syncing": "syncing model and dataset to compute node",
            "queued": "waiting for GPU slot",
            "running": "training with LLaMA-Factory",
            "completed": "training completed",
            "failed": "training stopped",
        }
        progress = int(task.get("progress", 0) or 0)
        eta = "--" if status in {"completed", "failed"} else f"{max(1, math.ceil((100 - progress) / 10))} min"
        return {
            "status": status,
            "progress": progress,
            "step": labels.get(status, status),
            "speed": task.get("train_speed") or "--",
            "eta": eta,
        }

    def compute_nodes(self) -> list[dict[str, Any]]:
        self.refresh_runtime_state()
        with self.connect() as conn:
            running = conn.execute(
                "SELECT compute_node_id, COUNT(*) AS cnt FROM fine_tune_tasks WHERE status IN ('syncing','queued','running') GROUP BY compute_node_id"
            ).fetchall()
            running_map = {r["compute_node_id"]: r["cnt"] for r in running}
            rows = conn.execute("SELECT * FROM compute_nodes ORDER BY scheduler_weight DESC, code").fetchall()
            return [
                {
                    **dict(row),
                    "enabled": bool(row["enabled"]),
                    "tags": json_loads(row["tags"], []),
                    "health_detail": json_loads(row["health_detail"], {}),
                    "current_running_jobs": running_map.get(row["id"], 0),
                }
                for row in rows
            ]

    def update_compute_node(self, node_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        current = next((n for n in self.compute_nodes() if n["id"] == node_id), None)
        if not current:
            raise KeyError(node_id)
        merged = {**current, **payload}
        with self.connect() as conn:
            conn.execute(
                """
                UPDATE compute_nodes
                SET name=?, api_base_url=?, file_gateway_url=?, enabled=?, scheduler_status=?,
                    scheduler_weight=?, tags=?, max_parallel_jobs=?, last_health_check_at=?
                WHERE id=?
                """,
                (
                    merged["name"],
                    merged["api_base_url"],
                    merged["file_gateway_url"],
                    1 if merged["enabled"] else 0,
                    merged["scheduler_status"],
                    merged["scheduler_weight"],
                    json_dumps(merged["tags"]),
                    merged["max_parallel_jobs"],
                    utcnow(),
                    node_id,
                ),
            )
        return next(n for n in self.compute_nodes() if n["id"] == node_id)

    def create_compute_node(self, payload: dict[str, Any]) -> dict[str, Any]:
        node_id = payload.get("id") or new_id("node")
        now = utcnow()
        tags = payload.get("tags") or []
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO compute_nodes
                (id, code, name, api_base_url, file_gateway_url, enabled, scheduler_status,
                 scheduler_weight, tags, gpu_count, current_running_jobs, max_parallel_jobs,
                 data_root, model_root, log_root, last_health_check_at, health_detail)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?)
                """,
                (
                    node_id,
                    payload["code"],
                    payload.get("name") or payload["code"],
                    payload["api_base_url"],
                    payload.get("file_gateway_url") or payload["api_base_url"],
                    1 if payload.get("enabled", True) else 0,
                    payload.get("scheduler_status", "offline"),
                    int(payload.get("scheduler_weight", 100)),
                    json_dumps(tags),
                    int(payload.get("gpu_count", 0)),
                    int(payload.get("max_parallel_jobs", 1)),
                    payload.get("data_root", "/data/yg-ft"),
                    payload.get("model_root", "/models"),
                    payload.get("log_root", "/data/yg-ft/training-logs"),
                    now,
                    json_dumps(payload.get("health_detail") or {"status": "registered"}),
                ),
            )
        return next(node for node in self.compute_nodes() if node["id"] == node_id)

    def gpus(self) -> list[dict[str, Any]]:
        self.refresh_runtime_state()
        with self.connect() as conn:
            rows = conn.execute(
                """
                SELECT g.*, n.code AS node_code, n.name AS node_name
                FROM gpus g JOIN compute_nodes n ON n.id = g.node_id
                ORDER BY n.code, g.gpu_index
                """
            ).fetchall()
            running_tasks = [
                self._task(row)
                for row in conn.execute(
                    "SELECT * FROM fine_tune_tasks WHERE status IN ('syncing','queued','running')"
                ).fetchall()
            ]
            items = []
            for row in rows:
                task = next(
                    (
                        t
                        for t in running_tasks
                        if t.get("compute_node_id") == row["node_id"] and row["gpu_index"] in (t.get("gpus") or [])
                    ),
                    None,
                )
                busy = task is not None and task.get("status") == "running"
                reserved = task is not None and task.get("status") in {"syncing", "queued"}
                memory_used = round(row["memory_total_gb"] * (0.72 if busy else 0.18 if reserved else 0.04), 1)
                gpu_percent = 86 if busy else 22 if reserved else 3
                items.append(
                    {
                        "id": row["gpu_index"],
                        "node_id": row["node_id"],
                        "node_code": row["node_code"],
                        "node_name": row["node_name"],
                        "name": row["name"],
                        "uuid": row["uuid"],
                        "gpu_percent": gpu_percent,
                        "memory_used_gb": memory_used,
                        "memory_total_gb": row["memory_total_gb"],
                        "memory_percent": round(memory_used / row["memory_total_gb"] * 100, 1),
                        "temperature": row["base_temperature"] + (21 if busy else 6 if reserved else 0),
                        "power_w": round(row["power_limit_w"] * (0.7 if busy else 0.25 if reserved else 0.08), 1),
                        "power_limit_w": row["power_limit_w"],
                        "status": "busy" if busy else "reserved" if reserved else "idle",
                        "processes": [
                            {
                                "pid": task["process_id"],
                                "name": "llamafactory-cli",
                                "memory_used_gb": memory_used,
                                "task_name": task["name"],
                                "user": "admin",
                            }
                        ]
                        if task
                        else [],
                    }
                )
            return items

    def system_info(self) -> dict[str, Any]:
        gpus = self.gpus()
        busy = len([g for g in gpus if g["status"] in {"busy", "reserved"}])
        cpu_percent = 18 + busy * 9
        memory_percent = 37 + busy * 4
        return {
            "timestamp": utcnow(),
            "cpu": {
                "percent": min(cpu_percent, 95),
                "cores": 32,
                "percents": [min(cpu_percent + (i % 7) - 3, 99) for i in range(32)],
                "model": "Platform x86_64 CPU",
                "frequency_mhz": 2600,
                "load_1m": round(cpu_percent / 10, 2),
            },
            "memory": {
                "used_gb": round(256 * memory_percent / 100, 1),
                "total_gb": 256,
                "percent": min(memory_percent, 95),
                "available_gb": round(256 * (100 - memory_percent) / 100, 1),
                "cached_gb": 32,
            },
            "disk": {
                "used_gb": 840,
                "total_gb": 2048,
                "percent": 41,
                "read_mb_s": 120 if busy else 8,
                "write_mb_s": 95 if busy else 5,
            },
            "gpu": gpus,
            "network": {
                "download_mb_s": 12 if busy else 1.2,
                "upload_mb_s": 7 if busy else 0.8,
                "download_mb": 8024,
                "upload_mb": 1732,
            },
            "system": {
                "uptime_seconds": int(time.time() % 100000),
                "process_count": 248 + busy,
                "os": "Linux platform compute image",
            },
        }

    def health_metrics(self) -> dict[str, float]:
        info = self.system_info()
        return {
            "cpu_percent": info["cpu"]["percent"],
            "memory_percent": info["memory"]["percent"],
            "disk_percent": info["disk"]["percent"],
        }

    def queue(self) -> list[dict[str, Any]]:
        return [
            {
                "id": task["id"],
                "name": task["name"],
                "status": task["status"],
                "progress": task.get("progress", 0),
                "compute_node_id": task.get("compute_node_id"),
                "gpus": task.get("gpus", []),
                "create_time": task.get("create_time"),
            }
            for task in self.tasks()
            if task["status"] in {"pending", "syncing", "queued", "running"}
        ]

    def replicas(self, node_id: str) -> list[dict[str, Any]]:
        with self.connect() as conn:
            rows = conn.execute(
                "SELECT * FROM resource_replicas WHERE node_id=? ORDER BY create_time DESC",
                (node_id,),
            ).fetchall()
            return [dict(row) for row in rows]

    def sync_job(self, sync_id: str) -> dict[str, Any]:
        self.refresh_runtime_state()
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM resource_sync_jobs WHERE id=?", (sync_id,)).fetchone()
            if not row:
                raise KeyError(sync_id)
            return {**dict(row), "resources": json_loads(row["resources"], [])}

    def training_log_files(self) -> list[dict[str, Any]]:
        return [
            {
                "file": f"train_{task['id']}_pid{task['process_id'] or 0}.log",
                "name": task["name"],
                "pid": task.get("process_id") or 0,
                "size": f"{max(1, int((task.get('progress', 0) or 0) * 1.5))} KB",
                "date": task.get("create_time", "")[:10],
            }
            for task in self.tasks()
        ]

    def training_log_content(self, file_name: str) -> dict[str, Any]:
        task = next((t for t in self.tasks() if t["id"] in file_name), None)
        if not task:
            raise KeyError(file_name)
        content = self.generate_training_log(task)
        return {"file": file_name, "content": content, "size": f"{max(1, len(content.encode('utf-8')) // 1024)} KB"}

    def generate_training_log(self, task: dict[str, Any]) -> str:
        progress = int(task.get("progress", 0) or 0)
        points = max(1, min(80, progress))
        lines = [
            f"[INFO] task={task['name']} engine=llama_factory status={task['status']}",
            f"[INFO] base_model={task.get('base_model')} dataset={task.get('train_dataset_id')} gpus={task.get('gpus', [])}",
            "[INFO] command=llamafactory-cli train --stage sft --finetuning_type lora --do_train true",
        ]
        for step in range(1, points + 1):
            if step % 3 != 0 and step != points:
                continue
            loss = max(0.12, 2.4 * math.exp(-step / 42))
            grad_norm = 0.45 + (step % 8) * 0.03
            lr = float(task.get("learning_rate") or 0.0002) * max(0.05, 1 - step / 120)
            epoch = round(step / max(1, points) * float(task.get("n_epochs") or 3), 4)
            lines.append(
                "{"
                f"'loss': {loss:.4f}, 'grad_norm': {grad_norm:.4f}, "
                f"'learning_rate': {lr:.8f}, 'epoch': {epoch:.4f}"
                "}"
            )
        if task.get("status") == "completed":
            lines.extend(
                [
                    "***** train metrics *****",
                    f"epoch = {task.get('n_epochs', 3)}",
                    "train_loss = 0.1248",
                    f"train_runtime = {task.get('train_duration') or '1m 10s'}",
                    "***** train metrics end *****",
                ]
            )
        return "\n".join(lines)

    def log_files(self, date: str | None = None) -> list[dict[str, Any]]:
        today = date or utcnow()[:10]
        return [
            {"file": f"backend-{today}.log", "name": f"backend-{today}.log", "size": "32 KB", "date": today},
            {"file": f"error-{today}.log", "name": f"error-{today}.log", "size": "1 KB", "date": today},
        ]

    def log_content(self, file_name: str) -> dict[str, Any]:
        lines = [
            json_dumps(
                {
                    "timestamp": utcnow(),
                    "level": "INFO",
                    "logger": "platform",
                    "file": "backend/app/api/v1/endpoints/platform.py",
                    "line": 1,
                    "message": "Platform log stream is available.",
                }
            )
        ]
        return {"file": file_name, "content": "\n".join(lines), "size": "1 KB"}


_store: PlatformStore | None = None


def get_platform_store() -> PlatformStore:
    global _store
    if _store is None:
        _store = PlatformStore()
    return _store