app/api/user_backups_v6.py

"""v6 备份 API：/u/api/backups/v6/*

端点（12 个）：
- GET    /sets                           列出当前用户备份点
- POST   /sets                           创建待提交备份点
- GET    /sets/{setId}                   获取备份点 manifest
- DELETE /sets/{setId}                   删除备份点（快照模型：任意点可删）
- POST   /sets/{setId}/manifest          提交 manifest（原子生效）
- POST   /sets/{setId}/rename            备份点改名
- GET    /sets/{setId}/archive/info      ZIP 分卷信息
- GET    /sets/{setId}/archive?volume=N  下载第 N 卷 ZIP（request.download 通道）
- POST   /blobs/upload                   上传单个 blob（multipart，小文件）
- POST   /blobs/upload_part              大文件 base64 分片上传
- POST   /blobs/batch_check              卸载重装兜底（path+size+mtime 比对最新备份点）
- DELETE /legacy-v5                      清理当前用户的 v5 遗留数据（被污染，整体废弃）
"""
from __future__ import annotations

import asyncio
import logging

from fastapi import APIRouter, File, Form, Request, UploadFile
from fastapi.responses import FileResponse, StreamingResponse
from typing import Optional

from ..errors import HttpError
from ..services import backup_v6_store
from ._common import CORS_HEADERS, ok_with_cors, read_json_body
from .user_files import require_user, require_user_for_download

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/u/api/backups/v6", tags=["user-backups-v6"])

# 单 blob 上限：框架单文件硬限制 10MB，但恢复走 ZIP 分卷（每卷预算 8MB），
# 单 blob 必须给卷内 manifest.json 留余量，与前端 backup-v6.js 的 MAX_BLOB_SIZE 一致
MAX_BLOB_SIZE = 7 * 1024 * 1024

# 后台 Hub 同步任务引用（保存强引用防 task 被 GC 中途回收）
_sync_task_ref = None


def _err(code: str, message: str, status: int = 400, **extra) -> HttpError:
    return HttpError(message, status=status, code=code, details=extra or None)


def _user_id(request: Request) -> str:
    payload = require_user(request)
    uid = str(payload.get("sub") or "")
    if not uid:
        raise _err("unauthorized", "missing user", status=401)
    return uid


# ===== sets =====

@router.get("/sets")
async def list_sets(request: Request):
    uid = _user_id(request)
    try:
        limit = int(request.query_params.get("limit") or "30")
        offset = int(request.query_params.get("offset") or "0")
    except Exception:
        limit, offset = 30, 0
    # 本地 DB 为空时触发 Hub 同步（HF Space 重建后 DB 丢失）
    if backup_v6_store.count_sets(uid) == 0:
        global _sync_task_ref
        if _sync_task_ref is None or _sync_task_ref.done():
            _sync_task_ref = asyncio.create_task(backup_v6_store._sync_sets_background())
    items = backup_v6_store.list_sets(uid, limit=limit, offset=offset)
    return ok_with_cors({"items": items, "count": len(items), "total": len(items)})


@router.post("/sets")
async def create_set(request: Request):
    uid = _user_id(request)
    body = await read_json_body(request)
    alias = str(body.get("alias") or "").strip()[:200]
    try:
        result = backup_v6_store.create_pending_set(uid, alias=alias)
    except ValueError as e:
        msg = str(e)
        if msg == "quota_exceeded":
            raise _err("quota_exceeded", f"已达备份点上限（{backup_v6_store.MAX_SETS_PER_USER} 份），请先删除旧的备份", status=409)
        raise _err("create_failed", msg, status=500)
    # 小天才 fetch.fetch 对非 200 的 2xx（如 201）会触发 fail 回调，统一返回 200
    return ok_with_cors(result, status=200)


@router.get("/sets/{set_id}")
async def get_set(set_id: str, request: Request):
    uid = _user_id(request)
    owner = backup_v6_store.get_set_owner(set_id)
    if not owner:
        raise _err("set_not_found", "备份点不存在", status=404)
    if owner != uid:
        raise _err("forbidden", "无权访问他人备份", status=403)
    manifest = backup_v6_store.get_set_manifest(set_id)
    if manifest is None:
        raise _err("manifest_missing", "manifest 缺失", status=500)
    return ok_with_cors({"manifest": manifest})


@router.delete("/sets/{set_id}")
async def delete_set(set_id: str, request: Request):
    uid = _user_id(request)
    owner = backup_v6_store.get_set_owner(set_id)
    if not owner:
        raise _err("set_not_found", "备份点不存在", status=404)
    if owner != uid:
        raise _err("forbidden", "无权删除他人备份", status=403)
    try:
        result = backup_v6_store.delete_set(set_id, uid)
    except ValueError as e:
        msg = str(e)
        if msg == "set_not_found":
            raise _err("set_not_found", "备份点不存在", status=404)
        if msg == "forbidden":
            raise _err("forbidden", "无权删除他人备份", status=403)
        raise _err("delete_failed", msg, status=500)
    return ok_with_cors({"deleted": True, "setId": set_id, **result})


@router.post("/sets/{set_id}/manifest")
async def commit_manifest(set_id: str, request: Request):
    uid = _user_id(request)
    body = await read_json_body(request)
    manifest = body.get("manifest")
    if not isinstance(manifest, dict):
        raise _err("invalid_manifest", "manifest 必须是 JSON 对象")
    if str(manifest.get("setId") or "") != set_id:
        manifest["setId"] = set_id
    blobs = manifest.get("blobs")
    if not isinstance(blobs, list) or not blobs:
        raise _err("invalid_manifest_blobs", "manifest.blobs 必须是非空数组")
    keys = [str(b.get("key") or "") for b in blobs if isinstance(b, dict)]
    if not keys:
        raise _err("empty_blobs", "manifest.blobs 不能为空")
    # blob 可用性校验（含 Hub 回源，网络往返放工作线程避免阻塞事件循环）
    try:
        still_missing = await asyncio.to_thread(backup_v6_store.validate_blob_keys, keys)
    except Exception as e:
        raise _err("validate_failed", str(e), status=500)
    if still_missing:
        # 本地与 Hub 都没有：清理孤儿 DB 记录，让客户端重传
        await asyncio.to_thread(backup_v6_store.cleanup_missing_blob_records, still_missing)
        raise _err(
            "blob_missing", "manifest 引用了未上传的 blob",
            missing_keys=still_missing[:20],
        )
    # ★ 提交前强制确认所有 blob 已落 Hub（堵住静默丢失：本地有但 Hub 没有，
    # Space 重启清盘后 blob 永久丢失。失败则拒绝提交，客户端可重试）
    try:
        push_failed = await backup_v6_store.ensure_blobs_on_hub(keys)
    except Exception as e:
        raise _err("blob_push_failed", f"blob 同步到 Hub 失败: {e}", status=503)
    if push_failed:
        raise _err(
            "blob_push_failed", "部分 blob 同步到 Hub 失败，请稍后重试提交",
            push_failed=push_failed[:20],
            status=503,
        )
    try:
        result = backup_v6_store.commit_set(set_id, manifest, uid)
    except ValueError as e:
        msg = str(e)
        if msg == "set_not_pending":
            raise _err("set_not_pending", "备份点不在待提交状态（已提交或已过期）")
        if msg == "invalid_manifest":
            raise _err("invalid_manifest", "manifest 格式无效")
        if msg == "invalid_manifest_version":
            raise _err("invalid_manifest_version", "manifest.version 必须为 6")
        if msg in ("invalid_manifest_blobs", "empty_blobs"):
            raise _err(msg, "manifest.blobs 必须是非空数组")
        if msg.startswith("blob_missing:"):
            missing = msg.split(":", 1)[1].split(",")
            raise _err("blob_missing", "manifest 引用了未上传的 blob", missing_keys=missing)
        if msg == "quota_exceeded":
            raise _err("quota_exceeded", "已达备份点上限", status=409)
        raise _err("commit_failed", msg, status=500)
    return ok_with_cors(result)


@router.post("/sets/{set_id}/rename")
async def rename_set(set_id: str, request: Request):
    uid = _user_id(request)
    body = await read_json_body(request)
    alias = str(body.get("alias") or "").strip()[:200]
    if not alias:
        raise _err("invalid_alias", "别名不能为空")
    try:
        result = backup_v6_store.rename_set(set_id, uid, alias)
    except ValueError as e:
        msg = str(e)
        if msg == "set_not_found":
            raise _err("set_not_found", "备份点不存在", status=404)
        if msg == "forbidden":
            raise _err("forbidden", "无权操作他人备份", status=403)
        raise _err("rename_failed", msg, status=500)
    return ok_with_cors(result)


# ===== ZIP 分卷下载 =====

@router.get("/sets/{set_id}/archive/info")
async def archive_info(set_id: str, request: Request):
    uid = _user_id(request)
    owner = backup_v6_store.get_set_owner(set_id)
    if not owner:
        raise _err("set_not_found", "备份点不存在", status=404)
    if owner != uid:
        raise _err("forbidden", "无权访问他人备份", status=403)
    try:
        info = backup_v6_store.archive_info(set_id)
    except ValueError as e:
        if str(e) == "set_not_found":
            raise _err("set_not_found", "备份点不存在", status=404)
        raise _err("archive_info_failed", str(e), status=500)
    return ok_with_cors(info)


@router.get("/sets/{set_id}/archive")
async def download_archive(set_id: str, request: Request):
    payload = require_user_for_download(request)
    uid = str(payload.get("sub") or "")
    if not uid:
        raise _err("unauthorized", "missing user", status=401)
    owner = backup_v6_store.get_set_owner(set_id)
    if not owner:
        raise _err("set_not_found", "备份点不存在", status=404)
    if owner != uid:
        raise _err("forbidden", "无权下载他人备份", status=403)
    try:
        volume = int(request.query_params.get("volume") or "0")
    except Exception:
        volume = 0
    try:
        zip_path, zip_size = await asyncio.to_thread(
            backup_v6_store.build_volume_zip, set_id, volume
        )
    except ValueError as e:
        msg = str(e)
        if msg == "set_not_found":
            raise _err("set_not_found", "备份点不存在", status=404)
        if msg == "volume_out_of_range":
            raise _err("volume_out_of_range", "分卷号超出范围", status=400)
        raise _err("zip_build_failed", msg, status=500)

    filename = f"backup_{set_id[:8]}_v{volume}.zip"
    generator = backup_v6_store.zip_iter_chunks(zip_path)
    headers = {
        "Content-Disposition": f'attachment; filename="{filename}"',
        "Content-Length": str(zip_size),
        **CORS_HEADERS,
    }
    return StreamingResponse(generator, media_type="application/zip", headers=headers)


# ===== blobs =====

@router.post("/blobs/upload")
async def upload_blob(
    request: Request,
    file: UploadFile = File(...),
    path: str = Form(""),
    type: str = Form(""),
):
    uid = _user_id(request)
    content = await file.read()
    if len(content) > MAX_BLOB_SIZE:
        raise _err("blob_too_large", f"blob 超过 {MAX_BLOB_SIZE} 字节", status=413)
    mime = str(file.content_type or "")
    if not mime:
        mime = backup_v6_store._guess_mime(path or file.filename or "")
    try:
        result = backup_v6_store.store_blob(content, mime)
    except ValueError as e:
        if str(e) == "blob_too_large":
            raise _err("blob_too_large", "blob 过大", status=413)
        raise _err("store_failed", str(e), status=500)
    return ok_with_cors(result)


@router.post("/blobs/upload_part")
async def upload_blob_part(request: Request):
    """大文件分片上传：{uploadId, seq, dataB64, last} → {key, size, exists}。

    流程：POST /blobs/upload_part/start 拿 uploadId → 逐片 POST → last=true 收尾。
    """
    uid = _user_id(request)
    body = await read_json_body(request)
    upload_id = str(body.get("uploadId") or "")
    seq = body.get("seq")
    data_b64 = body.get("dataB64")
    last = body.get("last") is True
    if not upload_id:
        raise _err("invalid_upload_id", "uploadId 不能为空")
    try:
        seq = int(seq)
    except Exception:
        raise _err("invalid_seq", "seq 必须是整数")
    if not isinstance(data_b64, str) or not data_b64:
        raise _err("invalid_data", "dataB64 不能为空")
    try:
        result = backup_v6_store.upload_part(upload_id, uid, seq, data_b64, last)
    except ValueError as e:
        msg = str(e)
        if msg == "upload_not_found":
            raise _err("upload_not_found", "上传会话不存在或已过期", status=404)
        if msg == "upload_expired":
            raise _err("upload_expired", "上传会话已过期，请重新开始")
        if msg == "part_too_large":
            raise _err("part_too_large", f"单分片解码后超过 {backup_v6_store.MAX_PART_SIZE} 字节", status=413)
        if msg == "invalid_base64":
            raise _err("invalid_base64", "dataB64 不是合法 base64")
        raise _err("upload_failed", msg, status=500)
    return ok_with_cors(result)


@router.post("/blobs/upload_part/start")
async def upload_blob_part_start(request: Request):
    uid = _user_id(request)
    body = await read_json_body(request)
    path = str(body.get("path") or "")
    type_ = str(body.get("type") or "")
    if not path:
        raise _err("invalid_path", "path 不能为空")
    upload_id = backup_v6_store.start_upload(uid, path, type_)
    return ok_with_cors({"uploadId": upload_id, "expiresIn": backup_v6_store.PENDING_SET_TTL_SEC})


@router.post("/blobs/batch_check")
async def batch_check(request: Request):
    uid = _user_id(request)
    body = await read_json_body(request)
    paths = body.get("paths")
    if not isinstance(paths, list):
        raise _err("invalid_paths", "paths 必须是数组")
    if len(paths) > backup_v6_store.BATCH_CHECK_MAX:
        raise _err("too_many_paths", f"单次最多 {backup_v6_store.BATCH_CHECK_MAX} 个 path")
    cleaned = []
    for p in paths:
        if not isinstance(p, dict):
            continue
        path = str(p.get("path") or "").strip()
        if not path:
            continue
        try:
            size = int(p.get("size") or 0)
        except Exception:
            size = 0
        try:
            mtime = int(p.get("mtime") or 0)
        except Exception:
            mtime = 0
        cleaned.append({"path": path, "size": size, "mtime": mtime})
    # batch_check 现含 blob 可用性校验（本地缺失时有一次 Hub 清单网络往返），
    # 必须放 to_thread，否则阻塞事件循环
    try:
        result = await asyncio.to_thread(backup_v6_store.batch_check, uid, cleaned)
    except Exception as e:
        raise _err("batch_check_failed", str(e), status=500)
    return ok_with_cors(result)


@router.get("/blobs/{key}")
async def download_blob(key: str, request: Request):
    payload = require_user_for_download(request)
    uid = str(payload.get("sub") or "")
    if not uid:
        raise _err("unauthorized", "missing user", status=401)
    if not _is_sha256_hex(key):
        raise _err("invalid_key", "key 必须是 64 位 sha256 hex", status=400)
    # get_blob_local_path 本地缺失时会同步走 Hub 回源（网络 IO），
    # 必须放 to_thread，否则阻塞事件循环拖垮整个服务
    blob_path = await asyncio.to_thread(backup_v6_store.get_blob_local_path, key)
    if not blob_path or not blob_path.exists():
        raise _err("blob_not_found", "blob 不存在", status=404)
    meta = backup_v6_store.get_blob_meta(key)
    mime = (meta or {}).get("mime") or "application/octet-stream"
    headers = {
        "Content-Disposition": f'attachment; filename="{blob_path.name}"',
        **CORS_HEADERS,
    }
    return FileResponse(str(blob_path), media_type=mime, headers=headers)


# ===== v5 遗留数据清理 =====

@router.delete("/legacy-v5")
async def purge_legacy_v5(request: Request):
    """清理当前用户的 v5 遗留数据（被污染 + 设计缺陷，已整体废弃）。

    删除该用户全部 v5 备份集、refs=0 的 v5 blob（本地 + Hub）。
    v6 数据不受任何影响。
    """
    uid = _user_id(request)
    try:
        result = await asyncio.to_thread(backup_v6_store.purge_legacy_v5, uid)
    except Exception as e:
        raise _err("purge_failed", str(e), status=500)
    return ok_with_cors({"ok": True, "purged": result})


def _is_sha256_hex(s: str) -> bool:
    if len(s) != 64:
        return False
    try:
        int(s, 16)
        return True
    except Exception:
        return False