app.py

"""
SyncServer broker: stores schematics in named pools, exposes HTTP + WebSocket for
Paper plugin sync, optional webhooks from the plugin, and scheduled backups.

Configuration: conf.txt next to this file (or path from SCHEMSYNC_CONF).
Precedence for each setting: environment variable > conf.txt > built-in default.

Pool model
----------
A "pool" is a named group of clients that share the same schematic set.
Pools are created automatically on first use (any authenticated request).
Schematics are stored under  data/schematics/<pool>/...
Backups are written to        data/backups/<pool>/schematics-YYYYMMDD-HHMMSS.zip
WebSocket connections are scoped to a single pool; broadcasts only reach that pool.
"""
from __future__ import annotations

import asyncio
import os
import re
import time
import zipfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Annotated, Optional

from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, UploadFile, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, Response

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

_POOL_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")


def _strip_inline_comment(value: str) -> str:
    value = value.strip()
    if "#" in value:
        value = value.split("#", 1)[0].strip()
    return value


def _load_conf_txt(path: Path) -> dict[str, str]:
    out: dict[str, str] = {}
    if not path.is_file():
        return out
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, rest = line.partition("=")
        key = key.strip().lower()
        val = _strip_inline_comment(rest)
        if key:
            out[key] = val
    return out


def _conf_path() -> Path:
    custom = os.environ.get("SCHEMSYNC_CONF", "").strip()
    if custom:
        return Path(custom).expanduser().resolve()
    return (Path(__file__).resolve().parent / "conf.txt").resolve()


def _pick(conf: dict[str, str], env_name: str, conf_keys: tuple[str, ...], default: str) -> str:
    env_val = os.environ.get(env_name)
    if env_val is not None and str(env_val).strip() != "":
        return str(env_val).strip()
    for ck in conf_keys:
        if ck in conf:
            return conf[ck]
    return default


@dataclass(frozen=True)
class BrokerSettings:
    host: str
    port: int
    data_dir: Path
    api_key: str
    backup_hours: float
    conf_path: Path


def load_settings() -> BrokerSettings:
    conf_path = _conf_path()
    conf = _load_conf_txt(conf_path)

    host = _pick(conf, "SCHEMSYNC_HOST", ("host", "bind", "listen_host"), "0.0.0.0")
    port_s = _pick(conf, "SCHEMSYNC_PORT", ("port",), "8756")
    try:
        port = int(port_s)
    except ValueError:
        port = 8756

    data_raw = _pick(conf, "SCHEMSYNC_DATA", ("data_dir", "data", "datadir"), "data")
    data_dir = Path(data_raw).expanduser()
    if not data_dir.is_absolute():
        data_dir = (Path(__file__).resolve().parent / data_dir).resolve()

    api_key = _pick(conf, "SCHEMSYNC_API_KEY", ("api_key", "apikey", "key"), "change-me-in-production")

    backup_s = _pick(conf, "SCHEMSYNC_BACKUP_HOURS", ("backup_hours", "backup_interval_hours"), "12")
    try:
        backup_hours = float(backup_s)
    except ValueError:
        backup_hours = 12.0

    return BrokerSettings(
        host=host,
        port=port,
        data_dir=data_dir,
        api_key=api_key,
        backup_hours=backup_hours,
        conf_path=conf_path,
    )


SETTINGS = load_settings()
DATA_DIR = SETTINGS.data_dir
SCHEMATICS_ROOT = DATA_DIR / "schematics"
BACKUPS_ROOT = DATA_DIR / "backups"
API_KEY = SETTINGS.api_key
BACKUP_INTERVAL_HOURS = SETTINGS.backup_hours

# ---------------------------------------------------------------------------
# Auth & path helpers
# ---------------------------------------------------------------------------

def require_api_key(x_api_key: Annotated[Optional[str], Header()] = None) -> None:
    if not API_KEY:
        return
    if not x_api_key or x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid or missing X-API-Key")


def validate_pool(pool: str) -> str:
    if not _POOL_RE.match(pool):
        raise HTTPException(status_code=400, detail="Pool name must be 1–64 alphanumeric/hyphen/underscore characters")
    return pool


def safe_relative_path(raw: str) -> Path:
    p = Path(raw.replace("\\", "/")).as_posix()
    parts = Path(p).parts
    if ".." in parts or p.startswith("/"):
        raise HTTPException(status_code=400, detail="Invalid path")
    return Path(p)


def pool_schematics_dir(pool: str) -> Path:
    d = SCHEMATICS_ROOT / pool
    d.mkdir(parents=True, exist_ok=True)
    return d


def pool_backups_dir(pool: str) -> Path:
    d = BACKUPS_ROOT / pool
    d.mkdir(parents=True, exist_ok=True)
    return d


def _mtime_ms(path: Path) -> int:
    return int(path.stat().st_mtime * 1000)

# ---------------------------------------------------------------------------
# FastAPI app + WebSocket state
# ---------------------------------------------------------------------------

app = FastAPI(title="SyncServer", version="2.0.0")

# pool name → set of connected WebSocket clients
_ws_pools: dict[str, set[WebSocket]] = {}


def _pool_clients(pool: str) -> set[WebSocket]:
    if pool not in _ws_pools:
        _ws_pools[pool] = set()
    return _ws_pools[pool]


async def _broadcast(pool: str, msg: dict) -> None:
    clients = _pool_clients(pool)
    dead: list[WebSocket] = []
    for ws in list(clients):
        try:
            await ws.send_json(msg)
        except Exception:
            dead.append(ws)
    for ws in dead:
        clients.discard(ws)


async def broadcast_schematic(pool: str, path: str) -> None:
    await _broadcast(pool, {"event": "schematic_updated", "path": path.replace("\\", "/")})


async def broadcast_repo_resync(pool: str) -> None:
    await _broadcast(pool, {"event": "repo_resync"})

# ---------------------------------------------------------------------------
# Startup / backup loop
# ---------------------------------------------------------------------------

@app.on_event("startup")
async def startup() -> None:
    SCHEMATICS_ROOT.mkdir(parents=True, exist_ok=True)
    BACKUPS_ROOT.mkdir(parents=True, exist_ok=True)
    asyncio.create_task(backup_loop())


async def backup_loop() -> None:
    while True:
        await asyncio.sleep(max(3600.0, BACKUP_INTERVAL_HOURS * 3600.0))
        try:
            ts = time.strftime("%Y%m%d-%H%M%S")
            if not SCHEMATICS_ROOT.is_dir():
                continue
            for pool_dir in SCHEMATICS_ROOT.iterdir():
                if not pool_dir.is_dir():
                    continue
                pool = pool_dir.name
                out = pool_backups_dir(pool) / f"schematics-{ts}.zip"
                with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zf:
                    for f in pool_dir.rglob("*"):
                        if f.is_file():
                            zf.write(f, arcname=f.relative_to(pool_dir).as_posix())
        except Exception:
            pass

# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------

@app.get("/health")
def health() -> dict:
    pools = [d.name for d in SCHEMATICS_ROOT.iterdir() if d.is_dir()] if SCHEMATICS_ROOT.is_dir() else []
    return {"ok": True, "conf": str(SETTINGS.conf_path), "pools": sorted(pools)}


@app.get("/v1/{pool}/list", dependencies=[Depends(require_api_key)])
def list_schematics(pool: str) -> list:
    validate_pool(pool)
    schematics_dir = pool_schematics_dir(pool)
    items: list[dict] = []
    for f in schematics_dir.rglob("*"):
        if f.is_file():
            rel = f.relative_to(schematics_dir).as_posix()
            items.append({"path": rel, "mtime_ms": _mtime_ms(f), "size": f.stat().st_size})
    return items


@app.get("/v1/{pool}/file/{full_path:path}", dependencies=[Depends(require_api_key)])
def get_file(pool: str, full_path: str) -> FileResponse:
    validate_pool(pool)
    schematics_dir = pool_schematics_dir(pool)
    rel = safe_relative_path(full_path)
    target = (schematics_dir / rel).resolve()
    if not str(target).startswith(str(schematics_dir.resolve())) or not target.is_file():
        raise HTTPException(status_code=404, detail="Not found")
    return FileResponse(
        target,
        media_type="application/octet-stream",
        headers={"X-Mtime-Ms": str(_mtime_ms(target))},
    )


@app.post("/v1/{pool}/upload", dependencies=[Depends(require_api_key)])
async def upload(
    pool: str,
    path: Annotated[str, Form()],
    file: Annotated[UploadFile, File()],
    x_client_mtime_ms: Annotated[Optional[str], Header()] = None,
    x_skip_ws_broadcast: Annotated[Optional[str], Header()] = None,
) -> dict:
    validate_pool(pool)
    schematics_dir = pool_schematics_dir(pool)
    rel = safe_relative_path(path)
    dest = (schematics_dir / rel).resolve()
    if not str(dest).startswith(str(schematics_dir.resolve())):
        raise HTTPException(status_code=400, detail="Invalid path")
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_bytes(await file.read())
    if x_client_mtime_ms:
        try:
            ms = int(x_client_mtime_ms) / 1000.0
            os.utime(dest, (ms, ms))
        except Exception:
            pass
    p = rel.as_posix()
    if x_skip_ws_broadcast != "1":
        await broadcast_schematic(pool, p)
    return {"ok": True, "pool": pool, "path": p}


@app.post("/v1/{pool}/notify/repo-resync", dependencies=[Depends(require_api_key)])
async def notify_repo_resync(pool: str) -> dict:
    validate_pool(pool)
    pool_schematics_dir(pool)  # ensure pool dir exists
    await broadcast_repo_resync(pool)
    return {"ok": True, "pool": pool}


@app.post("/v1/{pool}/hook/plugin", dependencies=[Depends(require_api_key)])
async def hook_plugin(pool: str, body: dict) -> Response:
    validate_pool(pool)
    path = str(body.get("path", "")).replace("\\", "/")
    if path:
        await broadcast_schematic(pool, path)
    return Response(status_code=204)


@app.websocket("/v1/{pool}/ws")
async def websocket_endpoint(pool: str, ws: WebSocket) -> None:
    if not _POOL_RE.match(pool):
        await ws.close(code=4400)
        return
    token = ws.query_params.get("token")
    if API_KEY and token != API_KEY:
        await ws.close(code=4401)
        return
    await ws.accept()
    clients = _pool_clients(pool)
    clients.add(ws)
    try:
        while True:
            await ws.receive_text()
    except WebSocketDisconnect:
        pass
    finally:
        clients.discard(ws)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app:app", host=SETTINGS.host, port=SETTINGS.port, reload=False)